Repository navigation
perf: Add an adaptive integer membership prefilter for hash joins - #25363
Merged
Merged
Conversation
Use a shared word bitmap that retains NULL keys and applies membership only to hash lookups. Count all observed rows when deciding whether to pause, with bounded exponential backoff. Align configuration types and metrics while preserving optional memory accounting and fallback. Standardize the 13 payload-bearing benchmark workloads and extend join, projection, fetch, and SQL regression coverage.
github-actions
Bot
added
documentation
|
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 |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25363 +/- ##
==========================================
+ Coverage 82.75% 82.76% +0.01%
==========================================
Files 1147 1148 +1
Lines 449798 450368 +570
Branches 449798 450368 +570
==========================================
+ Hits 372210 372751 +541
Misses 54916 54916
- Partials 22672 22701 +29 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
sunchao
reviewed
Sep 16, 2026
sunchao
left a comment
sunchao
left a comment
Member
There was a problem hiding this comment.
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_prefiltertests 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.
sunchao
approved these changes
Sep 16, 2026
Contributor
|
Is this PR waiting on more review? |
Member
Contributor
Author
|
@sunchao Thanks! I've merged the latest main and resolved the conflicts. |
Omega359
pushed a commit
to Omega359/arrow-datafusion
that referenced
this pull request
Oct 11, 2026
…ache#25363) ## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes apache#123` indicates that this PR will close issue apache#123. --> - Closes #. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. Please explain the problem you are trying to solve in terms of the user-visible behavior, rather than the implementation. For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the implementation. "COUNT(DISTINCT) returns wrong results when the column contains nulls" is the user-visible problem. --> 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? <!-- There is no need to duplicate the description in the issue here, but it is sometimes worth providing a summary of the individual changes 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? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code Briefly describe how this PR is tested, and point to the specific tests you added. For example: 'This new feature is covered by the `sqllogictest` cases added in `foo.slt`'. If this PR does not add tests, explain why. For example, if the change is already covered by existing tests, please mention it. You should also check the `codecov` bot reply on this PR to confirm the changed code is exercised. --> - 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? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please add the `api change` label. --> The feature is opt-in and disabled by default. It can be enabled with: ```sql 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% |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
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
RightSemijoin onpart.p_partkey = lineitem.l_partkeyreceives approximately 6 million probe rows. With memtable inputs or the default Parquet configuration: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?
hash_join_probemicrobenchmark.What is the testing strategy for this PR?
hash_join_integer_prefilter.sltcovers 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:
Additional execution options control the maximum key range (default
262144), minimum pruning ratio (0.5), and sampling window (12288input rows). The metricsinteger_prefilter_created_countandprobe_prefilter_rows_prunedexpose whether the prefilter was created and how many probe rows skipped lookup.Query results are unchanged.
Benchmarks
Microbench:
hash_join_probeTPC-H with memtable
TPC-H with parquet
with the default
datafusion.execution.parquet.pushdown_filters=falseTarget HashJoin
join_timeThe following table measures the
CollectLeft/RightSemiHashMap join onpart.p_partkey = lineitem.l_partkeydescribed above. It reports operatorjoin_time, summed across the four probe partitions, separately from end-to-end query latency.