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 71.42857% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.37%. Comparing base (32badcb) to head (25dae23).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
.../src/joins/sort_merge_join/materializing_stream.rs	71.42%	13 Missing and 9 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24573    +/-   ##
========================================
  Coverage   81.37%   81.37%            
========================================
  Files        1116     1116            
  Lines      397509   397661   +152     
  Branches   397509   397661   +152     
========================================
+ Hits       323461   323591   +130     
- Misses      55110    55119     +9     
- Partials    18938    18951    +13     

☔ 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