fix: ensure deferred-filtered outer joins preserve streamed output order

[jayzhan211]

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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 88.50575% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.45%. Comparing base (e14bd12) to head (2e303b6).
⚠️ Report is 33 commits behind head on main.

Files with missing lines	Patch %	Lines
.../src/joins/sort_merge_join/materializing_stream.rs	88.50%	2 Missing and 8 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24573      +/-   ##
==========================================
+ Coverage   81.39%   81.45%   +0.05%     
==========================================
  Files        1118     1119       +1     
  Lines      398680   400421    +1741     
  Branches   398680   400421    +1741     
==========================================
+ Hits       324517   326152    +1635     
+ Misses      55196    55186      -10     
- Partials    18967    19083     +116     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kosiew]

@jayzhan211,

Thanks for working on this. Routing the final deferred-filtered output through the coalescer looks like the right direction, and the LEFT JOIN regression test captures the ordering issue well.

I found one performance regression in the matched-column materialization path that I think should be addressed before merging. I also left a non-blocking suggestion to add symmetric RIGHT JOIN coverage.

[kosiew]

Could we keep the previous HashMap approach here, or use another O(chunks) index map? This changes the batch-to-source lookup from O(chunks) construction to repeated linear searches, which can become O(chunks²) in this hot path. A same-key buffered group can span many input batches, and append_output_pair creates one chunk per buffered batch, so I don't think we can rely on the group containing only a handful of chunks. With a large, batch-fragmented duplicate-key group, this could result in a significant number of comparisons during a freeze.

[jayzhan211]

It seems that linear scan would be a better choice than hash map because the distinct sources is likely "small", I add the comment to show why linear scan is preferred

[kosiew]

Could we also add the symmetric RIGHT JOIN regression? RIGHT JOIN streams the opposite child and has a different output-column and nulling layout, while advertising maintains_input_order = [false, true]. A test asserting that the right-side keys remain ordered would help protect the ordering contract on both paths.

[jayzhan211]

Test added

[kosiew]

@jayzhan211
I will review again this after conflicts are resolved.

[jayzhan211]

conflicts resolved

[kosiew]

@jayzhan211,

Thanks for the follow-up. The streamed-ordering regressions look good, including the requested RIGHT JOIN case, and cargo test -p datafusion-physical-plan --lib preserves_streamed_order passes all 3 tests.

I still think the source lookup change needs another pass before this is ready. The current linear lookup keeps the common case fast, but it does not preserve the previous O(chunks) behavior for valid inputs that produce many small buffered batches.

In particular, SortMergeJoinExec can receive arbitrary ExecutionPlan children, so we cannot rely on upstream operators always emitting batches that are large relative to batch_size.

I would suggest keeping the HashMap index, or using the linear scan only as a small-source fast path with a map fallback after the measured crossover.

[kosiew]

I think the performance concern from the previous review still applies here.

source_batches.iter().position(...) does a linear scan for every matched chunk, so source-index construction becomes quadratic in the number of distinct buffered batches in a freeze.

The bound described above is len(source_batches) <= batch_size / R + 1, but R can validly be 1 if a child emits tiny batches. With the default batch_size of 8192, that can mean up to roughly 8192 sources and about 33 million comparisons in one freeze.

Since SortMergeJoinExec accepts arbitrary ExecutionPlan children, the batching behavior of the common in-tree producers is not a contract we can rely on here.

Could we retain the HashMap lookup, or use the linear scan as a small-source fast path and fall back to a map once the source count crosses the measured crossover?

[jayzhan211]

I got a better approach for this

[jayzhan211]

TL;DR — You're right, and it's worse than described: the chunk sequence cycles (scanning_reset fires per streamed row, not per freeze), so the chunk count isn't bounded by S at all and the scan is O(chunks × S) — 8.35 ms in a single freeze, measured. Fixed, but with a direct-addressed table rather than a HashMap: buffered_batch_idx is an index into buffered_data.batches, so the keys are dense small integers and hashing them is pure overhead. Result is ~4× faster and ~6.4× smaller than the map in the normal case, ties the linear scan at its best, and needs no crossover threshold or fallback branch. One honest tradeoff in the wrapped-freeze case, detailed at the end.

Confirming the diagnosis

I instrumented materialize_right_columns on a real join (6 one-row buffered batches, 2 streamed rows, batch_size 5) and dumped the chunk sequence per freeze:

FREEZE chunks: [0, 1, 2, 3, 4]
FREEZE chunks: [5, 0, 1, 2, 3]   <- wrapped
FREEZE chunks: [4, 5]            <- never sees batch 0

