fix: nine-issue batch — sequences, kv scans, windows, graph DSL, and one instant conversion

nodedb-cluster-tests/tests/common_suite/cases/shuffle_aggregate_cross_node.rs

@@ -107,6 +107,8 @@ fn producer_plan(rows: &[&Row]) -> Vec<u8> {
         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,

nodedb-cluster-tests/tests/common_suite/cases/shuffle_consume_cross_node.rs

@@ -104,6 +104,8 @@ fn provider_scan_plan(rows: &[&Row]) -> Vec<u8> {
         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,

nodedb-cluster-tests/tests/common_suite/cases/shuffle_produce_cross_node.rs

@@ -108,6 +108,8 @@ fn provider_scan_plan(rows: &[Vec<u8>]) -> Vec<u8> {
         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,

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<u32>,
+        /// Output column names (same format as DocumentOp::Scan). Empty =
+        /// return the whole row document.
+        #[serde(default)]
+        projection: Vec<String>,
+        /// Serialized `Vec<ComputedColumn>` applied per row after the scan
+        /// (same format as DocumentOp::Scan). Empty = none.
+        #[serde(default)]
+        computed_columns: Vec<u8>,
     },
 
     /// Set or update TTL on an existing key.

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-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]

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<Vec<usize>, 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<serde_json::Value>)> = 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,
+    }
+}

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,
         };

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();

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<ast::Expr>) {
     match expr {

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,