nodedb_sql/planner/dml/
upsert.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! UPSERT and `INSERT ... ON CONFLICT DO UPDATE` planning.
4
5use sqlparser::ast;
6
7use super::super::dml_helpers::{
8    KvInsertParams, VectorPrimaryInsertParams, build_kv_insert_plan,
9    build_vector_primary_insert_plan, check_declared_float_ranges_in_assignments,
10    check_declared_int_ranges_in_assignments, is_vector_primary, resolve_insert_columns,
11};
12use super::target::{
13    column_schema, insert_columns, resolve_target, target_scope, typed_rows, values_rows,
14};
15use crate::engine_rules::{self, UpsertParams};
16use crate::error::Result;
17use crate::planner::declared_type_coerce::coerce_assignments_to_declared_types;
18use crate::types::*;
19
20/// Plan an UPSERT statement (pre-processed from `UPSERT INTO` to `INSERT INTO`).
21///
22/// Same parsing as INSERT but routes through `engine_rules.plan_upsert()`.
23pub fn plan_upsert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
24    plan_upsert_rows(ins, catalog, "UPSERT", Vec::new())
25}
26
27/// Plan an `INSERT ... ON CONFLICT DO UPDATE SET` statement.
28pub(super) fn plan_upsert_with_on_conflict(
29    ins: &ast::Insert,
30    catalog: &dyn SqlCatalog,
31    on_conflict_updates: Vec<(String, SqlExpr)>,
32) -> Result<Vec<SqlPlan>> {
33    plan_upsert_rows(ins, catalog, "INSERT ... ON CONFLICT", on_conflict_updates)
34}
35
36/// The shared body of `UPSERT` and `INSERT ... ON CONFLICT DO UPDATE`.
37///
38/// The two differ only in `verb` (for error messages) and in whether
39/// `on_conflict_updates` carries per-row assignments for the Data Plane to
40/// apply against the existing row.
41fn plan_upsert_rows(
42    ins: &ast::Insert,
43    catalog: &dyn SqlCatalog,
44    verb: &str,
45    mut on_conflict_updates: Vec<(String, SqlExpr)>,
46) -> Result<Vec<SqlPlan>> {
47    let (table_name, info) = resolve_target(ins, verb, catalog)?;
48    let columns = insert_columns(&ins.columns, &target_scope(&table_name, &info)?)?;
49    let rows_ast = values_rows(ins, verb)?;
50
51    // KV: upsert is a PUT (natural overwrite), and `INSERT ... ON CONFLICT
52    // (key) DO UPDATE SET ...` is the same physical write with the optional
53    // per-row assignments carried through. Positional column binding
54    // (below) does not apply here — see `plan_insert`.
55    if info.engine == EngineType::KeyValue {
56        return build_kv_insert_plan(KvInsertParams {
57            collection: table_name,
58            columns: &columns,
59            rows_ast,
60            intent: KvInsertIntent::Put,
61            on_conflict_updates,
62            pk_col: info.primary_key.as_deref(),
63            declared_columns: &info.columns,
64            catalog,
65        });
66    }
67
68    // Positional UPSERT (no column list): bind to the collection's declared
69    // column order — see `plan_insert` for the full rationale.
70    let columns = resolve_insert_columns(columns, &info, rows_ast)?;
71
72    let typed = typed_rows(&info, &columns, rows_ast, catalog)?;
73    // `DO UPDATE SET col = <literal>` writes through the same path as the
74    // inserted row, so its literals carry the same declared-type contract.
75    coerce_assignments_to_declared_types(
76        &info.columns,
77        &mut on_conflict_updates,
78        info.primary_key.as_deref(),
79    )?;
80    check_declared_int_ranges_in_assignments(&info.columns, &on_conflict_updates)?;
81    check_declared_float_ranges_in_assignments(&info.columns, &on_conflict_updates)?;
82
83    // Vector-primary collection: an existing key is replaced, or patched by
84    // the carried assignments. The document upsert path never reaches the
85    // HNSW index, so it cannot serve this collection.
86    if is_vector_primary(&info)
87        && let Some(ref vpc) = info.vector_primary
88    {
89        return build_vector_primary_insert_plan(VectorPrimaryInsertParams {
90            collection: &table_name,
91            vpc,
92            rows: typed.rows,
93            volatile_defaults: typed.volatile_defaults,
94            intent: VectorPrimaryInsertIntent::Upsert,
95            on_conflict_updates,
96            primary_key: info.primary_key.clone(),
97        });
98    }
99
100    let column_schema = column_schema(&info);
101    let rules = engine_rules::resolve_engine_rules(info.engine);
102    rules.plan_upsert(UpsertParams {
103        collection: table_name,
104        columns,
105        rows: typed.rows,
106        volatile_defaults: typed.volatile_defaults,
107        on_conflict_updates,
108        column_schema,
109        primary_key: info.primary_key.clone(),
110    })
111}
nodedb_sql/planner/dml/upsert.rs

nodedb_sql/planner/dml/
upsert.rs
