feat(crdt): wire document-row DocUpsert/DocDelete ops end-to-end

Adds CrdtOp::DocUpsert/DocDelete physical ops for scalar field
insert/replace/partial-update/delete on top-level LoroMap rows in
crdt=true document collections, and threads them through every layer
a Data Plane write touches: a new self-describing CrdtDocOp WAL record
(intent-carrying, since the Control Plane has no LoroDoc to diff),
replay handlers that re-run the same live upsert/delete logic, Raft
WAL replication encode/decode, dispatch routing, write-class/
permission classification, CDC change-event extraction, and response
shaping.

Author: farhan-syah

nodedb-crdt/src/state/core.rs

@@ -50,6 +50,48 @@ impl CrdtState {
         Ok(Self { doc, peer_id })
     }
 
+    /// Fetch a row's existing `LoroMap` container, or create one if absent.
+    /// Shared by `upsert` and `set_fields` — both need the same row handle
+    /// before diverging on prune-vs-preserve semantics.
+    fn row_container(&self, collection: &str, row_id: &str) -> Result<LoroMap> {
+        let coll = self.doc.get_map(collection);
+        match coll.get(row_id) {
+            Some(ValueOrContainer::Container(loro::Container::Map(m))) => Ok(m),
+            _ => coll
+                .insert_container(row_id, LoroMap::new())
+                .map_err(|e| CrdtError::Loro(e.to_string())),
+        }
+    }
+
+    /// Write `fields` onto `row_container` as scalar LWW inserts, rejecting
+    /// any key that currently holds a container value. Shared by `upsert`
+    /// and `set_fields` — both write the same way, only the prune step
+    /// (upsert-only) differs.
+    fn write_scalar_fields(
+        row_container: &LoroMap,
+        collection: &str,
+        row_id: &str,
+        fields: &[(&str, LoroValue)],
+    ) -> Result<()> {
+        for (field, value) in fields {
+            // A container-valued key can never legitimately appear in the
+            // incoming scalar projection. Overwriting one would destroy the
+            // nested container; skipping it would silently discard the
+            // caller's write. Reject instead of doing either.
+            if key_is_container(row_container, field) {
+                return Err(CrdtError::ScalarFieldShadowsContainer {
+                    collection: collection.to_string(),
+                    row_id: row_id.to_string(),
+                    field: (*field).to_string(),
+                });
+            }
+            row_container
+                .insert(field, value.clone())
+                .map_err(|e| CrdtError::Loro(e.to_string()))?;
+        }
+        Ok(())
+    }
+
     /// Insert or update a row in a collection.
     ///
     /// This is a REPLACE for scalar fields — every caller passes the
@@ -66,13 +108,7 @@ impl CrdtState {
         row_id: &str,
         fields: &[(&str, LoroValue)],
     ) -> Result<()> {
-        let coll = self.doc.get_map(collection);
-        let row_container = match coll.get(row_id) {
-            Some(ValueOrContainer::Container(loro::Container::Map(m))) => m,
-            _ => coll
-                .insert_container(row_id, LoroMap::new())
-                .map_err(|e| CrdtError::Loro(e.to_string()))?,
-        };
+        let row_container = self.row_container(collection, row_id)?;
 
         let incoming_keys: HashSet<&str> = fields.iter().map(|(field, _)| *field).collect();
 
@@ -94,23 +130,21 @@ impl CrdtState {
                 .map_err(|e| CrdtError::Loro(e.to_string()))?;
         }
 
-        for (field, value) in fields {
-            // A container-valued key can never legitimately appear in the
-            // incoming scalar projection. Overwriting one would destroy the
-            // nested container; skipping it would silently discard the
-            // caller's write. Reject instead of doing either.
-            if key_is_container(&row_container, field) {
-                return Err(CrdtError::ScalarFieldShadowsContainer {
-                    collection: collection.to_string(),
-                    row_id: row_id.to_string(),
-                    field: (*field).to_string(),
-                });
-            }
-            row_container
-                .insert(field, value.clone())
-                .map_err(|e| CrdtError::Loro(e.to_string()))?;
-        }
-        Ok(())
+        Self::write_scalar_fields(&row_container, collection, row_id, fields)
+    }
+
+    /// Partial-merge write: set exactly the provided scalar `fields` on a row
+    /// (LWW-per-field), creating the row if absent, leaving every untouched
+    /// key intact. This is `upsert` WITHOUT the full-projection prune step —
+    /// the UPDATE-SET semantic for `CrdtOp::DocUpsert { partial: true }`.
+    pub fn set_fields(
+        &self,
+        collection: &str,
+        row_id: &str,
+        fields: &[(&str, LoroValue)],
+    ) -> Result<()> {
+        let row_container = self.row_container(collection, row_id)?;
+        Self::write_scalar_fields(&row_container, collection, row_id, fields)
     }
 
     /// Delete a row from a collection.
