feat(core): converge grouped *_over_time onto positional Aggregate.by (closes #8)

crates/core/src/intent_algebra/lower.rs

@@ -93,13 +93,18 @@ pub fn convert(
             input,
         } => {
             // 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).
+            // fuses into the canonical shape: a `Window` input becomes
+            // `Window { Aggregate { by: [] } }` (a per-series, label-preserving
+            // reduction — see `output_schema_in`'s `Window` arm). GROUP BY keys
+            // resolve *positionally* into `Aggregate.by` — the same shape SQL
+            // produces — whenever they're in scope: an instant selector, or a
+            // label-preserving per-series `rate`/`increase`/`*_over_time`. They
+            // fall back to a name-based `Partition` only when they can't be (a
+            // windowed reduction's own keys, or an unresolved name; see below).
+            // The reducer's input column resolves against the aggregate's
+            // *direct* input (the scan under any window). (SQL GROUP BY is
+            // tabular, so it skips this branch for the plain `Aggregate.by` path
+            // below.)
             if aggs.len() == 1 && having.is_none() && !input.leaf_is_tabular() {
                 let (agg_input_l2, window): (&LQueryExpr, Option<(_, _)>) = match input.as_ref() {
                     LQueryExpr::Window {

crates/core/src/intent_algebra/query_expr.rs

@@ -353,7 +353,24 @@ impl QueryExpr {
             // time-indexed Scan it preserves the time_index; over an Aggregate
             // (the canonical Window-over-Aggregate fused shape) the child has
             // already consumed the time axis, so the child schema passes through.
-            QueryExpr::Window { child, .. } => child.output_schema_in(scope),
+            // A time `Window` over an `Aggregate` is a per-series time-window
+            // reduction (`avg_over_time` / `quantile_over_time` / …): it keeps
+            // each series's labels and replaces only the sample value, so it is
+            // label-preserving — which lets an outer cross-series `Aggregate.by`
+            // group on those labels positionally (e.g. `sum by(x)(avg_over_time(…))`).
+            // A `Window` over a `Scan` (a bare range vector) passes through.
+            QueryExpr::Window { child, .. } => match child.as_ref() {
+                QueryExpr::Aggregate {
+                    by,
+                    aggs,
+                    child: agg_in,
+                    ..
+                } if by.is_empty() && aggs.len() == 1 => Ok(per_series_reduction_schema(
+                    &agg_in.output_schema_in(scope)?,
+                    &aggs[0],
+                )),
+                _ => child.output_schema_in(scope),
+            },
 
             QueryExpr::Aggregate {
                 by,
@@ -364,27 +381,13 @@ impl QueryExpr {
             } => {
                 let in_schema = child.output_schema_in(scope)?;
 
-                // Per-series range reduction (`rate`/`increase`): one value out
-                // per series, so it is *label-preserving* — every label column
-                // survives and only the sample value is replaced (kept named
-                // `value`, the PromQL convention). This is what lets an outer
-                // cross-series `Aggregate.by` resolve its group keys positionally
-                // over `sum by (job) (rate(...))`.
-                if by.is_empty() && !aggs.is_empty() && aggs.iter().all(|a| a.is_per_series()) {
-                    let value_idx = in_schema.column_id("value").or_else(|| {
-                        (0..in_schema.columns.len()).find(|&i| Some(i) != in_schema.time_index)
-                    });
-                    let mut columns = in_schema.columns.clone();
-                    if let Some(vi) = value_idx {
-                        let mut out = aggs[0].output_column(&columns[vi]);
-                        out.name = "value".into();
-                        columns[vi] = out;
-                    }
-                    return Ok(Schema {
-                        columns,
-                        time_index: in_schema.time_index,
-                        unique_keys: in_schema.unique_keys.clone(),
-                    });
+                // Per-series range reduction (`rate`/`increase`, whose window is
+                // carried in the intent so there is no enclosing `Window` node):
+                // one value out per series, so it is *label-preserving*. (The
+                // `*_over_time` reductions take the same shape but under a time
+                // `Window` — handled by the `Window` arm below.)
+                if by.is_empty() && aggs.len() == 1 && aggs[0].is_per_series() {
+                    return Ok(per_series_reduction_schema(&in_schema, &aggs[0]));
                 }
 
                 let mut out_cols: Vec<Column> = Vec::with_capacity(by.len() + aggs.len());
@@ -570,6 +573,28 @@ impl QueryExpr {
     }
 }
 
+/// Output schema of a *per-series* window/range reduction (`rate`/`increase`,
+/// or an `*_over_time` reducer under a time `Window`). Such a reduction emits
+/// one value per series, so every label column of `input` is preserved and only
+/// the sample value is replaced — kept named `value` so the PromQL sample-value
+/// convention (and any outer `SampleValue` reference) still resolves it by name.
+fn per_series_reduction_schema(input: &Schema, agg: &AggIntent) -> Schema {
+    let value_idx = input
+        .column_id("value")
+        .or_else(|| (0..input.columns.len()).find(|&i| Some(i) != input.time_index));
+    let mut columns = input.columns.clone();
+    if let Some(vi) = value_idx {
+        let mut out = agg.output_column(&columns[vi]);
+        out.name = "value".into();
+        columns[vi] = out;
+    }
+    Schema {
+        columns,
+        time_index: input.time_index,
+        unique_keys: input.unique_keys.clone(),
+    }
+}
+
 /// Infer the `(DataType, nullable)` a scalar [`L3Expr`] produces against an
 /// input [`Schema`]. Used by `Project` schema derivation. Approximate at L3:
 /// unknown columns and bare `FunctionCall`s fall back to a permissive default
@@ -770,6 +795,49 @@ mod tests {
         assert!(s.column_id("job").is_some(), "label survives the reduction");
     }
 
+    #[test]
+    fn windowed_over_time_reduction_preserves_labels() {
+        // `*_over_time` lowers to `Window { Aggregate{ by:[], [reducer] } }`: a
+        // per-series time-window reduction. It must be label-preserving too
+        // (same as `rate`), so an outer `sum by(job)(avg_over_time(...))` resolves
+        // its key positionally. The reducer (`Avg`) shares its `AggIntent` with
+        // cross-series `avg`; the enclosing `Window` is what marks it per-series.
+        let child = scan(
+            vec![
+                col("ts", DataType::Timestamp, false),
+                col("value", DataType::Float64, false),
+                col("job", DataType::Utf8, true),
+            ],
+            Some(0),
+            vec![],
+        );
+        let avg_over_time = QueryExpr::Window {
+            kind: WindowKind::Tumbling,
+            size: Duration::from_secs(300),
+            slide: None,
+            child: Box::new(QueryExpr::Aggregate {
+                by: vec![],
+                aggs: vec![AggIntent::Avg { col: None }],
+                output_names: vec![],
+                having: None,
+                child: Box::new(child),
+            }),
+        };
+        let s = avg_over_time.output_schema().unwrap();
+        assert_eq!(
+            s.columns
+                .iter()
+                .map(|c| c.name.as_str())
+                .collect::<Vec<_>>(),
+            vec!["ts", "value", "job"],
+            "windowed per-series reduction preserves labels; value kept named `value`"
+        );
+        assert!(
+            s.column_id("job").is_some(),
+            "outer Aggregate.by can resolve it"
+        );
+    }
+
     #[test]
     fn project_keeps_time_index_when_ts_passed_through() {
         let child = scan(

crates/lower/src/promql.rs

@@ -21,7 +21,7 @@
 //! | `rate/irate(m[w])` | `Aggregate{[Rate{w}]}` (no Window) — `irate` shares the `rate` *intent*; the avg-vs-last-two-samples difference is an L4 estimation method |
 //! | `increase(m[w])` | `Aggregate{[Increase{w}]}` (no Window) |
 //! | `changes` / `resets` / `group` / `offset` / `@` | **rejected** — distinct semantics with no intent-algebra representation yet |
-//! | `OUTER by (dims) (…)` | `Aggregate.keys = dims` (→ `Partition` in L3) |
+//! | `OUTER by (dims) (…)` | `Aggregate.keys = dims` (→ positional `Aggregate.by` in L3; `Partition` only as a fallback for windowed/unresolved keys) |
 //! | `count by (d) (…)` | `Aggregate{[CountDistinct], …}` (→ `Cardinality`) |
 //! | `topk(k, count_over_time(…))` | `TopK{k, by}` (heavy-hitter, one pass) |
 //! | `topk(k, <other>)` / `bottomk(k, …)` | `Sort{value} → Limit{k}` |

crates/lower/tests/promql_conformance.rs

@@ -313,6 +313,31 @@ fn sum_by_of_rate_groups_outer_level() {
     ));
 }
 
+#[test]
+fn sum_by_of_over_time_groups_outer_level() {
+    // Outer cross-series Sum grouped on positional `Aggregate.by` over an inner
+    // *per-series* `avg_over_time` — `Window { Aggregate{Avg} }` is label-
+    // preserving, so the key resolves positionally just like the rate case (no
+    // name-based Partition). Leaf = [ts, value, instance] → by = [2].
+    let qe = ok("sum by(instance) (avg_over_time(node_cpu_seconds_total[5m]))");
+    let QueryExpr::Aggregate {
+        by, aggs, child, ..
+    } = &qe
+    else {
+        panic!("expected outer Aggregate grouped by instance, got {qe:?}");
+    };
+    assert_eq!(by, &vec![2]);
+    assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
+    // child is the inner per-series reduction: Window over Aggregate{Avg}.
+    let QueryExpr::Window { child, .. } = child.as_ref() else {
+        panic!("expected Window (per-series avg_over_time) under the Sum, got {child:?}");
+    };
+    assert!(matches!(
+        child.as_ref(),
+        QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Avg { .. }])
+    ));
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // E. Aggregation over time (per-series)    (cheat sheet "Aggregating Over
 //    Time"; functions.test)

crates/lower/tests/promql_lowering.rs

@@ -82,23 +82,24 @@ fn quantile_over_time_is_window_over_aggregate() {
 }
 
 #[test]
-fn outer_sum_by_wraps_in_partition() {
-    // `sum by (host) (quantile_over_time(...))` is a two-level reduction: an
-    // inner per-series quantile-over-time, then an outer cross-series sum.
-    // Grouping rides on a `Partition` wrapping the outer Sum (backend model).
+fn outer_sum_by_over_quantile_over_time_groups_positionally() {
+    // `sum by (host) (quantile_over_time(...))`: inner per-series
+    // quantile-over-time (label-preserving), then an outer cross-series sum
+    // grouped on a positional `Aggregate.by` — the same shape SQL produces, not
+    // a name-based Partition. Leaf = [ts, value, host, service] (referenced
+    // names appended sorted) → host = col 2.
     let qe = lower(r#"sum by (host) (quantile_over_time(0.99, latency{service="web"}[5m]))"#);
-    let QueryExpr::Partition { keys, child } = &qe else {
-        panic!("expected Partition, got {qe:?}");
-    };
-    assert_eq!(keys, &PartitionKeys::By(vec!["host".into()]));
-    // Outer cross-series Sum.
-    let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
-        panic!("expected outer Aggregate{{Sum}} under Partition, got {child:?}");
+    let QueryExpr::Aggregate {
+        by, aggs, child, ..
+    } = &qe
+    else {
+        panic!("expected outer Aggregate grouped by host, got {qe:?}");
     };
+    assert_eq!(by, &vec![2]);
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
-    // Inner: Window over Aggregate{Quantile}.
+    // Inner: Window over Aggregate{Quantile} (the per-series over_time reduction).
     let QueryExpr::Window { child, .. } = child.as_ref() else {
-        panic!("expected Window under the outer Sum, got {child:?}");
+        panic!("expected Window (per-series over_time) under the outer Sum, got {child:?}");
     };
     assert!(matches!(
         child.as_ref(),
@@ -275,20 +276,20 @@ fn count_over_time_is_count_intent() {
 #[test]
 fn outer_count_is_cardinality() {
     // `count by (symbol) (count_over_time(...))`: inner per-series sample count
-    // over the window, outer cross-series cardinality grouped by symbol.
+    // over the window (label-preserving), outer cross-series cardinality grouped
+    // on a positional `Aggregate.by`. Leaf = [ts, value, symbol] → symbol = col 2.
     let qe = lower("count by (symbol) (count_over_time(financial_last_trade_price[5m]))");
-    let QueryExpr::Partition { keys, child } = &qe else {
-        panic!("expected Partition, got {qe:?}");
-    };
-    assert_eq!(keys, &PartitionKeys::By(vec!["symbol".into()]));
-    // Outer cardinality (count of series).
-    let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
-        panic!("expected outer Aggregate{{Cardinality}} under Partition, got {child:?}");
+    let QueryExpr::Aggregate {
+        by, aggs, child, ..
+    } = &qe
+    else {
+        panic!("expected outer Aggregate grouped by symbol, got {qe:?}");
     };
+    assert_eq!(by, &vec![2]);
     assert!(matches!(aggs.as_slice(), [AggIntent::Cardinality { .. }]));
     // Inner: Window over Aggregate{Count} (count_over_time).
     let QueryExpr::Window { child, .. } = child.as_ref() else {
-        panic!("expected Window under the outer cardinality, got {child:?}");
+        panic!("expected Window (per-series count_over_time) under the cardinality, got {child:?}");
     };
     assert!(matches!(
         child.as_ref(),