nodedb_sql/engine_rules/
timeseries.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Engine rules for timeseries collections.
4//!
5//! Timeseries is append-only. INSERT, UPDATE, and UPSERT are not supported.
6//! Data enters via INGEST (mapped to `TimeseriesIngest`).
7//! Scans route to `TimeseriesScan` for time-range-aware execution.
8
9use crate::engine_rules::timeseries_native::native_shape;
10use crate::engine_rules::*;
11use crate::error::{Result, SqlError};
12use crate::types::*;
13
14pub struct TimeseriesRules;
15
16impl EngineRules for TimeseriesRules {
17    fn plan_insert(&self, p: InsertParams) -> Result<Vec<SqlPlan>> {
18        // Timeseries INSERT routes to TimeseriesIngest — append-only semantics.
19        Ok(vec![SqlPlan::TimeseriesIngest(TimeseriesIngestPlan {
20            collection: p.collection,
21            rows: p.rows,
22            volatile_defaults: p.volatile_defaults,
23        })])
24    }
25
26    fn plan_insert_select(&self, p: InsertSelectParams) -> Result<Vec<SqlPlan>> {
27        Err(SqlError::Unsupported {
28            detail: format!(
29                "INSERT ... SELECT into timeseries collection '{}' is not supported; \
30                 the target must be a document collection",
31                p.collection
32            ),
33        })
34    }
35
36    fn plan_upsert(&self, _p: UpsertParams) -> Result<Vec<SqlPlan>> {
37        Err(SqlError::Unsupported {
38            detail: "UPSERT is not supported on timeseries collections (append-only)".into(),
39        })
40    }
41
42    fn plan_scan(&self, p: ScanParams) -> Result<SqlPlan> {
43        if p.temporal.is_temporal() && !p.bitemporal {
44            return Err(SqlError::Unsupported {
45                detail: format!(
46                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
47                     timeseries collection; '{}' was not created WITH bitemporal = true",
48                    p.collection
49                ),
50            });
51        }
52        // Timeseries scans use TimeseriesScan for time-range-aware execution.
53        let time_range = default_time_range();
54        Ok(SqlPlan::TimeseriesScan(TimeseriesScanPlan {
55            collection: p.collection,
56            time_range,
57            bucket_interval_ms: 0,
58            group_by: Vec::new(),
59            aggregates: Vec::new(),
60            filters: p.filters,
61            having: Vec::new(),
62            projection: p.projection,
63            gap_fill: String::new(),
64            // No LIMIT reads every row. The Data Plane bounds that read by
65            // the per-query byte budget and fails it loudly past the budget.
66            limit: p.limit.unwrap_or(usize::MAX),
67            sort_keys: p.sort_keys,
68            tiered: p.has_auto_tier,
69            temporal: p.temporal,
70        }))
71    }
72
73    fn plan_point_get(&self, _p: PointGetParams) -> Result<SqlPlan> {
74        Err(SqlError::Unsupported {
75            detail: "point lookups are not supported on timeseries collections; \
76                     use SELECT with a time range filter instead"
77                .into(),
78        })
79    }
80
81    fn plan_update(&self, _p: UpdateParams) -> Result<Vec<SqlPlan>> {
82        Err(SqlError::Unsupported {
83            detail: "UPDATE is not supported on timeseries collections; \
84                     timeseries data is append-only"
85                .into(),
86        })
87    }
88
89    fn plan_update_from(&self, p: UpdateFromParams) -> Result<Vec<SqlPlan>> {
90        Err(SqlError::Unsupported {
91            detail: format!(
92                "UPDATE ... FROM is not supported on timeseries collection '{}'; \
93                 timeseries data is append-only",
94                p.collection
95            ),
96        })
97    }
98
99    fn plan_delete(&self, _p: DeleteParams) -> Result<Vec<SqlPlan>> {
100        // Timeseries has no row-level delete; only whole-partition drop via
101        // retention policy. Routing to columnar deletes nothing.
102        Err(SqlError::Unsupported {
103            detail: "DELETE is not supported on timeseries collections; \
104                     expire rows with CREATE RETENTION POLICY, which drops \
105                     whole partitions older than the configured age"
106                .into(),
107        })
108    }
109
110    fn plan_truncate(&self, p: TruncateParams) -> Result<Vec<SqlPlan>> {
111        Ok(vec![SqlPlan::Truncate {
112            collection: p.collection,
113            engine: EngineType::Timeseries,
114            restart_identity: p.restart_identity,
115        }])
116    }
117
118    fn plan_aggregate(&self, p: AggregateParams) -> Result<SqlPlan> {
119        if p.temporal.is_temporal() && !p.bitemporal {
120            return Err(SqlError::Unsupported {
121                detail: format!(
122                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
123                     timeseries collection; '{}' was not created WITH bitemporal = true",
124                    p.collection
125                ),
126            });
127        }
128        let Some(shape) = native_shape(&p) else {
129            // The generic aggregate runs over the raw row source: memtable,
130            // flushed partitions, staged rows, read policy and bitemporal
131            // scope. On an auto-tier collection the scan refuses a range that
132            // only downsampled tiers hold, since those carry no raw rows.
133            let input = SqlPlan::TimeseriesScan(TimeseriesScanPlan {
134                collection: p.collection,
135                time_range: default_time_range(),
136                bucket_interval_ms: 0,
137                group_by: Vec::new(),
138                aggregates: Vec::new(),
139                filters: p.filters,
140                having: Vec::new(),
141                projection: Vec::new(),
142                gap_fill: String::new(),
143                limit: usize::MAX,
144                sort_keys: Vec::new(),
145                tiered: p.has_auto_tier,
146                temporal: p.temporal,
147            });
148            return Ok(SqlPlan::Aggregate {
149                input: Box::new(input),
150                group_by: p.group_by,
151                group_by_aliases: Vec::new(),
152                output_order: Vec::new(),
153                aggregates: p.aggregates,
154                having: p.having,
155                limit: p.limit,
156                grouping_sets: p.grouping_sets,
157                sort_keys: Vec::new(),
158            });
159        };
160        Ok(SqlPlan::TimeseriesScan(TimeseriesScanPlan {
161            collection: p.collection,
162            time_range: default_time_range(),
163            bucket_interval_ms: shape.bucket_interval_ms,
164            group_by: shape.group_columns,
165            aggregates: p.aggregates,
166            filters: p.filters,
167            having: p.having,
168            projection: Vec::new(),
169            gap_fill: String::new(),
170            limit: p.limit,
171            // ORDER BY is attached later by `apply_order_by`, which sees the
172            // outer SELECT's sort list.
173            sort_keys: Vec::new(),
174            tiered: p.has_auto_tier,
175            temporal: p.temporal,
176        }))
177    }
178
179    fn plan_merge(&self, p: MergeParams) -> Result<Vec<SqlPlan>> {
180        Err(SqlError::Unsupported {
181            detail: format!(
182                "MERGE is not supported on timeseries collection '{}'",
183                p.collection
184            ),
185        })
186    }
187
188    /// Every column is declared at creation and encoded per column.
189    fn accepts_undeclared_columns(&self) -> bool {
190        false
191    }
192}
193
194/// Default time range bounds for the SqlPlan IR.
195///
196/// Actual time range extraction from filter predicates happens during
197/// PhysicalPlan conversion (Origin: `value::extract_time_range`).
198/// At the SqlPlan level, we pass unbounded defaults.
199fn default_time_range() -> (i64, i64) {
200    (i64::MIN, i64::MAX)
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    fn scan(limit: Option<usize>) -> SqlPlan {
208        TimeseriesRules
209            .plan_scan(ScanParams {
210                collection: "metrics".into(),
211                alias: None,
212                filters: Vec::new(),
213                projection: Vec::new(),
214                sort_keys: Vec::new(),
215                limit,
216                offset: 0,
217                distinct: false,
218                window_functions: Vec::new(),
219                indexes: Vec::new(),
220                temporal: crate::temporal::TemporalScope::default(),
221                bitemporal: false,
222                has_auto_tier: false,
223            })
224            .expect("plans")
225    }
226
227    /// No LIMIT reads every row. A 10,000-row default cut the read silently.
228    #[test]
229    fn plan_scan_without_limit_is_unbounded() {
230        let SqlPlan::TimeseriesScan(plan) = scan(None) else {
231            panic!("expected a timeseries scan");
232        };
233        assert_eq!(plan.limit, usize::MAX);
234        let SqlPlan::TimeseriesScan(plan) = scan(Some(5)) else {
235            panic!("expected a timeseries scan");
236        };
237        assert_eq!(plan.limit, 5);
238    }
239}
nodedb_sql/engine_rules/timeseries.rs

nodedb_sql/engine_rules/
timeseries.rs
