nodedb 0.2.1

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

// SPDX-License-Identifier: BUSL-1.1

//! Accumulator and per-group state type definitions.
//!
//! Each `AggAccum` variant holds only the derived state needed to compute the
//! final aggregate result — no raw document bytes are retained.  Memory per
//! group is O(num_aggregates × accumulator_size) regardless of how many
//! documents match that group.

use std::collections::{HashMap, HashSet};

use crate::bridge::physical_plan::AggregateSpec;
use nodedb_types::Value;

/// Maximum items collected by materializing aggregates (`array_agg`,
/// `array_agg_distinct`, `percentile_cont`, `string_agg`).
pub(super) const ARRAY_AGG_CAP: usize = 10_000;

/// Per-(group, aggregate-spec) running accumulator.
///
/// Derives `Serialize` / `Deserialize` so that partial states can be spilled
/// to disk by `GroupBySpiller` and merged back during finalize.
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) enum AggAccum {
    /// count(*) or count(field).
    Count { n: u64 },
    /// sum / avg: Kahan-compensated running sum + count.
    SumAvg { sum: f64, comp: f64, n: u64 },
    /// sum(DISTINCT col) / avg(DISTINCT col): map each distinct input
    /// value (keyed by its raw msgpack bytes) to its parsed numeric
    /// value. The sum and count are derived at finalize time, so the
    /// state is order-independent and therefore mergeable across
    /// spilled runs. Memory: O(num_distinct).
    SumAvgDistinct { seen: HashMap<Vec<u8>, f64> },
    /// min.
    Min { best: Option<Value> },
    /// max.
    Max { best: Option<Value> },
    /// count_distinct: set of raw msgpack bytes.
    CountDistinct { seen: HashSet<Vec<u8>> },
    /// stddev / variance variants: Welford M2 accumulator.
    Welford { n: u64, mean: f64, m2: f64 },
    /// approx_count_distinct: HyperLogLog.
    Hll {
        hll: nodedb_types::approx::HyperLogLog,
    },
    /// approx_percentile: t-digest.
    TDigest {
        digest: nodedb_types::approx::TDigest,
    },
    /// approx_topk: space-saving.
    TopK {
        ss: nodedb_types::approx::SpaceSaving,
        k: usize,
    },
    /// array_agg (capped).
    ArrayAgg { values: Vec<Value> },
    /// array_agg_distinct (capped).
    ArrayAggDistinct {
        seen: HashSet<Vec<u8>>,
        values: Vec<Value>,
    },
    /// percentile_cont (capped).
    PercentileCont { values: Vec<f64>, pct: f64 },
    /// string_agg / group_concat (capped).
    StringAgg { parts: Vec<String> },
}

/// Per-group running state: one `AggAccum` per aggregate spec.
///
/// Serializable so that `GroupBySpiller` can spill partial states to disk.
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct GroupState {
    pub(super) accums: Vec<AggAccum>,
}

impl GroupState {
    pub(crate) fn new(aggregates: &[AggregateSpec]) -> Self {
        Self {
            accums: aggregates.iter().map(AggAccum::new).collect(),
        }
    }

    pub(crate) fn feed(&mut self, aggregates: &[AggregateSpec], doc: &[u8]) {
        for (accum, agg) in self.accums.iter_mut().zip(aggregates) {
            accum.feed(agg, doc);
        }
    }

    /// Merge a partial `GroupState` from a spilled run into `self`.
    ///
    /// Delegates to `merge::merge_group_state`.
    pub(crate) fn merge_from(&mut self, other: GroupState) {
        super::merge::merge_group_state(self, other);
    }

    pub(crate) fn finalize(self, aggregates: &[AggregateSpec]) -> Vec<(String, Value)> {
        self.accums
            .into_iter()
            .zip(aggregates)
            .map(|(accum, agg)| (agg.alias.clone(), accum.finalize(agg)))
            .collect()
    }
}