diff --git a/Cargo.lock b/Cargo.lock index a42a20af081..a00d5b123f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4559,6 +4559,7 @@ dependencies = [ "jsonb", "num", "num-traits", + "once_cell", "ordered-float 4.6.0", "parquet-variant", "parquet-variant-compute", diff --git a/src/datatypes/Cargo.toml b/src/datatypes/Cargo.toml index d1dd70acef1..555aac1813e 100644 --- a/src/datatypes/Cargo.toml +++ b/src/datatypes/Cargo.toml @@ -26,6 +26,7 @@ enum_dispatch = "0.3" jsonb.workspace = true num = "0.4" num-traits = "0.2" +once_cell.workspace = true ordered-float.workspace = true parquet-variant.workspace = true parquet-variant-compute.workspace = true diff --git a/src/datatypes/src/series/time.rs b/src/datatypes/src/series/time.rs index 5bff2d45493..966ee2f6e15 100644 --- a/src/datatypes/src/series/time.rs +++ b/src/datatypes/src/series/time.rs @@ -16,6 +16,7 @@ use std::borrow::Cow; use std::sync::Arc; use arrow::buffer::ScalarBuffer; +use once_cell::sync::OnceCell; use snafu::OptionExt; use crate::error::{InvalidSeriesSnafu, Result}; @@ -77,31 +78,27 @@ impl GridSpec { } } -/// A materialized, shareable timestamp grid. +/// A lazily materialized, shareable timestamp grid. #[derive(Debug)] pub struct EvalGrid { spec: GridSpec, - timestamps: ScalarBuffer, + timestamps: OnceCell>, } impl EvalGrid { /// Allocates the grid timestamps once for sharing across result series. pub fn try_new(spec: GridSpec) -> Result { - GridSpec::try_new(spec.start_ms, spec.step_ms, spec.len)?; - let len = - usize::try_from(spec.len).map_err(|_| invalid_series("grid length overflows usize"))?; - let mut values = Vec::new(); - values - .try_reserve_exact(len) - .map_err(|_| invalid_series("grid timestamp allocation failed"))?; - for index in 0..len { - // The validated endpoint bounds every point in this positive-step grid. - let timestamp = (spec.start_ms as i128) + (index as i128) * (spec.step_ms as i128); - values.push(timestamp as i64); - } + let grid = Self::try_deferred(spec)?; + grid.timestamps()?; + Ok(grid) + } + + /// Creates a validated grid without allocating its timestamp array. + pub fn try_deferred(spec: GridSpec) -> Result { + let spec = GridSpec::try_new(spec.start_ms, spec.step_ms, spec.len)?; Ok(Self { spec, - timestamps: values.into(), + timestamps: OnceCell::new(), }) } @@ -110,9 +107,32 @@ impl EvalGrid { self.spec } - /// Returns the shared timestamp buffer. - pub fn timestamps(&self) -> &ScalarBuffer { - &self.timestamps + /// Returns the shared timestamp buffer, materializing it on first access. + pub fn timestamps(&self) -> Result<&ScalarBuffer> { + self.timestamps.get_or_try_init(|| { + let len = usize::try_from(self.spec.len) + .map_err(|_| invalid_series("grid length overflows usize"))?; + let mut values = Vec::new(); + values + .try_reserve_exact(len) + .map_err(|_| invalid_series("grid timestamp allocation failed"))?; + for index in 0..len { + let timestamp = + (self.spec.start_ms as i128) + (index as i128) * (self.spec.step_ms as i128); + values.push(timestamp as i64); + } + Ok(values.into()) + }) + } + + fn timestamp_at(&self, position: usize) -> Option { + if position as u64 >= self.spec.len { + return None; + } + let timestamp = + (self.spec.start_ms as i128) + (position as i128) * (self.spec.step_ms as i128); + // GridSpec validation bounds every position, and i128 holds usize * i64. + Some(timestamp as i64) } } @@ -130,7 +150,10 @@ impl GridRange { let end = start.checked_add(len).context(InvalidSeriesSnafu { reason: "grid range overflow".to_string(), })?; - if end > grid.timestamps.len() { + if end + > usize::try_from(grid.spec.len) + .map_err(|_| invalid_series("grid length overflows usize"))? + { return invalid("grid range exceeds its grid"); } Ok(Self { grid, start, len }) @@ -153,8 +176,8 @@ impl GridRange { self.len == 0 } /// Returns this range as a zero-copy Arrow buffer slice. - pub fn timestamps(&self) -> ScalarBuffer { - self.grid.timestamps.slice(self.start, self.len) + pub fn timestamps(&self) -> Result> { + Ok(self.grid.timestamps()?.slice(self.start, self.len)) } } @@ -230,7 +253,8 @@ impl GridRanges { /// path expects, so callers outside this crate build axes through /// [`TimeAxis::from_grid_spans`], which keeps [`TimeAxis::Grid`] for a single span. pub(crate) fn try_new(grid: Arc, spans: Vec) -> Result { - let grid_len = grid.timestamps().len(); + let grid_len = usize::try_from(grid.spec.len) + .map_err(|_| invalid_series("grid length overflows usize"))?; let mut rows = Vec::new(); rows.try_reserve_exact(spans.len()) .map_err(|_| invalid_series("grid span allocation failed"))?; @@ -584,11 +608,11 @@ impl TimeAxis { pub fn timestamp(&self, index: usize) -> Option { match self { Self::Grid(range) => (index < range.len) - .then(|| range.grid.timestamps.get(range.start + index).copied()) + .then(|| range.grid.timestamp_at(range.start.checked_add(index)?)) .flatten(), Self::GridRanges(ranges) => { let (span, offset) = ranges.span_of(index)?; - ranges.grid.timestamps.get(span.start + offset).copied() + ranges.grid.timestamp_at(span.start.checked_add(offset)?) } Self::Explicit(values) => values.get(index).copied(), Self::ExplicitChunks(chunks) => { @@ -751,28 +775,29 @@ impl TimeAxis { /// /// A single-run grid axis lends its buffer without copying; an axis of several runs, and a /// chunked explicit axis, allocate one array of the carried rows. - pub fn timestamps(&self) -> Cow<'_, [i64]> { + pub fn timestamps(&self) -> Result> { match self { - Self::Grid(range) => { - Cow::Borrowed(&range.grid.timestamps()[range.start()..range.start() + range.len()]) - } + Self::Grid(range) => Ok(Cow::Borrowed( + &range.grid.timestamps()?[range.start()..range.start() + range.len()], + )), Self::GridRanges(ranges) => { let mut timestamps = Vec::with_capacity(ranges.len()); for (position, _) in self.positions() { - match ranges.grid.timestamps.get(position) { - Some(timestamp) => timestamps.push(*timestamp), - // Unreachable: the constructor checks every span against the grid. - None => break, - } + timestamps.push( + ranges + .grid + .timestamp_at(position) + .ok_or_else(|| invalid_series("grid span timestamp is out of range"))?, + ); } - Cow::Owned(timestamps) + Ok(Cow::Owned(timestamps)) } - Self::Explicit(values) => Cow::Borrowed(values), - Self::ExplicitChunks(_) => Cow::Owned( + Self::Explicit(values) => Ok(Cow::Borrowed(values)), + Self::ExplicitChunks(_) => Ok(Cow::Owned( (0..self.len()) .filter_map(|row| self.timestamp(row)) .collect(), - ), + )), } } diff --git a/src/datatypes/src/series/time/tests.rs b/src/datatypes/src/series/time/tests.rs index de3ba29f621..9e45bf38ee8 100644 --- a/src/datatypes/src/series/time/tests.rs +++ b/src/datatypes/src/series/time/tests.rs @@ -32,18 +32,99 @@ fn grid_specs_validate_and_ranges_are_inclusive_without_rounding_up() { assert!(GridSpec::try_new(0, -1, 1).is_err()); assert!(GridSpec::try_new(0, 1, 0).is_err()); assert!(GridSpec::try_new(i64::MAX, 1, 2).is_err()); + let extreme = + Arc::new(EvalGrid::try_new(GridSpec::try_new(i64::MIN, i64::MAX, 3).unwrap()).unwrap()); assert_eq!( - &EvalGrid::try_new(GridSpec::try_new(i64::MIN, i64::MAX, 3).unwrap()) - .unwrap() - .timestamps()[..], + &extreme.timestamps().unwrap()[..], &[i64::MIN, -1, i64::MAX - 1] ); + let axis = TimeAxis::Grid(GridRange::try_new(extreme, 0, 3).unwrap()); + assert_eq!(axis.timestamp(0), Some(i64::MIN)); + assert_eq!(axis.timestamp(1), Some(-1)); + assert_eq!(axis.timestamp(2), Some(i64::MAX - 1)); + assert_eq!(axis.timestamp(3), None); + let deferred_extreme = Arc::new( + EvalGrid::try_deferred(GridSpec::try_new(i64::MIN, i64::MAX, 3).unwrap()).unwrap(), + ); + let axis = TimeAxis::Grid(GridRange::try_new(deferred_extreme.clone(), 0, 3).unwrap()); + assert_eq!(axis.timestamp(0), Some(i64::MIN)); + assert_eq!(axis.timestamp(1), Some(-1)); + assert_eq!(axis.timestamp(2), Some(i64::MAX - 1)); + assert_eq!(axis.timestamp(3), None); + assert!(deferred_extreme.timestamps.get().is_none()); assert!(GridSpec::try_new(i64::MIN, i64::MAX, 4).is_err()); assert!(GridSpec::try_new(0, i64::MAX, u64::MAX).is_err()); assert!(GridSpec::try_new(i64::MIN, 1, 2).is_ok()); assert!(GridSpec::try_new(i64::MAX - 1, 1, 2).is_ok()); } +#[test] +fn deferred_grid_keeps_large_grid_unmaterialized_until_requested() { + let len = (isize::MAX as u64 / std::mem::size_of::() as u64) + 1; + let spec = GridSpec::try_new(i64::MIN, 1, len).unwrap(); + let grid = Arc::new(EvalGrid::try_deferred(spec).unwrap()); + assert!(grid.timestamps.get().is_none()); + let range = GridRange::try_new(grid.clone(), 0, len as usize).unwrap(); + let ranges = GridRanges::try_new(grid.clone(), vec![GridSpan::new(1, 2)]).unwrap(); + assert_eq!(ranges.len(), 2); + let range_axis = TimeAxis::Grid(range); + assert_eq!(range_axis.timestamp(1), Some(i64::MIN + 1)); + assert_eq!( + range_axis.timestamp(len as usize - 1), + Some(i64::MIN + (len as i64 - 1)) + ); + assert_eq!(range_axis.timestamp(len as usize), None); + let axis = TimeAxis::GridRanges(ranges); + assert_eq!(axis.timestamp(1), Some(i64::MIN + 2)); + assert_eq!(axis.timestamp(2), None); + assert_eq!(axis.timestamp(len as usize - 1), None); + assert!(grid.timestamps.get().is_none()); +} + +#[test] +fn deferred_grid_materialization_is_shared_across_concurrent_accesses() { + let grid = Arc::new(EvalGrid::try_deferred(GridSpec::try_new(-10, 2, 4).unwrap()).unwrap()); + let eager = EvalGrid::try_new(GridSpec::try_new(-10, 2, 4).unwrap()).unwrap(); + assert!(eager.timestamps.get().is_some()); + let threads: Vec<_> = (0..8) + .map(|_| { + let grid = grid.clone(); + std::thread::spawn(move || grid.timestamps().unwrap().as_ptr() as usize) + }) + .collect(); + let pointers: Vec = threads + .into_iter() + .map(|thread| thread.join().unwrap()) + .collect(); + assert!(pointers.iter().all(|pointer| *pointer == pointers[0])); + assert_eq!(&grid.timestamps().unwrap()[..], &[-10, -8, -6, -4]); + assert!(grid.timestamps.get().is_some()); +} + +#[test] +fn deferred_sparse_gather_materializes_only_requested_timestamps() { + let grid = Arc::new(EvalGrid::try_deferred(GridSpec::try_new(10, 10, 6).unwrap()).unwrap()); + let axis = TimeAxis::GridRanges( + GridRanges::try_new(grid.clone(), vec![GridSpan::new(1, 2), GridSpan::new(5, 1)]).unwrap(), + ); + assert_eq!(axis.timestamps().unwrap().as_ref(), &[20, 30, 60]); + assert!(grid.timestamps.get().is_none()); +} + +#[test] +fn deferred_grid_allocation_failure_is_reported_and_not_cached() { + let len = (isize::MAX as u64 / std::mem::size_of::() as u64) + 1; + let grid = + Arc::new(EvalGrid::try_deferred(GridSpec::try_new(i64::MIN, 1, len).unwrap()).unwrap()); + assert!(grid.timestamps().is_err()); + assert!(grid.timestamps.get().is_none()); + let range = GridRange::try_new(grid.clone(), 0, 1).unwrap(); + assert!(range.timestamps().is_err()); + assert!(TimeAxis::Grid(range).timestamps().is_err()); + assert!(grid.timestamps().is_err()); + assert!(grid.timestamps.get().is_none()); +} + #[test] fn grid_materialization_revalidates_public_specs_and_shares_ranges_by_value() { assert!( @@ -70,14 +151,26 @@ fn grid_materialization_revalidates_public_specs_and_shares_ranges_by_value() { }) .is_err() ); + assert!( + EvalGrid::try_deferred(GridSpec { + start_ms: 0, + step_ms: 0, + len: 1 + }) + .is_err() + ); let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(-5, 3, 5).unwrap()).unwrap()); - assert_eq!(&grid.timestamps()[..], &[-5, -2, 1, 4, 7]); + assert!(grid.timestamps.get().is_some()); + assert_eq!(&grid.timestamps().unwrap()[..], &[-5, -2, 1, 4, 7]); let a = GridRange::try_new(grid.clone(), 1, 3).unwrap(); let b = GridRange::try_new(Arc::new(EvalGrid::try_new(grid.spec()).unwrap()), 1, 3).unwrap(); assert_eq!(a, b); - let values = a.timestamps(); - assert_eq!(values.as_ptr(), grid.timestamps().as_ptr().wrapping_add(1)); + let values = a.timestamps().unwrap(); + assert_eq!( + values.as_ptr(), + grid.timestamps().unwrap().as_ptr().wrapping_add(1) + ); assert_eq!(&values[..], &[-2, 1, 4]); assert!(GridRange::try_new(grid.clone(), 5, 0).unwrap().is_empty()); assert!(GridRange::try_new(grid.clone(), 4, 2).is_err()); diff --git a/src/promql/src/aggregate/state.rs b/src/promql/src/aggregate/state.rs index d609334aa10..b70d5c7fe17 100644 --- a/src/promql/src/aggregate/state.rs +++ b/src/promql/src/aggregate/state.rs @@ -161,8 +161,8 @@ impl FloatMerger { .eval_grid .as_ref() .expect("checked at construction") - .timestamps() - .len(); + .spec() + .len as usize; if partial.op != self.op { return Err(DataFusionError::Execution(format!( "cannot merge {:?} state into {:?}", diff --git a/src/promql/src/aggregate/tests.rs b/src/promql/src/aggregate/tests.rs index b230427e2fc..fef8bb651ac 100644 --- a/src/promql/src/aggregate/tests.rs +++ b/src/promql/src/aggregate/tests.rs @@ -395,7 +395,7 @@ fn fields_are_aggregated_independently() { }) .collect() }; - let grid_len = multi.eval_grid.as_ref().unwrap().timestamps().len(); + let grid_len = multi.eval_grid.as_ref().unwrap().spec().len as usize; let q_values: Vec> = (0..grid_len) .map(|i| Some([0.0, 0.25, 0.5, 0.9, 1.0][i % 5])) .collect(); @@ -1149,7 +1149,7 @@ fn partials_merge_to_the_serial_result() { .collect(); let names: Vec<&str> = fields.iter().map(String::as_str).collect(); let schema = schema_with_grid_len(&names, rng.random_range(1..80)); - let points = schema.eval_grid.as_ref().unwrap().timestamps().len(); + let points = schema.eval_grid.as_ref().unwrap().spec().len as usize; let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap(); let mut series = Vec::new(); for index in 0..rng.random_range(1..12) { diff --git a/src/promql/src/golden.rs b/src/promql/src/golden.rs index f15f2b126c7..cab57aa3856 100644 --- a/src/promql/src/golden.rs +++ b/src/promql/src/golden.rs @@ -254,7 +254,7 @@ fn fold_one(pieces: Vec) -> Series { None => panic!("expected a grid series, got {:?}", pieces[0].time), }; let mut rows: Vec>>> = - (0..grid.timestamps().len()).map(|_| None).collect(); + (0..grid.spec().len as usize).map(|_| None).collect(); for piece in &pieces { let mut cursors = piece .fields diff --git a/src/promql/src/grid_series.rs b/src/promql/src/grid_series.rs index 15edb845c86..c9d8bc19d1f 100644 --- a/src/promql/src/grid_series.rs +++ b/src/promql/src/grid_series.rs @@ -163,10 +163,10 @@ impl GridSeriesBuilder { self.schema.fields.len() ))); } - if index < self.next || index >= self.grid.timestamps().len() { + if index < self.next || index >= self.grid.spec().len as usize { return Err(DataFusionError::Execution(format!( "grid series index {index} is not increasing or exceeds grid length {}", - self.grid.timestamps().len() + self.grid.spec().len as usize ))); } self.next = index + 1; diff --git a/src/promql/src/grid_series/reference.rs b/src/promql/src/grid_series/reference.rs index cee219b0dfd..fb59611ac2b 100644 --- a/src/promql/src/grid_series/reference.rs +++ b/src/promql/src/grid_series/reference.rs @@ -43,7 +43,7 @@ pub(crate) fn expand(series: &Series) -> Dense { panic!("expected a grid series, got {:?}", series.time); }; let fields = series.fields.as_slice(); - let mut dense = vec![vec![None; fields.len()]; grid.timestamps().len()]; + let mut dense = vec![vec![None; fields.len()]; grid.spec().len as usize]; let mut cursors = fields .iter() .map(|field| field.cursor()) @@ -124,7 +124,7 @@ fn assert_series_matches( last + 1 - first, "{context}: the result carries a row for a position that holds no value" ); - assert_eq!(grid.timestamps().len(), expected.len()); + assert_eq!(grid.spec().len as usize, expected.len()); assert_eq!(&actual.metric, metric); let bits = |dense: &Dense| -> Vec>> { dense @@ -148,7 +148,7 @@ pub(crate) fn stitch(pieces: &[Series]) -> Dense { panic!("expected a grid series, got {:?}", piece.time); }; let dense = dense - .get_or_insert_with(|| vec![vec![None; piece.fields.len()]; grid.timestamps().len()]); + .get_or_insert_with(|| vec![vec![None; piece.fields.len()]; grid.spec().len as usize]); let mut cursors = piece .fields .iter() diff --git a/src/promql/src/grid_series/tests.rs b/src/promql/src/grid_series/tests.rs index 7692513a036..fbc6ebd0dc1 100644 --- a/src/promql/src/grid_series/tests.rs +++ b/src/promql/src/grid_series/tests.rs @@ -312,12 +312,7 @@ fn runs_never_bridge_positions_without_a_value() { let series = builder.finish(metric(0)).unwrap().expect("two rows"); // One series over a million-position grid, with two runs, and the axis says where they sit. assert_eq!( - series - .time - .grid() - .expect("a grid series") - .timestamps() - .len(), + series.time.grid().expect("a grid series").spec().len as usize, SPAN, "the grid really holds a million positions" ); @@ -787,7 +782,7 @@ fn merge_of_many_segments_keeps_one_row_per_value() { assert_eq!(merged.len(), 1); let series = &merged[0]; assert_eq!( - series.time.grid().unwrap().timestamps().len(), + series.time.grid().unwrap().spec().len as usize, GRID, "the grid really holds a million positions" ); @@ -971,7 +966,10 @@ mod acceptance { // The axis lends the grid's buffer instead of copying it: the fast path is not only a // label, it is the same rows without per-series timestamp storage. assert!( - matches!(series.time.timestamps(), std::borrow::Cow::Borrowed(_)), + matches!( + series.time.timestamps().unwrap(), + std::borrow::Cow::Borrowed(_) + ), "a single range lends its timestamps" ); } diff --git a/src/promql/src/info.rs b/src/promql/src/info.rs index 85c9ce73e0c..295f483a59f 100644 --- a/src/promql/src/info.rs +++ b/src/promql/src/info.rs @@ -75,7 +75,7 @@ pub fn info( .eval_grid .as_ref() .ok_or_else(|| DataFusionError::Execution("info requires an evaluated grid".into()))?; - let points = grid.timestamps().len(); + let points = grid.spec().len as usize; for series in base { if series.time.grid().map(|grid| grid.spec()) != Some(grid.spec()) || series.fields.len() != schema.fields.len() diff --git a/src/promql/src/instant_eval.rs b/src/promql/src/instant_eval.rs index 2b980fe5c3d..0e76ec616a5 100644 --- a/src/promql/src/instant_eval.rs +++ b/src/promql/src/instant_eval.rs @@ -226,7 +226,7 @@ pub fn timestamp_of_evaluated( "timestamp input does not match its schema".into(), )); } - let timestamps = grid.timestamps(); + let timestamps = grid.timestamps().map_err(crate::grid_series::external)?; let mut builder = GridSeriesBuilder::try_new(schema.clone())?; let mut cursors: Vec<_> = series.fields.iter().map(|field| field.cursor()).collect(); let mut row = Vec::with_capacity(cursors.len()); diff --git a/src/promql/src/map/tests/broadcast.rs b/src/promql/src/map/tests/broadcast.rs index 7d8022ef967..8bb4fe2c5b2 100644 --- a/src/promql/src/map/tests/broadcast.rs +++ b/src/promql/src/map/tests/broadcast.rs @@ -376,8 +376,8 @@ fn a_vector_scalar_operation_of_sparse_operands_keeps_the_shared_positions() { .eval_grid .as_ref() .expect("the fixture schema has a grid") - .timestamps() - .len(); + .spec() + .len as usize; let mut expected: Dense = vec![vec![None]; points_on_grid]; expected[5] = vec![Some(52.0)]; expected[9] = vec![Some(93.0)]; diff --git a/src/promql/src/map/tests/prometheus.rs b/src/promql/src/map/tests/prometheus.rs index 386f153dafb..3db701129bf 100644 --- a/src/promql/src/map/tests/prometheus.rs +++ b/src/promql/src/map/tests/prometheus.rs @@ -174,7 +174,7 @@ fn evaluate(case: &Case) -> Vec { } fn grid_len(case: &Case) -> usize { - case.schema.eval_grid.as_ref().unwrap().timestamps().len() + case.schema.eval_grid.as_ref().unwrap().spec().len as usize } #[test] diff --git a/src/promql/src/map_time.rs b/src/promql/src/map_time.rs index c0a2c5b94c3..c962f46be49 100644 --- a/src/promql/src/map_time.rs +++ b/src/promql/src/map_time.rs @@ -170,7 +170,12 @@ fn eval_time_series(schema: &SeriesSchema, value: impl Fn(i64) -> f64) -> Result let grid = schema.eval_grid.clone().ok_or_else(|| { DataFusionError::Execution("time functions require an evaluation grid".into()) })?; - let values: Float64Array = grid.timestamps().iter().map(|&ms| value(ms)).collect(); + let values: Float64Array = grid + .timestamps() + .map_err(external)? + .iter() + .map(|&ms| value(ms)) + .collect(); let range = GridRange::try_new(grid.clone(), 0, values.len()).map_err(external)?; let fields = vec![Samples::from_float(values); schema.fields.len()]; Series::try_new( diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index c4842b74bfa..517cf4a8678 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -388,11 +388,11 @@ impl NativeNode { } ensure_not_interleaved(input)?; // What is checked here and just below is that the declared input grid equals the - // input's grid, and that the output grid matches the declared one. The grid is - // not checked to cover the window's read range - // `(out.start - offset - range, out.end - offset]`: only a subquery feeds an - // evaluated input to a window, and this compiler does not accept subqueries yet, - // so that coverage check waits for it. + // input's grid, and that the output grid matches the declared one. That the grid + // covers the window's read range + // `(out.start - offset - range, out.end - offset]` is not checked here: only a + // subquery feeds an evaluated input to a window, and deriving that coverage is the + // compiler's, which builds the subquery's child grid to span it. match (op.input_grid.as_ref(), schema.eval_grid.as_ref()) { (None, None) => {} (Some(bound), Some(actual)) if bound.spec() == actual.spec() => {} diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index 3a7e2384809..02af5288cdc 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -714,7 +714,7 @@ impl RangeEvaluator { } // The input may skip positions: it is ingested by the positions it carries, and the points // it skips hold no sample, which is exactly what a hole means. - let timestamps = time.timestamps(); + let timestamps = time.timestamps().map_err(external)?; let last_ts = timestamps.last().copied(); if !last && last_ts.is_none() { // An empty delivery inside a series decides nothing: it is indistinguishable from a @@ -1045,7 +1045,7 @@ impl RangeEvaluator { let points_read = self.points_read(); let state = self.series.as_mut().expect("checked by check_chunk"); let spec = self.grid.spec(); - let grid = self.grid.timestamps(); + let grid = self.grid.timestamps().map_err(external)?; let range = i128::from(self.window.range_ms); let mut empty_tail = false; while state.cursor < points_read { @@ -1212,7 +1212,7 @@ impl RangeEvaluator { /// only the first. fn points_read(&self) -> usize { match self.replay { - Replay::PerPoint => self.grid.timestamps().len(), + Replay::PerPoint => self.grid.spec().len as usize, Replay::Once | Replay::EveryPoint => 1, } } diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index 56c830841a0..0c9aba0e139 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -279,7 +279,7 @@ fn fold_chunks(chunks: &[EvaluatedChunk], schema: &Arc) -> Option< .metric .clone(); let width = chunks[0].series.fields.len(); - let grid_len = schema.eval_grid.as_ref().unwrap().timestamps().len(); + let grid_len = schema.eval_grid.as_ref().unwrap().spec().len as usize; let mut rows: Vec>>> = (0..grid_len).map(|_| None).collect(); let mut closing = false; let mut previous_end = 0; @@ -495,6 +495,7 @@ fn window_end(grid: &EvalGrid, window: RangeWindow, eval_ts: i64) -> i64 { /// Rows of `rows`, already on the evaluation timeline, inside each grid point's window. fn windows_of(rows: &[Row], grid: &EvalGrid, window: RangeWindow) -> Vec> { grid.timestamps() + .unwrap() .iter() .map(|&eval_ts| { let end = window_end(grid, window, eval_ts); @@ -520,6 +521,7 @@ fn reference( ) -> Dense { let (kept, windows) = reference_windows(rows, grid, window, drop_stale); grid.timestamps() + .unwrap() .iter() .zip(windows) .map(|(&eval_ts, rows)| { @@ -2610,7 +2612,7 @@ fn random_evaluated( fields: usize, host: u64, ) -> (Option, Vec) { - let mut dense = random_dense(rng, grid.timestamps().len(), fields); + let mut dense = random_dense(rng, grid.spec().len as usize, fields); for value in dense.iter_mut().flatten().flatten() { if rng.random_bool(0.05) { *value = stale(); @@ -2626,6 +2628,7 @@ fn random_evaluated( } let rows = grid .timestamps() + .unwrap() .iter() .zip(&dense) .filter(|(_, row)| row.iter().any(Option::is_some)) @@ -2750,7 +2753,7 @@ fn subquery_grid_covers_every_step_multiple_in_the_windows() { let fold = fold_offset(&outer, window); let last_eval = match window.anchor_ms { Some(_) => spec.start_ms, - None => *outer.timestamps().last().unwrap(), + None => *outer.timestamps().unwrap().last().unwrap(), }; let lower = spec.start_ms - fold - window.range_ms; let expected: Vec = ((lower.div_euclid(step) - 1) * step..=last_eval - fold) @@ -2759,7 +2762,13 @@ fn subquery_grid_covers_every_step_multiple_in_the_windows() { .collect(); let actual = subquery_grid(spec, window, step) .unwrap() - .map(|grid| EvalGrid::try_new(grid).unwrap().timestamps().to_vec()) + .map(|grid| { + EvalGrid::try_new(grid) + .unwrap() + .timestamps() + .unwrap() + .to_vec() + }) .unwrap_or_default(); assert_eq!(actual, expected, "seed {seed}"); } @@ -3764,17 +3773,12 @@ fn a_million_position_grid_with_two_values_is_one_series_of_two_rows() { // What the axis does carry is the shared grid: every result of the query addresses rows on it. assert_eq!( - series - .time - .grid() - .expect("a grid series") - .timestamps() - .len(), + series.time.grid().expect("a grid series").spec().len as usize, SPAN as usize, "the grid really holds a million positions" ); assert_eq!( - series.time.timestamps().len(), + series.time.timestamps().unwrap().len(), 2, "the axis hands out two timestamps, not one per grid position" ); @@ -4148,6 +4152,7 @@ fn a_prefix_hands_out_its_rows_before_the_series_ends() { fn dense_sum_over_time(rows: &[Row], grid: &EvalGrid, window: RangeWindow) -> Dense { assert_eq!(window.offset_ms, 0, "this reference folds no offset"); grid.timestamps() + .unwrap() .iter() .map(|&eval_ts| { let inside: Vec = rows @@ -4462,7 +4467,7 @@ fn two_composed_evaluators_agree_with_the_dense_reference() { .iter() .enumerate() .filter_map(|(index, row)| { - row[0].map(|value| (inner.timestamps()[index], vec![Some(value)])) + row[0].map(|value| (inner.timestamps().unwrap()[index], vec![Some(value)])) }) .collect(); let expected = dense_sum_over_time(&stage_one_rows, &outer, outer_window); diff --git a/src/promql/src/select.rs b/src/promql/src/select.rs index c65a7955c36..6d37d638d16 100644 --- a/src/promql/src/select.rs +++ b/src/promql/src/select.rs @@ -610,7 +610,7 @@ fn grid_points(schema: &SeriesSchema, series: &[Series]) -> Result { )); } } - Ok(grid.timestamps().len()) + Ok(grid.spec().len as usize) } /// The first field's value of `series` at grid `point`, looked up by position. diff --git a/src/promql/src/vector_match.rs b/src/promql/src/vector_match.rs index 83e50a71df2..e5e069f5485 100644 --- a/src/promql/src/vector_match.rs +++ b/src/promql/src/vector_match.rs @@ -720,7 +720,10 @@ fn binary_field( Cardinality::ManyToOne(include) => (lhs, rhs, &include[..], false), Cardinality::OneToMany(include) => (rhs, lhs, &include[..], true), }; - let points = grid.timestamps().len(); + let points = grid.spec().len as usize; + let timestamp_at = |point: usize| { + (grid.spec().start_ms as i128 + point as i128 * grid.spec().step_ms as i128) as i64 + }; let errors_before = errors.len(); // The pairing phase is entered once, before the field allocates anything: its presence flags, // its match groups, the widest label set a pairing builds and the histogram payload one row of @@ -948,7 +951,7 @@ fn binary_field( operation at {} ms; many-to-many matching not allowed: matching labels must be \ unique on one side", render(key), - grid.timestamps()[point] + timestamp_at(point) ), )); } @@ -961,7 +964,7 @@ fn binary_field( "multiple matches for labels {} at {} ms: many-to-one matching must be explicit \ (group_left/group_right)", render(key), - grid.timestamps()[point] + timestamp_at(point) ), )); } @@ -975,14 +978,14 @@ fn binary_field( "multiple matches for labels {} at {} ms: grouping labels must ensure \ unique matches", render(&duplicate.key), - grid.timestamps()[duplicate.point] + timestamp_at(duplicate.point) ), ) } else { ( duplicate.point, 2, - same_labelset(&duplicate.key, grid.timestamps()[duplicate.point]), + same_labelset(&duplicate.key, timestamp_at(duplicate.point)), ) }); } @@ -1151,7 +1154,9 @@ pub fn set_vectors( let rhs = Operand::try_new(rhs_schema, rhs, field)?; let mut results = set_field(op, labels, &lhs, &rhs)?; if let Some(duplicate) = results.sort_rows() { - let timestamp = grid.timestamps()[duplicate.point]; + let timestamp = (grid.spec().start_ms as i128 + + duplicate.point as i128 * grid.spec().step_ms as i128) + as i64; errors.push((duplicate.point, 0, same_labelset(&duplicate.key, timestamp))); } fields.push(results); diff --git a/src/promql/src/vector_match/tests.rs b/src/promql/src/vector_match/tests.rs index 52ef09d3435..ec7671c9eb6 100644 --- a/src/promql/src/vector_match/tests.rs +++ b/src/promql/src/vector_match/tests.rs @@ -130,7 +130,7 @@ fn random_side(rng: &mut StdRng, grid: &Arc, fields: usize) -> Side { } // Each label set starts its own series: only `begin_series` resets the builder. builder.begin_series(); - let values = random_dense(rng, grid.timestamps().len(), fields); + let values = random_dense(rng, grid.spec().len as usize, fields); for (index, row) in values.iter().enumerate() { builder.push(index, row).unwrap(); } @@ -489,7 +489,7 @@ fn binary_vectors_match_prometheus_reference() { fill: FillValues::default(), }; let fields = lhs_fields.min(rhs_fields); - let points = grid.timestamps().len(); + let points = grid.spec().len as usize; let expected = per_field(fields, points, |field| { reference_binary( &matching, @@ -561,7 +561,7 @@ fn set_vectors_match_prometheus_reference() { let rhs = random_side(&mut rng, &grid, fields); let op = [SetOp::And, SetOp::Or, SetOp::Unless][rng.random_range(0..3)]; let labels = random_labels(&mut rng); - let points = grid.timestamps().len(); + let points = grid.spec().len as usize; let expected = per_field(fields, points, |field| { reference_set( op, diff --git a/src/query/src/promql/native_compile.rs b/src/query/src/promql/native_compile.rs index 84ff6d039a5..7465be279a1 100644 --- a/src/query/src/promql/native_compile.rs +++ b/src/query/src/promql/native_compile.rs @@ -19,37 +19,64 @@ //! accepts is one recursive rule: //! //! ```text -//! e.g. x -//! ([]) e.g. rate(x[5m]) -//! (, []) e.g. quantile_over_time(0.9, x[5m]) -//! ([], [, ]) -//! e.g. predict_linear(x[5m], 1), -//! double_exponential_smoothing(x[5m], 0.5, 0.5) -//! ([, ]*) e.g. abs(rate(x[5m])), clamp_min(x, 0) -//! e.g. x * 100, x > bool 0 -//! e.g. 1 / x, 0 < x -//! - e.g. -x -//! [(by|without)()] () -//! e.g. avg by(job)(rate(x[5m]) * 100) -//! () [(by|without)()] -//! e.g. count (x) by(job) -//! quantile(, ) [(by|without)()] -//! e.g. quantile(0.9, x) by(job) +//! one of, recursively: +//! e.g. x, x offset -5m +//! () e.g. rate(x[5m]), rate(x[5m:1m] @ 1234) +//! quantile_over_time(, ) e.g. quantile_over_time(0.9, x[5m]) +//! predict_linear(, ) e.g. predict_linear(x[5m], 300) +//! double_exponential_smoothing(, , ) +//! e.g. double_exponential_smoothing(x[5m], .5, .1) +//! ([, ]*) e.g. abs(rate(x[5m])), clamp_min(x, 0) +//! e.g. x * 100, x > bool 0 +//! e.g. 1 / x, 0 < x +//! - e.g. -x +//! [(by|without)()] () +//! e.g. avg by(job)(rate(x[5m]) * 100) +//! () [(by|without)()] +//! e.g. count (x) by(job) +//! quantile(, ) [(by|without)()] +//! e.g. quantile(0.9, x) by(job) +//! +//! [] a range selector: raw samples +//! | [:] +//! a subquery: evaluated on a grid; +//! the step may be omitted as `[:]` +//! nothing, `offset `, `@ `, or +//! both, in either order — e.g. `x[5m] @ 1234 +//! offset 1m` //! ``` //! -//! where `` is `sum`, `avg`, `count`, `min`, `max`, `group`, `stddev` or `stdvar`. +//! where `` is `sum`, `avg`, `count`, `min`, `max`, `group`, `stddev`, `stdvar` or +//! `quantile`, the function names are [`range_signature`]'s and [`math_signature`]'s, and which side +//! of the matrix a range function's constants go on is [`RangeSignature`]'s: before it for +//! `quantile_over_time`, after it for `predict_linear` and `double_exponential_smoothing`. A math +//! function's constants always follow its vector. //! -//! which become `Input` (raw samples) → `Instant` (the last sample within the statement's lookback), -//! `Input` (raw samples) → `Window` (the range function over the statement's grid), +//! Parentheses are transparent wherever they appear — `rate((x[5m]))` is `rate(x[5m])` — because +//! they say how a statement was typed rather than what it asks for. +//! +//! A subquery's `` is any shape above, so a subquery nests inside a subquery on a grid +//! of its own, as in `rate(rate(x[5m:1m])[10m:30s])`; a `` left out means Prometheus's default +//! evaluation interval, one minute. Its `offset` and its `@` anchor are its own, not the inner +//! expression's, so `rate(x[5m:1m] @ end() offset 1m)` reads one fixed interval. A `` +//! outside a `` takes an `offset` and no anchor: the instant shape has no field for one, so +//! `x @ 100` is refused and `rate(x[5m] @ 100)` is not. +//! +//! Those shapes become `Input` (raw samples) → `Instant` (the last sample within the statement's +//! lookback), `Input` (raw samples) → `Window` (the range function over the statement's grid), +//! `` → `Window` whose input is evaluated (the subquery, whose input grid is its own), //! `` → `Map` (a pointwise math function over what its argument evaluates to) and, over //! any of those, `Binary` (an arithmetic or comparison operation with a constant on one side), //! `Scalar` (that constant) and `Aggregate` (one of the nine aggregators that fold a group's values -//! into one value per grid point, optionally grouped). A selector may carry -//! `offset` in either selection shape and, in the window shape, an `@` anchor. Everything else is a +//! into one value per grid point, optionally grouped). 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 compiler binds a subquery's child grid and `RangeWindow` independently of runtime execution: +//! compilation describes the anchored interval, while runtime support determines how that bound +//! plan is consumed. +//! //! Two decisions keep the recursion from growing a branch per query. A constant argument — a //! quantile, a rounding step, a clamp bound — is folded out of the statement by [`constant_value`], //! so `quantile_over_time(1 - 0.1, x[5m])` binds the number it means rather than the expression it @@ -100,31 +127,42 @@ use promql::native::plan::{ AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, }; -use promql::range_eval::{RangeFunction, RangeWindow}; +use promql::range_eval::{RangeFunction, RangeWindow, subquery_grid}; use promql::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels}; use promql_parser::parser::token::TokenType; use promql_parser::parser::value::ValueType; use promql_parser::parser::{ AggregateExpr, AtModifier, BinModifier, BinaryExpr, Call, EvalStmt, Expr as PromExpr, - LabelModifier, MatrixSelector, NumberLiteral, Offset, VectorMatchCardinality, + LabelModifier, MatrixSelector, NumberLiteral, Offset, SubqueryExpr, VectorMatchCardinality, VectorMatchFillValues, VectorSelector, token, }; use crate::promql::label_values::signed_millis_since_epoch; /// The shapes this compiler accepts, quoted in every rejection. -const SUPPORTED: &str = "a selector, a range function over a range selector — \ - `([])`, with constants before the selector \ - as in `quantile_over_time(0.9, x[5m])` or after it as in \ - `predict_linear(x[5m], 1)` and \ - `double_exponential_smoothing(x[5m], 0.5, 0.5)` — a math function over \ - any of them, a binary operation between one of them and a constant, a \ - negation of one of them, or one of the aggregators that folds a group's \ - values into one value per grid point — `sum`, `avg`, `count`, `min`, `max`, \ - `group`, `stddev`, `stdvar` and `quantile(, ...)` — over any of \ - them, optionally grouped by `by` or `without`, as in `x`, `rate(x[5m])`, \ - `quantile_over_time(0.9, x[5m])`, `abs(rate(x[5m]))`, `2 / (x + 1)`, \ - `-x`, `avg by(job)(rate(x[5m]) * 100)` or `quantile(0.9, x)`"; +/// +/// It is one sentence of grammar rather than a list of examples, because it is what a refused query +/// reads: the shapes have to be true of every query the compiler binds, and the examples only show +/// them. The two matrix shapes and the recursion through them are stated first, because that +/// recursion is what a reader is most likely to get wrong. +const SUPPORTED: &str = "a selector — `x`, optionally with an `offset` — a range function over \ + a matrix, where a matrix is `[]` as in `rate(x[5m])` or \ + `[:]` as in `rate(x[5m:1m])`, with the step \ + optionally omitted as `[:]`, where the subquery's `` \ + is any shape here, so a subquery nests on a grid of its own as in \ + `rate(rate(x[5m:1m])[10m:30s])`, and where either matrix shape carries the \ + `offset` and `@` anchor written on it as in `rate(x[5m] @ 1234)`; a math \ + function over any of them (`abs(rate(x[5m]))`, `clamp_min(x, 0)`), a binary \ + operation between one of them and a constant (`2 / (x + 1)`, \ + `x > bool 0`), a negation of one of them (`-x`), or one of the aggregators \ + that folds a group's values into one value per grid point — `sum`, `avg`, \ + `count`, `min`, `max`, `group`, `stddev`, `stdvar` and \ + `quantile(, )` — over any of them, optionally \ + grouped by `by` or `without` as in `avg by(job)(rate(x[5m]) * 100)`; a \ + range function's own constants sit where that function takes them, before \ + the matrix for `quantile_over_time(0.9, x[5m])` and after it for \ + `predict_linear(x[5m], 300)` and \ + `double_exponential_smoothing(x[5m], 0.5, 0.1)`"; /// Compiles `stmt` into a tree that reads `input` and evaluates the statement's expression over it. /// @@ -179,6 +217,9 @@ pub fn compile( stmt, input, ordering, + // The grid is built once for the whole compilation rather than per node: every node of a + // tree without a subquery lands on this one, and a subquery replaces it for its own subtree. + grid: statement_grid(stmt)?, }; // The whole tree is validated once, here: a node compiled as another node's input is checked as // part of the tree rather than on its own, so the same node cannot be legal in one place and @@ -198,6 +239,13 @@ struct Context<'a> { input: Arc, /// The ordering the caller's scan guarantees. ordering: SeriesOrdering, + /// The grid the expression being compiled is evaluated on. + /// + /// It is the statement's grid for everything except the inside of a subquery, which opens a grid + /// of its own: `rate(x[5m:1m])` evaluates `x` at one-minute steps over the five-minute lookback, + /// not at the statement's steps. Carrying it here is what lets [`compile_expr`] recurse into a + /// subquery without every node builder having to ask which grid it is on. + grid: Arc, } /// Compiles `expr` into the node it denotes, recursing into whatever it is made of. @@ -211,7 +259,7 @@ fn compile_expr(context: &Context, expr: &PromExpr) -> Result { // would have to know how to run. The fold is checked first so that no shape below has to say // what it does with a constant operand. if let Some(value) = constant_value(expr) { - return scalar_node(context, value); + return Ok(scalar_node(context, value)); } match expr { // Parentheses group, so they change how a statement was typed rather than what it asks for: @@ -226,17 +274,19 @@ fn compile_expr(context: &Context, expr: &PromExpr) -> Result { } } -/// Binds `value` as a constant scalar on the statement's grid. +/// Binds `value` as a constant scalar on the grid of the expression being compiled. /// -/// The grid is the statement's rather than the constant's own: a constant has no timestamps, so the -/// only positions it can be broadcast over are the ones the statement is evaluated at, and -/// [`NativeTree::validate`] compares a binary operation's two grids by spec rather than by -/// allocation. -fn scalar_node(context: &Context, value: f64) -> Result { - Ok(NativeNode { +/// The grid is the context's rather than the constant's own: a constant has no timestamps, so the +/// only positions it can be broadcast over are the ones the expression it sits next to is evaluated +/// at — the statement's grid, or a subquery's inside one. [`NativeTree::validate`] compares a binary +/// operation's two grids by spec rather than by allocation. +/// +/// Nothing here can fail: the grid is the one [`Context`] already carries, so a caller needs no `?`. +fn scalar_node(context: &Context, value: f64) -> NativeNode { + NativeNode { expr: NativeExpr::Scalar(BoundScalar { value }), - output: NativeType::Scalar(statement_grid(context.stmt)?), - }) + output: NativeType::Scalar(Arc::clone(&context.grid)), + } } /// Compiles a negation as the multiplication it is: `-x` is `-1 * x`. @@ -249,7 +299,7 @@ fn negate(context: &Context, operand: &PromExpr) -> Result { // The constant goes on the left, where the minus was: `-x` subtracts nothing, but a reader of // the tree should see the same order the query was written in. Ok(broadcast( - scalar_node(context, -1.0)?, + scalar_node(context, -1.0), compile_expr(context, operand)?, BinaryOp::Mul, false, @@ -542,11 +592,17 @@ fn compile_call(context: &Context, call: &Call) -> Result { map_call(context, func.name, op, vector) } -/// Binds a range function over the range selector it is given, and the window that selector opens. +/// Binds a range function over the matrix it is given, and the window that matrix opens. /// -/// The selector's place is the signature's, so the constants on either side of it are read as the -/// signature says: `quantile_over_time` writes its parameter in front of the selector, while -/// `predict_linear` and `double_exponential_smoothing` write theirs behind it. +/// Where the matrix sits is the signature's, not a convention: `quantile_over_time(0.9, x[5m])` writes +/// its constant before the matrix, while `predict_linear(x[5m], 300)` and +/// `double_exponential_smoothing(x[5m], 0.5, 0.1)` write theirs after it, and a function with no +/// constants at all reads the matrix first. [`RangeSignature::selector_index`] is that position, and +/// everything else the call wrote is a constant — which keeps the position it was written at, so a +/// rejection names the argument the query wrote rather than the constant's offset among the constants. +/// +/// The matrix is either shape: a range selector, which reads raw samples, or a subquery, which +/// evaluates its own expression on a grid of its own. fn window_call( context: &Context, name: &str, @@ -563,19 +619,24 @@ fn window_call( signature.arguments() ))); } - // The selector sits where the signature says it does, and everything else is a constant. Reading - // it by position rather than by "the last argument" is what lets `predict_linear(x[5m], 300)` and + // The matrix sits where the signature says it does, and everything else is a constant. Reading it + // by position rather than by "the last argument" is what lets `predict_linear(x[5m], 300)` and // `quantile_over_time(0.9, x[5m])` share one path. let selector_index = signature.selector_index(); + // The selector stays in its original argument position for signature binding, while either + // matrix form can provide the samples folded by the range function. let argument = &args[selector_index]; - let PromExpr::MatrixSelector(MatrixSelector { vs, range }) = unparen(argument) else { - return Err(not_implemented(&format!( - "a call to `{name}` whose range selector is `{argument}`, where it has to be a range \ - selector like `x[5m]`" - ))); - }; let function = signature.bind(name, args)?; - window_selection(context, function, vs, range) + match unparen(argument) { + PromExpr::MatrixSelector(MatrixSelector { vs, range }) => { + window_selection(context, function, vs, range) + } + PromExpr::Subquery(subquery) => subquery_window(context, name, function, subquery), + other => Err(not_implemented(&format!( + "a call to `{name}` whose matrix is `{other}`, where it has to be a range \ + selector like `x[5m]` or a subquery like `x[5m:1m]`" + ))), + } } /// Wraps `op` over the value of `vector`, which is the first argument of a math function. @@ -629,7 +690,7 @@ fn window_selection( let anchor_ms = selector_anchor_ms(&vs.at, context.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(context.stmt)?; + let output_grid = Arc::clone(&context.grid); // A range function reports one value per input field, so the layout is the input's. let output = NativeType::Samples(Arc::new(SeriesSchema { @@ -667,6 +728,115 @@ fn window_selection( }) } +/// The step a subquery names when it names none, as in `x[5m:]`. +/// +/// Prometheus falls back to the global evaluation interval, whose default is a minute, rather than to +/// the statement's own step. The existing planner folds the same default; the constant is restated +/// here because that one is private to the planner, and it is a Prometheus protocol constant rather +/// than a number either caller may choose. +const DEFAULT_SUBQUERY_STEP_MS: i64 = 60_000; + +/// Binds `function` over `subquery`, whose inner expression is evaluated on a grid of its own. +/// +/// A range function folds a matrix, and PromQL writes a matrix two ways. A range selector reads raw +/// samples from the scan; a subquery builds one by evaluating an expression. That is why this node +/// declares `input_grid`: `rate(x[5m:1m])` samples `x` at one-minute steps over the five-minute +/// lookback, and the window folds those points rather than raw samples. The inner expression is +/// compiled against the subquery's grid for the same reason — a selector inside it has to land where +/// the window will read it. +/// +/// An `@` anchor makes the subquery step-invariant: every output point reads the same interval, +/// `(anchor - offset - range, anchor - offset]`. The child grid is built for that interval, while +/// the output grid remains the statement's grid; the anchor, range and offset are bound together in +/// the window. +fn subquery_window( + context: &Context, + name: &str, + function: RangeFunction, + subquery: &SubqueryExpr, +) -> Result { + let SubqueryExpr { + expr, + range, + step, + offset, + at, + } = subquery; + // The anchor is resolved by the same function a range selector's is, so `start()`, `end()` and a + // literal — including a negative one — mean the same thing in both shapes. + let anchor_ms = selector_anchor_ms(at, context.stmt)?; + // A subquery evaluates an expression and hands the window one value per grid point per field, so + // the expression inside it has to be a vector: a scalar has no fields for the window to fold. + if expr.value_type() != ValueType::Vector { + return Err(not_implemented(&format!( + "a call to `{name}` over the subquery `{subquery}`, whose expression is a {}, where a \ + subquery has to evaluate a vector", + expr.value_type() + ))); + } + let range_ms = range_millis(range.as_millis())?; + // A subquery step of zero names no evaluation point at all, so it is reported as its own value + // rather than as an empty grid. + let step_ms = match step { + Some(step) => millis(step.as_millis()).ok_or_else(|| { + DataFusionError::Plan(format!( + "the subquery's step of {}ms is out of range", + step.as_millis() + )) + })?, + None => DEFAULT_SUBQUERY_STEP_MS, + }; + if step_ms <= 0 { + return Err(DataFusionError::Plan(format!( + "the subquery's step is {step_ms}ms, and a step has to be longer than zero" + ))); + } + let offset_ms = selector_offset_ms(offset)?; + let window = RangeWindow { + range_ms, + offset_ms, + anchor_ms, + }; + let child_spec = subquery_grid(context.grid.spec(), window, step_ms)?.ok_or_else(|| { + not_implemented(&format!( + "the subquery over {range_ms}ms at {step_ms}ms per step, which names no evaluation point \ + in its anchored interval or the statement's range" + )) + })?; + let child_grid = Arc::new(EvalGrid::try_deferred(child_spec).map_err(grid_error)?); + let inner = compile_expr( + &Context { + stmt: context.stmt, + input: Arc::clone(&context.input), + ordering: context.ordering, + grid: Arc::clone(&child_grid), + }, + expr, + )?; + + // A range function reports one value per input field, and the inner expression reports one per + // field of the same raw input, so the layout is the input's. + let output = NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::clone(&context.input.fields), + eval_grid: Some(Arc::clone(&context.grid)), + })); + + Ok(NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function, + window, + // The one field that tells this window from a raw one: its input was evaluated on the + // subquery's grid rather than read from the scan. + input_grid: Some(child_grid), + output_grid: Arc::clone(&context.grid), + }, + }, + output, + }) +} + /// 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 @@ -684,7 +854,7 @@ fn instant_selection(context: &Context, selector: &VectorSelector) -> Result Result> { {error}" )) })?; - Ok(Arc::new(EvalGrid::try_new(spec).map_err(grid_error)?)) + Ok(Arc::new(EvalGrid::try_deferred(spec).map_err(grid_error)?)) } /// The window length of a range selector, which has to be longer than zero to select anything. @@ -1064,14 +1234,18 @@ fn arity_error(name: &str, signature: MathSignature, given: usize) -> DataFusion /// case — has nowhere to go. It is refused rather than answered from the wrong number, and the /// rejection names the expression, because that is what the query has to change. /// +/// `name` is whoever reads the argument: a call's function name, or an aggregator's, since +/// `quantile(0.9, x)` reads a constant too. The rejection therefore names that and the position +/// rather than calling everything a call, which would be false for an aggregate. +/// /// A quantile outside `[0, 1]` is not refused here: Prometheus binds it and the kernel reports /// `-Infinity` below zero and `+Infinity` above one, so refusing it would be this compiler inventing /// a rule the query engine does not have. fn constant_argument(name: &str, argument: &PromExpr, position: usize) -> Result { constant_value(argument).ok_or_else(|| { not_implemented(&format!( - "a call to `{name}` whose {} is `{argument}`, where it has to be a constant expression \ - this compiler folds to a number literal", + "`{name}`, whose {} is `{argument}`, where it has to be a constant expression this \ + compiler folds to a number literal", argument_position(position) )) }) diff --git a/src/query/src/promql/native_compile/tests.rs b/src/query/src/promql/native_compile/tests.rs index 494021183a3..76697ce8633 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/src/query/src/promql/native_compile/tests.rs @@ -26,7 +26,8 @@ use promql::native::plan::{ use promql::range_eval::RangeFunction; use promql::vector_match::{BinaryMatching, BinaryOp}; use promql_parser::parser::{ - BinModifier, EvalStmt, Expr as PromExpr, NumberLiteral, VectorMatchFillValues, + AtModifier, BinModifier, EvalStmt, Expr as PromExpr, NumberLiteral, Offset, + VectorMatchFillValues, }; use super::compile; @@ -51,9 +52,17 @@ fn eval_stmt(query: &str) -> EvalStmt { /// 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 { + eval_stmt_between(query, Duration::ZERO, end) +} + +/// The statement `query` over `UNIX_EPOCH + start..=UNIX_EPOCH + end`, on a 5s grid. +/// +/// The start is a parameter because a grid that begins off the epoch is the only way to tell a +/// child grid built from absolute step boundaries from one built from the statement's start. +fn eval_stmt_between(query: &str, start: Duration, end: Duration) -> EvalStmt { EvalStmt { expr: promql_parser::parser::parse(query).unwrap(), - start: UNIX_EPOCH, + start: UNIX_EPOCH + start, end: UNIX_EPOCH + end, interval: Duration::from_secs(5), lookback_delta: Duration::from_secs(5), @@ -100,6 +109,15 @@ fn with_call_arguments(query: &str, edit: impl FnOnce(&mut Vec>)) stmt } +/// The bare selector `x`, as the parser builds it. +/// +/// The parser type-checks function arguments, so a vector selector where a matrix belongs can only be +/// handed to the compiler by editing a tree the parser accepted: this supplies one. +fn a_bare_selector() -> PromExpr { + let stmt = eval_stmt("x"); + stmt.expr +} + /// The range selector out of `rate(x[5m])`. /// /// The parser type-checks function arguments, so a range selector can only be handed to a function @@ -201,6 +219,21 @@ fn grid_of(node: &NativeNode) -> (i64, i64, u64) { (spec.start_ms, spec.step_ms, spec.len) } +/// A long valid statement exposes its checked grid without allocating query-sized timestamps. +#[test] +fn a_long_statement_grid_is_deferred() { + let stmt = EvalStmt { + expr: promql_parser::parser::parse("x").unwrap(), + start: UNIX_EPOCH, + end: UNIX_EPOCH + Duration::from_secs(86_400), + interval: Duration::from_millis(1), + lookback_delta: Duration::from_secs(5), + }; + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime).unwrap(); + + assert_eq!(grid_of(tree.root()), (0, 1, 86_400_001)); +} + /// `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() { @@ -880,24 +913,30 @@ fn an_interleaved_scan_is_refused_whatever_the_shape() { #[test] fn an_unsupported_shape_is_rejected_and_names_what_it_found() { let cases = [ - ("x[5m:]", "the expression `x[5m:]`"), - // Nested shapes are refused by the argument check, not skipped over. A subquery is the - // reachable case of an argument that is not a range selector: the parser type-checks - // function arguments, so `rate(x)` never gets past parsing. + // A bare subquery is a matrix, and a tree reports evaluated samples: only a range function + // can fold a matrix, so a subquery on its own has no node. + (eval_stmt("x[5m:]"), "x[5m:]", "the expression `x[5m:]`"), + // A matrix position that is neither a range selector nor a subquery: a bare vector selector + // where a matrix belongs. The parser type-checks function arguments, so this can only reach + // the compiler as an edited tree. ( - "rate(x[5m:])", - "a call to `rate` whose range selector is `x[5m:]`", + with_call_arguments("rate(x[5m])", |args| { + *args[0] = a_bare_selector(); + }), + "rate(x)", + "whose matrix is `x`", ), // A function with no kernel. `predict_linear` used to stand here as the parameterised case // and is now bound, so what is left is a name this compiler has no kernel for at all. ( + eval_stmt("absent(x)"), "absent(x)", "a call to `absent`, which has no native kernel", ), ]; - for (query, named) in cases { - let error = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) + for (stmt, query, named) in cases { + let error = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) .expect_err(&format!("`{query}` compiled, and it should not have")); assert!( matches!(error, DataFusionError::NotImplemented(_)), @@ -2484,3 +2523,722 @@ fn a_parameterised_window_nests_under_a_comparison_and_an_aggregate() { }; assert_eq!(op.function, RangeFunction::PredictLinear(300.0)); } + +/// The grid a subquery evaluates its inner expression on, as `(start_ms, step_ms, len)`. +fn child_grid(window: &BoundWindow) -> (i64, i64, u64) { + let grid = window + .input_grid + .as_ref() + .expect("a subquery's window declares the grid its input was evaluated on"); + let spec = grid.spec(); + (spec.start_ms, spec.step_ms, spec.len) +} + +/// `rate(x[5m:1m])` folds a subquery rather than a range selector: the inner selector is evaluated on +/// the subquery's own grid, and the window declares that grid as its input's. The two grids are +/// asserted apart, because a compiler that bound the child onto the statement's grid would still +/// produce a window. +#[test] +fn a_subquery_feeds_a_window_evaluated_on_its_own_grid() { + let tree = compile( + &eval_stmt("rate(x[5m:1m])"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .unwrap_or_else(|error| panic!("`rate(x[5m:1m])` was refused: {error}")); + + let (input, op) = root_window(&tree); + assert_eq!(op.function, RangeFunction::Rate); + assert_eq!(op.window.range_ms, 300_000, "the subquery range was lost"); + assert_eq!(op.window.offset_ms, 0); + assert_eq!(op.window.anchor_ms, None); + + // The output lands on 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 inner selector is evaluated at one-minute steps, and the earliest read window opens at + // `0 - 300000`, so the first step boundary strictly after it is `-240000`. + assert_eq!( + child_grid(op), + (-240_000, 60_000, 6), + "the subquery's own grid is wrong" + ); + // The input is that evaluation, not raw samples: this is the field that tells a subquery's window + // from one over a range selector. + let inner = evaluated(input); + assert_eq!( + inner.eval_grid.as_ref().unwrap().spec().start_ms, + -240_000, + "the inner expression was not evaluated on the subquery's grid" + ); + + // Below the window's input sits the one raw scan, so the tree still reads the scan exactly once. + let NativeExpr::Instant { input: scan, .. } = &input.expr else { + panic!( + "the subquery's input is not an instant selection: {:?}", + input.expr + ); + }; + let NativeExpr::Input(bound) = &scan.expr else { + panic!("the innermost node is not an input: {:?}", scan.expr); + }; + assert!( + matches!(&bound.output, NativeType::Samples(schema) if schema.eval_grid.is_none()), + "the subquery's innermost input is not raw samples: {:?}", + bound.output + ); +} + +/// The child grid is derived from the statement's grid, the range, the offset and the step, and each +/// of those changes it. The cases are asserted as whole specs, because a formula that dropped one +/// term would still produce a plausible grid. +#[test] +fn a_subquery_grid_follows_its_range_step_and_offset() { + let cases = [ + // (query, start_ms, step_ms, len) + // The plain case: one minute steps, five minute windows, no offset. + ("rate(x[5m:1m])", -240_000, 60_000, 6), + // The statement's own step is not the subquery's step. + ("rate(x[5m:5s])", -295_000, 5_000, 80), + // A step that does not divide the range, so the first boundary is not `-range`, and the last + // point is the last multiple of the step rather than the statement's end. + ("rate(x[5m:7s])", -294_000, 7_000, 57), + // A positive offset reads earlier samples, so the whole child grid moves back. + ("rate(x[5m:1m] offset 1m)", -300_000, 60_000, 6), + // A negative offset reads later ones, so it moves forward. + ("rate(x[5m:1m] offset -1m)", -180_000, 60_000, 6), + ]; + + for (query, start_ms, step_ms, len) in cases { + let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime) + .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); + let (_, op) = root_window(&tree); + assert_eq!( + child_grid(op), + (start_ms, step_ms, len), + "`{query}` bound the wrong child grid" + ); + // Whatever the child grid is, the window still reports the statement's. + let spec = op.output_grid.spec(); + assert_eq!( + (spec.start_ms, spec.step_ms, spec.len), + (0, 5_000, 21), + "`{query}` moved the output off the statement's grid" + ); + } +} + +/// A subquery whose child interval contains no step boundary is refused as unsupported. +#[test] +fn a_subquery_with_an_empty_child_grid_is_refused() { + let error = compile( + &eval_stmt("rate(x[1ms:1m] @ 0.001)"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .expect_err("an empty subquery child grid compiled"); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "an empty child grid is an unsupported subquery shape: {error}" + ); + assert!( + error.to_string().contains("names no evaluation point"), + "the refusal did not identify the empty child grid: {error}" + ); +} + +/// Large subquery child grids compile as deferred grids without allocating one timestamp per point. +#[test] +fn large_subquery_child_grids_compile_deferred() { + const DAY_STEP_GRID: (i64, i64, u64) = (-86_399_999, 1, 86_400_000); + let cases = [ + ("rate(x[1d:1ms] @ 0)", false), + ("rate(rate(x[1d:1ms] @ 0)[1d:1ms] @ 0)", true), + ]; + + for (query, nested) in cases { + let mut stmt = eval_stmt(query); + stmt.end = UNIX_EPOCH; + stmt.interval = Duration::from_millis(1); + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); + let (input, outer) = root_window(&tree); + assert_eq!(child_grid(outer), DAY_STEP_GRID, "`{query}` outer grid"); + if nested { + let NativeExpr::Window { op: inner, .. } = &input.expr else { + panic!( + "`{query}` inner expression is not a window: {:?}", + input.expr + ); + }; + assert_eq!(child_grid(inner), DAY_STEP_GRID, "`{query}` inner grid"); + } + } + + // This range makes the eager timestamp vector's capacity exceed isize::MAX bytes, while its + // child GridSpec remains valid and the deferred compiler path should require no such allocation. + let range_ms = (isize::MAX / 8 + 1) as u64; + let mut stmt = eval_stmt("rate(x[1ms:1ms] @ 0)"); + stmt.end = UNIX_EPOCH; + stmt.interval = Duration::from_millis(1); + let PromExpr::Call(call) = &mut stmt.expr else { + unreachable!("query parses to a call"); + }; + let PromExpr::Subquery(subquery) = &mut *call.args.args[0] else { + unreachable!("call argument parses to a subquery"); + }; + subquery.range = Duration::from_millis(range_ms); + + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .expect("a huge but valid deferred child grid should compile"); + let (_, window) = root_window(&tree); + let grid = child_grid(window); + assert_eq!(grid.0, -(range_ms as i64) + 1); + assert_eq!(grid.1, 1); + assert_eq!(grid.2, range_ms); +} + +/// A sampled child endpoint can fit even when the unrounded read-window endpoint does not. +#[test] +fn a_subquery_accepts_an_in_range_sampled_endpoint() { + let mut stmt = eval_stmt("rate(x[1s:1s] offset -1ms)"); + let endpoint = UNIX_EPOCH + Duration::from_millis(i64::MAX as u64); + stmt.start = endpoint; + stmt.end = endpoint; + stmt.interval = Duration::from_millis(1); + + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .expect("the in-range sampled endpoint should compile"); + let (_, op) = root_window(&tree); + let spec = op.input_grid.as_ref().unwrap().spec(); + assert_eq!( + (spec.start_ms, spec.step_ms, spec.len), + (9_223_372_036_854_775_000, 1_000, 1), + "the child grid should contain the last in-range 1-second sample" + ); +} + +/// An anchored fold offset that cannot be represented as i64 is a plan error. +#[test] +fn a_subquery_with_an_overflowing_fold_offset_is_refused() { + let mut stmt = eval_stmt("rate(x[5m:1m] @ 0)"); + let PromExpr::Call(call) = &mut stmt.expr else { + unreachable!("query parses to a call"); + }; + let PromExpr::Subquery(subquery) = &mut *call.args.args[0] else { + unreachable!("call argument parses to a subquery"); + }; + subquery.at = Some(AtModifier::At( + UNIX_EPOCH + Duration::from_millis(i64::MAX as u64), + )); + subquery.offset = Some(Offset::Neg(Duration::from_millis(1))); + + let error = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .expect_err("an overflowing subquery fold offset compiled"); + assert!( + matches!(error, DataFusionError::Plan(_)), + "an overflowing folded offset is a plan error: {error}" + ); + assert!( + error.to_string().contains("out of range"), + "the rejection did not identify the offset overflow: {error}" + ); +} + +/// A subquery step that names no point is refused: `x[5m:]` falls back to Prometheus's default +/// evaluation interval of one minute, and a step of zero has no point at all. The rejection names the +/// number rather than reporting an empty grid. +#[test] +fn a_subquery_step_must_be_positive() { + // The parser refuses `x[5m:0s]`, so a zero step can only be handed to the compiler as an edited + // tree — the same way the range and arity tests reach their shapes. + let mut stmt = eval_stmt("rate(x[5m:1m])"); + let PromExpr::Call(call) = &mut stmt.expr else { + unreachable!("`rate(x[5m:1m])` parses to a call"); + }; + let PromExpr::Subquery(subquery) = &mut *call.args.args[0] else { + unreachable!("`x[5m:1m]` parses to a subquery"); + }; + subquery.step = Some(Duration::ZERO); + + let error = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .expect_err("a zero subquery step compiled, and it should not have"); + assert!( + matches!(error, DataFusionError::Plan(_)), + "a zero step is a plan error, not a shape this compiler lacks: {error}" + ); + assert!( + error.to_string().contains("has to be longer than zero"), + "the rejection did not say what a step has to be: {error}" + ); +} + +/// A subquery with no step falls back to Prometheus's default evaluation interval of one minute, +/// which is not the statement's step: the statement here steps in five seconds, so a compiler that +/// reused it would produce a grid five times longer. +#[test] +fn a_subquery_without_a_step_uses_the_default_interval() { + let tree = compile( + &eval_stmt("rate(x[5m:])"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .unwrap_or_else(|error| panic!("`rate(x[5m:])` was refused: {error}")); + + let (_, op) = root_window(&tree); + assert_eq!(op.window.range_ms, 300_000); + assert_eq!( + child_grid(op).1, + 60_000, + "a subquery without a step did not use the default interval" + ); +} + +/// The child grid an anchored subquery must declare, derived from its read interval rather than from +/// the compiler's formula: the samples of `(anchor - offset - range, anchor - offset]` at `step`. +/// +/// The multiples are found by walking a two-step-wide margin around the interval and keeping the ones +/// the interval itself contains, so no division and no rounding convention appears here: a wrong +/// formula in the compiler cannot be reproduced by this helper, which is the whole point of it. +fn anchored_child_grid( + anchor_ms: i64, + range_ms: i64, + step_ms: i64, + offset_ms: i64, +) -> (i64, i64, u64) { + let end = anchor_ms - offset_ms; + let start = end - range_ms; + let mut first = None; + let mut len = 0_u64; + for k in (start / step_ms - 2)..=(end / step_ms + 2) { + let sample = k * step_ms; + if sample <= start || sample > end { + continue; + } + first.get_or_insert(sample); + len += 1; + } + ( + first.expect("the read interval names no sample at all"), + step_ms, + len, + ) +} + +/// An anchored subquery reads one fixed interval instead of one interval per output point, and the +/// interval is the anchor's: `(anchor - offset - range, anchor - offset]` at the subquery's step. +/// +/// The expected grid comes from the interval, not from the compiler, and the anchor itself is stated +/// in the case table even where the query writes `start()` or `end()`, so the two have to agree. +#[test] +fn an_anchored_subquery_reads_one_fixed_interval() { + // (query, anchor_ms, range_ms, step_ms, offset_ms, statement start_ms, statement end_ms) + let cases = [ + ( + "rate(x[5m:1m] @ 1234)", + 1_234_000, + 300_000, + 60_000, + 0, + 0, + 100_000, + ), + // `start()` and `end()` are the statement's own bounds, not points of its grid. + ( + "rate(x[5m:1m] @ start())", + 0, + 300_000, + 60_000, + 0, + 0, + 100_000, + ), + ( + "rate(x[5m:1m] @ end())", + 100_000, + 300_000, + 60_000, + 0, + 0, + 100_000, + ), + // A pre-epoch anchor, which a `duration_since(UNIX_EPOCH).unwrap_or_default()` would make 0. + ( + "rate(x[5m:1m] @ -1.5)", + -1_500, + 300_000, + 60_000, + 0, + 0, + 100_000, + ), + // The subquery's own offset moves the anchored interval, as it moves an unanchored one, in + // either order: the parser allows `@` and `offset` to be written both ways round. + ( + "rate(x[5m:1m] offset 1m @ 1234)", + 1_234_000, + 300_000, + 60_000, + 60_000, + 0, + 100_000, + ), + ( + "rate(x[5m:1m] @ 1234 offset 1m)", + 1_234_000, + 300_000, + 60_000, + 60_000, + 0, + 100_000, + ), + // A step that divides neither the range nor the anchor, so a grid rounded to either would + // name different samples. + ( + "rate(x[5m:7s] @ 1234)", + 1_234_000, + 300_000, + 7_000, + 0, + 0, + 100_000, + ), + // A statement whose end is not a grid point and whose start is off the child step, so neither + // `end()` nor the statement's start can stand in for the anchor or for the child grid. + ( + "rate(x[5m:1m] @ end())", + 103_000, + 300_000, + 60_000, + 0, + 3_000, + 103_000, + ), + ]; + + for (query, anchor_ms, range_ms, step_ms, offset_ms, start_ms, end_ms) in cases { + let stmt = eval_stmt_between( + query, + Duration::from_millis(start_ms as u64), + Duration::from_millis(end_ms as u64), + ); + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); + let (input, op) = root_window(&tree); + + assert_eq!( + op.window.anchor_ms, + Some(anchor_ms), + "`{query}` bound the wrong anchor" + ); + assert_eq!(op.window.range_ms, range_ms, "`{query}`: wrong range"); + assert_eq!(op.window.offset_ms, offset_ms, "`{query}`: wrong offset"); + + let expected = anchored_child_grid(anchor_ms, range_ms, step_ms, offset_ms); + assert_eq!( + child_grid(op), + expected, + "`{query}` bound the wrong child grid: the interval \ + ({}, {}] at {step_ms}ms per step holds {expected:?}", + anchor_ms - offset_ms - range_ms, + anchor_ms - offset_ms + ); + // The inner expression lands on that grid, so the window reads the points it declares. + let inner = evaluated(input); + assert_eq!( + inner.eval_grid.as_ref().unwrap().spec().start_ms, + expected.0, + "`{query}` evaluated its expression off the declared child grid" + ); + // The statement's grid is untouched: an anchor changes what is read, not what is reported. + let out = op.output_grid.spec(); + assert_eq!( + (out.start_ms, out.step_ms), + (start_ms, 5_000), + "`{query}` moved the output off the statement's grid" + ); + assert_eq!( + out.len as i64, + (end_ms - start_ms) / 5_000 + 1, + "`{query}` changed the statement's grid length" + ); + } +} + +/// An anchored child grid does not depend on the statement at all: the same anchored subquery in two +/// statements of different spans reads the same interval, while the two report on their own grids. +#[test] +fn an_anchored_subquery_grid_ignores_the_statement() { + let short = eval_stmt_between( + "rate(x[5m:1m] @ 1234)", + Duration::ZERO, + Duration::from_secs(100), + ); + let long = eval_stmt_between( + "rate(x[5m:1m] @ 1234)", + Duration::from_secs(10), + Duration::from_secs(600), + ); + + let mut grids = Vec::new(); + for stmt in [&short, &long] { + let tree = compile(stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .unwrap_or_else(|error| panic!("an anchored subquery was refused: {error}")); + let (_, op) = root_window(&tree); + grids.push((child_grid(op), grid_of(tree.root()))); + } + + let expected_child = anchored_child_grid(1_234_000, 300_000, 60_000, 0); + assert_eq!( + grids[0].0, expected_child, + "the short statement's child grid is wrong" + ); + assert_eq!( + grids[1].0, expected_child, + "the child grid moved with the statement, so the anchor is not what it reads" + ); + assert_ne!( + grids[0].1, grids[1].1, + "the two statements have the same output grid, so this case proves nothing" + ); +} + +/// A window folds one value per grid point, so the expression inside a subquery has to produce a +/// vector. A scalar has no fields to fold and no points at all, so it is refused by type. +#[test] +fn a_subquery_over_a_scalar_is_refused() { + let mut stmt = eval_stmt("rate(x[5m:1m])"); + let PromExpr::Call(call) = &mut stmt.expr else { + unreachable!("`rate(x[5m:1m])` parses to a call"); + }; + let PromExpr::Subquery(subquery) = &mut *call.args.args[0] else { + unreachable!("`x[5m:1m]` parses to a subquery"); + }; + *subquery.expr = PromExpr::NumberLiteral(NumberLiteral::new(1.0)); + + let error = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .expect_err("a subquery over a scalar compiled, and it should not have"); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "the refusal is not NotImplemented: {error}" + ); + assert!( + error.to_string().contains("has to evaluate a vector"), + "the refusal did not say what a subquery evaluates: {error}" + ); +} + +/// A subquery nests like any other matrix: a window over a subquery can itself sit inside a subquery, +/// and the whole chain still reads the scan once. The grids are asserted level by level, because a +/// compiler that passed the outer grid down would produce a tree that validates but reads the wrong +/// points. +#[test] +fn a_subquery_nests_and_each_level_has_its_own_grid() { + let tree = compile( + &eval_stmt("rate(rate(x[5m:1m])[10m:30s])"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .unwrap_or_else(|error| panic!("`rate(rate(x[5m:1m])[10m:30s])` was refused: {error}")); + + let (outer_input, outer) = root_window(&tree); + assert_eq!(outer.window.range_ms, 600_000); + // The outer subquery reads ten minutes back at thirty-second steps, so its first boundary after + // `-600000` is `-570000` and its last multiple of 30000 below the statement's end is `90000`. + assert_eq!( + child_grid(outer), + (-570_000, 30_000, 23), + "the outer subquery's grid is wrong" + ); + + let NativeExpr::Window { + input: inner_input, + op: inner, + } = &outer_input.expr + else { + panic!( + "the outer subquery's input is not a window: {:?}", + outer_input.expr + ); + }; + assert_eq!(inner.function, RangeFunction::Rate); + assert_eq!(inner.window.range_ms, 300_000); + // The inner subquery is evaluated on the outer one's grid, not on the statement's: its earliest + // read window opens at `-570000 - 300000`, whose first one-minute boundary is `-840000`. + assert_eq!( + child_grid(inner), + (-840_000, 60_000, 16), + "the inner subquery's grid is wrong, so the outer grid was not passed down" + ); + + // The innermost node is still the one raw scan. + let NativeExpr::Instant { input: scan, .. } = &inner_input.expr else { + panic!( + "the inner window's input is not an instant selection: {:?}", + inner_input.expr + ); + }; + assert!( + matches!(scan.expr, NativeExpr::Input(_)), + "the innermost node is not the scan: {:?}", + scan.expr + ); +} + +/// A subquery composes with the rest of the compiler: a sum over a rate over a subquery is a tree, +/// not a shape the compiler knows by name. +#[test] +fn a_subquery_composes_with_an_aggregate_and_a_comparison() { + let tree = compile( + &eval_stmt("sum by(job)(rate(x[5m:1m]) > 0)"), + raw_schema(), + SeriesOrdering::KeyThenTime, + ) + .unwrap_or_else(|error| panic!("`sum by(job)(rate(x[5m:1m]) > 0)` was refused: {error}")); + + let (input, aggregate) = aggregate_parts(tree.root()); + assert_eq!(aggregate.kind, AggregateKind::Sum); + let (left, _, matching) = binary_parts(input); + assert_eq!(matching.op, BinaryOp::Gt); + let NativeExpr::Window { op, .. } = &left.expr else { + panic!( + "the comparison's left operand is not a window: {:?}", + left.expr + ); + }; + assert_eq!(op.window.range_ms, 300_000); + assert!( + op.input_grid.is_some(), + "the window over a subquery forgot its input grid" + ); +} + +/// The grid a subquery declares has to hold every sample its windows read — and no sample none of +/// them can read. Both halves are checked here, against an oracle that is the query's own numbers +/// rather than the compiler's. +/// +/// Nothing but the grid itself is read from the produced window: the range, the step and the offset +/// come from the query text (stated again in the case table, so the two can disagree), the read +/// windows come from the statement's output grid as `p - offset - range < t <= p - offset`, and the +/// samples are every multiple of the step in such a window. The membership test is the modulo one, +/// because a grid whose step or phase is wrong still spans the right two bounds: interval +/// containment is what let a wrong step pass before. +#[test] +fn a_subquery_grid_holds_every_sample_its_windows_read() { + // The statement's end: the start is a column, because a grid that begins off the epoch is the + // only way to see a child grid built from the statement's start rather than from the epoch. + let end = Duration::from_secs(100); + + // (query, statement start, range_ms, step_ms, offset_ms, some point reads no sample) + // + // The last flag is asserted rather than assumed: the narrow case below passes trivially if the + // compiler refused to bind a window at all, so the test would not notice losing it. + let cases = [ + // A whole-minute step over a five-minute window. + ("rate(x[5m:1m])", 0, 300_000, 60_000, 0, false), + // A step that divides neither the range nor the statement's 5s step. + ("rate(x[5m:7s])", 0, 300_000, 7_000, 0, false), + // A step the statement's own step does divide, so the two cannot be confused either. + ("rate(x[5m:5s])", 0, 300_000, 5_000, 0, false), + // A step that divides neither the range nor the offset. + ( + "rate(x[90s:40s] offset 30s)", + 0, + 90_000, + 40_000, + 30_000, + false, + ), + // Offsets in both directions move every read window. + ( + "rate(x[5m:1m] offset 1m)", + 0, + 300_000, + 60_000, + 60_000, + false, + ), + ( + "rate(x[5m:1m] offset -1m)", + 0, + 300_000, + 60_000, + -60_000, + false, + ), + // A window narrower than the child step: some output points read no sample at all, so a + // grid that dropped the samples the other points read would still look plausible. + ("rate(x[30s:1m])", 0, 30_000, 60_000, 0, true), + // A statement whose start is three seconds off the child step, so a grid stepped from the + // statement's start would be phase-shifted by exactly that much. + ("rate(x[5m:1m])", 3_000, 300_000, 60_000, 0, false), + ("rate(x[30s:1m])", 3_000, 30_000, 60_000, 0, true), + ]; + + for (query, start, range_ms, step_ms, offset_ms, some_points_read_nothing) in cases { + let stmt = eval_stmt_between(query, Duration::from_millis(start as u64), end); + let tree = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime) + .unwrap_or_else(|error| panic!("`{query}` was refused: {error}")); + let (_, op) = root_window(&tree); + + // The grid under test, and the statement's own grid, which is not the subquery's. + let (child_start, child_step, child_len) = child_grid(op); + let child_last = child_start + (child_len as i64 - 1) * child_step; + let out = op.output_grid.spec(); + let out_last = out.start_ms + (out.len as i64 - 1) * out.step_ms; + + // Every sample a window can read lies in the union of the read windows: strictly after the + // earliest window's left end, and no later than the latest window's right one. A grid that + // reached before the earliest read would evaluate a sample no window folds. + let earliest = out.start_ms - offset_ms - range_ms; + let latest = out_last - offset_ms; + for index in 0..child_len as i64 { + let point = child_start + index * child_step; + assert!( + point > earliest && point <= latest, + "`{query}`: the child grid point {point}ms lies outside the read windows \ + {earliest}ms..={latest}ms, where no window can read it" + ); + } + + let mut every_point_reads = true; + let mut samples_checked = 0_usize; + for index in 0..out.len as i64 { + let point = out.start_ms + index * out.step_ms; + let read_start = point - offset_ms - range_ms; + let read_end = point - offset_ms; + // Every multiple of the step in this window, enumerated two steps wider on each side + // and filtered by the window itself: the slack absorbs the difference between + // truncating and floor division, so this enumeration assumes no rounding convention. + let mut read_anything = false; + for k in (read_start / step_ms - 2)..=(read_end / step_ms + 2) { + let sample = k * step_ms; + if sample <= read_start || sample > read_end { + continue; + } + read_anything = true; + samples_checked += 1; + assert!( + sample >= child_start + && sample <= child_last + && (sample - child_start) % child_step == 0, + "`{query}`: the window at {point}ms reads {sample}ms, which is not a point of \ + the declared child grid {child_start}ms..={child_last}ms at {child_step}ms per \ + step" + ); + } + every_point_reads &= read_anything; + } + + assert!( + samples_checked > 0, + "`{query}`: no point of the statement's grid reads a sample, so this case checks nothing" + ); + assert_eq!( + every_point_reads, !some_points_read_nothing, + "`{query}`: the case table says some point reads no sample, so every point reading one \ + (or vice versa) means the window is not the one the query wrote" + ); + } +}