fix(promql): refuse a histogram in a modified window when it is ingested

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

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<i128>) -> 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,

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<Vec<EvaluatedChunk>> {
+    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<Row> = 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();
+    }
+}