feat(promql): compile subqueries with deferred grid materialization (#9835)

Compile range functions over explicit/default-step, offset, anchored, and nested subqueries using the shared range-evaluator grid calculation. Retain complete sampled coverage and reject empty child grids explicitly.

Keep compilation allocation proportional to plan structure, not timestamp count: validate deferred root and child grids; preserve eager EvalGrid::try_new; materialize shared timestamps fallibly only when a consumer requests them. Migrate Rust timestamp-buffer callers to Result without changing wire or persisted formats.

Regression coverage includes grid phase/coverage, rounded sampled endpoints, offset overflow, huge and nested compile-only grids, sparse gathering, sharing, and allocation-error propagation. Deferred compilation does not bound execution memory. No public Fused entrypoint or Scalar runtime is added here.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

Cargo.lock

@@ -4559,6 +4559,7 @@ dependencies = [
  "jsonb",
  "num",
  "num-traits",
+ "once_cell",
  "ordered-float 4.6.0",
  "parquet-variant",
  "parquet-variant-compute",

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

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<i64>,
+    timestamps: OnceCell<ScalarBuffer<i64>>,
 }
 
 impl EvalGrid {
     /// Allocates the grid timestamps once for sharing across result series.
     pub fn try_new(spec: GridSpec) -> Result<Self> {
-        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<Self> {
+        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<i64> {
-        &self.timestamps
+    /// Returns the shared timestamp buffer, materializing it on first access.
+    pub fn timestamps(&self) -> Result<&ScalarBuffer<i64>> {
+        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<i64> {
+        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<i64> {
-        self.grid.timestamps.slice(self.start, self.len)
+    pub fn timestamps(&self) -> Result<ScalarBuffer<i64>> {
+        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<EvalGrid>, spans: Vec<GridSpan>) -> Result<Self> {
-        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<i64> {
         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<Cow<'_, [i64]>> {
         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(),
-            ),
+            )),
         }
     }
 

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::<i64>() 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<usize> = 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::<i64>() 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());

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 {:?}",

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<Option<f64>> = (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) {

src/promql/src/golden.rs

@@ -254,7 +254,7 @@ fn fold_one(pieces: Vec<Series>) -> Series {
         None => panic!("expected a grid series, got {:?}", pieces[0].time),
     };
     let mut rows: Vec<Option<Vec<Option<f64>>>> =
-        (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

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;

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<Vec<Option<u64>>> {
         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()

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"
         );
     }

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()

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());

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)];

src/promql/src/map/tests/prometheus.rs

@@ -174,7 +174,7 @@ fn evaluate(case: &Case) -> Vec<Series> {
 }
 
 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]

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(

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() => {}

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,
         }
     }

src/promql/src/range_eval/tests.rs

@@ -279,7 +279,7 @@ fn fold_chunks(chunks: &[EvaluatedChunk], schema: &Arc<SeriesSchema>) -> 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<Option<Vec<Option<GridSample>>>> = (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<Range<usize>> {
     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<Series>, Vec<Row>) {
-    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<i64> = ((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<f64> = 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);

src/promql/src/select.rs

@@ -610,7 +610,7 @@ fn grid_points(schema: &SeriesSchema, series: &[Series]) -> Result<usize> {
             ));
         }
     }
-    Ok(grid.timestamps().len())
+    Ok(grid.spec().len as usize)
 }
 
 /// The first field's value of `series` at grid `point`, looked up by position.

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);

src/promql/src/vector_match/tests.rs

