You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/29 13:23:17 UTC
(datafusion) 01/01: perf: Optimize hashing, null-free fast path for `percentile_cont`, `median` (#23954)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23954-1727b7d709630767032dea6d0eb1e43ca8e6ef04
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 043d97fb2a1ec2576f7456b53030cf009f9278c7
Author: Neil Conway <ne...@gmail.com>
AuthorDate: Wed Jul 29 09:22:54 2026 -0400
perf: Optimize hashing, null-free fast path for `percentile_cont`, `median` (#23954)
## Which issue does this PR close?
- Closes #23953
## Rationale for this change
This PR makes three improvements to non-distinct
`PercentileContAccumulator` and `MedianAccumulator`, inspired by recent
work on `percentile_cont(DISTINCT)` (#23946):
1. Switch from SipHash to foldhash for internal hash maps
2. Add a null-free fast path to `update_batch` and `retract_batch`
3. Raise an error if we attempt to retract an unknown value in
`retract_batch`, rather than silently ignoring it.
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)
## What changes are included in this PR?
See above.
## Are these changes tested?
Yes: existing tests pass, new tests added for the null-free fast path
and the improved error handling.
## Are there any user-facing changes?
No.
---
datafusion/functions-aggregate/src/median.rs | 88 ++++++++++++++++++++--
.../functions-aggregate/src/percentile_cont.rs | 80 ++++++++++++++++++--
2 files changed, 155 insertions(+), 13 deletions(-)
diff --git a/datafusion/functions-aggregate/src/median.rs b/datafusion/functions-aggregate/src/median.rs
index 7a399f73ec..81a3c076df 100644
--- a/datafusion/functions-aggregate/src/median.rs
+++ b/datafusion/functions-aggregate/src/median.rs
@@ -39,10 +39,11 @@ 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,
- internal_datafusion_err,
+ internal_datafusion_err, internal_err,
};
use datafusion_expr::function::StateFieldsArgs;
use datafusion_expr::{
@@ -288,7 +289,12 @@ impl<T: ArrowNumericType> Accumulator for MedianAccumulator<T> {
"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<T: ArrowNumericType> Accumulator for MedianAccumulator<T> {
}
fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let mut to_remove: HashMap<Hashable<T::Native>, usize> = HashMap::new();
+ let mut to_remove: HashMap<Hashable<T::Native>, usize, RandomState> =
+ HashMap::default();
let arr = values[0].as_primitive::<T>();
- 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;
@@ -335,6 +349,15 @@ impl<T: ArrowNumericType> Accumulator for MedianAccumulator<T> {
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(())
}
@@ -627,3 +650,58 @@ fn calculate_median<T: ArrowNumericType>(values: &mut [T::Native]) -> Option<T::
Some(*median)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::Float64Array;
+
+ fn median_accumulator() -> MedianAccumulator<Float64Type> {
+ MedianAccumulator {
+ data_type: DataType::Float64,
+ all_values: vec![],
+ }
+ }
+
+ #[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
+ // 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 baf1978ea2..3a98900bbb 100644
--- a/datafusion/functions-aggregate/src/percentile_cont.rs
+++ b/datafusion/functions-aggregate/src/percentile_cont.rs
@@ -429,7 +429,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(())
}
@@ -449,11 +454,19 @@ where
}
fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let mut to_remove: HashMap<Hashable<T::Native>, usize> = HashMap::new();
+ let mut to_remove: HashMap<Hashable<T::Native>, usize, RandomState> =
+ HashMap::default();
let arr = values[0].as_primitive::<T>();
- 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;
@@ -474,6 +487,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(())
}
@@ -876,18 +898,60 @@ where
#[cfg(test)]
mod tests {
- use super::calculate_percentile;
+ use super::*;
+ use arrow::array::Float64Array;
use half::f16;
+ #[test]
+ fn retract_batch_errors_on_untracked_value() {
+ let mut acc = PercentileContAccumulator::<Float64Type>::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
+ // 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::<Float64Type>::new(0.5);
+ dense_acc
+ .update_batch(std::slice::from_ref(&dense))
+ .unwrap();
+ let mut sparse_acc = PercentileContAccumulator::<Float64Type>::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::<arrow::datatypes::Float16Type>(&mut values, 0.5)
- .expect("non-empty input");
+ let result = calculate_percentile::<Float16Type>(&mut values, 0.5)
+ .expect("non-empty input");
let result_f = result.to_f32();
assert!(
!result_f.is_nan(),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org