feat(control): route DML on CRDT document collections to CrdtOp

INSERT, UPSERT, UPDATE, and DELETE on a `crdt = true` document
collection now lower to CrdtOp::DocUpsert / DocDelete instead of the
plain DocumentOp path, so writes converge via CRDT last-writer-wins
rather than bypassing sync. Only representable shapes are routed:
full-replace INSERT/UPSERT and PK-targeted UPDATE with literal SET
values map to DocUpsert (partial for UPDATE), PK-targeted DELETE maps
to DocDelete. Predicate UPDATE/DELETE, IF NOT EXISTS INSERT, ON
CONFLICT DO UPDATE, non-literal UPDATE RHS, and UPDATE ... RETURNING
are rejected outright rather than silently falling through to the
non-CRDT path.

A shared crdt_gate module detects CRDT collections once per statement
and builds the fields_json payload consumed by DocUpsert.

Author: farhan-syah

nodedb/src/control/planner/sql_plan_convert/dml/crdt_gate.rs

@@ -0,0 +1,89 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! CRDT-routing gate for document-collection DML.
+//!
+//! Detects `crdt = true` document collections and builds the `fields_json`
+//! payload that `CrdtOp::DocUpsert` consumes. Lives at the `dml` module level
+//! so INSERT / UPSERT (in `insert.rs`) and UPDATE / DELETE (in
+//! `update_delete/`) all reach the same detection + payload builders.
+//!
+//! Routing contract: on a CRDT document collection, PK-targeted
+//! INSERT / UPSERT (full replace) / UPDATE-SET (literal RHS) / DELETE lower to
+//! `CrdtOp::DocUpsert` / `DocDelete`; every unsupported shape (predicate
+//! UPDATE/DELETE, non-literal UPDATE RHS, UPDATE ... RETURNING, `IF NOT EXISTS`
+//! INSERT, explicit `ON CONFLICT DO UPDATE`) is rejected with a typed error.
+//! There is NO silent fallthrough to a non-CRDT `DocumentOp` — that would
+//! bypass CRDT convergence and is a correctness bug.
+
+use nodedb_sql::types::{SqlExpr, SqlValue};
+
+use crate::control::planner::sql_plan_convert::convert::ConvertContext;
+use crate::control::planner::sql_plan_convert::value::row_to_msgpack;
+
+/// `true` when `collection` (already db-qualified by the caller) is a CRDT
+/// document collection.
+///
+/// A genuine catalog READ error propagates: misrouting a write to the non-CRDT
+/// path would silently bypass CRDT convergence. An ABSENT credential store or
+/// catalog, or an absent collection row (`Ok(None)`), is treated as non-CRDT
+/// (`Ok(false)`). Mirrors `document_collection_is_edge_bearing` in
+/// `update_delete/shared.rs` exactly.
+pub(in crate::control::planner::sql_plan_convert::dml) fn document_collection_is_crdt(
+    ctx: &ConvertContext,
+    collection: &str,
+) -> crate::Result<bool> {
+    let Some(credentials) = ctx.credentials.as_ref() else {
+        return Ok(false);
+    };
+    let catalog = credentials.catalog();
+    Ok(catalog
+        .get_collection(ctx.database_id, ctx.tenant_id.as_u64(), collection)?
+        .map(|c| c.crdt)
+        .unwrap_or(false))
+}
+
+/// Full-row fields → JSON object string for `DocUpsert.fields_json`.
+///
+/// Reuses the DML msgpack writer + `json_from_msgpack` so the stored payload is
+/// byte-identical to what the non-CRDT `PointInsert` path would produce for the
+/// same row, then serializes to JSON text (the shape `DocUpsert` consumes).
+pub(in crate::control::planner::sql_plan_convert::dml) fn row_to_fields_json(
+    row: &[(String, SqlValue)],
+) -> crate::Result<String> {
+    let mpk = row_to_msgpack(row)?;
+    let json = nodedb_types::json_from_msgpack(&mpk).map_err(|e| crate::Error::Serialization {
+        format: "json".into(),
+        detail: format!("crdt fields_json encode: {e}"),
+    })?;
+    sonic_rs::to_string(&json).map_err(|e| crate::Error::Serialization {
+        format: "json".into(),
+        detail: format!("crdt fields_json encode: {e}"),
+    })
+}
+
+/// UPDATE SET (partial) `fields_json` from LITERAL assignments only.
+///
+/// An expression RHS is rejected: the Control Plane planner cannot evaluate a
+/// per-row expression here (same restriction the KV and columnar/spatial UPDATE
+/// paths enforce). Only the provided fields go into the object — `DocUpsert`
+/// with `partial = true` writes exactly these (LWW-per-field), leaving untouched
+/// keys intact.
+pub(in crate::control::planner::sql_plan_convert::dml) fn literal_assignments_to_fields_json(
+    assignments: &[(String, SqlExpr)],
+) -> crate::Result<String> {
+    let mut row: Vec<(String, SqlValue)> = Vec::with_capacity(assignments.len());
+    for (field, expr) in assignments {
+        match expr {
+            SqlExpr::Literal(v) => row.push((field.clone(), v.clone())),
+            _ => {
+                return Err(crate::Error::BadRequest {
+                    detail: format!(
+                        "UPDATE with non-literal RHS on CRDT collection (field '{field}') \
+                         is not supported; use a literal value"
+                    ),
+                });
+            }
+        }
+    }
+    row_to_fields_json(&row)
+}

