feat(crdt): support RETURNING on CRDT document UPDATE/DELETE

Adds a response-only ReturningSpec field to CrdtOp::DocUpsert and
DocDelete, threaded through the planner (UPDATE/DELETE now route to
DocUpsert/DocDelete with RETURNING instead of being rejected), the
pgwire returning-spec injection, response-shape classification, WAL
dispatch/replication, and the Data Plane handlers that project the
written or pre-deletion row. WAL and replicated records still encode
with returning=None since the projection is response-only.

Author: farhan-syah

nodedb-physical/src/physical_plan/crdt.rs

@@ -4,6 +4,8 @@
 
 use nodedb_types::Surrogate;
 
+use super::ReturningSpec;
+
 /// CRDT engine physical operations.
 #[derive(
     Debug,
@@ -199,12 +201,20 @@ pub enum CrdtOp {
         fields_json: String,
         surrogate: Surrogate,
         partial: bool,
+        /// Response-only RETURNING projection; None for a plain write. NOT
+        /// persisted/replicated — WAL/replication reconstruct with None.
+        #[serde(default)]
+        returning: Option<ReturningSpec>,
     },
     /// 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,
+        /// Response-only RETURNING projection; None for a plain write. NOT
+        /// persisted/replicated — WAL/replication reconstruct with None.
+        #[serde(default)]
+        returning: Option<ReturningSpec>,
     },
 }

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

