From 22da7c491baecebbd54bebd87f9e60265575e4ec Mon Sep 17 00:00:00 2001 From: Neil Conway Date: Fri, 24 Jul 2026 14:36:21 -0400 Subject: [PATCH 1/2] fix: sliding window MIN returns wrong value for all-NULL frame --- datafusion/functions-aggregate/src/min_max.rs | 121 ++++++++++++++---- datafusion/sqllogictest/test_files/window.slt | 14 ++ 2 files changed, 111 insertions(+), 24 deletions(-) diff --git a/datafusion/functions-aggregate/src/min_max.rs b/datafusion/functions-aggregate/src/min_max.rs index f4eaaab853464..360d6b87af178 100644 --- a/datafusion/functions-aggregate/src/min_max.rs +++ b/datafusion/functions-aggregate/src/min_max.rs @@ -392,28 +392,41 @@ impl SlidingMaxAccumulator { moving_max: MovingMax::::new(), }) } + + /// Sets `self.max` to the maximum of the current window, or a typed NULL + /// if the window contains no non-null values. NULL values are never pushed + /// into `self.moving_max`, so it may be empty even when the window frame + /// is not. + fn update_max(&mut self) -> Result<()> { + self.max = match self.moving_max.max() { + Some(res) => res.clone(), + None => ScalarValue::try_from(&self.max.data_type())?, + }; + Ok(()) + } } impl Accumulator for SlidingMaxAccumulator { fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { for idx in 0..values[0].len() { let val = ScalarValue::try_from_array(&values[0], idx)?; - self.moving_max.push(val); - } - if let Some(res) = self.moving_max.max() { - self.max = res.clone(); + if !val.is_null() { + self.moving_max.push(val); + } } - Ok(()) + self.update_max() } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { - for _idx in 0..values[0].len() { - (self.moving_max).pop(); + // We assume that values are retracted in the order they were added, so + // the retracted values must be the oldest elements of `moving_max`. + // NULLs are never pushed, so be sure to only pop once per non-NULL + // value. + let non_null = values[0].len() - values[0].logical_null_count(); + for _ in 0..non_null { + self.moving_max.pop(); } - if let Some(res) = self.moving_max.max() { - self.max = res.clone(); - } - Ok(()) + self.update_max() } fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { @@ -675,6 +688,18 @@ impl SlidingMinAccumulator { moving_min: MovingMin::::new(), }) } + + /// Sets `self.min` to the minimum of the current window, or a typed NULL + /// if the window contains no non-null values. NULL values are never pushed + /// into `self.moving_min`, so it may be empty even when the window frame + /// is not. + fn update_min(&mut self) -> Result<()> { + self.min = match self.moving_min.min() { + Some(res) => res.clone(), + None => ScalarValue::try_from(&self.min.data_type())?, + }; + Ok(()) + } } impl Accumulator for SlidingMinAccumulator { @@ -689,23 +714,19 @@ impl Accumulator for SlidingMinAccumulator { self.moving_min.push(val); } } - if let Some(res) = self.moving_min.min() { - self.min = res.clone(); - } - Ok(()) + self.update_min() } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { - for idx in 0..values[0].len() { - let val = ScalarValue::try_from_array(&values[0], idx)?; - if !val.is_null() { - (self.moving_min).pop(); - } + // We assume that values are retracted in the order they were added, so + // the retracted values must be the oldest elements of `moving_min`. + // NULLs are never pushed, so be sure to only pop once per non-NULL + // value. + let non_null = values[0].len() - values[0].logical_null_count(); + for _ in 0..non_null { + self.moving_min.pop(); } - if let Some(res) = self.moving_min.min() { - self.min = res.clone(); - } - Ok(()) + self.update_min() } fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { @@ -1195,6 +1216,58 @@ mod tests { Ok(()) } + #[test] + fn sliding_min_all_null_window() -> Result<()> { + let mut min_acc = SlidingMinAccumulator::try_new(&DataType::Int32)?; + + let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(3), None])); + min_acc.update_batch(&[Arc::clone(&values)])?; + assert_eq!(min_acc.evaluate()?, ScalarValue::Int32(Some(3))); + + // Retract `3`; the window now contains only the NULL + let retracted: ArrayRef = Arc::new(Int32Array::from(vec![Some(3)])); + min_acc.retract_batch(&[Arc::clone(&retracted)])?; + assert_eq!(min_acc.evaluate()?, ScalarValue::Int32(None)); + + // A subsequent non-null value must be picked up again + let update: ArrayRef = Arc::new(Int32Array::from(vec![Some(7)])); + min_acc.update_batch(&[Arc::clone(&update)])?; + assert_eq!(min_acc.evaluate()?, ScalarValue::Int32(Some(7))); + + // Retracting the NULL row must not pop the remaining value + let null_row: ArrayRef = Arc::new(Int32Array::from(vec![None::])); + min_acc.retract_batch(&[Arc::clone(&null_row)])?; + assert_eq!(min_acc.evaluate()?, ScalarValue::Int32(Some(7))); + + Ok(()) + } + + #[test] + fn sliding_max_all_null_window() -> Result<()> { + let mut max_acc = SlidingMaxAccumulator::try_new(&DataType::Int32)?; + + let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(3), None])); + max_acc.update_batch(&[Arc::clone(&values)])?; + assert_eq!(max_acc.evaluate()?, ScalarValue::Int32(Some(3))); + + // Retract `3`; the window now contains only the NULL + let retracted: ArrayRef = Arc::new(Int32Array::from(vec![Some(3)])); + max_acc.retract_batch(&[Arc::clone(&retracted)])?; + assert_eq!(max_acc.evaluate()?, ScalarValue::Int32(None)); + + // A subsequent non-null value must be picked up again + let update: ArrayRef = Arc::new(Int32Array::from(vec![Some(7)])); + max_acc.update_batch(&[Arc::clone(&update)])?; + assert_eq!(max_acc.evaluate()?, ScalarValue::Int32(Some(7))); + + // Retracting the NULL row must not disturb the remaining value + let null_row: ArrayRef = Arc::new(Int32Array::from(vec![None::])); + max_acc.retract_batch(&[Arc::clone(&null_row)])?; + assert_eq!(max_acc.evaluate()?, ScalarValue::Int32(Some(7))); + + Ok(()) + } + #[test] fn moving_min_tests() -> Result<()> { moving_min_i32(100, 10)?; diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt index cbbd9b74dfc00..fd477a3386b69 100644 --- a/datafusion/sqllogictest/test_files/window.slt +++ b/datafusion/sqllogictest/test_files/window.slt @@ -6842,3 +6842,17 @@ DROP TABLE issue_20194_t1; statement ok DROP TABLE issue_20194_t2; + +# Sliding-window MIN/MAX over a frame whose non-NULL values have all been +# retracted should yield NULL. +query IIII +SELECT id, x, + MIN(x) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS min_x, + MAX(x) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS max_x +FROM (VALUES (1, 3), (2, NULL), (3, NULL), (4, 7)) t(id, x) +ORDER BY id +---- +1 3 3 3 +2 NULL 3 3 +3 NULL NULL NULL +4 7 7 7 From b2d6d3d18b01732cd51ba659a5d6c44bffadfc8d Mon Sep 17 00:00:00 2001 From: Neil Conway Date: Fri, 24 Jul 2026 17:18:29 -0400 Subject: [PATCH 2/2] refactor: derive sliding MIN/MAX results from the window structure The sliding accumulators cached the current min/max in a ScalarValue field, refreshed after every update_batch and retract_batch. The cache cost up to two extra ScalarValue clones per output row and was the root cause of the stale-value bug fixed in the previous commit: a cached value can go stale, a derived one cannot. Replace the cache with an immutable precomputed typed NULL that fixes the accumulator's data type and serves as the result for frames with no non-null values; evaluate() and state() now read the current extremum directly from the window structure. --- datafusion/functions-aggregate/src/min_max.rs | 56 ++++++++----------- 1 file changed, 24 insertions(+), 32 deletions(-) diff --git a/datafusion/functions-aggregate/src/min_max.rs b/datafusion/functions-aggregate/src/min_max.rs index 360d6b87af178..1fa0aec6f3fa1 100644 --- a/datafusion/functions-aggregate/src/min_max.rs +++ b/datafusion/functions-aggregate/src/min_max.rs @@ -380,7 +380,8 @@ impl AggregateUDFImpl for Max { #[derive(Debug)] pub struct SlidingMaxAccumulator { - max: ScalarValue, + /// Typed NULL returned when the window contains no non-null values + empty_value: ScalarValue, moving_max: MovingMax, } @@ -388,21 +389,16 @@ impl SlidingMaxAccumulator { /// new max accumulator pub fn try_new(datatype: &DataType) -> Result { Ok(Self { - max: ScalarValue::try_from(datatype)?, + empty_value: ScalarValue::try_from(datatype)?, moving_max: MovingMax::::new(), }) } - /// Sets `self.max` to the maximum of the current window, or a typed NULL - /// if the window contains no non-null values. NULL values are never pushed - /// into `self.moving_max`, so it may be empty even when the window frame - /// is not. - fn update_max(&mut self) -> Result<()> { - self.max = match self.moving_max.max() { + fn current_max(&self) -> ScalarValue { + match self.moving_max.max() { Some(res) => res.clone(), - None => ScalarValue::try_from(&self.max.data_type())?, - }; - Ok(()) + None => self.empty_value.clone(), + } } } @@ -414,7 +410,7 @@ impl Accumulator for SlidingMaxAccumulator { self.moving_max.push(val); } } - self.update_max() + Ok(()) } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { @@ -426,7 +422,7 @@ impl Accumulator for SlidingMaxAccumulator { for _ in 0..non_null { self.moving_max.pop(); } - self.update_max() + Ok(()) } fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { @@ -434,11 +430,11 @@ impl Accumulator for SlidingMaxAccumulator { } fn state(&mut self) -> Result> { - Ok(vec![self.max.clone()]) + Ok(vec![self.current_max()]) } fn evaluate(&mut self) -> Result { - Ok(self.max.clone()) + Ok(self.current_max()) } fn supports_retract_batch(&self) -> bool { @@ -446,7 +442,7 @@ impl Accumulator for SlidingMaxAccumulator { } fn size(&self) -> usize { - size_of_val(self) - size_of_val(&self.max) + self.max.size() + size_of_val(self) - size_of_val(&self.empty_value) + self.empty_value.size() } } @@ -677,34 +673,30 @@ impl AggregateUDFImpl for Min { #[derive(Debug)] pub struct SlidingMinAccumulator { - min: ScalarValue, + /// Typed NULL returned when the window contains no non-null values + empty_value: ScalarValue, moving_min: MovingMin, } impl SlidingMinAccumulator { pub fn try_new(datatype: &DataType) -> Result { Ok(Self { - min: ScalarValue::try_from(datatype)?, + empty_value: ScalarValue::try_from(datatype)?, moving_min: MovingMin::::new(), }) } - /// Sets `self.min` to the minimum of the current window, or a typed NULL - /// if the window contains no non-null values. NULL values are never pushed - /// into `self.moving_min`, so it may be empty even when the window frame - /// is not. - fn update_min(&mut self) -> Result<()> { - self.min = match self.moving_min.min() { + fn current_min(&self) -> ScalarValue { + match self.moving_min.min() { Some(res) => res.clone(), - None => ScalarValue::try_from(&self.min.data_type())?, - }; - Ok(()) + None => self.empty_value.clone(), + } } } impl Accumulator for SlidingMinAccumulator { fn state(&mut self) -> Result> { - Ok(vec![self.min.clone()]) + Ok(vec![self.current_min()]) } fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { @@ -714,7 +706,7 @@ impl Accumulator for SlidingMinAccumulator { self.moving_min.push(val); } } - self.update_min() + Ok(()) } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { @@ -726,7 +718,7 @@ impl Accumulator for SlidingMinAccumulator { for _ in 0..non_null { self.moving_min.pop(); } - self.update_min() + Ok(()) } fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { @@ -734,7 +726,7 @@ impl Accumulator for SlidingMinAccumulator { } fn evaluate(&mut self) -> Result { - Ok(self.min.clone()) + Ok(self.current_min()) } fn supports_retract_batch(&self) -> bool { @@ -742,7 +734,7 @@ impl Accumulator for SlidingMinAccumulator { } fn size(&self) -> usize { - size_of_val(self) - size_of_val(&self.min) + self.min.size() + size_of_val(self) - size_of_val(&self.empty_value) + self.empty_value.size() } }