From aefe0657b11971fd7355769bc4acee13255fd805 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 28 Jul 2026 08:32:50 -0700 Subject: [PATCH 1/4] Address review feedback for percentile_cont(DISTINCT) accumulator Follow-up to the DistinctPercentileContAccumulator fix, addressing post-merge review comments: - Use the fast foldhash RandomState for the `counts` map instead of the standard library's default SipHash, matching GenericDistinctBuffer. - Compute size() via estimate_memory_size, mirroring the other distinct accumulators. - Add a null-free fast path to update_batch and retract_batch that skips the per-element validity check when the input array has no nulls. - Add ORDER BY id to the sliding-window regression query in aggregate.slt so its output row order is deterministic. Co-authored-by: Claude Code --- .../src/percentile_cont.rs | 39 +++++++++++++++---- .../sqllogictest/test_files/aggregate.slt | 3 +- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs index 4b5e892fb5cdf..b9acab88a1789 100644 --- a/datafusion/functions-aggregate/src/percentile_cont.rs +++ b/datafusion/functions-aggregate/src/percentile_cont.rs @@ -32,8 +32,10 @@ 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_common::utils::memory::estimate_memory_size; use datafusion_functions_aggregate_common::noop_accumulator::NoopAccumulator; use crate::min_max::{max_udaf, min_udaf}; @@ -672,7 +674,11 @@ where #[derive(Debug)] struct DistinctPercentileContAccumulator { /// Distinct value -> number of in-window rows carrying it. - counts: HashMap, usize>, + /// + /// Uses the same fast (foldhash) `RandomState` as the shared + /// `GenericDistinctBuffer` rather than the standard library's default + /// SipHash, which is considerably slower for this hot path. + counts: HashMap, usize, RandomState>, percentile: f64, } @@ -709,8 +715,15 @@ where // `values` may carry extra argument columns (e.g. the percentile // literal); only the first column holds the aggregated values. let arr = values[0].as_primitive::(); - for value in arr.iter().flatten() { - *self.counts.entry(Hashable(value)).or_default() += 1; + if arr.null_count() > 0 { + for value in arr.iter().flatten() { + *self.counts.entry(Hashable(value)).or_default() += 1; + } + } else { + // Fast path: no nulls, so skip the per-element validity check. + for value in arr.values().iter() { + *self.counts.entry(Hashable(*value)).or_default() += 1; + } } Ok(()) } @@ -733,9 +746,11 @@ where } fn size(&self) -> usize { - size_of_val(self) - + self.counts.capacity() - * (size_of::>() + size_of::()) + estimate_memory_size::<(Hashable, usize)>( + self.counts.capacity(), + size_of_val(self), + ) + .unwrap() } fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { @@ -744,13 +759,23 @@ where } let arr = values[0].as_primitive::(); - for value in arr.iter().flatten() { + let mut decrement = |value: T::Native| { if let Some(count) = self.counts.get_mut(&Hashable(value)) { *count -= 1; if *count == 0 { self.counts.remove(&Hashable(value)); } } + }; + if arr.null_count() > 0 { + for value in arr.iter().flatten() { + decrement(value); + } + } else { + // Fast path: no nulls, so skip the per-element validity check. + for value in arr.values().iter() { + decrement(*value); + } } Ok(()) } diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index f6380c9a5f06d..0d4c902ec8a6f 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -1122,7 +1122,8 @@ SELECT percentile_cont(DISTINCT x, 0.5) FROM distinct_pct; query IR SELECT id, percentile_cont(DISTINCT x, 0.5) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) -FROM distinct_pct; +FROM distinct_pct +ORDER BY id; ---- 1 5 2 5 From 4037b9443a6b4f5dad60ba5cf8fafd7fd824f6c8 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 28 Jul 2026 08:55:38 -0700 Subject: [PATCH 2/4] test: cover merge/state for distinct percentile_cont MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a grouped percentile_cont(DISTINCT ...) regression test. A GROUP BY with target_partitions=4 forces two-phase (Partial + FinalPartitioned) aggregation, so the distinct accumulator's state() and merge_batch() paths run — the paths the previous whole-table regression queries did not reach (flagged by patch-coverage review on #23913). Duplicate values within each group also verify multiplicity is preserved across the per-partition merge. Co-authored-by: Claude Code --- .../sqllogictest/test_files/aggregate.slt | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index 0d4c902ec8a6f..540e5edeb283e 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -1132,6 +1132,25 @@ ORDER BY id; statement ok DROP TABLE distinct_pct; +# Regression: grouped percentile_cont(DISTINCT ...) forces two-phase +# (Partial + FinalPartitioned) aggregation, exercising the distinct +# accumulator's state()/merge_batch() paths. Duplicate values within a +# group must be de-duplicated across the per-partition merge. +# Group 1 distinct {1,5,9} -> median 5; group 2 distinct {3,7} -> median 5. +statement ok +CREATE TABLE grp_distinct_pct(g INT, x DOUBLE) AS VALUES + (1, 5), (1, 5), (1, 9), (1, 1), + (2, 7), (2, 7), (2, 3); + +query IR +SELECT g, percentile_cont(DISTINCT x, 0.5) FROM grp_distinct_pct GROUP BY g ORDER BY g; +---- +1 5 +2 5 + +statement ok +DROP TABLE grp_distinct_pct; + query RT select approx_median(arrow_cast(col_f32, 'Float16')), arrow_typeof(approx_median(arrow_cast(col_f32, 'Float16'))) from median_table; ---- From 2876db42f308bcf7dd9fd11bb7f4691ecf660e4a Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 28 Jul 2026 08:59:11 -0700 Subject: [PATCH 3/4] fix: error on retracting an untracked value in percentile_cont(DISTINCT) Per review discussion: if `retract_batch` is asked to remove a value not in the count map, the accumulator's state has diverged from the window frame and continuing would silently produce wrong results. Return an `internal_err!` instead of skipping, so the inconsistency surfaces rather than yielding a wrong answer. Co-authored-by: Claude Code --- .../src/percentile_cont.rs | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs index b9acab88a1789..baf1978ea24af 100644 --- a/datafusion/functions-aggregate/src/percentile_cont.rs +++ b/datafusion/functions-aggregate/src/percentile_cont.rs @@ -760,21 +760,30 @@ where let arr = values[0].as_primitive::(); let mut decrement = |value: T::Native| { - if let Some(count) = self.counts.get_mut(&Hashable(value)) { - *count -= 1; - if *count == 0 { - self.counts.remove(&Hashable(value)); + match self.counts.get_mut(&Hashable(value)) { + Some(count) => { + *count -= 1; + if *count == 0 { + self.counts.remove(&Hashable(value)); + } + Ok(()) } + // Retracting a value that isn't tracked means the accumulator + // state has diverged from the window frame; continuing would + // silently produce wrong results, so surface it as an error. + None => internal_err!( + "percentile_cont(DISTINCT) retract_batch: retracted a value not present in the window" + ), } }; if arr.null_count() > 0 { for value in arr.iter().flatten() { - decrement(value); + decrement(value)?; } } else { // Fast path: no nulls, so skip the per-element validity check. for value in arr.values().iter() { - decrement(*value); + decrement(*value)?; } } Ok(()) From 11cec2d6cd4c14b3b77f00e689cb871979c0881a Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 28 Jul 2026 13:08:10 -0700 Subject: [PATCH 4/4] test: cover NULL slow path in distinct percentile_cont Add a sliding-window percentile_cont(DISTINCT ...) regression over data containing a NULL positioned so that both the update_batch (NULL row enters the frame) and retract_batch (NULL row leaves the frame) code paths take the null_count() > 0 branch. Verified via temporary eprintln that both branches execute; the previous distinct-percentile tests used no NULLs and left those branches uncovered. Co-authored-by: Claude Code --- .../sqllogictest/test_files/aggregate.slt | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index 540e5edeb283e..26b8a78f3921a 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -1151,6 +1151,27 @@ SELECT g, percentile_cont(DISTINCT x, 0.5) FROM grp_distinct_pct GROUP BY g ORDE statement ok DROP TABLE grp_distinct_pct; +# Regression: sliding-window percentile_cont(DISTINCT ...) over data with +# NULLs exercises the null_count() > 0 slow path in BOTH update_batch (a NULL +# row enters the frame) and retract_batch (a NULL row leaves the frame as the +# window slides). NULLs are ignored; distinct dedups the non-null values. +statement ok +CREATE TABLE distinct_pct_nulls(id INT, x DOUBLE) AS VALUES + (1, 5), (2, NULL), (3, 9), (4, 5); + +query IR +SELECT id, percentile_cont(DISTINCT x, 0.5) + OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) +FROM distinct_pct_nulls ORDER BY id; +---- +1 5 +2 5 +3 9 +4 7 + +statement ok +DROP TABLE distinct_pct_nulls; + query RT select approx_median(arrow_cast(col_f32, 'Float16')), arrow_typeof(approx_median(arrow_cast(col_f32, 'Float16'))) from median_table; ----