chore: merge feat/narthex-window-eval into feat/narthex-vector-match

src/promql/src/range_eval.rs

@@ -263,11 +263,10 @@ impl RangeEvaluator {
         let range = i128::from(self.window.range_ms);
         while state.cursor < grid.len() {
             let eval_ts = grid[state.cursor];
-            if ready.is_some_and(|ready| i128::from(eval_ts) > ready) {
-                break;
-            }
             let lower = i128::from(eval_ts) - range;
             let mut earliest = None;
+            // Later windows start no earlier, so samples behind this one go even when the point
+            // itself is not ready yet.
             for field in &mut self.fields {
                 field.head += field.timestamps[field.head..]
                     .partition_point(|&timestamp| i128::from(timestamp) <= lower);
@@ -275,6 +274,9 @@ impl RangeEvaluator {
                     earliest = Some(earliest.map_or(timestamp, |min: i64| min.min(timestamp)));
                 }
             }
+            if ready.is_some_and(|ready| i128::from(eval_ts) > ready) {
+                break;
+            }
             // No retained sample: either the series is done or the next one is not here yet.
             let Some(earliest) = earliest else {
                 break;

src/promql/src/range_eval/tests.rs

@@ -427,3 +427,28 @@ fn retains_only_samples_later_windows_need() {
         assert!(evaluator.fields[0].timestamps.len() <= 60);
     }
 }
+
+/// Samples behind the next window are dropped while waiting for it, so a long gap between grid
+/// points keeps at most one chunk.
+#[test]
+fn waiting_for_a_point_drops_samples_behind_its_window() {
+    let window = RangeWindow {
+        range_ms: 10,
+        offset_ms: 0,
+    };
+    let mut evaluator = probe_evaluator(1, grid(0, 100_000, 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();
+        assert!(
+            evaluator
+                .push(&chunk(&rows, 1, 0, false))
+                .unwrap()
+                .is_none()
+        );
+        assert!(evaluator.fields[0].timestamps.len() <= 100);
+    }
+    let last = chunk(&[(100_000, vec![Some(1.0)])], 1, 0, true);
+    assert!(evaluator.push(&last).unwrap().is_some());
+}