feat(nodedb): support INSERT ... ON CONFLICT DO UPDATE SET with expression evaluation

Introduce `UpdateValue` on the bridge — a tagged union carrying either
pre-encoded msgpack bytes (fast literal path) or a `SqlExpr` that the
Data Plane evaluates against the existing row at apply time.

Planner changes:
- `SqlPlan::Upsert` gains an `on_conflict_updates` field for the SET
  assignments extracted from `ON CONFLICT (...) DO UPDATE SET`.
- `dml::plan_upsert_with_on_conflict` routes `INSERT ... ON CONFLICT DO
  UPDATE SET` through the upsert path, carrying the assignments.
- `assignments_to_update_values` converts planner `SqlExpr` to
  `UpdateValue`, choosing `Literal` for constant RHS and `Expr` for
  anything that must be evaluated per-row.
- `ScanFilter` gains an `expr: Option<SqlExpr>` field so complex WHERE
  clauses (scalar functions, arithmetic, non-literal BETWEEN) that the
  planner cannot reduce to simple triples are shipped verbatim to the
  Data Plane for row-level evaluation.
- `sql_expr_to_bridge_expr` extended to cover `UnaryOp`, `IsNull`,
  `Between`, `Substring`, `Trim`, and `Case` expressions.
- `sql_plan_convert/filter.rs` populates the new `expr` field for
  general-purpose `FilterExpr::Expr` predicates.

Executor changes:
- `bulk_dml` and `upsert` handlers receive `&[(String, UpdateValue)]`
  instead of `&[(String, Vec<u8>)]`; literal arms are identical to the
  previous fast path, expression arms evaluate against a pre-update
  snapshot of the row (matching PostgreSQL semantics — assignments do
  not observe each other).
- Empty `filter_bytes` in `bulk_dml` now means "no WHERE clause" instead
  of a deserialization error.
- `check_generated_readonly` and `needs_recomputation` generified over
  the update-value type so both call sites compile without duplication.

All existing call sites updated to pass `expr: None` on the `ScanFilter`
constructor and supply the required `on_conflict_updates` slice.

Author: farhan-syah

nodedb-sql/src/engine_rules/document_schemaless.rs

@@ -22,6 +22,7 @@ impl EngineRules for SchemalessRules {
             engine: EngineType::DocumentSchemaless,
             rows: p.rows,
             column_defaults: p.column_defaults,
+            on_conflict_updates: p.on_conflict_updates,
         }])
     }
 

nodedb-sql/src/engine_rules/document_strict.rs

@@ -22,6 +22,7 @@ impl EngineRules for StrictRules {
             engine: EngineType::DocumentStrict,
             rows: p.rows,
             column_defaults: p.column_defaults,
+            on_conflict_updates: p.on_conflict_updates,
         }])
     }
 

nodedb-sql/src/engine_rules/mod.rs

@@ -59,6 +59,10 @@ pub struct UpsertParams {
     pub columns: Vec<String>,
     pub rows: Vec<Vec<(String, SqlValue)>>,
     pub column_defaults: Vec<(String, String)>,
+    /// `ON CONFLICT (...) DO UPDATE SET` assignments. Empty for plain
+    /// `UPSERT INTO ...`; populated when the caller is
+    /// `INSERT ... ON CONFLICT ... DO UPDATE SET`.
+    pub on_conflict_updates: Vec<(String, SqlExpr)>,
 }
 
 /// Parameters for planning an AGGREGATE operation.

nodedb-sql/src/planner/dml.rs

@@ -8,8 +8,45 @@ use crate::parser::normalize::{normalize_ident, normalize_object_name};
 use crate::resolver::expr::{convert_expr, convert_value};
 use crate::types::*;
 
+/// Extract `ON CONFLICT (...) DO UPDATE SET` assignments from an AST
+/// insert, or `None` if this is a plain INSERT.
+fn extract_on_conflict_updates(ins: &ast::Insert) -> Result<Option<Vec<(String, SqlExpr)>>> {
+    let Some(on) = ins.on.as_ref() else {
+        return Ok(None);
+    };
+    let ast::OnInsert::OnConflict(oc) = on else {
+        return Ok(None);
+    };
+    let ast::OnConflictAction::DoUpdate(do_update) = &oc.action else {
+        // DO NOTHING maps to "ignore conflict" — currently unsupported.
+        return Err(SqlError::Unsupported {
+            detail: "ON CONFLICT DO NOTHING is not yet supported".into(),
+        });
+    };
+    let mut pairs = Vec::with_capacity(do_update.assignments.len());
+    for a in &do_update.assignments {
+        let name = match &a.target {
+            ast::AssignmentTarget::ColumnName(obj) => normalize_object_name(obj),
+            _ => {
+                return Err(SqlError::Unsupported {
+                    detail: "ON CONFLICT DO UPDATE SET target must be a column name".into(),
+                });
+            }
+        };
+        let expr = convert_expr(&a.value)?;
+        pairs.push((name, expr));
+    }
+    Ok(Some(pairs))
+}
+
 /// Plan an INSERT statement.
 pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
