refactor(hash-aggr): Support spilling for partial and final mode aggregation - #24061
refactor(hash-aggr): Support spilling for partial and final mode aggregation#240612010YOUY01 wants to merge 1 commit into
partial and final mode aggregation#24061Conversation
2010YOUY01
changed the title
partial and final mode aggregation
| /// into an ordered streaming aggregation, which ensures bounded memory usage and | ||
| /// evaluates the final result. | ||
| /// - [`OrderedFinalAggregateStream`] is reused for the streaming aggregation. | ||
| pub(crate) struct PartialHashAggregateStream { |
There was a problem hiding this comment.
We could put partial and final stream to two different files, I plan to this after this PR.
| | 3 | 1 | 2.0 | | ||
| | 3 | 2 | 5.0 | | ||
| | 4 | 3 | 11.0 | | ||
| | 4 | 1 | 4.0 | |
There was a problem hiding this comment.
they're both valid partial aggregation output, due to early emitting under memory limit.
| // enlarge memory limit to let the final aggregation finish | ||
| new_spill_ctx(2, 2600) | ||
| // Enlarge the memory limit enough to replay spilled states. | ||
| new_spill_ctx(2, 4640) |
There was a problem hiding this comment.
The legacy implementation underestimate the memory usage somehow, so here it relaxes the memory budget needs further invstigation
|
|
||
| assert_eq!(3, output_rows); | ||
| if spill { | ||
| // When spilling, the output rows metrics become partial output size + final output size |
There was a problem hiding this comment.
This should be a bug in the legacy implementation. Final aggregation should return the same output_rows regardless of the spilling condition.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24061 +/- ##
========================================
Coverage 80.88% 80.88%
========================================
Files 1101 1101
Lines 375720 376048 +328
Branches 375720 376048 +328
========================================
+ Hits 303895 304164 +269
- Misses 53729 53774 +45
- Partials 18096 18110 +14 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
2 participants
Which issue does this PR close?
Part of #22710
Rationale for this change
This PR adds existing spilling feature into the new 2-staged (partial and final) aggregation. The high-level implementation idea is the same as the legacy implementation.
For the algorithm description for this feature, see top comment change at
datafusion/physical-plan/src/aggregates/hash_stream.rsWhat changes are included in this PR?
The key changes to the operator state machine are:
In file
datafusion/physical-plan/src/aggregates/hash_stream.rsPartialHashAggregateStream::poll_next()FinalHashAggregateStream::poll_next()Use this as the starting point, you can navigate to all the related changes, for example adding new states to implement larger-than-memory execution.
Are these changes tested?
Existing tests
codecovAre there any user-facing changes?
No