Enforce PRIMARY KEY NOT NULL and give every read path one row image

nodedb-cluster/src/rpc_codec/execute/types.rs

@@ -89,6 +89,16 @@ pub enum TypedClusterError {
     DataPlane {
         code: DataPlaneErrorCode,
     },
+    /// A Control-Plane constraint refusal (`crate::Error::RejectedConstraint`
+    /// on the executing node), carried verbatim so the coordinator renders
+    /// the same SQLSTATE (23502 vs 23505) local execution would. Without
+    /// this, `constraint` collapsed into `Internal`'s bare numeric code and
+    /// a NOT NULL refusal on a remote shard read back as unique_violation.
+    RejectedConstraint {
+        collection: String,
+        constraint: String,
+        detail: String,
+    },
 }
 
 /// One streamed chunk of an `ExecuteStreamRequest` result.

nodedb-physical/src/physical_plan/document/op.rs

@@ -160,6 +160,11 @@ pub enum DocumentOp {
         /// See `PointPut::resolved_sum_targets`.
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// The collection's declared `PRIMARY KEY` column, `Some` only for a
+        /// schemaless collection. The Data Plane refuses a post-image whose
+        /// value at this field is absent or JSON null.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Full collection scan with filtering, sorting, and pagination.
@@ -406,6 +411,9 @@ pub enum DocumentOp {
         /// covering both sides of a join-key change.
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// See `PointUpdate::declared_primary_key`.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Bulk update: scan + apply field updates to all matches.
@@ -435,6 +443,9 @@ pub enum DocumentOp {
         /// covering both sides of a join-key change.
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// See `PointUpdate::declared_primary_key`.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Bulk delete: scan + delete all matches.
@@ -502,6 +513,9 @@ pub enum DocumentOp {
         /// applies the difference (both sides on a join-key rewrite).
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// See `PointUpdate::declared_primary_key`.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Cursor-paginated scan for the clone materializer. Returns

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

@@ -108,7 +108,7 @@ pub(crate) fn build_kv_insert_plan(
         // position in the statement's column list.
         let key_val = match row.iter().find(|(name, _)| name == key_col_name) {
             Some((_, value)) => value.clone(),
-            None => SqlValue::String(String::new()),
+            None => SqlValue::Null,
         };
         if let Some((_, value)) = row.iter().find(|(name, _)| name == "ttl") {
             match value {

nodedb-types/src/error/code_table.rs

@@ -50,7 +50,7 @@ error_code_table! {
     msg = message;
 
     // Write path.
-    CONSTRAINT_VIOLATION => ConstraintViolation { collection: String::new() },
+    CONSTRAINT_VIOLATION => ConstraintViolation { collection: String::new(), constraint: String::new() },
     WRITE_CONFLICT => WriteConflict { collection: String::new(), document_id: String::new() },
     DEADLINE_EXCEEDED => DeadlineExceeded,
     PREVALIDATION_REJECTED => PrevalidationRejected { constraint: String::new() },

nodedb-types/src/error/ctors/write_path.rs

@@ -9,12 +9,24 @@ use super::super::details::ErrorDetails;
 use super::super::types::NodeDbError;
 
 impl NodeDbError {
-    pub fn constraint_violation(collection: impl Into<String>, detail: impl fmt::Display) -> Self {
+    /// `constraint` names the constraint kind (`"not_null"`, `"unique"`,
+    /// ...) so every hop — local or reconstructed from a remote node — can
+    /// pick the right SQLSTATE instead of guessing unique_violation for
+    /// everything.
+    pub fn constraint_violation(
+        collection: impl Into<String>,
+        constraint: impl Into<String>,
+        detail: impl fmt::Display,
+    ) -> Self {
         let collection = collection.into();
+        let constraint = constraint.into();
         Self {
             code: ErrorCode::CONSTRAINT_VIOLATION,
-            message: format!("constraint violation on {collection}: {detail}"),
-            details: ErrorDetails::ConstraintViolation { collection },
+            message: format!("constraint violation on {collection} ({constraint}): {detail}"),
+            details: ErrorDetails::ConstraintViolation {
+                collection,
+                constraint,
+            },
             cause: None,
         }
     }

nodedb-types/src/error/details.rs

@@ -15,7 +15,14 @@ use serde::{Deserialize, Serialize};
 pub enum ErrorDetails {
     // Write path
     #[serde(rename = "constraint_violation")]
-    ConstraintViolation { collection: String },
+    ConstraintViolation {
+        collection: String,
+        /// The constraint kind (`"not_null"`, `"unique"`, ...). Drives
+        /// SQLSTATE selection (23502 vs 23505) on every hop, local or remote.
+        /// Named `constraint`, not `kind`, because the enum's own
+        /// `#[serde(tag = "kind")]` already owns that JSON key.
+        constraint: String,
+    },
     #[serde(rename = "write_conflict")]
     WriteConflict {
         collection: String,

nodedb-types/src/error/mod.rs

@@ -14,8 +14,8 @@
 //! ```json
 //! {
 //!   "code": "NDB-1000",
-//!   "message": "constraint violation on users: duplicate email",
-//!   "details": { "kind": "constraint_violation", "collection": "users" }
+//!   "message": "constraint violation on users (unique): duplicate email",
+//!   "details": { "kind": "constraint_violation", "collection": "users", "constraint": "unique" }
 //! }
 //! ```
 //!

nodedb-types/src/error/msgpack/decode/from_messagepack.rs

@@ -18,8 +18,11 @@ impl<'a> FromMessagePack<'a> for ErrorDetails {
         let (tag, field_count) = read_header(reader)?;
         match tag {
             TAG_CONSTRAINT_VIOLATION => {
-                let (collection,) = read1_str(reader, field_count)?;
-                Ok(ErrorDetails::ConstraintViolation { collection })
+                let (collection, constraint) = read2_str(reader, field_count)?;
+                Ok(ErrorDetails::ConstraintViolation {
+                    collection,
+                    constraint,
+                })
             }
             TAG_WRITE_CONFLICT => {
                 let (collection, document_id) = read2_str(reader, field_count)?;
@@ -650,9 +653,6 @@ mod tests {
     #[test]
     fn single_string_field_roundtrip() {
         let variants = vec![
-            ErrorDetails::ConstraintViolation {
-                collection: "orders".into(),
-            },
             ErrorDetails::AppendOnlyViolation {
                 collection: "ledger".into(),
             },
@@ -705,6 +705,12 @@ mod tests {
             document_id: "u-99".into(),
         };
         assert_eq!(roundtrip(&v2), v2);
+
+        let v3 = ErrorDetails::ConstraintViolation {
+            collection: "orders".into(),
+            constraint: "not_null".into(),
+        };
+        assert_eq!(roundtrip(&v3), v3);
     }
 
     #[test]

nodedb-types/src/error/msgpack/encode.rs

@@ -81,9 +81,10 @@ where
 impl ToMessagePack for ErrorDetails {
     fn write<W: Write>(&self, writer: &mut W) -> zerompk::Result<()> {
         match self {
-            ErrorDetails::ConstraintViolation { collection } => {
-                write1(writer, TAG_CONSTRAINT_VIOLATION, collection)
-            }
+            ErrorDetails::ConstraintViolation {
+                collection,
+                constraint,
+            } => write2(writer, TAG_CONSTRAINT_VIOLATION, collection, constraint),
             ErrorDetails::WriteConflict {
                 collection,
                 document_id,

nodedb-types/src/error/types.rs

@@ -180,7 +180,7 @@ mod tests {
 
     #[test]
     fn error_display_includes_code() {
-        let e = NodeDbError::constraint_violation("users", "duplicate email");
+        let e = NodeDbError::constraint_violation("users", "unique", "duplicate email");
         let msg = e.to_string();
         assert!(msg.contains("NDB-1000"));
         assert!(msg.contains("constraint violation"));

nodedb/src/control/backup/orchestrator.rs

@@ -385,5 +385,16 @@ fn map_typed_error(err: TypedClusterError, node_id: u64) -> Error {
         // Keep the shard's verdict typed: a backup snapshot refused by the
         // Data Plane must not read as a generic internal backup fault.
         TypedClusterError::DataPlane { code } => Error::DataPlane(code.into()),
+        // A constraint verdict keeps its collection and kind, so the client
+        // reads the SQLSTATE the refusing shard meant.
+        TypedClusterError::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        } => Error::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        },
     }
 }

nodedb/src/control/backup/restore/remote.rs

@@ -81,5 +81,16 @@ pub(super) fn map_typed_error(err: TypedClusterError, node_id: u64) -> Error {
         // Keep the shard's verdict typed: a restore refused by the Data Plane
         // must not read as a generic internal restore fault.
         TypedClusterError::DataPlane { code } => Error::DataPlane(code.into()),
+        // A constraint verdict keeps its collection and kind, so the client
+        // reads the SQLSTATE the refusing shard meant.
+        TypedClusterError::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        } => Error::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        },
     }
 }

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

@@ -674,6 +674,7 @@ mod tests {
             rls_filters: Vec::new(),
             rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
             resolved_sum_targets: Vec::new(),
+            declared_primary_key: None,
         });
         assert!(matches!(plan_vshard(&plan), PlanRouting::Unroutable(_)));
     }
@@ -704,6 +705,7 @@ mod tests {
             rls_filters: Vec::new(),
             rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
             resolved_sum_targets: Vec::new(),
+            declared_primary_key: None,
         });
         assert!(matches!(plan_vshard(&plan), PlanRouting::Unroutable(_)));
     }

nodedb/src/control/cluster/data_plane_error_wire.rs

@@ -30,6 +30,18 @@ pub(crate) fn execution_error_to_typed(err: crate::Error) -> TypedClusterError {
         crate::Error::DeadlineExceeded { .. } => {
             TypedClusterError::DeadlineExceeded { elapsed_ms: 0 }
         }
+        // A Control-Plane constraint refusal crosses verbatim, same as a
+        // Data-Plane verdict, so the coordinator answers 23502 vs 23505
+        // instead of flattening both into one numeric class.
+        crate::Error::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        } => TypedClusterError::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        },
         other => {
             let message = other.to_string();
             let code = u32::from(nodedb_types::error::NodeDbError::from(other).code().0);

nodedb/src/control/gateway/dispatcher.rs

@@ -372,6 +372,17 @@ pub(super) fn map_typed_cluster_error(err: TypedClusterError, vshard_id: u64) ->
         // Remote Data-Plane verdict: keep the code so the client sees the
         // SQLSTATE local execution renders, not a generic internal error.
         TypedClusterError::DataPlane { code } => Error::DataPlane(code.into()),
+        // Remote constraint refusal: keep the kind so the client sees 23502
+        // vs 23505, exactly as a local refusal on this node would render.
+        TypedClusterError::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        } => Error::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        },
         TypedClusterError::Internal { message, .. } => Error::Internal { detail: message },
     }
 }

nodedb/src/control/insert_select/copy_rows.rs

@@ -5,10 +5,11 @@
 //! source page-by-page, apply the residual `WHERE`, assign a fresh
 //! target-keyed surrogate, emit `(target_doc_id, value, surrogate)`.
 //!
-//! Every step here assumes standard msgpack bodies — `MaterializeScan`
-//! already normalized a strict source's Binary Tuple on the Data Plane.
-//! Never re-add a Control-Plane decode here; it would silently corrupt the
-//! filter, PK extraction, and target write.
+//! Every step here assumes standard msgpack bodies carrying an `id` field —
+//! `MaterializeScan` already normalized a strict source's Binary Tuple and
+//! injected the row's storage-key identity on the Data Plane. Never re-add
+//! a Control-Plane decode here; it would silently corrupt the filter, PK
+//! extraction, and target write.
 
 use nodedb_types::{DatabaseId, Surrogate, TenantId};
 

nodedb/src/control/merge_orchestrator/expand_staged_merge.rs

@@ -123,6 +123,7 @@ async fn resolve_merge_arms(
         source_join_col,
         clauses,
         rls_write_check,
+        declared_primary_key,
         ..
     }) = &task.plan
     else {
@@ -163,6 +164,7 @@ async fn resolve_merge_arms(
             // Writes nothing, so folds no sum delta; the emitted point ops
             // carry their own resolution.
             resolved_sum_targets: Vec::new(),
+            declared_primary_key: declared_primary_key.clone(),
         })));
     // Passing `txn_id` lets the RESOLVE pass fold TARGET's staging overlay,
     // so a MERGE reuses a prior statement's row instead of duplicating it.

nodedb/src/control/merge_orchestrator/orchestrator.rs

@@ -56,6 +56,9 @@ pub struct MergeArgs<'a> {
     /// RLS write predicate, carried onto the apply pass which decides every
     /// arm's image against it. Separate from `rls_filters`: read vs write gate.
     pub rls_write_check: &'a nodedb_types::RlsWriteCheck,
+    /// Declared `PRIMARY KEY` column of the target, `None` for none declared.
+    /// Carried on both passes so the Data Plane's UPDATE-arm guard runs.
+    pub declared_primary_key: Option<&'a str>,
 }
 
 /// Consume an authorized autocommit `MERGE` at the orchestration boundary.
@@ -79,6 +82,7 @@ pub async fn run_authorized_merge(
         // Unresolved on the way in: the orchestrator's own RESOLVE pass is what
         // produces the join keys this is filled from.
         resolved_sum_targets: _,
+        declared_primary_key,
     }) = task.plan
     else {
         return Err(crate::Error::BadRequest {
@@ -99,6 +103,7 @@ pub async fn run_authorized_merge(
             returning: returning.as_ref(),
             rls_filters: &rls_filters,
             rls_write_check: &rls_write_check,
+            declared_primary_key: declared_primary_key.as_deref(),
         },
     )
     .await
@@ -274,6 +279,7 @@ fn merge_plan(
         rls_write_check: args.rls_write_check.clone(),
         // Empty on RESOLVE (writes nothing); APPLY carries the resolution.
         resolved_sum_targets,
+        declared_primary_key: args.declared_primary_key.map(str::to_string),
     };
     if resolve_only {
         PhysicalPlan::Document(DocumentOp::ResolveWrite(Box::new(merge)))

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

@@ -288,6 +288,7 @@ mod tests {
                 rls_filters: vec![],
                 rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
                 resolved_sum_targets: Vec::new(),
+                declared_primary_key: None,
             }),
             post_set_op: PostSetOp::None,
             txn_id: None,

nodedb/src/control/planner/calvin/submit/routed.rs

@@ -150,6 +150,22 @@ pub async fn submit_calvin_routed(
             error: Some(TypedClusterError::DataPlane { code }),
             ..
         })) => Err(Error::DataPlane(code.into())),
+        // A constraint refusal on the sequencer leader keeps its kind, so a
+        // NOT NULL refusal on a routed write reaches the client as 23502
+        // instead of collapsing into a generic internal error.
+        Ok(RaftRpc::SubmitCalvinTxnResponse(SubmitCalvinTxnResponse {
+            error:
+                Some(TypedClusterError::RejectedConstraint {
+                    collection,
+                    constraint,
+                    detail,
+                }),
+            ..
+        })) => Err(Error::RejectedConstraint {
+            collection,
+            constraint,
+            detail,
+        }),
         Ok(RaftRpc::SubmitCalvinTxnResponse(SubmitCalvinTxnResponse {
             error: Some(e), ..
         })) => Err(Error::Internal {

nodedb/src/control/planner/materialized_sum/stored.rs

@@ -354,6 +354,7 @@ mod tests {
             rls_filters: Vec::new(),
             rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
             resolved_sum_targets: Vec::new(),
+            declared_primary_key: None,
         }
     }