fix(sql): raise expression errors over constant derived tables

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<String>,
+        /// Serialized `Vec<ComputedColumn>` 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<u8>,
+        /// Serialized `Vec<WindowFuncSpec>` evaluated per partition after
+        /// computed columns (window over derived-table rows — issue #295
+        /// Gap 3). Empty = no window functions.
+        #[serde(default)]
+        window_functions: Vec<u8>,
         /// ORDER BY terms, each an expression. Empty = unordered.
         #[serde(default)]
         sort_keys: Vec<crate::physical_plan::SortKeySpec>,
@@ -118,6 +130,14 @@ pub enum QueryOp {
         /// Output column names to keep. Empty = emit all columns.
         #[serde(default)]
         projection: Vec<String>,
+        /// Serialized `Vec<ComputedColumn>` applied per row (see
+        /// `ProviderScan::computed_columns`).
+        #[serde(default)]
+        computed_columns: Vec<u8>,
+        /// Serialized `Vec<WindowFuncSpec>` (see
+        /// `ProviderScan::window_functions`).
+        #[serde(default)]
+        window_functions: Vec<u8>,
         /// ORDER BY terms, each an expression. Empty = unordered.
         #[serde(default)]
         sort_keys: Vec<crate::physical_plan::SortKeySpec>,

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,

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,

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,

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,

nodedb-sql/src/types/plan/variants.rs

@@ -525,6 +525,9 @@ pub enum SqlPlan {
         filters: Vec<Filter>,
         /// Outer projection (target list). Empty = inherit the body's columns.
         projection: Vec<Projection>,
+        /// Window functions evaluated over the materialized rows. Empty =
+        /// none.
+        window_functions: Vec<crate::types::WindowSpec>,
         /// Outer `ORDER BY` keys applied over the materialized rows.
         sort_keys: Vec<SortKey>,
         /// Outer `OFFSET` (0 = none).

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,

nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs

@@ -26,6 +26,7 @@ pub(super) fn dispatch_rest<V: PlanVisitor>(
             input,
             filters,
             projection,
+            window_functions,
             sort_keys,
             offset,
             distinct,
@@ -34,6 +35,7 @@ pub(super) fn dispatch_rest<V: PlanVisitor>(
             input,
             filters,
             projection,
+            window_functions,
             sort_keys,
             offset: *offset,
             distinct: *distinct,

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,

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,

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,

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,

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 (issue #295).
+    if !matches!(input, SqlPlan::Scan { .. }) {
+        let derived_group_specs = group_by_to_specs(group_by);
+        let derived_agg_specs: Vec<AggregateSpec> =
+            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);

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,

nodedb/src/control/planner/sql_plan_convert/expr.rs

@@ -280,6 +280,22 @@ pub(super) fn convert_sort_keys(keys: &[SortKey]) -> Vec<SortKeySpec> {
         .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 Gap 3).
+                        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,

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,