refactor(sql): split dml_helpers into a directory module

Convert the single dml_helpers.rs file into a directory of
single-concern modules (value_convert, range_check, insert_columns,
ast_extract, vector_primary_insert, kv_insert) to keep each file under
the per-concern size limit as the declared-width range checks grow.
Purely mechanical: no behavior changes.

Author: farhan-syah

nodedb-sql/src/planner/dml_helpers.rs

@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! AST extraction helpers shared by the DML planners: pulling a bare table
+//! name out of a join tree, and pulling primary-key point-lookup keys out of
+//! a `WHERE` clause.
+
+use sqlparser::ast;
+
+use super::value_convert::expr_to_sql_value;
+use crate::error::{Result, SqlError};
+use crate::parser::normalize::{normalize_ident, normalize_object_name_checked};
+use crate::types::*;
+
+pub(crate) fn extract_table_name_from_table_with_joins(
+    table: &ast::TableWithJoins,
+) -> Result<String> {
+    match &table.relation {
+        ast::TableFactor::Table { name, .. } => Ok(normalize_object_name_checked(name)?),
+        _ => Err(SqlError::Unsupported {
+            detail: "non-table target in DML".into(),
+        }),
+    }
+}
+
+/// Extract point-operation keys from WHERE clause (WHERE pk = literal OR pk IN (...)).
+pub fn extract_point_keys(selection: Option<&ast::Expr>, info: &CollectionInfo) -> Vec<SqlValue> {
+    let pk = match &info.primary_key {
+        Some(pk) => pk.clone(),
+        None => return Vec::new(),
+    };
+
+    let expr = match selection {
+        Some(e) => e,
+        None => return Vec::new(),
+    };
+
+    let mut keys = Vec::new();
+    collect_pk_equalities(expr, &pk, &mut keys);
+    keys
+}
+
+fn collect_pk_equalities(expr: &ast::Expr, pk: &str, keys: &mut Vec<SqlValue>) {
+    match expr {
+        ast::Expr::BinaryOp {
+            left,
+            op: ast::BinaryOperator::Eq,
+            right,
+        } => {
+            if is_column(left, pk)
+                && let Ok(v) = expr_to_sql_value(right)
+            {
+                keys.push(v);
+            } else if is_column(right, pk)
+                && let Ok(v) = expr_to_sql_value(left)
+            {
+                keys.push(v);
+            }
+        }
+        ast::Expr::BinaryOp {
+            left,
+            op: ast::BinaryOperator::Or,
+            right,
+        } => {
+            collect_pk_equalities(left, pk, keys);
+            collect_pk_equalities(right, pk, keys);
+        }
+        ast::Expr::InList {
+            expr: inner,
+            list,
+            negated: false,
+        } if is_column(inner, pk) => {
+            for item in list {
+                if let Ok(v) = expr_to_sql_value(item) {
+                    keys.push(v);
+                }
+            }
+        }
+        _ => {}
+    }
+}
+
+fn is_column(expr: &ast::Expr, name: &str) -> bool {
+    match expr {
+        ast::Expr::Identifier(ident) => normalize_ident(ident) == name,
+        // Three or more parts: schema.table.col — never matches a plain pk name.
+        ast::Expr::CompoundIdentifier(parts) if parts.len() >= 3 => false,
+        ast::Expr::CompoundIdentifier(parts) if parts.len() == 2 => {
+            normalize_ident(&parts[1]) == name
+        }
+        _ => false,
+    }
+}

nodedb-sql/src/planner/dml_helpers/ast_extract.rs

