perf: Add an adaptive integer membership prefilter for hash joins

[lyne7-sc]

Which issue does this PR close?

• Closes #.

Rationale for this change

Hash joins can spend substantial time looking up probe keys that do not exist on the build side. For a single integer key with a bounded range, an exact membership bitmap can reject these keys cheaply before hash table lookup. This helps when many nonmatching keys reach the join, including when dynamic filters cannot be pushed down to the input.

For example, in the measured TPC-H SF1 plans, the RightSemi join on part.p_partkey = lineitem.l_partkey receives approximately 6 million probe rows. With memtable inputs or the default Parquet configuration:

• Q8: Only 0.73% of probe rows match the filtered build side, resulting in approximately 5.96 million unsuccessful hash table lookups.
• Q9: Only 5.32% of probe rows match the filtered build side, resulting in approximately 5.68 million unsuccessful hash table lookups.

These low probe hit rates leave substantial room to avoid unsuccessful lookups, reflected in both the query-latency and operator-level measurements below.

The extra membership check is less useful when most keys match. An adaptive policy pauses filtering when the observed pruning ratio is low and periodically samples again as the input changes.

What changes are included in this PR?

• Build an optional integer membership bitmap for eligible single-key HashMap joins in both CollectLeft and Partitioned modes. ArrayMap and null-aware joins retain their existing paths.
• Generate a lookup mask for each probe batch, skipping hash table lookups for absent keys. Preserve the original probe rows and NULL handling so outer, anti, and mark join semantics remain unchanged. Hash computation is unchanged.
• Track pruning independently in each probe stream. Unproductive sampling windows trigger progressively longer pauses; productive samples reset the pause interval.
• Account for bitmap and mask memory through the memory pool, falling back to normal lookup when an optional reservation cannot be satisfied.
• Add configuration, documentation, creation/pruning metrics, and the hash_join_probe microbenchmark.

What is the testing strategy for this PR?

• Unit tests cover membership and integer boundaries, range overflow, bitmap boundaries, adaptive sampling/backoff, and sliced validity masks.
• HashJoin tests cover fallback paths, memory reservation failures, and independent probe-stream adaptation with a shared build bitmap.
• hash_join_integer_prefilter.slt covers duplicate matches, unmatched rows, NULL equality, inner/full/semi/anti joins, null-aware IN/NOT IN behavior, range limits, and creation/pruning metrics.

Are there any user-facing changes?

The feature is opt-in and disabled by default. It can be enabled with:

SET datafusion.execution.enable_join_integer_prefilter = true;

Additional execution options control the maximum key range (default 262144), minimum pruning ratio (0.5), and sampling window (12288 input rows). The metrics integer_prefilter_created_count and probe_prefilter_rows_pruned expose whether the prefilter was created and how many probe rows skipped lookup.

Query results are unchanged.

Benchmarks

Microbench: hash_join_probe

Workload	Main	Off	On	On vs main
RightSemi, 10% interleaved hits	5.335	5.293	3.985	-25.3%
RightAnti, 10% interleaved hits	10.849	10.933	9.479	-12.6%
Inner, 10% interleaved hits	6.114	6.164	4.802	-21.4%
RightSemi, 10% clustered hits	4.286	4.338	3.549	-17.2%
RightAnti, 10% clustered hits	10.172	11.633	8.969	-11.8%
RightSemi, all hits	13.066	12.897	13.061	-0.0%
RightAnti, all hits	5.920	5.924	6.112	+3.2%
Inner, all hits	19.235	19.567	19.565	+1.7%
RightSemi, high-to-low hit rate	9.096	9.405	9.203	+1.2%
RightAnti, high-to-low hit rate	8.480	8.478	8.408	-0.8%
RightSemi, single probe batch, all hits	0.205	0.217	0.258	+25.9%
RightSemi, dense 100K build, all hits	10.172	10.264	10.107	-0.6%
RightSemi, sparse 100K build, all hits	15.579	15.568	15.567	-0.1%

TPC-H with memtable

Query	Main	Off	On	On vs main
Q8	25.332	23.381	16.080	-36.5%
Q9	48.990	46.381	41.653	-15.0%
Q17	67.344	66.039	43.156	-35.9%

TPC-H with parquet

