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

Author: viirya

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<T: ArrowNumericType> {
     /// Distinct value -> number of in-window rows carrying it.
-    counts: HashMap<Hashable<T::Native>, 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<Hashable<T::Native>, 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::<T>();
-        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::<Hashable<T::Native>>() + size_of::<usize>())
+        estimate_memory_size::<(Hashable<T::Native>, 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::<T>();
-        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(())
     }

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