diff --git a/datafusion/functions-aggregate/src/min_max.rs b/datafusion/functions-aggregate/src/min_max.rs index f4eaaab85346..1fa0aec6f3fa 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,30 +389,38 @@ 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(), }) } + + fn current_max(&self) -> ScalarValue { + match self.moving_max.max() { + Some(res) => res.clone(), + None => self.empty_value.clone(), + } + } } 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(()) } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { - for _idx in 0..values[0].len() { - (self.moving_max).pop(); - } - if let Some(res) = self.moving_max.max() { - self.max = res.clone(); + // 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(); } Ok(()) } @@ -421,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 { @@ -433,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() } } @@ -664,22 +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(), }) } + + fn current_min(&self) -> ScalarValue { + match self.moving_min.min() { + Some(res) => res.clone(), + 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<()> { @@ -689,21 +706,17 @@ impl Accumulator for SlidingMinAccumulator { self.moving_min.push(val); } } - if let Some(res) = self.moving_min.min() { - self.min = res.clone(); - } Ok(()) } 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(); - } - } - if let Some(res) = self.moving_min.min() { - self.min = res.clone(); + // 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(); } Ok(()) } @@ -713,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 { @@ -721,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() } } @@ -1195,6 +1208,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 cbbd9b74dfc0..fd477a3386b6 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