with the default datafusion.execution.parquet.pushdown_filters=false

Query	Main	Off	On	On vs main
Q8	63.736	63.243	58.271	-8.6%
Q9	76.014	74.103	70.175	-7.7%

Target HashJoin join_time

The following table measures the CollectLeft / RightSemi HashMap join on part.p_partkey = lineitem.l_partkey described above. It reports operator join_time, summed across the four probe partitions, separately from end-to-end query latency.

Input	Query	Main	Off	On	On vs main
MemTable	Q8	29.330	29.625	12.045	-58.9%
MemTable	Q9	39.985	40.010	26.415	-33.9%
Parquet (default)	Q8	29.685	29.820	11.190	-62.3%
Parquet (default)	Q9	35.830	35.305	20.525	-42.7%

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-common v55.1.0 (current)
       Built [  29.514s] (current)
     Parsing datafusion-common v55.1.0 (current)
      Parsed [   0.058s] (current)
    Building datafusion-common v55.1.0 (baseline)
       Built [  28.538s] (baseline)
     Parsing datafusion-common v55.1.0 (baseline)
      Parsed [   0.054s] (baseline)
    Checking datafusion-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.903s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ExecutionOptions.enable_join_integer_prefilter in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894
  field ExecutionOptions.join_integer_prefilter_max_key_range in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894
  field ExecutionOptions.join_integer_prefilter_min_pruning_ratio in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894
  field ExecutionOptions.join_integer_prefilter_sample_rows in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  60.868s] datafusion-common
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  32.665s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.135s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  32.212s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.137s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.816s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  67.185s] datafusion-physical-plan
    Building datafusion-sqllogictest v55.1.0 (current)
       Built [  83.175s] (current)
     Parsing datafusion-sqllogictest v55.1.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-sqllogictest v55.1.0 (baseline)
       Built [  83.504s] (baseline)
     Parsing datafusion-sqllogictest v55.1.0 (baseline)
      Parsed [   0.021s] (baseline)
    Checking datafusion-sqllogictest v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.107s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 169.231s] datafusion-sqllogictest

[codecov-commenter]

Codecov Report

❌ Patch coverage is 93.14587% with 39 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.95%. Comparing base (a0631ed) to head (c9414e2).
⚠️ Report is 15 commits behind head on main.

Files with missing lines	Patch %	Lines
...tafusion/physical-plan/src/joins/hash_join/exec.rs	91.23%	4 Missing and 18 partials ⚠️
...usion/physical-plan/src/joins/integer_prefilter.rs	94.30%	10 Missing and 6 partials ⚠️
...fusion/physical-plan/src/joins/hash_join/stream.rs	97.29%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25363      +/-   ##
==========================================
+ Coverage   81.92%   81.95%   +0.03%     
==========================================
  Files        1135     1136       +1     
  Lines      427573   428843    +1270     
  Branches   427573   428843    +1270     
==========================================
+ Hits       350279   351474    +1195     
- Misses      56367    56383      +16     
- Partials    20927    20986      +59     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[sunchao]

Thanks @lyne7-sc. I reviewed c9414e2a02f4720b32d75b2b70091713d3b3a47e against merge base a0631edb774855faa999266003e23f939473d44e, including lookup semantics, integer boundaries, memory accounting, adaptive sampling, and the benchmark setup. I did not find an actionable defect.

Local validation:

• All 10 focused integer_prefilter tests passed, including optional-memory fallback and independent adaptation with a shared build bitmap.
• An independent physical-plan oracle passed 4,800 scenarios on the base and the same scenarios with the feature disabled and enabled on the PR head (14,400 executions total). It covered all 10 join types, both NULL equalities, residual filters, duplicate continuation with small output batches, reordered projections, fetch limits, empty/all-NULL inputs, sliced validity buffers, and CollectLeft/Partitioned execution. Enabled runs exercised bitmap creation and pruning.
• All 13 new benchmark workloads passed their result assertions with the feature both enabled and disabled.

Keeping the original probe rows and validity separate from lookup membership appears to preserve the outer/anti/mark semantics, and the reservation fallback tests passed. I have not independently reproduced the reported release-mode or TPC-H timings; the benchmark runs above checked results only. At review time, 40 checks had succeeded and three extended-suite checks were skipped.