diff --git a/nodedb-physical/src/physical_plan/query.rs b/nodedb-physical/src/physical_plan/query.rs index 96f087d3c..86b1ba995 100644 --- a/nodedb-physical/src/physical_plan/query.rs +++ b/nodedb-physical/src/physical_plan/query.rs @@ -78,6 +78,19 @@ 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 functions over derived-table rows must + /// evaluate after computed columns because window arguments may + /// reference computed aliases. Empty = no window functions. + #[serde(default)] + window_functions: Vec, /// ORDER BY terms, each an expression. Empty = unordered. #[serde(default)] sort_keys: Vec, @@ -118,6 +131,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 237ee8f23..c2f7df732 100644 --- a/nodedb/src/control/clone/resolver/rewrite.rs +++ b/nodedb/src/control/clone/resolver/rewrite.rs @@ -117,6 +117,8 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -139,6 +141,8 @@ pub fn rewrite_plan_for_source(params: RewriteForSourceParams<'_>) -> crate::Res input: child, 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, @@ -626,6 +630,8 @@ mod tests { input: Box::new(gather(plan)), 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 0df53ece6..e002ba5a0 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..56b3df789 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,69 @@ 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. + 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 body_plan = body_tasks.pop().expect("len == 1").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 7887e2056..11af24a6e 100644 --- a/nodedb/src/control/planner/sql_plan_convert/convert.rs +++ b/nodedb/src/control/planner/sql_plan_convert/convert.rs @@ -258,6 +258,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..caf5721f2 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 returning NULL. + 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/scan/core.rs b/nodedb/src/control/planner/sql_plan_convert/scan/core.rs index 5e56c2697..8429e4b9f 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, 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 4493c7f02..c569d84dd 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, @@ -245,6 +247,7 @@ pub(super) fn convert_subquery( input, filters, projection, + window_functions, sort_keys, offset, distinct, @@ -309,6 +312,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). 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 444ffb90b..bcf964ad7 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 @@ -28,6 +28,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, @@ -106,6 +108,8 @@ pub(super) async fn resolve_post_process( input, filters, projection, + computed_columns, + window_functions, sort_keys, limit, offset, @@ -235,6 +239,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 630b3d2b0..c990177f6 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 @@ -1517,6 +1517,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..228e0cec8 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,114 @@ 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. + 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. ─────────────────────────────────────────── + // Evaluate each window spec per partition after computed columns + // (window arguments may reference computed aliases) and before + // distinct/project (the window alias must exist in the row map). + // This ordering is required for derived tables; previously window + // specs were dropped and every window column returned NULL. + // 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 4c99430cf..e92ff1726 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 @@ -130,6 +130,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, @@ -277,6 +279,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, @@ -464,6 +468,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..c0b40f544 --- /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. 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 49316db90..84a2a3b20 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; @@ -158,14 +159,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; @@ -247,7 +244,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; @@ -266,7 +262,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:?}" + ); + } +}