perf(pwmj): build output batches from buffered runs and one streamed … - #26287

Which issue does this PR close?

• Part of Check PiecewiseMergeJoin performance for large tables #18221 (and the PiecewiseMergeJoin EPIC [EPIC]: Make PiecewiseMergeJoin work in Datafusion #17427).

Rationale for this change

On large inputs the classic PiecewiseMergeJoin (Inner/Left/Right/Full) can be slower than NestedLoopJoin. For example, before #25840, l JOIN r ON l.a < r.b over 100K×100K rows at the default target_partitions took 1.41s with PWMJ and 1.26s with NLJ.

A profile of the 1K×1M case (sample, target_partitions = 1) shows the time goes to copying output rather than to comparisons:

For every matched streamed row, the join takes UInt32Array::from_value(row, count), gathers the streamed row count times, slices the buffered suffix, and builds a RecordBatch from them. BatchCoalescer then copies all of it again, so the streamed columns are written twice. #25840 made finding each row's first match a binary search and settles never-matching rows before the sort, but it kept this emission path, so on current main the cost above is still paid for every matched streamed row.

What changes are included in this PR?

The classic join builds each output batch directly, and BatchCoalescer is removed from this path.

Matched rows:

• The output is kept as runs: (buffered start, len), paired with one streamed row index per output row.
• Once batch_size rows are pending, or the stream batch ends:
  • Buffered columns are copied run by run with MutableArrayData::try_extend, which is a memcpy for fixed-width types.
  • Streamed columns are gathered with one take per column.

Rows that match nothing are emitted with null columns for the other side, in batches of at most batch_size:

• Streamed rows that can never match (Right/Full), which perf(pwmj):settle never-matching streamed rows against the buffered extreme… #25840 settles before the sort, are selected with one filter_record_batch and emitted as slices of its output (push_unmatched_streamed). When every row of a stream batch is unmatched, filter returns the input columns as a slice and copies nothing.
• Unmatched buffered rows (Left/Full), in the final pass, are the prefix before the min_marked watermark plus any trailing NULL keys (fix: align nested piecewise merge join sorting and comparison #26043). They are copied out of the buffered batch (push_unmatched_buffered), not sliced: a slice would keep the whole buffered batch alive and report its full buffers to downstream memory accounting once per output batch

Benchmarks

The pwmj suite from #25839 (100K buffered keys, 2M streamed rows, five match regimes × INNER/LEFT/RIGHT/FULL). Reproduce with:

cargo run -p datafusion-benchmarks --release --bin benchmark_runner -- pwmj

What is the testing strategy for this PR?

Six new unit tests in classic_join.rs:

• join_full_less_than_splits_suffixes_across_batches: at batch_size = 4, a buffered suffix straddles an output batch, and never-matching streamed rows, matched rows and unmatched buffered rows all appear. It asserts both the rows and the batch sizes [1, 4, 2, 1].
• join_inner_less_than_one_row_completes_several_batches: one streamed row whose buffered suffix fills several output batches.
• join_right_less_than_empty_buffered: a Right join with an empty buffered side, where every streamed row is padded with nulls.
• join_right_less_than_never_matching_rows_split_by_batch_size: more never-matching streamed rows than batch_size, which must be split into batches of at most batch_size, in input order, ahead of the matched rows.
• join_left_less_than_unmatched_prefix_and_trailing_nulls: a Left join on < whose unmatched buffered rows are a prefix longer than batch_size plus trailing NULL keys, which sort last for <.
• join_left_unmatched_buffered_batches_report_own_memory: each unmatched-buffered output batch reports only its own rows' memory, not the whole buffered batch.

Are there any user-facing changes?

No API changes. EXPLAIN ANALYZE reports a smaller output_bytes for PiecewiseMergeJoin, because output batches no longer share the coalescer's buffers (explain_analyze.slt: 128.0 KB → 96.0 B).