jayzhan211 opened a new pull request, #25550:
URL: https://github.com/apache/datafusion/pull/25550
## Which issue does this PR close?
Follow-up to #25255, which left outer joins for later.
## Rationale for this change
#25255 lets `HashJoinExec` rewrite a parent filter over one side's join keys
onto the other side's keys and push it to both inputs, for inner and semi
joins. There every output row has equal keys on both sides, so the copy is as
good as the original.
Left, right and mark joins also emit unmatched rows of their preserved side,
so the copy is not as good as the original. It is still useful as a
**prune-only** filter on the other side's input:
```text
FilterExec: lk = 'aa' (or a dynamic filter from a
join above)
└── HashJoinExec: join_type=Left, on=[(lk, rk)]
├── DataSourceExec left predicate=lk = 'aa' <- direct, as before
└── DataSourceExec right predicate=rk = 'aa' <- new: transferred
copy, prunes only
```
Why this is safe. A right row that fails `rk = 'aa'` can only match left
rows that fail `lk = 'aa'`: their keys are equal, also under `NullEqualsNull`
where both values are the same NULL, and a join filter only removes pairs. So
every left row that passes the filter keeps exactly the same matches. A left
row that fails it may turn from matched into NULL-extended (or mark = false),
but every row derived from it still fails the filter, and the filter stays
above the join unless the *preserved* side's scan accepted it. The copy never
adds output rows, so a `fetch` on the join or above it cannot lose rows that
pass.
Rules:
- only from the preserved side to the other side. `LEFT JOIN ... WHERE r.k
IS NULL` must not become `l.k IS NULL`, because the non-preserved key is also
NULL for unmatched rows;
- the join reports the preserved child's answer only, so a copy accepted by
the other side never removes the parent filter;
- not for `Full`, not for null-aware joins (a NULL probe key decides the
whole NOT IN result, and the copy is not true for NULL);
- not for anti joins. Pruning the other side makes an anti join emit *more*
rows. They all fail the filter, but a `fetch` between the join and the filter
can be filled by them and displace rows that pass. Example: `l.k = {0, 0, 10,
11}`, `r.k = {10, 11}`, a TopK filter `k < 2` from another union branch, `fetch
= 2` on the join. Anti joins need fetch guards on the operators that forward
filters first.
The main beneficiary is a dynamic filter from a join above that lands on the
preserved key of a left or mark join below: it now prunes the other input too.
### Benchmarks
**TL;DR:** neutral on TPC-DS SF1, TPC-H SF10 and JOB in both parquet modes;
no regression. The change applies to few plans in these suites, and where it
applies (TPC-DS Q80) it prunes 94 % of two scans for a small, consistent gain.
M4 Pro (12 cores / 24 GB), release binaries from separate target dirs, base
f7e2db3d9, 2 rounds x 3 iterations, sides alternated per round, machine
otherwise idle (1-minute load never above the core count). Geomean of per-query
ratios, per round, next to how much each binary differs from *itself* between
rounds:
| mode | suite | round 1 (base first) | round 2 (branch first) | base r2/r1
| branch r2/r1 |
|---|---|---|---|---|---|
| default | TPC-DS (99 q) | 1.026 | 0.997 | 1.028 | 0.998 |
| default | JOB (113 q) | 0.998 | 0.997 | 0.996 | 0.994 |
| default | TPC-H SF10 (22 q) | 1.007 | 1.001 | 1.001 | 0.995 |
| pushdown | TPC-DS | 0.989 | 0.997 | 0.996 | 1.004 |
| pushdown | JOB | 0.999 | 1.000 | 1.005 | 1.006 |
| pushdown | TPC-H SF10 | 0.971 | 1.017 | 1.022 | 1.069 |
Every ratio is inside the same-binary band. The 1.026 for TPC-DS round 1 is
one fast base sample (base differs from itself by 1.028; the branch is stable
and round 2 is 0.997). Row counts are identical between base and branch for all
468 query/mode pairs.
Why it is flat: of the 745 hash joins in the 99 TPC-DS plans, 711 are inner
or semi joins (covered by #25255) and only 26 are the types added here (Right
11, Left 10, LeftMark 3, RightMark 2). A `fact LEFT JOIN small_dim` is planned
as a Right join with the dimension as build side, so the transfer points at the
small table. Only Q40 and Q80 (`sales LEFT JOIN returns`) gain a filter. Q80,
pushdown mode:
| scan | dynamic filters before -> after | output rows before -> after |
|---|---|---|
| store_returns | 0 -> 1 | 287.5 K -> 16.5 K |
| catalog_returns | 0 -> 1 | 144.1 K -> 8.2 K |
| web_returns | 0 -> 1 | 71.8 K -> 71.8 K |
Q80 goes from 45.5 / 46.0 ms to 44.0 / 44.5 ms (both branch samples below
both base samples); at SF1 the pruned scans are too small for more.
Observed, not yet explained: the two scans that prune sit under
`Partitioned` joins; `web_returns` (Q80) and `catalog_returns` (Q40) sit on the
build side of `CollectLeft` joins, carry the same populated filter after
execution, and prune nothing. An unpopulated dynamic filter is `true`, so this
is a missed optimization, never a wrong result. It looks like the build-side
scan starting before the ancestor join has produced the filter; to be
reproduced in isolation and tracked separately.
## What changes are included in this PR?
- `HashJoinExec::key_transfer` replaces `supports_key_transfer` and returns
`Exact` (inner, semi), `PruneOnly { preserved_child }` (left, right, left mark,
right mark, not null-aware) or `None`.
- `gather_filters_for_pushdown`: for `PruneOnly` only the preserved-to-other
key map is used, and a side that is not preserved receives nothing but
transferred entries.
- `handle_child_pushdown_result`: for `PruneOnly` the result is the
preserved child's verdict per filter instead of `if_any`. Without a transfer
this equals the old result, since direct pushes only ever went to the preserved
side.
- `transfer_filter_across_keys` no longer rewrites column-free filters,
which the plain routing already handles.
## What is the testing strategy for this PR?
- `test_hashjoin_prune_only_transfer_differential`: four join types x both
`NullEquality` values x with and without a join filter x `k = c`, `k IS NULL`,
`NOT (k = c)`, `k = c OR k IS NULL`, over NULL, duplicate and one-sided keys.
For every combination of which scans accept filters, the rows equal those of
the plan where no scan accepts anything.
- `test_hashjoin_prune_only_transfer_keeps_parent_filter`: the copy reaches
the other scan and does not remove the `FilterExec`; the preserved scan
accepting it does.
- `test_hashjoin_prune_only_transfer_negative_cases`: non-preserved key,
non-key column, mark column, full, anti and null-aware joins transfer nothing.
- `test_hashjoin_prune_only_transfer_with_fetch`: with a join `fetch` every
row still comes from the join without a fetch.
- `test_hashjoin_dynamic_filter_prune_only_through_left_join`: an upper
join's dynamic filter prunes the lower left join's other input, checked through
scan metrics, with the NULL-extended row intact.
- `join_dynamic_filter_transfer.slt`: LEFT JOIN plan shape and results. One
existing snapshot gains the transferred predicate.
## Are there any user-facing changes?
No new configuration. `EXPLAIN` may show a filter on the non-preserved scan
of a left, right or mark join.
--
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]
