fix(sql): propagate ORDER BY into aggregate plans and support derived-FROM subqueries

Three silent-correctness bugs fixed end-to-end through planner, converter, and executor:

ORDER BY after GROUP BY was silently dropped
  The planner's apply_order_by did not handle SqlPlan::Aggregate, so every
  GROUP BY … ORDER BY query returned groups in hash-map iteration order.
  SqlPlan::Aggregate gains a sort_keys field; apply_order_by populates it;
  the executor sorts finalized group rows before truncating to LIMIT. The
  converter rejects non-column sort expressions with a typed error rather
  than silently discarding them. ORDER BY on a CTE (used by derived-FROM
  subqueries) is now pushed into the outer plan.

Derived FROM subquery support
  FROM (SELECT …) AS t was not a recognised pattern; the resolver dropped
  the factor silently, the scope was empty, and the planner errored with
  "multi-table FROM without JOIN". try_plan_derived_from desugars the
  single non-LATERAL derived-table into a synthetic CTE so the outer
  SELECT plans normally against the alias.

Aggregate alias case mismatch
  UnnamedExpr aliases were kept in the case emitted by the AST formatter
  (e.g. COUNT(DISTINCT user_id)), but expr_column_names produced lowercase
  keys. The JSON row key never matched, and clients saw NULL. Alias
  generation now lowercases the expression string to align with the
  column-name lookup.

Author: farhan-syah

nodedb-sql/src/engine_rules/columnar.rs

@@ -131,6 +131,7 @@ impl EngineRules for ColumnarRules {
             having: p.having,
             limit: p.limit,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         })
     }
 

nodedb-sql/src/engine_rules/document_schemaless.rs

@@ -136,6 +136,7 @@ impl EngineRules for SchemalessRules {
             having: p.having,
             limit: p.limit,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         })
     }
 }

nodedb-sql/src/engine_rules/document_strict.rs

@@ -136,6 +136,7 @@ impl EngineRules for StrictRules {
             having: p.having,
             limit: p.limit,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         })
     }
 }

nodedb-sql/src/engine_rules/kv.rs

@@ -117,6 +117,7 @@ impl EngineRules for KvRules {
             having: p.having,
             limit: p.limit,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         })
     }
 

nodedb-sql/src/engine_rules/spatial.rs

@@ -123,6 +123,7 @@ impl EngineRules for SpatialRules {
             having: p.having,
             limit: p.limit,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         })
     }
 

nodedb-sql/src/planner/aggregate.rs

@@ -93,6 +93,7 @@ fn attach_grouping_sets(
             input,
             limit,
             grouping_sets: _,
+            sort_keys,
             ..
         } => SqlPlan::Aggregate {
             input,
@@ -101,6 +102,7 @@ fn attach_grouping_sets(
             having,
             limit,
             grouping_sets: Some(grouping_sets),
+            sort_keys,
         },
         other => {
             // Engine returned something other than Aggregate (e.g. TimeseriesIngest).
@@ -112,6 +114,7 @@ fn attach_grouping_sets(
                 having,
                 limit: 10000,
                 grouping_sets: Some(grouping_sets),
+                sort_keys: Vec::new(),
             }
         }
     }
@@ -356,7 +359,15 @@ pub fn extract_aggregates_from_projection(
     let mut aggregates = Vec::new();
     for item in items {
         let (expr, alias): (&ast::Expr, String) = match item {
-            ast::SelectItem::UnnamedExpr(expr) => (expr, format!("{expr}")),
+            // Lowercase the unparsed expression text so the resulting
+            // alias (which becomes the JSON row key after the
+            // `apply_user_aliases_to_rows` rename) matches the pgwire
+            // lookup key built by `expr_column_names`, which also
+            // 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()),
             ast::SelectItem::ExprWithAlias { expr, alias } => (expr, normalize_ident(alias)),
             _ => continue,
         };

nodedb-sql/src/planner/join/plan.rs

@@ -134,6 +134,7 @@ pub fn plan_join_from_select(
             having,
             limit: 10000,
             grouping_sets: None,
+            sort_keys: Vec::new(),
         }));
     }
 

nodedb-sql/src/planner/select/entry.rs

@@ -392,9 +392,9 @@ fn apply_limit(mut plan: SqlPlan, limit_clause: &Option<ast::LimitClause>) -> Sq
 }
 
 /// Catalog wrapper that resolves CTE names as schemaless document collections.
-struct CteCatalog<'a> {
-    inner: &'a dyn SqlCatalog,
-    cte_names: Vec<String>,
+pub(crate) struct CteCatalog<'a> {
+    pub(crate) inner: &'a dyn SqlCatalog,
+    pub(crate) cte_names: Vec<String>,
 }
 
 impl SqlCatalog for CteCatalog<'_> {

nodedb-sql/src/planner/select/order_by/apply.rs

@@ -90,6 +90,37 @@ pub(in crate::planner::select) fn apply_order_by(
             window_functions: window_functions.clone(),
             temporal: *temporal,
         }),
