fix: drain completed sort-merge join output before awaiting input

[sunchao]

Which issue does this PR close?

No issue is automatically closed. This PR includes a standalone regression for completed sort-merge join output being withheld behind pending input.

Rationale for this change

The bitwise sort-merge join can finish an output batch and then wait for more outer input without yielding that completed batch. This occurs when the inner input is exhausted or a matching key continues across outer batches, with or without a filter. A stalled input can therefore withhold ready results, and long partitions can accumulate completed output in the coalescer.

What changes are included in this PR?

• Pass the existing emitter through the matching-key helpers and exhausted-inner drainage.
• Drain completed coalescer batches immediately after emitting each outer batch, before awaiting more input.
• Preserve partial-batch coalescing, final flushing, existing spill handling, and the timing helper.
• Add six regressions covering LeftMark and RightMark for empty inner input, matching keys, and filtered matching keys. Each requires a ready batch before the next outer input becomes pending, then checks all rows and marker values after resumption.

This does not add an operator or change memory-reservation accounting. The broader coalescer accounting work in #24427 and the separate materializing-stream work in #24573 are outside this change.

Are these changes tested?

Validated against public main 4fcaa01c:

• All six new regressions fail on unchanged production code with Completed output must precede pending outer input: Pending; all six pass with the fix.
• All 1,790 default physical-plan unit tests pass.
• The extended workspace command passes: 10,796 Rust tests, eight existing ignored tests, and all 505 SQL logic files. This includes 118 extended fuzz tests.
• cargo fmt --all, cargo fmt --all -- --check, and the mandatory full-workspace cargo clippy --all-targets --all-features -- -D warnings pass.

cargo test --locked --profile ci -p datafusion-physical-plan \
  bitwise_emits_completed_batches_before_pending_outer
cargo test --locked --profile ci -p datafusion-physical-plan --lib
RUST_BACKTRACE=1 cargo test --locked --profile ci \
  --exclude datafusion-examples --exclude datafusion-benchmarks --exclude datafusion-cli \
  --workspace --lib --tests --bins \
  --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption \
  -- --test-threads=8

The extended run used a process-local file-descriptor limit of 65,536. With the environment's default limit of 1,024, the existing sort_fuzz::test_sort_10k_mem failed with OS error 24; the same failure and successful higher-limit control were reproduced on a clean main checkout.

Existing semi/anti benchmarks were compared using distinct verified main/patched executables, with 100 samples, three seconds of warmup, and five seconds requested measurement:

Benchmark	Main	Patch	Criterion reported change
left_semi_1to10/100000	982.64 us	914.97 us	-6.08%
left_anti_partial/100000	2.0535 ms	2.1284 ms	+3.65%

The anti benchmark is slower in this paired run. This is one microbenchmark comparison on a shared machine, not a claim of universal throughput improvement or no regression. The new pending-input tests are the direct correctness/streaming regression; no RSS/OOM measurement is claimed.

Are there any user-facing changes?

Completed batches can be consumed before later outer input is ready. Row contents and marker semantics are unchanged; partial batches still coalesce as before. There is no public API, configuration, or dependency change.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 91.30435% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.45%. Comparing base (4fcaa01) to head (b6edfc7).

Files with missing lines	Patch %	Lines
...l-plan/src/joins/sort_merge_join/bitwise_stream.rs	91.30%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24685      +/-   ##
==========================================
- Coverage   81.45%   81.45%   -0.01%     
==========================================
  Files        1118     1118              
  Lines      399685   399699      +14     
  Branches   399685   399699      +14     
==========================================
+ Hits       325576   325579       +3     
- Misses      55103    55110       +7     
- Partials    19006    19010       +4     

☔ 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.

[jayzhan211]

Thanks @sunchao, LGTM