Enable dynamic filters for range-partitioned joins - #23854
Open
peterxcli wants to merge 25 commits into
Open
Conversation
…/hash-join-dynamic-filter-with-range-partition
saadtajwar
reviewed
Jul 24, 2026
Member
Author
|
cc @gene-bordegaray @jayshrivastava @stuhood @adriangb — kindly pinging on this one. Please take a look if you have time, thanks! |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23854 +/- ##
==========================================
- Coverage 81.01% 81.00% -0.02%
==========================================
Files 1106 1106
Lines 383974 384464 +490
Branches 383974 384464 +490
==========================================
+ Hits 311085 311433 +348
- Misses 54554 54664 +110
- Partials 18335 18367 +32 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
run benchmark tpch tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing feat/hash-join-dynamic-filter-with-range-partition (1a141f3) to 1763417 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing feat/hash-join-dynamic-filter-with-range-partition (1a141f3) to 1763417 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Member
Author
@comphead We’d need to enable range partitioning on the join inputs for the effect to show up. |
getChan
reviewed
Jul 26, 2026
getChan
left a comment
getChan
left a comment
Contributor
There was a problem hiding this comment.
make sense to me!
If it's not too much trouble, could you share some performance comparison benchmark results?
Member
Author
this still need some additional plumbing to really work, but let me try to do that and run some benchmark |
Contributor
|
I believe there is much discussion going on around the future of dynamic filters here #23814 . I am unable to give PRs full attention this week but think syncing with @jayshrivastava will be very useful |
…mic-filter-with-range-partition
peterxcli
force-pushed
the
feat/hash-join-dynamic-filter-with-range-partition
branch
from
July 31, 2026 04:29
9056b30 to
bf4e6a1
Compare
|
Hi @peterxcli, thanks for the request (#23854 (comment)). Only whitelisted users can trigger benchmarks. Allowed users: 2010YOUY01, Dandandan, Fokko, Jefffrey, Omega359, Rachelint, adriangb, alamb, asubiotto, brunal, buraksenn, cetra3, codephage2020, coderfender, comphead, erenavsarogullari, etseidl, friendlymatthew, gabotechs, geoffreyclaude, grtlr, haohuaijin, jonathanc-n, kevinjqliu, klion26, kosiew, kumarUjjawal, kunalsinghdadhwal, liamzwbao, mbutrovich, mkleen, mzabaluev, neilconway, rluvaton, sdf-jkl, timsaucer, xudong963, zhuqi-lucas. File an issue against this benchmark runner |
Contributor
|
run benchmark tpch tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing feat/hash-join-dynamic-filter-with-range-partition (7d2589f) to 39d5064 (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 feat/hash-join-dynamic-filter-with-range-partition (7d2589f) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing feat/hash-join-dynamic-filter-with-range-partition (7d2589f) to 39d5064 (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 feat/hash-join-dynamic-filter-with-range-partition (7d2589f) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Member
Author
|
Sure, I will address your review in few hours. |
Contributor
|
Thanks -- I think by early next week also fine -- @timsaucer is hoping to make a RC mid next week |
peterxcli
marked this pull request as draft
August 8, 2026 21:44
peterxcli
commented
Aug 9, 2026
Member
Author
There was a problem hiding this comment.
@gene-bordegaray thanks for the review.
There are two inline comments that I didn't resolve. I think we need more discussion around "edge case that is comparing floating point 0.0 values" especially.
|
|
||
| #[tokio::test] | ||
| async fn test_hashjoin_dynamic_filter_pushdown_range_partitioned() { | ||
| use datafusion_common::JoinType; |
Member
Author
There was a problem hiding this comment.
I followed how test_hashjoin_hash_table_pushdown_partitioned's pattern
datafusion/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
Lines 948 to 949 in 18f9db3
it also import the similar things at the top of test function.
should I also move their use to module level, or just follow their convention?
| .iter() | ||
| .zip(partition_filters) | ||
| .map(|(split_point, then_expr)| { | ||
| let when_expr = build_lexicographic_filter( |
Member
Author
There was a problem hiding this comment.
IMO, because now we're using CASE WHEN expression to do the partition routing, so I'm not sure that 1:1 replicate from
datafusion/datafusion/common/src/utils/mod.rs
Lines 119 to 146 in 18f9db3
Maybe we should just reject this case in
|
|
||
| #[tokio::test] | ||
| async fn test_hashjoin_dynamic_filter_pushdown_range_partitioned() { | ||
| use datafusion_common::JoinType; |
Contributor
There was a problem hiding this comment.
since it seems we are repeating imports, would probably be good to add these to module level, thaknk you 🙇
| ); | ||
| } | ||
|
|
||
| fn hashjoin_pushdown_scans() -> ( |
Contributor
There was a problem hiding this comment.
really great thank you ❤️
| .iter() | ||
| .zip(partition_filters) | ||
| .map(|(split_point, then_expr)| { | ||
| let when_expr = build_lexicographic_filter( |
Contributor
There was a problem hiding this comment.
Well I think we can achieve this if we redesign the statement a bit.
Right now we do:
CASE
WHEN key <range split[0] THEN F0
WHEN key <range split[1] THEN F1
...
ELSE Fn
ENDand the floating point comparisns here are cuasing us to suffer.
Range partitioning guarentees that a row will be routed to a particular partition number based on the split points and the row values itself. So we can modify this statement to reflect that rather that comparing floating points directly in the case, with someting like this:
CASE range_partition(key)
WHEN 0 THEN F0
WHEN 1 THEN F1
...
ELSE Fn
ENDWe can extract the code that determines where a row will fall form the repartition code into a comon helper:
https://github.com/peterxcli/datafusion/blob/18f9db399331ef1c2163c8d5a230b7169d073e50/datafusion/physical-plan/src/repartition/mod.rs#L974-L992
Then we can use something similar to what the Hash case is doing with HashExpr here by adding a RangeExpr that will return the partition rows belong in. Then both the repartition and this code can call it to ensure behavior is the same.
I don't believe this will have any pruning regression from the prior form since the page-stats pruner doesnt understand CASE statements at all.
For CPU overhead we now have to extract a ScalarValue per key in every row but this was something we did in the repartitioner knowing we can improve in later iterations as benchmarks showed proof of needing better perf here. We can file an issue to benchmark this and samply it to find where / how we can improve. One idea is to turn the keys into Arrow arrays and do this once per batch.
@jayshrivastava @LiaCastaneda also lmk what you guys this
Member
|
I am hoping to start the release branch on Wednesday. I see this was moved to draft. Do you think it will be ready for 55.0.0 or should we expect to ship without it? There is a lot of good discussion in this PR and I haven't followed all of it in detail, so it would be helpful to me to get a feel for how close this is to ready. |
Contributor
@timsaucer Hey tim! I think it is actually quite close. There is only one real discssuion thread goihg on here about how we can change the case statement to ensure the dyn filter handles comparing rows the same as the repartitioner here: #23854 (comment) I think this shouldnt be a large refactor so I think it should be in by Wednesday 👍 @peterxcli lmk thoughts as well 😄 |
peterxcli
marked this pull request as ready for review
August 10, 2026 15:06
Member
Author
|
Gene’s review identified that the searched dynamic-filter While investigating signed zero, found a separate pre-existing issue: Range ordering can place two HashJoin-equal keys, such as |
|
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 |
| /// This uses the same routing function as [`BatchPartitioner`], so dynamic | ||
| /// filtering and repartitioning agree for every [`ScalarValue`] comparison. | ||
| #[derive(Debug, Hash, PartialEq, Eq)] | ||
| pub struct RangeExpr { |
Contributor
There was a problem hiding this comment.
this is a grat implementation. Thank you!
this file is growing quite large. maybe we make a follow up to break some of this up a bit 👍
| expr.with_new_children(vec![Arc::clone(&remapped), Arc::clone(&remapped)])?; | ||
|
|
||
| let children = rewritten.children(); | ||
| assert_eq!(children.len(), 2); |
Contributor
There was a problem hiding this comment.
we are only checking that the children exist here, could we check the range and sort properties as well
like the sort options, split points
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
11 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
Partitioned hash joins build one dynamic filter per build partition. Existing routing uses
hash(key) % N, which cannot reproduce a Range partitioning layout.Compatible Range co-partitioned joins instead need to route probe rows using their existing ordering and split points.
What changes are included in this PR?
CASEexpression that routes probe rows to the corresponding partition filter using the Range ordering and split points.Are these changes tested?
unit test.
Are there any user-facing changes?
Yes. This PR adds
RangeExprto the physical-expression protobuf model, which adds the publicExprType::RangeExprenum variant. Downstream Rust consumers that exhaustively matchExprTypemust handle the new variant.It also enables dynamic-filter pushdown for compatible Range-partitioned joins.