chore: merge feat/narthex-window-eval into feat/narthex-window-eval-rate

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

src/promql/src/range_eval.rs

@@ -354,6 +354,11 @@ impl RangeEvaluator {
             state.cursor += 1;
         }
         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.
+            if state.cursor == grid.len() {
+                field.head = field.timestamps.len();
+            }
             field.timestamps.drain(..field.head);
             field.values.drain(..field.head);
             field.head = 0;

src/promql/src/range_eval/tests.rs

@@ -818,3 +818,40 @@ fn quantile_keeps_the_window_order_of_signed_zeros() {
     };
     assert_eq!(1.0 / value, f64::NEG_INFINITY);
 }
+
+/// Chunks after the last grid point are only validated, so nothing accumulates and the outcome
+/// does not depend on where the series is split.
+#[test]
+fn samples_after_the_last_window_are_validated_but_not_retained() {
+    let window = RangeWindow {
+        range_ms: 10,
+        offset_ms: 0,
+    };
+    let mut evaluator = probe_evaluator(1, grid(0, 10, 2), window);
+    for chunk_index in 0..900 {
+        let rows: Vec<Row> = (0..100)
+            .map(|row| (1 + chunk_index * 100 + row, vec![Some(1.0)]))
+            .collect();
+        evaluator.push(&chunk(&rows, 1, 0, false)).unwrap();
+        assert!(evaluator.fields[0].timestamps.len() <= 100);
+    }
+
+    let window = RangeWindow {
+        range_ms: 10,
+        offset_ms: 10,
+    };
+    let late = i64::MAX - 5;
+    let mut evaluator = probe_evaluator(1, grid(0, 10, 4), window);
+    evaluator
+        .push(&chunk(
+            &[(15, vec![Some(1.0)]), (20, vec![Some(1.0)])],
+            1,
+            0,
+            false,
+        ))
+        .unwrap();
+    assert!(matches!(
+        evaluator.push(&chunk(&[(late, vec![Some(1.0)])], 1, 0, true)),
+        Err(DataFusionError::Execution(_))
+    ));
+}