From 55588cd3318278353704e459c9f2b34e7ba281e9 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 27 May 2026 23:18:13 -0400 Subject: [PATCH] added e2e --- Cargo.lock | 8 + Cargo.toml | 1 + crates/core/src/intent_algebra/agg_intent.rs | 23 +- crates/core/src/intent_algebra/lower.rs | 105 +++---- crates/core/src/intent_algebra/query_expr.rs | 105 ++++--- crates/e2e/Cargo.toml | 8 + crates/e2e/src/lib.rs | 38 +++ crates/lower/src/promql.rs | 23 +- crates/lower/tests/promql_conformance.rs | 81 +++--- crates/lower/tests/promql_lowering.rs | 275 +++++++++++++------ 10 files changed, 405 insertions(+), 262 deletions(-) create mode 100644 crates/e2e/Cargo.toml create mode 100644 crates/e2e/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index e922417e..d59452de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -321,6 +321,14 @@ dependencies = [ "tokio", ] +[[package]] +name = "asap-e2e" +version = "0.1.0" +dependencies = [ + "asap-control-core", + "asap-control-lower", +] + [[package]] name = "async-compression" version = "0.4.19" diff --git a/Cargo.toml b/Cargo.toml index 7feaef0f..751492b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,5 +2,6 @@ members = [ "crates/core", "crates/lower", + "crates/e2e", ] resolver = "2" diff --git a/crates/core/src/intent_algebra/agg_intent.rs b/crates/core/src/intent_algebra/agg_intent.rs index 73922f41..802e1853 100644 --- a/crates/core/src/intent_algebra/agg_intent.rs +++ b/crates/core/src/intent_algebra/agg_intent.rs @@ -9,8 +9,6 @@ //! `ORDER BY value LIMIT k` stays as the `QueryExpr::Sort + Limit` operator //! pair. L1→L2→L3 lowering picks one or the other deterministically. -use std::time::Duration; - use serde::{Deserialize, Serialize}; use crate::intent_algebra::query_expr::DataModel; @@ -79,14 +77,11 @@ pub enum AggIntent { }, // ── Time-series streaming derivatives ──────────────────────────────── - // Carry PromQL's counter-reset adjustment; not equivalent to Sum/Count - // over a Window. Kept distinct so delta-set aggregators bind directly. - Rate { - window: Duration, - }, - Increase { - window: Duration, - }, + // Counter-reset adjustment; not equivalent to Sum/Count over a window. + // The temporal range lives on the enclosing `QueryExpr::TimeRange` node, + // not in the intent — this keeps the intent vocabulary range-agnostic. + Rate, + Increase, } impl AggIntent { @@ -94,7 +89,7 @@ impl AggIntent { /// this to skip non-applicable intents (e.g. `Rate` over a tabular source). pub fn requires(&self) -> DataModel { match self { - Self::Rate { .. } | Self::Increase { .. } => DataModel::TimeSeries, + Self::Rate | Self::Increase => DataModel::TimeSeries, _ => DataModel::Any, } } @@ -105,7 +100,7 @@ impl AggIntent { /// `rate`/`increase` carry their window in the intent. (Cross-series /// reductions like `sum`/`avg` over a series set return `false`.) pub fn is_per_series(&self) -> bool { - matches!(self, Self::Rate { .. } | Self::Increase { .. }) + matches!(self, Self::Rate | Self::Increase) } /// The positional input column this intent reduces, if it carries one. @@ -146,8 +141,8 @@ impl AggIntent { // (the L4 sketch-bound IR upgrades the dtype). AggIntent::TopK { k, .. } => col(&format!("topk_{k}"), DataType::Utf8, false), AggIntent::Cardinality { .. } => col("cardinality", DataType::Int64, false), - AggIntent::Rate { .. } => col("rate", DataType::Float64, false), - AggIntent::Increase { .. } => col("increase", DataType::Float64, false), + AggIntent::Rate => col("rate", DataType::Float64, false), + AggIntent::Increase => col("increase", DataType::Float64, false), } } } diff --git a/crates/core/src/intent_algebra/lower.rs b/crates/core/src/intent_algebra/lower.rs index 0ad4caf3..3daa54a7 100644 --- a/crates/core/src/intent_algebra/lower.rs +++ b/crates/core/src/intent_algebra/lower.rs @@ -10,6 +10,8 @@ //! schema every `ColumnId` indexes into, so positional resolution downstream //! is total. +use std::time::Duration; + use thiserror::Error; use crate::intent_algebra::agg_intent::AggIntent; @@ -21,7 +23,7 @@ use crate::intent_algebra::expr_ir::{ColumnRef, L2Expr, L3Expr, L3Scalar}; use crate::intent_algebra::names::BindingName; use crate::intent_algebra::query_expr::{ PartitionKeys as CPartitionKeys, Predicate, ProjectItem, QueryExpr as CQueryExpr, SortKey, - Source, WindowKind, + Source, }; use crate::intent_algebra::relational::{AggFunc, QueryExpr as LQueryExpr, SourceSpec}; use crate::intent_algebra::schema::{ColumnId, Schema}; @@ -106,32 +108,52 @@ pub fn convert( // 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 { - duration, - slide, - input: win_input, - } => (win_input, Some((*duration, *slide))), - other => (other, None), - }; - let agg_child = convert(agg_input_l2, fallback, acc)?; - let agg_in_schema = agg_child.output_schema()?; + // Extract the temporal range. Two sources: + // 1. An L2 `Window` child (emitted for `*_over_time` functions). + // 2. `Rate`/`Increase` AggFunc (carry their own range; no L2 Window). + // The range becomes a `TimeRange` node wrapping the inner scan + // rather than a `Window` node above the aggregate — `Window` is + // reserved for streaming query-repetition semantics. + let (agg_input_l2, time_range): (&LQueryExpr, Option) = + match input.as_ref() { + LQueryExpr::Window { + duration, + input: win_input, + .. + } => (win_input, Some(*duration)), + other => { + let range = match &aggs[0].func { + AggFunc::Rate { window } | AggFunc::Increase { window } => { + Some(*window) + } + _ => None, + }; + (other, range) + } + }; + let agg_child_raw = convert(agg_input_l2, fallback, acc)?; + let agg_in_schema = agg_child_raw.output_schema()?; let intent = agg_func_to_intent( &aggs[0].func, acc, resolve_agg_col(&aggs[0].col, &agg_in_schema)?, ); + // Wrap the child in TimeRange when there is a range. + let agg_child = match time_range { + Some(range) => CQueryExpr::TimeRange { + range, + child: Box::new(agg_child_raw), + }, + None => agg_child_raw, + }; // Resolve the group keys positionally against the aggregate's // input so the grouping lives in `Aggregate.by` — the *same* - // shape SQL produces. Only when this is a *non-windowed* - // reduction: an instant aggregate (`sum by (job) (m)`) or a - // cross-series reduction over a label-preserving `rate`/ - // `increase` (`sum by (job) (rate(m[w]))`), where the key is in - // scope. A *windowed* reduction here is per-series (e.g. - // `avg_over_time`) — its keys belong to an enclosing level, so - // keep the legacy name-based `Partition` marker instead of - // folding them into a per-series `by`. - let by = if window.is_none() { + // shape SQL produces. Only for instant (non-range) aggregates: + // an instant aggregate (`sum by (job) (m)`) or a cross-series + // reduction over a label-preserving `rate`/`increase`. + // A range reduction's keys belong to an enclosing level, so fall + // back to a name-based `Partition` instead. + let by = if time_range.is_none() { resolve_column_refs(keys, &agg_in_schema).unwrap_or_default() } else { Vec::new() @@ -144,28 +166,12 @@ pub fn convert( having: None, child: Box::new(agg_child), }; - let sketch = match window { - Some((duration, slide)) => CQueryExpr::Window { - kind: if slide.is_some() { - WindowKind::Sliding - } else { - WindowKind::Tumbling - }, - size: duration, - slide, - child: Box::new(aggregate), - }, - None => aggregate, - }; return Ok(if grouped_positionally { - sketch + aggregate } else { - // Fallback (windowed per-series reduction): keep the legacy - // name-based `Partition`. These keys are PromQL labels - // (unqualified), so the bare names suffice. CQueryExpr::Partition { keys: CPartitionKeys::By(ref_names(keys)), - child: Box::new(sketch), + child: Box::new(aggregate), } }); } @@ -205,18 +211,12 @@ pub fn convert( } } + // Standalone L2 Window (bare range vector `m[5m]` not inside a + // recognized single-stat Aggregate): map to TimeRange. LQueryExpr::Window { - duration, - slide, - input, - } => CQueryExpr::Window { - kind: if slide.is_some() { - WindowKind::Sliding - } else { - WindowKind::Tumbling - }, - size: *duration, - slide: *slide, + duration, input, .. + } => CQueryExpr::TimeRange { + range: *duration, child: Box::new(convert(input, fallback, acc)?), }, @@ -521,8 +521,9 @@ fn agg_func_to_intent(func: &AggFunc, acc: &AccuracyTarget, col: Option AggIntent::Rate { window: *window }, - AggFunc::Increase { window } => AggIntent::Increase { window: *window }, + // Range is on the enclosing TimeRange node; intent carries no window. + AggFunc::Rate { .. } => AggIntent::Rate, + AggFunc::Increase { .. } => AggIntent::Increase, } } @@ -621,7 +622,7 @@ mod tests { let CQueryExpr::Aggregate { aggs, .. } = &l3 else { panic!("expected Aggregate, got {l3:?}"); }; - assert_eq!(aggs, &vec![AggIntent::Sum { col: None }]); + assert_eq!(aggs, &[AggIntent::Sum { col: None }]); } /// `SELECT region, SUM(bytes), COUNT(*) FROM logs JOIN meta … GROUP BY region` diff --git a/crates/core/src/intent_algebra/query_expr.rs b/crates/core/src/intent_algebra/query_expr.rs index eed720ea..2fc76f78 100644 --- a/crates/core/src/intent_algebra/query_expr.rs +++ b/crates/core/src/intent_algebra/query_expr.rs @@ -312,6 +312,19 @@ pub enum QueryExpr { child: Box, }, + /// Temporal range selection — "look back `range` of history for this + /// computation." Used for all range-vector functions: `rate`, `increase`, + /// `*_over_time`. The range is distinct from both a streaming `Window` + /// (which is for query-repetition) and a row-level `Filter`. + /// + /// Structural marker: an `Aggregate` whose direct child is a `TimeRange` + /// is a *per-series* reduction (label-preserving); one whose child is a + /// plain `Scan` or another `Aggregate` is a *cross-series* reduction. + TimeRange { + range: Duration, + child: Box, + }, + /// SQL analytic window function: `func(args) OVER (PARTITION BY … ORDER BY …)`. /// Output schema = child schema + one column named `output_name` (the name /// the enclosing `Project` references). Window frames are not modelled yet. @@ -349,28 +362,11 @@ impl QueryExpr { match self { QueryExpr::Scan { schema, .. } => Ok(schema.clone()), - // ψ — a window reshapes the time axis but not the column set. Over a - // 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. - // 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), - }, + // ψ — streaming window (tumbling / sliding / session) for query + // repetition. Does not change the column schema; passes through. + // Per-series range reductions (`rate`, `*_over_time`) now use the + // `TimeRange` node instead, so this arm is a simple pass-through. + QueryExpr::Window { child, .. } => child.output_schema_in(scope), QueryExpr::Aggregate { by, @@ -381,12 +377,14 @@ impl QueryExpr { } => { let in_schema = child.output_schema_in(scope)?; - // 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() { + // 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 { .. }); + 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])); } @@ -450,7 +448,8 @@ impl QueryExpr { | QueryExpr::Partition { child, .. } | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } - | QueryExpr::Subquery { child, .. } => child.output_schema_in(scope), + | QueryExpr::Subquery { child, .. } + | QueryExpr::TimeRange { child, .. } => child.output_schema_in(scope), // π — one output column per projection item. Each item's type is // inferred from its expression against the child schema; the name @@ -762,9 +761,9 @@ mod tests { fn per_series_rate_preserves_labels() { // A per-series range reduction (`rate`) is label-preserving: it produces // one value per series, so every label survives and only the sample - // value is replaced (kept named `value`). This is what lets an outer - // cross-series `Aggregate.by` group on those labels positionally. - let child = scan( + // value is replaced (kept named `value`). The TimeRange child is the + // structural marker; the outer Aggregate carries the Rate intent. + let scan_node = scan( vec![ col("ts", DataType::Timestamp, false), col("value", DataType::Float64, false), @@ -775,12 +774,13 @@ mod tests { ); let rate = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Rate { - window: Duration::from_secs(300), - }], + aggs: vec![AggIntent::Rate], output_names: vec![], having: None, - child: Box::new(child), + child: Box::new(QueryExpr::TimeRange { + range: Duration::from_secs(300), + child: Box::new(scan_node), + }), }; let s = rate.output_schema().unwrap(); assert_eq!( @@ -796,13 +796,12 @@ mod tests { } #[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( + fn over_time_reduction_preserves_labels() { + // `*_over_time` lowers to `Aggregate { by:[], [reducer], TimeRange { Scan } }`: + // a per-series time-range reduction. The TimeRange child confers per-series + // semantics on otherwise cross-series intents like `Avg`, so an outer + // `sum by(job)(avg_over_time(...))` resolves its key positionally. + let scan_node = scan( vec![ col("ts", DataType::Timestamp, false), col("value", DataType::Float64, false), @@ -811,16 +810,14 @@ mod tests { 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 avg_over_time = QueryExpr::Aggregate { + by: vec![], + aggs: vec![AggIntent::Avg { col: None }], + output_names: vec![], + having: None, + child: Box::new(QueryExpr::TimeRange { + range: Duration::from_secs(300), + child: Box::new(scan_node), }), }; let s = avg_over_time.output_schema().unwrap(); @@ -830,7 +827,7 @@ mod tests { .map(|c| c.name.as_str()) .collect::>(), vec!["ts", "value", "job"], - "windowed per-series reduction preserves labels; value kept named `value`" + "TimeRange-child marks per-series: labels preserved, value renamed" ); assert!( s.column_id("job").is_some(), diff --git a/crates/e2e/Cargo.toml b/crates/e2e/Cargo.toml new file mode 100644 index 00000000..37d45e9f --- /dev/null +++ b/crates/e2e/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "asap-e2e" +version = "0.1.0" +edition = "2021" + +[dependencies] +asap-control-core = { path = "../core" } +asap-control-lower = { path = "../lower" } diff --git a/crates/e2e/src/lib.rs b/crates/e2e/src/lib.rs new file mode 100644 index 00000000..aee3779f --- /dev/null +++ b/crates/e2e/src/lib.rs @@ -0,0 +1,38 @@ +//! Shared test fixtures for the ASAP e2e test suite. +//! +//! This crate owns the integration tests that verify (a) correct L3 IR output +//! for each input query workload, and (b) semantically equivalent queries in +//! different languages map to the same IR. +//! +//! `fixtures` provides column/schema constructors used across test files. +//! Expected IR trees are always hand-constructed inside each test — nothing +//! here derives or computes expected outputs. + +pub mod fixtures { + use asap_control_core::intent_algebra::schema::{Column, DataType, Schema}; + + pub fn ts_col() -> Column { + Column::new("ts", DataType::Timestamp, false) + } + + pub fn value_col() -> Column { + Column::new("value", DataType::Float64, false) + } + + pub fn label_col(name: &str) -> Column { + Column::new(name, DataType::Utf8, true) + } + + /// Canonical PromQL leaf schema: `(ts: Timestamp, value: Float64)` plus + /// any label columns referenced in the query, in the order the Binder + /// appends them (alphabetical after dedup). + pub fn metric_schema(labels: &[&str]) -> Schema { + let mut cols = vec![ts_col(), value_col()]; + cols.extend(labels.iter().map(|n| label_col(n))); + Schema { + columns: cols, + time_index: Some(0), + unique_keys: vec![], + } + } +} diff --git a/crates/lower/src/promql.rs b/crates/lower/src/promql.rs index 85bb46bf..c3b96981 100644 --- a/crates/lower/src/promql.rs +++ b/crates/lower/src/promql.rs @@ -405,11 +405,16 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { // order-by-value + limit. let heavy_hitter = descending && matches!(inner.func, Some(InnerFunc::Count)); if heavy_hitter { - let scan = window_scan(inner); + // Preserve the Count intent in L3 so the intent algebra is + // explicit about what is being computed. L4 may fuse the Count + // and TopK into a single-pass heavy-hitter sketch (SpaceSaving / + // CMS-with-heap), but that is a cost-model decision, not an L3 + // concern. + let count_agg = windowed_aggregate(inner, vec![], inner_func(&InnerFunc::Count)); Ok(L2::TopK { k, by: keys, - input: Box::new(scan), + input: Box::new(count_agg), }) } else { let func = match &inner.func { @@ -481,20 +486,6 @@ fn outer_aggregate(keys: Vec, func: AggFunc, input: L2) -> L2 { } } -/// `[Window{w}] → Filter(Source)` with no aggregate (the heavy-hitter TopK -/// child — the sketch counts directly off the scan). -fn window_scan(inner: Inner) -> L2 { - let base = filtered_source(inner.metric, inner.matchers); - match inner.window { - Some(w) => L2::Window { - duration: w, - slide: None, - input: Box::new(base), - }, - None => base, - } -} - fn filtered_source(metric: String, matchers: Vec) -> L2 { let source = L2::Source(SourceSpec::new(metric)); if matchers.is_empty() { diff --git a/crates/lower/tests/promql_conformance.rs b/crates/lower/tests/promql_conformance.rs index 4f4c780c..35fe869b 100644 --- a/crates/lower/tests/promql_conformance.rs +++ b/crates/lower/tests/promql_conformance.rs @@ -69,6 +69,7 @@ fn collect(e: &QueryExpr, out: &mut Vec) { collect(child, out); } QueryExpr::Window { child, .. } + | QueryExpr::TimeRange { child, .. } | QueryExpr::Partition { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } @@ -108,6 +109,7 @@ fn first_scan(e: &QueryExpr) -> (String, usize) { (name, predicates.len()) } QueryExpr::Window { child, .. } + | QueryExpr::TimeRange { child, .. } | QueryExpr::Aggregate { child, .. } | QueryExpr::Partition { child, .. } | QueryExpr::Filter { child, .. } @@ -153,13 +155,14 @@ fn name_label_selects_the_metric() { } #[test] -fn range_vector_selector_is_a_window() { +fn range_vector_selector_is_time_range() { // SEMANTICS: `[5m]` turns an instant vector into a range vector. + // In L3 this is a dedicated `TimeRange` node (not a streaming `Window`). let qe = ok("node_cpu_seconds_total[5m]"); - let QueryExpr::Window { size, .. } = &qe else { - panic!("expected Window for a range-vector selector, got {qe:?}"); + let QueryExpr::TimeRange { range, .. } = &qe else { + panic!("expected TimeRange for a range-vector selector, got {qe:?}"); }; - assert_eq!(*size, Duration::from_secs(300)); + assert_eq!(*range, Duration::from_secs(300)); } // ───────────────────────────────────────────────────────────────────────────── @@ -168,15 +171,18 @@ fn range_vector_selector_is_a_window() { // ───────────────────────────────────────────────────────────────────────────── #[test] -fn rate_carries_its_window_in_the_intent() { - // SEMANTICS: per-second average rate over the range; the window IS the rate - // parameter, so no separate Window node. +fn rate_range_lives_in_time_range_node() { + // SEMANTICS: per-second average rate; the temporal range lives on the + // enclosing `TimeRange` node, not inside the intent. let qe = ok("rate(http_requests_total[5m])"); - assert!(matches!(&qe, QueryExpr::Aggregate { .. })); - assert!(has( - &qe, - |i| matches!(i, AggIntent::Rate { window } if *window == Duration::from_secs(300)) - )); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate, got {qe:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Rate])); + let QueryExpr::TimeRange { range, .. } = child.as_ref() else { + panic!("expected TimeRange child, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(300)); } #[test] @@ -184,16 +190,21 @@ fn irate_maps_to_rate_intent() { // SEMANTICS: instant rate from the last two samples; same intent vocabulary. assert!(has(&ok("irate(http_requests_total[1m])"), |i| matches!( i, - AggIntent::Rate { .. } + AggIntent::Rate ))); } #[test] -fn increase_maps_to_increase_intent() { - assert!(has(&ok("increase(http_requests_total[1h])"), |i| matches!( - i, - AggIntent::Increase { window } if *window == Duration::from_secs(3600) - ))); +fn increase_range_lives_in_time_range_node() { + let qe = ok("increase(http_requests_total[1h])"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate, got {qe:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Increase])); + let QueryExpr::TimeRange { range, .. } = child.as_ref() else { + panic!("expected TimeRange child, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(3600)); } // ───────────────────────────────────────────────────────────────────────────── @@ -289,7 +300,7 @@ fn sum_of_rate_is_two_levels() { assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }])); assert!(matches!( child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }]) + QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate]) )); } @@ -309,7 +320,7 @@ fn sum_by_of_rate_groups_outer_level() { // child is the inner per-series Rate aggregate. assert!(matches!( child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }]) + QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate]) )); } @@ -328,14 +339,12 @@ fn sum_by_of_over_time_groups_outer_level() { }; 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:?}"); + // child is the inner per-series reduction: Aggregate{Avg} over TimeRange. + let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { + panic!("expected Aggregate (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 { .. }]) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Avg { .. }])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } // ───────────────────────────────────────────────────────────────────────────── @@ -344,9 +353,9 @@ fn sum_by_of_over_time_groups_outer_level() { // ───────────────────────────────────────────────────────────────────────────── #[test] -fn over_time_functions_window_then_reduce() { - // SEMANTICS: reduce the samples WITHIN each series over the range → Window - // over the matching reduce intent. +fn over_time_functions_reduce_over_time_range() { + // SEMANTICS: reduce the samples WITHIN each series over the range → + // Aggregate over TimeRange (per-series, label-preserving). for (q, want) in [ ("avg_over_time(go_goroutines[5m])", "avg"), ("max_over_time(process_resident_memory_bytes[1d])", "max"), @@ -356,8 +365,8 @@ fn over_time_functions_window_then_reduce() { ] { let qe = ok(q); assert!( - matches!(&qe, QueryExpr::Window { .. }), - "{q}: expected Window" + matches!(&qe, QueryExpr::Aggregate { .. }), + "{q}: expected Aggregate" ); let matched = intents(&qe).iter().any(|i| match want { "avg" => matches!(i, AggIntent::Avg { .. }), @@ -372,9 +381,9 @@ fn over_time_functions_window_then_reduce() { } #[test] -fn quantile_over_time_is_window_over_quantile() { +fn quantile_over_time_is_aggregate_over_time_range() { let qe = ok("quantile_over_time(0.9, request_latency_seconds[5m])"); - assert!(matches!(&qe, QueryExpr::Window { .. })); + assert!(matches!(&qe, QueryExpr::Aggregate { .. })); assert!(has( &qe, |i| matches!(i, AggIntent::Quantile { q, .. } if (*q - 0.9).abs() < 1e-9) @@ -394,7 +403,7 @@ fn histogram_quantile_over_rate() { panic!("expected Aggregate{{Quantile}}, got {qe:?}"); }; assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { q, .. }] if (*q - 0.9).abs() < 1e-9)); - assert!(has(&qe, |i| matches!(i, AggIntent::Rate { .. }))); + assert!(has(&qe, |i| matches!(i, AggIntent::Rate))); } #[test] @@ -571,7 +580,7 @@ fn subquery_wraps_inner_query() { // SEMANTICS: `[range:res]` evaluates the inner query across a range. let qe = ok("rate(demo_api_request_duration_seconds_count[5m])[1h:]"); assert!(matches!(&qe, QueryExpr::Subquery { .. })); - assert!(has(&qe, |i| matches!(i, AggIntent::Rate { .. }))); + assert!(has(&qe, |i| matches!(i, AggIntent::Rate))); } #[test] diff --git a/crates/lower/tests/promql_lowering.rs b/crates/lower/tests/promql_lowering.rs index c9ceee49..fe176112 100644 --- a/crates/lower/tests/promql_lowering.rs +++ b/crates/lower/tests/promql_lowering.rs @@ -3,8 +3,7 @@ use std::time::Duration; use asap_control_core::intent_algebra::{ - AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, L3Scalar, PartitionKeys, QueryExpr, - Source, WindowKind, + AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, L3Scalar, PartitionKeys, QueryExpr, Source, }; use asap_control_core::types::AccuracyTarget; use asap_control_core::workload::{ @@ -56,27 +55,23 @@ fn regex_matcher_lowers_to_regex_compareop() { assert!(matches!(right.as_ref(), L3Expr::Literal(L3Scalar::Utf8(v)) if v == "/api/.*")); } -// ── *_over_time → Window over Aggregate ───────────────────────────────────────── +// ── *_over_time → Aggregate over TimeRange ────────────────────────────────────── #[test] -fn quantile_over_time_is_window_over_aggregate() { +fn quantile_over_time_is_time_range_aggregate() { let qe = lower(r#"quantile_over_time(0.99, http_request_duration{env="prod"}[5m])"#); - let QueryExpr::Window { - kind, size, child, .. - } = &qe - else { - panic!("expected Window, got {qe:?}"); - }; - assert_eq!(*kind, WindowKind::Tumbling); - assert_eq!(*size, Duration::from_secs(300)); let QueryExpr::Aggregate { by, aggs, child, .. - } = child.as_ref() + } = &qe else { - panic!("expected Aggregate under Window, got {child:?}"); + panic!("expected Aggregate, got {qe:?}"); }; assert!(by.is_empty()); assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { q, .. }] if (*q - 0.99).abs() < 1e-9)); + let QueryExpr::TimeRange { range, child } = child.as_ref() else { + panic!("expected TimeRange child, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(300)); // The label matcher folded onto the Scan. assert!(matches!(child.as_ref(), QueryExpr::Scan { predicates, .. } if predicates.len() == 1)); } @@ -97,48 +92,54 @@ fn outer_sum_by_over_quantile_over_time_groups_positionally() { }; assert_eq!(by, &vec![2]); assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }])); - // Inner: Window over Aggregate{Quantile} (the per-series over_time reduction). - let QueryExpr::Window { child, .. } = child.as_ref() else { - panic!("expected Window (per-series over_time) under the outer Sum, got {child:?}"); + // Inner: Aggregate{Quantile} over TimeRange (per-series over_time reduction). + let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { + panic!("expected Aggregate (quantile_over_time) under the outer Sum, got {child:?}"); }; - assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } #[test] fn avg_over_time_maps_to_avg_intent() { let qe = lower("avg_over_time(cpu_seconds_total[10m])"); - let QueryExpr::Window { size, child, .. } = &qe else { - panic!("expected Window, got {qe:?}"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate, got {qe:?}"); }; - assert_eq!(*size, Duration::from_secs(600)); - assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Avg { .. }]) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Avg { .. }])); + let QueryExpr::TimeRange { range, .. } = child.as_ref() else { + panic!("expected TimeRange child, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(600)); } #[test] fn stddev_and_stdvar_over_time() { let qe = lower("stddev_over_time(m[5m])"); - let QueryExpr::Window { child, .. } = &qe else { - panic!("expected Window"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate"); }; assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::StdDev { population: false, .. }]) + aggs.as_slice(), + [AggIntent::StdDev { + population: false, + .. + }] )); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); let qe = lower("stdvar_over_time(m[5m])"); - let QueryExpr::Window { child, .. } = &qe else { - panic!("expected Window"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate"); }; assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Variance { population: false, .. }]) + aggs.as_slice(), + [AggIntent::Variance { + population: false, + .. + }] )); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } #[test] @@ -153,10 +154,18 @@ fn histogram_quantile_wraps_inner_in_quantile() { let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { panic!("expected inner Aggregate{{Rate}}, got {child:?}"); }; + assert!(matches!(aggs.as_slice(), [AggIntent::Rate])); + let QueryExpr::TimeRange { + range, + child: tr_child, + } = child.as_ref() + else { + panic!("expected TimeRange under Rate, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(300)); assert!( - matches!(aggs.as_slice(), [AggIntent::Rate { window }] if *window == Duration::from_secs(300)) + matches!(tr_child.as_ref(), QueryExpr::Scan { predicates, .. } if predicates.len() == 1) ); - assert!(matches!(child.as_ref(), QueryExpr::Scan { predicates, .. } if predicates.len() == 1)); } #[test] @@ -181,28 +190,29 @@ fn histogram_quantile_over_sum_by_le_preserves_grouping() { // ── rate / increase carry their own window (no Window node) ───────────────────── #[test] -fn rate_has_no_window_node() { +fn rate_has_time_range_child_not_window() { let qe = lower("rate(http_requests_total[5m])"); let QueryExpr::Aggregate { aggs, child, .. } = &qe else { - panic!("expected Aggregate (no Window) for rate, got {qe:?}"); + panic!("expected Aggregate for rate, got {qe:?}"); }; - assert!(matches!( - aggs.as_slice(), - [AggIntent::Rate { window }] if *window == Duration::from_secs(300) - )); - assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); + assert!(matches!(aggs.as_slice(), [AggIntent::Rate])); + let QueryExpr::TimeRange { range, .. } = child.as_ref() else { + panic!("expected TimeRange child (not Window), got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(300)); } #[test] fn increase_maps_to_increase_intent() { let qe = lower("increase(errors_total[1h])"); - let QueryExpr::Aggregate { aggs, .. } = &qe else { + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { panic!("expected Aggregate for increase, got {qe:?}"); }; - assert!(matches!( - aggs.as_slice(), - [AggIntent::Increase { window }] if *window == Duration::from_secs(3600) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Increase])); + let QueryExpr::TimeRange { range, .. } = child.as_ref() else { + panic!("expected TimeRange child, got {child:?}"); + }; + assert_eq!(*range, Duration::from_secs(3600)); } // ── outer aggregation over an inner range-vector func is two levels ───────────── @@ -219,10 +229,8 @@ fn sum_over_rate_keeps_both_levels() { let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { panic!("expected inner Aggregate{{Rate}}, got {child:?}"); }; - assert!( - matches!(aggs.as_slice(), [AggIntent::Rate { window }] if *window == Duration::from_secs(300)) - ); - assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); + assert!(matches!(aggs.as_slice(), [AggIntent::Rate])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } #[test] @@ -241,7 +249,7 @@ fn sum_by_over_rate_groups_the_outer_sum() { assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }])); assert!(matches!( child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }]) + QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate]) )); } @@ -255,7 +263,7 @@ fn count_over_rate_keeps_both_levels() { assert!(matches!(aggs.as_slice(), [AggIntent::Cardinality { .. }])); assert!(matches!( child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }]) + QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate]) )); } @@ -264,13 +272,11 @@ fn count_over_rate_keeps_both_levels() { #[test] fn count_over_time_is_count_intent() { let qe = lower("count_over_time(m[5m])"); - let QueryExpr::Window { child, .. } = &qe else { - panic!("expected Window"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected Aggregate"); }; - assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Count { .. }]) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Count { .. }])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } #[test] @@ -287,14 +293,12 @@ fn outer_count_is_cardinality() { }; 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 (per-series count_over_time) under the cardinality, got {child:?}"); + // Inner: Aggregate{Count} over TimeRange (per-series count_over_time). + let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { + panic!("expected Aggregate (count_over_time) under the cardinality, got {child:?}"); }; - assert!(matches!( - child.as_ref(), - QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Count { .. }]) - )); + assert!(matches!(aggs.as_slice(), [AggIntent::Count { .. }])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); } // ── topk / bottomk ──────────────────────────────────────────────────────────── @@ -312,11 +316,15 @@ fn topk_over_count_is_heavy_hitter_topk() { // `service` is the only group key → resolved to a positional ColumnId. assert_eq!(by.len(), 1); assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 10, .. }])); - // The heavy-hitter sketch counts directly off the windowed scan. - let QueryExpr::Window { size, child, .. } = child.as_ref() else { - panic!("expected Window under TopK Aggregate, got {child:?}"); + // The count_over_time under the TopK is a TimeRange-backed aggregate. + let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { + panic!("expected Aggregate (count_over_time) under TopK, got {child:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Count { .. }])); + let QueryExpr::TimeRange { range, child } = child.as_ref() else { + panic!("expected TimeRange under Count aggregate, got {child:?}"); }; - assert_eq!(*size, Duration::from_secs(60)); + assert_eq!(*range, Duration::from_secs(60)); assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); } @@ -339,6 +347,36 @@ fn topk_over_avg_is_generic_sort_limit() { ); } +#[test] +fn topk_over_sum_is_generic_sort_limit() { + // Only `count_over_time` triggers the heavy-hitter path; `sum_over_time` + // falls back to generic sort + limit. + let qe = lower("topk(5, sum_over_time(m[5m]))"); + assert!( + matches!(&qe, QueryExpr::Limit { .. }), + "expected Limit (generic sort), got {qe:?}" + ); + assert!(has_intent(&qe, |i| matches!(i, AggIntent::Sum { .. }))); + assert!(!has_intent(&qe, |i| matches!(i, AggIntent::TopK { .. }))); +} + +#[test] +fn bottomk_over_count_is_generic_sort_ascending() { + // `bottomk` is never a heavy-hitter (descending=false), even over count. + let qe = lower("bottomk(3, count_over_time(m[5m]))"); + let QueryExpr::Limit { n, child, .. } = &qe else { + panic!("expected Limit, got {qe:?}"); + }; + assert_eq!(*n, 3); + let QueryExpr::Sort { keys, .. } = child.as_ref() else { + panic!("expected Sort"); + }; + assert!(keys[0].ascending, "bottomk ranks ascending"); + // Count intent is still present (as the inner aggregate), no TopK. + assert!(has_intent(&qe, |i| matches!(i, AggIntent::Count { .. }))); + assert!(!has_intent(&qe, |i| matches!(i, AggIntent::TopK { .. }))); +} + #[test] fn bottomk_is_always_generic_sort_ascending() { let qe = lower("bottomk(3, count_over_time(m[5m]))"); @@ -352,6 +390,28 @@ fn bottomk_is_always_generic_sort_ascending() { assert!(keys[0].ascending, "bottomk ranks ascending"); } +#[test] +fn topk_count_output_schema_carries_group_key() { + // The inner Count is per-series (label-preserving), so the group-by key + // (`service`) flows through to the outer TopK's `by` column. Leaf schema = + // [ts, value, service] → TopK groups on service (col 2). + let qe = lower("topk by (service) (5, count_over_time(m[1m]))"); + let QueryExpr::Aggregate { + by, aggs, child, .. + } = &qe + else { + panic!("expected Aggregate{{TopK}}, got {qe:?}"); + }; + assert_eq!(by, &vec![2], "service is col 2 in [ts, value, service]"); + assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }])); + // Inner Count aggregate is visible with its TimeRange child. + let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else { + panic!("expected inner Aggregate{{Count}}, got {child:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Count { .. }])); + assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. })); +} + // ── binary ops ──────────────────────────────────────────────────────────────── #[test] @@ -362,10 +422,10 @@ fn binary_op_division() { }; assert_eq!(*op, BinaryOpKind::Arith(ArithOp::Div)); assert!( - matches!(lhs.as_ref(), QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }])) + matches!(lhs.as_ref(), QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate])) ); assert!( - matches!(rhs.as_ref(), QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate { .. }])) + matches!(rhs.as_ref(), QueryExpr::Aggregate { aggs, .. } if matches!(aggs.as_slice(), [AggIntent::Rate])) ); } @@ -402,6 +462,38 @@ fn binary_op_binds_each_branch_against_its_own_schema() { ); } +/// Collect every `AggIntent` in the tree, root-to-leaf. +fn all_intents(e: &QueryExpr) -> Vec { + let mut out = Vec::new(); + collect_intents(e, &mut out); + out +} + +fn collect_intents(e: &QueryExpr, out: &mut Vec) { + match e { + QueryExpr::Aggregate { aggs, child, .. } => { + out.extend(aggs.iter().cloned()); + collect_intents(child, out); + } + QueryExpr::Partition { child, .. } + | QueryExpr::Window { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } => collect_intents(child, out), + QueryExpr::BinaryOp { lhs, rhs, .. } => { + collect_intents(lhs, out); + collect_intents(rhs, out); + } + _ => {} + } +} + +/// True if any `AggIntent` anywhere in the tree satisfies `pred`. +fn has_intent bool>(e: &QueryExpr, pred: F) -> bool { + all_intents(e).iter().any(pred) +} + /// Column names on the first `Scan` reachable by descending single-child nodes. fn scan_columns(e: &QueryExpr) -> Vec { match e { @@ -409,6 +501,7 @@ fn scan_columns(e: &QueryExpr) -> Vec { QueryExpr::Partition { child, .. } | QueryExpr::Aggregate { child, .. } | QueryExpr::Window { child, .. } + | QueryExpr::TimeRange { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } => scan_columns(child), @@ -477,10 +570,7 @@ fn accuracy_target_flows_into_quantile_intent() { AccuracyTarget::Epsilon(0.01), ) .unwrap(); - let QueryExpr::Window { child, .. } = &qe else { - panic!("expected Window"); - }; - let QueryExpr::Aggregate { aggs, .. } = child.as_ref() else { + let QueryExpr::Aggregate { aggs, .. } = &qe else { panic!("expected Aggregate"); }; assert!(matches!( @@ -492,19 +582,23 @@ fn accuracy_target_flows_into_quantile_intent() { // ── schema flow (positional, carried on Scan; derived on demand) ───────────────── #[test] -fn aggregate_output_schema_is_single_quantile_column() { +fn aggregate_output_schema_preserves_time_axis_and_labels() { let qe = lower(r#"quantile_over_time(0.99, http_request_duration{env="prod"}[5m])"#); - // Window requires its child to carry a time axis; the Aggregate beneath it - // strips it, so derive the schema at the Aggregate node. - let QueryExpr::Window { child, .. } = &qe else { - panic!("expected Window"); - }; - let schema = child.output_schema().expect("aggregate schema"); + // Per-series reduction: the root is Aggregate { TimeRange { Scan } }. + // The Binder adds all referenced label names (group keys AND filter + // predicate columns) to the scan schema, so `env` appears as a column + // even though it is only used as a filter. + // per_series_reduction_schema preserves the time axis and all label columns. + let QueryExpr::Aggregate { .. } = &qe else { + panic!("expected Aggregate, got {qe:?}"); + }; + let schema = qe.output_schema().expect("aggregate schema"); let names: Vec<&str> = schema.columns.iter().map(|c| c.name.as_str()).collect(); - assert_eq!(names, vec!["quantile_0_99"]); - assert!( - schema.time_index.is_none(), - "aggregate strips the time axis" + assert_eq!(names, vec!["ts", "value", "env"]); + assert_eq!( + schema.time_index, + Some(0), + "per-series over_time preserves the time axis" ); } @@ -518,6 +612,7 @@ fn scan_schema_carries_ts_value_and_group_keys() { QueryExpr::Scan { .. } => n, QueryExpr::Partition { child, .. } | QueryExpr::Window { child, .. } + | QueryExpr::TimeRange { child, .. } | QueryExpr::Aggregate { child, .. } | QueryExpr::Filter { child, .. } => find_scan(child), other => panic!("unexpected node {other:?}"),