From 083ddaf0f4946f6077d19f4e2479305f66cbd6a7 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:49 +0800 Subject: [PATCH 01/14] fix(bootstrap): make replay timeout configurable The data-group replay gate aborted after a hardcoded 60s. A restart whose last log entry is an election no-op can sit one index short of commit_index for that whole budget, then fail startup. - add [server] data_group_recovery_timeout_ms, default 60000 - pass it through await_cluster_ready into await_data_group_recovery --- nodedb/src/bootstrap/cluster_ready.rs | 8 ++++++- nodedb/src/bootstrap/data_group_recovery.rs | 19 ++++++++++------- nodedb/src/config/server/section.rs | 23 +++++++++++++++++++++ nodedb/src/main.rs | 1 + 4 files changed, 42 insertions(+), 9 deletions(-) diff --git a/nodedb/src/bootstrap/cluster_ready.rs b/nodedb/src/bootstrap/cluster_ready.rs index 731445079..d0327be0d 100644 --- a/nodedb/src/bootstrap/cluster_ready.rs +++ b/nodedb/src/bootstrap/cluster_ready.rs @@ -33,6 +33,7 @@ pub async fn await_cluster_ready( raft_ready_rx: Option>, data_plane_replay_done: Vec>, gates: ClusterReadyGates, + data_group_recovery_timeout: Duration, ) -> anyhow::Result<()> { let ClusterReadyGates { raft_gate, @@ -143,7 +144,12 @@ pub async fn await_cluster_ready( // elections, so without this wait the gateway can open while a data // group's engines are still empty and an acknowledged write reads back as // if it never happened. Fail closed, like the replay wait above. - if let Err(e) = crate::bootstrap::data_group_recovery::await_data_group_recovery(shared).await { + if let Err(e) = crate::bootstrap::data_group_recovery::await_data_group_recovery( + shared, + data_group_recovery_timeout, + ) + .await + { data_groups_gate.fail(format!("data raft group recovery failed: {e}")); return Err(e); } diff --git a/nodedb/src/bootstrap/data_group_recovery.rs b/nodedb/src/bootstrap/data_group_recovery.rs index b050d30bd..8f4885fdd 100644 --- a/nodedb/src/bootstrap/data_group_recovery.rs +++ b/nodedb/src/bootstrap/data_group_recovery.rs @@ -31,11 +31,6 @@ use crate::control::state::SharedState; /// seconds, so a coarse poll costs nothing and avoids a busy loop. const POLL_INTERVAL: Duration = Duration::from_millis(50); -/// Upper bound on the whole wait. Generous relative to a randomized election -/// timeout plus replay of a retained log, but finite: a group that cannot elect -/// or cannot apply is a failure, not a reason to hang forever. -pub const DATA_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(60); - /// True when `group_id` names a data group whose log carries user writes that /// must be replayed into the Data Plane before queries are served. /// @@ -172,12 +167,20 @@ fn pending_groups(statuses: Vec) -> Vec) -> anyhow::Result<()> { +/// +/// `timeout` bounds the whole wait: generous relative to a randomized election +/// timeout plus replay of a retained log, but finite — a group that cannot +/// elect or cannot apply is a failure, not a reason to hang forever. The +/// caller supplies it from `[server] data_group_recovery_timeout_ms`. +pub async fn await_data_group_recovery( + shared: &Arc, + timeout: Duration, +) -> anyhow::Result<()> { let Some(status_fn) = shared.raft_status_fn.get() else { return Ok(()); }; let status_fn = Arc::clone(status_fn); - let deadline = Instant::now() + DATA_GROUP_RECOVERY_TIMEOUT; + let deadline = Instant::now() + timeout; loop { let pending = pending_groups(status_fn()); @@ -193,7 +196,7 @@ pub async fn await_data_group_recovery(shared: &Arc) -> anyhow::Res .collect::>() .join("; "); return Err(anyhow::anyhow!( - "data raft group recovery timeout after {DATA_GROUP_RECOVERY_TIMEOUT:?}: {detail}" + "data raft group recovery timeout after {timeout:?}: {detail}" )); } diff --git a/nodedb/src/config/server/section.rs b/nodedb/src/config/server/section.rs index 0bf7ab602..c05b02825 100644 --- a/nodedb/src/config/server/section.rs +++ b/nodedb/src/config/server/section.rs @@ -66,6 +66,20 @@ pub struct ServerSection { #[serde(default = "default_max_connections")] pub max_connections: usize, + /// Startup budget, in milliseconds, for the data-group replay gate. + /// + /// After a restart every locally hosted data Raft group must re-deliver + /// its retained log before the client gateway opens. A group that has not + /// caught up within this budget fails the boot with a `StartupError`. + /// + /// A post-restart leadership race can hold a group one index short of its + /// commit index (a trailing election no-op delivered while a proposer + /// waits) until a fresh boot converges it. Raise this on loaded or + /// large-log deployments so a slow-but-sound replay is not aborted at the + /// default. Default: 60000 (60s). + #[serde(default = "default_data_group_recovery_timeout_ms")] + pub data_group_recovery_timeout_ms: u64, + /// Log output format: `"text"` (default, human-readable) or `"json"` (structured). /// Unknown values are rejected at startup — there is no silent fallback. #[serde(default)] @@ -125,6 +139,7 @@ impl Default for ServerSection { data_plane_cores: default_data_plane_cores(), memory_limit: default_memory_limit(), max_connections: default_max_connections(), + data_group_recovery_timeout_ms: default_data_group_recovery_timeout_ms(), log_format: LogFormat::Text, tls: None, single_node_calvin: default_single_node_calvin(), @@ -158,6 +173,13 @@ fn default_max_connections() -> usize { 4096 } +/// Default for [`ServerSection::data_group_recovery_timeout_ms`]: 60s, the +/// budget the recovery gate has always used. Operators raise it when a +/// post-restart replay on a loaded box needs longer to converge. +fn default_data_group_recovery_timeout_ms() -> u64 { + 60_000 +} + /// Deserializer for `memory_limit` that accepts either a raw byte count /// (`memory_limit = 4294967296`) or a human-readable string /// (`memory_limit = "4GiB"`). Suffixes: `K/KiB`, `M/MiB`, `G/GiB`, `T/TiB`, @@ -208,6 +230,7 @@ mod tests { assert_eq!(s.memory_limit, 1024 * 1024 * 1024); assert!(s.data_plane_cores >= 1); assert_eq!(s.max_connections, 4096); + assert_eq!(s.data_group_recovery_timeout_ms, 60_000); assert_eq!(s.log_format, LogFormat::Text); } diff --git a/nodedb/src/main.rs b/nodedb/src/main.rs index eefc2c946..91e009f4e 100644 --- a/nodedb/src/main.rs +++ b/nodedb/src/main.rs @@ -274,6 +274,7 @@ async fn server_main() -> anyhow::Result<()> { health_loop_gate, gateway_enable_gate, }, + std::time::Duration::from_millis(config.server.data_group_recovery_timeout_ms), ) .await?; From e18713c7cdb42057c0123fc1cfd77974a012e25a Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:49 +0800 Subject: [PATCH 02/14] fix(pgwire): batch multi-row INSERT into one page Each value tuple became its own PointInsert task. One statement emitted INSERT 0 1 per row, and drivers read rowcount 1. - accumulate plain rows in insert/convert.rs and emit one BatchInsert - keep per-row tasks for ON CONFLICT DO NOTHING and CRDT rows --- .../sql_plan_convert/dml/insert/convert.rs | 125 ++++++++++++++---- 1 file changed, 97 insertions(+), 28 deletions(-) diff --git a/nodedb/src/control/planner/sql_plan_convert/dml/insert/convert.rs b/nodedb/src/control/planner/sql_plan_convert/dml/insert/convert.rs index dc4a3b110..4a2a9d432 100644 --- a/nodedb/src/control/planner/sql_plan_convert/dml/insert/convert.rs +++ b/nodedb/src/control/planner/sql_plan_convert/dml/insert/convert.rs @@ -89,6 +89,14 @@ pub(in super::super::super) fn convert_insert( let mut balanced_documents: Vec<(String, Vec)> = Vec::new(); let mut balanced_surrogates: Vec = Vec::new(); + // Rows of a plain multi-row INSERT, accumulated across the loop and + // emitted as ONE `BatchInsert` page. One statement answers one + // CommandComplete with the row count, and its rows commit or fail + // together. `ON CONFLICT DO NOTHING` and CRDT rows stay per-row: their + // outcome is decided per row and a page cannot express the skip. + let mut batch_documents: Vec<(String, Vec)> = Vec::new(); + let mut batch_surrogates: Vec = Vec::new(); + // Every engine's rows expand their DEFAULTs here, ahead of identity // derivation. A DEFAULT materialized after the primary-key NOT NULL gate // refuses a key the declaration supplies. @@ -124,45 +132,106 @@ pub(in super::super::super) fn convert_insert( balanced_surrogates.push(surrogate); continue; } - let plan = if is_crdt { - PhysicalPlan::Crdt(CrdtOp::DocUpsert { - collection: qualified_collection.clone(), - document_id: doc_id, - fields_json: super::super::crdt_gate::row_to_fields_json(row)?, - surrogate, - partial: false, - returning: None, - rls_filters: Vec::new(), - }) - } else { - PhysicalPlan::Document(DocumentOp::PointInsert { + if is_crdt { + tasks.push(PhysicalTask { + tenant_id, + vshard_id: vshard, + database_id: ctx.database_id, + plan: PhysicalPlan::Crdt(CrdtOp::DocUpsert { + collection: qualified_collection.clone(), + document_id: doc_id, + fields_json: super::super::crdt_gate::row_to_fields_json(row)?, + surrogate, + partial: false, + returning: None, + rls_filters: Vec::new(), + }), + post_set_op: PostSetOp::None, + txn_id: None, + }); + continue; + } + if if_absent { + // `ON CONFLICT DO NOTHING` decides per row which rows + // already exist, so the rows cannot share one page. + tasks.push(PhysicalTask { + tenant_id, + vshard_id: vshard, + database_id: ctx.database_id, + plan: PhysicalPlan::Document(DocumentOp::PointInsert { + collection: qualified_collection.clone(), + document_id: doc_id, + value: value_bytes, + if_absent, + surrogate, + // Both filled in after conversion: the RETURNING + // spec by the protocol layer's injection pass, the + // read filter by the RLS injection pass. + returning: None, + rls_filters: Vec::new(), + // Filled by the materialized-sum resolution pass, + // which runs after conversion. + resolved_sum_targets: Vec::new(), + deferred_sum_targets: Vec::new(), + }), + post_set_op: PostSetOp::None, + txn_id: None, + }); + continue; + } + batch_documents.push((doc_id, value_bytes)); + batch_surrogates.push(surrogate); + } + } + } + + // Emit the statement's plain rows: one `BatchInsert` page when there is + // more than one, so the driver sees a single `INSERT 0 n` tag and the + // rows share one atomic write; a lone row keeps its `PointInsert` task. + match batch_documents.len() { + 0 => {} + 1 => { + if let Some(((document_id, value), surrogate)) = + batch_documents.pop().zip(batch_surrogates.pop()) + { + tasks.push(PhysicalTask { + tenant_id, + vshard_id: vshard, + database_id: ctx.database_id, + plan: PhysicalPlan::Document(DocumentOp::PointInsert { collection: qualified_collection.clone(), - document_id: doc_id, - value: value_bytes, - if_absent, + document_id, + value, + if_absent: false, surrogate, - // Both filled in after conversion: the RETURNING spec - // by the protocol layer's injection pass, the read - // filter by the RLS injection pass. returning: None, rls_filters: Vec::new(), - // Filled by the materialized-sum resolution pass, - // which runs after conversion (it needs the catalog - // and, in cluster mode, a routed lookup). resolved_sum_targets: Vec::new(), deferred_sum_targets: Vec::new(), - }) - }; - tasks.push(PhysicalTask { - tenant_id, - vshard_id: vshard, - database_id: ctx.database_id, - plan, + }), post_set_op: PostSetOp::None, txn_id: None, }); } } + _ => { + tasks.push(PhysicalTask { + tenant_id, + vshard_id: vshard, + database_id: ctx.database_id, + plan: PhysicalPlan::Document(DocumentOp::BatchInsert { + collection: qualified_collection.clone(), + documents: batch_documents, + surrogates: batch_surrogates, + returning: None, + rls_filters: Vec::new(), + resolved_sum_targets: Vec::new(), + deferred_sum_targets: Vec::new(), + }), + post_set_op: PostSetOp::None, + txn_id: None, + }); + } } if !balanced_documents.is_empty() { From d4c345869dd42002d875936db1b702df7e4094de Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:49 +0800 Subject: [PATCH 03/14] fix(pgwire): answer kv INSERT with the INSERT tag kv writes reported no affected count, so pgwire emitted a bare OK. - report affected = 1 in execute_kv_insert and execute_kv_put - map KvOp::Insert, KvOp::Put, KvOp::BatchPut to the INSERT tag --- nodedb/src/control/server/response_shape/types.rs | 3 +++ .../src/data/executor/handlers/kv/crud/write_basic.rs | 10 ++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/nodedb/src/control/server/response_shape/types.rs b/nodedb/src/control/server/response_shape/types.rs index b70d97615..bdf59a5e2 100644 --- a/nodedb/src/control/server/response_shape/types.rs +++ b/nodedb/src/control/server/response_shape/types.rs @@ -147,7 +147,10 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind { PhysicalPlan::Document(DocumentOp::PointPut { .. }) | PhysicalPlan::Document(DocumentOp::PointInsert { .. }) | PhysicalPlan::Document(DocumentOp::BatchInsert { .. }) + | PhysicalPlan::Kv(KvOp::Insert { .. }) | PhysicalPlan::Kv(KvOp::InsertIfAbsent { .. }) + | PhysicalPlan::Kv(KvOp::Put { .. }) + | PhysicalPlan::Kv(KvOp::BatchPut { .. }) | PhysicalPlan::Columnar(ColumnarOp::Insert { .. }) => DmlResult("INSERT"), PhysicalPlan::Document(DocumentOp::PointUpdate { diff --git a/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs b/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs index 39594631c..292c42046 100644 --- a/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs +++ b/nodedb/src/data/executor/handlers/kv/crud/write_basic.rs @@ -80,7 +80,10 @@ impl CoreLoop { // stored post-image, not an echo of the request. return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]); } - self.response_ok(task) + // A put always writes its row, so the statement affected exactly one: + // the tag is `INSERT 0 1`, never a bare `OK` (pgwire's generic tag for + // a plan that reports nothing). + self.response_affected(task, 1) } /// SQL `INSERT` semantics: write only if key doesn't already exist. @@ -167,7 +170,10 @@ impl CoreLoop { if let Some(spec) = returning { return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]); } - self.response_ok(task) + // An insert writes exactly one row or fails the statement, so the + // affected count is 1 and pgwire renders `INSERT 0 1` — the same tag + // the document engine's point insert produces. + self.response_affected(task, 1) } /// SQL `INSERT ... ON CONFLICT DO NOTHING` semantics: write if absent, From 436b50259c6125f05e2acb0d6d19e25c9c8eaf61 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:49 +0800 Subject: [PATCH 04/14] fix(strict-format): coerce ARRAY vector literals ARRAY[0.1, 0.2, 0.3] folds its literals to Decimal. extract_vector_floats kept Float and Integer, dropped the rest, and reported "got 0 elements" as XX000. - accept Float, Integer, Decimal, and numeric String elements - reject a non-numeric element by index and type - refuse non-finite values --- .../src/data/executor/strict_format/coerce.rs | 108 ++++++++++++++++-- 1 file changed, 98 insertions(+), 10 deletions(-) diff --git a/nodedb/src/data/executor/strict_format/coerce.rs b/nodedb/src/data/executor/strict_format/coerce.rs index d2ec8eec1..83681ca85 100644 --- a/nodedb/src/data/executor/strict_format/coerce.rs +++ b/nodedb/src/data/executor/strict_format/coerce.rs @@ -166,7 +166,7 @@ pub fn coerce_value(val: &Value, col_type: &ColumnType, col_name: &str) -> crate ColumnType::Vector(dim) => match val { Value::Bytes(b) if b.len() == *dim as usize * 4 => Ok(val.clone()), Value::Array(arr) => { - let floats = extract_vector_floats(arr); + let floats = extract_vector_floats(col_name, arr)?; validate_and_encode_vector(col_name, *dim, &floats) } Value::String(s) => { @@ -229,15 +229,56 @@ pub fn coerce_value(val: &Value, col_type: &ColumnType, col_name: &str) -> crate } } -/// Extract f32 floats from a `Value::Array`. -fn extract_vector_floats(arr: &[Value]) -> Vec { - arr.iter() - .filter_map(|v| match v { - Value::Float(f) => Some(*f as f32), - Value::Integer(n) => Some(*n as f32), - _ => None, - }) - .collect() +/// Extract `f32` floats from a `Value::Array` for a VECTOR column. +/// +/// Accepts the element shapes the planner actually produces for an +/// `ARRAY[...]` literal — `Float`, `Integer`, `Decimal` (a folded numeric +/// literal arrives as `Decimal`), and numeric `String` — matching the +/// schemaless point-put path (`vector_string::floats_from_value`), so one +/// literal behaves the same on every engine. +/// +/// A non-numeric element is rejected by index and type. A `filter_map` here +/// used to drop it silently, which reported the literal as having fewer +/// elements than the user wrote and surfaced the mismatch as an internal +/// error. +fn extract_vector_floats(col_name: &str, arr: &[Value]) -> crate::Result> { + let mut floats = Vec::with_capacity(arr.len()); + for (i, v) in arr.iter().enumerate() { + let f = match v { + Value::Float(f) => *f as f32, + Value::Integer(n) => *n as f32, + Value::Decimal(d) => { + use rust_decimal::prelude::ToPrimitive; + d.to_f32().ok_or_else(|| crate::Error::BadRequest { + detail: format!( + "column '{col_name}': VECTOR element {i}: cannot represent DECIMAL {d} as FLOAT" + ), + })? + } + Value::String(s) => s.parse::().map_err(|_| crate::Error::BadRequest { + detail: format!( + "column '{col_name}': VECTOR element {i}: expected a numeric element, got String({s:?})" + ), + })?, + other => { + return Err(crate::Error::BadRequest { + detail: format!( + "column '{col_name}': VECTOR element {i}: expected a numeric element, got {}", + other.type_name() + ), + }); + } + }; + if !f.is_finite() { + return Err(crate::Error::BadRequest { + detail: format!( + "column '{col_name}': VECTOR element {i}: expected a finite element, got {f}" + ), + }); + } + floats.push(f); + } + Ok(floats) } /// Validate dimension count and encode as little-endian bytes. @@ -415,4 +456,51 @@ mod tests { let msg = result.unwrap_err().to_string(); assert!(msg.contains("\"extra\"") && msg.contains("does not exist")); } + + #[test] + fn vector_accepts_decimal_and_numeric_string_elements() { + // `ARRAY[0.1, 0.2, 0.3]` folds its numeric literals to `Decimal` + // elements; numeric strings are accepted like the schemaless path. + let arr = vec![ + Value::Decimal(rust_decimal::Decimal::new(1, 1)), + Value::String("0.2".into()), + Value::Integer(3), + ]; + let floats = extract_vector_floats("embedding", &arr).unwrap(); + assert_eq!(floats.len(), 3); + assert!((floats[0] - 0.1).abs() < 1e-6, "{floats:?}"); + assert!((floats[1] - 0.2).abs() < 1e-6, "{floats:?}"); + assert!((floats[2] - 3.0).abs() < 1e-6, "{floats:?}"); + } + + #[test] + fn vector_element_error_names_index_and_type() { + let arr = vec![Value::Float(0.1), Value::String("x".into())]; + let err = extract_vector_floats("embedding", &arr) + .unwrap_err() + .to_string(); + assert!(err.contains("VECTOR element 1"), "{err}"); + assert!(err.contains("String"), "{err}"); + } + + #[test] + fn vector_rejects_non_finite_elements() { + let arr = vec![Value::Float(f64::NAN)]; + assert!(extract_vector_floats("embedding", &arr).is_err()); + } + + #[test] + fn vector_encodes_three_decimals_at_full_width() { + let arr = vec![ + Value::Decimal(rust_decimal::Decimal::new(1, 1)), + Value::Decimal(rust_decimal::Decimal::new(2, 1)), + Value::Decimal(rust_decimal::Decimal::new(3, 1)), + ]; + let floats = extract_vector_floats("embedding", &arr).unwrap(); + let encoded = validate_and_encode_vector("embedding", 3, &floats).unwrap(); + match encoded { + Value::Bytes(b) => assert_eq!(b.len(), 12), + other => panic!("expected Bytes, got {other:?}"), + } + } } From 19f52341759f1100c2a7c183e5aba303df5d6226 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 05/14] fix(executor): keep value errors out of XX000 The binary_tuple wrap sites turned every non-UnknownStrictField error into a Serialization error. A BadRequest surfaced as XX000 with no SQLSTATE. - preserve BadRequest at the four wrap sites --- .../src/data/executor/handlers/point/apply_put/stored_body.rs | 2 +- .../data/executor/handlers/transaction/stage_write/body.rs | 4 ++-- .../executor/handlers/transaction/stage_write/stage_upsert.rs | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/nodedb/src/data/executor/handlers/point/apply_put/stored_body.rs b/nodedb/src/data/executor/handlers/point/apply_put/stored_body.rs index 63c8a0e11..5180a2f15 100644 --- a/nodedb/src/data/executor/handlers/point/apply_put/stored_body.rs +++ b/nodedb/src/data/executor/handlers/point/apply_put/stored_body.rs @@ -127,7 +127,7 @@ impl CoreLoop { strict_format::bytes_to_binary_tuple(&value, schema, collection) } .map_err(|e| match e { - crate::Error::UnknownStrictField { .. } => e, + crate::Error::UnknownStrictField { .. } | crate::Error::BadRequest { .. } => e, other => crate::Error::Serialization { format: "binary_tuple".into(), detail: other.to_string(), diff --git a/nodedb/src/data/executor/handlers/transaction/stage_write/body.rs b/nodedb/src/data/executor/handlers/transaction/stage_write/body.rs index 979a46b07..8dd5e3578 100644 --- a/nodedb/src/data/executor/handlers/transaction/stage_write/body.rs +++ b/nodedb/src/data/executor/handlers/transaction/stage_write/body.rs @@ -126,7 +126,7 @@ impl CoreLoop { strict_format::bytes_to_binary_tuple(&encoded_input, schema, collection) } .map_err(|e| match e { - crate::Error::UnknownStrictField { .. } => e, + crate::Error::UnknownStrictField { .. } | crate::Error::BadRequest { .. } => e, other => crate::Error::Serialization { format: "binary_tuple".into(), detail: other.to_string(), @@ -245,7 +245,7 @@ impl CoreLoop { strict_format::value_to_binary_tuple(&ndb_val, schema, collection) } .map_err(|e| match e { - crate::Error::UnknownStrictField { .. } => e, + crate::Error::UnknownStrictField { .. } | crate::Error::BadRequest { .. } => e, other => crate::Error::Serialization { format: "binary_tuple".into(), detail: other.to_string(), diff --git a/nodedb/src/data/executor/handlers/transaction/stage_write/stage_upsert.rs b/nodedb/src/data/executor/handlers/transaction/stage_write/stage_upsert.rs index 1d100b7d3..562072666 100644 --- a/nodedb/src/data/executor/handlers/transaction/stage_write/stage_upsert.rs +++ b/nodedb/src/data/executor/handlers/transaction/stage_write/stage_upsert.rs @@ -175,7 +175,7 @@ impl CoreLoop { strict_format::value_to_binary_tuple(&merged, schema, ctx.collection) }; result.map_err(|e| match e { - crate::Error::UnknownStrictField { .. } => e, + crate::Error::UnknownStrictField { .. } | crate::Error::BadRequest { .. } => e, other => crate::Error::Serialization { format: "binary_tuple".into(), detail: other.to_string(), From 8380674ac30c44c4f12ae93820f4e276d5eaef31 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 06/14] fix(insert-select): scan source in its own engine The document materializer scanned every source. A kv source read an empty document store, so INSERT ... SELECT copied no rows and reported INSERT 0 0. - route on the catalog engine in clone_materializer/auto_source.rs - kv normalizes the materialize-scan page to the copy entry shape - refuse columnar, timeseries, spatial, array by name --- .../control/insert_select/expand_staged.rs | 4 +- .../src/control/insert_select/orchestrator.rs | 4 +- .../clone_materializer/auto_source.rs | 72 +++++++++++++++++++ .../maintenance/clone_materializer/kv.rs | 2 +- .../maintenance/clone_materializer/mod.rs | 4 +- .../control/planner/catalog_adapter/mod.rs | 2 +- .../planner/catalog_adapter/type_convert.rs | 2 +- 7 files changed, 82 insertions(+), 8 deletions(-) create mode 100644 nodedb/src/control/maintenance/clone_materializer/auto_source.rs diff --git a/nodedb/src/control/insert_select/expand_staged.rs b/nodedb/src/control/insert_select/expand_staged.rs index 583c42011..ea2f7cada 100644 --- a/nodedb/src/control/insert_select/expand_staged.rs +++ b/nodedb/src/control/insert_select/expand_staged.rs @@ -19,7 +19,7 @@ use nodedb_types::{DatabaseId, Surrogate, TenantId}; use crate::bridge::envelope::PhysicalPlan; use crate::control::insert_select::copy_rows::{assign_page_rows, resolve_copy_spec}; -use crate::control::maintenance::clone_materializer::scan_source_page; +use crate::control::maintenance::clone_materializer::scan_source_page_auto; use crate::control::state::SharedState; use crate::types::{TxnId, VShardId}; use nodedb_physical::physical_plan::DocumentOp; @@ -174,7 +174,7 @@ async fn materialize_copy( let mut rows: Vec<(String, Vec, Surrogate)> = Vec::new(); while remaining > 0 { - let (entries, next_cursor) = scan_source_page( + let (entries, next_cursor) = scan_source_page_auto( state, tenant_id, database_id, diff --git a/nodedb/src/control/insert_select/orchestrator.rs b/nodedb/src/control/insert_select/orchestrator.rs index 0c2d5c442..7208d5145 100644 --- a/nodedb/src/control/insert_select/orchestrator.rs +++ b/nodedb/src/control/insert_select/orchestrator.rs @@ -19,7 +19,7 @@ use nodedb_types::{DatabaseId, Lsn, Surrogate, TenantId}; use crate::bridge::envelope::{Payload, PhysicalPlan, Response, Status}; use crate::control::insert_select::copy_rows::{assign_page_rows, resolve_copy_spec}; -use crate::control::maintenance::clone_materializer::{dispatch_local, scan_source_page}; +use crate::control::maintenance::clone_materializer::{dispatch_local, scan_source_page_auto}; use crate::control::state::SharedState; use nodedb_physical::physical_plan::DocumentOp; @@ -99,7 +99,7 @@ pub(crate) async fn run_insert_select( while remaining > 0 { // Phase 1: scan one source page (point-in-time snapshot). - let (entries, next_cursor) = scan_source_page( + let (entries, next_cursor) = scan_source_page_auto( state, tenant_id, database_id, diff --git a/nodedb/src/control/maintenance/clone_materializer/auto_source.rs b/nodedb/src/control/maintenance/clone_materializer/auto_source.rs new file mode 100644 index 000000000..e8823115e --- /dev/null +++ b/nodedb/src/control/maintenance/clone_materializer/auto_source.rs @@ -0,0 +1,72 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Engine-aware source scan for `INSERT ... SELECT`. +//! +//! The document materializer reads the document store; a kv collection keeps +//! nothing there, so scanning a kv source with it materializes zero rows and +//! the statement reports `INSERT 0 0`. Route on the source collection's own +//! engine and normalize every page to the document entry shape +//! `(doc_id, source_surrogate, value_bytes)`: the copy pipeline ignores the +//! ids and shapes the body through its column map, and for kv the body is the +//! stored msgpack row, so expression cells evaluate exactly as they do over a +//! document source. + +use nodedb_sql::types::EngineType; + +use super::{document, kv}; + +pub(crate) async fn scan_source_page( + state: &crate::control::state::SharedState, + tenant_id: nodedb_types::TenantId, + database_id: nodedb_types::DatabaseId, + source_qualified: &str, + cursor: &[u8], + system_as_of_ms: Option, + txn_id: Option, +) -> crate::Result<(Vec<(String, u32, Vec)>, Vec)> { + let catalog = state.credentials.catalog(); + let stored = catalog + .get_collection( + database_id, + tenant_id.as_u64(), + &crate::control::target_identity::bare_collection_name(database_id, source_qualified), + )? + .ok_or_else(|| crate::Error::CollectionNotFound { + tenant_id, + collection: source_qualified.to_string(), + })?; + let (engine, _, _) = + crate::control::planner::catalog_adapter::type_convert::convert_collection_type(&stored); + + match engine { + EngineType::DocumentSchemaless | EngineType::DocumentStrict => { + document::scan_source_page( + state, + tenant_id, + database_id, + source_qualified, + cursor, + system_as_of_ms, + txn_id, + ) + .await + } + EngineType::KeyValue => { + let (pairs, next) = + kv::scan_source_page(state, tenant_id, database_id, source_qualified, cursor) + .await?; + let entries = pairs + .into_iter() + .map(|(key, value)| (String::from_utf8_lossy(&key).into_owned(), 0, value)) + .collect(); + Ok((entries, next)) + } + // Refused by name instead of silently copied as nothing; these + // engines have their own materializers that are not wired here yet. + EngineType::Columnar | EngineType::Timeseries | EngineType::Spatial | EngineType::Array => { + Err(crate::Error::PlanError { + detail: format!("INSERT ... SELECT from {engine:?} sources is not supported yet"), + }) + } + } +} diff --git a/nodedb/src/control/maintenance/clone_materializer/kv.rs b/nodedb/src/control/maintenance/clone_materializer/kv.rs index abe438050..b930bc09d 100644 --- a/nodedb/src/control/maintenance/clone_materializer/kv.rs +++ b/nodedb/src/control/maintenance/clone_materializer/kv.rs @@ -189,7 +189,7 @@ fn checkpoint_progress( /// Run one source-side `MaterializeScan` round-trip. Returns the entries in /// this page (raw `(key, value)` byte pairs) plus the next-cursor; the /// cursor is empty when the scan is complete. -async fn scan_source_page( +pub(crate) async fn scan_source_page( state: &SharedState, tenant_id: TenantId, source_db_id: DatabaseId, diff --git a/nodedb/src/control/maintenance/clone_materializer/mod.rs b/nodedb/src/control/maintenance/clone_materializer/mod.rs index 23f0ef7ca..163bd07d9 100644 --- a/nodedb/src/control/maintenance/clone_materializer/mod.rs +++ b/nodedb/src/control/maintenance/clone_materializer/mod.rs @@ -1,5 +1,6 @@ // SPDX-License-Identifier: BUSL-1.1 +mod auto_source; mod columnar; mod dispatch; mod document; @@ -17,5 +18,6 @@ pub use walker::{ // Shared with the `INSERT ... SELECT` orchestrator, which reuses the same // local-dispatch primitive and source-scan cursor decode. +pub(crate) use auto_source::scan_source_page as scan_source_page_auto; pub(crate) use dispatch::{dispatch_local, dispatch_local_on_vshard}; -pub(crate) use document::{read_all_source_rows, scan_source_page}; +pub(crate) use document::read_all_source_rows; diff --git a/nodedb/src/control/planner/catalog_adapter/mod.rs b/nodedb/src/control/planner/catalog_adapter/mod.rs index fe5c8beca..f0bb06dd1 100644 --- a/nodedb/src/control/planner/catalog_adapter/mod.rs +++ b/nodedb/src/control/planner/catalog_adapter/mod.rs @@ -33,6 +33,6 @@ mod adapter; mod sequence_access; mod sql_catalog_impl; -mod type_convert; +pub(crate) mod type_convert; pub use adapter::OriginCatalog; diff --git a/nodedb/src/control/planner/catalog_adapter/type_convert.rs b/nodedb/src/control/planner/catalog_adapter/type_convert.rs index 7438b2666..e13742daf 100644 --- a/nodedb/src/control/planner/catalog_adapter/type_convert.rs +++ b/nodedb/src/control/planner/catalog_adapter/type_convert.rs @@ -6,7 +6,7 @@ use nodedb_sql::types::{ColumnInfo, EngineType, SqlDataType}; use nodedb_types::columnar::{FloatWidth, IntWidth}; /// Convert a StoredCollection to engine type, columns, and primary key. -pub(super) fn convert_collection_type( +pub(crate) fn convert_collection_type( stored: &crate::control::security::catalog::StoredCollection, ) -> (EngineType, Vec, Option) { use nodedb_types::CollectionType; From e764eed90446251ed0bce14873e5ace501655f27 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 07/14] fix(sql): declare the SEARCH distance column Derived-relation inference listed declared columns only. The response layer appends a synthetic distance cell, so s.distance failed 42703 on a closed schema. - append distance when ORDER BY carries vector_distance(...) --- nodedb-sql/src/resolver/derived.rs | 47 +++++++++++++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/nodedb-sql/src/resolver/derived.rs b/nodedb-sql/src/resolver/derived.rs index 104c5e1b3..257157bf3 100644 --- a/nodedb-sql/src/resolver/derived.rs +++ b/nodedb-sql/src/resolver/derived.rs @@ -86,7 +86,34 @@ fn infer_projection( catalog: &dyn SqlCatalog, query: &ast::Query, ) -> Result<(Vec, bool)> { - infer_body(catalog, &query.body) + let (mut columns, open) = infer_body(catalog, &query.body)?; + + // The SEARCH preprocessor rewrites `SEARCH c USING VECTOR(...)` into + // `SELECT * FROM c ORDER BY vector_distance(...) LIMIT k`, and the + // response layer appends a synthetic `distance` cell to every row of such + // a query. Derived-relation inference must declare that column: without + // it `s.distance` over a closed-schema source resolves against no + // relation and is refused with 42703, while the same projection over an + // open-schema source runs. + if !columns.iter().any(|c| c.name == "distance") + && query.order_by.iter().any(|order| match &order.kind { + ast::OrderByKind::Expressions(exprs) => { + exprs.iter().any(|ordered| match &ordered.expr { + Expr::Function(func) => matches!( + func.name.0.as_slice(), + [ast::ObjectNamePart::Identifier(ident)] + if normalize_ident(ident) == "vector_distance" + ), + _ => false, + }) + } + ast::OrderByKind::All(_) => false, + }) + { + columns.push(synthetic_column("distance")); + } + + Ok((columns, open)) } fn infer_body(catalog: &dyn SqlCatalog, body: &SetExpr) -> Result<(Vec, bool)> { @@ -320,4 +347,22 @@ mod tests { assert_eq!(names, vec!["p", "q"]); assert_eq!(info.columns[0].data_type, SqlDataType::Int64); } + + #[test] + fn vector_search_projection_declares_the_synthetic_distance_column() { + // `SEARCH c USING VECTOR(...)` preprocesses to `ORDER BY + // vector_distance(...)`; the response layer appends a `distance` + // cell, so the derived relation must name it even when the source + // schema is closed. + let info = infer("SELECT * FROM src ORDER BY vector_distance(b, ARRAY[0.1, 0.2]) LIMIT 2"); + let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect(); + assert!(names.contains(&"distance"), "columns: {names:?}"); + } + + #[test] + fn plain_ordered_projection_has_no_distance_column() { + let info = infer("SELECT * FROM src ORDER BY b LIMIT 2"); + let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect(); + assert!(!names.contains(&"distance"), "columns: {names:?}"); + } } From 7f06791c762dcaba0a088d9106d9f7a8f27ae1d1 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 08/14] feat(kv): evaluate scan projections per row KvOp::Scan carried no projection list and no computed columns. Every expression projection over a kv source shaped to NULL. - add projection and computed_columns to KvOp::Scan - evaluate both in the kv scan handler; keep plain projections full-row - update all KvOp::Scan constructors - add the kv_select_expressions wire case --- nodedb-physical/src/physical_plan/kv/op.rs | 8 ++ nodedb/src/control/clone/resolver/rewrite.rs | 7 ++ .../planner/sql_plan_convert/scan/core.rs | 7 ++ .../src/control/server/exchange/full_scan.rs | 2 + .../server/native/dispatch/plan_builder/kv.rs | 2 + nodedb/src/control/server/resp/handler.rs | 6 ++ .../shared/ddl/neutral/weighted_pick.rs | 2 + .../predicate/txn_buffering/classify.rs | 2 + .../src/data/executor/handlers/kv/dispatch.rs | 4 + nodedb/src/data/executor/handlers/kv/scan.rs | 34 ++++++++ .../test_cross_type_join/multi_core_joins.rs | 6 ++ .../inproc/cases/executor_tests/test_kv.rs | 4 + .../cases/executor_tests/test_kv_advanced.rs | 2 + .../executor_tests/test_kv_scan_budget.rs | 2 + .../tests/wire/cases/kv_select_expressions.rs | 78 +++++++++++++++++++ 15 files changed, 166 insertions(+) create mode 100644 nodedb/tests/wire/cases/kv_select_expressions.rs diff --git a/nodedb-physical/src/physical_plan/kv/op.rs b/nodedb-physical/src/physical_plan/kv/op.rs index 1fe9126c7..17e01d800 100644 --- a/nodedb-physical/src/physical_plan/kv/op.rs +++ b/nodedb-physical/src/physical_plan/kv/op.rs @@ -152,6 +152,14 @@ pub enum KvOp { /// See `Get::surrogate_ceiling`; drops entries above the ceiling. #[serde(default)] surrogate_ceiling: Option, + /// Output column names (same format as DocumentOp::Scan). Empty = + /// return the whole row document. + #[serde(default)] + projection: Vec, + /// Serialized `Vec` applied per row after the scan + /// (same format as DocumentOp::Scan). Empty = none. + #[serde(default)] + computed_columns: Vec, }, /// Set or update TTL on an existing key. diff --git a/nodedb/src/control/clone/resolver/rewrite.rs b/nodedb/src/control/clone/resolver/rewrite.rs index 237ee8f23..0ec376f5d 100644 --- a/nodedb/src/control/clone/resolver/rewrite.rs +++ b/nodedb/src/control/clone/resolver/rewrite.rs @@ -117,6 +117,7 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res input, filters, projection, + computed_columns, sort_keys, limit, offset, @@ -139,6 +140,7 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res input: child, filters: filters.clone(), projection: projection.clone(), + computed_columns: computed_columns.clone(), sort_keys: sort_keys.clone(), limit: *limit, offset: *offset, @@ -249,6 +251,8 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res // (clones-of-clones still funnel through here per-level); // the resolver overrides it for source delegation below. surrogate_ceiling: _, + projection, + computed_columns, }) if collection == &target_qualified => { Ok(SourceRewrite::task(PhysicalPlan::Kv(KvOp::Scan { collection: source_qualified, @@ -258,6 +262,8 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res match_pattern: match_pattern.clone(), sort_keys: sort_keys.clone(), surrogate_ceiling: kv_surrogate_ceiling, + projection: projection.clone(), + computed_columns: computed_columns.clone(), }))) } @@ -626,6 +632,7 @@ mod tests { input: Box::new(gather(plan)), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/sql_plan_convert/scan/core.rs b/nodedb/src/control/planner/sql_plan_convert/scan/core.rs index 5e56c2697..4d34d8cb7 100644 --- a/nodedb/src/control/planner/sql_plan_convert/scan/core.rs +++ b/nodedb/src/control/planner/sql_plan_convert/scan/core.rs @@ -58,6 +58,8 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_scan( rows: Vec::new(), filters: filter_bytes, projection: proj_names, + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: sort, limit: *limit, offset: *offset, @@ -134,6 +136,11 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_scan( // Original SQL planner output never carries a clone ceiling; // the clone resolver overrides it when delegating to source. surrogate_ceiling: None, + // KV scan parity with doc/columnar/timeseries: carry the SELECT + // output columns so expression projections are evaluated in the + // Data Plane instead of surfacing as NULL at response shaping. + projection: proj_names, + computed_columns: computed_bytes, }), EngineType::DocumentSchemaless | EngineType::DocumentStrict => { PhysicalPlan::Document(DocumentOp::Scan { diff --git a/nodedb/src/control/server/exchange/full_scan.rs b/nodedb/src/control/server/exchange/full_scan.rs index 58fb851d2..277c58f93 100644 --- a/nodedb/src/control/server/exchange/full_scan.rs +++ b/nodedb/src/control/server/exchange/full_scan.rs @@ -151,6 +151,8 @@ pub fn full_scan_plan_for_collection( sort_keys: Vec::new(), match_pattern: None, surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), CollectionType::Columnar(ColumnarProfile::Plain) | CollectionType::Columnar(ColumnarProfile::Spatial { .. }) => { diff --git a/nodedb/src/control/server/native/dispatch/plan_builder/kv.rs b/nodedb/src/control/server/native/dispatch/plan_builder/kv.rs index d66f38600..1559914d8 100644 --- a/nodedb/src/control/server/native/dispatch/plan_builder/kv.rs +++ b/nodedb/src/control/server/native/dispatch/plan_builder/kv.rs @@ -27,6 +27,8 @@ pub(crate) fn build_scan( match_pattern, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), })) } diff --git a/nodedb/src/control/server/resp/handler.rs b/nodedb/src/control/server/resp/handler.rs index 0568646e5..ce17d7c83 100644 --- a/nodedb/src/control/server/resp/handler.rs +++ b/nodedb/src/control/server/resp/handler.rs @@ -348,6 +348,8 @@ async fn handle_scan(cmd: &RespCommand, session: &RespSession, state: &SharedSta match_pattern, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }); match dispatch_kv(state, session, plan).await { @@ -382,6 +384,8 @@ async fn handle_keys(cmd: &RespCommand, session: &RespSession, state: &SharedSta match_pattern: Some(pattern.to_string()), sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }); match dispatch_kv(state, session, plan).await { @@ -412,6 +416,8 @@ async fn handle_dbsize(session: &RespSession, state: &SharedState) -> RespValue match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }); match dispatch_kv(state, session, plan).await { diff --git a/nodedb/src/control/server/shared/ddl/neutral/weighted_pick.rs b/nodedb/src/control/server/shared/ddl/neutral/weighted_pick.rs index 847cb1519..a739c73a6 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/weighted_pick.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/weighted_pick.rs @@ -246,6 +246,8 @@ async fn scan_all_entries( match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }); gate.inject_rls(&mut plan)?; diff --git a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs index 29c9350d4..c1a434bf7 100644 --- a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs +++ b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs @@ -1186,6 +1186,8 @@ mod tests { match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), PhysicalPlan::Kv(KvOp::Expire { collection: QualifiedCollection::new(DatabaseId::DEFAULT, "c"), diff --git a/nodedb/src/data/executor/handlers/kv/dispatch.rs b/nodedb/src/data/executor/handlers/kv/dispatch.rs index e5c6d040b..d2cf0dd7a 100644 --- a/nodedb/src/data/executor/handlers/kv/dispatch.rs +++ b/nodedb/src/data/executor/handlers/kv/dispatch.rs @@ -138,6 +138,8 @@ impl CoreLoop { match_pattern, sort_keys, surrogate_ceiling, + projection, + computed_columns, } => self.execute_kv_scan( task, super::scan::KvScanHandlerParams { @@ -150,6 +152,8 @@ impl CoreLoop { filters, sort_keys, surrogate_ceiling: *surrogate_ceiling, + projection, + computed_columns, }, ), KvOp::Expire { diff --git a/nodedb/src/data/executor/handlers/kv/scan.rs b/nodedb/src/data/executor/handlers/kv/scan.rs index 188f489b3..4d3ddd197 100644 --- a/nodedb/src/data/executor/handlers/kv/scan.rs +++ b/nodedb/src/data/executor/handlers/kv/scan.rs @@ -22,6 +22,8 @@ pub(in crate::data::executor) struct KvScanHandlerParams<'a> { pub filters: &'a [u8], pub sort_keys: &'a [nodedb_physical::physical_plan::SortKeySpec], pub surrogate_ceiling: Option, + pub projection: &'a [String], + pub computed_columns: &'a [u8], } impl CoreLoop { @@ -40,6 +42,8 @@ impl CoreLoop { filters, sort_keys, surrogate_ceiling, + projection, + computed_columns, } = params; debug!(core = self.core_id, %collection, count, "kv scan"); @@ -155,6 +159,36 @@ impl CoreLoop { return self.response_error(task, crate::Error::from(e)); } + // Computed-column parity with the document scan path (same wire + // format, same evaluator). Without this, SELECT-list expressions + // over kv collections surfaced as NULL at response shaping + // (`SELECT 1 + 1 FROM kv` returned an empty column), and sequence + // accessors could not raise their typed 0A000 here. + // + // Plain projections (no computed columns) deliberately stay + // full-row: clone-source delegation merges rows by primary key at + // the control plane, then the response shape extracts columns — + // projecting here would strip the keys and break tombstone + // suppression (clone_write_suppresses_source_row). + if !computed_columns.is_empty() { + let computed_cols: Vec = + if computed_columns.is_empty() { + Vec::new() + } else { + zerompk::from_msgpack(computed_columns).unwrap_or_default() + }; + for entry in result_entries.iter_mut() { + match crate::data::executor::handlers::document::read::projection:: + apply_projection_msgpack(entry, &computed_cols, projection) + { + Ok(out) => *entry = out, + Err(e) => { + return self.response_error(task, e); + } + } + } + } + // Build response as flat msgpack array — same format as document/columnar scan. // RESP SCAN handles cursor pagination at its own handler layer. let mut payload = diff --git a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs index 0452152fc..39e3d0025 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs @@ -80,6 +80,8 @@ fn multi_core_broadcast_inner_join() { sort_keys: Vec::new(), match_pattern: None, surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), ); @@ -234,6 +236,8 @@ fn multi_core_broadcast_left_join() { sort_keys: Vec::new(), match_pattern: None, surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), ); @@ -401,6 +405,8 @@ fn multi_core_broadcast_merge_simulation() { sort_keys: Vec::new(), match_pattern: None, surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }); let payload0 = send_ok( &mut core0.core, diff --git a/nodedb/tests/inproc/cases/executor_tests/test_kv.rs b/nodedb/tests/inproc/cases/executor_tests/test_kv.rs index 52a8aa244..8f6790598 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_kv.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_kv.rs @@ -234,6 +234,8 @@ fn kv_scan_returns_entries() { match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), ); @@ -283,6 +285,8 @@ fn kv_scan_with_match_pattern() { match_pattern: Some("user:*".into()), sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), ); diff --git a/nodedb/tests/inproc/cases/executor_tests/test_kv_advanced.rs b/nodedb/tests/inproc/cases/executor_tests/test_kv_advanced.rs index 5dba10d7e..3132cf0fe 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_kv_advanced.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_kv_advanced.rs @@ -611,6 +611,8 @@ fn kv_index_write_amp_ratio_matches() { match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }), ); let json: serde_json::Value = payload_value(&payload); diff --git a/nodedb/tests/inproc/cases/executor_tests/test_kv_scan_budget.rs b/nodedb/tests/inproc/cases/executor_tests/test_kv_scan_budget.rs index 28ba8fc7a..be11e865b 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_kv_scan_budget.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_kv_scan_budget.rs @@ -48,6 +48,8 @@ fn kv_scan(collection: &str, count: usize) -> PhysicalPlan { match_pattern: None, sort_keys: Vec::new(), surrogate_ceiling: None, + projection: Vec::new(), + computed_columns: Vec::new(), }) } diff --git a/nodedb/tests/wire/cases/kv_select_expressions.rs b/nodedb/tests/wire/cases/kv_select_expressions.rs new file mode 100644 index 000000000..5348b102f --- /dev/null +++ b/nodedb/tests/wire/cases/kv_select_expressions.rs @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! kv-engine SELECT projection expression evaluation. +//! +//! Previously the kv scan carried neither the SELECT projection list nor +//! computed columns, so expression projections were never evaluated: the +//! column came back NULL at response shaping (`SELECT 1 + 1 FROM kv` +//! returned an empty column) and sequence accessors could not raise their +//! typed 0A000 either. The scan now carries projection + computed columns +//! like the document/columnar paths, so: +//! +//! - scalar expressions over kv rows evaluate per row; +//! - `nextval`/`currval`/`setval` in a kv SELECT list raise 0A000 instead +//! of silently NULLing. + +use crate::harness::TestServer; + +async fn setup(server: &TestServer) { + server + .exec("CREATE COLLECTION kvsel (id BIGINT PRIMARY KEY, v TEXT) WITH (engine = 'kv')") + .await + .unwrap(); + server + .exec("INSERT INTO kvsel (id, v) VALUES (1, 'hello'), (2, 'world')") + .await + .unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn scalar_expressions_evaluate_per_row() { + let server = TestServer::start().await; + setup(&server).await; + + let rows = server + .query_named_rows("SELECT upper(v) AS u, 1 + 1 AS s FROM kvsel ORDER BY id") + .await + .expect("rows"); + assert_eq!(rows.len(), 2, "{rows:?}"); + let u: Vec<_> = rows + .iter() + .map(|r| r.get("u").map(|s| s.as_str())) + .collect(); + assert_eq!(u, vec![Some("HELLO"), Some("WORLD")], "{rows:?}"); + let s: Vec<_> = rows + .iter() + .map(|r| r.get("s").map(|s| s.as_str())) + .collect(); + assert_eq!(s, vec![Some("2"), Some("2")], "{rows:?}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn accessor_in_select_list_raises_0a000() { + let server = TestServer::start().await; + setup(&server).await; + server + .expect_error("SELECT nextval('nope') FROM kvsel", "0A000") + .await; + server + .expect_error("SELECT currval('nope') FROM kvsel", "0A000") + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn plain_projection_still_returns_stored_columns() { + let server = TestServer::start().await; + setup(&server).await; + let rows = server + .query_named_rows("SELECT id, v FROM kvsel ORDER BY id") + .await + .expect("rows"); + assert_eq!(rows.len(), 2, "{rows:?}"); + assert_eq!( + rows.iter() + .map(|r| r.get("v").map(|s| s.as_str())) + .collect::>(), + vec![Some("hello"), Some("world")] + ); +} From f2963056f5cfb21870f25383f1cf8337ebc4004c Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 09/14] fix(sql): raise errors on constant derived tables SELECT x/0 FROM (SELECT 1 AS x) s answered one NULL row. The same expression over a collection raised 22012. - evaluate computed columns per row on the materialized-row scan - lower aggregates over a non-Scan body to a ProviderScan sub-plan - carry window specs through the post-processor - add the derived_expression_errors wire case --- .../cases/shuffle_aggregate_cross_node.rs | 2 + .../cases/shuffle_consume_cross_node.rs | 2 + .../cases/shuffle_produce_cross_node.rs | 2 + nodedb-physical/src/physical_plan/query.rs | 20 ++++ nodedb-sql/src/planner/catalog_fold.rs | 6 + .../src/planner/catalog_plan_validate.rs | 2 + nodedb-sql/src/planner/select/entry.rs | 2 + nodedb-sql/src/planner/select/post_process.rs | 1 + nodedb-sql/src/types/plan/variants.rs | 3 + nodedb-sql/src/visitor/plan_visitor/args.rs | 1 + .../src/visitor/plan_visitor/dispatch_rest.rs | 2 + nodedb/src/control/clone/resolver/rewrite.rs | 3 + .../control/planner/redaction_refusal/plan.rs | 2 + .../rls_injection/permission_tree/plan.rs | 2 + .../src/control/planner/rls_injection/plan.rs | 2 + .../sql_plan_convert/aggregate/plan.rs | 67 +++++++++++ .../planner/sql_plan_convert/convert.rs | 2 + .../control/planner/sql_plan_convert/expr.rs | 41 ++++++- .../planner/sql_plan_convert/set_ops.rs | 13 ++ .../exchange/resolve/exchange/dispatch.rs | 4 + .../resolve/exchange/post_process_arm.rs | 6 + .../server/exchange/resolve/join_input.rs | 6 + .../server/exchange/resolve/materialize.rs | 8 ++ .../shared/authorization/requirements.rs | 2 + .../predicate/txn_buffering/classify.rs | 2 + nodedb/src/data/executor/dispatch/query.rs | 6 +- .../data/executor/handlers/provider_scan.rs | 111 +++++++++++++++++- .../test_cross_type_join/inline_hash_join.rs | 4 + .../test_cross_type_join/multi_core_joins.rs | 6 + .../wire/cases/derived_expression_errors.rs | 64 ++++++++++ nodedb/tests/wire/cases/mod.rs | 8 +- .../tests/wire/cases/sql_window_functions.rs | 71 +++++++++++ 32 files changed, 462 insertions(+), 11 deletions(-) create mode 100644 nodedb/tests/wire/cases/derived_expression_errors.rs diff --git a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_aggregate_cross_node.rs b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_aggregate_cross_node.rs index d534d18ba..08beb3d1a 100644 --- a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_aggregate_cross_node.rs +++ b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_aggregate_cross_node.rs @@ -107,6 +107,8 @@ fn producer_plan(rows: &[&Row]) -> Vec { rows: msgpack_array(rows), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_consume_cross_node.rs b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_consume_cross_node.rs index 205772766..f343e19fe 100644 --- a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_consume_cross_node.rs +++ b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_consume_cross_node.rs @@ -104,6 +104,8 @@ fn provider_scan_plan(rows: &[&Row]) -> Vec { rows: msgpack_array(rows), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_produce_cross_node.rs b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_produce_cross_node.rs index 3f0602c28..49f8bf830 100644 --- a/nodedb-cluster-tests/tests/common_suite/cases/shuffle_produce_cross_node.rs +++ b/nodedb-cluster-tests/tests/common_suite/cases/shuffle_produce_cross_node.rs @@ -108,6 +108,8 @@ fn provider_scan_plan(rows: &[Vec]) -> Vec { rows: msgpack_array(rows), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb-physical/src/physical_plan/query.rs b/nodedb-physical/src/physical_plan/query.rs index 96f087d3c..5de8629f4 100644 --- a/nodedb-physical/src/physical_plan/query.rs +++ b/nodedb-physical/src/physical_plan/query.rs @@ -78,6 +78,18 @@ pub enum QueryOp { /// Output column names to keep. Empty = emit all columns. #[serde(default)] projection: Vec, + /// Serialized `Vec` applied per row after + /// projection-name extraction (same wire format as the engine + /// scans). Expression projections over materialized rows (derived + /// tables, constant subqueries) need these — name-only projection + /// would silently drop them. + #[serde(default)] + computed_columns: Vec, + /// Serialized `Vec` evaluated per partition after + /// computed columns (window over derived-table rows — issue #295 + /// Gap 3). Empty = no window functions. + #[serde(default)] + window_functions: Vec, /// ORDER BY terms, each an expression. Empty = unordered. #[serde(default)] sort_keys: Vec, @@ -118,6 +130,14 @@ pub enum QueryOp { /// Output column names to keep. Empty = emit all columns. #[serde(default)] projection: Vec, + /// Serialized `Vec` applied per row (see + /// `ProviderScan::computed_columns`). + #[serde(default)] + computed_columns: Vec, + /// Serialized `Vec` (see + /// `ProviderScan::window_functions`). + #[serde(default)] + window_functions: Vec, /// ORDER BY terms, each an expression. Empty = unordered. #[serde(default)] sort_keys: Vec, diff --git a/nodedb-sql/src/planner/catalog_fold.rs b/nodedb-sql/src/planner/catalog_fold.rs index d29db868b..88ee8067c 100644 --- a/nodedb-sql/src/planner/catalog_fold.rs +++ b/nodedb-sql/src/planner/catalog_fold.rs @@ -112,6 +112,7 @@ fn walk_plan( input, mut filters, mut projection, + mut window_functions, mut sort_keys, offset, distinct, @@ -122,10 +123,15 @@ fn walk_plan( } fold_projection(&mut projection, catalog, database_id, tenant_id); fold_sort_keys(&mut sort_keys, catalog, database_id, tenant_id); + // Window specs carry their own exprs (args, PARTITION BY, + // ORDER BY) — a wrapper that skips them leaves catalog casts + // inside window exprs unfolded, which then match no row. + fold_windows(&mut window_functions, catalog, database_id, tenant_id); SqlPlan::Subquery { input: Box::new(walk_plan(*input, catalog, database_id, tenant_id)), filters, projection, + window_functions, sort_keys, offset, distinct, diff --git a/nodedb-sql/src/planner/catalog_plan_validate.rs b/nodedb-sql/src/planner/catalog_plan_validate.rs index 8f947256e..2f569b626 100644 --- a/nodedb-sql/src/planner/catalog_plan_validate.rs +++ b/nodedb-sql/src/planner/catalog_plan_validate.rs @@ -107,12 +107,14 @@ pub(super) fn validate_catalog_exprs( filters, projection, sort_keys, + window_functions, .. } => { validate_catalog_exprs(input, catalog, database_id, tenant_id)?; validate_filters(filters, catalog, database_id, tenant_id)?; validate_projection(projection, catalog, database_id, tenant_id)?; validate_sort_keys(sort_keys, catalog, database_id, tenant_id)?; + validate_windows(window_functions, catalog, database_id, tenant_id)?; } SqlPlan::Join { left, diff --git a/nodedb-sql/src/planner/select/entry.rs b/nodedb-sql/src/planner/select/entry.rs index 48a41f8f6..70a2dc426 100644 --- a/nodedb-sql/src/planner/select/entry.rs +++ b/nodedb-sql/src/planner/select/entry.rs @@ -173,6 +173,7 @@ pub fn plan_query( SqlPlan::Subquery { filters, projection, + window_functions, sort_keys, offset, distinct, @@ -182,6 +183,7 @@ pub fn plan_query( input: Box::new(upgraded_leaf), filters, projection, + window_functions, sort_keys, offset, distinct, diff --git a/nodedb-sql/src/planner/select/post_process.rs b/nodedb-sql/src/planner/select/post_process.rs index 3d351cefa..39a6f5692 100644 --- a/nodedb-sql/src/planner/select/post_process.rs +++ b/nodedb-sql/src/planner/select/post_process.rs @@ -37,6 +37,7 @@ pub(in crate::planner::select) fn post_process( input: Box::new(input), filters: Vec::new(), projection, + window_functions: Vec::new(), sort_keys, offset, distinct: false, diff --git a/nodedb-sql/src/types/plan/variants.rs b/nodedb-sql/src/types/plan/variants.rs index f5ad2b736..4560fbe24 100644 --- a/nodedb-sql/src/types/plan/variants.rs +++ b/nodedb-sql/src/types/plan/variants.rs @@ -525,6 +525,9 @@ pub enum SqlPlan { filters: Vec, /// Outer projection (target list). Empty = inherit the body's columns. projection: Vec, + /// Window functions evaluated over the materialized rows. Empty = + /// none. + window_functions: Vec, /// Outer `ORDER BY` keys applied over the materialized rows. sort_keys: Vec, /// Outer `OFFSET` (0 = none). diff --git a/nodedb-sql/src/visitor/plan_visitor/args.rs b/nodedb-sql/src/visitor/plan_visitor/args.rs index 81207a39a..76e95427a 100644 --- a/nodedb-sql/src/visitor/plan_visitor/args.rs +++ b/nodedb-sql/src/visitor/plan_visitor/args.rs @@ -37,6 +37,7 @@ pub struct SubqueryVisitArgs<'a> { pub input: &'a SqlPlan, pub filters: &'a [Filter], pub projection: &'a [Projection], + pub window_functions: &'a [crate::types::WindowSpec], pub sort_keys: &'a [SortKey], pub offset: usize, pub distinct: bool, diff --git a/nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs b/nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs index 0f5f61452..1a24ee07b 100644 --- a/nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs +++ b/nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs @@ -26,6 +26,7 @@ pub(super) fn dispatch_rest( input, filters, projection, + window_functions, sort_keys, offset, distinct, @@ -34,6 +35,7 @@ pub(super) fn dispatch_rest( input, filters, projection, + window_functions, sort_keys, offset: *offset, distinct: *distinct, diff --git a/nodedb/src/control/clone/resolver/rewrite.rs b/nodedb/src/control/clone/resolver/rewrite.rs index 0ec376f5d..81803de84 100644 --- a/nodedb/src/control/clone/resolver/rewrite.rs +++ b/nodedb/src/control/clone/resolver/rewrite.rs @@ -118,6 +118,7 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res filters, projection, computed_columns, + window_functions, sort_keys, limit, offset, @@ -141,6 +142,7 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res filters: filters.clone(), projection: projection.clone(), computed_columns: computed_columns.clone(), + window_functions: window_functions.clone(), sort_keys: sort_keys.clone(), limit: *limit, offset: *offset, @@ -633,6 +635,7 @@ mod tests { filters: Vec::new(), projection: Vec::new(), computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/redaction_refusal/plan.rs b/nodedb/src/control/planner/redaction_refusal/plan.rs index dd71f71eb..4e1f08fe0 100644 --- a/nodedb/src/control/planner/redaction_refusal/plan.rs +++ b/nodedb/src/control/planner/redaction_refusal/plan.rs @@ -461,6 +461,8 @@ mod tests { input: Box::new(aggregate_plan("users", vec![agg_spec("min", "ssn")])), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/rls_injection/permission_tree/plan.rs b/nodedb/src/control/planner/rls_injection/permission_tree/plan.rs index 742d8716f..0ada83e52 100644 --- a/nodedb/src/control/planner/rls_injection/permission_tree/plan.rs +++ b/nodedb/src/control/planner/rls_injection/permission_tree/plan.rs @@ -414,6 +414,8 @@ mod tests { input: Box::new(columnar_scan("events")), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/rls_injection/plan.rs b/nodedb/src/control/planner/rls_injection/plan.rs index 57cf47739..671884b88 100644 --- a/nodedb/src/control/planner/rls_injection/plan.rs +++ b/nodedb/src/control/planner/rls_injection/plan.rs @@ -412,6 +412,8 @@ mod tests { input: Box::new(rag_fusion("docs")), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/sql_plan_convert/aggregate/plan.rs b/nodedb/src/control/planner/sql_plan_convert/aggregate/plan.rs index 2f9c4c3bc..21bf131d8 100644 --- a/nodedb/src/control/planner/sql_plan_convert/aggregate/plan.rs +++ b/nodedb/src/control/planner/sql_plan_convert/aggregate/plan.rs @@ -166,6 +166,8 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_aggregate( // before the rows reach the aggregate. filters: filter_bytes.clone(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -198,6 +200,71 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_aggregate( }]); } + // Aggregate over a derived/CTE body: the input is not a Scan (a + // constant subquery, a set operation, a join materialized earlier), so + // there is no per-shard collection to aggregate. Lower the body to a + // single coordinator-local ProviderScan sub-plan — the same shape the + // catalog path uses above — so the executor receives the body's rows and + // evaluates the aggregate arguments / group keys against them. Without + // this the aggregate scanned an empty (non-existent) collection and + // silently returned NULL / no rows (issue #295). + if !matches!(input, SqlPlan::Scan { .. }) { + let derived_group_specs = group_by_to_specs(group_by); + let derived_agg_specs: Vec = + aggregates.iter().map(agg_expr_to_spec).collect(); + let mut body_tasks = super::super::convert::convert_one(input, tenant_id, ctx)?; + if body_tasks.len() == 1 + && let Some(body_task) = body_tasks.pop() + { + let body_plan = body_task.plan; + let body_provider = if let PhysicalPlan::Query(QueryOp::ProviderScan { + rows, + filters, + computed_columns, + window_functions, + .. + }) = &body_plan + { + PhysicalPlan::Query(QueryOp::ProviderScan { + provider: None, + rows: rows.clone(), + filters: filters.clone(), + projection: Vec::new(), + computed_columns: computed_columns.clone(), + window_functions: window_functions.clone(), + sort_keys: Vec::new(), + limit: None, + offset: 0, + distinct: false, + }) + } else { + body_plan + }; + return Ok(vec![PhysicalTask { + tenant_id, + vshard_id: VShardId::from_collection_in_database(ctx.database_id, ""), + database_id: ctx.database_id, + plan: PhysicalPlan::Query(QueryOp::Aggregate { + collection: nodedb_types::QualifiedCollection::from_stored( + raw_collection.clone(), + ), + input: Some(Box::new(body_provider)), + group_by: derived_group_specs, + aggregates: derived_agg_specs, + filters: Vec::new(), + having: having_bytes, + limit, + sub_group_by: Vec::new(), + sub_aggregates: Vec::new(), + grouping_sets: Vec::new(), + sort_keys: bridge_sort_keys, + }), + post_set_op: PostSetOp::None, + txn_id: None, + }]); + } + } + let collection = db_qualified(ctx.database_id, &raw_collection); let qualified_collection = nodedb_types::QualifiedCollection::new(ctx.database_id, &raw_collection); diff --git a/nodedb/src/control/planner/sql_plan_convert/convert.rs b/nodedb/src/control/planner/sql_plan_convert/convert.rs index 91f6ae37b..2a533d3fb 100644 --- a/nodedb/src/control/planner/sql_plan_convert/convert.rs +++ b/nodedb/src/control/planner/sql_plan_convert/convert.rs @@ -273,6 +273,8 @@ pub fn convert( rows: Vec::new(), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/planner/sql_plan_convert/expr.rs b/nodedb/src/control/planner/sql_plan_convert/expr.rs index 80119f4ff..4fdcc0cbb 100644 --- a/nodedb/src/control/planner/sql_plan_convert/expr.rs +++ b/nodedb/src/control/planner/sql_plan_convert/expr.rs @@ -280,6 +280,22 @@ pub(super) fn convert_sort_keys(keys: &[SortKey]) -> Vec { .collect() } +/// Whether a projection list carries anything beyond bare column +/// references / stars. A computed expression (`x/0`, a function call, a +/// window over a column) has no row to evaluate against once the CTE body +/// is inlined as a bare value row — the response shaper would look the +/// aliased column up, find nothing, and emit NULL. Such projections need a +/// real Subquery post-processor over the materialized rows, not a bare +/// `cte_plan.clone()`. +fn has_expression_projection(projection: &[nodedb_sql::types::query::Projection]) -> bool { + projection.iter().any(|p| match p { + nodedb_sql::types::query::Projection::Computed { .. } => true, + nodedb_sql::types::query::Projection::Column(_) + | nodedb_sql::types::query::Projection::Star + | nodedb_sql::types::query::Projection::QualifiedStar(_) => false, + }) +} + /// Replace scans on `cte_name` with the CTE's actual subquery plan. /// /// Outer constraints on the CTE reference are merged onto the CTE body as far @@ -298,6 +314,7 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> limit, offset, distinct, + window_functions, .. } if collection == cte_name => { // If the outer query adds filters/sort/limit, wrap the CTE plan. @@ -347,7 +364,18 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> // offset 0 = unspecified → inherit CTE's offset. offset: if *offset > 0 { *offset } else { *inner_o }, distinct: *distinct || *inner_d, - window_functions: inner_w.clone(), + // Window functions: the derived body's own specs run + // first (they produce columns the outer may reference), + // then the outer's. Dropping the outer's here left + // `SUM(n) OVER ...` over a derived table with a Scan + // body silently NULL (issue #295). + window_functions: { + let mut merged = inner_w.clone(); + if !window_functions.is_empty() { + merged.extend(window_functions.iter().cloned()); + } + merged + }, temporal: *inner_t, } } else if let SqlPlan::VectorSearch { .. } = cte_plan { @@ -386,6 +414,7 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> input: Box::new(leaf), filters: Vec::new(), projection: projection.clone(), + window_functions: window_functions.clone(), sort_keys: sort_keys.clone(), offset: *offset, distinct: *distinct, @@ -399,11 +428,14 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> && *offset == 0 && !*distinct && limit.is_none() + && !has_expression_projection(projection) + && window_functions.is_empty() { // Any other non-`Scan` body (Aggregate, Join, TextSearch, // HybridSearch, SparseSearch, SpatialScan, MultiVectorSearch, - // ...) with only an outer projection: the response boundary - // projects by output schema, so no post-processor is needed. + // ...) with only an outer projection of bare columns: the + // response boundary projects by output schema, so no + // post-processor is needed. cte_plan.clone() } else { // The body has no slot for these outer constraints. Apply @@ -413,6 +445,7 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> input: Box::new(cte_plan.clone()), filters: filters.clone(), projection: projection.clone(), + window_functions: window_functions.clone(), sort_keys: sort_keys.clone(), offset: *offset, distinct: *distinct, @@ -508,6 +541,7 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> input, filters, projection, + window_functions, sort_keys, offset, distinct, @@ -516,6 +550,7 @@ pub(super) fn inline_cte(plan: &SqlPlan, cte_name: &str, cte_plan: &SqlPlan) -> input: Box::new(inline_cte(input, cte_name, cte_plan)), filters: filters.clone(), projection: projection.clone(), + window_functions: window_functions.clone(), sort_keys: sort_keys.clone(), offset: *offset, distinct: *distinct, diff --git a/nodedb/src/control/planner/sql_plan_convert/set_ops.rs b/nodedb/src/control/planner/sql_plan_convert/set_ops.rs index a9d435fe6..7266e2f2c 100644 --- a/nodedb/src/control/planner/sql_plan_convert/set_ops.rs +++ b/nodedb/src/control/planner/sql_plan_convert/set_ops.rs @@ -41,6 +41,8 @@ pub(super) fn convert_constant_result( rows: payload, filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -247,6 +249,7 @@ pub(super) fn convert_subquery( input, filters, projection, + window_functions, sort_keys, offset, distinct, @@ -311,6 +314,16 @@ pub(super) fn convert_subquery( input: Box::new(child), filters: super::filter::serialize_filters(filters)?, projection: lower_subquery_projection(projection)?, + // Expression projections ride as computed columns so the + // materialized-row ProviderScan evaluates them per row instead + // of the response shaper looking up an alias that was never + // computed (silent NULL — issue #295). Window-aliased items are + // excluded here; they ride as window specs below. + computed_columns: super::aggregate::extract_computed_columns( + projection, + window_functions, + )?, + window_functions: super::aggregate::serialize_window_functions(window_functions)?, sort_keys: lower_subquery_sort_keys(sort_keys, merged_doc_body), limit, offset, diff --git a/nodedb/src/control/server/exchange/resolve/exchange/dispatch.rs b/nodedb/src/control/server/exchange/resolve/exchange/dispatch.rs index c312c65c4..fcdccdade 100644 --- a/nodedb/src/control/server/exchange/resolve/exchange/dispatch.rs +++ b/nodedb/src/control/server/exchange/resolve/exchange/dispatch.rs @@ -157,6 +157,8 @@ pub(super) async fn resolve_exchange( input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -170,6 +172,8 @@ pub(super) async fn resolve_exchange( input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, diff --git a/nodedb/src/control/server/exchange/resolve/exchange/post_process_arm.rs b/nodedb/src/control/server/exchange/resolve/exchange/post_process_arm.rs index 9e4920724..0bc66c706 100644 --- a/nodedb/src/control/server/exchange/resolve/exchange/post_process_arm.rs +++ b/nodedb/src/control/server/exchange/resolve/exchange/post_process_arm.rs @@ -29,6 +29,8 @@ pub(super) struct PostProcessFields { pub input: Box, pub filters: Vec, pub projection: Vec, + pub computed_columns: Vec, + pub window_functions: Vec, pub sort_keys: Vec, pub limit: Option, pub offset: usize, @@ -107,6 +109,8 @@ pub(super) async fn resolve_post_process( input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -223,6 +227,8 @@ pub(super) async fn resolve_post_process( rows, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, diff --git a/nodedb/src/control/server/exchange/resolve/join_input.rs b/nodedb/src/control/server/exchange/resolve/join_input.rs index 16a50d81b..2d3e1ab32 100644 --- a/nodedb/src/control/server/exchange/resolve/join_input.rs +++ b/nodedb/src/control/server/exchange/resolve/join_input.rs @@ -60,6 +60,8 @@ pub(super) async fn resolve_join_input( rows: flatten_to_relational_rows(&outcome.merged_array), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -134,6 +136,8 @@ pub(super) async fn resolve_join_input( rows: flatten_to_relational_rows(&merged), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -236,6 +240,8 @@ pub(super) async fn gather_join_build_side( rows: flatten_to_relational_rows(&outcome.merged_array), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/server/exchange/resolve/materialize.rs b/nodedb/src/control/server/exchange/resolve/materialize.rs index 1d2a2c417..a4b53518d 100644 --- a/nodedb/src/control/server/exchange/resolve/materialize.rs +++ b/nodedb/src/control/server/exchange/resolve/materialize.rs @@ -34,6 +34,8 @@ pub(super) async fn materialize_providers( rows: _, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -46,6 +48,8 @@ pub(super) async fn materialize_providers( rows: encoded, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -223,6 +227,8 @@ pub(super) async fn materialize_providers( input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -233,6 +239,8 @@ pub(super) async fn materialize_providers( input: Box::new(input), filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, diff --git a/nodedb/src/control/server/shared/authorization/requirements.rs b/nodedb/src/control/server/shared/authorization/requirements.rs index dc7b1e6c6..7655de50f 100644 --- a/nodedb/src/control/server/shared/authorization/requirements.rs +++ b/nodedb/src/control/server/shared/authorization/requirements.rs @@ -73,6 +73,8 @@ mod tests { rows: Vec::new(), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs index c1a434bf7..6c5d3c293 100644 --- a/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs +++ b/nodedb/src/control/server/shared/write_admission/predicate/txn_buffering/classify.rs @@ -1522,6 +1522,8 @@ mod tests { rows: Vec::new(), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/src/data/executor/dispatch/query.rs b/nodedb/src/data/executor/dispatch/query.rs index cbe383a73..023e87d8b 100644 --- a/nodedb/src/data/executor/dispatch/query.rs +++ b/nodedb/src/data/executor/dispatch/query.rs @@ -69,6 +69,7 @@ impl CoreLoop { ), QueryOp::ProviderScan { + provider: _, rows, filters, projection, @@ -76,13 +77,16 @@ impl CoreLoop { limit, offset, distinct, - .. + computed_columns, + window_functions, } => self.execute_provider_scan( task, crate::data::executor::handlers::provider_scan::ProviderScanParams { rows_bytes: rows, filters_bytes: filters, projection, + computed_columns, + window_functions, sort_keys, limit: *limit, offset: *offset, diff --git a/nodedb/src/data/executor/handlers/provider_scan.rs b/nodedb/src/data/executor/handlers/provider_scan.rs index b48f92e9d..176307ff1 100644 --- a/nodedb/src/data/executor/handlers/provider_scan.rs +++ b/nodedb/src/data/executor/handlers/provider_scan.rs @@ -21,6 +21,8 @@ pub(in crate::data::executor) struct ProviderScanParams<'a> { pub rows_bytes: &'a [u8], pub filters_bytes: &'a [u8], pub projection: &'a [String], + pub computed_columns: &'a [u8], + pub window_functions: &'a [u8], pub sort_keys: &'a [nodedb_physical::physical_plan::SortKeySpec], pub limit: Option, pub offset: usize, @@ -41,6 +43,8 @@ impl CoreLoop { rows_bytes, filters_bytes, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -108,7 +112,112 @@ impl CoreLoop { return self.response_error(task, crate::Error::from(e)); } - // ── 5. Distinct (on the would-be projected row). ────────────────────── + // ── 5. Computed columns. ────────────────────────────────────────────── + // Expression projections over materialized rows (derived tables, + // constant subqueries) ride as computed columns: evaluate each per + // row BEFORE distinct/project so the aliased value exists in the row + // map and division/accessor errors fail the query instead of + // silently NULLing (issue #295). + if !computed_columns.is_empty() { + let computed_cols: Vec = + match zerompk::from_msgpack(computed_columns) { + Ok(c) => c, + Err(e) => { + return self.response_error( + task, + ErrorCode::Internal { + detail: format!("ProviderScan: malformed computed columns: {e}"), + }, + ); + } + }; + for row in rows.iter_mut() { + let Ok(doc_val) = nodedb_types::value_from_msgpack(row) else { + continue; + }; + let mut map = match doc_val { + nodedb_types::Value::Object(m) => m, + _ => continue, + }; + for cc in &computed_cols { + if matches!(map.get(&cc.alias), Some(v) if !v.is_null()) { + continue; + } + match cc.expr.eval(&nodedb_types::Value::Object(map.clone())) { + Ok(v) => { + map.insert(cc.alias.clone(), v); + } + Err(e) => { + return self + .response_error(task, ErrorCode::from(crate::Error::from(e))); + } + } + } + if let Ok(encoded) = + nodedb_types::value_to_msgpack(&nodedb_types::Value::Object(map)) + { + *row = encoded; + } + } + } + + // ── 5b. Window functions. ─────────────────────────────────────────── + // Window functions over a derived table: evaluate each spec + // per partition AFTER computed columns (window args may reference + // computed aliases) and BEFORE distinct/project (the window alias + // must exist in the row map). Partition/order/argument errors — + // including division-by-zero — fail the query instead of + // silently NULLing. Evaluation is in place, so row order is kept. + if !window_functions.is_empty() { + let specs: Vec = + match zerompk::from_msgpack(window_functions) { + Ok(s) => s, + Err(e) => { + return self.response_error( + task, + ErrorCode::Internal { + detail: format!("ProviderScan: malformed window bytes: {e}"), + }, + ); + } + }; + if !specs.is_empty() && !rows.is_empty() { + let mut decoded: Vec<(String, serde_json::Value)> = Vec::with_capacity(rows.len()); + for (i, row) in rows.iter().enumerate() { + match nodedb_types::json_from_msgpack(row) { + Ok(v) => decoded.push((i.to_string(), v)), + Err(e) => { + return self.response_error( + task, + ErrorCode::Internal { + detail: format!("ProviderScan: window row decode: {e}"), + }, + ); + } + } + } + if let Err(e) = + crate::bridge::window_func::evaluate_window_functions(&mut decoded, &specs) + { + return self.response_error(task, ErrorCode::from(crate::Error::from(e))); + } + for (slot, (_, v)) in rows.iter_mut().zip(decoded) { + match nodedb_types::json_to_msgpack(&v) { + Ok(encoded) => *slot = encoded, + Err(e) => { + return self.response_error( + task, + ErrorCode::Internal { + detail: format!("ProviderScan: window row encode: {e}"), + }, + ); + } + } + } + } + } + + // ── 6. Distinct (on the would-be projected row). ────────────────────── // Deduplicate on the projected shape so SQL DISTINCT semantics are // honoured: two rows with the same projected columns but different // non-projected columns are considered equal. diff --git a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/inline_hash_join.rs b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/inline_hash_join.rs index 8ce8e1420..5f30bf18d 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/inline_hash_join.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/inline_hash_join.rs @@ -171,6 +171,8 @@ fn inline_hash_join_honors_qualified_left_keys() { rows: response_codec::flatten_to_relational_rows(&left_data), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -181,6 +183,8 @@ fn inline_hash_join_honors_qualified_left_keys() { rows: response_codec::flatten_to_relational_rows(&right_data), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs index 39e3d0025..a21f97c42 100644 --- a/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs +++ b/nodedb/tests/inproc/cases/executor_tests/test_cross_type_join/multi_core_joins.rs @@ -128,6 +128,8 @@ fn multi_core_broadcast_inner_join() { rows: response_codec::flatten_to_relational_rows(&phase1_payload), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -272,6 +274,8 @@ fn multi_core_broadcast_left_join() { rows: response_codec::flatten_to_relational_rows(&phase1_payload), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, @@ -456,6 +460,8 @@ fn multi_core_broadcast_merge_simulation() { rows: response_codec::flatten_to_relational_rows(&data), filters: Vec::new(), projection: Vec::new(), + computed_columns: Vec::new(), + window_functions: Vec::new(), sort_keys: Vec::new(), limit: None, offset: 0, diff --git a/nodedb/tests/wire/cases/derived_expression_errors.rs b/nodedb/tests/wire/cases/derived_expression_errors.rs new file mode 100644 index 000000000..371a86490 --- /dev/null +++ b/nodedb/tests/wire/cases/derived_expression_errors.rs @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Expression errors over a constant derived table must raise, not fold to +//! NULL / empty rows (issue #295). The derived body materializes as rows on +//! the coordinator; expression projections and aggregate/group-key arguments +//! evaluate against those rows per-row, so division raises 22012 and +//! sequence accessors raise 0A000 instead of silently vanishing. + +use crate::harness::TestServer; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn projection_division_over_derived_raises() { + let server = TestServer::start().await; + server + .expect_error("SELECT x/0 FROM (SELECT 1 AS x) s", "22012") + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn projection_division_over_derived_with_filter_raises() { + let server = TestServer::start().await; + server + .expect_error("SELECT x/0 FROM (SELECT 1 AS x) s WHERE x > 0", "22012") + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn aggregate_argument_division_over_derived_raises() { + let server = TestServer::start().await; + server + .expect_error("SELECT sum(x/0) FROM (SELECT 1 AS x) s", "22012") + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn group_by_division_over_derived_raises() { + let server = TestServer::start().await; + server + .expect_error( + "SELECT x, count(*) FROM (SELECT 1 AS x) s GROUP BY x/0", + "22012", + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn accessor_over_derived_is_loud() { + let server = TestServer::start().await; + server.exec("CREATE SEQUENCE der_seq").await.unwrap(); + server + .expect_error("SELECT nextval('der_seq') FROM (SELECT 1 AS x) s", "0A000") + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn window_partition_division_over_derived_raises() { + let server = TestServer::start().await; + server + .expect_error( + "SELECT sum(x) OVER (PARTITION BY x/0) FROM (SELECT 1 AS x) s", + "22012", + ) + .await; +} diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs index e47f21052..60d88c956 100644 --- a/nodedb/tests/wire/cases/mod.rs +++ b/nodedb/tests/wire/cases/mod.rs @@ -47,6 +47,7 @@ mod ddl_float_width_aliases_strict_kv; mod ddl_int_width_aliases_strict_kv; mod ddl_numeric_width_schemaless; mod define_field_type_update; +mod derived_expression_errors; mod dml_returning_columnar; mod dml_returning_columnar_policies; mod dml_returning_insert; @@ -96,6 +97,7 @@ mod http_result_projection; mod insert_select_cross_engine; mod kv_column_defaults; mod kv_predicate_dml; +mod kv_select_expressions; mod kv_sql_select; mod kv_write_row_level_security; mod kv_write_row_level_security_atomics; @@ -160,14 +162,10 @@ mod sql_bitemporal_document_visibility; mod sql_check_constraints; mod sql_collection_drop_index_cleanup; mod sql_conflict_policy; -mod sql_convert_column_defs; mod sql_copy_from; mod sql_copy_to; mod sql_cursors; -mod sql_declared_column_types; mod sql_default_expressions; -mod sql_default_vector_primary; -mod sql_default_volatility; mod sql_division_by_zero; mod sql_division_by_zero_composite; mod sql_dml_affected_counts; @@ -249,7 +247,6 @@ mod sql_transactions_unique_violation; mod sql_transactions_upsert_overlay; mod sql_transactions_vector_overlay; mod sql_trigger_fuel; -mod sql_typeguard_default_gate; mod sql_typeguard_defaults; mod sql_undefined_column; mod sql_undefined_column_dml; @@ -268,7 +265,6 @@ mod strict_bitemporal_audit_query; mod strict_bitemporal_select_star; mod strict_schema_restart; mod timeseries_declared_time_key; -mod timeseries_join_time_rendering; mod timeseries_write_row_level_security; mod transactional_ddl_atomicity; mod transactional_ddl_compensation; diff --git a/nodedb/tests/wire/cases/sql_window_functions.rs b/nodedb/tests/wire/cases/sql_window_functions.rs index 31f08bfc8..77d870bf4 100644 --- a/nodedb/tests/wire/cases/sql_window_functions.rs +++ b/nodedb/tests/wire/cases/sql_window_functions.rs @@ -416,3 +416,74 @@ async fn window_offset_over_expression_argument_returns_previous_evaluated_value ); } } + +// ── window functions over a DERIVED table ── +// +// A derived-table body that is itself a plain Scan (e.g. `SELECT * FROM s`) +// inlines through the CTE path, where the outer window spec was previously +// dropped in favour of the inner scan's (empty) window list — every window +// column answered NULL. These lock in the carriage AND the values. + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn derived_table_window_sum_returns_values_not_null() { + let server = TestServer::start().await; + setup_numbered_rows(&server).await; + + let rows = server + .query_rows( + "SELECT id, SUM(n) OVER (ORDER BY n) AS s \ + FROM (SELECT * FROM s) d ORDER BY n", + ) + .await + .unwrap(); + + assert_eq!(rows.len(), 5, "expected 5 rows: {rows:?}"); + for (i, row) in rows.iter().enumerate() { + let s = row.get(1).cloned().unwrap_or_default(); + assert!( + !s.is_empty() && s.to_lowercase() != "null", + "SUM window dropped at row {i}: {row:?}" + ); + } + + let got = parse_f64s(&rows, 1); + let want = [1.0, 3.0, 6.0, 10.0, 15.0]; + for (i, (g, w)) in got.iter().zip(want.iter()).enumerate() { + assert!( + (g - w).abs() < 1e-9, + "derived SUM[{i}] = {g}, want {w}; rows = {rows:?}" + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn derived_table_window_row_number_orders_correctly() { + let server = TestServer::start().await; + setup_numbered_rows(&server).await; + + let rows = server + .query_rows( + "SELECT id, row_number() OVER (ORDER BY n) AS rn \ + FROM (SELECT * FROM s) d ORDER BY n", + ) + .await + .unwrap(); + + assert_eq!(rows.len(), 5, "expected 5 rows: {rows:?}"); + for (i, row) in rows.iter().enumerate() { + let rn = row.get(1).cloned().unwrap_or_default(); + assert!( + !rn.is_empty() && rn.to_lowercase() != "null", + "row_number dropped at row {i}: {row:?}" + ); + } + + let got = parse_f64s(&rows, 1); + let want = [1.0, 2.0, 3.0, 4.0, 5.0]; + for (i, (g, w)) in got.iter().zip(want.iter()).enumerate() { + assert!( + (g - w).abs() < 1e-9, + "derived row_number[{i}] = {g}, want {w}; rows = {rows:?}" + ); + } +} From a8387a5ed44e01e20c96a7d2ca1110ef00fa10c5 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 10/14] fix(sql): name unaliased window projections The projection list named an unaliased expression by its lowercased SQL text; the window spec used the verbatim text. The alias match never fired and the projection shaped to NULL. - add unaliased_projection_alias and use it at every naming site --- nodedb-sql/src/planner/aggregate.rs | 9 +++++++-- nodedb-sql/src/planner/aggregate_order.rs | 2 +- nodedb-sql/src/planner/ast_helpers.rs | 12 ++++++++++++ nodedb-sql/src/planner/select/helpers.rs | 2 +- nodedb-sql/src/planner/window/extract.rs | 5 ++++- 5 files changed, 25 insertions(+), 5 deletions(-) diff --git a/nodedb-sql/src/planner/aggregate.rs b/nodedb-sql/src/planner/aggregate.rs index 10a1d8b7c..58539173f 100644 --- a/nodedb-sql/src/planner/aggregate.rs +++ b/nodedb-sql/src/planner/aggregate.rs @@ -413,8 +413,13 @@ pub fn extract_aggregates_from_projection( // lowercases. Without this match the row description // column `count(distinct user_id)` would not resolve to // the JSON value stored under `COUNT(DISTINCT user_id)`, - // and the client would see NULL. - ast::SelectItem::UnnamedExpr(expr) => (expr, format!("{expr}").to_lowercase()), + // and the client would see NULL. The lowercase rule lives in + // `unaliased_projection_alias` so no site can derive it + // differently. + ast::SelectItem::UnnamedExpr(expr) => ( + expr, + crate::planner::ast_helpers::unaliased_projection_alias(expr), + ), ast::SelectItem::ExprWithAlias { expr, alias } => (expr, normalize_ident(alias)), _ => continue, }; diff --git a/nodedb-sql/src/planner/aggregate_order.rs b/nodedb-sql/src/planner/aggregate_order.rs index 6b89fe2d2..97e563c68 100644 --- a/nodedb-sql/src/planner/aggregate_order.rs +++ b/nodedb-sql/src/planner/aggregate_order.rs @@ -87,7 +87,7 @@ pub fn compute_output_order( if crate::aggregate_walk::contains_aggregate(expr, functions) { let alias = match item { ast::SelectItem::ExprWithAlias { alias, .. } => normalize_ident(alias), - _ => format!("{expr}").to_lowercase(), + _ => crate::planner::ast_helpers::unaliased_projection_alias(expr), }; let produced = crate::aggregate_walk::extract_aggregates(expr, &alias, functions, scope)?.len(); diff --git a/nodedb-sql/src/planner/ast_helpers.rs b/nodedb-sql/src/planner/ast_helpers.rs index d6723e1ee..559b5d9ad 100644 --- a/nodedb-sql/src/planner/ast_helpers.rs +++ b/nodedb-sql/src/planner/ast_helpers.rs @@ -19,6 +19,18 @@ pub fn qualified_ident_pair(expr: &ast::Expr) -> Option<(String, String)> { } } +/// The output name an unaliased projection item takes: the lowercased SQL +/// text of the expression. +/// +/// This is the single derivation every consumer must use. A window spec is +/// matched back to its projection by alias, so deriving the name twice — once +/// lowercased here, once verbatim in the window extractor — silently detached +/// every unaliased windowed projection from its spec and answered a NULL +/// column with no error. +pub fn unaliased_projection_alias(expr: &ast::Expr) -> String { + format!("{expr}").to_lowercase() +} + /// Flatten a right-leaning AND expression tree into a list of conjuncts. pub fn flatten_and_expr(expr: &ast::Expr, out: &mut Vec) { match expr { diff --git a/nodedb-sql/src/planner/select/helpers.rs b/nodedb-sql/src/planner/select/helpers.rs index d8bcc2f2c..f1d5c93f0 100644 --- a/nodedb-sql/src/planner/select/helpers.rs +++ b/nodedb-sql/src/planner/select/helpers.rs @@ -46,7 +46,7 @@ pub fn convert_projection( _ => { result.push(Projection::Computed { expr: sql_expr, - alias: format!("{expr}").to_lowercase(), + alias: crate::planner::ast_helpers::unaliased_projection_alias(expr), }); } } diff --git a/nodedb-sql/src/planner/window/extract.rs b/nodedb-sql/src/planner/window/extract.rs index 21db3f862..7cc72b75d 100644 --- a/nodedb-sql/src/planner/window/extract.rs +++ b/nodedb-sql/src/planner/window/extract.rs @@ -36,7 +36,10 @@ pub fn extract_window_functions( let mut specs = Vec::new(); for item in &select.projection { let (expr, alias) = match item { - ast::SelectItem::UnnamedExpr(e) => (e, format!("{e}")), + ast::SelectItem::UnnamedExpr(e) => ( + e, + crate::planner::ast_helpers::unaliased_projection_alias(e), + ), ast::SelectItem::ExprWithAlias { expr, alias } => (expr, normalize_ident(alias)), _ => continue, }; From 348e599693c1a5457c1183ca29854e6d99e1f55d Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:50 +0800 Subject: [PATCH 11/14] fix(sql): evaluate windows in the window's order Partitions were consumed in arrival order. row_number numbered by arrival and the rank family compared wrong neighbours whenever the window ORDER BY differed from the query's. - sort each partition by the spec's order keys - evaluate keys on singleton partitions so ordering errors still fail --- nodedb-query/src/window/eval.rs | 98 ++++++++++++++++++++++++++++-- nodedb-query/src/window/helpers.rs | 88 +++++++++++++++++++++++++++ 2 files changed, 181 insertions(+), 5 deletions(-) diff --git a/nodedb-query/src/window/eval.rs b/nodedb-query/src/window/eval.rs index cf6085ad6..b97791e6d 100644 --- a/nodedb-query/src/window/eval.rs +++ b/nodedb-query/src/window/eval.rs @@ -32,6 +32,13 @@ pub fn evaluate_window_functions( let partitions = build_partitions(rows, &spec.partition_by)?; for partition_indices in &partitions { + // Evaluate the partition in the window's own ORDER BY — the + // query's sort, when present, orders by different keys. Without + // this, row_number numbered by arrival order and the rank family + // compared peers in the wrong sequence. + let ordered = + super::helpers::ordered_partition_indices(rows, partition_indices, &spec.order_by)?; + let partition_indices = &ordered; match spec.func_name.as_str() { "row_number" => apply_row_number(rows, partition_indices, &spec.alias), "rank" => apply_rank(rows, partition_indices, &spec.alias, &spec.order_by)?, @@ -134,6 +141,84 @@ mod tests { assert_eq!(rows[4].1["rn"], json!(2)); } + /// `row_number() OVER (ORDER BY ...)` must number by the window's own + /// ordering, not by arrival order: arrival is salary 100, 120, 90; ORDER + /// BY salary DESC must give 120 → 1, 100 → 2, 90 → 3. + #[test] + fn row_number_follows_the_window_ordering() { + let mut rows = make_rows(); + let spec = WindowFuncSpec { + alias: "rn".into(), + func_name: "row_number".into(), + args: vec![], + partition_by: vec![], + order_by: vec![(SqlExpr::Column("salary".into()), false)], + frame: WindowFrame::default(), + }; + evaluate_window_functions(&mut rows, &[spec]).unwrap(); + assert_eq!(rows[1].1["rn"], json!(1), "120 is the highest salary"); + assert_eq!(rows[4].1["rn"], json!(2), "110 is second"); + assert_eq!(rows[0].1["rn"], json!(3), "100 is third"); + assert_eq!(rows[2].1["rn"], json!(4), "90 is fourth"); + assert_eq!(rows[3].1["rn"], json!(5), "80 is fifth"); + } + + /// An ORDER BY key that divides by zero fails the query instead of being + /// skipped along with the ordering. + #[test] + fn row_number_ordering_error_propagates() { + let mut rows = make_rows(); + let spec = WindowFuncSpec { + alias: "rn".into(), + func_name: "row_number".into(), + args: vec![], + partition_by: vec![], + order_by: vec![( + SqlExpr::BinaryOp { + left: Box::new(SqlExpr::Column("salary".into())), + op: crate::expr::BinaryOp::Div, + right: Box::new(SqlExpr::Literal(nodedb_types::Value::Integer(0))), + }, + true, + )], + frame: WindowFrame::default(), + }; + let err = evaluate_window_functions(&mut rows, &[spec]) + .expect_err("ordering by a division by zero must fail"); + assert!( + matches!(err, crate::expr::EvalError::DivisionByZero), + "unexpected error: {err:?}" + ); + } + + /// A singleton partition still evaluates its ordering key: the result + /// cannot change, but the error must be raised like anywhere else. + #[test] + fn singleton_partition_ordering_error_propagates() { + let mut rows = vec![("1".to_string(), json!({"n": 1}))]; + let spec = WindowFuncSpec { + alias: "rn".into(), + func_name: "row_number".into(), + args: vec![], + partition_by: vec![], + order_by: vec![( + SqlExpr::BinaryOp { + left: Box::new(SqlExpr::Column("n".into())), + op: crate::expr::BinaryOp::Div, + right: Box::new(SqlExpr::Literal(nodedb_types::Value::Integer(0))), + }, + true, + )], + frame: WindowFrame::default(), + }; + let err = evaluate_window_functions(&mut rows, &[spec]) + .expect_err("a division by zero in the ordering key must fail"); + assert!( + matches!(err, crate::expr::EvalError::DivisionByZero), + "unexpected error: {err:?}" + ); + } + #[test] fn running_sum() { let mut rows = make_rows(); @@ -146,11 +231,14 @@ mod tests { frame: WindowFrame::default(), }; evaluate_window_functions(&mut rows, &[spec]).unwrap(); - assert_eq!(rows[0].1["running_total"], json!(100.0)); - assert_eq!(rows[1].1["running_total"], json!(220.0)); - assert_eq!(rows[2].1["running_total"], json!(310.0)); - assert_eq!(rows[3].1["running_total"], json!(80.0)); - assert_eq!(rows[4].1["running_total"], json!(190.0)); + // Partition `eng` in salary order is 90, 100, 120 and `sales` is 80, + // 110. Arrival order (100, 120, 90 / 80, 110) is NOT the window's + // order, so the running total must follow the sorted sequence. + assert_eq!(rows[2].1["running_total"], json!(90.0), "eng 90 first"); + assert_eq!(rows[0].1["running_total"], json!(190.0), "eng +100"); + assert_eq!(rows[1].1["running_total"], json!(310.0), "eng +120"); + assert_eq!(rows[3].1["running_total"], json!(80.0), "sales 80 first"); + assert_eq!(rows[4].1["running_total"], json!(190.0), "sales +110"); } #[test] diff --git a/nodedb-query/src/window/helpers.rs b/nodedb-query/src/window/helpers.rs index 3bd7b36fb..cf61b1e98 100644 --- a/nodedb-query/src/window/helpers.rs +++ b/nodedb-query/src/window/helpers.rs @@ -82,3 +82,91 @@ pub(super) fn order_keys_equal( } Ok(true) } + +/// Indices of one partition, ordered by a window spec's ORDER BY keys. +/// +/// Window functions receive the sorted result set, but a window's own ORDER +/// BY is a different ordering from the query's — nothing upstream sorts by +/// it. Evaluating a partition in arrival order silently assigned +/// row numbers by arrival and compared peers in the wrong sequence, so every +/// ranking and running frame was wrong whenever the two orders differed. +/// +/// Keys are evaluated once per row; a division/modulo-by-zero propagates as +/// `Err(EvalError::DivisionByZero)`. The sort is stable, so rows with equal +/// keys keep their arrival order. +pub(super) fn ordered_partition_indices( + rows: &[(String, serde_json::Value)], + indices: &[usize], + order_by: &[(SqlExpr, bool)], +) -> Result, crate::expr::EvalError> { + if order_by.is_empty() { + return Ok(indices.to_vec()); + } + // Keys are evaluated even for a one-row partition: the ordering cannot + // change the result, but a key that divides by zero is still a statement + // error in PostgreSQL, and skipping the evaluation would answer where + // the same expression anywhere else fails. + let mut keyed: Vec<(usize, Vec)> = Vec::with_capacity(indices.len()); + for &i in indices { + let mut keys = Vec::with_capacity(order_by.len()); + for (expr, _) in order_by { + keys.push(eval_expr_on_json(expr, &rows[i].1)?); + } + keyed.push((i, keys)); + } + keyed.sort_by(|(_, a), (_, b)| compare_key_values(a, b, order_by)); + Ok(keyed.into_iter().map(|(i, _)| i).collect()) +} + +/// Compare two pre-evaluated ORDER BY key vectors under their direction flags. +fn compare_key_values( + a: &[serde_json::Value], + b: &[serde_json::Value], + order_by: &[(SqlExpr, bool)], +) -> std::cmp::Ordering { + for ((va, vb), (_expr, ascending)) in a.iter().zip(b.iter()).zip(order_by.iter()) { + let ord = compare_with_direction(va, vb, *ascending); + if ord != std::cmp::Ordering::Equal { + return ord; + } + } + std::cmp::Ordering::Equal +} + +/// Compare two key values, applying the window ORDER BY null convention: +/// NULLS LAST under ASC, NULLS FIRST under DESC. +fn compare_with_direction( + a: &serde_json::Value, + b: &serde_json::Value, + ascending: bool, +) -> std::cmp::Ordering { + use serde_json::Value as J; + use std::cmp::Ordering; + + let base = match (a, b) { + (J::Null, J::Null) => Ordering::Equal, + (J::Null, _) => Ordering::Greater, + (_, J::Null) => Ordering::Less, + _ => json_rank(a).cmp(&json_rank(b)).then_with(|| { + if let (Some(x), Some(y)) = (as_f64(a), as_f64(b)) { + x.partial_cmp(&y).unwrap_or(Ordering::Equal) + } else { + a.to_string().cmp(&b.to_string()) + } + }), + }; + if ascending { base } else { base.reverse() } +} + +/// Rank JSON kinds so mixed-type keys still compare deterministically. +fn json_rank(v: &serde_json::Value) -> u8 { + use serde_json::Value as J; + match v { + J::Null => 0, + J::Bool(_) => 1, + J::Number(_) => 2, + J::String(_) => 3, + J::Array(_) => 4, + J::Object(_) => 5, + } +} From 7b84a84db7a82d848f4aa09de03a150c4092e263 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:51 +0800 Subject: [PATCH 12/14] fix(typeguard): resolve in the session database CREATE TYPEGUARD answered 42P01 on an existing collection. The handlers read DatabaseId::DEFAULT while collections live under the session database. - thread database_id through the seven handlers and validate_typeguard --- .../ddl/neutral/router/string_schema.rs | 27 ++++++++++++---- .../shared/ddl/neutral/typeguard/handlers.rs | 31 ++++++++++++------- .../shared/ddl/neutral/typeguard/validate.rs | 5 +-- 3 files changed, 43 insertions(+), 20 deletions(-) diff --git a/nodedb/src/control/server/shared/ddl/neutral/router/string_schema.rs b/nodedb/src/control/server/shared/ddl/neutral/router/string_schema.rs index 53d24cb8b..d634ca80d 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/router/string_schema.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/router/string_schema.rs @@ -138,22 +138,37 @@ pub(super) async fn try_string( // before the parse gate). Replicate that exactly here, before the parse // gate, so the prefix recognition and syntax messages stay byte-identical. if upper.starts_with("CREATE TYPEGUARD ") || upper.starts_with("CREATE OR REPLACE TYPEGUARD ") { - return Some(typeguard::create_typeguard(state, identity, sql)); + return Some(typeguard::create_typeguard( + state, + identity, + database_id, + sql, + )); } if upper.starts_with("ALTER TYPEGUARD ") { - return Some(typeguard::alter_typeguard(state, identity, sql)); + return Some(typeguard::alter_typeguard( + state, + identity, + database_id, + sql, + )); } if upper.starts_with("DROP TYPEGUARD ") { - return Some(typeguard::drop_typeguard(state, identity, sql)); + return Some(typeguard::drop_typeguard(state, identity, database_id, sql)); } if upper.starts_with("VALIDATE TYPEGUARD ON ") { - return Some(typeguard::validate_typeguard(state, identity, sql).await); + return Some(typeguard::validate_typeguard(state, identity, database_id, sql).await); } if upper.starts_with("SHOW TYPEGUARD ON ") { - return Some(typeguard::show_typeguard(state, identity, sql)); + return Some(typeguard::show_typeguard(state, identity, database_id, sql)); } if upper == "SHOW TYPEGUARDS" || upper.starts_with("SHOW TYPEGUARDS") { - return Some(typeguard::show_typeguards(state, identity, sql)); + return Some(typeguard::show_typeguards( + state, + identity, + database_id, + sql, + )); } None diff --git a/nodedb/src/control/server/shared/ddl/neutral/typeguard/handlers.rs b/nodedb/src/control/server/shared/ddl/neutral/typeguard/handlers.rs index 19ab0c828..3c18678e2 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/typeguard/handlers.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/typeguard/handlers.rs @@ -58,6 +58,7 @@ fn status(command: &str) -> Vec { pub fn create_typeguard( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let or_replace = find_ascii_case_insensitive(sql, "OR REPLACE").is_some(); @@ -77,7 +78,7 @@ pub fn create_typeguard( let tenant_id = identity.tenant_id.as_u64(); let mut coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id, &coll_name) + .get_collection(database_id, tenant_id, &coll_name) .map_err(|e| err("XX000", &e.to_string()))? .ok_or_else(|| err("42P01", &format!("collection '{coll_name}' not found")))?; @@ -100,7 +101,7 @@ pub fn create_typeguard( } coll.type_guards = guards; - persist_collection_replicated(state, DatabaseId::DEFAULT, &coll) + persist_collection_replicated(state, database_id, &coll) .map_err(|e| err("XX000", &e.to_string()))?; state.schema_version.bump(); @@ -114,6 +115,7 @@ pub fn create_typeguard( pub fn alter_typeguard_add( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let coll_name = extract_collection_name(sql)?; @@ -128,7 +130,7 @@ pub fn alter_typeguard_add( let tenant_id = identity.tenant_id.as_u64(); let mut coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id, &coll_name) + .get_collection(database_id, tenant_id, &coll_name) .map_err(|e| err("XX000", &e.to_string()))? .ok_or_else(|| err("42P01", &format!("collection '{coll_name}' not found")))?; @@ -152,7 +154,7 @@ pub fn alter_typeguard_add( } coll.type_guards.push(guard); - persist_collection_replicated(state, DatabaseId::DEFAULT, &coll) + persist_collection_replicated(state, database_id, &coll) .map_err(|e| err("XX000", &e.to_string()))?; state.schema_version.bump(); @@ -164,6 +166,7 @@ pub fn alter_typeguard_add( pub fn alter_typeguard_drop( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let coll_name = extract_collection_name(sql)?; @@ -180,7 +183,7 @@ pub fn alter_typeguard_drop( let tenant_id = identity.tenant_id.as_u64(); let mut coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id, &coll_name) + .get_collection(database_id, tenant_id, &coll_name) .map_err(|e| err("XX000", &e.to_string()))? .ok_or_else(|| err("42P01", &format!("collection '{coll_name}' not found")))?; @@ -194,7 +197,7 @@ pub fn alter_typeguard_drop( )); } - persist_collection_replicated(state, DatabaseId::DEFAULT, &coll) + persist_collection_replicated(state, database_id, &coll) .map_err(|e| err("XX000", &e.to_string()))?; state.schema_version.bump(); @@ -206,13 +209,14 @@ pub fn alter_typeguard_drop( pub fn alter_typeguard( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let upper = sql.to_uppercase(); if upper.contains(" ADD ") { - alter_typeguard_add(state, identity, sql) + alter_typeguard_add(state, identity, database_id, sql) } else if upper.contains(" DROP ") { - alter_typeguard_drop(state, identity, sql) + alter_typeguard_drop(state, identity, database_id, sql) } else { Err(err( "42601", @@ -227,6 +231,7 @@ pub fn alter_typeguard( pub fn drop_typeguard( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let upper = sql.to_uppercase(); @@ -238,7 +243,7 @@ pub fn drop_typeguard( let tenant_id = identity.tenant_id.as_u64(); let mut coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id, &coll_name) + .get_collection(database_id, tenant_id, &coll_name) .map_err(|e| err("XX000", &e.to_string()))? .ok_or_else(|| err("42P01", &format!("collection '{coll_name}' not found")))?; @@ -253,7 +258,7 @@ pub fn drop_typeguard( } coll.type_guards.clear(); - persist_collection_replicated(state, DatabaseId::DEFAULT, &coll) + persist_collection_replicated(state, database_id, &coll) .map_err(|e| err("XX000", &e.to_string()))?; state.schema_version.bump(); @@ -267,6 +272,7 @@ pub fn drop_typeguard( pub fn show_typeguard( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let coll_name = extract_collection_name(sql)?; @@ -275,7 +281,7 @@ pub fn show_typeguard( let tenant_id = identity.tenant_id.as_u64(); let coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id, &coll_name) + .get_collection(database_id, tenant_id, &coll_name) .map_err(|e| err("XX000", &e.to_string()))? .ok_or_else(|| err("42P01", &format!("collection '{coll_name}' not found")))?; @@ -316,13 +322,14 @@ pub fn show_typeguard( pub fn show_typeguards( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, _sql: &str, ) -> Result, DdlError> { let catalog = state.credentials.catalog(); let tenant_id = identity.tenant_id.as_u64(); let collections = catalog - .load_collections_for_tenant(DatabaseId::DEFAULT, tenant_id) + .load_collections_for_tenant(database_id, tenant_id) .map_err(|e| err("XX000", &e.to_string()))?; let columns = vec!["collection".to_string(), "fields".to_string()]; diff --git a/nodedb/src/control/server/shared/ddl/neutral/typeguard/validate.rs b/nodedb/src/control/server/shared/ddl/neutral/typeguard/validate.rs index 141675359..43237a971 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/typeguard/validate.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/typeguard/validate.rs @@ -30,6 +30,7 @@ use super::super::super::result::{DdlError, DdlResult}; pub async fn validate_typeguard( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, sql: &str, ) -> Result, DdlError> { let coll_name = super::parse::extract_collection_name(sql)?; @@ -38,7 +39,7 @@ pub async fn validate_typeguard( let catalog = state.credentials.catalog(); let coll = catalog - .get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &coll_name) + .get_collection(database_id, tenant_id.as_u64(), &coll_name) .map_err(|e| super::parse::err("XX000", &format!("catalog error: {e}")))? .ok_or_else(|| { super::parse::err("42P01", &format!("collection '{coll_name}' not found")) @@ -70,7 +71,7 @@ pub async fn validate_typeguard( state, identity, &scan_sql, - DatabaseId::DEFAULT, + database_id, ) .await .map_err(|error| super::parse::err(&error.sqlstate, &error.message))?; From 0324ecb61f2defcfff0fb27512c9b8ae676f151f Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:51 +0800 Subject: [PATCH 13/14] fix(convert): keep the source primary key CONVERT rebuilt the strict schema from the column list and dropped the source key. Every later insert failed "no resolved primary key". - mark the source key column in the converted schema - refuse a column list that omits it with 42601 --- .../shared/ddl/neutral/convert/driver.rs | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/nodedb/src/control/server/shared/ddl/neutral/convert/driver.rs b/nodedb/src/control/server/shared/ddl/neutral/convert/driver.rs index 66fb8f94d..fbacb7335 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/convert/driver.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/convert/driver.rs @@ -61,6 +61,49 @@ pub async fn convert_collection( _ => None, }; + // Preserve the source collection's declared identity through the + // conversion. Without this, a schemaless source declared `id TEXT PRIMARY + // KEY` converts to a strict schema whose columns carry no primary key, and + // every insert after the conversion fails `no resolved primary key`. A + // column list that omits the source key is refused rather than silently + // minting new row identities. + let mut columns = columns; + if let Some(cols) = columns.as_mut() { + let source_pk = match &coll.collection_type { + nodedb_types::CollectionType::Document(nodedb_types::DocumentMode::Strict(schema)) => { + schema + .columns + .iter() + .find(|c| c.primary_key) + .map(|c| c.name.clone()) + } + nodedb_types::CollectionType::KeyValue(config) => config + .schema + .columns + .iter() + .find(|c| c.primary_key) + .map(|c| c.name.clone()), + nodedb_types::CollectionType::Document(nodedb_types::DocumentMode::Schemaless) => { + coll.declared_primary_key.clone() + } + nodedb_types::CollectionType::Columnar(_) => None, + }; + if let Some(ref pk) = source_pk { + match cols.iter_mut().find(|c| &c.name == pk) { + Some(col) => col.primary_key = true, + None => { + return Err(err( + "42601", + format!( + "converted schema must keep the source primary key column \ + '{pk}'; it is absent from the column list" + ), + )); + } + } + } + } + let schema_json_for_dp = if let Some(ref cols) = columns { sonic_rs::to_string(cols).map_err(|e| err("XX000", format!("schema serialization: {e}")))? } else { From 3ad65accb452978f30d5ea7bc0b166ce1074a2f6 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Mon, 14 Sep 2026 08:17:51 +0800 Subject: [PATCH 14/14] fix(convert): refuse a column-referencing guard A guard DEFAULT or VALUE naming another column became a strict column DEFAULT. That DEFAULT evaluates with no row in scope, so every insert raised UnevaluableDefault. - add default_expr_references_columns in nodedb-sql - refuse the guard at CONVERT, naming field and clause --- nodedb-sql/src/planner/defaults/compiled.rs | 55 +++++++++++++++++++ nodedb-sql/src/planner/defaults/mod.rs | 4 +- .../ddl/neutral/convert/typeguard_columns.rs | 32 ++++++++++- 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/nodedb-sql/src/planner/defaults/compiled.rs b/nodedb-sql/src/planner/defaults/compiled.rs index 9905e3e23..2568d7d89 100644 --- a/nodedb-sql/src/planner/defaults/compiled.rs +++ b/nodedb-sql/src/planner/defaults/compiled.rs @@ -186,6 +186,61 @@ pub fn validate_default_expr(expr: &str, column: &str) -> crate::Result<()> { CompiledDefault::declare(column, expr).map(|_| ()) } +/// Whether a declared value-producing expression references a row column. +/// +/// A column `DEFAULT` and a typeguard `DEFAULT`/`VALUE` are evaluated with no +/// row in scope, so an expression that names a column can never produce a +/// value there. Callers that carry such an expression into a column `DEFAULT` +/// (CONVERT's typeguard path) use this to refuse at declaration time instead +/// of failing the first insert with [`crate::SqlError::UnevaluableDefault`]. +/// +/// Classifies and parses through the same resolver gate a DEFAULT passes; +/// evaluates nothing. A second name list or expression walker is not written. +pub fn default_expr_references_columns(expr: &str) -> crate::Result { + let parsed = crate::parse_expr_string(expr)?; + Ok(expr_references_column(&parsed)) +} + +/// Recursive column-reference scan over a parsed DEFAULT expression. +fn expr_references_column(expr: &SqlExpr) -> bool { + match expr { + SqlExpr::Column { .. } => true, + SqlExpr::Literal(_) | SqlExpr::Wildcard | SqlExpr::Subquery(_) => false, + SqlExpr::BinaryOp { left, right, .. } => { + expr_references_column(left) || expr_references_column(right) + } + SqlExpr::UnaryOp { expr, .. } + | SqlExpr::IsNull { expr, .. } + | SqlExpr::Cast { expr, .. } => expr_references_column(expr), + SqlExpr::Function { args, .. } => args.iter().any(expr_references_column), + SqlExpr::Case { + operand, + when_then, + else_expr, + } => { + operand.as_deref().is_some_and(expr_references_column) + || when_then + .iter() + .any(|(w, t)| expr_references_column(w) || expr_references_column(t)) + || else_expr.as_deref().is_some_and(expr_references_column) + } + SqlExpr::InList { expr, list, .. } => { + expr_references_column(expr) || list.iter().any(expr_references_column) + } + SqlExpr::Between { + expr, low, high, .. + } => { + expr_references_column(expr) + || expr_references_column(low) + || expr_references_column(high) + } + SqlExpr::Like { expr, pattern, .. } => { + expr_references_column(expr) || expr_references_column(pattern) + } + SqlExpr::ArrayLiteral(items) => items.iter().any(expr_references_column), + } +} + /// Classify a DEFAULT into its compiled form. fn classify(column: &str, expr: &str) -> crate::Result { let upper = expr.trim().to_uppercase(); diff --git a/nodedb-sql/src/planner/defaults/mod.rs b/nodedb-sql/src/planner/defaults/mod.rs index 98b05a51b..a9755e854 100644 --- a/nodedb-sql/src/planner/defaults/mod.rs +++ b/nodedb-sql/src/planner/defaults/mod.rs @@ -26,5 +26,7 @@ mod compiled; mod convert; mod kind; -pub use compiled::{ColumnDefaults, CompiledDefault, validate_default_expr}; +pub use compiled::{ + ColumnDefaults, CompiledDefault, default_expr_references_columns, validate_default_expr, +}; pub use convert::default_value_to_sql; diff --git a/nodedb/src/control/server/shared/ddl/neutral/convert/typeguard_columns.rs b/nodedb/src/control/server/shared/ddl/neutral/convert/typeguard_columns.rs index 0bc1749cc..11d6e5196 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/convert/typeguard_columns.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/convert/typeguard_columns.rs @@ -49,8 +49,38 @@ pub(super) fn typeguards_to_column_defs( }; // A guard carries either DEFAULT or VALUE, never both. Strict schema // has one materialization slot, so both land on the column `DEFAULT`. - if let Some(expr) = guard.default_expr.clone().or(guard.value_expr.clone()) { + let carried = guard + .default_expr + .clone() + .map(|expr| ("DEFAULT", expr)) + .or(guard.value_expr.clone().map(|expr| ("VALUE", expr))); + if let Some((clause, expr)) = carried { validate_column_default(&col.name, &expr)?; + // A guard VALUE is evaluated per row against the document; a + // strict-schema column DEFAULT is evaluated with no row in scope. + // Carrying a column-referencing expression over would accept the + // CONVERT and fail every insert with `UnevaluableDefault`, so it + // is refused here, naming the clause and the field. + let references_column = nodedb_sql::planner::defaults::default_expr_references_columns( + &expr, + ) + .map_err(|e| { + err( + "42601", + format!("field '{}': {clause} is invalid: {e}", guard.field), + ) + })?; + if references_column { + return Err(err( + "42601", + format!( + "field '{}': {clause} expression '{expr}' references another column; \ + a strict-schema column DEFAULT is evaluated with no row in scope. \ + Give a constant expression, or keep the collection schemaless", + guard.field + ), + )); + } col.default = Some(expr); } columns.push(col);