Sunninsky opened a new pull request, #27358: URL: https://github.com/apache/datafusion/pull/27358
## Which issue does this PR close? - Part of #26868 (the 55.3.0 patch release tracking issue). - Backport of #24573 to `branch-55`. ## Rationale for this change `LEFT`, `RIGHT` and `FULL` sort-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_order` is `[true, false]` for `LEFT`), so an operator above is allowed to rely on it. Deferred-filtered joins stage their output in a second `BatchCoalescer` (`self.output`), but the final flush at end of input bypassed it and emitted its batch directly, so rows still buffered in `output` from an earlier flush came out *after* it. A `LEFT JOIN` where 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` (labelled `v56.0.0`); it is on `branch-56` but not on `branch-55`, so it is in neither 55.1.0 nor 55.2.0. It was requested for 55.3.0 on #26868 (https://github.com/apache/datafusion/issues/26868#issuecomment-6791317393). ## What changes are included in this PR? - A `git cherry-pick -x` of d1fe98894238dd908a874a88c21c12a102ef5b92 (#24573, by @jayzhan211). - `datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs`: the final filtered batch is pushed into `self.output` instead of being emitted directly, and `emit_completed_output` drains every completed batch from `self.output`. Also from #24573, with no behaviour change: `join_arrays` returns `Result` instead of `unwrap()`-ing (`StreamedBatch::new` / `BufferedBatch::new` become `try_new`, both `pub(super)`), `materialize_right_columns` maps buffered batch indices to interleave sources with a direct-addressed table instead of a `HashMap`, and `new_output_coalescer` replaces four copies of the same construction. - `datafusion/physical-plan/src/joins/sort_merge_join/tests.rs` and `datafusion/physical-plan/benches/sort_merge_join.rs`: the tests and benchmark case from #24573. - One conflict, in `tests.rs`: the cherry-pick also carried two unrelated tests from `main` (`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 #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 #24573 added. - With `materializing_stream.rs` reverted to this branch's version and the tests kept, `cargo test -p datafusion-physical-plan --lib sort_merge_join` fails 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 --benches` compiles the changed benchmark. - `cargo fmt --all -- --check` and `cargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warnings` pass. ## 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](https://claude.com/claude-code) -- 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]
