From fff0a2e3c93868a4e43a0a067f7de4e68e212fb6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 6 Jul 2026 09:33:03 -0600 Subject: [PATCH] feat(promql): lower offset / @ time-shift modifiers to a TimeShift node (#40) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `m offset 1h` and `m @ 1609746000` were rejected outright — there was no IR concept for a per-selector time shift. Neither modifier changes a selector's schema; both only move *when* it is evaluated, so they lower to a pass-through `TimeShift` wrapper over the selector's `Scan` (mirroring how the range-vector `[5m]` is a `TimeRange` wrapper rather than a Scan field). - asap-ir: `TimeShift { offset_ms: i64, at: Option }` + `AtModifier::{Start, End, Timestamp(i64)}`, and a `QueryExpr::TimeShift { shift, child }` node whose output schema is the child's (offset/@ never touch columns). Offset is signed ms (a negative offset shifts forward); `@ ` scales PromQL seconds → ms. - L2 `SourceSpec` gains a defaulted `shift` field (constructors set the identity, so no construction churn); the converter's `scan()` lifts a non-identity shift into the `TimeShift` wrapper. - The PromQL front end's `vs_parts` now returns the shift (threaded through `Inner`/`extract_matrix`/`filtered_source`); a ranged selector `m[5m] offset 1h` shifts *under* its `TimeRange`, so the 5m window is taken at the shifted time. This unblocks the common week-over-week / baseline pattern `rate(m[5m]) - rate(m[5m] offset 1w)`. Tests: TimeShift schema pass-through + identity/serde; conformance for offset (signed), `@ `/`start()`/`end()`, offset+@ composition, and the under-TimeRange nesting; exact-tree e2e pins (week-over-week + `@`); the former rejection tests flipped (conformance, equivalence, the info-composition test). Docs: the lowering map gains an offset/@ row. Closes #40 Co-Authored-By: Claude Fable 5 --- crates/e2e/tests/nested.rs | 67 +++++++++- crates/frontend-promql/src/promql.rs | 117 ++++++++++++------ .../awesome_prometheus_alerts.rs | 1 + .../tests/promql_conformance.rs | 84 ++++++++++--- .../tests/promql_equivalence.rs | 28 ++--- .../synthetic_packet_trace.rs | 1 + crates/ir/src/intent_algebra/mod.rs | 7 +- crates/ir/src/intent_algebra/query_expr.rs | 102 ++++++++++++++- crates/l2/src/canonicalize.rs | 1 + crates/l2/src/lower.rs | 12 +- crates/l2/src/relational.rs | 15 ++- 11 files changed, 360 insertions(+), 75 deletions(-) diff --git a/crates/e2e/tests/nested.rs b/crates/e2e/tests/nested.rs index a7856ca1..287c6e71 100644 --- a/crates/e2e/tests/nested.rs +++ b/crates/e2e/tests/nested.rs @@ -10,13 +10,13 @@ use std::time::Duration; +use asap_e2e::fixtures::metric_schema; +use asap_frontend_promql::lower_promql; use asap_ir::intent_algebra::{ - AggIntent, ArithOp, BinaryOpKind, CompareOp, GroupKeys, L3Expr, L3Scalar, Predicate, QueryExpr, - Source, VectorMatch, VectorMatchKind, + AggIntent, ArithOp, AtModifier, BinaryOpKind, CompareOp, GroupKeys, L3Expr, L3Scalar, + Predicate, QueryExpr, Source, TimeShift, VectorMatch, VectorMatchKind, }; use asap_ir::types::AccuracyTarget; -use asap_frontend_promql::lower_promql; -use asap_e2e::fixtures::metric_schema; fn lower(q: &str) -> QueryExpr { lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}")) @@ -277,6 +277,65 @@ fn q39_sum_without_instance_over_rate() { ); } +// #40 — week-over-week: `rate(m[5m]) - rate(m[5m] offset 1w)`. Only the RHS +// selector is time-shifted, so its scan is wrapped in a `TimeShift` under the +// `TimeRange`; the LHS is a bare rate. Both sides bound independently. +#[test] +fn q40_week_over_week_offset() { + let rate_over = |shift: Option| { + let scan = QueryExpr::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: metric_schema(&[]), + }; + let ranged = match shift { + Some(ms) => QueryExpr::TimeShift { + shift: TimeShift { + offset_ms: ms, + at: None, + }, + child: Box::new(scan), + }, + None => scan, + }; + agg( + vec![], + AggIntent::Rate, + QueryExpr::TimeRange { + range: Duration::from_secs(300), + child: Box::new(ranged), + }, + ) + }; + let expected = QueryExpr::BinaryOp { + op: BinaryOpKind::Arith(ArithOp::Sub), + lhs: Box::new(rate_over(None)), + rhs: Box::new(rate_over(Some(604_800_000))), // 1w + vector_match: None, + }; + assert_eq!(lower("rate(m[5m]) - rate(m[5m] offset 1w)"), expected,); +} + +// #40 — `@` anchor: `up @ 1609746000` pins the eval time to an absolute instant +// (seconds → ms); a bare selector wrapped in a `TimeShift` carrying the anchor. +#[test] +fn q40_at_modifier_absolute() { + let expected = QueryExpr::TimeShift { + shift: TimeShift { + offset_ms: 0, + at: Some(AtModifier::Timestamp(1_609_746_000_000)), + }, + child: Box::new(QueryExpr::Scan { + source: Source::TimeSeries { + metric: "up".into(), + }, + predicates: vec![], + schema: metric_schema(&[]), + }), + }; + assert_eq!(lower("up @ 1609746000"), expected); +} + // #24 — sum by job over rate over a filtered scan // same schema [ts, value, job, status]; rate is label-preserving, // so outer sum by job still finds job at col 2 diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 57ff7701..39b036a4 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -44,17 +44,19 @@ //! | `m{f}` | `Filter(Source)` | //! | `a OP b` | `BinaryOp{vector_match}` | //! | `expr[r:res]` | `PromQLSubquery{r, res}` | +//! | ` offset ` / ` @ `/`start()`/`end()` | `TimeShift{shift}` over the selector's `Scan` — pass-through schema; a ranged selector shifts under its `TimeRange` (issue #40) | -use std::time::Duration; +use std::time::{Duration, SystemTime}; use promql_parser::label::{MatchOp, Matcher}; use promql_parser::parser::{ - self, token, AggregateExpr, BinaryExpr, Call, Expr, LabelModifier, VectorMatchCardinality, - VectorSelector, + self, token, AggregateExpr, AtModifier, BinaryExpr, Call, Expr, LabelModifier, Offset, + VectorMatchCardinality, VectorSelector, }; use asap_ir::intent_algebra::query_expr::{ - BinaryOpKind, GroupSide, VectorGrouping, VectorMatch, VectorMatchKind, + AtModifier as L3AtModifier, BinaryOpKind, GroupSide, TimeShift, VectorGrouping, VectorMatch, + VectorMatchKind, }; use asap_l2::relational::{ AggFunc, AggItem, L2SortKey, QueryExpr as L2, SourceSpec, @@ -137,6 +139,8 @@ struct Inner { matchers: Vec, window: Option, func: Option, + /// `offset` / `@` on the selector, carried to the `Source` (issue #40). + shift: TimeShift, } /// Maximum PromQL expression nesting depth the walker accepts. Real queries @@ -233,15 +237,15 @@ fn walk(expr: &Expr) -> Result { input: Box::new(walk(&sq.expr)?), }), Expr::VectorSelector(vs) => { - let (metric, matchers) = vs_parts(vs)?; - Ok(filtered_source(metric, matchers)) + let (metric, matchers, shift) = vs_parts(vs)?; + Ok(filtered_source(metric, matchers, shift)) } Expr::MatrixSelector(ms) => { - let (metric, matchers) = vs_parts(&ms.vs)?; + let (metric, matchers, shift) = vs_parts(&ms.vs)?; Ok(L2::Window { duration: ms.range, slide: None, - input: Box::new(filtered_source(metric, matchers)), + input: Box::new(filtered_source(metric, matchers, shift)), }) } // A number literal is a scalar leaf (`v > 5`, or a bare scalar query @@ -904,14 +908,14 @@ fn histogram_arg_is_sketchable(arg: &Expr) -> bool { fn collect_metric_names(expr: &Expr, out: &mut Vec) { match expr { Expr::VectorSelector(vs) => { - if let Ok((metric, _)) = vs_parts(vs) { + if let Ok((metric, ..)) = vs_parts(vs) { if !metric.is_empty() { out.push(metric); } } } Expr::MatrixSelector(ms) => { - if let Ok((metric, _)) = vs_parts(&ms.vs) { + if let Ok((metric, ..)) = vs_parts(&ms.vs) { if !metric.is_empty() { out.push(metric); } @@ -1008,21 +1012,23 @@ fn walk_binary(bin: &BinaryExpr) -> Result { fn lower_inner(expr: &Expr) -> Result { match expr { Expr::VectorSelector(vs) => { - let (metric, matchers) = vs_parts(vs)?; + let (metric, matchers, shift) = vs_parts(vs)?; Ok(Inner { metric, matchers, window: None, func: None, + shift, }) } Expr::MatrixSelector(ms) => { - let (metric, matchers) = vs_parts(&ms.vs)?; + let (metric, matchers, shift) = vs_parts(&ms.vs)?; Ok(Inner { metric, matchers, window: Some(ms.range), func: None, + shift, }) } Expr::Paren(p) => lower_inner(&p.expr), @@ -1037,41 +1043,45 @@ fn lower_inner(expr: &Expr) -> Result { fn lower_inner_call(call: &Call) -> Result { let name = call.func.name; let at0 = |func: InnerFunc| -> Result { - let (metric, matchers, window) = extract_matrix(arg(call, 0)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 0)?)?; Ok(Inner { metric, matchers, window: Some(window), func: Some(func), + shift, }) }; match name { "rate" | "irate" => { - let (metric, matchers, window) = extract_matrix(arg(call, 0)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 0)?)?; Ok(Inner { metric, matchers, window: Some(window), func: Some(InnerFunc::Rate(window)), + shift, }) } "increase" => { - let (metric, matchers, window) = extract_matrix(arg(call, 0)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 0)?)?; Ok(Inner { metric, matchers, window: Some(window), func: Some(InnerFunc::Increase(window)), + shift, }) } "quantile_over_time" => { let phi = quantile_param(num_arg(call, 0)?)?; - let (metric, matchers, window) = extract_matrix(arg(call, 1)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 1)?)?; Ok(Inner { metric, matchers, window: Some(window), func: Some(InnerFunc::Quantile(phi)), + shift, }) } "avg_over_time" => at0(InnerFunc::Avg), @@ -1100,18 +1110,19 @@ fn lower_inner_call(call: &Call) -> Result { "ts_of_first_over_time" => at0(InnerFunc::TsOfFirstOverTime), "ts_of_last_over_time" => at0(InnerFunc::TsOfLastOverTime), "predict_linear" => { - let (metric, matchers, window) = extract_matrix(arg(call, 0)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 0)?)?; let seconds = num_arg(call, 1)?; Ok(Inner { metric, matchers, window: Some(window), func: Some(InnerFunc::PredictLinear(seconds)), + shift, }) } // `holt_winters` is the legacy spelling of `double_exponential_smoothing`. "double_exponential_smoothing" | "holt_winters" => { - let (metric, matchers, window) = extract_matrix(arg(call, 0)?)?; + let (metric, matchers, window, shift) = extract_matrix(arg(call, 0)?)?; let smoothing = num_arg(call, 1)?; let trend = num_arg(call, 2)?; Ok(Inner { @@ -1119,6 +1130,7 @@ fn lower_inner_call(call: &Call) -> Result { matchers, window: Some(window), func: Some(InnerFunc::DoubleExp { smoothing, trend }), + shift, }) } other => Err(LoweringError::UnsupportedFunction(other.to_string())), @@ -1130,7 +1142,7 @@ fn lower_inner_call(call: &Call) -> Result { fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { match outer { Outer::None => match &inner.func { - None => Ok(filtered_source(inner.metric, inner.matchers)), + None => Ok(filtered_source(inner.metric, inner.matchers, inner.shift)), Some(f) => { let func = inner_func(f); Ok(windowed_aggregate(inner, keys, func)) @@ -1175,7 +1187,7 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { // wrapped in a reducing aggregate (issue #86). let base = match inner.func.as_ref().map(inner_func) { Some(func) => windowed_aggregate(inner, vec![], func), - None => filtered_source(inner.metric, inner.matchers), + None => filtered_source(inner.metric, inner.matchers, inner.shift), }; Ok(L2::Sample { keys, @@ -1221,7 +1233,7 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { // `Sort.partition_by` can rank within each group (issue #12). let base = match inner.func.as_ref().map(inner_func) { Some(func) => windowed_aggregate(inner, vec![], func), - None => filtered_source(inner.metric, inner.matchers), + None => filtered_source(inner.metric, inner.matchers, inner.shift), }; let sorted = L2::Sort { keys: vec![L2SortKey { @@ -1247,7 +1259,7 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result { fn windowed_aggregate(inner: Inner, keys: Vec, func: AggFunc) -> L2 { let skip_window = matches!(func, AggFunc::Rate { .. } | AggFunc::Increase { .. }); let window = inner.window; - let base = filtered_source(inner.metric, inner.matchers); + let base = filtered_source(inner.metric, inner.matchers, inner.shift); let input = match window { Some(w) if !skip_window => L2::Window { duration: w, @@ -1290,8 +1302,8 @@ fn outer_aggregate(keys: Vec, func: AggFunc, input: L2) -> L2 { } } -fn filtered_source(metric: String, matchers: Vec) -> L2 { - let source = L2::Source(SourceSpec::new(metric)); +fn filtered_source(metric: String, matchers: Vec, shift: TimeShift) -> L2 { + let source = L2::Source(SourceSpec::new(metric).with_shift(shift)); if matchers.is_empty() { source } else { @@ -1405,15 +1417,7 @@ fn resolve_group(agg: &AggregateExpr) -> Result<(Vec, bool)> { // ── Free helpers ────────────────────────────────────────────────────────────── -fn vs_parts(vs: &VectorSelector) -> Result<(String, Vec)> { - // `offset` / `@` shift the evaluation/lookback time. The intent algebra has - // no representation for either, so silently lowering them (as if absent) - // would change the query's meaning. Reject rather than mislower. - if vs.offset.is_some() || vs.at.is_some() { - return Err(LoweringError::UnsupportedFeature( - "`offset` / `@` time-shift modifiers have no intent-algebra representation".into(), - )); - } +fn vs_parts(vs: &VectorSelector) -> Result<(String, Vec, TimeShift)> { // A non-equality `__name__` matcher (`=~` / `!~` / `!=`) selects *across* // metric names. The L3 `Source::TimeSeries { metric }` carries a single // concrete metric name, so there is no representation for a regex/negated @@ -1451,7 +1455,46 @@ fn vs_parts(vs: &VectorSelector) -> Result<(String, Vec)> { .collect(); ms.sort_by(|a, b| a.name.cmp(&b.name).then_with(|| a.value.cmp(&b.value))); let matchers = ms.into_iter().map(matcher_to_l3expr).collect(); - Ok((metric, matchers)) + let shift = time_shift(vs.offset.as_ref(), vs.at.as_ref())?; + Ok((metric, matchers, shift)) +} + +/// Convert the parser's `offset` / `@` modifiers into a [`TimeShift`] (issue +/// #40). Offset is signed milliseconds; `@ ` (parser seconds → ms) becomes +/// an absolute anchor, `@ start()`/`@ end()` the range bounds. +fn time_shift(offset: Option<&Offset>, at: Option<&AtModifier>) -> Result { + let offset_ms = match offset { + None => 0, + Some(Offset::Pos(d)) => duration_ms(*d)?, + Some(Offset::Neg(d)) => -duration_ms(*d)?, + }; + let at = match at { + None => None, + Some(AtModifier::Start) => Some(L3AtModifier::Start), + Some(AtModifier::End) => Some(L3AtModifier::End), + Some(AtModifier::At(t)) => Some(L3AtModifier::Timestamp(system_time_ms(*t)?)), + }; + Ok(TimeShift { offset_ms, at }) +} + +/// A `Duration` as `i64` milliseconds, rejecting an overflow rather than +/// silently truncating a pathologically large `offset`. +fn duration_ms(d: Duration) -> Result { + i64::try_from(d.as_millis()).map_err(|_| { + LoweringError::InvalidParameter("offset duration overflows i64 milliseconds".into()) + }) +} + +/// A `SystemTime` (`@ `) as `i64` milliseconds since the Unix epoch, signed +/// so pre-epoch anchors (the parser permits them) are preserved. +fn system_time_ms(t: SystemTime) -> Result { + let ms = match t.duration_since(std::time::UNIX_EPOCH) { + Ok(d) => i64::try_from(d.as_millis()), + Err(e) => i64::try_from(e.duration().as_millis()).map(|ms| -ms), + }; + ms.map_err(|_| { + LoweringError::InvalidParameter("`@` timestamp overflows i64 milliseconds".into()) + }) } fn matcher_to_l3expr(m: &Matcher) -> L2Expr { @@ -1468,11 +1511,11 @@ fn matcher_to_l3expr(m: &Matcher) -> L2Expr { } } -fn extract_matrix(expr: &Expr) -> Result<(String, Vec, Duration)> { +fn extract_matrix(expr: &Expr) -> Result<(String, Vec, Duration, TimeShift)> { match expr { Expr::MatrixSelector(ms) => { - let (metric, matchers) = vs_parts(&ms.vs)?; - Ok((metric, matchers, ms.range)) + let (metric, matchers, shift) = vs_parts(&ms.vs)?; + Ok((metric, matchers, ms.range, shift)) } Expr::Paren(p) => extract_matrix(&p.expr), // A range-vector function argument must be a (parenthesised) matrix diff --git a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs index ea8f4453..f2a9319b 100644 --- a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs +++ b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs @@ -57,6 +57,7 @@ fn intents(e: &QueryExpr) -> Vec { } QueryExpr::Window { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs index e3db2e79..b5b77fb8 100644 --- a/crates/frontend-promql/tests/promql_conformance.rs +++ b/crates/frontend-promql/tests/promql_conformance.rs @@ -35,7 +35,8 @@ use std::time::Duration; use asap_ir::intent_algebra::schema::DataType; use asap_ir::intent_algebra::{ - AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, MathFunc, QueryExpr, SampleKind, Source, TimeFunc, + AggIntent, ArithOp, AtModifier, BinaryOpKind, CompareOp, L3Expr, MathFunc, QueryExpr, + SampleKind, Source, TimeFunc, }; use asap_ir::types::AccuracyTarget; use asap_frontend_promql::{lower_promql, PromqlError as LoweringError}; @@ -71,6 +72,7 @@ fn collect(e: &QueryExpr, out: &mut Vec) { } QueryExpr::Window { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } @@ -116,6 +118,7 @@ fn first_scan(e: &QueryExpr) -> (String, usize) { } QueryExpr::Window { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Aggregate { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } @@ -143,6 +146,7 @@ fn negates_via_scalar(e: &QueryExpr) -> bool { | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Subquery { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Project { child, .. } => negates_via_scalar(child), @@ -987,18 +991,68 @@ fn nested_subquery_from_prometheus_docs() { // ───────────────────────────────────────────────────────────────────────────── #[test] -fn offset_modifier_is_rejected() { - // SEMANTICS (PromQL): `offset 5m` shifts the lookback 5m into the past. The - // intent algebra can't represent it, so we reject rather than silently drop - // it (which would change the query's meaning). - let _ = rejected("http_requests_total offset 5m"); +fn offset_modifier_lowers_to_a_time_shift() { + // SEMANTICS (PromQL, issue #40): `offset 5m` shifts the lookback 5m into the + // past — a `TimeShift` wrapper over the selector (signed ms; a negative + // offset shifts forward). Schema is unchanged (the shift only moves *when*). + let qe = ok("http_requests_total offset 5m"); + let QueryExpr::TimeShift { shift, child } = &qe else { + panic!("expected a TimeShift, got {qe:?}"); + }; + assert_eq!(shift.offset_ms, 300_000); + assert!(shift.at.is_none()); + assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); + + // `offset -5m` shifts forward → negative ms. + let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total offset -5m") else { + panic!("expected a TimeShift"); + }; + assert_eq!(shift.offset_ms, -300_000); +} + +#[test] +fn at_modifier_lowers_to_a_time_shift() { + // SEMANTICS (PromQL, issue #40): `@ ` pins the evaluation to an absolute + // instant (PromQL seconds → IR milliseconds); `@ start()` / `@ end()` anchor + // to the query range bounds. + let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total @ 1609746000") else { + panic!("expected a TimeShift for `@ `"); + }; + assert_eq!(shift.at, Some(AtModifier::Timestamp(1_609_746_000_000))); + assert_eq!(shift.offset_ms, 0); + + let QueryExpr::TimeShift { shift, .. } = &ok("http_requests_total @ start()") else { + panic!("expected a TimeShift for `@ start()`"); + }; + assert_eq!(shift.at, Some(AtModifier::Start)); + + // Offset and `@` compose: `@ end() offset 5m` carries both. + let qe = ok("http_requests_total @ end() offset 5m"); + let QueryExpr::TimeShift { shift, .. } = &qe else { + panic!("expected a TimeShift, got {qe:?}"); + }; + assert_eq!(shift.at, Some(AtModifier::End)); + assert_eq!(shift.offset_ms, 300_000); } #[test] -fn at_modifier_is_rejected() { - // SEMANTICS (PromQL): `@ ` pins the evaluation time. Rejected for the - // same reason as `offset`. - let _ = rejected("http_requests_total @ 1609746000"); +fn offset_on_a_ranged_selector_wraps_inside_the_time_range() { + // `rate(m[5m] offset 1h)` — the offset is on the ranged selector, so the + // `TimeShift` sits *under* the `TimeRange` (the 5m window is taken at the + // shifted time), and the whole thing under the per-series `Rate` (#40). + let qe = ok("rate(http_requests_total[5m] offset 1h)"); + let QueryExpr::Aggregate { aggs, child, .. } = &qe else { + panic!("expected the rate Aggregate, got {qe:?}"); + }; + assert!(matches!(aggs.as_slice(), [AggIntent::Rate])); + let QueryExpr::TimeRange { child, .. } = child.as_ref() else { + panic!("expected a TimeRange under rate, got {child:?}"); + }; + let QueryExpr::TimeShift { shift, child } = child.as_ref() else { + panic!("expected a TimeShift under the TimeRange, got {child:?}"); + }; + assert_eq!(shift.offset_ms, 3_600_000); + assert!(matches!(child.as_ref(), QueryExpr::Scan { .. })); } // ───────────────────────────────────────────────────────────────────────────── @@ -1542,15 +1596,16 @@ fn info_selector_carries_the_info_side_matchers() { } #[test] -fn info_composes_under_an_aggregation_and_rejects_time_shift() { +fn info_composes_under_an_aggregation_and_over_a_time_shift() { // `sum(info(m))` — enrichment first, then a cross-series sum over it. assert!(has(&ok("sum(info(node_uname_info))"), |i| matches!( i, AggIntent::Sum { .. } ))); - // `offset` / `@` on the input still have no representation → rejected. - let _ = rejected("info(metric @ 60)"); - let _ = rejected("info(metric offset 1m)"); + // `offset` / `@` on the input now lower to a `TimeShift` under the info-join + // (issue #40) — the enrichment composes over the shifted selector. + assert!(matches!(ok("info(metric @ 60)"), QueryExpr::InfoJoin { .. })); + assert!(matches!(ok("info(metric offset 1m)"), QueryExpr::InfoJoin { .. })); } // ───────────────────────────────────────────────────────────────────────────── @@ -1700,6 +1755,7 @@ fn first_relabel(e: &QueryExpr) -> &QueryExpr { QueryExpr::Aggregate { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Window { child, .. } => first_relabel(child), other => panic!("no Relabel reachable from {other:?}"), } diff --git a/crates/frontend-promql/tests/promql_equivalence.rs b/crates/frontend-promql/tests/promql_equivalence.rs index 9702c3e2..7f4dc9c8 100644 --- a/crates/frontend-promql/tests/promql_equivalence.rs +++ b/crates/frontend-promql/tests/promql_equivalence.rs @@ -47,15 +47,6 @@ fn assert_distinct(a: &str, b: &str) { ); } -/// A query whose semantics we can't faithfully represent must be rejected -/// (never silently mislowered into a different meaning). -fn assert_rejected(q: &str) { - assert!( - lower_promql(q, AccuracyTarget::Exact).is_err(), - "{q:?} should be rejected, not silently lowered" - ); -} - // ───────────────────────────────────────────────────────────────────────────── // 1. Equivalence classes the lowering canonicalises to one L3. // ───────────────────────────────────────────────────────────────────────────── @@ -182,9 +173,18 @@ fn group_is_not_sum() { #[test] fn offset_and_at_are_not_dropped() { - // Time-shift modifiers change the query's meaning; they previously lowered - // identically to the un-shifted query (silent loss). - assert_rejected("http_requests_total offset 5m"); - assert_rejected("http_requests_total @ 1609746000"); - assert_rejected("rate(http_requests_total[5m] offset 1h)"); + // Time-shift modifiers change the query's meaning. They now lower to a + // `TimeShift` wrapper (issue #40) — the point is they stay DISTINCT from the + // un-shifted query rather than collapsing onto it (the former silent loss). + assert_distinct("http_requests_total offset 5m", "http_requests_total"); + assert_distinct("http_requests_total @ 1609746000", "http_requests_total"); + assert_distinct( + "rate(http_requests_total[5m] offset 1h)", + "rate(http_requests_total[5m])", + ); + // Different shifts are also distinct from each other. + assert_distinct( + "http_requests_total offset 5m", + "http_requests_total offset 10m", + ); } diff --git a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs index 2aea1c84..69207bee 100644 --- a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs +++ b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs @@ -79,6 +79,7 @@ fn intents(e: &QueryExpr) -> Vec { } QueryExpr::Window { child, .. } | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } | QueryExpr::Filter { child, .. } | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } diff --git a/crates/ir/src/intent_algebra/mod.rs b/crates/ir/src/intent_algebra/mod.rs index 1bdb8557..522c47d2 100644 --- a/crates/ir/src/intent_algebra/mod.rs +++ b/crates/ir/src/intent_algebra/mod.rs @@ -27,8 +27,9 @@ pub use agg_intent::{ pub use expr_ir::{ArithOp, ColumnRef, CompareOp, Expr, L2Expr, L3Expr, L3Scalar}; pub use names::{BindingName, QueryId}; pub use query_expr::{ - aggregate_output_schema, BinaryOpKind, BindingScope, DataModel, GroupKeys, GroupSide, - InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError, SampleKind, SetOpKind, - SortKey, Source, VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind, WindowKind, + aggregate_output_schema, AtModifier, BinaryOpKind, BindingScope, DataModel, GroupKeys, + GroupSide, InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError, SampleKind, + SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch, VectorMatchKind, + WindowFuncKind, WindowKind, }; pub use schema::{cse_reuse_is_legal, Column, ColumnId, CseError, DataType, Schema}; diff --git a/crates/ir/src/intent_algebra/query_expr.rs b/crates/ir/src/intent_algebra/query_expr.rs index c0c638fe..7d3011d7 100644 --- a/crates/ir/src/intent_algebra/query_expr.rs +++ b/crates/ir/src/intent_algebra/query_expr.rs @@ -307,6 +307,42 @@ pub struct VectorMatch { pub grouping: Option, } +/// PromQL `@` modifier — pins a selector's evaluation time to an anchor instead +/// of the query evaluation time (issue #40). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum AtModifier { + /// `@ start()` — the query range's start instant. + Start, + /// `@ end()` — the query range's end instant. + End, + /// `@ ` — an absolute instant, milliseconds since the Unix epoch (may be + /// negative). PromQL writes the timestamp in seconds; the front end scales it. + Timestamp(i64), +} + +/// PromQL per-selector **time-shift** modifiers — `offset` and `@` (issue #40). +/// Neither changes a selector's *schema*; both move *when* it is evaluated, so +/// the shift is a pass-through wrapper ([`QueryExpr::TimeShift`]) over the +/// selector rather than a new leaf shape. The runtime resolves the anchor and +/// applies the offset. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct TimeShift { + /// `offset ` as signed milliseconds — a positive value shifts the + /// lookback *back* in time (`offset 5m`), a negative value shifts it + /// *forward* (`offset -5m`). `0` = no offset. + pub offset_ms: i64, + /// `@` anchor; `None` = evaluate at the query time. + pub at: Option, +} + +impl TimeShift { + /// Whether this shift is the identity (no `offset`, no `@`) — the state of + /// every selector that carries neither modifier. + pub fn is_identity(&self) -> bool { + self.offset_ms == 0 && self.at.is_none() + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum VectorMatchKind { On, @@ -535,6 +571,19 @@ pub enum QueryExpr { child: Box, }, + /// PromQL `offset` / `@` **time shift** on a selector (issue #40). A + /// pass-through wrapper: it moves *when* `child` is evaluated (the runtime + /// resolves the `@` anchor and applies the offset) but leaves its schema + /// unchanged. Wraps the shifted selector directly — `m offset 1h` → + /// `TimeShift { Scan }`; a ranged selector `m[5m] offset 1h` → + /// `TimeRange { 5m, TimeShift { Scan } }` (the range is taken at the shifted + /// time). Never carries the identity shift (the converter emits a bare + /// selector when neither modifier is present). + TimeShift { + shift: TimeShift, + 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. @@ -631,7 +680,10 @@ impl QueryExpr { // Info enrichment adds runtime info labels — the statically-known // schema is the child's (open), so it passes through (#84). | QueryExpr::InfoJoin { child, .. } - | QueryExpr::TimeRange { child, .. } => child.output_schema_in(scope), + | QueryExpr::TimeRange { child, .. } + // A time shift (`offset`/`@`) moves *when* the child is evaluated, + // never its columns — schema passes through (#40). + | QueryExpr::TimeShift { child, .. } => child.output_schema_in(scope), // ρ — relabel preserves every input column and writes one label // `dst` (Utf8): overwritten in place if it already exists, else @@ -1238,6 +1290,54 @@ mod tests { assert!(s.unique_keys.is_empty(), "kept set unknown → no unique key"); } + #[test] + fn time_shift_is_schema_pass_through() { + // `offset`/`@` move *when* a selector is evaluated, never its columns — + // a `TimeShift` output schema equals its child's (issue #40). + let scan_node = scan( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("job", DataType::Utf8, true), + ], + Some(0), + vec![], + ); + let shifted = QueryExpr::TimeShift { + shift: TimeShift { + offset_ms: 3_600_000, + at: Some(AtModifier::Timestamp(1_609_746_000_000)), + }, + child: Box::new(scan_node.clone()), + }; + assert_eq!( + shifted.output_schema().unwrap(), + scan_node.output_schema().unwrap(), + ); + } + + #[test] + fn time_shift_identity_and_serde() { + let offset_only = TimeShift { + offset_ms: 1, + at: None, + }; + let at_only = TimeShift { + offset_ms: 0, + at: Some(AtModifier::End), + }; + assert!(TimeShift::default().is_identity()); + assert!(!offset_only.is_identity()); + assert!(!at_only.is_identity()); + // Round-trip the shift + anchor. + let s = TimeShift { + offset_ms: -300_000, + at: Some(AtModifier::Timestamp(60_000)), + }; + let back: TimeShift = serde_json::from_str(&serde_json::to_string(&s).unwrap()).unwrap(); + assert_eq!(back, s); + } + #[test] fn per_series_rate_preserves_labels() { // A per-series range reduction (`rate`) is label-preserving: it produces diff --git a/crates/l2/src/canonicalize.rs b/crates/l2/src/canonicalize.rs index 985765df..20b9f9a5 100644 --- a/crates/l2/src/canonicalize.rs +++ b/crates/l2/src/canonicalize.rs @@ -69,6 +69,7 @@ fn children_mut(expr: &mut QueryExpr) -> Vec<&mut QueryExpr> { | Distinct { child, .. } | Subquery { child, .. } | TimeRange { child, .. } + | TimeShift { child, .. } | WindowFunc { child, .. } | Sample { child, .. } | InfoJoin { child, .. } diff --git a/crates/l2/src/lower.rs b/crates/l2/src/lower.rs index 506d09a6..951f7c6d 100644 --- a/crates/l2/src/lower.rs +++ b/crates/l2/src/lower.rs @@ -604,10 +604,20 @@ fn scan( .iter() .map(|e| -> Result { Ok(Predicate(resolve_expr(e, &schema)?)) }) .collect::, _>>()?; - Ok(CQueryExpr::Scan { + let scan = CQueryExpr::Scan { source, predicates, schema, + }; + // A non-identity `offset`/`@` lifts into a `TimeShift` wrapper over the scan; + // an unshifted selector stays a bare `Scan` (issue #40). + Ok(if spec.shift.is_identity() { + scan + } else { + CQueryExpr::TimeShift { + shift: spec.shift, + child: Box::new(scan), + } }) } diff --git a/crates/l2/src/relational.rs b/crates/l2/src/relational.rs index 1acd3a7a..9730dfe3 100644 --- a/crates/l2/src/relational.rs +++ b/crates/l2/src/relational.rs @@ -11,7 +11,7 @@ use std::time::Duration; pub use asap_ir::intent_algebra::expr_ir::{ColumnRef, L2Expr}; use asap_ir::intent_algebra::agg_intent::{MathFunc, TimeFunc}; -use asap_ir::intent_algebra::query_expr::{InfoMatcher, SampleKind}; +use asap_ir::intent_algebra::query_expr::{InfoMatcher, SampleKind, TimeShift}; pub use asap_ir::intent_algebra::query_expr::{BinaryOpKind, VectorMatch, WindowFuncKind}; use asap_ir::intent_algebra::schema::Schema; @@ -42,6 +42,11 @@ pub struct SourceSpec { /// labels). The presence of a schema also selects the L3 `Source` variant: /// `Some` → `Source::Table`, `None` → `Source::TimeSeries`. pub schema: Option, + /// PromQL `offset` / `@` time shift on this selector (issue #40). The + /// converter lifts a non-identity shift into an L3 [`TimeShift`] wrapper over + /// the `Scan`. `TimeShift::default()` (the identity) for every unshifted + /// selector and every SQL table. + pub shift: TimeShift, } impl SourceSpec { @@ -50,6 +55,7 @@ impl SourceSpec { Self { name: name.into(), schema: None, + shift: TimeShift::default(), } } @@ -58,8 +64,15 @@ impl SourceSpec { Self { name: name.into(), schema: Some(schema), + shift: TimeShift::default(), } } + + /// This PromQL leaf with a time-shift modifier attached (issue #40). + pub fn with_shift(mut self, shift: TimeShift) -> Self { + self.shift = shift; + self + } } /// One aggregate function in a GROUP BY / AGGREGATE node.