sunchao commented on code in PR #26726:
URL: https://github.com/apache/datafusion/pull/26726#discussion_r4379973279
##########
datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs:
##########
@@ -319,13 +321,10 @@ impl Aggregating {
let output = match
reservation.try_resize(self.table.memory_size()) {
Ok(()) => self.table.take_completed_state_batch()?,
Err(oom @ DataFusionError::ResourcesExhausted(_)) => {
- // Partial group clustering may have an unbounded active
key range.
- // The final stage can merge incomplete states emitted
here.
- if matches!(self.table.group_clustering(),
GroupClustering::Full(_)) {
-
Some(self.table.take_completed_state_batch()?.ok_or(oom)?)
- } else {
- Some(self.table.take_state_batch()?.ok_or(oom)?)
- }
+ // Emit every state, including the active groups, and
reset the
+ // table to release the capacity it grew for this batch.
The
+ // final stage merges the states of a group split across
outputs.
+ Some(self.table.take_state_batch()?.ok_or(oom)?)
Review Comment:
Splitting the active group makes the advertised `Unique(k)` constraint
incorrect even within one fully clustered partition. I used the two sorted
batches from `completed_groups_under_memory_pressure`, with `v=9` and `v=1` for
the two occurrences of `k=256`. With a 13,000-byte limit, `SortExec(k ASC,
sum_state ASC)` directly above the partial returned `(256, 9)` then `(256, 1)`.
`ordering_satisfy` accepted the secondary ordering because `k` was marked
unique, so `SortExec` passed the rows through. Restoring the base emission gave
`(256, 10)` at the same limit. Could you keep the ordering by `k` but withhold
group-key uniqueness for partial outputs that can split, and add a regression
for the secondary sort? [Constraint
declaration](https://github.com/apache/datafusion/blob/67ce0599bcd40999539c767859daac1da4e40d64/datafusion/physical-plan/src/aggregates/mod.rs#L2010-L2025)
· [Sort fast
path](https://github.com/apache/datafusion/blob/67ce0599bcd40999539c767859daac1da4e40d64/datafusi
on/physical-plan/src/sorts/sort.rs#L1607-L1619)
##########
datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs:
##########
@@ -319,13 +321,10 @@ impl Aggregating {
let output = match
reservation.try_resize(self.table.memory_size()) {
Ok(()) => self.table.take_completed_state_batch()?,
Err(oom @ DataFusionError::ResourcesExhausted(_)) => {
- // Partial group clustering may have an unbounded active
key range.
- // The final stage can merge incomplete states emitted
here.
- if matches!(self.table.group_clustering(),
GroupClustering::Full(_)) {
-
Some(self.table.take_completed_state_batch()?.ok_or(oom)?)
- } else {
- Some(self.table.take_state_batch()?.ok_or(oom)?)
- }
+ // Emit every state, including the active groups, and
reset the
+ // table to release the capacity it grew for this batch.
The
Review Comment:
The reset still leaves batch-sized scratch in `APPROX_DISTINCT`. I ran a
fully clustered partial with `APPROX_DISTINCT(k)` and `APPROX_DISTINCT(v)`, one
sorted batch of 8,192 distinct `Int64` keys, and `v=42`. Each
`HllGroupsAccumulator` keeps a 64 KiB `hashes` vector after
`state(EmitTo::All)`, so reserving the empty table still fails at the PR’s 100
KiB limit. Releasing those vectors on the full drain let the same run emit all
8,192 rows. Could the pressure reset release this accumulator scratch as well,
with a regression for this case? [State
emission](https://github.com/apache/datafusion/blob/67ce0599bcd40999539c767859daac1da4e40d64/datafusion/functions-aggregate/src/approx_distinct.rs#L593-L605)
· [Scratch
accounting](https://github.com/apache/datafusion/blob/67ce0599bcd40999539c767859daac1da4e40d64/datafusion/functions-aggregate/src/approx_distinct.rs#L640-L644)
--
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]
