nodedb_sql/engine_rules/
continuous_aggregate.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Engine rules for a continuous aggregate read as a relation.
4//!
5//! A continuous aggregate is read-only: its rows are the finalized buckets
6//! the aggregate maintains from its source. Every write refuses with
7//! SQLSTATE `42809`. A read plans as a scan of the aggregate's rows, and an
8//! aggregate query over it aggregates those rows. A key equality never
9//! becomes a point lookup: the rows have no primary key.
10
11use crate::engine_rules::*;
12use crate::error::{Result, SqlError};
13use crate::types::*;
14
15pub struct ContinuousAggregateRules;
16
17/// The refusal every write to continuous aggregate `name` gets.
18fn read_only(op: &str, name: &str) -> SqlError {
19    SqlError::WrongObjectType {
20        detail: format!("continuous aggregate '{name}' is read-only; {op} is not allowed"),
21    }
22}
23
24fn refuse_temporal(collection: &str, temporal: &crate::temporal::TemporalScope) -> Result<()> {
25    if temporal.is_temporal() {
26        return Err(SqlError::Unsupported {
27            detail: format!(
28                "FOR SYSTEM_TIME / FOR VALID_TIME is not supported on continuous \
29                 aggregate '{collection}'"
30            ),
31        });
32    }
33    Ok(())
34}
35
36impl EngineRules for ContinuousAggregateRules {
37    fn plan_insert(&self, p: InsertParams) -> Result<Vec<SqlPlan>> {
38        Err(read_only("INSERT", &p.collection))
39    }
40
41    fn plan_insert_select(&self, p: InsertSelectParams) -> Result<Vec<SqlPlan>> {
42        Err(read_only("INSERT ... SELECT", &p.collection))
43    }
44
45    fn plan_upsert(&self, p: UpsertParams) -> Result<Vec<SqlPlan>> {
46        Err(read_only("UPSERT", &p.collection))
47    }
48
49    fn plan_scan(&self, p: ScanParams) -> Result<SqlPlan> {
50        refuse_temporal(&p.collection, &p.temporal)?;
51        Ok(SqlPlan::Scan {
52            collection: p.collection,
53            alias: p.alias,
54            engine: EngineType::ContinuousAggregate,
55            filters: p.filters,
56            projection: p.projection,
57            sort_keys: p.sort_keys,
58            limit: p.limit,
59            offset: p.offset,
60            distinct: p.distinct,
61            window_functions: p.window_functions,
62            temporal: p.temporal,
63        })
64    }
65
66    /// The optimizer never rewrites a continuous-aggregate read into a point
67    /// lookup (`has_point_lookup`), so no plan reaches here.
68    fn plan_point_get(&self, p: PointGetParams) -> Result<SqlPlan> {
69        Err(SqlError::Unsupported {
70            detail: format!(
71                "continuous aggregate '{}' has no primary key to look up",
72                p.collection
73            ),
74        })
75    }
76
77    fn plan_update(&self, p: UpdateParams) -> Result<Vec<SqlPlan>> {
78        Err(read_only("UPDATE", &p.collection))
79    }
80
81    fn plan_update_from(&self, p: UpdateFromParams) -> Result<Vec<SqlPlan>> {
82        Err(read_only("UPDATE ... FROM", &p.collection))
83    }
84
85    fn plan_delete(&self, p: DeleteParams) -> Result<Vec<SqlPlan>> {
86        Err(read_only("DELETE", &p.collection))
87    }
88
89    fn plan_truncate(&self, p: TruncateParams) -> Result<Vec<SqlPlan>> {
90        Err(read_only("TRUNCATE", &p.collection))
91    }
92
93    fn plan_aggregate(&self, p: AggregateParams) -> Result<SqlPlan> {
94        refuse_temporal(&p.collection, &p.temporal)?;
95        let base_scan = SqlPlan::Scan {
96            collection: p.collection,
97            alias: p.alias,
98            engine: EngineType::ContinuousAggregate,
99            filters: p.filters,
100            projection: Vec::new(),
101            sort_keys: Vec::new(),
102            limit: None,
103            offset: 0,
104            distinct: false,
105            window_functions: Vec::new(),
106            temporal: p.temporal,
107        };
108        Ok(SqlPlan::Aggregate {
109            input: Box::new(base_scan),
110            group_by: p.group_by,
111            group_by_aliases: Vec::new(),
112            output_order: Vec::new(),
113            aggregates: p.aggregates,
114            having: p.having,
115            limit: p.limit,
116            grouping_sets: p.grouping_sets,
117            sort_keys: Vec::new(),
118        })
119    }
120
121    fn plan_merge(&self, p: MergeParams) -> Result<Vec<SqlPlan>> {
122        Err(read_only("MERGE", &p.collection))
123    }
124
125    /// The columns are the bucket, the GROUP BY columns and the outputs.
126    fn accepts_undeclared_columns(&self) -> bool {
127        false
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134
135    #[test]
136    fn every_write_is_refused_as_read_only() {
137        let r = ContinuousAggregateRules;
138        let err = r
139            .plan_insert(InsertParams {
140                collection: "cpu_1m".into(),
141                columns: vec![],
142                rows: vec![],
143                volatile_defaults: false,
144                if_absent: false,
145                column_schema: vec![],
146                primary_key: None,
147            })
148            .unwrap_err();
149        assert!(
150            matches!(&err, SqlError::WrongObjectType { detail } if detail.contains("read-only")),
151            "{err:?}"
152        );
153        assert!(matches!(
154            r.plan_truncate(TruncateParams {
155                collection: "cpu_1m".into(),
156                restart_identity: false,
157            })
158            .unwrap_err(),
159            SqlError::WrongObjectType { .. }
160        ));
161    }
162}
nodedb_sql/engine_rules/continuous_aggregate.rs

nodedb_sql/engine_rules/
continuous_aggregate.rs
