feat(promql): range-vector functions over a sub-query argument (#55)

crates/frontend-promql/src/promql.rs

@@ -210,52 +210,84 @@ fn walk(expr: &Expr) -> Result<L2> {
 /// `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)?));
-        }
+    if let Some(l2) = range_fn_over_subquery(call)? {
+        return Ok(l2);
     }
     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)?)))
+/// A range-vector function applied to a **sub-query** — `f(<inst>[range:res])`.
+///
+/// Covers the whole range-vector family: the `*_over_time` reducers,
+/// `rate`/`irate`/`increase`, and the counter-derivatives
+/// (`changes`/`delta`/`idelta`/`deriv`/`resets`/`predict_linear`/
+/// `double_exponential_smoothing`). Each lowers to a per-series `Aggregate{[f]}`
+/// directly over the `PromQLSubquery` — the sub-query is the range context, so
+/// there is no separate `Window`/`TimeRange` (the L2→L3 converter treats the
+/// `Subquery` as the range marker). Returns `None` when `call` isn't a range
+/// function or its argument isn't a sub-query, so the flat matrix-selector
+/// template still handles `f(m[w])` (issues #42, #55).
+fn range_fn_over_subquery(call: &Call) -> Result<Option<L2>> {
+    // `rate`/`increase`/`irate` carry their window in the `AggFunc`; over a
+    // sub-query that window is the sub-query's own range.
+    if let "rate" | "irate" | "increase" = call.func.name {
+        let arg_expr = arg(call, 0)?;
+        let Some(range) = subquery_range(arg_expr) else {
+            return Ok(None);
+        };
+        let inner = if call.func.name == "increase" {
+            InnerFunc::Increase(range)
+        } else {
+            InnerFunc::Rate(range)
+        };
+        return Ok(Some(outer_aggregate(vec![], inner_func(&inner), walk(arg_expr)?)));
+    }
+
+    // `*_over_time` reducers + counter-derivatives: the func-kind, and the index
+    // of the matrix/sub-query argument (`quantile_over_time` reads φ from arg 0,
+    // so its matrix is arg 1; the rest take arg 0 + trailing scalar params).
+    let (inner, matrix_idx): (InnerFunc, usize) = match call.func.name {
+        "avg_over_time" => (InnerFunc::Avg, 0),
+        "min_over_time" => (InnerFunc::Min, 0),
+        "max_over_time" => (InnerFunc::Max, 0),
+        "sum_over_time" => (InnerFunc::Sum, 0),
+        "stddev_over_time" => (InnerFunc::StdDev, 0),
+        "stdvar_over_time" => (InnerFunc::Variance, 0),
+        "count_over_time" => (InnerFunc::Count, 0),
+        "quantile_over_time" => (InnerFunc::Quantile(quantile_param(num_arg(call, 0)?)?), 1),
+        "changes" => (InnerFunc::Changes, 0),
+        "delta" => (InnerFunc::Delta, 0),
+        "idelta" => (InnerFunc::IDelta, 0),
+        "deriv" => (InnerFunc::Deriv, 0),
+        "resets" => (InnerFunc::Resets, 0),
+        "predict_linear" => (InnerFunc::PredictLinear(num_arg(call, 1)?), 0),
+        "double_exponential_smoothing" | "holt_winters" => (
+            InnerFunc::DoubleExp {
+                smoothing: num_arg(call, 1)?,
+                trend: num_arg(call, 2)?,
+            },
+            0,
+        ),
+        _ => return Ok(None),
     };
-    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),
+    let arg_expr = arg(call, matrix_idx)?;
+    if !is_subquery(arg_expr) {
+        return Ok(None);
     }
+    Ok(Some(outer_aggregate(vec![], inner_func(&inner), walk(arg_expr)?)))
 }
 
 /// A (parenthesised) PromQL sub-query — `<inst>[range:res]`.
 fn is_subquery(expr: &Expr) -> bool {
+    subquery_range(expr).is_some()
+}
+
+/// The `range` of a (parenthesised) sub-query argument, if it is one.
+fn subquery_range(expr: &Expr) -> Option<Duration> {
     match expr {
-        Expr::Subquery(_) => true,
-        Expr::Paren(p) => is_subquery(&p.expr),
-        _ => false,
+        Expr::Subquery(sq) => Some(sq.range),
+        Expr::Paren(p) => subquery_range(&p.expr),
+        _ => None,
     }
 }
 

crates/frontend-promql/tests/promql_conformance.rs

@@ -907,11 +907,45 @@ fn counter_derivative_composes_in_binary_ops() {
 }
 
 #[test]
