bench: classic pwmj sql benchmark - #25839

…xtreme… (apache#25840)

## Which issue does this PR close?

- Part of apache#18221 (and the PWMJ epic apache#17427).

It doesn't close apache#18221. On the issue's 100K × 100K shape, the join's
time goes to its huge output, so this change leaves it about the same.
See the benchmarks below.

## Rationale for this change

The classic `PiecewiseMergeJoin` stream (`INNER` / `LEFT` / `RIGHT` /
`FULL`) sorts each streamed batch, then walks the sorted buffered side
row by row to find each streamed row's first match. Two costs are
avoidable:

- **Rows that can't match anything still pay full price.** They are
sorted, gathered and walked through the buffered side like every other
row. This covers keys below every buffered key for `<`, and NULL keys.
When most streamed rows can't match, which is common for a selective
range predicate, the operator does all this work and outputs nothing for
them.
- **The first match is found by a linear walk.** Every match set is a
suffix `[k, len)` of the sorted buffered side, so `k` can be found with
a binary search instead of stepping over every non-matching buffered
row.

## What changes are included in this PR?

All changes are in `joins/piecewise_merge_join/`, plus a new benchmark.

- **Rows that can't match are removed before the sort
(`classic_join.rs`).** Because every match set is a suffix, a streamed
row matches anything at all only if it matches the last buffered key.
- One vectorized comparison against that key, cached once when the
buffered side is ready, finds those rows.
- They are emitted as unmatched rows for `Right`/`Full` and dropped for
`Inner`/`Left`. Only the remaining rows are sorted, gathered and
scanned.
- When no rows were removed, the path is unchanged except for that one
comparison per batch. When some were, the sort indices are mapped back
to positions in the batch, so it is still gathered only once.
- An empty buffered side, or one whose keys are all NULL, short-circuits
without the comparison.
- **The comparison must agree exactly with the scan, or it would change
results.**
- Flat keys use one `apply_cmp`, which normalizes `-0.0`/`+0.0` just as
the scan's `JoinKeyComparator` does.
- Nested keys are decided with the scan's own comparator. `apply_cmp`
orders NULL elements inside a key ascending, while the comparator
applies the sort options, descending for `<`/`<=`, at every nesting
level.
- The scan's nested ordering for `<`/`<=` gives different results from
`NestedLoopJoinExec` when keys contain NULL elements. That is a
pre-existing bug on `main`, which I'll file and fix separately. This PR
keeps PWMJ's current results unchanged.
- **Binary search for each row's first match (`utils.rs`).**
- `first_match(lo, hi, matches)` replaces the linear walk in
`resolve_classic_join`.
- It searches `[previous row's match, len)`. The streamed batch is
sorted, so each row's first match is at or after the previous row's.
- The existence stream already did an inline binary search. It now calls
the same helper, with no change in behavior.
- The operator dispatch moves into two small helpers, `matches_on_equal`
and `is_match`, used by both streams.
- **Streamed NULL keys never reach the scan now,** since the new
pre-sort step removes them. The scan's streamed-NULL skip was dead code
and is replaced by a `debug_assert`. The buffered-side NULL skip stays.

### Benchmarks

The `pwmj` suite from apache#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
```

| regime | `main` (median) | Inner | Left | Right | Full |
|---|---|---|---|---|---|
| `no_match`: no streamed row matches | 47–50 ms | −95.9% (25×) | −95.9%
(24×) | −94.0% (17×) | −93.7% (16×) |
| `null_heavy`: half NULL, rest `no_match` | 45–50 ms | −95.7% (23×) |
−95.7% (23×) | −94.0% (17×) | −93.7% (16×) |
| `half_match`: half `no_match`, half `selective` | 371–375 ms | −9.2% |
−8.7% | −8.2% | −8.4% |
| `selective`: every row matches the 1–4 smallest buffered keys |
694–699 ms | −2.9% | −3.9% | −3.6% | −3.2% |
| `all_match`: every row matches every buffered key | 162–163 ms | +0.1%
| −0.6% | −0.0% | +0.3% |

## What is the testing strategy for this PR?

- `first_match_agrees_with_linear_scan` (`utils.rs`): exhaustive over
every range, answer and start position up to 40. It checks the result
against a linear scan and bounds the number of comparisons by `⌈log2(hi
− lo + 1)⌉`.
- `matchable_rows_agrees_with_scan` (`classic_join.rs`): the pre-sort
check against a brute-force version of the scan's own definition of a
match. It covers 12 key types, 4 operators, and a normal, all-NULL and
empty buffered side.
- The key types: Int32, Float64 with `±0.0`/NaN/`±inf`, Utf8, Utf8View,
Binary, Dictionary, Decimal128, Date32, Timestamp, Boolean, and two List
cases with NULL elements.
  - With the nested-key branch removed, this test fails.
- `join_right_less_than_signed_zero_prefilter_agrees_with_scan`: end to
end with `-0.0`/`+0.0` keys.
- Existing tests:
  - all PWMJ unit tests;
- the NLJ differential fuzz test `fuzz_pwmj_matches_nested_loop`
(`--features extended_tests`), covering Inner/Left/Right/Full, NULL
keys, small batches and multiple partitions;
  - the PWMJ and join sqllogictests.

## Are there any user-facing changes?

No API or configuration changes.

…xtreme… (apache#25840)

## Which issue does this PR close?

- Part of apache#18221 (and the PWMJ epic apache#17427).

It doesn't close apache#18221. On the issue's 100K × 100K shape, the join's
time goes to its huge output, so this change leaves it about the same.
See the benchmarks below.

## Rationale for this change

The classic `PiecewiseMergeJoin` stream (`INNER` / `LEFT` / `RIGHT` /
`FULL`) sorts each streamed batch, then walks the sorted buffered side
row by row to find each streamed row's first match. Two costs are
avoidable:

- **Rows that can't match anything still pay full price.** They are
sorted, gathered and walked through the buffered side like every other
row. This covers keys below every buffered key for `<`, and NULL keys.
When most streamed rows can't match, which is common for a selective
range predicate, the operator does all this work and outputs nothing for
them.
- **The first match is found by a linear walk.** Every match set is a
suffix `[k, len)` of the sorted buffered side, so `k` can be found with
a binary search instead of stepping over every non-matching buffered
row.

## What changes are included in this PR?

All changes are in `joins/piecewise_merge_join/`, plus a new benchmark.

- **Rows that can't match are removed before the sort
(`classic_join.rs`).** Because every match set is a suffix, a streamed
row matches anything at all only if it matches the last buffered key.
- One vectorized comparison against that key, cached once when the
buffered side is ready, finds those rows.
- They are emitted as unmatched rows for `Right`/`Full` and dropped for
`Inner`/`Left`. Only the remaining rows are sorted, gathered and
scanned.
- When no rows were removed, the path is unchanged except for that one
comparison per batch. When some were, the sort indices are mapped back
to positions in the batch, so it is still gathered only once.
- An empty buffered side, or one whose keys are all NULL, short-circuits
without the comparison.
- **The comparison must agree exactly with the scan, or it would change
results.**
- Flat keys use one `apply_cmp`, which normalizes `-0.0`/`+0.0` just as
the scan's `JoinKeyComparator` does.
- Nested keys are decided with the scan's own comparator. `apply_cmp`
orders NULL elements inside a key ascending, while the comparator
applies the sort options, descending for `<`/`<=`, at every nesting
level.
- The scan's nested ordering for `<`/`<=` gives different results from
`NestedLoopJoinExec` when keys contain NULL elements. That is a
pre-existing bug on `main`, which I'll file and fix separately. This PR
keeps PWMJ's current results unchanged.
- **Binary search for each row's first match (`utils.rs`).**
- `first_match(lo, hi, matches)` replaces the linear walk in
`resolve_classic_join`.
- It searches `[previous row's match, len)`. The streamed batch is
sorted, so each row's first match is at or after the previous row's.
- The existence stream already did an inline binary search. It now calls
the same helper, with no change in behavior.
- The operator dispatch moves into two small helpers, `matches_on_equal`
and `is_match`, used by both streams.
- **Streamed NULL keys never reach the scan now,** since the new
pre-sort step removes them. The scan's streamed-NULL skip was dead code
and is replaced by a `debug_assert`. The buffered-side NULL skip stays.

### Benchmarks

The `pwmj` suite from apache#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
```

| regime | `main` (median) | Inner | Left | Right | Full |
|---|---|---|---|---|---|
| `no_match`: no streamed row matches | 47–50 ms | −95.9% (25×) | −95.9%
(24×) | −94.0% (17×) | −93.7% (16×) |
| `null_heavy`: half NULL, rest `no_match` | 45–50 ms | −95.7% (23×) |
−95.7% (23×) | −94.0% (17×) | −93.7% (16×) |
| `half_match`: half `no_match`, half `selective` | 371–375 ms | −9.2% |
−8.7% | −8.2% | −8.4% |
| `selective`: every row matches the 1–4 smallest buffered keys |
694–699 ms | −2.9% | −3.9% | −3.6% | −3.2% |
| `all_match`: every row matches every buffered key | 162–163 ms | +0.1%
| −0.6% | −0.0% | +0.3% |

## What is the testing strategy for this PR?

- `first_match_agrees_with_linear_scan` (`utils.rs`): exhaustive over
every range, answer and start position up to 40. It checks the result
against a linear scan and bounds the number of comparisons by `⌈log2(hi
− lo + 1)⌉`.
- `matchable_rows_agrees_with_scan` (`classic_join.rs`): the pre-sort
check against a brute-force version of the scan's own definition of a
match. It covers 12 key types, 4 operators, and a normal, all-NULL and
empty buffered side.
- The key types: Int32, Float64 with `±0.0`/NaN/`±inf`, Utf8, Utf8View,
Binary, Dictionary, Decimal128, Date32, Timestamp, Boolean, and two List
cases with NULL elements.
  - With the nested-key branch removed, this test fails.
- `join_right_less_than_signed_zero_prefilter_agrees_with_scan`: end to
end with `-0.0`/`+0.0` keys.
- Existing tests:
  - all PWMJ unit tests;
- the NLJ differential fuzz test `fuzz_pwmj_matches_nested_loop`
(`--features extended_tests`), covering Inner/Left/Right/Full, NULL
keys, small batches and multiple partitions;
  - the PWMJ and join sqllogictests.

## Are there any user-facing changes?

No API or configuration changes.

apache#26287)

## Which issue does this PR close?

- Part of apache#18221 (and the PiecewiseMergeJoin EPIC apache#17427).

## Rationale for this change

On large inputs the classic `PiecewiseMergeJoin` (Inner/Left/Right/Full)
can be slower than `NestedLoopJoin`. For example, before apache#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:

| where | share of busy samples |
|---|---|
| `take` repeating each matched streamed row | 32% |
| `BatchCoalescer` `memmove` of every column | 30% |
| downstream `sum` (the query itself) | 10% |
| per-streamed-row fixed cost (`RecordBatch`, allocations, coalescer
bookkeeping) | ~13% |

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. apache#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 apache#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
(apache#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 apache#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`

| regime | main (ms) | this PR (ms) | speedup |
|---|---|---|---|
| no_match, null_heavy: Inner, Left | 2.2–2.3 | 2.1 | 1.05× |
| no_match, null_heavy: Right, Full | 3.2–3.3 | 2.4–2.5 | **1.29–1.33×**
|
| half_match | 231–233 | 27–29 | **8.1–8.4×** |
| selective | 453–455 | 47–48 | **9.5–9.7×** |
| all_match (200M output rows) | 167–168 | 159–161 | 1.05–1.06× |

## 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).