From b4659f54666eeb393573a4467cf4e0e961c7e7b7 Mon Sep 17 00:00:00 2001 From: Neil Conway Date: Tue, 28 Jul 2026 16:43:37 -0400 Subject: [PATCH 1/3] perf: speed up sliding-window percentile_cont and median accumulators The non-distinct PercentileContAccumulator and MedianAccumulator used the default SipHash hasher for their internal HashMaps. This is slow; switching to foldhash is significantly faster. Also, add a null-free fast path to `update_batch` and `retract_batch` in both accumulators. Benchmarks: percentile_cont no_nulls window=256 -61.1% (249.0 -> 96.6 us) percentile_cont with_nulls window=256 -59.4% (232.7 -> 94.5 us) percentile_cont no_nulls window=4096 -50.9% (756.3 -> 370.4 us) percentile_cont with_nulls window=4096 -48.2% (552.2 -> 286.2 us) percentile_cont no_nulls window=16384 -47.3% (2.311 -> 1.218 ms) percentile_cont with_nulls window=16384 -42.0% (1.465 -> 0.851 ms) median no_nulls window=256 -62.8% (247.0 -> 92.0 us) median with_nulls window=256 -59.8% (228.8 -> 92.0 us) median no_nulls window=4096 -40.0% (411.4 -> 246.6 us) median with_nulls window=4096 -39.4% (364.4 -> 221.0 us) median no_nulls window=16384 -31.5% (1.107 -> 0.759 ms) median with_nulls window=16384 -32.9% (0.947 -> 0.637 ms) --- datafusion/functions-aggregate/src/median.rs | 60 +++++++++++++++++-- .../src/percentile_cont.rs | 55 ++++++++++++++--- 2 files changed, 103 insertions(+), 12 deletions(-) diff --git a/datafusion/functions-aggregate/src/median.rs b/datafusion/functions-aggregate/src/median.rs index 7a399f73ec8e2..51f18d657fa71 100644 --- a/datafusion/functions-aggregate/src/median.rs +++ b/datafusion/functions-aggregate/src/median.rs @@ -39,6 +39,7 @@ use arrow::datatypes::{ ArrowNativeType, ArrowPrimitiveType, Decimal32Type, Decimal64Type, FieldRef, }; +use datafusion_common::hash_utils::RandomState; use datafusion_common::types::{NativeType, logical_float64}; use datafusion_common::{ DataFusionError, Result, ScalarValue, assert_eq_or_internal_err, exec_datafusion_err, @@ -288,7 +289,12 @@ impl Accumulator for MedianAccumulator { "failed to reserve {additional} values for median accumulator: {e}" ) })?; - self.all_values.extend(values.iter().flatten()); + if values.null_count() > 0 { + self.all_values.extend(values.iter().flatten()); + } else { + // Fast path: no nulls, so the values buffer can be appended wholesale. + self.all_values.extend_from_slice(values.values()); + } Ok(()) } @@ -310,11 +316,19 @@ impl Accumulator for MedianAccumulator { } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { - let mut to_remove: HashMap, usize> = HashMap::new(); + let mut to_remove: HashMap, usize, RandomState> = + HashMap::default(); let arr = values[0].as_primitive::(); - for value in arr.iter().flatten() { - *to_remove.entry(Hashable(value)).or_default() += 1; + if arr.null_count() > 0 { + for value in arr.iter().flatten() { + *to_remove.entry(Hashable(value)).or_default() += 1; + } + } else { + // Fast path: no nulls, so skip the per-element validity check. + for value in arr.values().iter() { + *to_remove.entry(Hashable(*value)).or_default() += 1; + } } let mut i = 0; @@ -627,3 +641,41 @@ fn calculate_median(values: &mut [T::Native]) -> Option MedianAccumulator { + MedianAccumulator { + data_type: DataType::Float64, + all_values: vec![], + } + } + + #[test] + fn update_batch_with_and_without_nulls_agree() { + // The null-free fast path must accumulate the same values as the + // general path. + let dense: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])); + let sparse: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(1.0), + None, + Some(2.0), + None, + Some(3.0), + ])); + + let mut dense_acc = median_accumulator(); + dense_acc + .update_batch(std::slice::from_ref(&dense)) + .unwrap(); + let mut sparse_acc = median_accumulator(); + sparse_acc + .update_batch(std::slice::from_ref(&sparse)) + .unwrap(); + + assert_eq!(dense_acc.all_values, sparse_acc.all_values); + } +} diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs index 4b5e892fb5cdf..6bb5438fda397 100644 --- a/datafusion/functions-aggregate/src/percentile_cont.rs +++ b/datafusion/functions-aggregate/src/percentile_cont.rs @@ -32,6 +32,7 @@ use arrow::{ use num_traits::AsPrimitive; use arrow::array::ArrowNativeTypeOp; +use datafusion_common::hash_utils::RandomState; use datafusion_common::internal_err; use datafusion_common::types::{NativeType, logical_float64}; use datafusion_functions_aggregate_common::noop_accumulator::NoopAccumulator; @@ -427,7 +428,12 @@ where "failed to reserve {additional} values for percentile_cont accumulator: {e}" ) })?; - self.all_values.extend(values.iter().flatten()); + if values.null_count() > 0 { + self.all_values.extend(values.iter().flatten()); + } else { + // Fast path: no nulls, so the values buffer can be appended wholesale. + self.all_values.extend_from_slice(values.values()); + } Ok(()) } @@ -447,11 +453,19 @@ where } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { - let mut to_remove: HashMap, usize> = HashMap::new(); + let mut to_remove: HashMap, usize, RandomState> = + HashMap::default(); let arr = values[0].as_primitive::(); - for value in arr.iter().flatten() { - *to_remove.entry(Hashable(value)).or_default() += 1; + if arr.null_count() > 0 { + for value in arr.iter().flatten() { + *to_remove.entry(Hashable(value)).or_default() += 1; + } + } else { + // Fast path: no nulls, so skip the per-element validity check. + for value in arr.values().iter() { + *to_remove.entry(Hashable(*value)).or_default() += 1; + } } let mut i = 0; @@ -842,18 +856,43 @@ where #[cfg(test)] mod tests { - use super::calculate_percentile; + use super::*; + use arrow::array::Float64Array; use half::f16; + #[test] + fn update_batch_with_and_without_nulls_agree() { + // The null-free fast path must accumulate the same values as the + // general path. + let dense: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])); + let sparse: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(1.0), + None, + Some(2.0), + None, + Some(3.0), + ])); + + let mut dense_acc = PercentileContAccumulator::::new(0.5); + dense_acc + .update_batch(std::slice::from_ref(&dense)) + .unwrap(); + let mut sparse_acc = PercentileContAccumulator::::new(0.5); + sparse_acc + .update_batch(std::slice::from_ref(&sparse)) + .unwrap(); + + assert_eq!(dense_acc.all_values, sparse_acc.all_values); + } + #[test] fn f16_interpolation_does_not_overflow_to_nan() { // Regression test for https://github.com/apache/datafusion/issues/18945 // Interpolating between 0 and the max finite f16 value previously overflowed // intermediate f16 computations and produced NaN. let mut values = vec![f16::from_f32(0.0), f16::from_f32(65504.0)]; - let result = - calculate_percentile::(&mut values, 0.5) - .expect("non-empty input"); + let result = calculate_percentile::(&mut values, 0.5) + .expect("non-empty input"); let result_f = result.to_f32(); assert!( !result_f.is_nan(), From b7c8c5be644ac8c32bb349c21f95213d533a1aa6 Mon Sep 17 00:00:00 2001 From: Neil Conway Date: Tue, 28 Jul 2026 16:44:22 -0400 Subject: [PATCH 2/3] fix: error handling for retracking untracked values If `retract_batch` is asked to remove values the accumulator is not tracking, its state has diverged from the window frame and continuing would silently produce wrong results; return an internal error rather than silently ignoring. --- datafusion/functions-aggregate/src/median.rs | 28 ++++++++++++++++- .../src/percentile_cont.rs | 31 +++++++++++++++++-- 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/datafusion/functions-aggregate/src/median.rs b/datafusion/functions-aggregate/src/median.rs index 51f18d657fa71..81a3c076dffbe 100644 --- a/datafusion/functions-aggregate/src/median.rs +++ b/datafusion/functions-aggregate/src/median.rs @@ -43,7 +43,7 @@ use datafusion_common::hash_utils::RandomState; use datafusion_common::types::{NativeType, logical_float64}; use datafusion_common::{ DataFusionError, Result, ScalarValue, assert_eq_or_internal_err, exec_datafusion_err, - internal_datafusion_err, + internal_datafusion_err, internal_err, }; use datafusion_expr::function::StateFieldsArgs; use datafusion_expr::{ @@ -349,6 +349,15 @@ impl Accumulator for MedianAccumulator { i += 1; } } + + // Retracting values that are not tracked means the accumulator state + // has diverged from the window frame; continuing would silently + // produce wrong results, so surface it as an error. + if !to_remove.is_empty() { + return internal_err!( + "median retract_batch: retracted value(s) not present in the window" + ); + } Ok(()) } @@ -654,6 +663,23 @@ mod tests { } } + #[test] + fn retract_batch_errors_on_untracked_value() { + let mut acc = median_accumulator(); + let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0])); + acc.update_batch(std::slice::from_ref(&values)).unwrap(); + + let retract: ArrayRef = Arc::new(Float64Array::from(vec![3.0])); + let err = acc + .retract_batch(std::slice::from_ref(&retract)) + .unwrap_err() + .to_string(); + assert!( + err.contains("not present in the window"), + "unexpected error: {err}" + ); + } + #[test] fn update_batch_with_and_without_nulls_agree() { // The null-free fast path must accumulate the same values as the diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs index 6bb5438fda397..45be3209922c5 100644 --- a/datafusion/functions-aggregate/src/percentile_cont.rs +++ b/datafusion/functions-aggregate/src/percentile_cont.rs @@ -486,6 +486,15 @@ where i += 1; } } + + // Retracting values that are not tracked means the accumulator state + // has diverged from the window frame; continuing would silently + // produce wrong results, so surface it as an error. + if !to_remove.is_empty() { + return internal_err!( + "percentile_cont retract_batch: retracted value(s) not present in the window" + ); + } Ok(()) } @@ -860,6 +869,23 @@ mod tests { use arrow::array::Float64Array; use half::f16; + #[test] + fn retract_batch_errors_on_untracked_value() { + let mut acc = PercentileContAccumulator::::new(0.5); + let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0])); + acc.update_batch(std::slice::from_ref(&values)).unwrap(); + + let retract: ArrayRef = Arc::new(Float64Array::from(vec![3.0])); + let err = acc + .retract_batch(std::slice::from_ref(&retract)) + .unwrap_err() + .to_string(); + assert!( + err.contains("not present in the window"), + "unexpected error: {err}" + ); + } + #[test] fn update_batch_with_and_without_nulls_agree() { // The null-free fast path must accumulate the same values as the @@ -891,8 +917,9 @@ mod tests { // Interpolating between 0 and the max finite f16 value previously overflowed // intermediate f16 computations and produced NaN. let mut values = vec![f16::from_f32(0.0), f16::from_f32(65504.0)]; - let result = calculate_percentile::(&mut values, 0.5) - .expect("non-empty input"); + let result = + calculate_percentile::(&mut values, 0.5) + .expect("non-empty input"); let result_f = result.to_f32(); assert!( !result_f.is_nan(), From 886dbef697559db7c1b1a0dcb13eb8a17705e0de Mon Sep 17 00:00:00 2001 From: Neil Conway Date: Wed, 29 Jul 2026 07:42:57 -0400 Subject: [PATCH 3/3] chore: apply rustfmt --- datafusion/functions-aggregate/src/percentile_cont.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs index 45be3209922c5..53eda31f6e512 100644 --- a/datafusion/functions-aggregate/src/percentile_cont.rs +++ b/datafusion/functions-aggregate/src/percentile_cont.rs @@ -917,9 +917,8 @@ mod tests { // Interpolating between 0 and the max finite f16 value previously overflowed // intermediate f16 computations and produced NaN. let mut values = vec![f16::from_f32(0.0), f16::from_f32(65504.0)]; - let result = - calculate_percentile::(&mut values, 0.5) - .expect("non-empty input"); + let result = calculate_percentile::(&mut values, 0.5) + .expect("non-empty input"); let result_f = result.to_f32(); assert!( !result_f.is_nan(),