Repository navigation
perf: replace the routed hash join dynamic filter with one global filter (InList or bloom filter) - #26442
Draft
adriangb wants to merge 4 commits into
Draft
perf: replace the routed hash join dynamic filter with one global filter (InList or bloom filter)#26442adriangb wants to merge 4 commits into
adriangb wants to merge 4 commits into
Conversation
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 |
Contributor
Author
|
run benchmarks |
Contributor
Author
|
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 |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
Contributor
Author
|
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:
|
Contributor
Author
|
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 (6a2eada) to 9e0a08f (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 running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (6a2eada) to 9e0a08f (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 |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 2
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 2
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 128
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 128
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
Contributor
Author
|
run benchmark tpcds env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 2
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
Contributor
Author
|
run benchmark tpcds env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 2
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
Contributor
Author
|
run benchmark tpcds env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 128
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: false |
Contributor
Author
|
run benchmark tpcds env:
DATAFUSION_EXECUTION_TARGET_PARTITIONS: 128
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (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 hash-join-global-bloom-filter (6a2eada) to 9e0a08f (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 (6a2eada) to 9e0a08f (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 (6a2eada) to 9e0a08f (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 running (GKE) | trigger CPU Details (lscpu)Comparing hash-join-global-bloom-filter (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpch10
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "2"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 (6a2eada) to 9e0a08f (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
DATAFUSION_EXECUTION_TARGET_PARTITIONS: "128"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
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?
Depends on apache/arrow-rs#11641 (
Sbbf::insert_hashes,Sbbf::check_hashes,Sbbf::union) and apache/arrow-rs#11088 (Sbbf::estimated_fpp). Until they are released, the first commit patches arrow-rs toadriangb/arrow-rs:sbbf-batch-union-apis-60(the60.0.0tag plus those two PRs). Remove that commit before merge.Rationale for this change
For a partitioned hash join with many build keys, the dynamic filter on
mainis a routedhash_lookup: each probe row is hashed to its build partition and looked up in that partition's hash table. This has three costs:This PR replaces it with one global filter that does not depend on partitioning:
Example (
dynamic_row_group_pruning.slt):Benchmark bot, total time against
main(negative is faster):The tpch10 regression with pushdown off is in the bloom filter build. A profile of Q7 and Q18 puts the extra CPU in
Sbbf::insert_hash: cache misses into partition filters that are larger than needed (see Open questions).What changes are included in this PR?
Review commit by commit:
chore: patch arrow-rs: temporary, remove before merge.perf: replace the routed hash join dynamic filter with one global filter:BloomFilterExpr: a parquetSbbfof the join key hashes (HASH_JOIN_SEED, the hashes the probe side already computes). It supports every join key type, including multi-column keys.Sbbf::unionand folded to the target fpp. No locks or shared state.Sbbf::check_hashes.HashJoinBloomFilterreservation. If a reservation fails, the filter has no bloom filter and the query continues.KeyRangeBitmaps are merged, so parquet row group and page pruning still works, also without a bloom filter and withpushdown_filters = false.lit(true), because it holds runtime state.refactor: drop the routed hash table lookup dynamic filter: removesHashTableLookupExpr,PartitionRouterand the routedCASEfallback. When a build partition is canceled, its keys are unknown, so the filter istrue.refactor: allow the hash join dynamic filter for any input partitioning: removes the partitioning checks fromallow_join_dynamic_filter_pushdown.New options:
hash_join_dynamic_filter_inlist_max_distinct_values20hash_join_bloom_filter_fpp0.01hash_join_bloom_filter_max_bytes8388608(8 MiB)hash_join_bloom_filter_max_fpp0.15All four are under
datafusion.optimizer. The existing per-partitionhash_join_inlist_pushdown_max_distinct_valueskeeps its meaning.datafusion-physical-plannow depends onparquet(forSbbf).What is the testing strategy for this PR?
bloom_filter.rs: membership, NULL keys, merge of filters of different sizes, size and fpp limits, memory reservation, multi-column and float keys, proto.shared_bounds.rs: filter shape, InList vs bloom filter at the limit, canceled partitions, memory pool fallback, merged bounds and bitmap.exec.rs: the filter is allowed for any partitioning. After the plan is dropped, the memory pool holds only the bloom filter.dynamic_row_group_pruning.slt: row groups are pruned by the merged bitmap withpushdown_filters = false.filter_pushdown.rsand several.sltfiles.Are there any user-facing changes?
EXPLAINshowsbloom_filterinstead ofhash_lookupin hash join dynamic filters.HashTableLookupExprandPartitionRouter(public, doc-hidden) are removed.Open questions
🤖 Generated with Claude Code