Function reduce_agg_partials

nodedb-cluster/src/distributed_array/merge.rs:134–146

Merge per-shard partial aggregates into one result per group-by key. Groups by `group_key`; uses `ArrayAggPartial::merge` for each group.

(shard_resps: &[ArrayShardAggResp])

///
/// Groups by `group_key`; uses `ArrayAggPartial::merge` for each group.
pub fn reduce_agg_partials(shard_resps: &[ArrayShardAggResp]) -> Vec<ArrayAggPartial> {
    use std::collections::BTreeMap;
    let mut buckets: BTreeMap<i64, ArrayAggPartial> = BTreeMap::new();
    for resp in shard_resps {
        for partial in &resp.partials {
            buckets
                .entry(partial.group_key)
                .and_modify(|existing| existing.merge(partial))
                .or_insert_with(|| partial.clone());
        }
    }
    buckets.into_values().collect()
}

#[cfg(test)]
mod tests {