nodedb/src/control/planner/sql_plan_convert/dml/insert.rs

@@ -195,6 +195,19 @@ pub(in super::super) fn convert_insert(
     let mut tasks = Vec::new();
     let mut columnar_rows: Vec<&Vec<(String, SqlValue)>> = Vec::new();
 
+    // Detect CRDT document collections once (never re-hit the catalog per row).
+    // `IF NOT EXISTS` (ON CONFLICT DO NOTHING → `if_absent`) cannot be honored by
+    // `CrdtOp::DocUpsert`, which is an unconditional LWW full-replace: reject.
+    let is_crdt = super::crdt_gate::document_collection_is_crdt(ctx, collection)?;
+    if is_crdt && if_absent {
+        return Err(crate::Error::BadRequest {
+            detail: format!(
+                "INSERT ... IF NOT EXISTS on CRDT collection '{collection}' is not supported; \
+                 CRDT documents converge via last-writer-wins full replace"
+            ),
+        });
+    }
+
     let mut expanded_rows: Vec<Vec<(String, SqlValue)>> = Vec::with_capacity(rows.len());
     for row in rows {
         if column_defaults.is_empty() {
@@ -252,17 +265,28 @@ pub(in super::super) fn convert_insert(
                     let s = assign_for_pk(ctx, collection, doc_id.as_bytes())?;
                     (doc_id, s)
                 };
-                tasks.push(PhysicalTask {
-                    tenant_id,
-                    vshard_id: vshard,
-                    database_id: ctx.database_id,
-                    plan: PhysicalPlan::Document(DocumentOp::PointInsert {
+                let plan = if is_crdt {
+                    PhysicalPlan::Crdt(CrdtOp::DocUpsert {
+                        collection: collection.into(),
+                        document_id: doc_id,
+                        fields_json: super::crdt_gate::row_to_fields_json(row)?,
+                        surrogate,
+                        partial: false,
+                    })
+                } else {
+                    PhysicalPlan::Document(DocumentOp::PointInsert {
                         collection: collection.into(),
                         document_id: doc_id,
                         value: value_bytes,
                         if_absent,
                         surrogate,
-                    }),
+                    })
+                };
+                tasks.push(PhysicalTask {
+                    tenant_id,
+                    vshard_id: vshard,
+                    database_id: ctx.database_id,
+                    plan,
                     post_set_op: PostSetOp::None,
                     txn_id: None,
                 });
@@ -341,6 +365,20 @@ pub(in super::super) fn convert_upsert(
     let vshard = VShardId::from_collection_in_database(ctx.database_id, collection);
     let mut tasks = Vec::new();
 
+    // Detect CRDT document collections once. An explicit `ON CONFLICT DO UPDATE
+    // SET ...` cannot be honored: CRDT conflict resolution IS the LWW
+    // full-replace `DocUpsert` performs, so a caller-supplied merge clause has
+    // no place to run. Reject rather than silently ignore it.
+    let is_crdt = super::crdt_gate::document_collection_is_crdt(ctx, collection)?;
+    if is_crdt && !on_conflict_updates.is_empty() {
+        return Err(crate::Error::BadRequest {
+            detail: format!(
+                "UPSERT with ON CONFLICT DO UPDATE on CRDT collection '{collection}' is not \
+                 supported; CRDT documents converge via last-writer-wins full replace"
+            ),
+        });
+    }
+
     let on_conflict_values = if on_conflict_updates.is_empty() {
         Vec::new()
     } else {
@@ -368,17 +406,28 @@ pub(in super::super) fn convert_upsert(
                     let s = assign_for_pk(ctx, collection, doc_id.as_bytes())?;
                     (doc_id, s)
                 };
-                tasks.push(PhysicalTask {
-                    tenant_id,
-                    vshard_id: vshard,
-                    database_id: ctx.database_id,
-                    plan: PhysicalPlan::Document(DocumentOp::Upsert {
+                let plan = if is_crdt {
+                    PhysicalPlan::Crdt(CrdtOp::DocUpsert {
+                        collection: collection.into(),
+                        document_id: doc_id,
+                        fields_json: super::crdt_gate::row_to_fields_json(row)?,
+                        surrogate,
+                        partial: false,
+                    })
+                } else {
+                    PhysicalPlan::Document(DocumentOp::Upsert {
                         collection: collection.into(),
                         document_id: doc_id,
                         value: value_bytes,
                         on_conflict_updates: on_conflict_values.clone(),
                         surrogate,
-                    }),
+                    })
+                };
+                tasks.push(PhysicalTask {
+                    tenant_id,
+                    vshard_id: vshard,
+                    database_id: ctx.database_id,
+                    plan,
                     post_set_op: PostSetOp::None,
                     txn_id: None,
                 });

nodedb/src/control/planner/sql_plan_convert/dml/mod.rs

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
+mod crdt_gate;
 mod insert;
 mod kv_and_vector;
 mod merge;

nodedb/src/control/planner/sql_plan_convert/dml/update_delete/delete.rs

@@ -69,6 +69,21 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_delete(
         }]);
     }
 
+    // CRDT GATE: a DELETE on a `crdt = true` document collection routes to
+    // `CrdtOp::DocDelete` (tombstone + sparse-store removal). Only a PK-targeted
+    // DELETE is representable; a predicate (non-PK) DELETE is rejected — there is
+    // NO silent fallthrough to a `DocumentOp`, which would bypass CRDT
+    // convergence. Read the flag ONCE, before the PK loop.
+    let is_crdt = super::super::crdt_gate::document_collection_is_crdt(ctx, collection)?;
+    if is_crdt && target_keys.is_empty() {
+        return Err(crate::Error::BadRequest {
+            detail: format!(
+                "predicate (non-primary-key) DELETE on CRDT collection '{collection}' is not \
+                 supported; target rows by primary key"
+            ),
+        });
+    }
+
     // EDGE-BEARING GATE: a PK-equality delete on a schemaless-document
     // collection that carries implicit edges must NOT lower to a static
     // `PointDelete` — that op bypasses the dependent-predicate (OLLP) path
@@ -79,7 +94,8 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_delete(
     // collections keep the fast `PointDelete` path below. Reached only for
     // document engines (the KV case returned above); strict/columnar/etc.
     // never set `has_implicit_edges`, so the flag naturally scopes this.
-    if !target_keys.is_empty() && document_collection_is_edge_bearing(ctx, collection)? {
+    if !is_crdt && !target_keys.is_empty() && document_collection_is_edge_bearing(ctx, collection)?
+    {
         let effective_filter = delete_effective_filter(filters, target_keys)?;
         return Ok(vec![PhysicalTask {
             tenant_id,
@@ -109,17 +125,26 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_delete(
                 },
                 None => Surrogate::ZERO,
             };
-            tasks.push(PhysicalTask {
-                tenant_id,
-                vshard_id: vshard,
-                database_id: ctx.database_id,
-                plan: PhysicalPlan::Document(DocumentOp::PointDelete {
+            let plan = if is_crdt {
+                PhysicalPlan::Crdt(CrdtOp::DocDelete {
+                    collection: collection.into(),
+                    document_id: pk_string,
+                    surrogate,
+                })
+            } else {
+                PhysicalPlan::Document(DocumentOp::PointDelete {
                     collection: collection.into(),
                     document_id: pk_string,
                     surrogate,
                     pk_bytes,
                     returning: None,
-                }),
+                })
+            };
+            tasks.push(PhysicalTask {
+                tenant_id,
+                vshard_id: vshard,
+                database_id: ctx.database_id,
+                plan,
                 post_set_op: PostSetOp::None,
                 txn_id: None,
             });