Repository navigation
perf: replace the routed hash join dynamic filter with one global filter (InList or bloom filter) - #26442
perf: replace the routed hash join dynamic filter with one global filter (InList or bloom filter)#26442adriangb wants to merge 3 commits into
Conversation
|
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 |
|
run benchmarks |
|
run benchmarks env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (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 hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpcds
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 hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (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 hash-join-global-bloom-filter (885155a) to 5f21582 (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 completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (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 hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (merge-base) diff Run configurationrun benchmark tpch10
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 hash-join-global-bloom-filter (885155a) to 5f21582 (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 hash-join-global-bloom-filter (885155a) to 5f21582 (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 completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (885155a) to 5f21582 (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 |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #26442 +/- ##
========================================
Coverage 82.92% 82.93%
========================================
Files 1154 1155 +1
Lines 457872 458727 +855
Branches 457872 458727 +855
========================================
+ Hits 379712 380465 +753
- Misses 55136 55217 +81
- Partials 23024 23045 +21 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
run benchmark tpch10 baseline:
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false
changed:
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU 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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU 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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 4
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 4
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 16
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 16
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 64
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 64
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "4"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "64"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "4"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "16"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "16"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "64"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "16"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "16"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 hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "64"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "4"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "4"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing hash-join-global-bloom-filter (a57d1f4) to 9411463 (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "64"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
Which issue does this PR close?
Depends on apache/arrow-rs#11641 (new
SbbfAPIs). Until arrow-rs releases them, this PR patches the workspace with[patch.crates-io]git dependencies toadriangb/arrow-rs:sbbf-batch-union-apis-60: the arrow-rs60.0.0tag (the version inCargo.lock) + apache/arrow-rs#11088 (Sbbf::estimated_fpp) + the changes of apache/arrow-rs#11641 (Sbbf::insert_hashes,Sbbf::check_hashes,Sbbf::union). No other arrow-rsmainchanges are included. The patch (first commit) must be removed before merge.Rationale for this change
The dynamic filter of a partitioned hash join routes each probe row to its partition with a
CASEon the repartition hash. Profiles show that theCASEmachinery costs about 6 / 50 / 150 ns per probe row at 1 / 16 / 128 partitions, and it is most of the query CPU time for join-heavy queries.Before (snapshot of
test_hashjoin_dynamic_filter_pushdown_partitioned, 12 partitions, two key columns):After:
With more build keys than the InList limit, the membership test is
bloom_filter(washash_lookupinside eachCASEbranch):TPC-H SF100, wall time as a ratio to
main(lower is better), 3 rounds. The A/A control (mainagainst itself) stayed within 0.99–1.04.main(0.98–1.01). The remaining 3–7% wall time is the wait at the build barrier.Sbbf::unionmove the fill into the parallel build tasks: fill takes 5–10 ms per partition, union ~0.5 ms, fold <1 ms.hash_join_bloom_filter_max_bytesis now 8 MiB.Methodology: dedicated GCP
c4a-standard-16VM (Arm, 64 GB), TPC-H SF100 parquet. Each config runs 3 A/B rounds with alternating order, plus an A/A control ofmainagainst itself in the same run. Ratios are paired per round, with a bootstrap 95% CI.What changes are included in this PR?
One global filter for all probe rows, with no routing:
CollectLeftjoins.IS NULL ORis kept for null-aware (NOT IN) and null-equal joins, as before.CASEfilter:BloomFilterExpr, new):Sbbfof the join key hashes (create_hasheswithHASH_JOIN_SEED, the same hashes the probe side computes). Multi-column keys use one filter of the combined hash. All key types the join supports work, including floats (-0.0and0.0hash the same).Sbbf::insert_hashes, sized for its distinct key count times the number of partitions (an upper bound for the whole build side), rounded to a power of two.Sbbf::union(a larger filter is folded on the fly), thenfold_to_target_fpp(fpp). No OS threads, locks or atomics. The build is eager.Sbbf::check_hashes.fppwould be larger thanhash_join_bloom_filter_max_bytes.estimated_fpp() > hash_join_bloom_filter_max_fpp(saturated; safety net).HashJoinBloomFilterreservation that the expression owns (released on drop). If a reservation fails, no bloom filter is built and the query does not fail.lit(true), likeHashTableLookupExpr(it holds runtime build state).KeyRangeBitmaps of all partitions (Prune hash join probe side via a build-side key range bitmap #25602) are merged withKeyRangeBitmap::try_unionand attached to the membership expression, so parquet keeps row group and page pruning, also withpushdown_filters = falseand when no bloom filter is built (key_range_pruninginEXPLAIN).New options:
datafusion.optimizer.hash_join_dynamic_filter_inlist_max_distinct_values20datafusion.optimizer.hash_join_bloom_filter_fpp0.01datafusion.optimizer.hash_join_bloom_filter_max_bytes8388608(8 MiB)datafusion.optimizer.hash_join_bloom_filter_max_fpp0.15hash_join_inlist_pushdown_max_distinct_values(default 150) keeps its meaning: the per-partition InList limit. It still decides whether each partition collects an InList, and the global InList needs every partition to have one. So a separate option for the global limit is cleaner than changing the meaning of the existing one.datafusion-physical-plannow depends onparquet(forSbbf).What is the testing strategy for this PR?
bloom_filter.rs: build keys are found and others rejected, NULL probe keys, merged filter equals one filter of all keys, partition filters of different sizes, too-big build side has no filter, saturated filter is not used, memory is reserved and released, small memory pool skips the filter, multi-column keys, float keys, proto encodes aslit(true).shared_bounds.rs: global bloom filter shape and results, InList vs bloom at the global limit, range partitioning uses the global filter, float keys use the global bloom filter, canceled partition falls back to the routed filter, small memory pool falls back to bounds, merged pruning bitmap, merged bounds ignore NULL bounds. Existing null-aware / null-equal / canceled tests pass.key_range_bitmap.rs:try_unionnever drops a key.dynamic_row_group_pruning.slt: newEXPLAIN ANALYZEof a partitioned join withpushdown_filters = false: the global filter, and 2 of 4 row groups pruned by the merged bitmap.filter_pushdown.rs,push_down_filter_parquet.slt,range_partitioning.slt,dynamic_row_group_pruning.slt,information_schema.slt.Are there any user-facing changes?
EXPLAINoutput of hash join dynamic filters changes (noCASE hash_repartition,bloom_filterorkey_range_pruninginstead ofhash_lookup).Decisions
Sbbf::unionand folds. No shared mutable filter, OS threads or lazy build.hash_join_bloom_filter_max_bytes(too big to be useful). The 15% saturation check stays as a safety net.HashTableLookupExpr.parquetdependency ofdatafusion-physical-planis accepted for now.Open questions
num_distinct × partitionscapped byhash_join_bloom_filter_max_bytes.🤖 Generated with Claude Code