@@ -315,3 +349,55 @@ impl RowLookup for CrdtState {
         self.field_value_exists_live(collection, field, value, exclude_row_id)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const COLL: &str = "c";
+    const ROW: &str = "r";
+
+    #[test]
+    fn set_fields_preserves_untouched_keys_and_upsert_prunes() {
+        let state = CrdtState::new(0).expect("state");
+
+        // Full projection {a:1, b:2}.
+        state
+            .upsert(
+                COLL,
+                ROW,
+                &[("a", LoroValue::I64(1)), ("b", LoroValue::I64(2))],
+            )
+            .expect("upsert");
+
+        // Partial-merge {b:9}: `a` must survive untouched, `b` overwritten.
+        state
+            .set_fields(COLL, ROW, &[("b", LoroValue::I64(9))])
+            .expect("set_fields");
+        assert_eq!(
+            state.read_field(COLL, ROW, "a"),
+            Some(LoroValue::I64(1)),
+            "set_fields must leave the untouched key `a` intact"
+        );
+        assert_eq!(
+            state.read_field(COLL, ROW, "b"),
+            Some(LoroValue::I64(9)),
+            "set_fields must overwrite `b` to 9"
+        );
+
+        // Full-projection replace {a:5}: absent key `b` must be pruned.
+        state
+            .upsert(COLL, ROW, &[("a", LoroValue::I64(5))])
+            .expect("upsert replace");
+        assert_eq!(
+            state.read_field(COLL, ROW, "a"),
+            Some(LoroValue::I64(5)),
+            "upsert must set `a` to 5"
+        );
+        assert_eq!(
+            state.read_field(COLL, ROW, "b"),
+            None,
+            "upsert must prune key `b` absent from the projection"
+        );
+    }
+}

nodedb-physical/src/physical_plan/crdt.rs

@@ -180,4 +180,31 @@ pub enum CrdtOp {
         /// Surrogate of the parent document hosting this block list.
         surrogate: Surrogate,
     },
+
+    // ─── Document row (top-level LoroMap) field-carrying ops ─────────
+    /// Insert-or-replace / partial-update a document row's scalar fields,
+    /// server-built from `fields_json` (a JSON object). Mirrors the
+    /// `ListInsert` intent-carrying contract: the Control Plane has no
+    /// `LoroDoc`, so the Data Plane handler builds the Loro mutation.
+    ///
+    /// `partial = false`: INSERT / full replace — scalar keys absent from
+    /// `fields_json` are pruned (LWW full-projection replace).
+    /// `partial = true`: UPDATE SET — only the provided fields are written
+    /// (LWW-per-field), untouched keys survive.
+    ///
+    /// `surrogate` is this row's OWN stable top-level cross-engine identity.
+    DocUpsert {
+        collection: String,
+        document_id: String,
+        fields_json: String,
+        surrogate: Surrogate,
+        partial: bool,
+    },
+    /// Delete a document row: tombstone in the collection's Loro doc + remove
+    /// from the sparse document store.
+    DocDelete {
+        collection: String,
+        document_id: String,
+        surrogate: Surrogate,
+    },
 }

