nodedb_sql/planner/
array_dml.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Planner for `INSERT INTO ARRAY` and `DELETE FROM ARRAY`.
4//!
5//! Validation against the catalog: array exists, coord arity matches
6//! dim count, attr arity matches attr count, type tags coerce. Type
7//! coercion is purposely loose at the SQL level — `Int → Float` is
8//! accepted, the converter performs the actual cast on the way to the
9//! engine's typed `CoordValue` / `CellValue`.
10
11use crate::catalog::{ArrayCatalogView, SqlCatalog};
12use crate::error::{Result, SqlError};
13use crate::parser::array_stmt::{DeleteArrayAst, InsertArrayAst};
14use crate::types::SqlPlan;
15use crate::types::{DeleteArrayPlan, InsertArrayPlan};
16use crate::types_array::{ArrayAttrLiteral, ArrayAttrType, ArrayCoordLiteral, ArrayDimType};
17
18pub fn plan_insert_array(ast: &InsertArrayAst, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
19    let view = catalog
20        .lookup_array(&ast.name)
21        .ok_or_else(|| SqlError::Parse {
22            detail: format!("INSERT INTO ARRAY {}: array not found", ast.name),
23        })?;
24    if ast.rows.is_empty() {
25        return Err(SqlError::Parse {
26            detail: format!("INSERT INTO ARRAY {}: at least one row required", ast.name),
27        });
28    }
29    for (ri, row) in ast.rows.iter().enumerate() {
30        validate_coords(&ast.name, ri, &row.coords, &view)?;
31        validate_attrs(&ast.name, ri, &row.attrs, &view)?;
32    }
33    Ok(vec![SqlPlan::InsertArray(InsertArrayPlan {
34        name: ast.name.clone(),
35        rows: ast.rows.clone(),
36    })])
37}
38
39pub fn plan_delete_array(ast: &DeleteArrayAst, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
40    let view = catalog
41        .lookup_array(&ast.name)
42        .ok_or_else(|| SqlError::Parse {
43            detail: format!("DELETE FROM ARRAY {}: array not found", ast.name),
44        })?;
45    if ast.coords.is_empty() {
46        return Err(SqlError::Parse {
47            detail: format!(
48                "DELETE FROM ARRAY {}: at least one coord tuple required",
49                ast.name
50            ),
51        });
52    }
53    for (ri, row) in ast.coords.iter().enumerate() {
54        validate_coords(&ast.name, ri, row, &view)?;
55    }
56    Ok(vec![SqlPlan::DeleteArray(DeleteArrayPlan {
57        name: ast.name.clone(),
58        coords: ast.coords.clone(),
59    })])
60}
61
62fn validate_coords(
63    array: &str,
64    row: usize,
65    coords: &[ArrayCoordLiteral],
66    view: &ArrayCatalogView,
67) -> Result<()> {
68    if coords.len() != view.dims.len() {
69        return Err(SqlError::Parse {
70            detail: format!(
71                "ARRAY {array} row {row}: coord arity {} != dim count {}",
72                coords.len(),
73                view.dims.len()
74            ),
75        });
76    }
77    for (i, c) in coords.iter().enumerate() {
78        if !coord_compatible(c, view.dims[i].dtype) {
79            return Err(SqlError::TypeMismatch {
80                detail: format!(
81                    "ARRAY {array} row {row}: coord for dim `{}` (declared {:?}) is incompatible",
82                    view.dims[i].name, view.dims[i].dtype
83                ),
84            });
85        }
86    }
87    Ok(())
88}
89
90fn validate_attrs(
91    array: &str,
92    row: usize,
93    attrs: &[ArrayAttrLiteral],
94    view: &ArrayCatalogView,
95) -> Result<()> {
96    if attrs.len() != view.attrs.len() {
97        return Err(SqlError::Parse {
98            detail: format!(
99                "ARRAY {array} row {row}: attr arity {} != attr count {}",
100                attrs.len(),
101                view.attrs.len()
102            ),
103        });
104    }
105    for (i, a) in attrs.iter().enumerate() {
106        let spec = &view.attrs[i];
107        match a {
108            ArrayAttrLiteral::Null if !spec.nullable => {
109                return Err(SqlError::TypeMismatch {
110                    detail: format!("ARRAY {array} row {row}: attr `{}` is NOT NULL", spec.name),
111                });
112            }
113            ArrayAttrLiteral::Null => {}
114            other if !attr_compatible(other, spec.dtype) => {
115                return Err(SqlError::TypeMismatch {
116                    detail: format!(
117                        "ARRAY {array} row {row}: attr `{}` (declared {:?}) is incompatible",
118                        spec.name, spec.dtype
119                    ),
120                });
121            }
122            _ => {}
123        }
124    }
125    Ok(())
126}
127
128fn coord_compatible(c: &ArrayCoordLiteral, dtype: ArrayDimType) -> bool {
129    matches!(
130        (c, dtype),
131        (ArrayCoordLiteral::Int64(_), ArrayDimType::Int64)
132            | (ArrayCoordLiteral::Int64(_), ArrayDimType::TimestampMs)
133            | (ArrayCoordLiteral::Int64(_), ArrayDimType::Float64)
134            | (ArrayCoordLiteral::Float64(_), ArrayDimType::Float64)
135            | (ArrayCoordLiteral::String(_), ArrayDimType::String)
136    )
137}
138
139fn attr_compatible(a: &ArrayAttrLiteral, dtype: ArrayAttrType) -> bool {
140    matches!(
141        (a, dtype),
142        (ArrayAttrLiteral::Int64(_), ArrayAttrType::Int64)
143            | (ArrayAttrLiteral::Int64(_), ArrayAttrType::Float64)
144            | (ArrayAttrLiteral::Float64(_), ArrayAttrType::Float64)
145            | (ArrayAttrLiteral::String(_), ArrayAttrType::String)
146            | (ArrayAttrLiteral::Bytes(_), ArrayAttrType::Bytes)
147    )
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use crate::catalog::SqlCatalogError;
154    use crate::types::CollectionInfo;
155    use crate::types_array::{ArrayAttrAst, ArrayDimAst};
156
157    struct StubCatalog {
158        view: Option<ArrayCatalogView>,
159    }
160    impl SqlCatalog for StubCatalog {
161        fn get_collection(
162            &self,
163            _: nodedb_types::DatabaseId,
164            _: &str,
165        ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> {
166            Ok(None)
167        }
168        fn lookup_array(&self, _: &str) -> Option<ArrayCatalogView> {
169            self.view.clone()
170        }
171    }
172
173    fn view_2d() -> ArrayCatalogView {
174        ArrayCatalogView {
175            name: "g".into(),
176            dims: vec![
177                ArrayDimAst {
178                    name: "chrom".into(),
179                    dtype: ArrayDimType::Int64,
180                    lo: crate::types_array::ArrayDomainBound::Int64(1),
181                    hi: crate::types_array::ArrayDomainBound::Int64(23),
182                },
183                ArrayDimAst {
184                    name: "pos".into(),
185                    dtype: ArrayDimType::Int64,
186                    lo: crate::types_array::ArrayDomainBound::Int64(0),
187                    hi: crate::types_array::ArrayDomainBound::Int64(10_000_000),
188                },
189            ],
190            attrs: vec![ArrayAttrAst {
191                name: "v".into(),
192                dtype: ArrayAttrType::Float64,
193                nullable: true,
194            }],
195            tile_extents: vec![1, 1_000_000],
196        }
197    }
198
199    #[test]
200    fn insert_unknown_array() {
201        let cat = StubCatalog { view: None };
202        let ast = InsertArrayAst {
203            name: "g".into(),
204            rows: vec![],
205        };
206        assert!(plan_insert_array(&ast, &cat).is_err());
207    }
208
209    #[test]
210    fn insert_arity_mismatch_rejected() {
211        let cat = StubCatalog {
212            view: Some(view_2d()),
213        };
214        let ast = InsertArrayAst {
215            name: "g".into(),
216            rows: vec![crate::types_array::ArrayInsertRow {
217                coords: vec![ArrayCoordLiteral::Int64(1)],
218                attrs: vec![ArrayAttrLiteral::Float64(1.0)],
219            }],
220        };
221        assert!(plan_insert_array(&ast, &cat).is_err());
222    }
223
224    #[test]
225    fn insert_happy() {
226        let cat = StubCatalog {
227            view: Some(view_2d()),
228        };
229        let ast = InsertArrayAst {
230            name: "g".into(),
231            rows: vec![crate::types_array::ArrayInsertRow {
232                coords: vec![ArrayCoordLiteral::Int64(1), ArrayCoordLiteral::Int64(100)],
233                attrs: vec![ArrayAttrLiteral::Float64(99.5)],
234            }],
235        };
236        let plans = plan_insert_array(&ast, &cat).unwrap();
237        assert_eq!(plans.len(), 1);
238        assert!(matches!(plans[0], SqlPlan::InsertArray { .. }));
239    }
240
241    #[test]
242    fn delete_happy() {
243        let cat = StubCatalog {
244            view: Some(view_2d()),
245        };
246        let ast = DeleteArrayAst {
247            name: "g".into(),
248            coords: vec![vec![
249                ArrayCoordLiteral::Int64(1),
250                ArrayCoordLiteral::Int64(100),
251            ]],
252        };
253        let plans = plan_delete_array(&ast, &cat).unwrap();
254        assert_eq!(plans.len(), 1);
255        assert!(matches!(plans[0], SqlPlan::DeleteArray { .. }));
256    }
257}
nodedb_sql/planner/array_dml.rs

nodedb_sql/planner/
array_dml.rs
