feat(lower): support *_over_time over a sub-query (#27)

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]));
                 }

crates/lower/src/promql.rs

@@ -152,7 +152,7 @@ fn walk(expr: &Expr) -> Result<L2> {
     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<L2> {
     }
 }
 
+/// 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<L2> {
+    if let Some((func, arg_expr)) = over_time_reducer(call)? {
+        if is_subquery(arg_expr) {
+            // `<agg>_over_time(<subquery>)` 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<Option<(AggFunc, &Expr)>> {
+    let simple = |f: InnerFunc| -> Result<Option<(AggFunc, &Expr)>> {
+        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 — `<inst>[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<L2> {
     let keys = resolve_group(agg)?;
     let outer = outer_kind(agg)?;

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 { .. }));
 }
 
 // ─────────────────────────────────────────────────────────────────────────────