Repository navigation
feat(perf): compute routing and lookup hashes once - #26249
Merged
Merged
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #26249 +/- ##
========================================
Coverage 82.92% 82.92%
========================================
Files 1154 1154
Lines 457790 458001 +211
Branches 457790 458001 +211
========================================
+ Hits 379637 379815 +178
- Misses 55135 55162 +27
- Partials 23018 23024 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
devanbenz
changed the title
Contributor
There was a problem hiding this comment.
From what I can gather, I like the approach of a specialized partitioning expression and after looking at the code for some time it makes sense. I also can't think of an easy way from the top of my head to improve the CASE evaluation in such a way that this specialized expression isn't necessary.
I've left a few notes that you can think of integrating. As I am not too familiar with the joining ecosystem I can't really approve it but to me the approach makes sense.
I think the only thing where this could be counter-productive is, if the data source can apply pruning with the CASE expression but not with the new special-purpose expression. I am not sure whether this happening or is even worth doing. Probably an optimization that transforms small multi-partition build sides into an IN LIST expression is the best way for this scenario.
Contributor
Author
|
@tschwarzinger I'm actually thinking more about this and how I can get away from specialization that requires adding a whole new Expr. I was thinking about maybe folding this functionality in to https://docs.rs/datafusion/latest/datafusion/physical_plan/joins/struct.HashTableLookupExpr.html and just slightly modifying the existing operator 🤔 I'll get back to this PR later today or tomorrow, but, that's where my head is at right now. |
Contributor
We already emit in list expressions for small builds sides, is there something I'm missing? |
Contributor
Author
I was able to reproduce this and generate a table + query (with the help of my buddy claude) to get pruning in the CASE expr working, but, not working in my specialization: Parquet file information: Results of pruning: Where the query is like so (including building parquet files): Going to continue playing with this and getting it to a much better state. |
Contributor
What I meant was over multiple partitions. I am not sure whether we have anything there. For example: becomes |
Contributor
|
Exploration also needs to consider number of partitions involved (i've looked at implementations before that are faster at 1->64 partitions but get slower after that), selectivity of the join, size of each side, pruning potential, size/length of the values being hashed, etc. There are many factors 😓 |
Contributor
|
Thank you for investigating that! From what I can tell, the pruning for hash lookups is done via |
adriangb
reviewed
Oct 16, 2026
adriangb
left a comment
adriangb
left a comment
Contributor
There was a problem hiding this comment.
Thank you for this PR, @devanbenz.
I ran some benchmarks, results look good. TPC-H SF100, CPU time vs main, lower is better. N is the number of partitions:
| Query | N=16 | N=128 |
|---|---|---|
| Q3 | 0.74 | 0.52 |
| Q5 | 0.72 | 0.46 |
| Q9 | 0.64 | 0.38 |
| Q21 | 0.67 | 0.41 |
This PR also looses pushdown:
-- Data: probe sorted by k, small row groups and pages; build has 5,000 sparse keys.
COPY (SELECT value AS k, value * 2 AS v FROM generate_series(0, 1999999) ORDER BY k)
TO '/tmp/mre/probe.parquet' OPTIONS (max_row_group_size 20000, data_page_row_count_limit 2000);
COPY (SELECT value AS k FROM generate_series(0, 1999999) WHERE value % 200000 < 500)
TO '/tmp/mre/build.parquet';
CREATE EXTERNAL TABLE probe STORED AS PARQUET LOCATION '/tmp/mre/probe.parquet';
CREATE EXTERNAL TABLE build STORED AS PARQUET LOCATION '/tmp/mre/build.parquet';
SET datafusion.execution.target_partitions = 16;
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
EXPLAIN ANALYZE SELECT count(*) FROM build b JOIN probe p ON b.k = p.k;Probe DataSourceExec metrics:
main |
this PR | |
|---|---|---|
| predicate | CASE hash_repartition % 16 WHEN 0 THEN ... AND hash_lookup ... |
k@0 >= 0 AND k@0 <= 1800499 AND partitioned_hash_lookup(16) |
| pruning_predicate | min/max AND KEY_RANGE_BITMAP(...) per partition |
min/max only |
| row_groups_pruned_statistics | 100 total → 14 matched | 100 total → 91 matched |
| page_index_rows_pruned | 280.0 K total → 26.05 K matched | 1.82 M total → 1.80 M matched |
| output_rows | 26.05 K | 1.80 M |
Some more issues we should resolve before moving forward with this in review comments.
| lookups.push(PartitionLookup::Map(Arc::clone(map))); | ||
| Some(lookups) | ||
| } | ||
| _ => None, |
Contributor
There was a problem hiding this comment.
This PR falls back to CASE at high target_partitions. That is where the CASE path costs the most: its cost grows about linearly with N (microbenchmark: about 6 / 50 / 150 ns/row at N=1 / 16 / 128).
Since this arm returns None as soon as one partition is not a hash map, and at high N each partition gets few keys, so some partitions pick InList (the per-partition 150-distinct threshold). Then the whole filter falls back to CASE hash_repartition % N.
MRE (uses probe.parquet from the other comment):
SET datafusion.execution.target_partitions = 128;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
EXPLAIN ANALYZE SELECT count(*)
FROM (SELECT value AS k FROM generate_series(1, 19200)) b
JOIN probe p ON b.k * 97 = p.k;| HASH_JOIN_SEED, | ||
| lookups, | ||
| )) as Arc<dyn PhysicalExpr>; | ||
| let bounds = merged_bounds |
Contributor
There was a problem hiding this comment.
The merged bounds ignore canceled partitions; is that intended? It looks safe because a canceled partition’s rows have no consumer.
A CanceledUnknown partition pushes PartitionLookup::Const(true) but adds nothing to the bounds. The bounds predicate is then ANDed for every probe row. So a row routed to the canceled partition is removed if its key is outside the other partitions' range. That row can have a match in the canceled partition.
The CASE path keeps these rows, because each partition has its own bounds and a canceled partition gives true.
Possible fix (needs checking): do not use the merged bounds when a partition is canceled.
Suggested change
| let bounds = merged_bounds | |
| let bounds = merged_bounds.filter(|_| !has_canceled_unknown) |
| on_columns: Vec<PhysicalExprRef>, | ||
| routing_state: SeededRandomState, | ||
| lookup_state: SeededRandomState, | ||
| partitions: Vec<PartitionLookup>, |
Contributor
There was a problem hiding this comment.
Related to the pruning regression (see the comment on shared_bounds.rs): this expression has no equivalent of HashTableLookupExpr::pruning_bitmap(), so PruningPredicate cannot use it.
Suggestion:
- Store one combined
Option<Arc<KeyRangeBitmap>>here. It is the union of the per-partition bitmaps. If any partition has no bitmap (or isConst(true)), useNone. - Expose it with
pruning_bitmap(), and letpruning_predicate.rshandlePartitionedLookupExprthe same way asHashTableLookupExpr.
pub(crate) struct PartitionedLookupExpr {
on_columns: Vec<PhysicalExprRef>,
routing_state: SeededRandomState,
lookup_state: SeededRandomState,
partitions: Vec<PartitionLookup>,
/// Union of the per-partition key-range bitmaps, for row group and page pruning.
pruning_bitmap: Option<Arc<KeyRangeBitmap>>,
}
Contributor
Author
There was a problem hiding this comment.
@adriangb I'm actually thinking about utilizing HashTableLookupExpr instead of creating a new specialized operator--while keeping it's API surface the same/internals the same if possible so I can utilize PruningPredicate and all the other niceties of HashTableLookupExpr.
Contributor
Author
There was a problem hiding this comment.
Also see: #26249 (comment) where my operator failed to prune while CASE did it as expected.
Contributor
Author
There was a problem hiding this comment.
New changes allow pruning, I went ahead and utilized HashTableLookupExpr. Using
-- Data: probe sorted by k, small row groups and pages; build has 5,000 sparse keys.
COPY (SELECT value AS k, value * 2 AS v FROM generate_series(0, 1999999) ORDER BY k)
TO '/tmp/mre/probe.parquet' OPTIONS (max_row_group_size 20000, data_page_row_count_limit 2000);
COPY (SELECT value AS k FROM generate_series(0, 1999999) WHERE value % 200000 < 500)
TO '/tmp/mre/build.parquet';
CREATE EXTERNAL TABLE probe STORED AS PARQUET LOCATION '/tmp/mre/probe.parquet';
CREATE EXTERNAL TABLE build STORED AS PARQUET LOCATION '/tmp/mre/build.parquet';
SET datafusion.execution.target_partitions = 16;
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
EXPLAIN ANALYZE SELECT count(*) FROM build b JOIN probe p ON b.k = p.k;Here is the DataSourceExec
| | DataSourceExec: file_groups={16 groups: [[tmp/mre/probe.parquet:0..657492], [tmp/mre/probe.parquet:657492..1314984], [tmp/mre/probe.parquet:1314984..1972476], [tmp/mre/probe.parquet:1972476..2629968], [tmp/mre/probe.parquet:2629968..3287460], ...]}, projection=[k], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ k@0 >= 0 AND k@0 <= 1800499 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=k_null_count@1 != row_count@2 AND k_max@0 >= 0 AND k_null_count@1 != row_count@2 AND k_min@3 <= 1800499 AND k_null_count@1 != row_count@2 AND KEY_RANGE_BITMAP(k_min@3, k_max@0, 2508/1048576 buckets), required_guarantees=[], metrics=[output_rows=26.05 K, elapsed_compute=18.74ms, output_bytes=896.0 KB, output_batches=14, files_ranges_pruned_statistics=16 total → 16 matched, row_groups_pruned_statistics=100 total → 14 matched, page_index_pages_pruned=140 total → 14 matched, page_index_rows_pruned=280.0 K total → 26.05 K matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_processed=10.0 MB, bytes_scanned=328.0 KB, file_open_errors=0, file_scan_errors=0, files_opened=16, files_processed=16, num_predicate_creation_errors=0, page_index_load_skipped=5, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=0, predicate_cache_records=0, bloom_filter_eval_time=82.82µs, metadata_load_time=2.66ms, page_index_eval_time=2.14ms, row_pushdown_eval_time=43ns, statistics_eval_time=5.01ms, time_elapsed_opening=54.47ms, time_elapsed_processing=78.27ms, time_elapsed_scanning_total=42.34ms, time_elapsed_scanning_until_data=32.50ms, output_rows_skew=37.62%, scan_efficiency_ratio=3.19% (335.9 K/10.52 M)] |
| |
More specifically the pruning metrics:
metrics=[output_rows=26.05 K, elapsed_compute=18.74ms, output_bytes=896.0 KB, output_batches=14, files_ranges_pruned_statistics=16 total → 16 matched, row_groups_pruned_statistics=100 total → 14 matched, page_index_pages_pruned=140 total → 14 matched, page_index_rows_pruned=280.0 K total → 26.05 K matched,
github-actions
Bot
added
core
Building one hash table over every build partition's keys ran on a single thread at the dynamic filter barrier, so every probe stream waited on it (about 0.7s for a 15M key build at TPC-H SF10). HashTableLookupExpr now has a routed mode: each probe row is looked up in the table of the build partition its key was repartitioned to (hash or range routing), sharing the partitions' tables as they are. It stays a HashTableLookupExpr, so pruning keeps using its merged pruning bitmap. Removes union_hash_maps and JoinHashMapType::hash_values.
Contributor
Author
|
@adriangb When you get a moment could you please run your same benchmarks against this branch again. I pushed up some changes that I was cooking on over the past couple days to improve performance.
Locally I'm seeing:
* Caused by one outlier round (241.8s vs ~125s in the other two rounds). Using the following machine i7-12700K (20 threads, x86), 32 GB RAM With pushdown filters off - all tpch:
With pushdown filters on - all tpch
|
adriangb
approved these changes
Oct 28, 2026
adriangb
left a comment
adriangb
left a comment
Contributor
There was a problem hiding this comment.
Overall this looks good to me. If you're able to rebase, fix review comments and mark as ready we can move forward with this. Even if we end up adding a bloom filter option (because it's faster or maybe just because it can be serialized across the wire), and even if this doesn't really fully resolve the issues with hash join dynamic filter overhead, it's certainly better than the status quo.
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark hash_join_semi_antiCPU Details (lscpu)Details
Resource Usagehash_join_semi_anti — base (merge-base)
hash_join_semi_anti — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing partition_lookup_experiment (7c6a3f5) to 611311c (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Benchmark summary for
|
| Suite | Pushdown filters | Query | HEAD | PR | Change |
|---|---|---|---|---|---|
| tpch10 | default | — | no change >10% | ||
| tpch10 | off | — | no change >10% | ||
| tpch10 | on | Q2 | 107.46 ms | 92.52 ms | 1.16x faster |
| tpch10 | on | Q3 | 308.31 ms | 242.11 ms | 1.27x faster |
| tpch10 | on | Q4 | 297.67 ms | 206.97 ms | 1.44x faster |
| tpch10 | on | Q5 | 439.05 ms | 312.88 ms | 1.40x faster |
| tpch10 | on | Q7 | 442.02 ms | 345.36 ms | 1.28x faster |
| tpch10 | on | Q10 | 374.52 ms | 320.44 ms | 1.17x faster |
| tpch10 | on | Q12 | 303.00 ms | 251.53 ms | 1.20x faster |
| tpch10 | on | Q13 | 401.97 ms | 340.31 ms | 1.18x faster |
| tpch10 | on | Q16 | 88.62 ms | 73.69 ms | 1.20x faster |
| tpch10 | on | Q17 | 491.60 ms | 296.74 ms | 1.66x faster |
| tpch10 | on | Q18 | 1629.72 ms | 1330.86 ms | 1.22x faster |
| tpch10 | on | Q20 | 279.70 ms | 231.31 ms | 1.21x faster |
| tpch10 | on | Q21 | 1234.26 ms | 941.86 ms | 1.31x faster |
| tpch10 | on | Q22 | 134.42 ms | 78.97 ms | 1.70x faster |
| tpch (sf1) | default / off | — | no change >10% | ||
| tpch (sf1) | on | Q12 | 40.35 ms | 34.35 ms | 1.17x faster |
| tpch (sf1) | on | Q17 | 71.86 ms | 44.79 ms | 1.60x faster |
| tpch (sf1) | on | Q18 | 141.28 ms | 111.12 ms | 1.27x faster |
| tpcds | default (2 runs) | Q65 | ~130 ms | ~73 ms | 1.75–1.78x faster |
| tpcds | default (1 of 2 runs) | Q46 | 12.63 ms | 11.35 ms | 1.11x faster |
| tpcds | off | Q65 | 129.23 ms | 70.82 ms | 1.82x faster |
| tpcds | on | Q17 / Q25 / Q29 / Q50 | 2.14x / 2.20x / 1.86x / 2.02x faster | ||
| tpcds | on | Q18 / Q64 / Q65 / Q72 | 1.43x / 1.56x / 1.69x / 1.27x faster | ||
| tpcds | on | Q75 / Q85 / Q95 | 1.17x / 1.40x / 1.33x faster | ||
| clickbench_partitioned | default (1 of 2 runs) | Q35 | 296.48 ms | 330.52 ms | 1.11x slower |
| clickbench_partitioned | default (1 of 2 runs) | Q39 | 139.09 ms | 155.91 ms | 1.12x slower |
| clickbench_partitioned | on | Q37 | 35.47 ms | 39.34 ms | 1.11x slower |
| clickbench_partitioned | off | Q38 | 39.32 ms | 43.60 ms | 1.11x slower |
| prepared_hash_join | default | — | all within ±6% | ||
| hash_join_semi_anti | default | — | all within ±5% |
Takeaways
- @devanbenz: with filter pushdown off, tpch10 shows no query changing by more than 10% in either direction. The earlier regression did not reproduce.
- With pushdown on, tpch10 and tpcds both get much faster across the board, by up to about 2x.
- The only slowdowns are on four clickbench queries, each about 1.11x. Each one showed up in only one of the runs, so I read them as noise.
Contributor
|
LGTM! Can you:
|
# Conflicts: # xtask/src/ci_steps.rs
adriangb
reviewed
Nov 2, 2026
adriangb
approved these changes
Nov 2, 2026
adriangb
enabled auto-merge
November 2, 2026 17:54
github-merge-queue
Bot
removed this pull request from the merge queue due to failed status checks
Nov 2, 2026
adriangb
pushed a commit
to adriangb/datafusion
that referenced
this pull request
Nov 5, 2026
…olds it (apache#26994) ## Which issue does this PR close? - Closes apache#26980. ## Rationale for this change A `Partitioned` hash join frees its build reservation when the partition's stream ends, but the dynamic filter and the build accumulator keep the build map until the plan is dropped, so a finished join's maps stay allocated while the pool counts them as free. In a query with several stages, later operators reserve the whole pool on top of them, and the process can be killed by the OS instead of failing with `ResourcesExhausted`. apache#26980 has the cause and the measurements. This takes the first fix direction from the issue: the reservation travels with the map, so the memory stays charged for as long as anything holds the map. This does not reduce memory use or free the maps sooner. A memory-limited query that only fit because a finished join's maps were not counted can now fail with `ResourcesExhausted`, and the finished join then shows up among the top consumers in the error. ## What changes are included in this PR? - `ReservedMap` (`joins/mod.rs`): a `Map` and the reservation its memory is charged to, in one `Arc`. `JoinBuildData`, `PushdownStrategy::Map` and `HashTableLookupExpr` hold it in place of `Arc<Map>`, so every holder of the map holds its charge. - `build_in_memory`: charges the hash table or array map and the pruning bitmap to a reservation created with `new_empty()` from the build reservation, so the consumer name and error text do not change. The batch, key arrays and visited bitmap stay on the build reservation and are released with the stream as before. - `HashTableLookupExpr::new`: keeps its signature and wraps each map without a reservation, for the `datafusion-proto` and `datafusion-pruning` tests. `new_reserved` is the crate-internal constructor the join uses. - `PreparedHashJoinBuild::reserved_bytes`: sums both reservations, so its value is unchanged. Not in this PR: - Releasing the maps once every partition has finished probing, the second fix direction in apache#26980, frees the memory sooner but changes when the filter is updated, so it is left for a follow-up. I can open an issue for it. - `PushdownStrategy::InList` arrays share the prepared build batch and have the same lifetime gap, bounded by `hash_join_inlist_pushdown_max_size` (128 KiB per partition by default). - *Added after merging apache#26249:* the hash tables that the routed lookup builds from IN-list partitions (`in_list_hash_map`) and the merged pruning bitmap are not reserved either. The hash tables grow with the IN lists, which the same limit caps, and the bitmap is at most 128 KiB. ## What is the testing strategy for this PR? - `build_map_stays_reserved_with_dynamic_filter` in `hash_join/exec.rs`, over `CollectLeft` and `Partitioned`. After the join's stream is dropped, `pool.reserved()` equals the map plus pruning bitmap bytes (`Partitioned`) or the whole build (`CollectLeft`). After the plan is dropped, the filter is the last holder and both modes equal the map plus bitmap bytes. After the filter is dropped, it is 0. On `main` the `Partitioned` case fails at the first assertion with 0 reserved against 136,288 expected; ~~`CollectLeft` passes on `main` too.~~ *Corrected in review:* `CollectLeft` fails after the plan is dropped, with 0 reserved against 136,288 expected, so the test also covers a filter that outlives the plan. - *Added in review:* `array_map_stays_reserved_with_dynamic_filter`, over the same two modes. Its keys are dense, unlike the first test's, so the build map is an array map without a pruning bitmap. After the plan is dropped, `pool.reserved()` equals the `ArrayMap` estimate; after the filter is dropped, it is 0. On `main` both modes fail after the plan is dropped with 0 reserved against 160,000 expected. With only the `array_map_key_range` argument reverted to the build reservation, this test fails and the other 1,820 `joins::` tests pass. - *Added after merging apache#26249:* `routed_maps_stay_reserved_with_dynamic_filter` runs the `Partitioned` join over two hash partitions, so the filter is the routed lookup. The maps stay reserved after every stream ends, the amount does not change when the plan is dropped, and it is 0 after the filter is dropped. On `main` nothing is reserved once the streams end. If the routed lookup holds the maps without their reservation, this test fails and no other `joins::` test does. - The test from apache#26980, run on this branch (not added here: it needs a global allocator): | Run | Allocated after the stream is dropped | Reserved | Reserved on `main` | |---|---|---|---| | 1. `Partitioned`, dynamic filter off | 0.2 MiB | 0.0 MiB | 0.0 MiB | | 2. `Partitioned`, dynamic filter on | 151.5 MiB | 151.3 MiB | 0.0 MiB | | 3. `CollectLeft`, dynamic filter on | 182.0 MiB | 181.8 MiB | 181.8 MiB | Dropping the plan frees everything in all three runs. ## Are there any user-facing changes? No API change; `HashTableLookupExpr::new` is unchanged. Under a memory limit, a query whose finished joins' maps were not counted before can now fail with `ResourcesExhausted`, and spilling operators under the same limit can spill earlier. AI assistance: written with Claude Code (Claude Fable 5.1). I reported the original case, reproduced the issue, reviewed the diff and ran the tests above. --------- Co-authored-by: Jay Zhan <[email protected]>
vim89
pushed a commit
to vim89/datafusion
that referenced
this pull request
Nov 11, 2026
…h join filter (apache#27091) ## Which issue does this PR close? - This PR does not close an issue. It follows up on apache#26249, which merged the build partitions of a `Partitioned` hash join into one routed lookup. > [!NOTE] > Stacked on apache#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 `Partitioned` hash join on dense integer keys runs much slower with Parquet filter pushdown on than with it off. On TPC-H SF10 the `hj` Q26 join takes 93.5 ms with `pushdown_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 apache#26249 accepts only hash tables, so these joins fall back to a `CASE` with one branch per partition. The scan evaluates that `CASE` on every row. ## What changes are included in this PR? 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 by hash. Every non-empty partition can now merge into the routed lookup. The `CASE` fallback remains only for canceled partitions. A follow-up PR removes it. ## What is the testing strategy for this PR? `routed_lookup_fuzz` now builds an `ArrayMap` for about half of its partitions. It compares the routed lookup with one lookup over the whole build side. All `datafusion-physical-plan` tests and all sqllogictests pass. `hj` on TPC-H SF10, 16 partitions, median of 10 iterations: | Query | `pushdown_filters` | `main` + apache#27090 | This PR | |---|---|---|---| | Q26 (dense keys) | `true` | 93.5 ms | 44.7 ms | | Q26 (dense keys) | `false` | 50.0 ms | 48.2 ms | | Q27 (sparse keys) | `true` | 59.2 ms | 57.3 ms | | Q27 (sparse keys) | `false` | 54.5 ms | 55.6 ms | Only Q26 with pushdown on changes. Its probe filter goes from `CASE hash_repartition % 16 WHEN 0 THEN ... hash_lookup ...` to one `hash_lookup`. ## Are there any user-facing changes? No API changes. A `Partitioned` hash join on dense integer keys runs faster when Parquet filter pushdown is on. 🤖 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?
Performance
This PR improves TPC-H queries across the board with pushdown filters enabled
pushdown_filterstruefalsemainvsmain: 1.01x)without pushdown filters there does not appear to be a regression and differences are due to benchmark noise most likely.
Per-query SF10 results
TPC-H SF10 parquet, i7-12700K, 20 partitions,
bench.sh run tpch10.main(49c67c296) and this PR (5c72efaf2) were run alternately, so background load affects both equally.pushdown_filters = truepushdown_filters = falseRationale for this change
For
Partitionedhash joins, the probe-side dynamic filter is aCASE hash_repartition % N WHEN i THEN bounds_i AND hash_lookup_i ..., whose per-row cost grows withN. It was the main cost of TPC-H Q17 withpushdown_filterson.What changes are included in this PR?
When every non-empty partition has an IN list or a hash table and none was canceled, the
CASEis replaced by one merged filter:HashTableLookupExprthat routes each probe row (hash % N, orRangeExpr) to its partition's existing hash table. No merged table is built.[min, max]bounds, plus a mergedKeyRangeBitmap(KeyRangeBitmap::union), so row-group and page pruning matchmain.CASEis kept when a partition was canceled or uses anArrayMap.HashTableLookupExpr::new(#[doc(hidden)]) is now the single constructor, takingOption<PartitionRouter>and per-partition maps.What is the testing strategy for this PR?
routed_lookup_fuzz: routed lookup vs. a single-table lookup over all keys.union_fuzz:KeyRangeBitmap::unionvs.KeyRangeBitmap::try_newover all keys.shared_bounds.rsandpruning_predicate.rs, andrange_partitioning.slt.spill-pool-fuzzstep is nowfuzzand runs both fuzzers with 10k seeds.Are there any user-facing changes?
No change to query results.
EXPLAINshowsk >= min AND k <= max AND hash_lookupinstead of theCASE.