perf: keep the hash join unique-key fast path when the build side has NULL keys - #26018

… NULL keys (apache#26018)

## Which issue does this PR close?

- None. There is no separate issue; the problem is described below.

## Rationale for this change

When every build key is unique, the hash join probe uses a faster loop
that skips walking the duplicate-key chain. That loop was enabled by
`map.len() == next.len()`, which means "every build row is in the map".
Under `NullEqualsNothing`, build rows with a NULL key are not inserted,
because they can never match. So a single NULL build key turns the fast
path off, and every probe walks the chain even though every chain has
length 1.

This affects the hash-map path: string keys, multi-column keys and
wide-range integer keys. It does not affect `ArrayMap`. A nullable
unique key on the build side is enough to trigger it.

## What changes are included in this PR?

- `update_from_iter` returns whether it chained a row onto an existing
hash value. `JoinHashMapU32` and `JoinHashMapU64` keep that result in a
`chained` flag across batches.
- `get_matched_indices_with_limit_offset` now takes the uniqueness
decision as a parameter instead of computing `map.len() ==
next_chain.len()` itself. The two join hash maps pass `!chained`.
`PruningJoinHashMap` (symmetric hash join) still passes `map.len() ==
next.len()`, so its behavior is unchanged.
- Nothing changes when no build rows are skipped, because `map.len() ==
next.len()` held exactly when no row was chained. `NullEqualsNull` is
also unchanged: NULL rows are inserted and chain as before.

## What is the testing strategy for this PR?

New unit tests:
- `join_hash_map::tests::test_unique_fast_path_with_uninserted_row`:
unique keys plus one row that was not inserted now take the fast path.
The test checks the fast path's offset behavior with `limit = 2`; before
this change it failed with `Some((1, Some(0)))`.
-
`join_hash_map::tests::test_chain_from_earlier_update_disables_fast_path`:
a chain created by an earlier batch keeps the chain walk.
- `utils::tests::update_hash_null_key_does_not_change_matches`:
`update_hash` plus a chunked probe give the same matches with and
without a NULL build key.

Benchmark: `datafusion-cli` built with `release-nonlto`, base vs. this
PR, on an Apple M4 Pro (12 cores, 24 GB) with default
`target_partitions`. Each query is `SELECT count(*) FROM probe p JOIN
build b ON <keys>`. The build side has 5M unique keys (`Utf8View`, or
`(Int64 wide range, Utf8View)`) plus one NULL-key row. The probe side
has 20M rows in shuffled order and matches every key. Runs were
interleaved; each value is the median of 18 runs (6 rounds × 3), started
only when the 1-minute load average was below 4. Under default
thresholds the 5M-row build side runs in `Partitioned` mode. The
`CollectLeft` rows raise `hash_join_single_partition_threshold{,_rows}`
so that the whole build side is one map. `EXPLAIN ANALYZE` shows
`array_map_created_count=0` in every case except where noted.

| Case | Mode | base (s) | PR (s) | PR/base |
|---|---|---|---|---|
| Utf8View key, unique + 1 NULL | CollectLeft | 0.237 | 0.214 | **0.90**
|
| (Int64, Utf8View) key, unique + 1 NULL | CollectLeft | 0.284 | 0.259 |
**0.91** |
| Utf8View key, unique + 1 NULL | Partitioned | 0.144 | 0.143 | 0.99 |
| (Int64, Utf8View) key, unique + 1 NULL | Partitioned | 0.192 | 0.189 |
0.99 |
| Control: Utf8View key, unique, no NULL | Partitioned | 0.134 | 0.137 |
1.02 |
| Control: Utf8View key, unique, no NULL | CollectLeft | 0.205 | 0.210 |
1.03 |
| Control: (Int64, Utf8View) key, unique, no NULL | CollectLeft | 0.253
| 0.253 | 1.00 |
| Control: 2 rows per key + 1 NULL | Partitioned | 0.163 | 0.161 | 0.99
|
| Control: 2 rows per key + 1 NULL | CollectLeft | 0.260 | 0.264 | 1.02
|
| Control: `hj` Q13 (TPC-H SF10, 1% density, hash map), `count(*)` |
CollectLeft | 0.067 | 0.068 | 1.02 |

`HashJoinExec` `join_time` from `EXPLAIN ANALYZE` (summed over
partitions, median of 7 runs):

| Case | base | PR |
|---|---|---|
| Utf8View key + 1 NULL, CollectLeft | 1.14 s | 0.84 s (-26%) |
| (Int64, Utf8View) key + 1 NULL, CollectLeft | 1.31 s | 1.01 s (-23%) |
| Utf8View key + 1 NULL, Partitioned | 729 ms | 715 ms |
| Control: no NULL, CollectLeft | 802 ms | 809 ms |
| Control: 2 rows per key, CollectLeft | 1.15 s | 1.12 s |

With one NULL key, `join_time` now matches the no-NULL control. In
`Partitioned` mode all NULL-key build rows hash to a single partition,
so only that partition lost the fast path, and the end-to-end change
there is within noise. Run-to-run spread on this shared machine is about
±3%, which covers every control.

## Are there any user-facing changes?

No. `join_hash_map` is `#[doc(hidden)]` internal API, although the
signatures of its free functions `update_from_iter` and
`get_matched_indices_with_limit_offset` change.

## Which issue does this PR close?

- Closes apache#25332 .

## Rationale for this change

`MapOffset` was a `(usize, Option<u64>)` tuple that encoded the resume
point of chunked hash map lookups by convention. One position had two
encodings (`(row, Some(0))` and `(row + 1, None)`), the unique-key fast
paths read only `offset.0`, and `HashJoinStream` relied on an
undocumented invariant (`offset == (0, None)` only before the first
chunk).

## What changes are included in this PR?

- Replace the tuple with `pub enum ProbeOffset { AtRow { row }, MidChain
{ row, next } }` in `join_hash_map.rs`, re-exported from `joins::utils`
next to `JoinHashMapType`.
- Normalize at the producer: `traverse_chain` returns `AtRow { row: row
+ 1 }` once a row's chain is finished, so every position has one
encoding and readers no longer handle `Some(0)`.

  | Old | New |
  |---|---|
  | `(row, None)` | `AtRow { row }` |
  | `(row, Some(next))`, `next != 0` | `MidChain { row, next }` |
  | `(row, Some(0))` | `AtRow { row: row + 1 }` |

- Unique-key fast paths (`JoinHashMap`, `ArrayMap`) `debug_assert!` that
they only receive `AtRow`. The invariant holds by construction (`unique`
is fixed when the build side is built, and every chunk of a probe batch
uses the same map), so a check in debug builds and CI is enough to catch
a regression without adding a panic path to query execution.
- Add `ProcessProbeBatchState::is_first_chunk()` and use it at the three
first-chunk checks. The invariant it relies on (a returned offset is
never `AtRow { row: 0 }`) is documented on `ProbeOffset`.

## What is the testing strategy for this PR?

Covered by existing tests, with expected offsets updated to the new
encoding.

`test_unique_fast_path_with_uninserted_row` (apache#26018) told the fast path
apart from the chain walk by their different offset encodings, which
this PR removes. This PR added an unmatched probe row so the two paths
now emit different results (the fast path limits by probe rows, the
chain walk by matches), and checked that the test fails when the fast
path is disabled. cc @jayzhan211


## Are there any user-facing changes?

No behavior change. 
API change: `JoinHashMapType::get_matched_indices_with_limit_offset` and
the public `get_matched_indices_with_limit_offset` function now take and
return `ProbeOffset` instead of `(usize, Option<u64>)`.

Co-authored-by: Jay Zhan <[email protected]>