Type row identity: StorageKey internal, RowIdentity client-visible

nodedb-cluster-tests/tests/common_suite/cases/cross_node_pk_lookup.rs

@@ -8,8 +8,8 @@
 //! (`surrogate_pk{,_rev}_v3`) is SHARDED to the collection's data-group
 //! members. `document_strict` collections are single-vShard-homed, so when the
 //! coordinator is NOT a member of that group, resolution misses → the
-//! coordinator ships `Surrogate::ZERO` to the owner → the owner does
-//! `surrogate_to_doc_id(ZERO)` → the row is NOT FOUND. So cross-node PK reads
+//! coordinator ships `Surrogate::ZERO` to the owner → the owner renders
+//! `StorageKey::for_surrogate(ZERO)` → the row is NOT FOUND. So cross-node PK reads
 //! from a non-member coordinator silently returned EMPTY.
 //!
 //! Scans are unaffected: they route + scan on the owner with no surrogate

nodedb-cluster-tests/tests/common_suite/cases/vector_index_dispatch_cross_node.rs

@@ -288,11 +288,14 @@ async fn search_probe(node: &TestClusterNode, query: Vec<f32>) -> Result<Vec<u32
     let Ok(Value::Array(items)) = nodedb_types::value_from_msgpack(payload) else {
         return Err(format!("undecodable hits payload {payload:02x?}"));
     };