The batch_size bound in my comment bounded S, but not the number of chunks — and the scan runs once per chunk. pair_streamed_row_with_group calls scanning_reset() per streamed row, not per freeze, so the sequence cycles and one freeze can contain many passes over the same sources. Cost is O(chunks × S); with one-row batches that's your 33M comparisons, which I measured at 8.35 ms in a single freeze. My "degrades gradually rather than falling off a cliff" line was simply wrong.

Why a direct-addressed table rather than the map

The keys here aren't opaque. buffered_batch_idx is literally an index into buffered_data.batches, a VecDeque — so the key space is dense, small, non-negative integers bounded by the deque length. Hashing those is pure overhead. A Vec indexed by batch_idx - min is the natural map, and it beats the HashMap on both axes.

Speed — 8192 rows in 2048 chunks; the last row is the one-row-per-batch shape:

distinct sources	hashmap	linear scan	direct table
4	19.7 µs	4.5 µs	4.8 µs
32	20.7 µs	13.0 µs	5.0 µs
128	23.5 µs	42.7 µs	5.1 µs
1024	48.1 µs	281.7 µs	5.8 µs
8192	293.3 µs	8347.6 µs	16.9 µs

Memory — peak bytes held by the lookup structure alone, measured with a tracking allocator (source_batches and interleave_indices excluded, since every variant needs them):

shape	distinct	span	map (presized)	direct
dense, 32 sources	32	32	1.1 KB	256 B
dense, 1024 sources	1024	1024	34.0 KB	8.0 KB
dense, 8192 sources	8192	8192	272.0 KB	64.0 KB

~6.4× less, because HashMap<usize, usize> pays 17 bytes/bucket at a 7/8 load factor rounded up to a power of two, while the table pays 8 bytes/slot flat. (Without pre-sizing, the map peaks at 408 KB during rehash.) It also ties the linear scan where the scan is at its best, so there's no crossover to tune and no fallback branch.

The two cases

Keys within a freeze are contiguous only within a streamed-row pass:

• Dense freeze (the norm) — span == distinct. The table is optimal on both time and memory.
• Wrapped freeze — the freeze straddles a scanning_reset, holding the tail of one pass plus the head of the next. The window wraps and leaves a gap, so span = the group's batch count while distinct = batch_size.

The tradeoff

The wrapped case is where the table gives something up, and I want to be straight about it. Memory crosses over at a span of ~35k batches — beyond that the table's transient exceeds the map's fixed 272 KB, reaching 4 MB at a 524288-batch group.

Time still favours the table, because scanning_reset fires once per pass, so at most one freeze per pass wraps and the O(span) cost amortizes against the O(group) of useful work the rest of the pass does:

group spans	freezes	hashmap	direct
8192	2	264.3 µs	26.2 µs
131072	17	2894.0 µs	304.1 µs
524288	65	11252.4 µs	1168.4 µs

Both linear, table ~10× ahead. The other tradeoff is one allocation per multi-source freeze that the scan didn't have — that's the 4.8 vs 4.5 µs at 4 sources. The single-source path returns before reaching any of this, so the common case is untouched.

I also tried an epoch-stamped table kept as reusable scratch on the stream: only ~15% faster, and it converts a transient allocation into a permanent one of the same size. Not worth the extra state.

If you'd like a hard cap on that transient, a one-line guard falling back to the map when span > k * matched_chunks.len() would bound it cheaply — happy to add it, though I'd argue 4 MB against a buffer already holding 524288 BufferedBatches with retained join-key arrays isn't the binding constraint.

Also in the PR

A regression test covering all three freeze shapes above (including the wrapped one and the min > 0 one), plus a debug_assert that every key addresses the live deque — the property the dense key space rests on.

[jayzhan211]

Happy to hear any other input or ideas you have!

[kosiew]

@jayzhan211,

Thanks for the follow-up. This addresses my previous concern about the repeated linear source-batch lookup. The direct-addressed table keeps the lookup efficient without relying on assumptions about how child plans batch their output, and the source offset and null sentinel handling look consistent with the interleave inputs.

The symmetric RIGHT JOIN ordering regression is also in place, along with coverage for rescanned and wrapped multi-batch groups and a partially passing filter.

There is still a small non-blocking test gap around combining a wrapped multi-source freeze with a null buffered index. Given that unmatched streamed rows are currently kept in separate chunks and the sentinel path is straightforward, I don't think that needs to block this PR.

Looks good to me. Thanks for iterating on this!

[jayzhan211]

Thanks for the review! I added test and benchmark