jayzhan211 opened a new pull request, #24573:
URL: https://github.com/apache/datafusion/pull/24573
## Rationale for this change
`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]`.
## What changes are included in this PR?
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.
## Are these changes tested?
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.
## Are there any user-facing changes?
Yes — outer sort-merge joins with a join filter now return rows in the order
the
operator claims to produce them. No API changes.
--
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]
