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

nodedb_types/timeseries/
continuous_agg.rs
