From ee4c06d49815bfc60dbe67404c50cd0217ccce45 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 21 Oct 2026 16:57:22 +0800 Subject: [PATCH] test(promql): hold the batch path to the per-window one on random input (#9739) * test(promql): hold the batch path to the per-window one on random input Two comments claimed the opposite of what the code relies on: a batch's spans were described as absolute indices that survive a release, but a span is recorded against the retained arrays as they stand, and `advance` drains and shifts them. Nothing enforces the difference -- the batch is correct only because `evaluate` flushes it before advancing, so a later change that let a batch outlive one call would return wrong counter values with no error. Both now say that, and `PendingBatch::spans` carries the invariant where a reader looks for it. The batch path had no randomized comparison at all: the differential tests drive a probe kernel and the function tests cover no counter function, so only fixed cases exercised it. A new differential test runs rate and increase over 200 seeds of one to three fields, three series through one evaluator, random chunk splits and mixed histogram samples, comparing the batch path against the same evaluator with its batch form removed. Two assertions that could only have failed on a broken helper are gone, and the counter-reset flag is documented as governing both readers of the index, not just the per-window one. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): assert the batch is flushed before the sample drain `PendingBatch.spans` index the retained sample arrays, which `FieldSamples::advance` drains and shifts. The batch is correct only because `evaluate` flushes it just above that drain, and nothing enforced the ordering: a future change that let a batch outlive the call would compute counter values against shifted samples with no error. The assertion holds on every tested batch path. It is skipped when no batch kernel is installed, since `pending` is only ever filled on the kernel branch. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(promql): state the flush-before-advance rule once, and assert it plainly The batch assert could not fail on its `batch.is_none()` disjunct: `accumulate` is reached only from inside `if let Some(kernel) = self.batch.as_mut()`, and `batch` is never cleared, so a non-empty `pending` implies a kernel. Drop the disjunct and name who guarantees the order in the message instead. The rule that a batch must flush before `advance` was restated in six places. Keep it on `PendingBatch::spans`, where a reader asks why a span dies with the batch, and in the assertion message. The other four now state only the site-specific fact each needs. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(promql): drop the last restatement of flush-before-advance The rule now lives only on PendingBatch::spans and in the debug_assert message, as the review asked; the comment above the final flush restated it a third time. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/promql/src/range_eval.rs | 24 +-- src/promql/src/range_eval/function.rs | 5 + src/promql/src/range_eval/tests.rs | 210 ++++++++++++++++++++++---- 3 files changed, 201 insertions(+), 38 deletions(-) diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index 428f9f0f9af..1564fec0d0a 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -80,7 +80,8 @@ pub(crate) struct FloatWindows<'a> { pub resets: &'a [CounterReset], /// Absolute position of `timestamps[0]`: a reset position is `base` plus a sample index. pub base: u64, - /// One range per window, ascending and left-open at the start. + /// One range per window, ascending. Each is a half-open index range `[start, end)`; the time + /// window it was derived from is left-open, `(T - range_ms, T]`. pub windows: &'a [Range], /// The evaluation timestamp of each window. pub eval_ts: &'a [i64], @@ -90,8 +91,9 @@ pub(crate) struct FloatWindows<'a> { impl FloatWindows<'_> { /// The resets inside window `index`. /// - /// `windows` are absolute indices into `timestamps`, so a window whose start has already - /// been released still maps to the right reset. + /// `windows` index the retained arrays as they stand in this batch, not absolutely: each + /// span is relative to `timestamps` and `values`, which is why `base` plus a span start is + /// the absolute position of the sample the span starts at. pub(crate) fn resets_of(&self, index: usize) -> &[CounterReset] { let span = &self.windows[index]; resets_between( @@ -167,6 +169,10 @@ struct PendingBatch { /// Evaluation timestamp of each accumulated point. eval_ts: Vec, /// The retained-sample range of each point's window, one list per field. + /// + /// A span is relative to the retained arrays at the time it was recorded, not an absolute + /// index, so [`FieldSamples::advance`] invalidates it: that is why `evaluate` flushes the + /// batch before advancing. spans: Vec>>, /// One output column per field, written by the batch kernel. columns: Vec>>, @@ -202,9 +208,6 @@ impl PendingBatch { /// Records one point for the next batch call: its evaluation timestamp and, per field, the range /// of retained samples its window covers. -/// -/// The ranges are relative to the retained samples, so samples dropped before the batch flushes -/// would shift them; [`flush_batch`] therefore runs before any `advance`. fn accumulate(pending: &mut PendingBatch, index: usize, eval_ts: i64, fields: &[FieldSamples]) { debug_assert_eq!(pending.spans.len(), fields.len()); pending.indices.push(index); @@ -236,8 +239,6 @@ fn flush_batch( *batch_calls += 1; } for (index, field) in fields.iter().enumerate() { - // `pending.spans` index the retained arrays absolutely, so no adjustment is needed for - // samples `advance` has released: those sit before every span start. let batch = FloatWindows { timestamps: &field.timestamps, values: &field.values, @@ -1043,8 +1044,6 @@ impl RangeEvaluator { } state.cursor += 1; } - // The accumulated spans index the retained arrays, which `advance` below rewrites, so the - // batch has to flush first. if let Some(kernel) = self.batch.as_mut() { flush_batch( kernel, @@ -1073,6 +1072,11 @@ impl RangeEvaluator { } state.cursor = decided; } + debug_assert!( + self.pending.is_empty(), + "`evaluate` flushes the batch before `advance` rewrites the retained arrays a \ + pending span indexes" + ); for field in &mut self.fields { // After the last window, samples are still ingested so that the overflow and // histogram checks do not depend on how the series is chunked, but none are kept. diff --git a/src/promql/src/range_eval/function.rs b/src/promql/src/range_eval/function.rs index f2765d1a543..88888bb15f8 100644 --- a/src/promql/src/range_eval/function.rs +++ b/src/promql/src/range_eval/function.rs @@ -73,6 +73,11 @@ impl RangeFunction { /// the same as saying `IS_COUNTER` in `extrapolated_rate`: `rate` and `increase`. A kernel /// that corrects counters without the index being maintained would silently drop every /// correction, so a new counter function has to be added here as well. + /// + /// It governs both readers of the index: the per-window kernel above, and the batch kernel, + /// which reads [`FloatWindows::resets_of`]. When it returns `false`, `resets` stays empty and + /// `resets_of` yields nothing, so a batch kernel for a counter function would skip every + /// correction and report `delta`-like values. pub(crate) fn indexes_counter_resets(self) -> bool { matches!(self, Self::Rate | Self::Increase) } diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index c11af5e3c02..f3ea5d3d952 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -721,6 +721,185 @@ fn functions_match_range_udfs() { } } +/// Replaces the float sample of `field` at every row index in `rows` with a native histogram +/// sample, so the windows holding one take the mixed-window path. +fn mix_in_histograms(chunks: &mut [RawSeriesChunk], field: usize, rows: &BTreeSet) { + let histogram = "0/0/0/4014000000000000/4014000000000000/1/0_1/4014000000000000///"; + let view = OwnedHistogram::parse(histogram); + let mut start = 0; + for piece in chunks { + let len = piece.time.len(); + let local: Vec = rows + .range(start..start + len) + .map(|&row| row - start) + .collect(); + start += len; + if local.is_empty() { + continue; + } + let mut builder = HistogramBatchBuilder::new(); + let indices: UInt32Array = (0..len) + .map(|row| { + local + .binary_search(&row) + .is_ok() + .then(|| builder.push(view.view()).unwrap()) + }) + .collect(); + // A position is a float or a histogram, never both, so the float sample of a mixed-in row + // goes. + let values: Float64Array = piece.fields.as_slice()[field] + .iter() + .enumerate() + .map(|(row, sample)| match sample { + _ if local.binary_search(&row).is_ok() => None, + SampleRef::Float(value) => Some(value), + SampleRef::Missing => None, + SampleRef::Histogram(_) => unreachable!("histograms are mixed in once"), + }) + .collect(); + let histograms = HistogramSamples::try_new(indices, builder.finish().unwrap()).unwrap(); + let mixed = Samples::try_from_mixed(values, histograms).unwrap(); + let fields = piece + .fields + .as_slice() + .iter() + .enumerate() + .map(|(index, samples)| { + if index == field { + mixed.clone() + } else { + samples.clone() + } + }) + .collect(); + piece.fields = SeriesFields::try_from_vec(fields).unwrap(); + } +} + +/// Asserts two evaluator results are identical, including native histogram samples, which the +/// dense form [`expand`] rejects. +fn assert_same_series(actual: Option<&Series>, expected: Option<&Series>, context: &str) { + match (actual, expected) { + (None, None) => {} + (Some(actual), Some(expected)) => { + assert_eq!(actual.metric, expected.metric, "{context}: metric"); + let (TimeAxis::Grid(actual_range), TimeAxis::Grid(expected_range)) = + (&actual.time, &expected.time) + else { + panic!( + "{context}: expected grid series, got {:?} and {:?}", + actual.time, expected.time + ); + }; + assert_eq!(actual_range, expected_range, "{context}: grid range"); + for (field, (got, want)) in actual + .fields + .as_slice() + .iter() + .zip(expected.fields.as_slice()) + .enumerate() + { + for (row, (got, want)) in got.iter().zip(want.iter()).enumerate() { + let context = format!("{context}: field {field} row {row}"); + match (got, want) { + (SampleRef::Missing, SampleRef::Missing) => {} + (SampleRef::Float(got), SampleRef::Float(want)) => { + assert_eq!(got.to_bits(), want.to_bits(), "{context}: {got} vs {want}") + } + (SampleRef::Histogram(got), SampleRef::Histogram(want)) => { + let got = FloatHistogram::try_from_view(&got).unwrap(); + let want = FloatHistogram::try_from_view(&want).unwrap(); + // Both paths run the same histogram code over the same samples, so + // this is an exact comparison: an epsilon of zero. + assert_histogram_close(&got, &want, 0.0, &context); + } + (SampleRef::Missing, _) => { + panic!("{context}: only the per-window path produced a sample") + } + (_, SampleRef::Missing) => { + panic!("{context}: only the batch path produced a sample") + } + _ => panic!("{context}: a float on one path and a histogram on the other"), + } + } + } + } + (actual, expected) => panic!( + "{context}: one path produced a series, the other none: {actual:?} vs {expected:?}" + ), + } +} + +/// `rate` and `increase` over random input, computed twice: by the batch kernel and by the +/// per-window path. +/// +/// Both evaluators are built by `try_new` and differ in nothing but the dispatch shape, and both +/// kernels reach `extrapolated_rate_of`, so the results have to be bit-identical. A difference +/// means the batch read a different set of samples, evaluated at a different time, or corrected a +/// different reset: a window bound that disagrees with the per-window path. +/// +/// The rows mix gaps, stale markers, counter resets and native histogram samples, the last so +/// that some grid points take the mixed-window fallback, where a pending batch has to flush +/// before the point is decided on the per-window path. The random chunk splits are what make +/// `advance` drop samples mid-series, so spans that outlived one would read shifted samples. +#[test] +fn batch_counter_functions_match_the_per_window_path_on_random_input() { + let mut batch_calls = 0; + for seed in 0..200 { + let mut rng = StdRng::seed_from_u64(seed); + let fields = rng.random_range(1..=3); + let (grid, mut window) = random_window(&mut rng); + // An anchored selector reads one window and installs no batch kernel, so both paths would + // run the same code; every seed stays on the grid-driven batch path. + window.anchor_ms = None; + let function = if rng.random_bool(0.5) { + RangeFunction::Rate + } else { + RangeFunction::Increase + }; + let mut batched = + RangeEvaluator::try_new(&raw_schema(fields), grid.clone(), window, function).unwrap(); + let mut per_window = + RangeEvaluator::try_new(&raw_schema(fields), grid.clone(), window, function).unwrap(); + // The per-window path is the same evaluator without the batch kernel. `with_kernel`, + // which the other differential tests use, rejects native histogram samples, and these + // series carry some, so removing the batch form is the only switch that leaves the rest + // of the evaluation identical. + per_window.batch = None; + // One evaluator serves several series, so state must not leak between them. + for host in 0..3 { + let rows = random_rows(&mut rng, fields); + let histogram_field = rng.random_range(0..fields); + let histogram_rows: BTreeSet = + (0..rows.len()).filter(|_| rng.random_bool(0.1)).collect(); + let mut chunks = random_chunks(&mut rng, &rows, fields, host); + mix_in_histograms(&mut chunks, histogram_field, &histogram_rows); + // A push hands out the positions it decided, not only on the last chunk, so what each + // path produced is every chunk it handed out, folded back into one series. + let batched_chunks = collect_chunks(&mut batched, &chunks); + let per_window_chunks = collect_chunks(&mut per_window, &chunks); + + let batched_result = fold_chunks(&batched_chunks, batched.output_schema()); + let per_window_result = fold_chunks(&per_window_chunks, per_window.output_schema()); + + let context = format!("seed {seed} host {host}"); + assert_same_series( + batched_result.as_ref(), + per_window_result.as_ref(), + &context, + ); + } + batch_calls += batched.batch_call_count(); + } + // The comparison is worth nothing unless the batch kernel ran at all: counting the calls is + // what tells a batch-path comparison from one that silently fell back to per-window. + assert!( + batch_calls > 300, + "only {batch_calls} batch kernel calls: the batch path was barely reached" + ); +} + /// A chunk of entirely present float columns is appended in bulk, and that is the path the test /// asserts: the per-sample loop produces the same output, so comparing values alone could not /// tell whether the bulk path ran. @@ -786,31 +965,6 @@ fn a_chunk_with_histograms_is_not_appended_in_bulk() { assert_eq!(evaluator.bulk_ingest_count(), 0); } -/// `rate` over a float-only series takes the batch path. The per-window path produces the same -/// output, so a test comparing values alone could not tell which one ran. -#[test] -fn rate_over_floats_is_evaluated_in_batches() { - let rows: Vec = (0..20) - .map(|index| (index as i64 * 10_000, vec![Some(index as f64)])) - .collect(); - let mut evaluator = RangeEvaluator::try_new( - &raw_schema(1), - grid(0, 10_000, 20), - RangeWindow { - range_ms: 30_000, - offset_ms: 0, - anchor_ms: None, - }, - RangeFunction::Rate, - ) - .unwrap(); - evaluator.push(&chunk(&rows, 1, 0, true)).unwrap(); - assert!( - evaluator.batch_call_count() > 0, - "rate over floats must reach the batch kernel" - ); -} - /// A batch spanning more grid points than `EVALUATION_BATCH` flushes in the middle of one /// evaluation, and each field's windows come from that field's own samples. /// @@ -881,7 +1035,6 @@ fn a_batch_spanning_fields_and_the_flush_threshold_matches_the_per_window_path() .push(&chunk(&rows, fields, 0, true)) .unwrap() .expect("the series completes"); - assert_eq!(per_window.batch_call_count(), 0); // Fold the chunk back into one series first: a chunk's range reports the positions its push // decided, which can include leading and trailing positions holding no value, while the @@ -899,8 +1052,9 @@ fn a_batch_spanning_fields_and_the_flush_threshold_matches_the_per_window_path() /// A window holding a native histogram sample keeps its grid point on the per-window path, since /// the batch kernel reads the floats alone and cannot reproduce the mixed-window rules. /// -/// The series is laid out so every grid point's window contains the histogram: one point fewer -/// than the first test, so no point can fall before it. +/// The series is laid out so every grid point's window contains the histogram: three samples +/// 10 s apart with the histogram on the first, and a window reaching 30 s back, so a window +/// covering any grid point reaches timestamp 0. #[test] fn a_histogram_window_keeps_rate_on_the_per_window_path() { let histogram = "0/0/0/4014000000000000/4014000000000000/1/0_1/4014000000000000///";