fix(sql): raise 23502 for NULL primary keys instead of committing them

nodedb-sql/src/error.rs

@@ -22,6 +22,12 @@ pub enum SqlError {
     #[error("unknown column '{column}' in table '{table}'")]
     UnknownColumn { table: String, column: String },
 
+    /// A write supplied NULL (explicitly or by omission) for a declared
+    /// PRIMARY KEY column. PRIMARY KEY implies NOT NULL on every engine;
+    /// PostgreSQL rejects the same write with SQLSTATE `23502`.
+    #[error("null value in column '{column}' violates not-null constraint in table '{table}'")]
+    NotNullViolation { table: String, column: String },
+
     #[error("ambiguous column '{column}' — qualify with table name")]
     AmbiguousColumn { column: String },
 

nodedb-sql/src/planner/dml.rs

@@ -97,6 +97,74 @@ fn classify_on_conflict(ins: &ast::Insert, scope: &TableScope) -> Result<OnConfl
     }
 }
 
+/// PRIMARY KEY implies NOT NULL on every engine.
+///
+/// A row that omits the declared primary-key column or binds it to NULL
+/// must raise `23502` (not_null_violation) rather than commit. On the
+/// document path such a row used to fall through to "fresh surrogate" —
+/// silently minting a new identity for what the caller declared as the row's
+/// key, so several NULL-key rows coexisted and readback of them disagreed
+/// between scan and aggregate paths. Collections whose key is synthetic
+/// (`_rowid`, or a schemaless collection carrying only the auto-injected PK
+/// column) legitimately mint fresh identities and are exempt.
+fn enforce_pk_not_null(
+    engine: EngineType,
+    collection: &str,
+    primary_key: Option<&str>,
+    columns: &[ColumnInfo],
+    rows: &[Vec<(String, SqlValue)>],
+) -> Result<()> {
+    let Some(pk) = primary_key else {
+        return Ok(());
+    };
+    if pk == "_rowid" {
+        return Ok(());
+    }
+    // Closed to the engines #293 names: document_strict and kv reject NULL
+    // keys today, and a declared schemaless column list closes the document
+    // engine. columnar/timeseries/spatial are deliberately NOT gated: their
+    // identity contract mints a surrogate for an omitted key (SERIAL-style
+    // auto identity), so a missing key is legitimate there, not a violation.
+    // Closed to the engines #293 names: document_strict and kv reject NULL
+    // keys today; a schemaless document collection closes exactly when the
+    // user DECLARED its primary key (the auto-injected `id` is synthesized,
+    // `raw_type: None`, and omitting it legitimately mints an auto identity
+    // even when other fields are declared — upstream exercises that with
+    // `CREATE COLLECTION ... FIELDS (...)`). columnar/timeseries/spatial are
+    // deliberately NOT gated: their identity contract and the Data Plane
+    // handle NULL keys themselves.
+    let closed = match engine {
+        EngineType::DocumentSchemaless => columns
+            .iter()
+            .any(|c| c.is_primary_key && c.raw_type.is_some()),
+        EngineType::KeyValue | EngineType::DocumentStrict => true,
+        EngineType::Columnar | EngineType::Timeseries | EngineType::Spatial | EngineType::Array => {
+            false
+        }
+    };
+    if !closed {
+        return Ok(());
+    }
+    let pk_has_default = columns.iter().any(|c| c.name == pk && c.default.is_some());
+    for row in rows {
+        let cell = row.iter().find(|(name, _)| name == pk).map(|(_, v)| v);
+        let missing = cell.is_none();
+        let null = matches!(cell, Some(SqlValue::Null));
+        // A declared DEFAULT materializes at conversion; only absence is
+        // exempt, an explicit NULL still violates the constraint.
+        if missing && pk_has_default {
+            continue;
+        }
+        if missing || null {
+            return Err(SqlError::NotNullViolation {
+                table: collection.to_string(),
+                column: pk.to_string(),
+            });
+        }
+    }
+    Ok(())
+}
+
 /// Plan an INSERT statement.
 pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
     let table_name = match &ins.table {
@@ -215,6 +283,13 @@ pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
         .iter()
         .filter_map(|c| c.raw_type.as_ref().map(|t| (c.name.clone(), t.clone())))
         .collect();
+    enforce_pk_not_null(
+        info.engine,
+        &table_name,
+        info.primary_key.as_deref(),
+        &info.columns,
+        &rows,
+    )?;
     let rules = engine_rules::resolve_engine_rules(info.engine);
     rules.plan_insert(InsertParams {
         collection: table_name,
@@ -297,6 +372,13 @@ pub fn plan_upsert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
         .iter()
         .filter_map(|c| c.raw_type.as_ref().map(|t| (c.name.clone(), t.clone())))
         .collect();
+    enforce_pk_not_null(
+        info.engine,
+        &table_name,
+        info.primary_key.as_deref(),
+        &info.columns,
+        &rows,
+    )?;
     let rules = engine_rules::resolve_engine_rules(info.engine);
     rules.plan_upsert(engine_rules::UpsertParams {
         collection: table_name,
@@ -391,6 +473,13 @@ fn plan_upsert_with_on_conflict(
         .iter()
         .filter_map(|c| c.raw_type.as_ref().map(|t| (c.name.clone(), t.clone())))
         .collect();
+    enforce_pk_not_null(
+        info.engine,
+        &table_name,
+        info.primary_key.as_deref(),
+        &info.columns,
+        &rows,
+    )?;
     let rules = engine_rules::resolve_engine_rules(info.engine);
     rules.plan_upsert(engine_rules::UpsertParams {
         collection: table_name,

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

@@ -100,6 +100,21 @@ pub(crate) fn build_kv_insert_plan(
     check_declared_int_ranges_in_assignments(declared_columns, &on_conflict_updates)?;
     check_declared_float_ranges_in_assignments(declared_columns, &on_conflict_updates)?;
 
+    // PRIMARY KEY implies NOT NULL. A row that omits the key column or
+    // binds it to NULL would commit an empty-keyed entry — unique against
+    // nothing and unreadable — instead of raising. Defaults were already
+    // materialized above, so a present NULL is an explicit violation and
+    // an absent key means the column list never mentioned it.
+    for row in &coerced_rows {
+        let cell = row.iter().find(|(name, _)| name == key_col_name);
+        if cell.is_none() || matches!(cell, Some((_, SqlValue::Null))) {
+            return Err(SqlError::NotNullViolation {
+                table: table_name.clone(),
+                column: key_col_name.to_string(),
+            });
+        }
+    }
+
     let mut entries = Vec::with_capacity(coerced_rows.len());
     let mut ttl_secs: u64 = 0;
     for row in &coerced_rows {

nodedb-types/src/error/code.rs

@@ -63,6 +63,8 @@ impl ErrorCode {
     pub const UNDEFINED_COLUMN: Self = Self(1206);
     /// A bare column name resolves against more than one relation in scope.
     pub const AMBIGUOUS_COLUMN: Self = Self(1207);
+    /// A NOT NULL column received an explicit NULL or no value at all.
+    pub const NOT_NULL_VIOLATION: Self = Self(1208);
 
     // Engine ops (1300–1399)
     pub const ARRAY: Self = Self(1300);

nodedb-types/src/error/code_table.rs

@@ -91,6 +91,7 @@ error_code_table! {
     AMBIGUOUS_COLUMN => AmbiguousColumn { column: String::new() },
     DIVISION_BY_ZERO => DivisionByZero,
     INVALID_LIMIT_VALUE => InvalidLimitValue { clause: "remote".into(), value: message.to_owned() },
+    NOT_NULL_VIOLATION => NotNullViolation { table: "remote".into(), column: message.to_owned() },
 
     // Auth / tenant quota.
     AUTHORIZATION_DENIED => AuthorizationDenied { resource: String::new() },

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

@@ -160,6 +160,22 @@ impl NodeDbError {
         }
     }
 
+    /// A NOT NULL column received an explicit NULL or no value at all.
+    /// Distinct from `constraint_violation` so clients can match on the
+    /// specific code (SQLSTATE `23502`, `not_null_violation`).
+    pub fn not_null_violation(table: impl Into<String>, column: impl Into<String>) -> Self {
+        let table = table.into();
+        let column = column.into();
+        Self {
+            code: ErrorCode::NOT_NULL_VIOLATION,
+            message: format!(
+                "null value in column '{column}' violates not-null constraint in table '{table}'"
+            ),
+            details: ErrorDetails::NotNullViolation { table, column },
+            cause: None,
+        }
+    }
+
     /// Expression evaluation divided or took a modulus by zero. Distinct
     /// from `plan_error` so clients can match on the specific code
     /// (SQLSTATE `22012`, `division_by_zero`) rather than parsing the

nodedb-types/src/error/details.rs

@@ -113,6 +113,9 @@ pub enum ErrorDetails {
     /// A LIMIT/OFFSET/FETCH bound resolved outside `[0, usize::MAX]`.
     #[serde(rename = "invalid_limit_value")]
     InvalidLimitValue { clause: String, value: String },
+    /// A NOT NULL column received an explicit NULL or no value at all.
+    #[serde(rename = "not_null_violation")]
+    NotNullViolation { table: String, column: String },
 
     // Auth
     #[serde(rename = "authorization_denied")]

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

@@ -165,3 +165,4 @@ pub(super) const TAG_CANNOT_DROP_DEFAULT_DATABASE: u16 = 77;
 pub(super) const TAG_INVALID_LIMIT_VALUE: u16 = 78;
 pub(super) const TAG_UNDEFINED_COLUMN: u16 = 79;
 pub(super) const TAG_AMBIGUOUS_COLUMN: u16 = 80;
+pub(super) const TAG_NOT_NULL_VIOLATION: u16 = 81;

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

@@ -139,6 +139,10 @@ impl<'a> FromMessagePack<'a> for ErrorDetails {
                 let (column,) = read1_str(reader, field_count)?;
                 Ok(ErrorDetails::AmbiguousColumn { column })
             }
+            TAG_NOT_NULL_VIOLATION => {
+                let (table, column) = read2_str(reader, field_count)?;
+                Ok(ErrorDetails::NotNullViolation { table, column })
+            }
             TAG_DIVISION_BY_ZERO => {
                 skip_fields(reader, field_count)?;
                 Ok(ErrorDetails::DivisionByZero)

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

@@ -165,6 +165,9 @@ impl ToMessagePack for ErrorDetails {
             ErrorDetails::AmbiguousColumn { column } => {
                 write1(writer, TAG_AMBIGUOUS_COLUMN, column)
             }
+            ErrorDetails::NotNullViolation { table, column } => {
+                write2(writer, TAG_NOT_NULL_VIOLATION, table, column)
+            }
             ErrorDetails::DivisionByZero => write_unit(writer, TAG_DIVISION_BY_ZERO),
             ErrorDetails::InvalidLimitValue { clause, value } => {
                 write2(writer, TAG_INVALID_LIMIT_VALUE, clause, value)

nodedb/src/control/planner/catalog_adapter/type_convert.rs

@@ -61,13 +61,27 @@ pub(super) fn convert_collection_type(
                 .declared_primary_key
                 .clone()
                 .unwrap_or_else(|| "id".to_string());
+            // `raw_type: None` is the documented marker for a synthesized
+            // (auto-injected) primary key. A key the DDL actually declared
+            // carries its declared type token here instead, so the planner
+            // can tell "user-declared identity" from "auto id" — the NULL-pk
+            // write gate (#293) closes only the former.
+            let pk_declared_type = stored
+                .fields
+                .iter()
+                .find(|(n, _)| n.eq_ignore_ascii_case(&pk_name))
+                .and_then(|(_, ts)| {
+                    ts.split_whitespace()
+                        .next()
+                        .map(|t| t.trim_end_matches(',').to_string())
+                });
             let mut columns = vec![ColumnInfo {
                 name: pk_name.clone(),
                 data_type: SqlDataType::String,
                 nullable: false,
                 is_primary_key: true,
                 default: None,
-                raw_type: None,
+                raw_type: pk_declared_type,
                 int_width: None,
                 float_width: None,
             }];

nodedb/src/control/planner/context/query/planning.rs

@@ -42,6 +42,9 @@ fn map_plan_error(error: nodedb_sql::SqlError, tenant_id: crate::types::TenantId
         nodedb_sql::SqlError::UndefinedFunction { name } => {
             crate::Error::UndefinedFunction { name }
         }
+        nodedb_sql::SqlError::NotNullViolation { table, column } => {
+            crate::Error::NotNullViolation { table, column }
+        }
         // A constant expression that divides by zero is the same condition the
         // row-scope evaluator raises, so it carries the same code.
         nodedb_sql::SqlError::DivisionByZero => crate::Error::DivisionByZero,

nodedb/src/control/server/pgwire/types/error_map.rs

@@ -63,6 +63,14 @@ pub fn error_to_sqlstate(err: &crate::Error) -> (&'static str, &'static str, Str
         crate::Error::UnknownStrictField { .. } => {
             ("ERROR", sqlstate::UNDEFINED_COLUMN, err.to_string())
         }
+
+        crate::Error::NotNullViolation { table, column } => (
+            "ERROR",
+            sqlstate::NOT_NULL_VIOLATION,
+            format!(
+                "null value in column '{column}' violates not-null constraint in table '{table}'"
+            ),
+        ),
         crate::Error::DivisionByZero => ("ERROR", sqlstate::DIVISION_BY_ZERO, err.to_string()),
         crate::Error::InvalidLimitValue { .. } => {
             ("ERROR", sqlstate::INVALID_LIMIT_VALUE, err.to_string())

nodedb/src/error/types.rs

@@ -301,6 +301,12 @@ pub enum Error {
     #[error("column \"{column}\" of collection \"{collection}\" does not exist")]
     UnknownStrictField { collection: String, column: String },
 
+    /// A NOT NULL column received an explicit NULL or no value at all.
+    /// Propagated from the INSERT/UPSERT planner; the pgwire layer renders
+    /// this as SQLSTATE `23502` (not_null_violation).
+    #[error("null value in column '{column}' violates not-null constraint in table '{table}'")]
+    NotNullViolation { table: String, column: String },
+
     /// Expression evaluation divided or took a modulus by zero. Rendered as
     /// SQLSTATE `22012` (division_by_zero) at the pgwire layer.
     #[error("division by zero")]

nodedb/src/error_classify.rs

@@ -151,6 +151,10 @@ pub(crate) fn classify(e: &Error) -> NodeDbError {
         Error::UndefinedColumn { column } => NodeDbError::undefined_column(column.clone()),
         Error::AmbiguousColumn { column } => NodeDbError::ambiguous_column(column.clone()),
         Error::UnknownStrictField { column, .. } => NodeDbError::undefined_column(column.clone()),
+
+        Error::NotNullViolation { table, column } => {
+            NodeDbError::not_null_violation(table.clone(), column.clone())
+        }
         Error::DivisionByZero => NodeDbError::division_by_zero(),
         Error::InvalidLimitValue { clause, value } => {
             NodeDbError::invalid_limit_value(*clause, value.clone())

nodedb/tests/wire/cases/mod.rs

@@ -104,6 +104,7 @@ mod merge_insert_renamed_source_column;
 mod merge_insert_surrogate_stability;
 mod move_tenant_idempotent;
 mod move_tenant_round_trip;
+mod not_null_pk_23502;
 mod object_literal_dml_row_level_security;
 mod object_literal_trailing_clause;
 mod pg_catalog_oid_stability;