nodedb-wal/src/record/types.rs

@@ -93,6 +93,21 @@ pub enum RecordType {
     /// was acknowledged to the client.
     CrdtListOp = 21 | 0x8000,
 
+    /// CRDT engine: document-row intent op (`DocUpsert` / `DocDelete`) —
+    /// field-carrying insert-or-replace / partial-update / delete of a
+    /// top-level `LoroMap` row for SQL DML on a `crdt='true'` collection.
+    ///
+    /// Like `CrdtListOp`, the Data Plane builds the Loro mutation server-side
+    /// and the Control Plane has no `LoroDoc` to compute a delta from, so the
+    /// record carries the **intent** (collection, document, surrogate, fields,
+    /// partial flag) rather than a Loro delta; replay re-executes the exact
+    /// same live handler that ran on first application.
+    ///
+    /// Required: skipping this record on replay silently drops an
+    /// acknowledged document write, diverging row content from what was
+    /// acknowledged to the client.
+    CrdtDocOp = 22 | 0x8000,
+
     /// Timeseries engine: metric sample batch.
     TimeseriesBatch = 30,
 
@@ -293,6 +308,7 @@ impl RecordType {
             x if x == 17 | 0x8000 => Some(Self::MultiVectorDelete),
             x if x == 20 | 0x8000 => Some(Self::CrdtDelta),
             x if x == 21 | 0x8000 => Some(Self::CrdtListOp),
+            x if x == 22 | 0x8000 => Some(Self::CrdtDocOp),
             x if x == 50 | 0x8000 => Some(Self::Transaction),
             x if x == 58 | 0x8000 => Some(Self::TransactionRedo),
             x if x == 51 | 0x8000 => Some(Self::SurrogateAlloc),
@@ -346,6 +362,7 @@ mod tests {
             RecordType::GraphNodeLabelRemove as u32
         ));
         assert!(RecordType::is_required(RecordType::CrdtListOp as u32));
+        assert!(RecordType::is_required(RecordType::CrdtDocOp as u32));
     }
 
     #[test]