@@ -130,7 +130,7 @@ fn random_side(rng: &mut StdRng, grid: &Arc<EvalGrid>, 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,

src/query/src/promql/native_compile.rs

@@ -19,37 +19,64 @@
 //! accepts is one recursive rule:
 //!
 //! ```text
-//! <selector>                                        e.g. x
-//! <range function>(<selector>[<range>])             e.g. rate(x[5m])
-//! <range function>(<constant>, <selector>[<range>])   e.g. quantile_over_time(0.9, x[5m])
-//! <range function>(<selector>[<range>], <constant>[, <constant>])
-//!                                                   e.g. predict_linear(x[5m], 1),
-//!                                                        double_exponential_smoothing(x[5m], 0.5, 0.5)
-//! <math function>(<expression>[, <constant>]*)      e.g. abs(rate(x[5m])), clamp_min(x, 0)
-//! <expression> <operator> <constant>                e.g. x * 100, x > bool 0
-//! <constant> <operator> <expression>                e.g. 1 / x, 0 < x
-//! -<expression>                                     e.g. -x
-//! <aggregator> [(by|without)(<labels>)] (<expression>)
-//!                                                   e.g. avg by(job)(rate(x[5m]) * 100)
-//! <aggregator> (<expression>) [(by|without)(<labels>)]
-//!                                                   e.g. count (x) by(job)
-//! quantile(<constant>, <expression>) [(by|without)(<labels>)]
-//!                                                   e.g. quantile(0.9, x) by(job)
+//! <expression>                                     one of, recursively:
+//!   <selector>                                     e.g. x, x offset -5m
+//!   <range function>(<matrix>)                     e.g. rate(x[5m]), rate(x[5m:1m] @ 1234)
+//!   quantile_over_time(<constant>, <matrix>)       e.g. quantile_over_time(0.9, x[5m])
+//!   predict_linear(<matrix>, <constant>)           e.g. predict_linear(x[5m], 300)
+//!   double_exponential_smoothing(<matrix>, <constant>, <constant>)
+//!                                                  e.g. double_exponential_smoothing(x[5m], .5, .1)
+//!   <math function>(<expression>[, <constant>]*)   e.g. abs(rate(x[5m])), clamp_min(x, 0)
+//!   <expression> <operator> <constant>             e.g. x * 100, x > bool 0
+//!   <constant> <operator> <expression>             e.g. 1 / x, 0 < x
+//!   -<expression>                                  e.g. -x
+//!   <aggregator> [(by|without)(<labels>)] (<expression>)
+//!                                                  e.g. avg by(job)(rate(x[5m]) * 100)
+//!   <aggregator> (<expression>) [(by|without)(<labels>)]
+//!                                                  e.g. count (x) by(job)
+//!   quantile(<constant>, <expression>) [(by|without)(<labels>)]
+//!                                                  e.g. quantile(0.9, x) by(job)
+//!
+//! <matrix>    <selector>[<range>]<modifiers>       a range selector: raw samples
+//!           | <expression>[<range>:<step>]<modifiers>
+//!                                                  a subquery: <expression> evaluated on a grid;
+//!                                                  the step may be omitted as `[<range>:]`
+//! <modifiers>                                      nothing, `offset <duration>`, `@ <anchor>`, or
+//!                                                  both, in either order — e.g. `x[5m] @ 1234
+//!                                                  offset 1m`
 //! ```
 //!
-//! where `<aggregator>` is `sum`, `avg`, `count`, `min`, `max`, `group`, `stddev` or `stdvar`.
+//! where `<aggregator>` 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 `<expression>` 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 `<step>` 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 `<selector>`
+//! outside a `<matrix>` 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),
+//! `<expression>` → `Window` whose input is evaluated (the subquery, whose input grid is its own),
 //! `<expression>` → `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 — \
-                         `<range function>(<selector>[<range>])`, 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(<constant>, ...)` — 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 `<selector>[<range>]` as in `rate(x[5m])` or \
+                         `<expression>[<range>:<step>]` as in `rate(x[5m:1m])`, with the step \
+                         optionally omitted as `[<range>:]`, where the subquery's `<expression>` \
+                         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(<constant>, <expression>)` — 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<SeriesSchema>,
     /// 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<EvalGrid>,
 }
 
 /// 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<NativeNode> {
     // 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<NativeNode> {
     }
 }
 
-/// 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<NativeNode> {
-    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<NativeNode> {
     // 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<NativeNode> {
     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<NativeNode> {
+    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<Nat
     }
     let lookback_ms = lookback_millis(context.stmt.lookback_delta.as_millis())?;
     let offset_ms = selector_offset_ms(&selector.offset)?;
-    let output_grid = statement_grid(context.stmt)?;
+    let output_grid = Arc::clone(&context.grid);
 
     // An instant selection reports one value per input field, so the layout is the input's.
     let output = NativeType::Samples(Arc::new(SeriesSchema {
@@ -735,7 +905,7 @@ fn statement_grid(stmt: &EvalStmt) -> Result<Arc<EvalGrid>> {
              {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<f64> {
     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)
         ))
     })

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<Box<PromExpr>>))
     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"
+        );
+    }
+}