From 734fd740c74b22a4a6cf249d677fac8987958708 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 2 Jul 2026 07:35:34 -0600 Subject: [PATCH] feat(lower): support *_over_time over a sub-query (#27) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `max_over_time(rate(m[5m])[1h:])` and the rest of the `*_over_time`/`quantile_over_time`-of-sub-query family were rejected — `extract_matrix` only accepts a (parenthesised) matrix selector, not the `PromQLSubquery` a sub-query argument produces. `walk_call` now intercepts a `*_over_time` call whose argument is a sub-query, lowers the sub-query recursively, and reduces it per series: max_over_time(rate(m[5m])[1h:]) -> Aggregate{Max} -> Subquery{1h} -> Aggregate{Rate} -> TimeRange{5m} -> Scan Non-sub-query calls fall through to the existing flat template unchanged, so plain `max_over_time(m[w])` and heavy-hitter `topk(k, count_over_time(m[w]))` are untouched. Per-series correctness: an empty-keys aggregate over a `Subquery` is now recognised as a label-preserving per-series range reduction, mirroring the existing `TimeRange` marker. This is sound because a genuine cross-series aggregation operator over a range vector (`sum(rate(m[5m])[1h:])`) is a PromQL type error the parser already rejects — so an `Aggregate` directly over a `Subquery` only ever arises from this per-series `*_over_time` shape. Verified by `sum by (job) (max_over_time(rate(m[5m])[1h:]))`: the inner `Max` preserves `job` for the outer `sum by (job)` to group on. Flips the `over_time_of_subquery_is_rejected__GAP` conformance test into three passing tests (per-series reduction, `quantile_over_time` phi, outer aggregation label preservation). Full workspace suite green; clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/core/src/intent_algebra/query_expr.rs | 17 ++++-- crates/lower/src/promql.rs | 60 +++++++++++++++++++- crates/lower/tests/promql_conformance.rs | 58 +++++++++++++++++-- 3 files changed, 124 insertions(+), 11 deletions(-) diff --git a/crates/core/src/intent_algebra/query_expr.rs b/crates/core/src/intent_algebra/query_expr.rs index bceadebf..33fefec9 100644 --- a/crates/core/src/intent_algebra/query_expr.rs +++ b/crates/core/src/intent_algebra/query_expr.rs @@ -428,11 +428,18 @@ impl QueryExpr { // Per-series range reduction: `rate`/`increase` (is_per_series) // OR any single aggregate whose direct child is a `TimeRange` - // (`*_over_time` functions). Both produce one value per series - // and are label-preserving — the `TimeRange` child is the - // structural marker that confers per-series semantics on - // otherwise cross-series intents like `Avg`/`Sum`/`Count`. - let is_range_child = matches!(child.as_ref(), QueryExpr::TimeRange { .. }); + // (`*_over_time` functions) or a `Subquery` (`*_over_time` over a + // sub-query, e.g. `max_over_time(rate(m[5m])[1h:])`). All produce + // one value per series and are label-preserving — the range child + // is the structural marker that confers per-series semantics on + // otherwise cross-series intents like `Avg`/`Sum`/`Count`. A + // cross-series aggregation operator over a range vector is a + // PromQL type error the parser rejects, so an `Aggregate` over a + // `Subquery` is only ever this per-series `*_over_time` shape. + let is_range_child = matches!( + child.as_ref(), + QueryExpr::TimeRange { .. } | QueryExpr::Subquery { .. } + ); if by.is_empty() && aggs.len() == 1 && (aggs[0].is_per_series() || is_range_child) { return Ok(per_series_reduction_schema(&in_schema, &aggs[0])); } diff --git a/crates/lower/src/promql.rs b/crates/lower/src/promql.rs index f5a02d63..15828c9a 100644 --- a/crates/lower/src/promql.rs +++ b/crates/lower/src/promql.rs @@ -152,7 +152,7 @@ fn walk(expr: &Expr) -> Result { match expr { Expr::Aggregate(agg) => walk_aggregate(agg), Expr::Call(call) if call.func.name == "histogram_quantile" => walk_histogram_quantile(call), - Expr::Call(call) => build(lower_inner_call(call)?, vec![], Outer::None), + Expr::Call(call) => walk_call(call), Expr::Binary(bin) => walk_binary(bin), Expr::Paren(p) => walk(&p.expr), // `UnaryExpr` is built only by negation (`Neg`); unary `+` is folded to @@ -190,6 +190,64 @@ fn walk(expr: &Expr) -> Result { } } +/// Lower a bare function call (`rate(m[5m])`, `max_over_time(m[5m])`, …). +/// +/// The common case routes through the flat `lower_inner_call` template. The one +/// exception is a `*_over_time`/`quantile_over_time` function applied to a +/// **sub-query** (`max_over_time(rate(m[5m])[1h:])`): its argument is a +/// `PromQLSubquery`, not a matrix selector, so the flat template's +/// `extract_matrix` can't accept it. Lower the sub-query recursively and reduce +/// it per series (issue #27). +fn walk_call(call: &Call) -> Result { + if let Some((func, arg_expr)) = over_time_reducer(call)? { + if is_subquery(arg_expr) { + // `_over_time()` reduces the sub-query's range vector + // *per series* over time — label-preserving, no group keys. The + // `PromQLSubquery` child is the structural range marker that confers + // per-series semantics on the otherwise cross-series reducer (mirrors + // the `TimeRange` marker for `*_over_time(m[w])`). A cross-series + // aggregation operator (`sum`/`avg`) over a range vector is a PromQL + // type error the parser already rejects, so an `Aggregate` directly + // over a `PromQLSubquery` only ever arises here. + return Ok(outer_aggregate(vec![], func, walk(arg_expr)?)); + } + } + build(lower_inner_call(call)?, vec![], Outer::None) +} + +/// The per-series reducer for a `*_over_time` range-vector function together +/// with the expression in its matrix/sub-query argument slot (`φ` for +/// `quantile_over_time` is read from arg 0). Returns `None` for any other call, +/// so non-`*_over_time` functions fall through to the flat template. +fn over_time_reducer(call: &Call) -> Result> { + let simple = |f: InnerFunc| -> Result> { + Ok(Some((inner_func(&f), arg(call, 0)?))) + }; + match call.func.name { + "avg_over_time" => simple(InnerFunc::Avg), + "min_over_time" => simple(InnerFunc::Min), + "max_over_time" => simple(InnerFunc::Max), + "sum_over_time" => simple(InnerFunc::Sum), + "stddev_over_time" => simple(InnerFunc::StdDev), + "stdvar_over_time" => simple(InnerFunc::Variance), + "count_over_time" => simple(InnerFunc::Count), + "quantile_over_time" => { + let phi = quantile_param(num_arg(call, 0)?)?; + Ok(Some((inner_func(&InnerFunc::Quantile(phi)), arg(call, 1)?))) + } + _ => Ok(None), + } +} + +/// A (parenthesised) PromQL sub-query — `[range:res]`. +fn is_subquery(expr: &Expr) -> bool { + match expr { + Expr::Subquery(_) => true, + Expr::Paren(p) => is_subquery(&p.expr), + _ => false, + } +} + fn walk_aggregate(agg: &AggregateExpr) -> Result { let keys = resolve_group(agg)?; let outer = outer_kind(agg)?; diff --git a/crates/lower/tests/promql_conformance.rs b/crates/lower/tests/promql_conformance.rs index 87bc5840..05cfbf28 100644 --- a/crates/lower/tests/promql_conformance.rs +++ b/crates/lower/tests/promql_conformance.rs @@ -648,11 +648,59 @@ fn subquery_wraps_inner_query() { } #[test] -fn over_time_of_subquery_is_rejected__GAP() { - // SEMANTICS (PromQL): `max_over_time(rate(...)[1h:])` chains a subquery into - // a range-vector function. `extract_matrix` doesn't accept a subquery arg, - // so this canonical pattern is rejected today. - let _ = rejected("max_over_time(rate(demo_api_request_duration_seconds_count[5m])[1h:])"); +fn over_time_of_subquery_reduces_per_series() { + // SEMANTICS (PromQL): `max_over_time(rate(...)[1h:])` chains a sub-query into + // a range-vector function — the sub-query evaluates `rate` across a 1h range, + // then `max_over_time` takes the max of those samples *per series*. It lowers + // to a per-series `Max` reduction over a `Subquery` (issue #27). + let qe = ok("max_over_time(rate(demo_api_request_duration_seconds_count[5m])[1h:])"); + let QueryExpr::Aggregate { by, aggs, child, .. } = &qe else { + panic!("expected an Aggregate at the root, got {qe:?}"); + }; + assert!(by.is_empty(), "`*_over_time` has no grouping — reduces per series"); + assert!(matches!(aggs.as_slice(), [AggIntent::Max { .. }])); + // The reduction rides directly on the sub-query (the structural range marker + // that keeps it label-preserving), which wraps the inner `rate`. + assert!( + matches!(child.as_ref(), QueryExpr::Subquery { .. }), + "the `Max` reduces over a Subquery, got {child:?}" + ); + assert!(intents(&qe).iter().any(|i| matches!(i, AggIntent::Rate))); +} + +#[test] +fn quantile_over_time_of_subquery_carries_phi() { + // The `quantile_over_time` φ parameter is read from arg 0; the sub-query is + // arg 1. It lowers to a per-series `Quantile(φ)` over the `Subquery`. + let qe = ok("quantile_over_time(0.9, rate(demo[5m])[1h:])"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected an Aggregate, got {qe:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { q, .. }] if (*q - 0.9).abs() < 1e-9)); + assert!(matches!(child.as_ref(), QueryExpr::Subquery { .. })); +} + +#[test] +fn aggregation_over_over_time_of_subquery_keeps_labels() { + // `sum by (job) (max_over_time(rate(m[5m])[1h:]))` — the inner + // `max_over_time` is per-series (label-preserving), so the `job` label + // survives for the OUTER cross-series `sum by (job)` to group on. If the + // inner `Max` collapsed labels, `job` would not resolve here. + let qe = ok("sum by (job) (max_over_time(rate(demo{job=\"api\"}[5m])[1h:]))"); + let QueryExpr::Aggregate { by, aggs, child, .. } = &qe else { + panic!("expected outer Aggregate, got {qe:?}"); + }; + assert!(!by.is_empty(), "outer `sum by (job)` groups on a label"); + assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }])); + // Inner node is the per-series `max_over_time` reduction over the subquery. + let QueryExpr::Aggregate { by: inner_by, aggs: inner_aggs, child: inner_child, .. } = + child.as_ref() + else { + panic!("expected inner Aggregate, got {child:?}"); + }; + assert!(inner_by.is_empty()); + assert!(matches!(inner_aggs.as_slice(), [AggIntent::Max { .. }])); + assert!(matches!(inner_child.as_ref(), QueryExpr::Subquery { .. })); } // ─────────────────────────────────────────────────────────────────────────────