+        // ORDER BY applied to a GROUP BY result: stash the sort keys
+        // on the Aggregate plan; the executor sorts the finalized
+        // group rows before returning. Without this branch the sort
+        // is silently dropped — every `… GROUP BY x ORDER BY x` query
+        // comes back in hash-map iteration order, which is a
+        // data-correctness bug for any downstream consumer.
+        SqlPlan::Aggregate {
+            input,
+            group_by,
+            aggregates,
+            having,
+            limit,
+            grouping_sets,
+            ..
+        } => Ok(SqlPlan::Aggregate {
+            input: input.clone(),
+            group_by: group_by.clone(),
+            aggregates: aggregates.clone(),
+            having: having.clone(),
+            limit: *limit,
+            grouping_sets: grouping_sets.clone(),
+            sort_keys,
+        }),
+        // Cte wraps an inner outer plan; push ORDER BY into that outer
+        // so derived-table queries (`SELECT … FROM (…) AS t ORDER BY …`)
+        // honour the sort. inline_cte downstream merges the outer Scan
+        // with the inner subquery plan; the sort_keys ride along.
+        SqlPlan::Cte { definitions, outer } => Ok(SqlPlan::Cte {
+            definitions: definitions.clone(),
+            outer: Box::new(apply_order_by(outer, order_by, functions, select_items)?),
+        }),
         _ => Ok(plan.clone()),
     }
 }

nodedb-sql/src/planner/select/select_stmt.rs

@@ -9,13 +9,16 @@ use super::helpers::{convert_projection, convert_where_to_filters, eval_constant
 use super::where_search::try_extract_where_search;
 use crate::error::{Result, SqlError};
 use crate::functions::registry::FunctionRegistry;
+use crate::parser::normalize::normalize_ident;
 use crate::planner::lateral::plan::{
     is_lateral_derived, lateral_alias_from_factor, plan_lateral_join, subquery_from_factor,
 };
 use crate::resolver::columns::TableScope;
 use crate::temporal::TemporalScope;
 use crate::types::*;
 
+use super::entry::{CteCatalog, plan_query};
+
 /// Plan a single SELECT statement (no UNION, no CTE wrapper).
 pub(super) fn plan_select(
     select: &Select,
@@ -31,6 +34,19 @@ pub(super) fn plan_select(
         return Ok(plan);
     }
 
+    // 0.5. Derived FROM subquery: `FROM (SELECT ...) AS t`.
+    //
+    // Plan the inner subquery first, then desugar into a synthetic CTE
+    // so the outer SELECT — which may reference `t` like any other
+    // relation — plans against a catalog that resolves the alias to a
+    // schemaless source. Until this branch existed the resolver
+    // dropped non-LATERAL derived factors silently, the scope ended
+    // up empty, and the planner errored with "multi-table FROM
+    // without JOIN".
+    if let Some(plan) = try_plan_derived_from(select, catalog, functions, temporal)? {
+        return Ok(plan);
+    }
+
     // 1. Resolve FROM tables.
     let scope = TableScope::resolve_from(catalog, &select.from)?;
 
@@ -345,6 +361,71 @@ fn has_aggregation(select: &Select, functions: &FunctionRegistry) -> bool {
     false
 }
 
+/// Desugar `FROM (SELECT ...) AS alias` into a synthetic single-CTE plan.
+///
+/// Recognises the single-source, non-LATERAL derived-table pattern. The
+/// inner subquery is planned with the original catalog; the outer
+/// SELECT is replanned with a `CteCatalog` that resolves the alias to
+/// a schemaless source. The result is wrapped as `SqlPlan::Cte` so the
+/// `convert_cte` lowering takes care of execution.
+///
+/// Returns `Ok(None)` when the FROM clause is not a single derived
+/// table, so the caller falls through to the regular planning path.
+fn try_plan_derived_from(
+    select: &Select,
+    catalog: &dyn SqlCatalog,
+    functions: &FunctionRegistry,
+    temporal: TemporalScope,
+) -> Result<Option<SqlPlan>> {
+    if select.from.len() != 1 {
+        return Ok(None);
+    }
+    let from = &select.from[0];
+    if !from.joins.is_empty() {
+        return Ok(None);
+    }
+    let (subquery, alias_ident) = match &from.relation {
+        ast::TableFactor::Derived {
+            lateral: false,
+            subquery,
+            alias: Some(alias),
+            ..
+        } => (subquery, alias),
+        _ => return Ok(None),
+    };
+
+    let alias_name = normalize_ident(&alias_ident.name);
+    let inner_plan = plan_query(subquery, catalog, functions, temporal)?;
+
+    // Replan the outer SELECT against a catalog that resolves the alias
+    // as a schemaless source. The outer can reference `alias.col`
+    // qualified or unqualified — the resolver treats CTE rows as a
+    // schemaless document so any projected column flows through.
+    let derived_catalog = CteCatalog {
+        inner: catalog,
+        cte_names: vec![alias_name.clone()],
+    };
+    let mut outer_select = select.clone();
+    outer_select.from[0].relation = ast::TableFactor::Table {
+        name: ast::ObjectName::from(vec![ast::Ident::new(alias_name.clone())]),
+        alias: None,
+        args: None,
+        with_hints: Vec::new(),
+        version: None,
+        with_ordinality: false,
+        partitions: Vec::new(),
+        json_path: None,
+        sample: None,
+        index_hints: Vec::new(),
+    };
+    let outer_plan = plan_select(&outer_select, &derived_catalog, functions, temporal)?;
+
+    Ok(Some(SqlPlan::Cte {
+        definitions: vec![(alias_name, inner_plan)],
+        outer: Box::new(outer_plan),
+    }))
+}
+
 /// Dispatch to the JOIN planner if the FROM contains joins.
 fn try_plan_join(
     select: &Select,