chore: refactor SortMergeJoin bitwise stream to generators and simplify to be textbook like as possible - #23761
Conversation
|
run benchmark smj |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing let-claude-try-refactor-bitwise-stream-to-use-generator (d3e32e9) to ed22a4a (merge-base) diff using: smj File an issue against this benchmark runner |
|
Love to see a negative line count! |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagesmj — base (merge-base)
smj — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #23761 +/- ##
==========================================
- Coverage 80.67% 80.66% -0.01%
==========================================
Files 1094 1095 +1
Lines 371770 372197 +427
Branches 371770 372197 +427
==========================================
+ Hits 299907 300233 +326
- Misses 53964 54032 +68
- Partials 17899 17932 +33 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…e when waiting for inputs
Now its not due to added tests to make sure we are not counting child/consumer time, but the stream file still have negative line count! |
|
@mbutrovich Can you please review? 😅 |
I'll find some time this week, definitely. Thanks for the PR! |
|
Out of curiosity, is the non-bitwise SMJ stream up next? |
if I can make it more readable yes, afterward I will try to migrate the new hash aggregate exec |
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks for tackling this, @rluvaton! It reads much better now.
Traced the control flow by hand against the removed poll-based version: every branch (outer/inner key groups crossing batch boundaries, filtered and unfiltered, spill buffer vs spill file, inner exhaustion mid-group) produces the same state transitions. PendingBoundary, batch_emitted, buffering_inner_pending, and spill_stream_has_data only existed to survive unwinding out of poll_join on Pending; with await the function suspends in place, so removing them is correct, not just simpler.
The two new sync fast paths (try_skip_inner_key_group, try_process_key_match) skip the async machinery when a key group does not reach a batch boundary. Checked their output against the general path for that case and they compute the same cursor and matched-bit state.
CI is green (clippy, cargo test, extended_tests), the existing semi/anti/mark fuzz tests in join_fuzz.rs exercise this stream end to end, and the posted smj benchmark shows no regression.
One request before merge: datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:744 notes that the clock now keeps running across spill read-back, where before a genuine Pending from the spill stream stopped the per-call timer early and that wait was invisible to join_time. That is a real change to what the metric measures, not just an implementation detail. Please add a line to the PR description calling this out, since it currently says "no user-facing changes" and join_time is visible in EXPLAIN ANALYZE.
4 participants
Which issue does this PR close?
N/A
Rationale for this change
SortMergeJoin bitwise stream implementation is very complex and hard to understand while on paper it should be pretty simple.
the reason for that is we have to store state between polls (we had
boundary) and handle the case where both can getPoll::Pendingfrom child andPoll::Readyfrom child which further complicate the codeWhat changes are included in this PR?
(this was entirely written by Claude Fable, sorry, I tried manually but the code was too complex to hold in my head 😅 )
Are these changes tested?
existing tests
Are there any user-facing changes?
The join_time now includes the time to read from the async spill stream between pending which is arguable more correct since this time is part of the operator, although long waits between pending calls will be counted in the op
join_timewhile the alternative is not counting the read from file and decoding...