chore: convert PartialHashAggregateStream to async generators and cleanup - #24017
Open
rluvaton wants to merge 16 commits into
Open
chore: convert PartialHashAggregateStream to async generators and cleanup#24017rluvaton wants to merge 16 commits into
PartialHashAggregateStream to async generators and cleanup#24017rluvaton wants to merge 16 commits into
Conversation
Member
Author
|
run benchmarks |
13 tasks
Member
Author
|
run benchmark tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing move-final-hash-stream (8a49479) to 541caab (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing move-final-hash-stream (8a49479) to 541caab (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing move-final-hash-stream (8a49479) to 541caab (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing move-final-hash-stream (8a49479) to 541caab (merge-base) diff using: tpcds File an issue against this benchmark runner |
16 tasks
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #24017 +/- ##
==========================================
- Coverage 81.62% 81.62% -0.01%
==========================================
Files 1123 1123
Lines 409524 409400 -124
Branches 409524 409400 -124
==========================================
- Hits 334294 334179 -115
+ Misses 55594 55572 -22
- Partials 19636 19649 +13 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
Could we delay those cleanup PRs after #22710 is functionally complete? Now we're in a inconsistent state, and I think it's better to fully remove the legacy implementation sooner, and next we can continue cleaning things up. I estimate it's ~3 PRs away from that, I'll update in the EPIC issue when it's ready. |
alamb
marked this pull request as draft
August 3, 2026 20:34
# Conflicts: # datafusion/physical-plan/src/aggregates/hash_stream.rs
# Conflicts: # datafusion/physical-plan/src/aggregates/hash_stream.rs
rluvaton
marked this pull request as ready for review
September 2, 2026 09:29
2010YOUY01
reviewed
Sep 2, 2026
| /// States for final hash aggregation processing. | ||
| // The typestate pattern is used in case the inner logic becomes more complex in | ||
| // the future. | ||
| enum FinalHashAggregateState { |
Contributor
There was a problem hiding this comment.
Is it possible to keep the explicit states, and only use the emitter pattern to get rid of the verbose control flow implementation?
If we measure complexity as the cognitive load to understand everything, instead of LoC, I think this explicit state representation approach is better.
As I currently understand it, the heuristic to follow is:
- If the state space is simple (2-3 states, trivial to understand, like the cross join we did in refactor: CrossJoinStream (simplifying, less state, async generator pattern) #24291), such an enum is unnecessary.
- If we have to draw a state machine graph to understand the inner logic, like here, keep the
enumstates.
Additionally, I didn't try to polish the state splits originally, so there might be some room to simplify them, like merging sibling states to make it less verbose.
Member
Author
There was a problem hiding this comment.
The comment kept for high level understanding, I did not measure by LoC but by cognitive load, with the async generators it is much easier for me understand the code, and the flow is clearer.
Anyway, I extracted the final hash aggregate to a separate pr with updated comments
Contributor
There was a problem hiding this comment.
Interesting... I feel the opposite way.
If we don't encode the states explicitly, then the states/transitions have to be encoded with local variables and hidden assumptions, and to understand that we have to load everything into short-term memory, which is hard.
The benefit of an explicit state representation is that it's a 1:1 mapping from the mental model to the code:
- It's possible to understand each state separately
- It's safer: impossible states are unrepresentable, and it's easy to guard with sanity checks
It's quite understandable for me now, because I'm already familiar with it; for someone without prior knowledge trying to understand it, or if it later evolves into a different shape, I think it will be more challenging to understand.
Member
Author
There was a problem hiding this comment.
the state machine is unneded since it is almost linear with async generators for the final stream
you read from input, if you dont have memory, spill and continue reading
once finished, if have spills, spill the in pending data in hash table and switch to ordered final aggregate stream
if no spills, just output batches from hash table.
Flow
Contributor
There was a problem hiding this comment.
the state machine is unneded since it is almost linear with async generators for the final stream
This point makes sense to me. If we can model the state graph to a linear (no backwards edge) shape, maybe it's good to eliminate enum states.
I'd like to do some prototyping myself to figure out the pattern.
Contributor
|
#24877 tries to apply generator pattern together with explicit states, I'm wondering how you guys think. I'm lean towards keeping the explicit states, because it's more like controlling complexity with a mechanism instead of with discipline. It's also a pattern that can be used brainlessly for more complex state graphs, like joins with spilling (lots of loop edges), and we don't have to invent different patterns for different operators for consistency. I think this PR is pretty clean now, and the quality is maintained by good practice. However, if we don't rely on explicit states, I think it's more likely to get poisoned in the future (e.g. adding mysterious local flags for new features, making it confusing again). |
# Conflicts: # datafusion/physical-plan/src/aggregates/hash_stream.rs
rluvaton
changed the title
PartialHashAggregateStream and FinalHashAggregateStream to async generators and cleanupPartialHashAggregateStream to async generators and cleanup
Member
Author
|
@2010YOUY01 I've cleaned up and add something like state machine, can you please take another look |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Related to:
Rationale for this change
Cleanup the code and remove state
What changes are included in this PR?
changed
PartialHashAggregateStreamto async generator and remove unneeded code due to thatAre these changes tested?
existing tests
Are there any user-facing changes?
nope