Split the partitioned hash join dynamic filter into bounds AND membership - #24235
Draft
adriangb wants to merge 3 commits into
Draft
Split the partitioned hash join dynamic filter into bounds AND membership#24235adriangb wants to merge 3 commits into
adriangb wants to merge 3 commits into
Conversation
`handle_child_pushdown_result` reached for the single self filter it had pushed and `dynamic_expressions_produced` cloned the single field behind it. Route both through one `HashJoinExecDynamicFilter::produced_expressions` helper and pop the driven filter off the pushed list instead of indexing position zero, so a second pushed filter is a matter of adding a field rather than of rewriting the plumbing. No behaviour change: exactly one self filter is still pushed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…ship
A partitioned join used to push one `DynamicFilter[CASE hash % n WHEN i THEN
(bounds_i AND membership_i) ... END]`. Everything the join knows was sealed
inside a single opaque conjunct: `split_conjunction` splits only on a
top-level `BinaryExpr(And)` and does not descend into a
`DynamicFilterPhysicalExpr`, so the Parquet reader saw one `ArrowPredicate`
that had to compute the routing hash for every probe row before it could
reject any of them, and row-group pruning could make nothing of a `CASE`.
Push two filters instead:
DynamicFilter[ merged bounds ] AND DynamicFilter[ routing CASE ]
Two *wrappers* is the load-bearing part, not two expressions inside one
wrapper. Split in two, `row_filter.rs` turns each into its own
`ArrowPredicate` and arrow-rs applies them in sequence against an
accumulating `RowSelection`, building a fresh array reader per predicate over
only the still-selected rows. The cheap vectorized range check therefore runs
first and the routing hash and hash-table lookup only see the survivors. The
range half is also plain enough for `PruningPredicate` to use against
row-group statistics, which a `CASE` never was.
The merged bounds are the union of the per-partition ranges (see
`bounds_union`), which is a relaxation: it admits keys that route to a
partition that does not hold them. That is sound because the membership half
behind it is exact. The bounds stay inside the `CASE` — exactly as before —
whenever the union cannot describe the build side (a canceled partition), the
merge is degenerate, or the second filter did not survive pushdown, so no
plan loses selectivity it used to have. `CollectLeft` joins have no routing to
hoist out of and are left untouched.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
When every non-empty build partition pushes an `InList`, the routing `CASE` is redundant. Routing is a deterministic function of the key columns, so every build row holding key `K` lands in the same partition a probe row holding `K` routes to; testing `K` against the union of all the lists therefore accepts and rejects *precisely* what the `CASE` does. This is an equality, not the relaxation the bounds merge is. Collapsing drops the routing hash from the probe path entirely and leaves an `InListExpr`, which unlike a `CASE` is a shape `datafusion/pruning` understands (up to `max_in_list_size`). The union is capped at 1 MiB: each partition's list is independently limited by `hash_join_inlist_pushdown_max_size`, so the concatenation grows with the partition count, and past some size the routed `CASE` — where a probe row only ever probes one list — is the cheaper shape. Partitions that push a hash map, and builds where a canceled partition makes the union incomplete, keep the `CASE`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
github-actions
Bot
added
sqllogictest
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing prototype/split-dynamic-filter (8768c31) to 33ad1cc (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
Contributor
Author
|
show benchmark queue |
|
Hi @adriangb, you asked to view the benchmark queue (#24235 (comment)).
File an issue against this benchmark runner |
|
Benchmark for this request failed. Run configurationrun benchmark tpchLast 20 lines of output: Click to expandFile an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing prototype/split-dynamic-filter (8768c31) to 33ad1cc (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 |
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
2 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
A dynamic filter produced by a hash-partitioned join currently reaches the scan as a single opaque conjunct:
Everything the join knows is sealed inside one expression.
split_conjunctionsplits only on a top-levelBinaryExpr(And)and does not descend into aDynamicFilterPhysicalExpr, sorow_filter.rsbuilds exactly oneArrowPredicate: every probe row must have its routing hash computed before any row can be rejected, and row-group pruning can make nothing of aCASE.This PR pushes two filters instead:
Two wrappers, not two expressions inside one wrapper, is the load-bearing detail. Split in two, each becomes its own
ArrowPredicate, and arrow-rs applies them in sequence against an accumulatingRowSelection— building a fresh array reader per predicate over only the still-selected rows. The intent is that the cheap vectorized range check runs first and the routing hash plus hash-table lookup only see the survivors.Honest summary: on TPC-H, the bounds half does not pay for itself
The idea above is sound but the premise fails on this workload, for a structural reason worth stating plainly. Hash routing scatters keys uniformly across partitions, so every partition's key range converges on the global domain. The union of those ranges is then very close to the full column domain. TPC-H q3's merged bound is
l_orderkey ∈ [96, 5999975]against a ~6M-row domain: it rejects roughly 0.002% of rows. The cheap first predicate compacts nothing, so all it can do is add cost.The part that carries real value is the
InListcollapse (third commit). When every non-empty build partition pushes anInList, the routingCASEis redundant — routing is a deterministic function of the key columns, so testing a key against the union of the lists accepts and rejects precisely what theCASEdoes. That is an equality, not a relaxation, and it removes the routing hash from the probe path entirely, leaving anInListExprthatdatafusion/pruningactually understands. That is where the measured wins come from.The catch is that it fires rarely:
hash_join_inlist_pushdown_max_sizeis 128 KiB with a 150-distinct-value cap, per partition, so most joins never produce an all-InListbuild.Measurements
Re-measured from scratch, because an earlier round of numbers was not trustworthy. Protocol:
--iterations 5, TPC-H SF=1 parquet. The machine was not idle and was not quiesced.Primary statistic is the paired per-round delta (median over 12 rounds of
branch/base − 1); pairing cancels round-level machine drift, which matters because the unpaired cross-round spread exceeds 100% on some queries. Positive = branch slower.pis an exact two-sided sign test over the 12 paired rounds.pushdown_filters = false— control noise floor 5.09%pushdown_filters = true— control noise floor 1.99%Reading these:
pushdown_filters=true" did NOT survive correct measurement. It is not a smaller cost than reported — it is not there. On q3 the paired delta is −0.72%, i.e. the sign flipped, with 7 of 12 rounds actually favouring the branch (p = 0.77); on q5 it is −1.14%, 6/12, p = 1.00. Both sit inside a 1.99% control floor, so neither direction is resolvable. That earlier ~4% figure was an artifact of the confounded protocol described above (chiefly the cross-worktree build), and it should not be carried forward.pushdown_filters=trueis unambiguous: 12/12 rounds, both orderings agree to within 1.5 points, far outside any floor. q18 −11.4% is consistent in direction across both orderings (9/12 rounds). Both are theInListcollapse, not the bounds split.pushdown_filters=false, on q9 (+4.7%, only 1 of 12 rounds faster, p = 0.006) and q17 (+8.3%, 2/12, p = 0.039). To be explicit about how much weight these carry: the control floor in this mode is 5.09%, so both rest on the paired sign test — the consistency of the direction across rounds and across both orderings — rather than on their magnitude. q9's +4.7% is below that floor outright; what makes it worth reporting is that the branch lost 11 of 12 paired rounds, not that the number is large. Read them as "there is probably a small real cost here", not as calibrated cost estimates.pushdown_filters=falseto −33.8% atpushdown_filters=true. That is the expected shape rather than a contradiction: theInListcollapse only pays when the dynamic filter is actually evaluated per row inside the scan, which is precisely whatpushdown_filters=trueturns on. With pushdown off, the filter never reaches the row-level evaluation path where removing the routing hash would help — but the extra wrapper, the bounds union, and the secondDynamicFilterare constructed and charged either way. So the same change is a small cost in one mode and a large win in the other.pushdown_filters=true(+1.99% pooled, but −10.18% branch-first vs +10.02% base-first in the unpaired view) is a good advertisement for reporting orderings separately — a query this change provably cannot touch can still show a double-digit unpaired swing.The lever this branch did not pull
The obvious next move, which is not attempted here: gate the bounds conjunct on a degeneracy check using probe-side column statistics from
right_child.partition_statistics(). When the merged union covers (or nearly covers) the probe column's own min/max domain, the conjunct is known to be useless before a single row is read, and it should simply not be pushed. That is the most likely way to turn the bounds half from a cost into a neutral — it keeps the win in the cases where per-partition ranges genuinely are narrow (correlated or pre-clustered build sides) and stops paying for it on uniformly hashed keys like TPC-H's. I would rather land that gate than tune anything here to make the current numbers look better.What changes are included in this PR?
Three commits, each building green on its own:
refactor: let a hash join carry more than one self filter— routeshandle_child_pushdown_resultanddynamic_expressions_producedthrough oneHashJoinExecDynamicFilter::produced_expressionshelper and pops the driven filter off the pushed list instead of indexing position zero. No behaviour change; exactly one self filter is still pushed.Split the partitioned hash join dynamic filter into bounds AND membership— addsbounds_union.rs, which computes the set-theoretic union of the per-partition ranges, and pushes it as a second, routing-freeDynamicFilterwrapper. The union is a relaxation (it admits keys that route to a partition not holding them), which is sound because the membership half behind it is exact. Multi-column keys are merged per column and emitted as a product of ranges, a superset of the true union. The bounds stay inside theCASEexactly as before whenever the union cannot describe the build side (a cancelled partition), the merge is degenerate, or the second filter did not survive pushdown — so no plan loses selectivity it previously had.CollectLeftjoins have no routing to hoist and are untouched.Collapse an all-InList partitioned membership check into one InList— when every non-empty build partition pushed anInList, replace the routingCASEwith a singleInListExprover the union. Capped at 1 MiB, since each partition's list is independently limited and the concatenation grows with partition count; past some size the routedCASE(where a probe row only ever probes one list) is the cheaper shape. Partitions pushing a hash map, and builds where a cancelled partition makes the union incomplete, keep theCASE.Are these changes tested?
Yes.
bounds_union.rsships unit tests for the union arithmetic, including the multi-column product, degenerate merges, and the relaxation estimate.cargo test --profile ci -p datafusion-physical-plan --lib— 1675 passed, 0 failed.cargo test --profile ci -p datafusion-sqllogictest --test sqllogictests— 502/502 files pass..sltfiles have updatedEXPLAINoutput, showing the intendedDynamicFilter [ empty ] AND DynamicFilter [ empty ]shape on the probe-side scan:preserve_file_partitioning.slt,push_down_filter_parquet.slt,statistics_registry.slt. These are plan-shape changes only; every result set is unchanged.cargo fmt --allclean;cargo clippy --profile ci --all-targets --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption --workspace -- -D warningsclean.Are there any user-facing changes?
No API changes and no changes to query results.
The one visible difference is in
EXPLAINoutput: the probe-side scan of a partitioned hash join now shows twoDynamicFilterconjuncts where it previously showed one. Anything asserting on that exact string will need updating.No new configuration options. The existing
hash_join_inlist_pushdown_max_size(128 KiB) andhash_join_inlist_pushdown_max_distinct_values(150), both per-partition, continue to govern whether theInListpath is reachable at all.