viirya commented on code in PR #23913:
URL: https://github.com/apache/datafusion/pull/23913#discussion_r3667027909
##########
datafusion/functions-aggregate/src/percentile_cont.rs:
##########
@@ -662,16 +662,24 @@ where
}
}
+/// Sliding-window–capable accumulator for `percentile_cont(DISTINCT ...)`.
+///
+/// Distinct values are tracked with a per-value multiplicity count (how many
+/// rows currently in the window carry that value) rather than a plain set, so
+/// that `retract_batch` only drops a value once *all* of its occurrences have
+/// left the window frame. The percentile is then computed over the set of keys
+/// with a positive count.
#[derive(Debug)]
struct DistinctPercentileContAccumulator<T: ArrowNumericType> {
- distinct_values: GenericDistinctBuffer<T>,
+ /// Distinct value -> number of in-window rows carrying it.
+ counts: HashMap<Hashable<T::Native>, usize>,
Review Comment:
Good catch — switched the count map to the foldhash `RandomState` that
`GenericDistinctBuffer` uses, instead of the std default SipHash. Fixed in
#23946.
##########
datafusion/sqllogictest/test_files/aggregate.slt:
##########
@@ -1104,6 +1104,33 @@ ORDER BY tags, timestamp;
statement ok
DROP TABLE median_window_test;
+# Regression: percentile_cont(DISTINCT ...) used to forward the extra
+# percentile-argument column into the distinct-values buffer (which asserts a
+# single input array), panicking on every distinct query. Plain aggregate:
+statement ok
+CREATE TABLE distinct_pct(id INT, x DOUBLE) AS VALUES
+ (1, 5), (2, 5), (3, 9);
+
+query R
+SELECT percentile_cont(DISTINCT x, 0.5) FROM distinct_pct;
+----
+7
+
+# Regression: distinct sliding-window percentile must count value multiplicity
+# on retract. Row 3's frame is {5, 9}; the row-1 `5` leaves the frame but the
+# row-2 `5` remains, so the distinct set is still {5, 9} (median 7), not {9}.
+query IR
+SELECT id, percentile_cont(DISTINCT x, 0.5)
+ OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+FROM distinct_pct;
Review Comment:
Added `ORDER BY id` to make the row order deterministic. Fixed in #23946.
##########
datafusion/functions-aggregate/src/percentile_cont.rs:
##########
@@ -684,26 +692,50 @@ where
f64: AsPrimitive<T::Native>,
{
fn state(&mut self) -> Result<Vec<ScalarValue>> {
- self.distinct_values.state()
+ // Emit the distinct keys as a single List scalar, matching the state
+ // shape declared in `state_fields` (a List of the input type). Counts
+ // are window-local bookkeeping and are intentionally not serialized:
+ // cross-partition merges only need the distinct key set.
+ let arr = Arc::new(
+ PrimitiveArray::<T>::from_iter_values(self.counts.keys().map(|v|
v.0))
+ .with_data_type(T::DATA_TYPE),
+ );
+ Ok(vec![
+ SingleRowListArrayBuilder::new(arr).build_list_scalar(),
+ ])
}
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.distinct_values.update_batch(values)
+ // `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;
+ }
+ Ok(())
}
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
- self.distinct_values.merge_batch(states)
+ let list = states[0].as_list::<i32>();
+ for values in list.iter().flatten() {
+ let arr = values.as_primitive::<T>();
+ for value in arr.iter().flatten() {
+ *self.counts.entry(Hashable(value)).or_default() += 1;
+ }
+ }
+ Ok(())
}
fn evaluate(&mut self) -> Result<ScalarValue> {
- let mut values: Vec<T::Native> =
- self.distinct_values.values.iter().map(|v| v.0).collect();
+ let mut values: Vec<T::Native> = self.counts.keys().map(|v|
v.0).collect();
let value = calculate_percentile::<T>(&mut values, self.percentile);
ScalarValue::new_primitive::<T>(value, &T::DATA_TYPE)
}
fn size(&self) -> usize {
- size_of_val(self) + self.distinct_values.size()
+ size_of_val(self)
+ + self.counts.capacity()
+ * (size_of::<Hashable<T::Native>>() + size_of::<usize>())
Review Comment:
Switched to `estimate_memory_size::<(Hashable<T::Native>,
usize)>(self.counts.capacity(), size_of_val(self))`. Fixed in #23946.
##########
datafusion/functions-aggregate/src/percentile_cont.rs:
##########
@@ -684,26 +692,50 @@ where
f64: AsPrimitive<T::Native>,
{
fn state(&mut self) -> Result<Vec<ScalarValue>> {
- self.distinct_values.state()
+ // Emit the distinct keys as a single List scalar, matching the state
+ // shape declared in `state_fields` (a List of the input type). Counts
+ // are window-local bookkeeping and are intentionally not serialized:
+ // cross-partition merges only need the distinct key set.
+ let arr = Arc::new(
+ PrimitiveArray::<T>::from_iter_values(self.counts.keys().map(|v|
v.0))
+ .with_data_type(T::DATA_TYPE),
+ );
+ Ok(vec![
+ SingleRowListArrayBuilder::new(arr).build_list_scalar(),
+ ])
}
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- self.distinct_values.update_batch(values)
+ // `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;
+ }
Review Comment:
Adopted the same null-free fast path in `update_batch` and `retract_batch`.
Fixed in #23946.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