+    // `INSERT ... ON CONFLICT DO UPDATE SET` reroutes to the upsert path
+    // with the assignments carried through. Detected before any other
+    // work so both planning paths share the `ast::Insert` decode below.
+    if let Some(on_conflict_updates) = extract_on_conflict_updates(ins)? {
+        return plan_upsert_with_on_conflict(ins, catalog, on_conflict_updates);
+    }
     let table_name = match &ins.table {
         ast::TableObject::TableName(name) => normalize_object_name(name),
         ast::TableObject::TableFunction(_) => {
@@ -190,6 +227,60 @@ pub fn plan_upsert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
         columns,
         rows,
         column_defaults,
+        on_conflict_updates: Vec::new(),
+    })
+}
+
+/// Plan an `INSERT ... ON CONFLICT DO UPDATE SET` statement. Identical to
+/// `plan_upsert` except the assignments are carried onto the upsert plan
+/// so the Data Plane can evaluate them against the existing row instead
+/// of merging the would-be-inserted values.
+fn plan_upsert_with_on_conflict(
+    ins: &ast::Insert,
+    catalog: &dyn SqlCatalog,
+    on_conflict_updates: Vec<(String, SqlExpr)>,
+) -> Result<Vec<SqlPlan>> {
+    let table_name = match &ins.table {
+        ast::TableObject::TableName(name) => normalize_object_name(name),
+        ast::TableObject::TableFunction(_) => {
+            return Err(SqlError::Unsupported {
+                detail: "INSERT ... ON CONFLICT on a table function is not supported".into(),
+            });
+        }
+    };
+    let info = catalog
+        .get_collection(&table_name)?
+        .ok_or_else(|| SqlError::UnknownTable {
+            name: table_name.clone(),
+        })?;
+
+    let columns: Vec<String> = ins.columns.iter().map(normalize_ident).collect();
+
+    let source = ins.source.as_ref().ok_or_else(|| SqlError::Parse {
+        detail: "INSERT ... ON CONFLICT requires VALUES".into(),
+    })?;
+    let rows_ast = match &*source.body {
+        ast::SetExpr::Values(values) => &values.rows,
+        _ => {
+            return Err(SqlError::Unsupported {
+                detail: "INSERT ... ON CONFLICT source must be VALUES".into(),
+            });
+        }
+    };
+
+    let rows = convert_value_rows(&columns, rows_ast)?;
+    let column_defaults: Vec<(String, String)> = info
+        .columns
+        .iter()
+        .filter_map(|c| c.default.as_ref().map(|d| (c.name.clone(), d.clone())))
+        .collect();
+    let rules = engine_rules::resolve_engine_rules(info.engine);
+    rules.plan_upsert(engine_rules::UpsertParams {
+        collection: table_name,
+        columns,
+        rows,
+        column_defaults,
+        on_conflict_updates,
     })
 }
 

nodedb-sql/src/types.rs

@@ -65,6 +65,11 @@ pub enum SqlPlan {
         engine: EngineType,
         rows: Vec<Vec<(String, SqlValue)>>,
         column_defaults: Vec<(String, String)>,
+        /// `ON CONFLICT (...) DO UPDATE SET field = expr` assignments.
+        /// When empty, upsert is a plain merge: new columns overwrite existing.
+        /// When non-empty, the engine applies these per-row against the
+        /// *existing* document instead of merging the inserted values.
+        on_conflict_updates: Vec<(String, SqlExpr)>,
     },
     InsertSelect {
         target: String,

nodedb/src/bridge/physical_plan/document.rs

@@ -2,6 +2,54 @@
 
 use nodedb_types::columnar::StrictSchema;
 
+/// Right-hand side of an UPDATE ... SET field = <...> assignment.
+///
+/// The planner turns each assignment into one of these before it crosses
+/// the SPSC bridge:
+///
+/// - `Literal` — pre-encoded msgpack bytes for constant RHS. This is the
+///   fast path: the Data Plane can merge these at the binary level for
+///   non-strict collections without decoding the current row.
+/// - `Expr` — a `SqlExpr` that must be evaluated against the *current*
+///   document at apply time. Used for arithmetic (`col + 1`), functions
+///   (`LOWER(col)`, `NOW()`), `CASE`, concatenation, and anything else
+///   whose result depends on the row being updated.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub enum UpdateValue {
+    Literal(Vec<u8>),
+    Expr(crate::bridge::expr_eval::SqlExpr),
+}
+
+impl zerompk::ToMessagePack for UpdateValue {
+    fn write<W: zerompk::Write>(&self, writer: &mut W) -> zerompk::Result<()> {
+        writer.write_array_len(2)?;
+        match self {
+            UpdateValue::Literal(bytes) => {
+                writer.write_u8(0)?;
+                bytes.write(writer)
+            }
+            UpdateValue::Expr(expr) => {
+                writer.write_u8(1)?;
+                expr.write(writer)
+            }
+        }
+    }
+}
+
+impl<'a> zerompk::FromMessagePack<'a> for UpdateValue {
+    fn read<R: zerompk::Read<'a>>(reader: &mut R) -> zerompk::Result<Self> {
+        reader.check_array_len(2)?;
+        let tag = reader.read_u8()?;
+        match tag {
+            0 => Ok(UpdateValue::Literal(Vec::<u8>::read(reader)?)),
+            1 => Ok(UpdateValue::Expr(crate::bridge::expr_eval::SqlExpr::read(
+                reader,
+            )?)),
+            _ => Err(zerompk::Error::InvalidMarker(tag)),
+        }
+    }
+}
+
 /// Storage encoding mode for a document collection.
 ///
 /// Determines how documents are serialized before storage in the sparse engine.
