From 804d050c2b6d01ff138e099f61e8a1f92f77f83e Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Fri, 23 Oct 2026 16:09:32 +0800 Subject: [PATCH] fix(promql): refuse a histogram in a modified window when it is ingested Signed-off-by: Dennis Zhuang --- src/promql/src/range_eval.rs | 50 +++++++++++++++---- src/promql/src/range_eval/tests.rs | 77 ++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+), 8 deletions(-) diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index 59c8faa08f3..11aa896690a 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -1065,6 +1065,20 @@ impl RangeEvaluator { self.window.offset_ms )) })?; + // Refused here rather than when a point is evaluated: a smoothed point can be decided + // before the rest of its window arrives, so whether a later histogram is seen would + // depend on how the input is chunked. + if let Some(modifier) = self.window.modifier + && row + .iter() + .any(|sample| matches!(sample, SampleRef::Histogram(_))) + && self.decided_window_reads(shifted) + { + return Err(DataFusionError::NotImplemented(format!( + "the {} range modifier over native histogram samples", + modifier.name() + ))); + } for (field, sample) in self.fields.iter_mut().zip(&row) { match sample { SampleRef::Missing => {} @@ -1088,6 +1102,34 @@ impl RangeEvaluator { Ok(()) } + /// Whether a sample at `timestamp` on the evaluation timeline lies in the window of a point + /// this evaluator decides: a window ending at `E` reads `(E - range - before, E + after]`, and + /// only the first point's window is read when the result is replayed. + fn decided_window_reads(&self, timestamp: i64) -> bool { + let (before, after) = self.window.modifier.map_or((0, 0), |modifier| { + ( + i128::from(modifier.before_ms()), + i128::from(modifier.after_ms()), + ) + }); + let spec = self.grid.spec(); + let ends = match self.replay { + Replay::PerPoint => i128::from(spec.len), + Replay::Once | Replay::EveryPoint => 1, + }; + let (start, step) = (i128::from(spec.start_ms), i128::from(spec.step_ms)); + // The window ends that read the sample are those in `[timestamp - after, timestamp + + // range + before)`; the first grid point at or after the lower bound decides. + let lowest = i128::from(timestamp) - after; + let highest = i128::from(timestamp) + i128::from(self.window.range_ms) + before; + let first = if lowest <= start { + 0 + } else { + (lowest - start + step - 1) / step + }; + first < ends && start + first * step < highest + } + /// Evaluates grid points up to `ready` on the evaluation timeline, or all remaining points /// when `ready` is `None`. fn evaluate(&mut self, ready: Option) -> Result<()> { @@ -1332,14 +1374,6 @@ impl RowEvaluator<'_> { let histograms = &field.histograms[field.histogram_head..]; let histograms = &histograms [..histograms.partition_point(|(timestamp, _)| *timestamp <= window_end)]; - if let Some(modifier) = self.modifier - && !histograms.is_empty() - { - return Err(DataFusionError::NotImplemented(format!( - "the {} range modifier over native histogram samples", - modifier.name() - ))); - } row.push(match self.function { Some(function) if !histograms.is_empty() => histogram::evaluate( function, diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index 825222cc54e..2d2e8b7c3ad 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -4663,3 +4663,80 @@ fn smoothed_waits_for_the_sample_after_a_point_its_chunk_ended_on() { "smoothed delta at 61s", ); } + +/// `delta(m[2s] smoothed)` with a 3s lookback at 61s: floats at 59s and 61s, then `histogram` at +/// `histogram_ts`, either as one chunk or with the histogram in a closing chunk of its own. +fn smoothed_delta_with_histogram( + histogram_ts: i64, + one_chunk: bool, +) -> datafusion::error::Result> { + let window = RangeWindow { + range_ms: 2_000, + offset_ms: 0, + anchor_ms: None, + modifier: Some(RangeModifier::Smoothed { lookback_ms: 3_000 }), + }; + let schema = raw_schema(1); + let mut evaluator = RangeEvaluator::try_new( + &schema, + grid(61_000, 1_000, 1), + window, + RangeFunction::Delta, + ) + .unwrap(); + let floats: Vec = vec![(59_000, vec![Some(2.0)]), (61_000, vec![Some(4.0)])]; + let histogram = histogram_chunk(&schema, 1, histogram_ts); + let pieces = if one_chunk { + let float_chunk = chunk(&floats, 1, 0, false); + let field = Samples::try_from_chunks(vec![ + float_chunk.fields.as_slice()[0].clone(), + histogram.fields.as_slice()[0].clone(), + ]) + .unwrap(); + let time = + RawTimestamps::try_new(ScalarBuffer::from(vec![59_000, 61_000, histogram_ts])).unwrap(); + vec![ + RawSeriesChunk::try_new( + &schema, + float_chunk.primary_key.clone(), + float_chunk.metric.clone(), + time, + SeriesFields::try_from_vec(vec![field]).unwrap(), + true, + ) + .unwrap(), + ] + } else { + let mut closing = histogram; + closing.is_last = true; + vec![chunk(&floats, 1, 0, false), closing] + }; + let mut chunks = Vec::new(); + for piece in &pieces { + chunks.extend(evaluator.push(piece)?); + } + Ok(chunks) +} + +/// The smoothed point at 61s is decided by the float at 61s, before a histogram at 62s inside its +/// extended window arrives. The histogram is still refused, however the input is chunked, rather +/// than refused in one chunk and swallowed after the point was handed over in another. +#[test] +fn a_smoothed_window_refuses_a_histogram_whatever_the_chunking() { + for one_chunk in [true, false] { + let error = smoothed_delta_with_histogram(62_000, one_chunk).unwrap_err(); + assert!( + matches!(error, DataFusionError::NotImplemented(_)), + "one chunk: {one_chunk}: {error}" + ); + } +} + +/// A histogram past the extended window of every point is not read by any of them, so it is +/// accepted: the 61s window with a 3s lookback reads up to 64s. +#[test] +fn a_histogram_outside_every_smoothed_window_is_accepted() { + for one_chunk in [true, false] { + smoothed_delta_with_histogram(65_000, one_chunk).unwrap(); + } +}