@@ -272,6 +272,7 @@ pub(in super::super) fn convert_insert(
                         fields_json: super::crdt_gate::row_to_fields_json(row)?,
                         surrogate,
                         partial: false,
+                        returning: None,
                     })
                 } else {
                     PhysicalPlan::Document(DocumentOp::PointInsert {
@@ -413,6 +414,7 @@ pub(in super::super) fn convert_upsert(
                         fields_json: super::crdt_gate::row_to_fields_json(row)?,
                         surrogate,
                         partial: false,
+                        returning: None,
                     })
                 } else {
                     PhysicalPlan::Document(DocumentOp::Upsert {

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

@@ -130,6 +130,7 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_delete(
                     collection: collection.into(),
                     document_id: pk_string,
                     surrogate,
+                    returning: None,
                 })
             } else {
                 PhysicalPlan::Document(DocumentOp::PointDelete {

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

@@ -293,16 +293,34 @@ fn non_literal_rhs_update_on_crdt_rejects() {
 }
 
 #[test]
-fn update_returning_on_crdt_rejects() {
+fn update_returning_on_crdt_routes_to_doc_upsert() {
+    // RETURNING is stripped before planning and re-injected downstream
+    // (pgwire `inject_returning_spec` attaches the spec to the CrdtOp, and the
+    // DP handler emits the projected rows). So at the convert layer a RETURNING
+    // UPDATE on a crdt collection routes to `DocUpsert` exactly like a plain
+    // UPDATE — it is NOT rejected. `returning` on the op is `None` here; the
+    // spec is attached later at the protocol boundary.
     let (ctx, _dir) = ctx_with_catalog();
     let assignments = vec![(
         "name".to_string(),
         SqlExpr::Literal(SqlValue::String("bob".to_string())),
     )];
     let keys = vec![SqlValue::String("k1".to_string())];
-    let err = convert_update(update_params("crdt_coll", &assignments, &keys, true, &ctx))
-        .expect_err("UPDATE ... RETURNING on crdt must reject");
-    assert!(matches!(err, crate::Error::BadRequest { .. }));
+    let tasks = convert_update(update_params("crdt_coll", &assignments, &keys, true, &ctx))
+        .expect("UPDATE ... RETURNING on crdt must route, not reject");
+    assert_eq!(tasks.len(), 1);
+    match &tasks[0].plan {
+        PhysicalPlan::Crdt(CrdtOp::DocUpsert {
+            partial, returning, ..
+        }) => {
+            assert!(partial, "UPDATE SET must be a partial DocUpsert");
+            assert!(
+                returning.is_none(),
+                "RETURNING spec is attached downstream, not at convert"
+            );
+        }
+        other => panic!("expected partial CrdtOp::DocUpsert, got {other:?}"),
+    }
 }
 
 #[test]

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

@@ -41,7 +41,7 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_update(
         assignments,
         filters,
         target_keys,
-        returning,
+        returning: _returning,
         tenant_id,
         ctx,
     } = params;
@@ -146,26 +146,20 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_update(
 
     // CRDT GATE: an UPDATE on a `crdt = true` document collection routes to
     // `CrdtOp::DocUpsert` with `partial = true` (LWW-per-field). Only PK-targeted
-    // SET with literal RHS is representable. Predicate (non-PK) UPDATE, non-literal
-    // RHS, and UPDATE ... RETURNING are rejected — there is NO silent fallthrough
-    // to a `DocumentOp`, which would bypass CRDT convergence. Read the flag ONCE.
+    // SET with literal RHS is representable. Predicate (non-PK) UPDATE and
+    // non-literal RHS are rejected — there is NO silent fallthrough to a
+    // `DocumentOp`, which would bypass CRDT convergence. `RETURNING` IS
+    // supported: the response-only projection is injected into the plan later
+    // and emitted by the Data Plane handler, exactly like `PointUpdate`. Read
+    // the flag ONCE.
     let is_crdt = super::super::crdt_gate::document_collection_is_crdt(ctx, collection)?;
-    if is_crdt {
-        if returning {
-            return Err(crate::Error::BadRequest {
-                detail: format!(
-                    "UPDATE ... RETURNING on CRDT collection '{collection}' is not supported"
-                ),
-            });
-        }
-        if target_keys.is_empty() {
-            return Err(crate::Error::BadRequest {
-                detail: format!(
-                    "predicate (non-primary-key) UPDATE on CRDT collection '{collection}' is not \
-                     supported; target rows by primary key"
-                ),
-            });
-        }
+    if is_crdt && target_keys.is_empty() {
+        return Err(crate::Error::BadRequest {
+            detail: format!(
+                "predicate (non-primary-key) UPDATE on CRDT collection '{collection}' is not \
+                 supported; target rows by primary key"
+            ),
+        });
     }
     // Partial-update payload for the CRDT path, built ONCE from the literal SET
     // assignments (non-literal RHS rejected inside the builder).
@@ -246,6 +240,7 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_update(
                     fields_json: fields_json.clone(),
                     surrogate,
                     partial: true,
+                    returning: None,
                 })
             } else {
                 PhysicalPlan::Document(DocumentOp::PointUpdate {

nodedb/src/control/server/pgwire/handler/routing/planning.rs

@@ -257,14 +257,15 @@ pub(super) fn consistency_for_tasks(tasks: &[PhysicalTask]) -> crate::types::Rea
 
 /// Inject a RETURNING spec into a DML physical plan variant.
 ///
-/// Only `PointUpdate`, `BulkUpdate`, `PointDelete`, and `BulkDelete` are
-/// affected. All other plan variants are left unchanged.
+/// Only `PointUpdate`, `BulkUpdate`, `PointDelete`, `BulkDelete`,
+/// `UpdateFromJoin`, and the CRDT `DocUpsert` / `DocDelete` ops are affected.
+/// All other plan variants are left unchanged.
 pub(super) fn inject_returning_spec(
     plan: &mut crate::bridge::envelope::PhysicalPlan,
     spec: nodedb_physical::physical_plan::ReturningSpec,
 ) {
     use crate::bridge::envelope::PhysicalPlan;
-    use nodedb_physical::physical_plan::DocumentOp;
+    use nodedb_physical::physical_plan::{CrdtOp, DocumentOp};
 
     match plan {
         PhysicalPlan::Document(DocumentOp::PointUpdate { returning, .. }) => {
@@ -282,6 +283,12 @@ pub(super) fn inject_returning_spec(
         PhysicalPlan::Document(DocumentOp::UpdateFromJoin { returning, .. }) => {
             *returning = Some(spec);
         }
+        PhysicalPlan::Crdt(CrdtOp::DocUpsert { returning, .. }) => {
+            *returning = Some(spec);
+        }
+        PhysicalPlan::Crdt(CrdtOp::DocDelete { returning, .. }) => {
+            *returning = Some(spec);
+        }
         _ => {}
     }
 }

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

@@ -29,6 +29,13 @@ pub enum PlanKind {
 
 pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
     match plan {
+        PhysicalPlan::Crdt(CrdtOp::DocUpsert {
+            returning: Some(_), ..
+        })
+        | PhysicalPlan::Crdt(CrdtOp::DocDelete {
+            returning: Some(_), ..
+        }) => PlanKind::ReturningRows,
+
         PhysicalPlan::Document(DocumentOp::PointGet { .. })
         | PhysicalPlan::Crdt(CrdtOp::Read { .. })
         | PhysicalPlan::Crdt(CrdtOp::GetPolicy { .. })
@@ -167,7 +174,8 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
 
         // Default: opaque execution result. The specific arms above take
         // precedence; these inner wildcards catch every unmatched op of each
-        // engine plus the engines with no arms here (Crdt, Meta, ClusterArray).
+        // engine (including the remaining `Crdt` ops not covered above) plus
+        // the engines with no arms at all here (Meta, ClusterArray).
         // Exhaustive so a new PhysicalPlan variant forces a decision.
         PhysicalPlan::Document(_)
         | PhysicalPlan::Graph(_)

nodedb/src/control/server/shared/write_admission/predicate/txn_buffering.rs

@@ -1003,11 +1003,13 @@ mod tests {
                 fields_json: "{}".into(),
                 surrogate: Surrogate::ZERO,
                 partial: false,
+                returning: None,
             }),
             PhysicalPlan::Crdt(CrdtOp::DocDelete {
                 collection: "c".into(),
                 document_id: "d".into(),
                 surrogate: Surrogate::ZERO,
+                returning: None,
             }),
         ];
         for p in &plans {

nodedb/src/control/server/wal_dispatch/crdt.rs

@@ -125,6 +125,7 @@ pub(super) fn wal_append_crdt_op(
             fields_json,
             surrogate,
             partial,
+            returning: _,
         } => {
             // Intent-logged like the block-list ops: the Data Plane builds the
             // Loro mutation and the Control Plane has no `LoroDoc`, so the
@@ -144,6 +145,7 @@ pub(super) fn wal_append_crdt_op(
             collection,
             document_id,
             surrogate,
+            returning: _,
         } => {
             let payload = crate::wal::CrdtDocOpWalRecord::Delete {
                 collection: collection.clone(),
@@ -281,6 +283,7 @@ mod tests {
             fields_json: r#"{"a":1}"#.to_string(),
             surrogate: Surrogate::new(3),
             partial: false,
+            returning: None,
         });
 
         let outcome = super::super::wal_append_if_write(

nodedb/src/control/wal_replication/decode/crdt.rs

@@ -131,6 +131,7 @@ pub(super) fn doc_upsert(
         fields_json: fields_json.to_owned(),
         surrogate: nodedb_types::Surrogate::new(surrogate),
         partial,
+        returning: None,
     })
 }
 
@@ -141,6 +142,7 @@ pub(super) fn doc_delete(collection: &str, document_id: &str, surrogate: u32) ->
         collection: collection.to_owned(),
         document_id: document_id.to_owned(),
         surrogate: nodedb_types::Surrogate::new(surrogate),
+        returning: None,
     })
 }