Function apply_post_aggregation

nodedb/src/control/server/post_aggregate.rs:18–68

Apply GROUP BY + aggregate functions on a join response payload. The input response payload is a msgpack array of maps (merged from all cores). Returns a new response with aggregated results (also msgpack).

(
    resp: Response,
    group_by: &[String],
    aggregates: &[(String, String)],
)

/// The input response payload is a msgpack array of maps (merged from all cores).
/// Returns a new response with aggregated results (also msgpack).
pub fn apply_post_aggregation(
    resp: Response,
    group_by: &[String],
    aggregates: &[(String, String)],
) -> crate::Result<Response> {
    let payload_bytes = resp.payload.as_bytes();

    // Parse the msgpack array of row-maps.
    let rows = parse_msgpack_rows(payload_bytes)?;

    // Group rows by the GROUP BY columns.
    let mut groups: HashMap<Vec<String>, Vec<&[u8]>> = HashMap::new();
    for row in &rows {
        let key: Vec<String> = group_by
            .iter()
            .map(|col| extract_field_str(row, col).unwrap_or_default())
            .collect();
        groups.entry(key).or_default().push(row);
    }

    // Build result as msgpack array.
    use nodedb_query::msgpack_scan::writer;
    let mut buf = Vec::with_capacity(payload_bytes.len());
    writer::write_array_header(&mut buf, groups.len());

    for (key, group_rows) in &groups {
        let field_count = group_by.len() + aggregates.len();
        writer::write_map_header(&mut buf, field_count);

        // Write GROUP BY columns.
        for (i, col) in group_by.iter().enumerate() {
            writer::write_kv_str(&mut buf, col, &key[i]);
        }

        // Compute and write each aggregate.
        for (op, field) in aggregates {
            let agg_key = canonical_agg_key(op, field);
            let value = compute_aggregate(op, field, group_rows);
            match value {
                AggValue::Int(n) => writer::write_kv_i64(&mut buf, &agg_key, n),
                AggValue::Float(f) => writer::write_kv_f64(&mut buf, &agg_key, f),
                AggValue::Null => writer::write_kv_null(&mut buf, &agg_key),
            }
        }
    }

    Ok(Response {
        payload: Payload::from_vec(buf),
        ..resp
    })
}

enum AggValue {
    Int(i64),