@@ -139,8 +187,8 @@ pub enum DocumentOp {
     PointUpdate {
         collection: String,
         document_id: String,
-        /// Field name → new JSON value.
-        updates: Vec<(String, Vec<u8>)>,
+        /// Field name → assignment RHS (literal bytes or row-scope expression).
+        updates: Vec<(String, UpdateValue)>,
         /// If true, return the post-update document as payload (for RETURNING clause).
         returning: bool,
     },
@@ -217,18 +265,22 @@ pub enum DocumentOp {
         source_limit: usize,
     },
 
-    /// Upsert: insert or merge.
+    /// Upsert: insert or merge. When `on_conflict_updates` is non-empty,
+    /// the conflict branch evaluates those assignments against the
+    /// *existing* document instead of merging the inserted value —
+    /// the `INSERT ... ON CONFLICT DO UPDATE SET ...` path.
     Upsert {
         collection: String,
         document_id: String,
         value: Vec<u8>,
+        on_conflict_updates: Vec<(String, UpdateValue)>,
     },
 
     /// Bulk update: scan + apply field updates to all matches.
     BulkUpdate {
         collection: String,
         filters: Vec<u8>,
-        updates: Vec<(String, Vec<u8>)>,
+        updates: Vec<(String, UpdateValue)>,
         /// If true, return updated documents as JSON array payload (for RETURNING clause).
         returning: bool,
     },

nodedb/src/bridge/physical_plan/mod.rs

@@ -20,7 +20,7 @@ pub use columnar::ColumnarOp;
 pub use crdt::CrdtOp;
 pub use document::{
     BalancedDef, DocumentOp, EnforcementOptions, GeneratedColumnSpec, MaterializedSumBinding,
-    PeriodLockConfig, StorageMode,
+    PeriodLockConfig, StorageMode, UpdateValue,
 };
 pub use graph::GraphOp;
 pub use kv::KvOp;

nodedb/src/control/planner/rls_injection.rs

@@ -352,6 +352,7 @@ fn inject_permission_tree_for_plan(
                 .collect(),
         ),
         clauses: Vec::new(),
+        expr: None,
     };
 
     let filter_bytes =

nodedb/src/control/planner/sql_plan_convert/aggregate.rs

@@ -345,8 +345,43 @@ pub(super) fn serialize_window_functions(
 
 #[cfg(test)]
 mod tests {
-    use super::{extract_computed_columns, extract_projection_names};
-    use nodedb_sql::types::{Projection, SqlExpr, WindowSpec};
+    use super::{agg_expr_to_spec, extract_computed_columns, extract_projection_names};
+    use nodedb_sql::types::{AggregateExpr, BinaryOp, Projection, SqlExpr, SqlValue, WindowSpec};
+
+    #[test]
+    fn aggregate_spec_preserves_alias_and_case_expression() {
+        let agg = AggregateExpr {
+            function: "sum".into(),
+            args: vec![SqlExpr::Case {
+                operand: None,
+                when_then: vec![(
+                    SqlExpr::BinaryOp {
+                        left: Box::new(SqlExpr::Column {
+                            table: None,
+                            name: "category".into(),
+                        }),
+                        op: BinaryOp::Eq,
+                        right: Box::new(SqlExpr::Literal(SqlValue::String("tools".into()))),
+                    },
+                    SqlExpr::Literal(SqlValue::Int(1)),
+                )],
+                else_expr: Some(Box::new(SqlExpr::Literal(SqlValue::Int(0)))),
+            }],
+            alias: "tools_count".into(),
+            distinct: false,
+        };
+
+        let spec = agg_expr_to_spec(&agg);
+
+        assert_eq!(spec.function, "sum");
+        assert_eq!(spec.alias, "sum(*)");
+        assert_eq!(spec.user_alias.as_deref(), Some("tools_count"));
+        assert_eq!(spec.field, "*");
+        assert!(matches!(
+            spec.expr,
+            Some(crate::bridge::expr_eval::SqlExpr::Case { .. })
+        ));
+    }
 
     #[test]
     fn window_aliases_stay_in_projection_and_out_of_computed_columns() {

nodedb/src/control/planner/sql_plan_convert/convert.rs

@@ -84,7 +84,15 @@ pub(super) fn convert_one(
             engine,
             rows,
             column_defaults,
-        } => super::dml::convert_upsert(collection, engine, rows, column_defaults, tenant_id),
+            on_conflict_updates,
+        } => super::dml::convert_upsert(
+            collection,
+            engine,
+            rows,
+            column_defaults,
+            on_conflict_updates,
+            tenant_id,
+        ),
 
         SqlPlan::KvInsert {
             collection,