nodedb_sql/planner/dml/
insert.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! INSERT planning.
4
5use sqlparser::ast;
6
7use super::super::dml_helpers::{
8    KvInsertParams, VectorPrimaryInsertParams, bind_insert_select_columns, build_kv_insert_plan,
9    build_vector_primary_insert_plan, insert_select_targets, is_vector_primary,
10    name_insert_select_outputs, refuse_vector_primary_shape, resolve_insert_columns,
11};
12use super::target::{
13    OnConflict, classify_on_conflict, column_schema, insert_columns, resolve_target, target_scope,
14    typed_rows, values_rows,
15};
16use super::upsert::plan_upsert_with_on_conflict;
17use crate::engine_rules::{self, InsertParams, InsertSelectParams};
18use crate::error::Result;
19use crate::types::*;
20
21/// Plan an INSERT statement.
22pub fn plan_insert(
23    ins: &ast::Insert,
24    catalog: &dyn SqlCatalog,
25    functions: &crate::functions::registry::FunctionRegistry,
26    temporal: crate::TemporalScope,
27) -> Result<Vec<SqlPlan>> {
28    let (table_name, info) = resolve_target(ins, "INSERT", catalog)?;
29    let target_scope = target_scope(&table_name, &info)?;
30
31    // `INSERT ... ON CONFLICT DO UPDATE SET` reroutes to the upsert path
32    // with the assignments carried through. `DO NOTHING` stays on the
33    // INSERT path with `if_absent=true`.
34    let if_absent = match classify_on_conflict(ins, &target_scope)? {
35        OnConflict::None => false,
36        OnConflict::DoNothing => true,
37        OnConflict::DoUpdate(updates) => {
38            return plan_upsert_with_on_conflict(ins, catalog, updates);
39        }
40    };
41
42    let columns = insert_columns(&ins.columns, &target_scope)?;
43
44    // Check for INSERT...SELECT.
45    if let Some(source) = &ins.source
46        && let ast::SetExpr::Select(select) = &*source.body
47    {
48        return plan_insert_select(
49            InsertSelectTarget {
50                table_name,
51                info: &info,
52                columns: &columns,
53            },
54            source,
55            select,
56            catalog,
57            functions,
58            temporal,
59        );
60    }
61
62    let rows_ast = values_rows(ins, "INSERT")?;
63
64    // KV engine: key and value are fundamentally separate — handle directly.
65    // Positional column binding (below) does not apply here: the KV path
66    // matches columns by name against `pk_col`/`"key"`/`"ttl"`, which is
67    // orthogonal to declared column order.
68    if info.engine == EngineType::KeyValue {
69        let intent = if if_absent {
70            KvInsertIntent::InsertIfAbsent
71        } else {
72            KvInsertIntent::Insert
73        };
74        return build_kv_insert_plan(KvInsertParams {
75            collection: table_name,
76            columns: &columns,
77            rows_ast,
78            intent,
79            on_conflict_updates: Vec::new(),
80            pk_col: info.primary_key.as_deref(),
81            declared_columns: &info.columns,
82            catalog,
83        });
84    }
85
86    // Positional INSERT (no column list): bind values to the collection's
87    // declared column order so named projections/predicates can find them.
88    // No-op for named inserts and schemaless collections.
89    let columns = resolve_insert_columns(columns, &info, rows_ast)?;
90
91    // One typing pass for every remaining engine: defaults materialized,
92    // then every cell coerced and range-checked against its declared type.
93    let typed = typed_rows(&info, &columns, rows_ast, catalog)?;
94
95    // Vector-primary collection: bypass document encoding. The row's
96    // existence intent travels with the plan, as it does for KV.
97    if is_vector_primary(&info)
98        && let Some(ref vpc) = info.vector_primary
99    {
100        let intent = if if_absent {
101            VectorPrimaryInsertIntent::InsertIfAbsent
102        } else {
103            VectorPrimaryInsertIntent::Insert
104        };
105        return build_vector_primary_insert_plan(VectorPrimaryInsertParams {
106            collection: &table_name,
107            vpc,
108            rows: typed.rows,
109            volatile_defaults: typed.volatile_defaults,
110            intent,
111            on_conflict_updates: Vec::new(),
112            primary_key: info.primary_key.clone(),
113        });
114    }
115
116    // All other engines: delegate to engine rules.
117    let column_schema = column_schema(&info);
118    let rules = engine_rules::resolve_engine_rules(info.engine);
119    rules.plan_insert(InsertParams {
120        collection: table_name,
121        columns,
122        rows: typed.rows,
123        volatile_defaults: typed.volatile_defaults,
124        if_absent,
125        column_schema,
126        primary_key: info.primary_key.clone(),
127    })
128}
129
130/// The target side of an `INSERT ... SELECT`.
131struct InsertSelectTarget<'a> {
132    table_name: String,
133    info: &'a CollectionInfo,
134    /// The explicit target column list, empty when the statement gives none.
135    columns: &'a [String],
136}
137
138/// Plan an `INSERT ... SELECT`.
139///
140/// The source SELECT plans as an ordinary query, with every output column
141/// named so the copy can pair it with its target column. The target engine
142/// decides through `EngineRules::plan_insert_select` whether it accepts the
143/// copy. A vector-primary target is refused: the copy writes document
144/// bodies, which its vector index never reads.
145fn plan_insert_select(
146    target: InsertSelectTarget<'_>,
147    source: &ast::Query,
148    select: &ast::Select,
149    catalog: &dyn SqlCatalog,
150    functions: &crate::functions::registry::FunctionRegistry,
151    temporal: crate::TemporalScope,
152) -> Result<Vec<SqlPlan>> {
153    refuse_vector_primary_shape(target.info, "INSERT ... SELECT")?;
154    let names = insert_select_targets(target.columns, select, target.info)?;
155    let (named_source, source_columns) = name_insert_select_outputs(source, &names)?;
156    let source_plan =
157        super::super::select::plan_query(&named_source, catalog, functions, temporal)?;
158    // A plain document scan is read as stored rows, so each target column
159    // is bound as an expression over the stored row as well.
160    let column_map = match &source_plan {
161        SqlPlan::Scan {
162            engine: EngineType::DocumentSchemaless | EngineType::DocumentStrict,
163            ..
164        } => bind_insert_select_columns(catalog, functions, temporal, &names, select)?,
165        _ => Vec::new(),
166    };
167    engine_rules::resolve_engine_rules(target.info.engine).plan_insert_select(InsertSelectParams {
168        collection: target.table_name,
169        source: Box::new(source_plan),
170        column_map,
171        source_columns,
172    })
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178    use crate::catalog::{SqlCatalog, SqlCatalogError};
179    use crate::parser::statement::parse_sql;
180    use nodedb_types::columnar::IntWidth;
181    use nodedb_types::datetime::NdbDateTime;
182
183    fn column(name: &str, data_type: SqlDataType, default: Option<&str>) -> ColumnInfo {
184        ColumnInfo {
185            name: name.to_string(),
186            data_type,
187            nullable: true,
188            is_primary_key: false,
189            default: default.map(str::to_string),
190            raw_type: None,
191            int_width: None,
192            float_width: None,
193        }
194    }
195
196    fn collection(engine: EngineType, columns: Vec<ColumnInfo>) -> CollectionInfo {
197        let primary_key = columns
198            .iter()
199            .find(|c| c.is_primary_key)
200            .map(|c| c.name.clone());
201        CollectionInfo {
202            name: "t".into(),
203            engine,
204            columns,
205            primary_key,
206            has_auto_tier: false,
207            indexes: Vec::new(),
208            bitemporal: false,
209            primary: nodedb_types::PrimaryEngine::Document,
210            vector_primary: None,
211            partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
212            open_schema: CollectionInfo::open_schema_for(engine),
213        }
214    }
215
216    /// A catalog with one collection named `t`.
217    struct OneCollection(CollectionInfo);
218
219    impl SqlCatalog for OneCollection {
220        fn get_collection(
221            &self,
222            _: nodedb_types::DatabaseId,
223            name: &str,
224        ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> {
225            Ok((name == "t").then(|| self.0.clone()))
226        }
227    }
228
229    fn plan(sql: &str, catalog: &dyn SqlCatalog) -> Result<SqlPlan> {
230        let statements = parse_sql(sql)?;
231        let sqlparser::ast::Statement::Insert(ins) = &statements[0] else {
232            panic!("expected an INSERT statement");
233        };
234        let mut plans = plan_insert(
235            ins,
236            catalog,
237            &crate::functions::registry::FunctionRegistry::new(),
238            crate::TemporalScope::default(),
239        )?;
240        Ok(plans.remove(0))
241    }
242
243    /// `2020-03-05T10:00:00Z` as microseconds since the Unix epoch.
244    const EARLY_MICROS: i64 = 1_583_402_400_000_000;
245
246    fn cell<'a>(row: &'a [(String, SqlValue)], name: &str) -> &'a SqlValue {
247        &row.iter()
248            .find(|(n, _)| n == name)
249            .unwrap_or_else(|| panic!("row carries {name}"))
250            .1
251    }
252
253    /// A numeric DEFAULT on a TIMESTAMP column is materialized, then coerced
254    /// like a supplied literal: the plan carries the typed instant.
255    #[test]
256    fn a_numeric_timestamp_default_is_materialized_then_coerced() {
257        let mut id = column("id", SqlDataType::String, None);
258        id.is_primary_key = true;
259        let catalog = OneCollection(collection(
260            EngineType::DocumentSchemaless,
261            vec![
262                id,
263                column("at", SqlDataType::Timestamp, Some("1583402400000")),
264            ],
265        ));
266        let plan = plan("INSERT INTO t (id) VALUES ('r1')", &catalog).expect("plans");
267        let SqlPlan::Insert(InsertPlan {
268            rows,
269            volatile_defaults,
270            ..
271        }) = plan
272        else {
273            panic!("expected SqlPlan::Insert, got {plan:?}");
274        };
275        assert!(!volatile_defaults, "a literal DEFAULT is not volatile");
276        assert_eq!(
277            cell(&rows[0], "at"),
278            &SqlValue::Timestamp(NdbDateTime::from_micros(EARLY_MICROS))
279        );
280    }
281
282    /// A DEFAULT the column cannot hold is refused at the insert, naming the
283    /// column, exactly as the same literal in VALUES is.
284    #[test]
285    fn a_default_out_of_declared_range_is_refused() {
286        let mut id = column("id", SqlDataType::String, None);
287        id.is_primary_key = true;
288        let mut small = column("s", SqlDataType::Int64, Some("999999"));
289        small.int_width = Some(IntWidth::I16);
290        let catalog = OneCollection(collection(EngineType::DocumentStrict, vec![id, small]));
291        let err = plan("INSERT INTO t (id) VALUES ('r1')", &catalog)
292            .expect_err("DEFAULT 999999 does not fit SMALLINT");
293        assert!(
294            matches!(err, crate::SqlError::IntegerOutOfRange { ref column, .. } if column == "s"),
295            "{err}"
296        );
297    }
298
299    /// A timeseries plan carries the declared default of an omitted column
300    /// in its rows, the TIME_KEY column included.
301    #[test]
302    fn a_timeseries_plan_carries_defaults_in_its_rows() {
303        let catalog = OneCollection(collection(
304            EngineType::Timeseries,
305            vec![
306                column("at", SqlDataType::Timestamp, Some("1583402400000")),
307                column("host", SqlDataType::String, Some("'h0'")),
308                column("v", SqlDataType::Float64, None),
309            ],
310        ));
311        let plan = plan("INSERT INTO t (v) VALUES (1.0)", &catalog).expect("plans");
312        let SqlPlan::TimeseriesIngest(TimeseriesIngestPlan {
313            rows,
314            volatile_defaults,
315            ..
316        }) = plan
317        else {
318            panic!("expected SqlPlan::TimeseriesIngest, got {plan:?}");
319        };
320        assert!(!volatile_defaults);
321        assert_eq!(
322            cell(&rows[0], "at"),
323            &SqlValue::Timestamp(NdbDateTime::from_micros(EARLY_MICROS))
324        );
325        assert_eq!(cell(&rows[0], "host"), &SqlValue::String("h0".into()));
326        assert_eq!(cell(&rows[0], "v"), &SqlValue::Float(1.0));
327    }
328
329    /// A catalog with a target `t` and a source `s`.
330    struct TwoCollections {
331        target: CollectionInfo,
332        source: CollectionInfo,
333    }
334
335    impl SqlCatalog for TwoCollections {
336        fn get_collection(
337            &self,
338            _: nodedb_types::DatabaseId,
339            name: &str,
340        ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> {
341            Ok(match name {
342                "t" => Some(self.target.clone()),
343                "s" => Some(self.source.clone()),
344                _ => None,
345            })
346        }
347    }
348
349    fn copy_catalog(target: EngineType, source: EngineType) -> TwoCollections {
350        let mut id = column("id", SqlDataType::String, None);
351        id.is_primary_key = true;
352        let columns = vec![id, column("v", SqlDataType::String, None)];
353        let mut source = collection(source, columns.clone());
354        source.name = "s".into();
355        TwoCollections {
356            target: collection(target, columns),
357            source,
358        }
359    }
360
361    fn source_columns(plan: &SqlPlan) -> &[(String, String)] {
362        let SqlPlan::InsertSelect { source_columns, .. } = plan else {
363            panic!("expected SqlPlan::InsertSelect, got {plan:?}");
364        };
365        source_columns
366    }
367
368    /// Every unaliased SELECT-list item gets an output name the copy reads
369    /// back, paired with its target column. A document scan source also
370    /// binds each target column over the stored row.
371    #[test]
372    fn insert_select_names_every_output_column() {
373        let catalog = copy_catalog(
374            EngineType::DocumentSchemaless,
375            EngineType::DocumentSchemaless,
376        );
377        let plan = plan("INSERT INTO t (id, v) SELECT id, v FROM s", &catalog).expect("plans");
378        assert_eq!(
379            source_columns(&plan),
380            [
381                ("id".to_string(), "__nodedb_insert_select_0".to_string()),
382                ("v".to_string(), "__nodedb_insert_select_1".to_string()),
383            ]
384        );
385        let SqlPlan::InsertSelect { column_map, .. } = &plan else {
386            unreachable!()
387        };
388        assert_eq!(column_map.len(), 2);
389    }
390
391    /// An aliased item keeps its alias, so an ORDER BY over the alias keeps
392    /// its meaning.
393    #[test]
394    fn insert_select_keeps_a_user_alias() {
395        let catalog = copy_catalog(
396            EngineType::DocumentSchemaless,
397            EngineType::DocumentSchemaless,
398        );
399        let plan = plan(
400            "INSERT INTO t (id, v) SELECT id, v AS score FROM s ORDER BY score",
401            &catalog,
402        )
403        .expect("plans");
404        assert_eq!(
405            source_columns(&plan)[1],
406            ("v".to_string(), "score".to_string())
407        );
408    }
409
410    #[test]
411    fn insert_select_with_a_repeated_alias_is_refused() {
412        let catalog = copy_catalog(
413            EngineType::DocumentSchemaless,
414            EngineType::DocumentSchemaless,
415        );
416        let err = plan(
417            "INSERT INTO t (id, v) SELECT id AS x, v AS x FROM s",
418            &catalog,
419        )
420        .expect_err("two outputs named x");
421        assert!(matches!(err, crate::SqlError::Unsupported { .. }), "{err}");
422    }
423
424    /// A key-value source plans as a query: no stored-row binding, only the
425    /// output pairing.
426    #[test]
427    fn insert_select_from_a_kv_source_pairs_outputs_only() {
428        let catalog = copy_catalog(EngineType::DocumentSchemaless, EngineType::KeyValue);
429        let plan = plan("INSERT INTO t (id, v) SELECT id, v FROM s", &catalog).expect("plans");
430        let SqlPlan::InsertSelect {
431            column_map,
432            source_columns,
433            ..
434        } = &plan
435        else {
436            panic!("expected SqlPlan::InsertSelect, got {plan:?}");
437        };
438        assert!(column_map.is_empty());
439        assert_eq!(source_columns.len(), 2);
440    }
441
442    /// The target engine refuses at plan time, before any row is read.
443    #[test]
444    fn insert_select_into_a_non_document_target_is_refused() {
445        for engine in [
446            EngineType::KeyValue,
447            EngineType::Columnar,
448            EngineType::Spatial,
449            EngineType::Timeseries,
450        ] {
451            let catalog = copy_catalog(engine, EngineType::DocumentSchemaless);
452            let err = plan("INSERT INTO t (id, v) SELECT id, v FROM s", &catalog)
453                .expect_err("a non-document target refuses the copy");
454            assert!(
455                matches!(err, crate::SqlError::Unsupported { .. }),
456                "{engine:?}: {err}"
457            );
458        }
459    }
460
461    /// A vector-primary target is refused: the copy writes document bodies
462    /// its vector index never reads.
463    #[test]
464    fn insert_select_into_a_vector_primary_target_is_refused() {
465        let mut catalog = copy_catalog(
466            EngineType::DocumentSchemaless,
467            EngineType::DocumentSchemaless,
468        );
469        catalog.target.primary = nodedb_types::PrimaryEngine::Vector;
470        catalog.target.vector_primary = Some(nodedb_types::VectorPrimaryConfig::default());
471        let err = plan("INSERT INTO t (id, v) SELECT id, v FROM s", &catalog)
472            .expect_err("a vector-primary target refuses the copy");
473        assert!(
474            matches!(err, crate::SqlError::Unsupported { ref detail } if detail.contains("vector-primary")),
475            "{err}"
476        );
477    }
478
479    /// A volatile DEFAULT marks the plan so the plan cache never replays it.
480    #[test]
481    fn a_volatile_default_marks_the_plan() {
482        let mut id = column("id", SqlDataType::String, Some("UUID_V7"));
483        id.is_primary_key = true;
484        let catalog = OneCollection(collection(
485            EngineType::DocumentStrict,
486            vec![id, column("n", SqlDataType::Int64, None)],
487        ));
488        let plan = plan("INSERT INTO t (n) VALUES (1)", &catalog).expect("plans");
489        assert!(
490            !plan.cache_eligibility().is_cacheable(),
491            "a plan carrying a volatile DEFAULT is never cached"
492        );
493    }
494}
nodedb_sql/planner/dml/insert.rs

nodedb_sql/planner/dml/
insert.rs
