nodedb_types/timeseries/
continuous_agg.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Continuous aggregate definition types.
4//!
5//! Shared between Origin and Lite. Origin uses these for SQL DDL parsing
6//! and the continuous aggregate manager. Lite uses them for the embedded
7//! continuous aggregate engine and its DDL handler.
8
9use serde::{Deserialize, Serialize};
10
11/// Definition of a continuous aggregate.
12///
13/// Map-encoded so fields (e.g. `database_id`) can be added with
14/// `#[msgpack(default)]` and older persisted definitions still decode
15/// without a migration.
16///
17/// `Default` gives an empty definition whose `incarnation` is `Hlc::ZERO`.
18/// No CREATE stamps that incarnation, and Origin never installs persisted
19/// state into, or from, a definition that carries it.
20#[derive(
21    Debug,
22    Clone,
23    Default,
24    PartialEq,
25    Serialize,
26    Deserialize,
27    zerompk::ToMessagePack,
28    zerompk::FromMessagePack,
29)]
30#[msgpack(map)]
31pub struct ContinuousAggregateDef {
32    /// Owning database. Scopes the Data-Plane manager's per-name maps and
33    /// catalog keys so an aggregate named identically in two databases
34    /// never collides or materializes against the wrong database's storage.
35    ///
36    /// `#[msgpack(default)]`: definitions persisted before database scoping
37    /// decode with `0` (`DatabaseId::DEFAULT`) — the database those legacy
38    /// rows lived in — so no migration is required.
39    #[msgpack(default)]
40    pub database_id: u64,
41    /// Name of this aggregate (e.g., "metrics_1m").
42    pub name: String,
43    /// Source collection or aggregate to read from.
44    pub source: String,
45    /// Time bucket interval string (e.g., "1m", "1h", "1d").
46    pub bucket_interval: String,
47    /// Bucket interval in milliseconds (computed from bucket_interval).
48    pub bucket_interval_ms: i64,
49    /// Columns to GROUP BY (tag/symbol columns).
50    pub group_by: Vec<String>,
51    /// Aggregate expressions to compute.
52    pub aggregates: Vec<AggregateExpr>,
53    /// When to refresh.
54    pub refresh_policy: RefreshPolicy,
55    /// Retention period in milliseconds (0 = infinite, independent of source).
56    pub retention_period_ms: u64,
57    /// Whether this aggregate is currently stale (schema change invalidation).
58    pub stale: bool,
59    /// Identity of the CREATE that made this definition. A re-created
60    /// aggregate of the same name never loads the persisted buckets of a
61    /// dropped one.
62    #[msgpack(default)]
63    #[serde(default)]
64    pub incarnation: crate::Hlc,
65}
66
67/// An aggregate expression: function + source column → result column.
68#[derive(
69    Debug,
70    Clone,
71    PartialEq,
72    Serialize,
73    Deserialize,
74    zerompk::ToMessagePack,
75    zerompk::FromMessagePack,
76)]
77pub struct AggregateExpr {
78    /// Aggregate function.
79    pub function: AggFunction,
80    /// Source column name (e.g., "cpu"). "*" for COUNT.
81    pub source_column: String,
82    /// Output column name (e.g., "cpu_avg"). Auto-generated if empty.
83    pub output_column: String,
84}
85
86/// Supported aggregate functions.
87#[derive(
88    Debug,
89    Clone,
90    PartialEq,
91    Serialize,
92    Deserialize,
93    zerompk::ToMessagePack,
94    zerompk::FromMessagePack,
95)]
96#[serde(rename_all = "snake_case")]
97#[non_exhaustive]
98pub enum AggFunction {
99    #[serde(rename = "sum")]
100    Sum,
101    #[serde(rename = "count")]
102    Count,
103    #[serde(rename = "min")]
104    Min,
105    #[serde(rename = "max")]
106    Max,
107    #[serde(rename = "avg")]
108    Avg,
109    #[serde(rename = "first")]
110    First,
111    #[serde(rename = "last")]
112    Last,
113    /// Approximate count distinct via HyperLogLog.
114    #[serde(rename = "count_distinct")]
115    CountDistinct,
116    /// Approximate percentile via TDigest. Inner value is the quantile (0.0–1.0).
117    #[serde(rename = "percentile")]
118    Percentile(f64),
119    /// Approximate top-K heavy hitters via SpaceSaving. Inner value is K.
120    #[serde(rename = "topk")]
121    TopK(usize),
122}
123
124impl AggFunction {
125    pub fn as_str(&self) -> &'static str {
126        match self {
127            Self::Sum => "sum",
128            Self::Count => "count",
129            Self::Min => "min",
130            Self::Max => "max",
131            Self::Avg => "avg",
132            Self::First => "first",
133            Self::Last => "last",
134            Self::CountDistinct => "count_distinct",
135            Self::Percentile(_) => "percentile",
136            Self::TopK(_) => "topk",
137        }
138    }
139
140    /// Whether this function requires sketch state in PartialAggregate.
141    pub fn uses_sketch(&self) -> bool {
142        matches!(
143            self,
144            Self::CountDistinct | Self::Percentile(_) | Self::TopK(_)
145        )
146    }
147}
148
149/// When to refresh the aggregate.
150#[derive(
151    Debug,
152    Clone,
153    Default,
154    PartialEq,
155    Eq,
156    Serialize,
157    Deserialize,
158    zerompk::ToMessagePack,
159    zerompk::FromMessagePack,
160)]
161#[serde(rename_all = "snake_case")]
162#[non_exhaustive]
163pub enum RefreshPolicy {
164    /// Refresh on every memtable flush. Lowest latency.
165    #[default]
166    #[serde(rename = "on_flush")]
167    OnFlush,
168    /// Refresh when a partition is sealed. Lower CPU cost.
169    #[serde(rename = "on_seal")]
170    OnSeal,
171    /// Refresh every N milliseconds.
172    #[serde(rename = "periodic")]
173    Periodic(u64),
174    /// Only refresh via explicit REFRESH command.
175    #[serde(rename = "manual")]
176    Manual,
177}
nodedb_types/timeseries/continuous_agg.rs

nodedb_types/timeseries/
continuous_agg.rs
