feat(core): route tabular GROUP BY through positional Aggregate.by (key in schema)

The single-aggregate fused path wraps GROUP BY keys in a name-based Partition
(the PromQL streaming-sketch canonical shape) whose output schema is the child's
— so the group key is not an output column. That's correct for time series but
wrong for SQL, where `SELECT k, agg(...) GROUP BY k` projects `k`.

The fused path is now gated on a *time-series* leaf (`relational::leaf_is_tabular`
= the leftmost Source carries a resolved schema, i.e. Source::Table). Tabular
single-agg GROUP BY falls through to the positional `Aggregate.by` path, so the
key lands in the output schema and the enclosing SELECT projection resolves it.
PromQL's fused-Partition shape is unchanged.

- relational: factor out `leaf_source`; add `leaf_is_tabular`; `source_name` now
  delegates to `leaf_source`.

Test: `SELECT service, SUM(bytes) GROUP BY service` → Aggregate.by=[1],
Sum{col:3}, and the root projection schema is [service:Utf8, sum:Int64]. 111
tests green, clippy -D warnings clean, fmt clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: zzylol

crates/core/src/intent_algebra/lower.rs

@@ -87,11 +87,15 @@ pub fn convert(
             having,
             input,
         } => {
-            // Single-statistic aggregate (no HAVING) fuses: a `Window` input
-            // becomes `Window { Aggregate { by: [] } }`; GROUP BY keys wrap the
-            // result in a `Partition`. The reducer's input column resolves
-            // against the aggregate's *direct* input (the scan under any window).
-            if aggs.len() == 1 && having.is_none() {
+            // Single-statistic aggregate (no HAVING) over a *time-series* leaf
+            // fuses: a `Window` input becomes `Window { Aggregate { by: [] } }`;
+            // GROUP BY keys wrap the result in a `Partition` (the streaming
+            // sketch canonical shape). Tabular (SQL) GROUP BY instead falls
+            // through to the positional `Aggregate.by` path below, so the group
+            // keys land in the output schema (a SELECT projects them). The
+            // reducer's input column resolves against the aggregate's *direct*
+            // input (the scan under any window).
+            if aggs.len() == 1 && having.is_none() && !input.leaf_is_tabular() {
                 let (agg_input_l2, window): (&LQueryExpr, Option<(_, _)>) = match input.as_ref() {
                     LQueryExpr::Window {
                         duration,

crates/core/src/intent_algebra/relational.rs

@@ -223,10 +223,10 @@ impl QueryExpr {
         }
     }
 
-    /// Outermost metric/table name from the first `Source` leaf.
-    pub fn source_name(&self) -> Option<&str> {
+    /// The leftmost `Source` leaf.
+    pub fn leaf_source(&self) -> Option<&SourceSpec> {
         match self {
-            QueryExpr::Source(s) => Some(&s.name),
+            QueryExpr::Source(s) => Some(s),
             QueryExpr::Filter { input, .. }
             | QueryExpr::Project { input, .. }
             | QueryExpr::Aggregate { input, .. }
@@ -236,13 +236,27 @@ impl QueryExpr {
             | QueryExpr::TopK { input, .. }
             | QueryExpr::Sort { input, .. }
             | QueryExpr::Limit { input, .. }
-            | QueryExpr::PromQLSubquery { input, .. } => input.source_name(),
-            QueryExpr::Merge { inputs } => inputs.first()?.source_name(),
+            | QueryExpr::PromQLSubquery { input, .. } => input.leaf_source(),
+            QueryExpr::Merge { inputs } => inputs.first()?.leaf_source(),
             QueryExpr::Join { left, .. }
             | QueryExpr::SetOp { left, .. }
-            | QueryExpr::BinaryOp { lhs: left, .. } => left.source_name(),
-            QueryExpr::LetBinding { body, .. } => body.source_name(),
+            | QueryExpr::BinaryOp { lhs: left, .. } => left.leaf_source(),
+            QueryExpr::LetBinding { body, .. } => body.leaf_source(),
             QueryExpr::Ref(_) => None,
         }
     }
+
+    /// Outermost metric/table name from the first `Source` leaf.
+    pub fn source_name(&self) -> Option<&str> {
+        self.leaf_source().map(|s| s.name.as_str())
+    }
+
+    /// Whether the leftmost `Source` leaf carries a resolved schema — i.e. it is
+    /// a SQL table (`Source::Table`). Time-series (PromQL) leaves return
+    /// `false`. The converter uses this to keep the time-series fused-Partition
+    /// canonical shape for PromQL while routing tabular GROUP BY through a
+    /// positional `Aggregate.by` (so group keys land in the output schema).
+    pub fn leaf_is_tabular(&self) -> bool {
+        self.leaf_source().is_some_and(|s| s.schema.is_some())
+    }
 }

crates/lower/tests/sql_lowering.rs

@@ -134,6 +134,27 @@ async fn projection_over_aggregate_resolves_output_types_via_output_names() {
     );
 }
 
+#[tokio::test]
+async fn single_agg_group_by_keeps_key_in_output_schema() {
+    // A tabular single-aggregate GROUP BY routes through the positional
+    // Aggregate.by path (not the PromQL fused-Partition shape), so the group
+    // key is a real output column the enclosing SELECT projection resolves.
+    let qe = lower("SELECT service, SUM(bytes) FROM metrics GROUP BY service").await;
+    let (by, aggs) = find_aggregate(&qe).expect("expected an Aggregate (not a Partition)");
+    assert_eq!(by, &vec![1], "GROUP BY service → Aggregate.by column 1");
+    assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: Some(3) }]));
+
+    // Both the group key and the aggregate resolve in the root projection schema.
+    let schema = qe.output_schema().expect("root projection schema");
+    assert_eq!(schema.columns.len(), 2);
+    assert_eq!(
+        schema.columns[0].dtype,
+        DataType::Utf8,
+        "service is in the output"
+    );
+    assert_eq!(schema.columns[1].dtype, DataType::Int64, "SUM(bytes)");
+}
+
 #[tokio::test]
 async fn count_star_is_count_intent() {
     let qe = lower("SELECT COUNT(*) FROM metrics").await;