+    // A bound vector hit carries its storage key as `id`.
     Ok(items
         .iter()
         .filter_map(|hit| match hit {
             Value::Object(map) => match map.get("id") {
-                Some(Value::Integer(id)) => Some(*id as u32),
+                Some(Value::String(key)) => {
+                    nodedb_types::StorageKey::parse(key).map(|k| k.surrogate().as_u32())
+                }
                 _ => None,
             },
             _ => None,

nodedb-cluster-tests/tests/misc_suite/cases/cluster_triggers.rs

@@ -181,7 +181,7 @@ fn event_source_preserved_through_write_event() {
         database_id: nodedb::types::DatabaseId::DEFAULT,
         collection: Arc::from("orders"),
         op: WriteOp::Insert,
-        row_id: RowId::new("doc-1"),
+        row_id: RowId::row(nodedb_types::RowIdentity::from_user_key("doc-1")),
         lsn: Lsn::new(100),
         tenant_id: TenantId::new(1),
         vshard_id: VShardId::new(0),

nodedb-cluster/src/rpc_codec/data_plane_error.rs

@@ -112,4 +112,13 @@ pub enum DataPlaneErrorCode {
         limit: u64,
     },
     DivisionByZero,
+    /// A period-lock reference row exists but does not carry the
+    /// configured `status_column` — a misconfigured column name, not a
+    /// locked period.
+    PeriodLockMisconfigured {
+        collection: String,
+        ref_table: String,
+        status_column: String,
+        row_identity: String,
+    },
 }

nodedb-graph/src/csr/index/interning.rs

@@ -130,7 +130,7 @@ impl CsrIndex {
     ///
     /// A graph node and its same-pk document share one global surrogate, so this
     /// is the bridge from a MATCH binding's node name to the document storage key
-    /// (`surrogate_to_doc_id`) used to fetch the node's properties.
+    /// (`StorageKey::for_surrogate`) used to fetch the node's properties.
     pub fn node_surrogate(&self, node: &str) -> Option<nodedb_types::Surrogate> {
         let &local_id = self.node_to_id.get(node)?;
         let raw = self.node_surrogate_raw(local_id);

nodedb-physical/src/physical_plan/columnar.rs

@@ -42,6 +42,11 @@ pub enum ColumnarInsertIntent {
     /// assignments (with `EXCLUDED.col` bound to the incoming row), and
     /// writes the merged result.
     Put,
+    /// Plain `INSERT` on a collection whose `PRIMARY KEY` is declared on a
+    /// natural key column (not `id` / `document_id`). Duplicate PK refuses
+    /// the row with `RejectedConstraint` (SQLSTATE 23505) instead of the
+    /// `Insert` tombstone-and-append behavior above.
+    InsertUnique,
 }
 
 /// Base columnar physical operations.

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

@@ -328,6 +328,10 @@ pub enum DocumentOp {
         /// divergence, before writing.
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// See `PointUpdate::declared_primary_key`. Names the column each
+        /// removed row's identity is read from for its event and redo entry.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Estimate count via HLL cardinality stats.
@@ -473,6 +477,10 @@ pub enum DocumentOp {
         /// See `PointPut::resolved_sum_targets`.
         #[serde(default)]
         resolved_sum_targets: Vec<ResolvedSumTarget>,
+        /// See `PointUpdate::declared_primary_key`. Names the column each
+        /// removed row's identity is read from when staged.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// MERGE: join-based multi-action DML (INSERT/UPDATE/DELETE per WHEN
@@ -557,6 +565,10 @@ pub enum DocumentOp {
         join_column: String,
         /// Join value that resolved to `surrogate`.
         join_value: String,
+        /// The TARGET collection's declared `PRIMARY KEY` column, when it
+        /// has one. Names the target row in its event and redo entry.
+        #[serde(default)]
+        declared_primary_key: Option<String>,
     },
 
     /// Read-only resolve pass over the wrapped write op: runs its full

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

@@ -150,6 +150,11 @@ pub struct MaterializedSumBinding {
     pub join_column: String,
     /// Expression evaluated against the source INSERT row to compute the delta.
     pub value_expr: nodedb_query::expr::SqlExpr,
+    /// The TARGET collection's declared `PRIMARY KEY` column, when it has
+    /// one. Resolved from the catalog at plan time. Names the target row in
+    /// the event and redo entry its balance write produces.
+    #[serde(default)]
+    pub declared_primary_key: Option<String>,
 }
 
 /// Period lock configuration propagated to Data Plane.

nodedb-physical/src/physical_plan/meta.rs

@@ -71,10 +71,17 @@ pub enum MetaOp {
     ///
     /// `target_type`: "document_schemaless", "document_strict", "kv".
     /// `schema_json`: for "document_strict"/"kv", JSON-serialized column definitions.
+    /// `source_storage_mode`: the collection's storage mode BEFORE this
+    /// conversion, read from the catalog by the Control Plane dispatcher.
+    /// The Data Plane's own `doc_configs` cache still reflects the OLD mode
+    /// at dispatch time (the catalog flip and re-register happen after this
+    /// op returns), so the handler cannot resolve the source format from
+    /// that cache. It must take it from the plan instead.
     ConvertCollection {
         collection: QualifiedCollection,
         target_type: String,
         schema_json: String,
+        source_storage_mode: super::document::StorageMode,
     },
 
     /// Snapshot a tenant's data from the sparse engine.

nodedb-physical/src/surrogate.rs

@@ -51,18 +51,19 @@ pub trait SurrogateAssigner: Send + Sync {
 
     /// Allocate a FRESH, never-before-issued surrogate for a row that has no
     /// content primary key — i.e. a collection whose primary key is the
-    /// auto-generated `_rowid` (no `PRIMARY KEY` was declared at CREATE). Each
-    /// call returns a new value; there is no `pk_bytes` to content-address on,
-    /// so repeated calls do NOT collapse to the same surrogate (which is
-    /// exactly the bug that content-addressing an empty key would cause).
+    /// auto-generated `_rowid` (no `PRIMARY KEY` was declared at CREATE), or
+    /// a timeseries row.
     ///
-    /// The Data Plane sets the row's `_rowid` equal to this surrogate, so
-    /// implementations should bind the surrogate to its own value for reverse
-    /// `_rowid = N` point lookups.
+    /// Every call allocates a new value. There is no `pk_bytes` to
+    /// content-address on, so repeated calls never collapse onto one
+    /// surrogate.
+    ///
+    /// The returned `String` is the bound identity. The caller uses it
+    /// verbatim and never re-derives it.
     fn assign_fresh(
         &self,
         database_id: DatabaseId,
         tenant_id: TenantId,
         collection: &str,
-    ) -> Result<Surrogate, SurrogateAssignError>;
+    ) -> Result<(Surrogate, String), SurrogateAssignError>;
 }