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

@@ -61,6 +61,61 @@ fn classify_on_conflict(ins: &ast::Insert) -> Result<OnConflict> {
     }
 }
 
+/// 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(());
+    }
+    let closed = match engine {
+        EngineType::DocumentSchemaless => columns.len() > 1,
+        EngineType::KeyValue
+        | EngineType::DocumentStrict
+        | EngineType::Columnar
+        | EngineType::Timeseries
+        | EngineType::Spatial => true,
+        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>> {
     // `INSERT ... ON CONFLICT DO UPDATE SET` reroutes to the upsert path
@@ -179,6 +234,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,
@@ -265,6 +327,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,
@@ -363,6 +432,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

@@ -59,6 +59,8 @@ impl ErrorCode {
     pub const DIVISION_BY_ZERO: Self = Self(1204);
     /// A LIMIT/OFFSET/FETCH bound resolved outside `[0, usize::MAX]`.
     pub const INVALID_LIMIT_VALUE: Self = Self(1205);
+    /// A NOT NULL column received an explicit NULL or no value at all.
+    pub const NOT_NULL_VIOLATION: Self = Self(1207);
 
     // Engine ops (1300–1399)
     pub const ARRAY: Self = Self(1300);

nodedb-types/src/error/code_table.rs

@@ -89,6 +89,7 @@ error_code_table! {
     UNDEFINED_FUNCTION => UndefinedFunction { name: 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

@@ -134,6 +134,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

@@ -107,6 +107,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

@@ -161,3 +161,4 @@ pub(super) const TAG_OBJECT_NOT_READY: u16 = 75;
 pub(super) const TAG_NOT_FOUND: u16 = 76;
 pub(super) const TAG_CANNOT_DROP_DEFAULT_DATABASE: u16 = 77;
 pub(super) const TAG_INVALID_LIMIT_VALUE: u16 = 78;
+pub(super) const TAG_NOT_NULL_VIOLATION: u16 = 80;

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

@@ -131,6 +131,10 @@ impl<'a> FromMessagePack<'a> for ErrorDetails {
                 let (name,) = read1_str(reader, field_count)?;
                 Ok(ErrorDetails::UndefinedFunction { name })
             }
+            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

@@ -159,6 +159,9 @@ impl ToMessagePack for ErrorDetails {
             ErrorDetails::UndefinedFunction { name } => {
                 write1(writer, TAG_UNDEFINED_FUNCTION, name)
             }
+            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/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

@@ -50,6 +50,13 @@ pub fn error_to_sqlstate(err: &crate::Error) -> (&'static str, &'static str, Str
             sqlstate::UNDEFINED_FUNCTION,
             format!("function {name}(...) does not exist"),
         ),
+        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

@@ -282,6 +282,12 @@ pub enum Error {
     #[error("function {name}(...) does not exist")]
     UndefinedFunction { name: 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

@@ -148,6 +148,9 @@ pub(crate) fn classify(e: &Error) -> NodeDbError {
         }
         Error::PlanError { detail } => NodeDbError::plan_error(detail),
         Error::UndefinedFunction { name } => NodeDbError::undefined_function(name.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;