Repository navigation
[branch-55] fix: ensure deferred-filtered outer joins preserve streamed output order (#24573) - #27358
Merged
Merged
Conversation
…der (apache#24573) `LEFT`/`RIGHT`/`FULL` sort-merge joins with a join filter could return rows out of order. These join types advertise that they preserve the ordering of one input (`maintains_input_order` is `[true, false]` for `LEFT`), so downstream operators are allowed to rely on it. Deferred-filtered joins stage their output in a second `BatchCoalescer` (`self.output`) because the filter correction step emits ragged batch sizes. The final flush at end-of-input bypassed that buffer and emitted its batch directly, so any rows still buffered in `output` from an earlier flush were emitted *after* it. A `LEFT JOIN` where some keys match large buffered groups and the trailing keys match a single row each reproduces this: the large groups trip the flush gate and push sub-threshold batches that stay buffered, while the trailing keys never trip the gate and land in the final flush. Streamed keys came back as `[5, 6, 0, 1, 2, 3, 4]` instead of `[0, 1, 2, 3, 4, 5, 6]`. Bug fix: - `on_children_exhausted` now pushes the final filtered batch into `self.output` instead of emitting it directly, so all deferred-filtered output leaves through a single buffer and stays in order. Cleanups in the same file, no behavior change: - `emit_completed_output` drains every completed batch from `self.output`; previously each flush emitted at most one and left the rest buffered. - `join_arrays` returns `Result` instead of `unwrap()`-ing. A failing join-key expression previously panicked the worker thread. `StreamedBatch::new` and `BufferedBatch::new` became `try_new`. - `materialize_right_columns` maps buffered batch indices to interleave sources with a linear scan instead of a `HashMap` — a key group spans a handful of batches at most, and this ran per matched chunk. - Extracted `new_output_coalescer`, replacing four copies of the same `BatchCoalescer::new(..).with_biggest_coalesce_batch_size(..)` construction. Yes. Added `left_join_with_filter_preserves_streamed_order`, which builds the mixed group-size shape described above and asserts the streamed key column comes back in order. It fails on `main` with `[5, 6, 0, 1, 2, 3, 4]`. Also ran the full `datafusion-physical-plan` test suite and the joins sqllogictests. Yes — outer sort-merge joins with a join filter now return rows in the order the operator claims to produce them. No API changes. (cherry picked from commit d1fe988) Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
9 of 23 tasks
alamb
approved these changes
Nov 12, 2026
alamb
left a comment
alamb
left a comment
Contributor
There was a problem hiding this comment.
Looks good -- thank you. I verified this is a clean cherry pick
Codecov Report❌ Patch coverage is
@@ Coverage Diff @@
## branch-55 #27358 +/- ##
=============================================
- Coverage 81.24% 81.24% -0.01%
=============================================
Files 1110 1110
Lines 389341 389351 +10
Branches 389341 389351 +10
=============================================
+ Hits 316324 316331 +7
+ Misses 54446 54440 -6
- Partials 18571 18580 +9
... and 7 files with indirect coverage changes 🚀 New features to boost your workflow:
|
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
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?
55.3.0(patch) Release (Nov 2026) #26868 (the 55.3.0 patch release tracking issue).branch-55.Rationale for this change
LEFT,RIGHTandFULLsort-merge joins with a join filter can return rows out of order. These join types advertise that they preserve the order of one input (maintains_input_orderis[true, false]forLEFT), so an operator above is allowed to rely on it. Deferred-filtered joins stage their output in a secondBatchCoalescer(self.output), but the final flush at end of input bypassed it and emitted its batch directly, so rows still buffered inoutputfrom an earlier flush came out after it.A
LEFT JOINwhere some keys match large buffered groups and the trailing keys match one row each reproduces it: the streamed keys come back as[5, 6, 0, 1, 2, 3, 4]instead of[0, 1, 2, 3, 4, 5, 6]. Through DataFusion Comet we saw a native aggregate above such a join trust the advertised order, close a group early and return it twice (8 rows where Spark returns 7). This is a wrong result, which makes it eligible for backport.#24573 fixed it on
main(labelledv56.0.0); it is onbranch-56but not onbranch-55, so it is in neither 55.1.0 nor 55.2.0. It was requested for 55.3.0 on #26868 (#26868 (comment)).What changes are included in this PR?
git cherry-pick -xof d1fe988 (fix: ensure deferred-filtered outer joins preserve streamed output order #24573, by @jayzhan211).datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs: the final filtered batch is pushed intoself.outputinstead of being emitted directly, andemit_completed_outputdrains every completed batch fromself.output. Also from fix: ensure deferred-filtered outer joins preserve streamed output order #24573, with no behaviour change:join_arraysreturnsResultinstead ofunwrap()-ing (StreamedBatch::new/BufferedBatch::newbecometry_new, bothpub(super)),materialize_right_columnsmaps buffered batch indices to interleave sources with a direct-addressed table instead of aHashMap, andnew_output_coalescerreplaces four copies of the same construction.datafusion/physical-plan/src/joins/sort_merge_join/tests.rsanddatafusion/physical-plan/benches/sort_merge_join.rs: the tests and benchmark case from fix: ensure deferred-filtered outer joins preserve streamed output order #24573.tests.rs: the cherry-pick also carried two unrelated tests frommain(swap_inputs_swaps_the_projection,an_empty_projection_keeps_the_rows) that cover code which is not on this branch. They are dropped. The diffstat equals fix: ensure deferred-filtered outer joins preserve streamed output order #24573's (3 files, +571/-91).Are these changes tested?
Yes. On this branch (Rust 1.95.0) I ran:
cargo test -p datafusion-physical-plan --lib: 1705 passed, including the five tests fix: ensure deferred-filtered outer joins preserve streamed output order #24573 added.materializing_stream.rsreverted to this branch's version and the tests kept,cargo test -p datafusion-physical-plan --lib sort_merge_joinfails the three streamed-order tests (left_join_with_filter_preserves_streamed_order,right_join_with_filter_preserves_streamed_order,left_join_with_partial_filter_preserves_streamed_order).cargo test -p datafusion-sqllogictest --test sqllogictests -- join: all 16 matching files pass.cargo check -p datafusion-physical-plan --benchescompiles the changed benchmark.cargo fmt --all -- --checkandcargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warningspass.Are there any user-facing changes?
Outer sort-merge joins with a join filter return rows in the order they advertise. No public API change.
🤖 Generated with Claude Code