Repository navigation
perf: route probe rows to ArrayMap build partitions in the merged hash join filter - #27091
Merged
Merged
Conversation
A `Partitioned` hash join merges its partitions' build maps into one routed `hash_lookup` filter. The routed lookup accepted only hash tables, so a join whose partitions built an `ArrayMap` (dense integer keys) fell back to a `CASE` over one filter per partition. The routed lookup now probes an `ArrayMap` partition with the key values of the rows routed to it, because an `ArrayMap` is keyed by value, not hash. Every non-empty partition can now merge. The `CASE` fallback remains only for canceled partitions. The routed lookup fuzz test now builds an `ArrayMap` for about half of the partitions. Co-Authored-By: Claude Opus 5.5 <[email protected]>
adriangb
force-pushed
the
routed-lookup-array-map
branch
from
November 4, 2026 21:56
c133310 to
97db09b
Compare
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmark hj |
Contributor
Author
|
run benchmark tpch10 |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: Runner log (last 40 lines)Kubernetes messageFile an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: Runner log (last 40 lines)Kubernetes messageFile an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: Runner log (last 40 lines)Kubernetes messageFile an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing routed-lookup-array-map (97db09b) to 70f2ed0 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing routed-lookup-array-map (97db09b) to 70f2ed0 (merge-base) diff Run configurationrun benchmark tpch10Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing routed-lookup-array-map (97db09b) to 70f2ed0 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing routed-lookup-array-map (97db09b) to 70f2ed0 (merge-base) diff Run configurationrun benchmark tpch10CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
…ed-lookup-array-map # Conflicts: # datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs # datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
jayzhan211
requested changes
Nov 7, 2026
jayzhan211
left a comment
jayzhan211
left a comment
Contributor
There was a problem hiding this comment.
@adriangb thanks for this. In shared_bounds.rs around line 994, please drop the all-Range early return in merged_pruning_bitmap. Dense partitions usually have no bitmap, so this now returns None for them, and Range-partitioned joins stop pruning the gaps between partitions: base prunes 4 of 4 probe files, this PR prunes 1 of 4, and removing the return prunes 4 of 4. union already returns None when every bucket is set.
- /// bounds. `None` when no partition narrows its bounds, as the bounds
- /// predicate already prunes that much.
+ /// bounds. `None` when nothing can be excluded.- if parts
- .iter()
- .all(|part| matches!(part, KeyRangePart::Range { .. }))
- {
- return None;
- }
KeyRangeBitmap::union(&merged.min, &merged.max, &parts).map(Arc::new)Repro setup and a regression test (the test needs `array_map_of` from my comment on the merge gate)
Repro: Partitioned join, Range splits (10, 20, 30), hash_join_inlist_pushdown_max_distinct_values = 0. Build keys {7,8,9} {17,18,19} {27,28,29} {37,38,39}, probe range_partitioned. Bytes scanned: base 0 B, this PR 126 B, with the fix 0 B.
#[test]
fn disjoint_array_map_partitions_keep_gap_pruning() {
let acc = make_partitioned_expr_accumulator_for_test(2);
let members = [
(0, array_map_of(&[7, 8, 9]), bounds(7, 9)),
(1, array_map_of(&[37, 38, 39]), bounds(37, 39)),
];
let bitmap = acc
.merged_pruning_bitmap(&members, Some(&bounds(7, 39)))
.expect("the gap between the partitions can be pruned");
assert!(bitmap.may_contain_range(7, 9));
assert!(!bitmap.may_contain_range(10, 36));
}Optional cleanup: a few comments still describe hash tables only, now that partitions can be array maps. They are the Lookup::Routed doc (partitioned_hash_eval.rs line 284), the evaluate_routed doc (line 421), and the merged_membership_filter doc (shared_bounds.rs line 860).
| // An array map is keyed by value, not hash, so it probes | ||
| // the run's own key values. | ||
| Map::ArrayMap(map) => { | ||
| let indices = UInt32Array::from(rows[run.clone()].to_vec()); |
Contributor
There was a problem hiding this comment.
When every partition is an array map, with_hashes and sorted_hashes do per-row work that nothing reads. Skipping them took lookup cost from 4.8 to 4.1 ns/row at 16 partitions and batch 8192, measured with a local variant on a busy laptop. It's fine to handle this in a follow-up, since it only saves time and needs the closure reshaped. The condition to hash:
let needs_hashes = maps.iter().flatten().any(|map| {
let map: &Map = map;
matches!(map, Map::HashMap(_))
});
Contributor
Author
There was a problem hiding this comment.
Left for a follow-up, as you suggested: this PR keeps hashing every row.
🤖 Addressed by Claude Code
Array map partitions now join the merged lookup, and a dense partition usually has no pruning bitmap, so it contributes its whole bounds. The merged pruning bitmap returned `None` when every partition contributed only bounds, so a Range-partitioned join stopped pruning the probe containers that fall in the gaps between partitions. `union` already returns `None` when no bucket can be excluded, so the early return goes. This also adds tests that array map partitions reach the merged filter under hash and range routing. The fuzz test keeps hash tables under `force_hash_collisions`, where a hash table matches every key and an array map would disagree with the reference. The remaining changes are doc wording for array maps, `take_arrays`, and a simpler test helper. Co-Authored-By: Claude Opus 5.5 <[email protected]>
Contributor
Author
|
@jayzhan211 could you take another look here? Thanks! |
Contributor
Author
Thank you for helping review! |
geoffreyclaude
pushed a commit
to geoffreyclaude/datafusion
that referenced
this pull request
Nov 12, 2026
…ray map (apache#27324) ## Which issue does this PR close? - Follow-up to apache#27091, from review feedback there. ## Rationale for this change A `Partitioned` hash join with dense integer keys builds an `ArrayMap` in every partition. Its dynamic filter then computes a hash for every probe row, but an `ArrayMap` looks keys up by value, so nothing reads those hashes. Skipping them makes the filter cheaper on every probe batch. ## What changes are included in this PR? `HashTableLookupExpr::evaluate_routed` hashes the probe keys only when at least one partition is a hash table. The counting sort by partition runs either way. ## What is the testing strategy for this PR? - A new unit test builds every partition as an `ArrayMap` and checks exact matches, including a null probe row. - The routed lookup fuzz test passes with and without `force_hash_collisions`. - Lookup cost on a GCP VM (16 partitions, batch 8192): 7.49 → 6.56 ns/row. ## Are there any user-facing changes? No. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <[email protected]>
andygrove
pushed a commit
to andygrove/datafusion
that referenced
this pull request
Nov 12, 2026
…he#27092) ## Which issue does this PR close? - This PR does not close an issue. It follows up on apache#21666, which added canceled partitions to the hash join dynamic filter to fix apache#21625. ## Rationale for this change A `Partitioned` hash join cancels one of its partitions when a parent operator drops that partition's stream before it reports its build side. For example, a parent join partition whose own build side is empty drops its probe-side stream unpolled. The dynamic filter keeps every probe row routed to a canceled partition, because that partition's build keys are unknown. Keeping those rows buys nothing. A stream starts to probe only after its report is delivered, so a canceled stream has emitted no row and never will. The shared probe input still reads the rows routed to it, and `RepartitionExec` then discards them. Keeping them also costs something: - A canceled partition forces the routed `CASE` fallback instead of the merged lookup. After apache#27091 it is the only thing that does. - Under `NullEqualsNull`, a canceled partition makes every live partition keep probe-side NULL rows. ## What changes are included in this PR? Three commits: 1. `test:` Two dynamic filter tests checked range routing for `-0.0` and for composite nullable keys through a canceled partition's `CASE` branch. The commit moves both cases into `RangeRouter` tests, before the next commit removes that branch. 2. `fix:` A canceled partition counts as empty, so the filter rejects the rows routed to it. The commit rewrites the tests that asserted the old behavior. 3. `refactor:` The `CASE` fallback has no input left. The commit removes it, the per-partition filters it used, and the test assertions that a filter is not a `CASE`. When exactly one partition holds rows, the filter is still that partition's own IN list or hash lookup, so the probe scan keeps bloom filter pruning. 4. `fix:` Restores the branch for a lone non-empty partition, which the refactor first dropped. Adds a test for it and for a canceled partition next to live ones, and removes the remaining `CASE` wording. This is +155 / −345 lines. ## What is the testing strategy for this PR? - The new `RangeRouter` tests keep the routing cases that the removed tests covered. - `partitioned_canceled_and_empty_partitions_reject_every_row` and `partitioned_canceled_partition_does_not_widen_filter` cover the new behavior. - `test_partitioned_dynamic_filter_reports_empty_canceled_partitions`, the regression test for apache#21625, still passes. - All `datafusion-physical-plan` tests, all sqllogictests, and `core_integration` pass. I also checked the behavior on a query that cancels partitions. In `SELECT x.k FROM (SELECT big.k FROM mid JOIN big ON mid.k = big.k) x WHERE x.k IN (SELECT k FROM tiny)` with 4 partitions, the parent `RightSemi` join cancels two child partitions. The result is the same with and without this PR. ## Are there any user-facing changes? No. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <[email protected]>
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
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?
Partitionedhash join into one routed lookup.Note
Stacked on #27090, which adds the benchmark queries used below. Review only the last commit,
perf: route probe rows to ArrayMap build partitions in the merged filter.Rationale for this change
A
Partitionedhash join on dense integer keys runs much slower with Parquet filter pushdown on than with it off. On TPC-H SF10 thehjQ26 join takes 93.5 ms withpushdown_filters=true, against 50.0 ms with it off. A similar join on sparse keys (Q27) does not slow down this way.The cause is the filter the join pushes into its probe-side scan. Each build partition of a dense-key join builds an
ArrayMap. The routed lookup from #26249 accepts only hash tables, so these joins fall back to aCASEwith one branch per partition. The scan evaluates thatCASEon every row.What changes are included in this PR?
The routed lookup now probes an
ArrayMappartition with the key values of the rows routed to it, because anArrayMapis keyed by value, not by hash. Every non-empty partition can now merge into the routed lookup. TheCASEfallback remains only for canceled partitions. A follow-up PR removes it.What is the testing strategy for this PR?
routed_lookup_fuzznow builds anArrayMapfor about half of its partitions. It compares the routed lookup with one lookup over the whole build side. Alldatafusion-physical-plantests and all sqllogictests pass.hjon TPC-H SF10, 16 partitions, median of 10 iterations:pushdown_filtersmain+ #27090truefalsetruefalseOnly Q26 with pushdown on changes. Its probe filter goes from
CASE hash_repartition % 16 WHEN 0 THEN ... hash_lookup ...to onehash_lookup.Are there any user-facing changes?
No API changes. A
Partitionedhash join on dense integer keys runs faster when Parquet filter pushdown is on.🤖 Generated with Claude Code