fix: clear stale sliding aggregate state for empty RANGE frames

datafusion/physical-expr/src/window/sliding_aggregate.rs

@@ -210,6 +210,23 @@ impl AggregateWindowExpr for SlidingAggregateWindowExpr {
         filter_mask: Option<&BooleanArray>,
     ) -> Result<ScalarValue> {
         if cur_range.start == cur_range.end {
+            // Keep the accumulator synchronized with `last_range`. RANGE frames
+            // can become empty between two non-empty frames when the ORDER BY
+            // values contain gaps.
+            let retract_bound = last_range.end - last_range.start;
+            if retract_bound > 0 {
+                let slice_mask =
+                    filter_mask.map(|m| m.slice(last_range.start, retract_bound));
+                let retract: Vec<ArrayRef> = value_slice
+                    .iter()
+                    .map(|v| v.slice(last_range.start, retract_bound))
+                    .map(|arr| match &slice_mask {
+                        Some(m) => filter_array(&arr, m),
+                        None => Ok(arr),
+                    })
+                    .collect::<Result<Vec<_>>>()?;
+                accumulator.retract_batch(&retract)?
+            }
             self.aggregate
                 .default_value(self.aggregate.field().data_type())
         } else {

datafusion/sqllogictest/test_files/window.slt

@@ -6629,6 +6629,23 @@ FROM (
 2 1
 3 1
 
+# A RANGE frame can transition from non-empty to empty and back to non-empty
+# when the ORDER BY values contain gaps. The sliding accumulator must discard
+# the state from the previous non-empty frame.
+query IIIII
+SELECT k,
+       SUM(v) OVER w,
+       COUNT(v) OVER w,
+       MIN(v) OVER w,
+       MAX(v) OVER w
+FROM (VALUES (0, 100), (10, 90), (30, 10), (40, 20)) AS t(k, v)
+WINDOW w AS (ORDER BY k RANGE BETWEEN 10 PRECEDING AND 5 PRECEDING);
+----
+0 NULL 0 NULL NULL
+10 100 1 100 100
+30 NULL 0 NULL NULL
+40 10 1 10 10
+
 # AVG over a sliding window must yield NULL when the frame has no non-NULL
 # values — including frames that became empty via `retract_batch`. Covers
 # Float64, Decimal, and the narrow-frame retract-to-empty case.
@@ -6856,4 +6873,4 @@ ORDER BY id
 1 3 3 3 3 3
 2 NULL 3 3 3 3
 3 NULL NULL NULL NULL NULL
-4 7 7 7 7 7
+4 7 7 7 7 7