feat: add opt-in probe selection for partitioned inner hash joins

[QinXi-ai]

Which issue does this PR close?

Partially addresses #15382. This draft seeks feedback on the design direction and covers the probe side of eligible partitioned inner hash joins; it does not close the broader issue.

Rationale for this change

Hash repartitioning currently copies every probe payload column before the join knows which rows match. For wide inputs with few matches, much of that copying produces rows the join immediately discards.

This change provides an opt-in path that shares the original probe batch across partitions and materializes payload rows only when producing join results. Join keys are still gathered into compact arrays for lookup and collision checks.

What changes are included in this PR?

• Add datafusion.execution.enable_hash_join_probe_selection, defaulting to false, with configuration documentation and SHOW ALL expectations.
• Use a private selection exchange when a partitioned INNER HashJoinExec directly consumes an unordered hash RepartitionExec on matching column keys, with more than one partition and an unlimited memory pool. Other plans retain the existing exchange, including finite-memory execution and dynamic/null-aware filtering.
• Keep payload ownership and its memory reservation shared across bounded output queues. Tie producer lifetime to consumers and release unclaimed output queues on cancellation. Forward source errors and task panics to consumers.
• Probe compact keys, resolve hash collisions, then translate selected positions back to original rows before join filters and output projection.
• Add physical-plan differential/lifecycle tests, SQL integration coverage, and a reproducible full-operator benchmark with path-activation and output-cardinality assertions.
• Include two Windows-only lint compatibility cleanups: remove an existing redundant fallback semicolon and acknowledge the async signature of an example whose Windows branch is synchronous.

The public RecordBatchStream contract and serialized plan format are unchanged. The private exchange avoids exposing selection-bearing batches to operators that expect ordinary batches. Build-side selection, spilling selection batches, other join types, and generalized selection propagation remain follow-up work.

What is the testing strategy for this PR?

Differential tests compare complete output multisets with the materialized path across seeds, partition counts, map implementations, null equality, duplicate keys, filters, and small output batches. Additional tests cover unsupported-plan fallback, cancellation with an exchange retained, source errors/panics with idle outputs, and reservation failure. SQL tests exercise Int64, Utf8 and Utf8View keys with filters and embedded projection, asserting both exact rows and the path-activation metric.

Windows validation passed full-target/full-feature Clippy with warnings denied, formatting, the five selection tests with forced hash collisions, SQL integration, and the changed SHOW ALL expectations. The core library's 467 tests also passed. The full extended workspace run was stopped after an existing spill test exceeded its 20-second attempt budget under load; the same test passed all 12 attempts after heavy concurrent work stopped, without changing its threshold. The complete information_schema run was stopped after three of four files completed; its changed SHOW ALL section plus original configuration cleanup passed separately. Full workspace and full-file SLT completion remain unverified, and Linux CI has not run.

The benchmark measurements and broader validation above were recorded for the original implementation commit 07f916c, based on c0b2ebe; they are not a fresh benchmark of later upstream changes. Benchmark measurements include both exchanges, build, probe, output collection, and reservation release; input generation is outside timing. Cases vary payload width/type, partition count, match coverage and skew. The path remains disabled by default because narrow and skewed cases can regress. Reported reservation peaks are engine accounting; process working-set measurements are reported separately.

In the final 1,344-measurement Windows run (32 cases, 21 alternating samples per path), median paired speedups were 11.27x for Utf8 and 2.62x for wide integer payloads at 8 partitions and 10% uniform match coverage. Narrow 32-partition cases regressed to about 0.80x–0.85x. These are operator benchmarks, not general SQL or production claims. Default-off/base comparisons were noisy: three full-scan median time ratios were 1.05/1.03/1.13; a follow-up with adjacent, order-alternated processes on four representative cases gave 1.00/1.03/0.95. These measurements do not establish zero default-path overhead.

Memory accounting also has a tradeoff: the Utf8/8/10% uniform case increased peak reservations from 9.77 MB to 44.29 MB because original batches remain retained. Separate process runs covering uniform and skewed inputs observed approximately 91–94 MB peak Windows working set with the option off and 77 MB with it on. Reservations and process working set measure different things; this is another reason to keep the initial path opt-in and restricted to unlimited pools.

After merging upstream 5e42247 in 83a712b, formatting, full-target/full-feature Clippy, all five selection tests with forced hash collisions, and the SQL integration test passed again on the synchronized head. The draft status and wider validation limitations remain unchanged.

Are there any user-facing changes?

Yes: the new opt-in execution configuration and probe_selection_partitions metric. Existing behavior remains the default. No public API or wire-format changes are required.

[QinXi-ai]

I am opening this as a draft to get feedback on the design direction before extending the implementation.

The main questions I would appreciate feedback on are:

1. Is a private, probe-side selection exchange owned by HashJoinExec an acceptable first step for Support zero copy hash repartitioning for Hash Join #15382, or should this wait for a reusable selection representation across operators?
2. Is the initial scope useful: opt-in partitioned INNER joins, direct matching hash repartition, and unlimited memory pools? In particular, the original-batch retention increases peak reservations in the low-match Utf8 case even though the measured process working set decreases.
3. What benchmark and validation gates would you want before considering this ready? The Windows operator results show substantial gains for wide, low-match inputs, regressions for some narrow inputs, and noisy default-off comparisons. Full workspace/SLT completion and Linux CI are still outstanding as detailed in the description.

The intent is to evaluate this limited design, not to close the broader issue or enable it by default. I would be happy to split out the benchmark or adjust the integration boundary based on maintainer feedback.