-fn counter_derivative_over_a_subquery_is_rejected__GAP() {
-    // Unlike `*_over_time` (issue #42), the counter-derivative functions do not
-    // yet accept a sub-query argument — only a bare matrix selector. This is
-    // valid PromQL and rejects cleanly (never mislowered); wiring them into the
-    // sub-query path is a follow-up to #42/#44.
-    let _ = rejected("changes(rate(m[5m])[1h:])");
-    let _ = rejected("delta(sum(m)[5m:])");
+fn range_functions_over_a_subquery_reduce_per_series() {
+    // Issue #55 — the whole range-vector family accepts a sub-query argument
+    // (generalizing `*_over_time`, #42): `rate`/`increase`/`irate` and the
+    // counter-derivatives. Each lowers to a per-series `Aggregate{[f]}` directly
+    // over the `Subquery` — the sub-query is the range context, so there is NO
+    // separate `TimeRange` (that would double the range).
+    for (q, want) in [
+        ("rate(sum(m)[5m:])", AggIntent::Rate),
+        ("increase(sum(m)[5m:])", AggIntent::Increase),
+        ("irate(sum(m)[5m:])", AggIntent::Rate), // irate shares the Rate intent
+        ("changes(rate(m[5m])[1h:])", AggIntent::Changes),
+        ("delta(sum(m)[5m:])", AggIntent::Delta),
+        ("deriv(sum(m)[10m:])", AggIntent::Deriv),
+        ("resets(sum(m)[5m:])", AggIntent::Resets),
+    ] {
+        let qe = ok(q);
+        let QueryExpr::Aggregate { by, aggs, child, .. } = &qe else {
+            panic!("{q}: expected an Aggregate, got {qe:?}");
+        };
+        assert!(by.is_empty(), "{q}: per-series, no grouping");
+        assert_eq!(aggs.as_slice(), std::slice::from_ref(&want), "{q}: wrong intent");
+        assert!(
+            matches!(child.as_ref(), QueryExpr::Subquery { .. }),
+            "{q}: reduces directly over the Subquery (no TimeRange), got {child:?}"
+        );
+    }
+}
+
+#[test]
+fn predict_linear_and_double_exp_over_a_subquery_carry_params() {
+    // The scalar params survive the sub-query path.
+    let pl = ok("predict_linear(sum(m)[1h:], 3600)");
+    assert!(intents(&pl)
+        .iter()
+        .any(|i| matches!(i, AggIntent::PredictLinear { seconds } if (*seconds - 3600.0).abs() < 1e-9)));
+    let de = ok("double_exponential_smoothing(sum(m)[10m:], 0.5, 0.3)");
+    assert!(intents(&de).iter().any(|i| matches!(
+        i,
+        AggIntent::DoubleExpSmoothing { smoothing, trend }
+            if (*smoothing - 0.5).abs() < 1e-9 && (*trend - 0.3).abs() < 1e-9
+    )));
 }

crates/l2/src/lower.rs

@@ -136,6 +136,12 @@ pub fn convert(
                             input: win_input,
                             ..
                         } => (win_input, Some(*duration)),
+                        // A `PromQLSubquery` is itself the range context (a range
+                        // function over a sub-query, `f(<inst>[range:res])` —
+                        // issues #42/#55). No separate `TimeRange`, and
+                        // `Rate`/`Increase`'s carried window is subsumed by the
+                        // sub-query's range.
+                        other @ LQueryExpr::PromQLSubquery { .. } => (other, None),
                         other => {
                             let range = match &aggs[0].func {
                                 AggFunc::Rate { window } | AggFunc::Increase { window } => {
@@ -147,12 +153,13 @@ pub fn convert(
                         }
                     };
                 // A counter-derivative range function is per-series over a range
-                // and always arrives under an L2 `Window` from the front ends.
-                // If one reaches here range-less (no `Window`, and unlike
-                // `Rate`/`Increase` it carries no window in its `AggFunc`),
-                // emitting it without a `TimeRange` would silently drop its
-                // window — reject the malformed tree instead (issue #71).
+                // and always arrives under an L2 `Window` (or, for a sub-query
+                // argument, over a `PromQLSubquery` — handled above). If one
+                // reaches here range-less over anything else, emitting it without
+                // a `TimeRange` would silently drop its window — reject the
+                // malformed tree instead (issue #71).
                 if time_range.is_none()
+                    && !matches!(agg_input_l2, LQueryExpr::PromQLSubquery { .. })
                     && matches!(
                         &aggs[0].func,
                         AggFunc::Changes