fix(sql): reject positional INSERT/UPSERT into KV collections

KV splits key/value by matching column names against pk_col/"key"/"ttl",
so a positional VALUES row with no column list has no principled key to
bind to and would silently produce an empty-keyed, empty-valued row
(all such rows colliding). Reject with a dedicated error instead.

Author: farhan-syah

nodedb-sql/src/error.rs

@@ -49,6 +49,20 @@ pub enum SqlError {
         declared: usize,
     },
 
+    /// A positional `INSERT`/`UPSERT` into a Key-Value collection supplied
+    /// no explicit column list. The KV path splits key/value by matching
+    /// column *names* against `pk_col`/`"key"`/`"ttl"`, so without a column
+    /// list there is no principled key/value split — binding positionally
+    /// would silently write an empty-keyed, empty-valued row (all such rows
+    /// collide). Rejected rather than guessed at, mirroring
+    /// [`InsertColumnArityMismatch`].
+    #[error(
+        "INSERT/UPSERT into KV collection '{collection}': supply an explicit \
+         column list (KV needs named key/value columns; a positional VALUES \
+         row has no key to bind to)"
+    )]
+    PositionalKvInsertUnsupported { collection: String },
+
     /// A descriptor the planner depends on is being drained by
     /// an in-flight DDL. Callers (pgwire handlers) should retry
     /// the whole statement after a short backoff. Propagated

nodedb-sql/src/planner/dml_helpers.rs

@@ -278,6 +278,16 @@ pub(super) fn build_kv_insert_plan(
     on_conflict_updates: Vec<(String, SqlExpr)>,
     pk_col: Option<&str>,
 ) -> Result<Vec<SqlPlan>> {
+    // Positional KV insert (no column list): the key/value split below is
+    // driven entirely by matching column *names* against `key_col_name`/
+    // `"ttl"`. With an empty `columns` list there is no key to bind to, so
+    // every row would silently become an empty-keyed, empty-valued entry
+    // (all colliding). Reject rather than corrupt.
+    if columns.is_empty() {
+        return Err(SqlError::PositionalKvInsertUnsupported {
+            collection: table_name,
+        });
+    }
     let key_col_name = pk_col.unwrap_or("key");
     let key_idx = columns.iter().position(|c| c == key_col_name);
     let ttl_idx = columns.iter().position(|c| c == "ttl");

nodedb-sql/tests/positional_insert_column_binding.rs

@@ -203,29 +203,18 @@ fn positional_upsert_binds_declared_column_names() {
     );
 }
 
-/// KV collections separate key/value by name match against `pk_col`/
-/// `"key"`/`"ttl"`, not by declared column order — this fix intentionally
-/// leaves that path untouched (see judgment call in the PR description).
-///
-/// Pre-existing (out-of-scope) behaviour pinned here: `build_kv_insert_plan`
-/// walks `columns.iter().enumerate()` to build the value map, so an empty
-/// (positional) column list yields an EMPTY value map and an empty-string
-/// key — it never falls back to `col{i}` the way `convert_value_rows` does,
-/// because it isn't built from that helper. That is a real gap for
-/// positional KV inserts, but a separate, wider problem than #202
-/// (there is no principled key/value split to fall back to without a
-/// column list at all); this test only pins that this fix does not change
-/// that existing behaviour.
+/// KV collections split key/value by matching column *names* against
+/// `pk_col`/`"key"`/`"ttl"`, not by declared column order. A positional
+/// insert supplies no column list, so there is no key to bind to —
+/// `build_kv_insert_plan` would otherwise emit an empty-keyed, empty-valued
+/// entry (every such row colliding, all data silently dropped). It is
+/// rejected outright, mirroring the arity-overflow decision.
 #[test]
-fn positional_insert_on_kv_collection_is_unaffected() {
-    let plan = plan_one("INSERT INTO kv_probe VALUES ('k1', 'v1')");
-    match plan {
-        SqlPlan::KvInsert { entries, .. } => {
-            assert_eq!(entries.len(), 1);
-            let (key, value_cols) = &entries[0];
-            assert_eq!(*key, SqlValue::String(String::new()));
-            assert_eq!(value_cols, &Vec::<(String, SqlValue)>::new());
-        }
-        other => panic!("expected a KvInsert plan, got {other:?}"),
-    }
+fn positional_insert_on_kv_collection_is_rejected() {
+    let err = plan_sql("INSERT INTO kv_probe VALUES ('k1', 'v1')", &Catalog)
+        .expect_err("positional KV insert has no key/value column names to bind to");
+    assert!(
+        matches!(err, SqlError::PositionalKvInsertUnsupported { .. }),
+        "expected PositionalKvInsertUnsupported, got {err:?}"
+    );
 }