Function compute_aggregate_binary

nodedb-query/src/msgpack_scan/aggregate.rs:25–237

Compute an aggregate function over raw MessagePack documents. Each entry in `docs` is a complete MessagePack map (the raw bytes from storage). Returns the result as `Value` — conversion to JSON happens at the response boundary only.

(
    op: &str,
    field: &str,
    expr: Option<&crate::expr::SqlExpr>,
    docs: &[&[u8]],
)

/// Returns the result as `Value` — conversion to JSON happens at the
/// response boundary only.
pub fn compute_aggregate_binary(
    op: &str,
    field: &str,
    expr: Option<&crate::expr::SqlExpr>,
    docs: &[&[u8]],
) -> Value {
    match op {
        "count" => {
            if field == "*" && expr.is_none() {
                Value::Integer(docs.len() as i64)
            } else {
                let count = docs
                    .iter()
                    .filter_map(|d| extract_as_value(d, field, expr))
                    .filter(|v| !v.is_null())
                    .count();
                Value::Integer(count as i64)
            }
        }

        "sum" => {
            let total: f64 = docs
                .iter()
                .filter_map(|d| extract_f64_val(d, field, expr))
                .sum();
            Value::Float(total)
        }

        "avg" => {
            let (sum, count) = docs
                .iter()
                .filter_map(|d| extract_f64_val(d, field, expr))
                .fold((0.0f64, 0u64), |(s, c), v| (s + v, c + 1));
            if count == 0 {
                Value::Null
            } else {
                Value::Float(sum / count as f64)
            }
        }

        "min" => find_minmax(docs, field, expr, false),
        "max" => find_minmax(docs, field, expr, true),

        "count_distinct" => {
            let mut seen = HashSet::new();
            for doc in docs {
                if let Some(bytes) = extract_value_bytes(doc, field, expr)
                    && !value_bytes_are_null(&bytes)
                {
                    seen.insert(bytes);
                }
            }
            Value::Integer(seen.len() as i64)
        }

        "stddev" | "stddev_pop" => {
            stat_aggregate(docs, field, expr, |variance, _n| variance.sqrt(), true)
        }