nodedb_sql/planner/
array_ddl.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Planner for `CREATE ARRAY` and `DROP ARRAY`.
4//!
5//! Validation is engine-agnostic: name non-empty, dims non-empty, attrs
6//! non-empty, `tile_extents.len() == dims.len()`, no duplicate dim/attr
7//! names, and per-dim domain bounds are well formed (lo <= hi). The
8//! schema-hash + the typed `ArraySchema` are computed in the Origin
9//! converter where `nodedb-array` is available.
10
11use crate::types::{AlterArrayPlan, CreateArrayPlan};
12use std::collections::HashSet;
13
14use crate::error::{Result, SqlError};
15use crate::parser::array_stmt::{AlterArrayAst, CreateArrayAst, DropArrayAst};
16use crate::types::SqlPlan;
17use crate::types_array::{ArrayDimType, ArrayDomainBound};
18
19pub fn plan_create_array(ast: &CreateArrayAst) -> Result<SqlPlan> {
20    if ast.name.is_empty() {
21        return Err(SqlError::Parse {
22            detail: "CREATE ARRAY: array name must not be empty".into(),
23        });
24    }
25    if ast.dims.is_empty() {
26        return Err(SqlError::Parse {
27            detail: format!("CREATE ARRAY {}: at least one dim is required", ast.name),
28        });
29    }
30    if ast.attrs.is_empty() {
31        return Err(SqlError::Parse {
32            detail: format!("CREATE ARRAY {}: at least one attr is required", ast.name),
33        });
34    }
35    if ast.tile_extents.len() != ast.dims.len() {
36        return Err(SqlError::Parse {
37            detail: format!(
38                "CREATE ARRAY {}: TILE_EXTENTS arity {} != DIMS arity {}",
39                ast.name,
40                ast.tile_extents.len(),
41                ast.dims.len()
42            ),
43        });
44    }
45    let mut seen_dims: HashSet<&str> = HashSet::new();
46    for d in &ast.dims {
47        if d.name.is_empty() {
48            return Err(SqlError::Parse {
49                detail: format!("CREATE ARRAY {}: empty dim name", ast.name),
50            });
51        }
52        if !seen_dims.insert(d.name.as_str()) {
53            return Err(SqlError::Parse {
54                detail: format!("CREATE ARRAY {}: duplicate dim name `{}`", ast.name, d.name),
55            });
56        }
57        if !domain_bounds_match_type(&d.lo, &d.hi, d.dtype) {
58            return Err(SqlError::Parse {
59                detail: format!(
60                    "CREATE ARRAY {}: domain bounds for dim `{}` do not match dim type",
61                    ast.name, d.name
62                ),
63            });
64        }
65        if !domain_lo_le_hi(&d.lo, &d.hi) {
66            return Err(SqlError::Parse {
67                detail: format!("CREATE ARRAY {}: dim `{}` lo > hi", ast.name, d.name),
68            });
69        }
70    }
71    let mut seen_attrs: HashSet<&str> = HashSet::new();
72    for a in &ast.attrs {
73        if a.name.is_empty() {
74            return Err(SqlError::Parse {
75                detail: format!("CREATE ARRAY {}: empty attr name", ast.name),
76            });
77        }
78        if !seen_attrs.insert(a.name.as_str()) {
79            return Err(SqlError::Parse {
80                detail: format!(
81                    "CREATE ARRAY {}: duplicate attr name `{}`",
82                    ast.name, a.name
83                ),
84            });
85        }
86    }
87    for (i, te) in ast.tile_extents.iter().enumerate() {
88        if *te <= 0 {
89            return Err(SqlError::Parse {
90                detail: format!(
91                    "CREATE ARRAY {}: tile extent for dim `{}` must be > 0 (got {te})",
92                    ast.name, ast.dims[i].name
93                ),
94            });
95        }
96    }
97
98    if !(1..=16).contains(&ast.prefix_bits) {
99        return Err(SqlError::Parse {
100            detail: format!(
101                "CREATE ARRAY {}: prefix_bits {} is not in range 1–16",
102                ast.name, ast.prefix_bits
103            ),
104        });
105    }
106    Ok(SqlPlan::CreateArray(CreateArrayPlan {
107        name: ast.name.clone(),
108        dims: ast.dims.clone(),
109        attrs: ast.attrs.clone(),
110        tile_extents: ast.tile_extents.clone(),
111        cell_order: ast.cell_order,
112        tile_order: ast.tile_order,
113        prefix_bits: ast.prefix_bits,
114        audit_retain_ms: ast.audit_retain_ms,
115        minimum_audit_retain_ms: ast.minimum_audit_retain_ms,
116    }))
117}
118
119pub fn plan_alter_array(ast: &AlterArrayAst) -> Result<SqlPlan> {
120    if ast.name.is_empty() {
121        return Err(SqlError::Parse {
122            detail: "ALTER ARRAY: array name must not be empty".into(),
123        });
124    }
125    if ast.set.is_empty() {
126        return Err(SqlError::Parse {
127            detail: format!("ALTER ARRAY {}: SET clause is empty", ast.name),
128        });
129    }
130
131    let mut audit_retain_ms: Option<Option<i64>> = None;
132    let mut minimum_audit_retain_ms: Option<u64> = None;
133
134    for (key, value) in &ast.set {
135        match key.as_str() {
136            "audit_retain_ms" => {
137                audit_retain_ms = Some(*value);
138            }
139            "minimum_audit_retain_ms" => {
140                let n = value.ok_or_else(|| SqlError::Parse {
141                    detail: format!(
142                        "ALTER ARRAY {}: minimum_audit_retain_ms cannot be NULL",
143                        ast.name
144                    ),
145                })?;
146                minimum_audit_retain_ms = Some(n as u64);
147            }
148            other => {
149                return Err(SqlError::Parse {
150                    detail: format!(
151                        "ALTER ARRAY {}: unknown SET key `{other}`; \
152                         expected `audit_retain_ms` or `minimum_audit_retain_ms`",
153                        ast.name
154                    ),
155                });
156            }
157        }
158    }
159
160    Ok(SqlPlan::AlterArray(AlterArrayPlan {
161        name: ast.name.clone(),
162        audit_retain_ms,
163        minimum_audit_retain_ms,
164    }))
165}
166
167pub fn plan_drop_array(ast: &DropArrayAst) -> Result<SqlPlan> {
168    if ast.name.is_empty() {
169        return Err(SqlError::Parse {
170            detail: "DROP ARRAY: name must not be empty".into(),
171        });
172    }
173    Ok(SqlPlan::DropArray {
174        name: ast.name.clone(),
175        if_exists: ast.if_exists,
176    })
177}
178
179fn domain_bounds_match_type(
180    lo: &ArrayDomainBound,
181    hi: &ArrayDomainBound,
182    dtype: ArrayDimType,
183) -> bool {
184    matches!(
185        (lo, hi, dtype),
186        (
187            ArrayDomainBound::Int64(_),
188            ArrayDomainBound::Int64(_),
189            ArrayDimType::Int64
190        ) | (
191            ArrayDomainBound::Float64(_),
192            ArrayDomainBound::Float64(_),
193            ArrayDimType::Float64
194        ) | (
195            ArrayDomainBound::TimestampMs(_),
196            ArrayDomainBound::TimestampMs(_),
197            ArrayDimType::TimestampMs,
198        ) | (
199            ArrayDomainBound::String(_),
200            ArrayDomainBound::String(_),
201            ArrayDimType::String
202        )
203    )
204}
205
206fn domain_lo_le_hi(lo: &ArrayDomainBound, hi: &ArrayDomainBound) -> bool {
207    match (lo, hi) {
208        (ArrayDomainBound::Int64(a), ArrayDomainBound::Int64(b)) => a <= b,
209        (ArrayDomainBound::Float64(a), ArrayDomainBound::Float64(b)) => a <= b,
210        (ArrayDomainBound::TimestampMs(a), ArrayDomainBound::TimestampMs(b)) => a <= b,
211        (ArrayDomainBound::String(a), ArrayDomainBound::String(b)) => a <= b,
212        _ => false,
213    }
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219    use crate::types_array::{ArrayAttrAst, ArrayAttrType, ArrayDimAst};
220
221    fn ok_ast() -> CreateArrayAst {
222        CreateArrayAst {
223            name: "g".into(),
224            dims: vec![ArrayDimAst {
225                name: "x".into(),
226                dtype: ArrayDimType::Int64,
227                lo: ArrayDomainBound::Int64(0),
228                hi: ArrayDomainBound::Int64(10),
229            }],
230            attrs: vec![ArrayAttrAst {
231                name: "v".into(),
232                dtype: ArrayAttrType::Int64,
233                nullable: false,
234            }],
235            tile_extents: vec![4],
236            cell_order: Default::default(),
237            tile_order: Default::default(),
238            prefix_bits: 8,
239            audit_retain_ms: None,
240            minimum_audit_retain_ms: None,
241        }
242    }
243
244    #[test]
245    fn happy_path() {
246        let plan = plan_create_array(&ok_ast()).unwrap();
247        assert!(matches!(plan, SqlPlan::CreateArray { .. }));
248    }
249
250    #[test]
251    fn rejects_extent_arity() {
252        let mut a = ok_ast();
253        a.tile_extents = vec![1, 2];
254        assert!(plan_create_array(&a).is_err());
255    }
256
257    #[test]
258    fn rejects_dup_dim() {
259        let mut a = ok_ast();
260        a.dims.push(a.dims[0].clone());
261        a.tile_extents = vec![4, 4];
262        assert!(plan_create_array(&a).is_err());
263    }
264
265    #[test]
266    fn rejects_lo_gt_hi() {
267        let mut a = ok_ast();
268        a.dims[0].lo = ArrayDomainBound::Int64(5);
269        a.dims[0].hi = ArrayDomainBound::Int64(1);
270        assert!(plan_create_array(&a).is_err());
271    }
272}
nodedb_sql/planner/array_ddl.rs

nodedb_sql/planner/
array_ddl.rs
