neilconway commented on code in PR #23913:
URL: https://github.com/apache/datafusion/pull/23913#discussion_r3666314619
##########
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:
This needs an `ORDER BY` to be deterministic, or else use `rowsort` SLT
directive.
##########
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:
`GenericDistinctBuffer` has a fast-path for null-free arrays; might be worth
adopting here as well.
##########
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:
`GenericDistinctBuffer` previously used foldhash, but this will switch to
using siphash, which is a lot slower.
##########
datafusion/functions-aggregate/src/percentile_cont.rs:
##########
@@ -713,7 +745,12 @@ where
let arr = values[0].as_primitive::<T>();
for value in arr.iter().flatten() {
- self.distinct_values.values.remove(&Hashable(value));
+ if let Some(count) = self.counts.get_mut(&Hashable(value)) {
+ *count -= 1;
+ if *count == 0 {
+ self.counts.remove(&Hashable(value));
+ }
+ }
Review Comment:
wdyt of raising an error if the caller attempts to retract a value that
isn't in the map?
##########
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:
`estimate_memory_size::<(Hashable<T::Native>,
usize)>(self.counts.capacity(), size_of_val(self))` is better I think.
--
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]