@@ -364,6 +381,7 @@ mod tests {
             RecordType::MultiVectorDelete,
             RecordType::CrdtDelta,
             RecordType::CrdtListOp,
+            RecordType::CrdtDocOp,
             RecordType::TimeseriesBatch,
             RecordType::LogBatch,
             RecordType::ArrayPut,

nodedb/src/control/cluster/calvin/scheduler/driver/core/routing.rs

@@ -253,6 +253,8 @@ fn crdt_routing(op: &CrdtOp) -> PlanRouting {
         | CrdtOp::ListInsert { collection, .. }
         | CrdtOp::ListDelete { collection, .. }
         | CrdtOp::ListMove { collection, .. }
+        | CrdtOp::DocUpsert { collection, .. }
+        | CrdtOp::DocDelete { collection, .. }
         | CrdtOp::SetConstraints { collection, .. }
         | CrdtOp::DropConstraints { collection, .. }
         | CrdtOp::RestoreToVersion { collection, .. }

nodedb/src/control/gateway/version_set.rs

@@ -330,7 +330,9 @@ pub fn touched_collections(plan: &PhysicalPlan) -> Vec<String> {
                 | RestoreToVersion { collection, .. }
                 | ListInsert { collection, .. }
                 | ListDelete { collection, .. }
-                | ListMove { collection, .. } => out.push(collection.clone()),
+                | ListMove { collection, .. }
+                | DocUpsert { collection, .. }
+                | DocDelete { collection, .. } => out.push(collection.clone()),
 
                 // `ImportSnapshot` is a whole-tenant Loro import and
                 // `SetConstraints` / `DropConstraints` install validator rules —

nodedb/src/control/planner/calvin/write_class.rs

@@ -199,6 +199,8 @@ fn crdt_is_write(op: &CrdtOp) -> bool {
         | CrdtOp::ListInsert { .. }
         | CrdtOp::ListDelete { .. }
         | CrdtOp::ListMove { .. }
+        | CrdtOp::DocUpsert { .. }
+        | CrdtOp::DocDelete { .. }
         | CrdtOp::SetConstraints { .. }
         | CrdtOp::DropConstraints { .. }
         | CrdtOp::RestoreToVersion { .. }

nodedb/src/control/security/identity/plan_permission.rs

@@ -118,7 +118,9 @@ pub fn required_permission(plan: &crate::bridge::envelope::PhysicalPlan) -> Perm
             | CrdtOp::RestoreToVersion { .. }
             | CrdtOp::ListInsert { .. }
             | CrdtOp::ListDelete { .. }
-            | CrdtOp::ListMove { .. },
+            | CrdtOp::ListMove { .. }
+            | CrdtOp::DocUpsert { .. }
+            | CrdtOp::DocDelete { .. },
         ) => Permission::Write,
 
         PhysicalPlan::Vector(

nodedb/src/control/server/dispatch_utils/change_events/extract.rs

@@ -361,6 +361,31 @@ pub(super) fn extract_write_metadata(
         PhysicalPlan::Crdt(CrdtOp::ImportSnapshot { collection, .. }) => {
             vec![(collection.clone(), "*".into(), ChangeOperation::Update)]
         }
+        // Document-row field-carrying ops: a full replace / partial-update is
+        // an Insert / Update respectively; a delete is a Delete.
+        PhysicalPlan::Crdt(CrdtOp::DocUpsert {
+            collection,
+            document_id,
+            partial,
+            ..
+        }) => vec![(
+            collection.clone(),
+            document_id.clone(),
+            if *partial {
+                ChangeOperation::Update
+            } else {
+                ChangeOperation::Insert
+            },
+        )],
+        PhysicalPlan::Crdt(CrdtOp::DocDelete {
+            collection,
+            document_id,
+            ..
+        }) => vec![(
+            collection.clone(),
+            document_id.clone(),
+            ChangeOperation::Delete,
+        )],
         // Read (Read/ReadConstraints/GetPolicy/ReadAtVersion/
         // GetVersionVector/ExportDelta), history maintenance
         // (CompactAtVersion), and config/DDL (SetConstraints/

nodedb/src/control/server/response_shape/types.rs

@@ -31,7 +31,9 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
     match plan {
         PhysicalPlan::Document(DocumentOp::PointGet { .. })
         | PhysicalPlan::Crdt(CrdtOp::Read { .. })
-        | PhysicalPlan::Crdt(CrdtOp::GetPolicy { .. }) => PlanKind::SingleDocument,
+        | PhysicalPlan::Crdt(CrdtOp::GetPolicy { .. })
+        | PhysicalPlan::Crdt(CrdtOp::DocUpsert { .. })
+        | PhysicalPlan::Crdt(CrdtOp::DocDelete { .. }) => PlanKind::SingleDocument,
 
         PhysicalPlan::Vector(VectorOp::Search { .. })
         | PhysicalPlan::Vector(VectorOp::MultiSearch { .. })

nodedb/src/control/server/shared/plan_util.rs

@@ -19,6 +19,8 @@ pub(crate) fn extract_collection(plan: &PhysicalPlan) -> Option<&str> {
         | PhysicalPlan::Document(DocumentOp::RangeScan { collection, .. })
         | PhysicalPlan::Crdt(CrdtOp::Read { collection, .. })
         | PhysicalPlan::Crdt(CrdtOp::Apply { collection, .. })
+        | PhysicalPlan::Crdt(CrdtOp::DocUpsert { collection, .. })
+        | PhysicalPlan::Crdt(CrdtOp::DocDelete { collection, .. })
         | PhysicalPlan::Vector(VectorOp::Insert { collection, .. })
         | PhysicalPlan::Vector(VectorOp::BatchInsert { collection, .. })
         | PhysicalPlan::Vector(VectorOp::MultiSearch { collection, .. })