@@ -0,0 +1,48 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Effective-column resolution for positional `VALUES`-clause inserts.
+
+use sqlparser::ast;
+
+use crate::error::{Result, SqlError};
+use crate::types::*;
+
+/// Resolve the effective column list for a `VALUES`-clause INSERT/UPSERT.
+///
+/// A *positional* insert — `INSERT INTO t VALUES (...)` with no explicit
+/// column list — must still bind each value to the collection's declared
+/// column names. Left alone, `convert_value_rows` falls back to synthetic
+/// `col0`, `col1`, ... names for every value (see its `col{i}` fallback
+/// below): the row stores fine, but named projections and WHERE predicates
+/// can never find it again (#202).
+///
+/// Named inserts (`columns` already non-empty) and schemaless collections
+/// (`info.columns` empty — there is no declared order to bind to) pass
+/// through unchanged; the `col{i}` fallback remains the last resort for
+/// those.
+///
+/// Fewer values than declared columns binds by position against the
+/// leading columns, consistent with a partial named insert (e.g.
+/// `INSERT INTO t (id) VALUES (1)` on a wider table also only binds
+/// `id`). More values than declared columns is rejected outright:
+/// inventing a `colN` slot for the overflow would reproduce the exact
+/// unaddressable-column failure this fix closes.
+pub(crate) fn resolve_insert_columns(
+    columns: Vec<String>,
+    info: &CollectionInfo,
+    rows: &[Vec<ast::Expr>],
+) -> Result<Vec<String>> {
+    if !columns.is_empty() || info.columns.is_empty() {
+        return Ok(columns);
+    }
+
+    let declared: Vec<String> = info.columns.iter().map(|c| c.name.clone()).collect();
+    if let Some(row) = rows.iter().find(|row| row.len() > declared.len()) {
+        return Err(SqlError::InsertColumnArityMismatch {
+            collection: info.name.clone(),
+            given: row.len(),
+            declared: declared.len(),
+        });
+    }
+    Ok(declared)
+}

nodedb-sql/src/planner/dml_helpers/insert_columns.rs

