diff --git a/src/query/src/promql/native_compile.rs b/src/query/src/promql/native_compile.rs index 89fbe95d611..831f10eab6d 100644 --- a/src/query/src/promql/native_compile.rs +++ b/src/query/src/promql/native_compile.rs @@ -16,20 +16,24 @@ //! //! This is the front half of the native evaluation model: it reads the parsed statement and writes //! down what it means as bound nodes. It runs nothing, and nothing calls it yet, so the subset it -//! accepts is deliberately one shape: +//! accepts is deliberately two shapes: //! //! ```text +//! e.g. x //! ([]) e.g. rate(x[5m]) //! ``` //! -//! which becomes `Input` (raw samples) → `Window` (the range function over the statement's grid). -//! Everything else is a [`DataFusionError::NotImplemented`] that names what was found and quotes -//! the shape above, because rejecting a query the compiler does not understand is the only honest -//! alternative to compiling it into something else. +//! which become `Input` (raw samples) → `Instant` (the last sample within the statement's lookback) +//! and `Input` (raw samples) → `Window` (the range function over the statement's grid). A selector +//! may carry `offset` in either shape and, in the window shape, an `@` anchor. Everything else is a +//! [`DataFusionError::NotImplemented`] that names what was found and quotes the shapes above, +//! because rejecting a query the compiler does not understand is the only honest alternative to +//! compiling it into something else. //! -//! The statement's grid and the selector's window are the only numbers derived here; the legality -//! of the tree is not checked here. [`NativeTree::try_new`] is the single authority on that, so a -//! node added later cannot be legal in one place and illegal in another. +//! The statement's grid, the selector's lookback, window, offset and anchor are the only numbers +//! derived here; the legality of the tree is not checked here. [`NativeTree::try_new`] is the +//! single authority on that, so a node added later cannot be legal in one place and illegal in +//! another. use std::sync::Arc; @@ -38,17 +42,21 @@ use datatypes::series::SeriesOrdering; use datatypes::series::samples::SeriesSchema; use datatypes::series::time::{EvalGrid, GridSpec}; use promql::native::plan::{ - BoundInput, BoundWindow, NativeExpr, NativeNode, NativeTree, NativeType, + BoundInput, BoundInstant, BoundWindow, NativeExpr, NativeNode, NativeTree, NativeType, }; use promql::range_eval::{RangeFunction, RangeWindow}; -use promql_parser::parser::{Call, EvalStmt, Expr as PromExpr, MatrixSelector, Offset}; +use promql_parser::parser::{ + AtModifier, Call, EvalStmt, Expr as PromExpr, MatrixSelector, Offset, VectorSelector, +}; use crate::promql::label_values::signed_millis_since_epoch; -/// The one shape this compiler accepts, quoted in every rejection. -const SUPPORTED: &str = "`([])`, for example `rate(x[5m])`"; +/// The shapes this compiler accepts, quoted in every rejection. +const SUPPORTED: &str = "a bare selector or `([])`, for example \ + `x` or `rate(x[5m])`"; -/// Compiles `stmt` into a tree that reads `input` and folds it with the statement's one window. +/// Compiles `stmt` into a tree that reads `input` and either selects from it or folds it with the +/// statement's one window. /// /// `input` is the raw series schema the caller bound from the table, and the caller has to apply the /// selector's matchers before binding it: `__field__`, `__schema__` and `__database__` select rows @@ -75,6 +83,10 @@ pub fn compile( input: Arc, ordering: SeriesOrdering, ) -> Result { + // A bare selector is the other shape: an instant selection over the same raw input. + if let PromExpr::VectorSelector(selector) = unparen(&stmt.expr) { + return instant_selection(stmt, selector, input, ordering); + } let PromExpr::Call(Call { func, args }) = unparen(&stmt.expr) else { return Err(not_implemented(&format!("the expression `{}`", stmt.expr))); }; @@ -107,18 +119,11 @@ pub fn compile( series may not interleave", )); } - // An `@`-anchored selector fixes its window instead of following the evaluation grid, which the - // native model binds as an absolute millisecond anchor. Resolving `@` into that value is not - // part of this subset, so the anchor is refused rather than dropped. - if vs.at.is_some() { - return Err(not_implemented(&format!( - "a call to `{}` over an `@`-anchored selector, whose anchor is not resolved into an \ - absolute millisecond value yet and so cannot be bound", - func.name - ))); - } + // An `@` anchor is step-invariant: every step folds the same samples, which the window's + // `anchor_ms` tells `RangeEvaluator` to replay over the grid. let range_ms = range_millis(range.as_millis())?; let offset_ms = selector_offset_ms(&vs.offset)?; + let anchor_ms = selector_anchor_ms(&vs.at, stmt)?; // The grid is the last thing built: a query refused above, or a window that cannot exist, // never pays for a grid nothing would read. let output_grid = statement_grid(stmt)?; @@ -148,8 +153,7 @@ pub fn compile( window: RangeWindow { range_ms, offset_ms, - // Only `@` sets an anchor, and an anchored selector is rejected above. - anchor_ms: None, + anchor_ms, }, // The input is raw samples, so it has no grid of its own. input_grid: None, @@ -160,6 +164,58 @@ pub fn compile( }) } +/// Compiles a bare selector into an instant selection over the raw input. +/// +/// The lookback, the offset and the grid are the whole of an instant selection: it reports the last +/// sample at or before each grid point, within the statement's lookback. +fn instant_selection( + stmt: &EvalStmt, + selector: &VectorSelector, + input: Arc, + ordering: SeriesOrdering, +) -> Result { + // An `@` anchor is bound on a range window, which carries `anchor_ms`; an instant selection has + // no such field, so the anchor has nowhere to go. Refusing it is the only alternative to + // dropping it, which would silently select around every grid point instead of one anchor. + if selector.at.is_some() { + return Err(not_implemented(&format!( + "the `@`-anchored selector `{selector}`, whose anchor the instant selection node has \ + no field for: an instant selection binds a lookback, an offset and a grid, while \ + anchoring is bound on a range window, as in `rate(x[5m] @ 1234)`" + ))); + } + let lookback_ms = lookback_millis(stmt.lookback_delta.as_millis())?; + let offset_ms = selector_offset_ms(&selector.offset)?; + let output_grid = statement_grid(stmt)?; + + // An instant selection reports one value per input field, so the layout is the input's. + let output = NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::clone(&input.fields), + eval_grid: Some(Arc::clone(&output_grid)), + })); + + // One allocation shared by the binding and the node's output, as in the window shape. + let raw = NativeType::Samples(input); + + NativeTree::try_new(NativeNode { + expr: NativeExpr::Instant { + input: Box::new(NativeNode { + expr: NativeExpr::Input(BoundInput { + output: raw.clone(), + ordering, + }), + output: raw, + }), + op: BoundInstant { + lookback_ms, + offset_ms, + output_grid, + }, + }, + output, + }) +} + /// The grid the statement is evaluated on: `(end - start) / step + 1` points, the last of which /// does not pass `end`. fn statement_grid(stmt: &EvalStmt) -> Result> { @@ -199,6 +255,37 @@ fn range_millis(range: u128) -> Result { Ok(range_ms) } +/// The statement's lookback in milliseconds, which is how far back an instant selection reads. +fn lookback_millis(lookback: u128) -> Result { + millis(lookback).ok_or_else(|| { + DataFusionError::Plan(format!( + "the statement's lookback of {lookback}ms is out of range" + )) + }) +} + +/// The selector's `@` anchor in milliseconds since the Unix epoch, or `None` when it has none. +/// +/// `start()` and `end()` stay symbolic in the AST, because only the statement knows which evaluation +/// range they name, so they are resolved against `stmt.start` and `stmt.end` here. The offset is not +/// folded in: [`RangeWindow`] keeps anchor and offset apart and ends the window at +/// `anchor_ms - offset_ms` at run time. +/// +/// An anchor decouples the window's read range from the statement's: it reads +/// `(anchor_ms - offset_ms - range_ms, anchor_ms - offset_ms]`, which for an anchor like `@ 1234` can +/// lie entirely outside `[stmt.start - range_ms, stmt.end]`. Whoever binds the scan therefore has to +/// derive its range from the window's anchor, offset and range rather than from `stmt.start` and +/// `stmt.end`; a scan bound to the statement alone leaves such a window with no samples, and the +/// query silently returns empty instead of failing. +fn selector_anchor_ms(at: &Option, stmt: &EvalStmt) -> Result> { + match at { + None => Ok(None), + Some(AtModifier::Start) => Ok(Some(signed_millis_since_epoch(stmt.start)?)), + Some(AtModifier::End) => Ok(Some(signed_millis_since_epoch(stmt.end)?)), + Some(AtModifier::At(time)) => Ok(Some(signed_millis_since_epoch(*time)?)), + } +} + /// The selector's offset in milliseconds; positive reads earlier samples. fn selector_offset_ms(offset: &Option) -> Result { let Some(offset) = offset else { diff --git a/src/query/src/promql/native_compile/tests.rs b/src/query/src/promql/native_compile/tests.rs index c0917236f41..1918ce3ec2e 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/src/query/src/promql/native_compile/tests.rs @@ -18,7 +18,9 @@ use std::time::{Duration, UNIX_EPOCH}; use datafusion::error::DataFusionError; use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; -use promql::native::plan::{BoundWindow, NativeExpr, NativeNode, NativeTree, NativeType}; +use promql::native::plan::{ + BoundInstant, BoundWindow, NativeExpr, NativeNode, NativeTree, NativeType, +}; use promql::range_eval::RangeFunction; use promql_parser::parser::{EvalStmt, Expr as PromExpr}; @@ -36,10 +38,18 @@ fn raw_schema() -> Arc { /// A statement over 100 seconds in 5-second steps, which is 21 grid points. fn eval_stmt(query: &str) -> EvalStmt { + eval_stmt_over(query, Duration::from_secs(100)) +} + +/// The statement `query` over `UNIX_EPOCH..=end`, on a 5s grid. +/// +/// The end is a parameter because a grid stops before an end it does not land on, which is the only +/// way `end()` and the last grid point can be told apart. +fn eval_stmt_over(query: &str, end: Duration) -> EvalStmt { EvalStmt { expr: promql_parser::parser::parse(query).unwrap(), start: UNIX_EPOCH, - end: UNIX_EPOCH + Duration::from_secs(100), + end: UNIX_EPOCH + end, interval: Duration::from_secs(5), lookback_delta: Duration::from_secs(5), } @@ -59,7 +69,7 @@ fn with_selector_range(range: Duration) -> EvalStmt { stmt } -/// The window at the root of `tree`, which is the only shape `compile` produces. +/// The window at the root of `tree`, for the range-function shape. fn root_window(tree: &NativeTree) -> (&NativeNode, &BoundWindow) { let NativeExpr::Window { input, op } = &tree.root().expr else { panic!("the root is not a window: {:?}", tree.root().expr); @@ -67,6 +77,17 @@ fn root_window(tree: &NativeTree) -> (&NativeNode, &BoundWindow) { (input, op) } +/// The instant selection at the root of `tree`, for the bare-selector shape. +fn root_instant(tree: &NativeTree) -> (&NativeNode, &BoundInstant) { + let NativeExpr::Instant { input, op } = &tree.root().expr else { + panic!( + "the root is not an instant selection: {:?}", + tree.root().expr + ); + }; + (input, op) +} + /// `rate(x[5m])` becomes a window over the raw input, on the statement's grid. #[test] fn a_rate_over_a_selector_becomes_a_window_over_raw_input() { @@ -148,6 +169,173 @@ fn the_supported_range_functions_map_to_their_kernels() { } } +/// `x` becomes an instant selection over the raw input, on the statement's grid, with the +/// statement's lookback and the input's fields. +#[test] +fn a_bare_selector_becomes_an_instant_over_raw_input() { + let input = raw_schema(); + // A lookback the statement chose, so the binding cannot be a default the compiler invented. + let stmt = EvalStmt { + lookback_delta: Duration::from_secs(120), + ..eval_stmt("x") + }; + let tree = compile(&stmt, Arc::clone(&input), SeriesOrdering::KeyThenTime).unwrap(); + + let (input_node, op) = root_instant(&tree); + assert_eq!(op.lookback_ms, 120_000); + assert_eq!(op.offset_ms, 0); + + // The statement's grid: 0, 5s, ..., 100s. + let spec = op.output_grid.spec(); + assert_eq!( + (spec.start_ms, spec.step_ms, spec.len), + (0, 5_000, 21), + "the instant selection does not land on the statement's grid" + ); + + // The output is the input's fields, evaluated on the statement's grid. + let output = tree + .root() + .output + .samples() + .expect("an instant selection produces samples"); + assert!( + Arc::ptr_eq(&output.fields, &input.fields), + "the instant selection did not keep the input's fields" + ); + let output_grid = output + .eval_grid + .as_ref() + .expect("an instant selection produces evaluated values"); + assert!( + Arc::ptr_eq(output_grid, &op.output_grid), + "the output grid is not the bound grid" + ); + + let NativeExpr::Input(bound) = &input_node.expr else { + panic!( + "the instant selection's input is not an input: {:?}", + input_node.expr + ); + }; + assert!( + matches!(&bound.output, NativeType::Samples(schema) if schema.eval_grid.is_none()), + "the input is not raw samples: {:?}", + bound.output + ); + assert_eq!(bound.ordering, SeriesOrdering::KeyThenTime); +} + +/// A bare selector's offset reaches the instant binding with its sign. +#[test] +fn an_offset_on_a_bare_selector_keeps_its_sign() { + let cases = [("x offset 1m", 60_000), ("x offset -1m", -60_000)]; + + for (query, offset_ms) in cases { + let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime).unwrap(); + let (_, op) = root_instant(&tree); + assert_eq!(op.offset_ms, offset_ms, "`{query}` lost the offset's sign"); + assert_eq!(op.lookback_ms, 5_000, "`{query}` lost the lookback"); + } +} + +/// An `@`-anchored bare selector is refused because the instant node has no anchor field, not +/// because anchoring is unsupported: the window shape binds an anchor. +#[test] +fn an_anchored_instant_selector_is_refused_for_the_missing_anchor_field() { + let error = compile( + &eval_stmt("x @ 100"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .expect_err("`x @ 100` compiled, and it should not have"); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "`x @ 100` did not report NotImplemented: {error}" + ); + let message = error.to_string(); + assert!( + message.contains("whose anchor the instant selection node has no field for"), + "`x @ 100` did not say why its anchor cannot be bound: {message}" + ); +} + +/// A window's `@` anchor is bound as an absolute millisecond timestamp: `start()` and `end()` are +/// the statement's own bounds, and a number is the unix second it names. +#[test] +fn an_anchor_on_a_window_resolves_to_absolute_milliseconds() { + let cases = [ + // `@ `: 1234 is unix second 1234, which is 1_234_000 ms. + ("rate(x[5m] @ 1234)", 1_234_000), + // A sub-second anchor resolves to the millisecond it names. + ("rate(x[5m] @ 1.234)", 1_234), + // Prometheus allows a pre-epoch anchor: `signed_millis_since_epoch` takes `Err(earlier)`, + // where a `duration_since(UNIX_EPOCH).unwrap_or_default()` helper would silently give 0. + ("rate(x[5m] @ -1.5)", -1_500), + // The statement here is UNIX_EPOCH..100s, which is what `start()` and `end()` mean. + ("rate(x[5m] @ start())", 0), + ("rate(x[5m] @ end())", 100_000), + ]; + + for (query, anchor_ms) in cases { + let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime).unwrap(); + let (_, op) = root_window(&tree); + assert_eq!( + op.window.anchor_ms, + Some(anchor_ms), + "`{query}` bound the wrong anchor" + ); + assert_eq!(op.window.offset_ms, 0, "`{query}` invented an offset"); + } +} + +/// `end()` is the statement's end, not the last grid point. +/// +/// The grid stops before an end it does not land on, so an end of 103s over a 5s step has its last +/// point at 100s. Binding `end()` to that point instead of to the statement's own end would pass a +/// test whose end is a multiple of the step; here the two numbers differ. +#[test] +fn end_resolves_to_the_statement_end_not_the_last_grid_point() { + let stmt = eval_stmt_over("rate(x[5m] @ end())", Duration::from_secs(103)); + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime).unwrap(); + let (_, op) = root_window(&tree); + assert_eq!( + op.window.anchor_ms, + Some(103_000), + "`@ end()` bound something other than the statement's end" + ); + + // The grid the same statement compiles onto ends at 100s, so the two are not the same number. + let spec = op.output_grid.spec(); + let last_point = spec.start_ms + (spec.len as i64 - 1) * spec.step_ms; + assert_eq!( + last_point, 100_000, + "the test's own grid does not stop before the statement end" + ); +} + +/// An anchor and an offset are bound apart: the window ends at `anchor - offset`, which the +/// runtime folds, so both numbers have to survive compilation. +#[test] +fn an_anchor_and_an_offset_are_bound_apart() { + let cases = [ + ("rate(x[5m] @ 1.234 offset 1m)", 1_234, 60_000), + ("rate(x[5m] @ 1234 offset 1m)", 1_234_000, 60_000), + ("rate(x[5m] @ 1.234 offset -1m)", 1_234, -60_000), + ]; + + for (query, anchor_ms, offset_ms) in cases { + let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime).unwrap(); + let (_, op) = root_window(&tree); + assert_eq!( + op.window.anchor_ms, + Some(anchor_ms), + "`{query}` lost the anchor" + ); + assert_eq!(op.window.offset_ms, offset_ms, "`{query}` lost the offset"); + } +} + /// The numbers a tree is made of, at the edges: a span the step does not divide, offsets in either /// direction, ranges that cannot be a window, and an ordering the scan did not default to. #[test] @@ -212,6 +400,16 @@ fn the_numeric_edges_of_a_statement_are_bound_as_given() { SeriesOrdering::KeyThenTime, Expected::Rejected(Rejection::Plan), ), + // An instant selection's lookback is narrowed like every other duration. + ( + "a lookback wider than i64 milliseconds", + EvalStmt { + lookback_delta: Duration::from_millis(u64::MAX), + ..eval_stmt("x") + }, + SeriesOrdering::KeyThenTime, + Expected::Rejected(Rejection::Plan), + ), // A range window reads one series to its last chunk before the next, so a scan whose // chunks of different series interleave is refused instead of recorded onto an input the // window would fail on. @@ -293,8 +491,6 @@ fn the_numeric_edges_of_a_statement_are_bound_as_given() { #[test] fn an_unsupported_shape_is_rejected_and_names_what_it_found() { let cases = [ - // A bare selector reads every sample; there is no window to bind. - ("x", "the expression `x`"), ("x + 1", "the expression `x + 1`"), ("sum(x)", "the expression `sum(x)`"), ("x[5m:]", "the expression `x[5m:]`"), @@ -314,9 +510,6 @@ fn an_unsupported_shape_is_rejected_and_names_what_it_found() { "absent(x)", "a call to `absent`, which has no native range kernel", ), - // An anchored selector needs its anchor resolved to an absolute millisecond value, which - // this compiler does not do yet. - ("rate(x[5m] @ 100)", "`@`-anchored selector"), ]; for (query, named) in cases {