QinXi-ai opened a new pull request, #25159: URL: https://github.com/apache/datafusion/pull/25159
## Which issue does this PR close? Partially addresses #15382. This PR 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. 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. ## 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