@@ -0,0 +1,132 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Plan construction for the KV engine's `VALUES`-clause insert paths
+//! (plain `INSERT`, `UPSERT`, and `INSERT ... ON CONFLICT DO UPDATE`).
+
+use sqlparser::ast;
+
+use super::range_check::check_declared_int_ranges;
+use super::value_convert::expr_to_sql_value;
+use crate::error::{Result, SqlError};
+use crate::planner::declared_type_coerce::{
+    coerce_assignments_to_declared_types, coerce_row_to_declared_types,
+};
+use crate::types::*;
+
+/// Build a `SqlPlan::KvInsert` from a VALUES clause. Shared by plain INSERT,
+/// UPSERT, and `INSERT ... ON CONFLICT (key) DO UPDATE` — the three paths
+/// differ only in `intent` and `on_conflict_updates`, never in how entries
+/// are extracted from the row exprs.
+///
+/// `pk_col` is the schema-defined primary-key column name from
+/// `CollectionInfo::primary_key`.  When supplied, that column is used as
+/// the KV key regardless of whether it is named `"key"`.  Falls back to
+/// the literal name `"key"` when `pk_col` is `None` (legacy / generic
+/// KV collections that use the built-in key/value column convention).
+pub(crate) fn build_kv_insert_plan(
+    table_name: String,
+    columns: &[String],
+    rows_ast: &[Vec<ast::Expr>],
+    intent: KvInsertIntent,
+    mut on_conflict_updates: Vec<(String, SqlExpr)>,
+    pk_col: Option<&str>,
+    declared_columns: &[ColumnInfo],
+) -> Result<Vec<SqlPlan>> {
+    // Positional KV insert (no column list): the key/value split below is
+    // driven entirely by matching column *names* against `key_col_name`/
+    // `"ttl"`. With an empty `columns` list there is no key to bind to, so
+    // every row would silently become an empty-keyed, empty-valued entry
+    // (all colliding). Reject rather than corrupt.
+    if columns.is_empty() {
+        return Err(SqlError::PositionalKvInsertUnsupported {
+            collection: table_name,
+        });
+    }
+    let key_col_name = pk_col.unwrap_or("key");
+    let key_idx = columns.iter().position(|c| c == key_col_name);
+    let ttl_idx = columns.iter().position(|c| c == "ttl");
+    // When using a named primary-key column (e.g. `k STRING PRIMARY KEY`), we
+    // store the key bytes in the KV key slot AND also keep the column in the
+    // value map.  This allows scan filters on the primary-key column (e.g.
+    // `WHERE k = 'x'`) and projection (e.g. `SELECT k FROM ...`) to work
+    // without teaching the KV scan handler to inspect the raw key bytes.
+    // The only column we exclude from the value map is the built-in `"key"`
+    // sentinel (used by raw key/value KV collections) and `"ttl"`.
+    let exclude_from_value: std::collections::HashSet<usize> = {
+        let mut s = std::collections::HashSet::new();
+        // Exclude the raw "key" sentinel column (not a named PK column).
+        if key_col_name == "key"
+            && let Some(idx) = key_idx
+        {
+            s.insert(idx);
+        }
+        if let Some(idx) = ttl_idx {
+            s.insert(idx);
+        }
+        s
+    };
+    // Resolve every row's literals once, then coerce each cell to its declared
+    // column type. Unlike the strict, columnar, and timeseries engines, KV has
+    // no typed write path: its engine stores the bytes it is handed and the
+    // declared schema exists only here, in the catalog. Without this the
+    // stored cell's type is whatever the literal happened to resolve to — a
+    // fractional literal is an exact `Decimal`, which serializes as a msgpack
+    // string — while `RowDescription` advertises the declared numeric type, and
+    // the read path can only encode SQL NULL for it. See
+    // `declared_type_coerce` for the full rationale.
+    let mut coerced_rows: Vec<Vec<(String, SqlValue)>> = Vec::with_capacity(rows_ast.len());
+    for row_exprs in rows_ast {
+        let mut row: Vec<(String, SqlValue)> = Vec::with_capacity(columns.len());
+        for (i, col) in columns.iter().enumerate() {
+            let Some(expr) = row_exprs.get(i) else { break };
+            row.push((col.clone(), expr_to_sql_value(expr)?));
+        }
+        // The key column is exempt — see `coerce_rows_to_declared_types`.
+        coerce_row_to_declared_types(declared_columns, &mut row, Some(key_col_name))?;
+        coerced_rows.push(row);
+    }
+    // KV returns early from every INSERT/UPSERT entry point, so the declared
+    // width check lives here rather than at the call sites — otherwise a
+    // fourth KV entry point could be added without it. It runs on the coerced
+    // values so a literal that only becomes an integer through coercion is
+    // still range-checked against its declared width.
+    check_declared_int_ranges(declared_columns, &coerced_rows)?;
+    // `ON CONFLICT DO UPDATE SET col = <literal>` writes through the same
+    // untyped KV path as the inserted row, so its literals need the same
+    // declared-type coercion.
+    coerce_assignments_to_declared_types(
+        declared_columns,
+        &mut on_conflict_updates,
+        Some(key_col_name),
+    )?;
+
+    let mut entries = Vec::with_capacity(coerced_rows.len());
+    let mut ttl_secs: u64 = 0;
+    for row in &coerced_rows {
+        let key_val = match key_idx.and_then(|idx| row.get(idx)) {
+            Some((_, value)) => value.clone(),
+            None => SqlValue::String(String::new()),
+        };
+        if let Some((_, value)) = ttl_idx.and_then(|idx| row.get(idx)) {
+            match value {
+                SqlValue::Int(n) => ttl_secs = (*n).max(0) as u64,
+                SqlValue::Float(f) => ttl_secs = f.max(0.0) as u64,
+                _ => {}
+            }
+        }
+        let value_cols: Vec<(String, SqlValue)> = row
+            .iter()
+            .enumerate()
+            .filter(|(i, _)| !exclude_from_value.contains(i))
+            .map(|(_, cell)| cell.clone())
+            .collect();
+        entries.push((key_val, value_cols));
+    }
+    Ok(vec![SqlPlan::KvInsert {
+        collection: table_name,
+        entries,
+        ttl_secs,
+        intent,
+        on_conflict_updates,
+    }])
+}

nodedb-sql/src/planner/dml_helpers/kv_insert.rs

@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! DML planning helpers, split by concern:
+//! - [`value_convert`] — `ast::Expr` -> `SqlValue` conversion
+//! - [`range_check`] — declared-width coercion + range validation
+//! - [`insert_columns`] — positional-insert column resolution
+//! - [`ast_extract`] — table-name / primary-key point-lookup extraction
+//! - [`vector_primary_insert`] — vector-primary collection insert plans
+//! - [`kv_insert`] — KV engine insert plans
+
+mod ast_extract;
+mod insert_columns;
+mod kv_insert;
+mod range_check;
+mod value_convert;
+mod vector_primary_insert;
+
+pub use ast_extract::extract_point_keys;
+pub(super) use ast_extract::extract_table_name_from_table_with_joins;
+pub(super) use insert_columns::resolve_insert_columns;
+pub(super) use kv_insert::build_kv_insert_plan;
+pub(super) use range_check::{check_declared_int_ranges_in_assignments, coerce_and_check_rows};
+pub(super) use value_convert::{convert_value_rows, expr_to_sql_value};
+pub(super) use vector_primary_insert::build_vector_primary_insert_plan;