adriangb opened a new pull request, #24235:
URL: https://github.com/apache/datafusion/pull/24235
## Which issue does this PR close?
- None. This is a prototype opened as a **draft, for evaluation only** — I
want the project's benchmark fleet to judge it, not to merge it as-is. See
"Honest summary" below: on TPC-H the bounds split does not pay for itself, and
I would rather have that measured on neutral hardware than argued about.
## Rationale for this change
A dynamic filter produced by a **hash-partitioned** join currently reaches
the scan as a single opaque conjunct:
```
DynamicFilter[ CASE hash_repartition % n
WHEN 0 THEN bounds_0 AND membership_0
WHEN 1 THEN bounds_1 AND membership_1
... END ]
```
Everything the join knows is sealed inside one expression.
`split_conjunction` splits only on a top-level `BinaryExpr(And)` and does not
descend into a `DynamicFilterPhysicalExpr`, so `row_filter.rs` builds exactly
one `ArrowPredicate`: every probe row must have its routing hash computed
before any row can be rejected, and row-group pruning can make nothing of a
`CASE`.
This PR pushes two filters instead:
```
DynamicFilter[ merged bounds ] AND DynamicFilter[ membership ]
```
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 accumulating `RowSelection` —
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 **`InList` collapse** (third
commit). When every non-empty build partition pushes an `InList`, the routing
`CASE` is 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 the `CASE` does. That is an equality, not a relaxation, and it removes the
routing hash from the probe path entirely, leaving an `InListExpr` that
`datafusion/pruning` actually understands. That is where the measured wins come
from.
The catch is that it fires rarely: `hash_join_inlist_pushdown_max_size` is
128 KiB with a 150-distinct-value cap, **per partition**, so most joins never
produce an all-`InList` build.
### Measurements
Re-measured from scratch, because an earlier round of numbers was not
trustworthy. Protocol:
- **Both binaries built from the same worktree and the same target dir**
(build change → copy binary aside → revert to base → rebuild). An earlier
comparison across two worktrees at different upstream commits attributed
hundreds of unrelated commits to this change; its tell was a 6.7% delta on a
control query containing no joins.
- **Counterbalanced ordering within each round** — (branch, base) on odd
rounds, (base, branch) on even — reported separately as well as pooled.
- **Controls**: TPC-H q1 and q6 contain no joins, so this PR cannot affect
them. Their delta *is* the noise floor. Nothing smaller than it is reported as
a result.
- 12 rounds × `--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. `p` is an exact two-sided sign test over the 12 paired rounds.
**`pushdown_filters = false`** — control noise floor **5.09%**
| query | paired Δ% | branch-first Δ% | base-first Δ% | rounds branch faster
| sign p |
|---|---|---|---|---|---|
| q1 *(control)* | +5.09 | +10.83 | −0.96 | 5/12 | 0.774 |
| q6 *(control)* | +1.10 | −9.51 | +2.69 | 5/12 | 0.774 |
| q3 | +2.08 | +0.26 | +4.06 | 5/12 | 0.774 |
| q5 | −4.28 | −4.97 | −2.02 | 7/12 | 0.774 |
| q9 | +4.74 | +7.09 | +4.74 | 1/12 | 0.006 |
| q17 | +8.33 | +10.50 | +6.24 | 2/12 | 0.039 |
| q18 | +1.35 | +2.26 | +1.35 | 5/12 | 0.774 |
**`pushdown_filters = true`** — control noise floor **1.99%**
| query | paired Δ% | branch-first Δ% | base-first Δ% | rounds branch faster
| sign p |
|---|---|---|---|---|---|
| q1 *(control)* | +1.99 | −1.47 | +7.16 | 4/12 | 0.388 |
| q6 *(control)* | −0.38 | −8.19 | +3.78 | 6/12 | 1.000 |
| q3 | −0.72 | −1.19 | +1.58 | 7/12 | 0.774 |
| q5 | −1.14 | −5.66 | +7.93 | 6/12 | 1.000 |
| q9 | +5.34 | +10.16 | +2.63 | 3/12 | 0.146 |
| q17 | **−33.76** | −34.61 | −33.08 | 12/12 | 0.000 |
| q18 | −11.44 | −11.14 | −12.55 | 9/12 | 0.146 |
Reading these:
- **The previously claimed "~4% cost at `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.
- **q17 −33.8% at `pushdown_filters=true` is 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 the
`InList` collapse, not the bounds split.
- **The residual cost instead shows up at `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.
- **q17 flips from +8.3% at `pushdown_filters=false` to −33.8% at
`pushdown_filters=true`.** That is the expected shape rather than a
contradiction: the `InList` collapse only pays when the dynamic filter is
actually evaluated per row inside the scan, which is precisely what
`pushdown_filters=true` turns 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 second `DynamicFilter` are
constructed and charged either way. So the same change is a small cost in one
mode and a large win in the other.
- q1's control row at `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:
1. **`refactor: let a hash join carry more than one self filter`** — routes
`handle_child_pushdown_result` and `dynamic_expressions_produced` through one
`HashJoinExecDynamicFilter::produced_expressions` helper and pops the driven
filter off the pushed list instead of indexing position zero. No behaviour
change; exactly one self filter is still pushed.
2. **`Split the partitioned hash join dynamic filter into bounds AND
membership`** — adds `bounds_union.rs`, which computes the set-theoretic union
of the per-partition ranges, and pushes it as a second, routing-free
`DynamicFilter` wrapper. 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 the
`CASE` exactly 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.
`CollectLeft` joins have no routing to hoist and are untouched.
3. **`Collapse an all-InList partitioned membership check into one InList`**
— when every non-empty build partition pushed an `InList`, replace the routing
`CASE` with a single `InListExpr` over 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 routed `CASE` (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 the `CASE`.
## Are these changes tested?
Yes.
- `bounds_union.rs` ships unit tests for the union arithmetic, including the
multi-column product, degenerate merges, and the relaxation estimate.
- Existing hash-join and dynamic-filter coverage passes: `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.
- Three `.slt` files have updated `EXPLAIN` output, showing the intended
`DynamicFilter [ 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 --all` clean; `cargo clippy --profile ci --all-targets
--features
avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption
--workspace -- -D warnings` clean.
## Are there any user-facing changes?
No API changes and no changes to query results.
The one visible difference is in `EXPLAIN` output: the probe-side scan of a
partitioned hash join now shows two `DynamicFilter` conjuncts 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) and
`hash_join_inlist_pushdown_max_distinct_values` (150), both per-partition,
continue to govern whether the `InList` path is reachable at all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
