From 8db99947667181912b1a914399dce274ca0cd3b8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 27 May 2026 12:24:19 -0600 Subject: [PATCH 1/2] feat(core): converge grouped *_over_time onto positional Aggregate.by (closes #8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The residual of review D: `sum by(x)(avg_over_time(m[w]))` (and quantile/count/ …_over_time) still parked its group key in a name-based `Partition`, because the inner `*_over_time` reduction — `Window { Aggregate{ by:[], [reducer] } }` — collapsed its output schema to `[reducer]`, dropping `x`, so the outer cross-series aggregate had nothing to resolve into `Aggregate.by`. Fix: recognize that a per-series time-window reduction is *label-preserving* — the same property `rate`/`increase` already have (D), just under a time `Window` rather than carried in the intent. In `output_schema_in`, the `Window`-over- `Aggregate` arm now re-derives the schema label-preservingly (keep every label, replace only the sample value, kept named `value`). The discriminator is purely structural — the enclosing `Window` — which sidesteps the shared-`AggIntent` problem (`avg_over_time` and cross-series `avg` are both `AggIntent::Avg`). With labels preserved, the converter's existing key-resolution lands `by=[x]` unchanged — no converter edit. So `sum/count/avg/quantile by(x)(*_over_time(…))` now emit the same positional shape as SQL. Naming the preserved value column `value` keeps `SampleValue` resolvable, so `topk by(h)(avg_over_time(…))` (the generic Sort+Limit path, grouping still via Partition) does not regress. Factored the per-series logic into `per_series_reduction_schema`, shared by the rate/increase (Aggregate) and *_over_time (Window) arms. Updated the two flipped shape tests; added a Window-over-Aggregate label-preservation unit test and a `sum by(instance)(avg_over_time(...))` conformance test. Full suite green (135), clippy -D warnings + fmt clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/core/src/intent_algebra/query_expr.rs | 112 +++++++++++++++---- crates/lower/tests/promql_conformance.rs | 25 +++++ crates/lower/tests/promql_lowering.rs | 45 ++++---- 3 files changed, 138 insertions(+), 44 deletions(-) diff --git a/crates/core/src/intent_algebra/query_expr.rs b/crates/core/src/intent_algebra/query_expr.rs index d8604028..eed720ea 100644 --- a/crates/core/src/intent_algebra/query_expr.rs +++ b/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 = 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!["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( diff --git a/crates/lower/tests/promql_conformance.rs b/crates/lower/tests/promql_conformance.rs index 32a29b33..4f4c780c 100644 --- a/crates/lower/tests/promql_conformance.rs +++ b/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) diff --git a/crates/lower/tests/promql_lowering.rs b/crates/lower/tests/promql_lowering.rs index d99fbd4e..c9ceee49 100644 --- a/crates/lower/tests/promql_lowering.rs +++ b/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(), From ed2c6c6694089f0c9b298b81c3133eb66bd61e02 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 27 May 2026 12:33:32 -0600 Subject: [PATCH 2/2] =?UTF-8?q?docs:=20correct=20stale=20"GROUP=20BY=20?= =?UTF-8?q?=E2=86=92=20Partition"=20comments=20after=20the=20*=5Fover=5Fti?= =?UTF-8?q?me=20convergence?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The converter fused-path comment and the promql.rs mapping table still described PromQL grouping as landing in a name-based `Partition`; after the D + #8 work it resolves positionally into `Aggregate.by` (Partition is only a fallback for windowed/unresolved keys). Doc-only — no behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/core/src/intent_algebra/lower.rs | 19 ++++++++++++------- crates/lower/src/promql.rs | 2 +- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/crates/core/src/intent_algebra/lower.rs b/crates/core/src/intent_algebra/lower.rs index 2238788d..0ad4caf3 100644 --- a/crates/core/src/intent_algebra/lower.rs +++ b/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 { diff --git a/crates/lower/src/promql.rs b/crates/lower/src/promql.rs index 654149c4..85bb46bf 100644 --- a/crates/lower/src/promql.rs +++ b/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, )` / `bottomk(k, …)` | `Sort{value} → Limit{k}` |