feat: prune-only transfer of parent and dynamic filters across HashJoinExec keys (left, right and mark joins)

[jayzhan211]

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:

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 f7e2db3, 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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 97.05882% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 82.41%. Comparing base (f7e2db3) to head (30a9139).
⚠️ Report is 8 commits behind head on main.

Files with missing lines	Patch %	Lines
...tafusion/physical-plan/src/joins/hash_join/exec.rs	97.05%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25550      +/-   ##
==========================================
+ Coverage   82.39%   82.41%   +0.02%     
==========================================
  Files        1138     1138              
  Lines      434592   435317     +725     
  Branches   434592   435317     +725     
==========================================
+ Hits       358073   358761     +688     
+ Misses      54853    54844       -9     
- Partials    21666    21712      +46     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kosiew]

@jayzhan211,

Thanks for working on this. The prune-only distinction looks like a useful direction, and the added differential, NULL, join-filter, dynamic-filter, and fetch coverage is helpful.

I found one correctness issue with outer joins when the join has a fetch. Pruning the non-preserved side can change which rows reach the fetch, which can then change the rows seen by the parent filter. I think we need to disable this transfer for Left and Right joins when fetch is set. Mark joins do not have the same multiplicity issue, so they can still support prune-only transfer with fetch.

I left the details and a concrete example inline.

[kosiew]

I think we need to guard PruneOnly for outer joins when this join has fetch. Pruning matching rows from the non-preserved side can remove rows that would otherwise fill the fetch, which can expose a later row that passes the parent filter and change the query result.

For example, consider left rows (bb, aa), right rows (bb, bb, aa), a parent filter lk = 'aa', and fetch = 2. Without transferring rk = 'aa', the two bb matches can fill the fetch, so the parent filter returns no rows. With the transferred filter, those right-side bb rows are pruned. The aa match can then reach the fetch, so the parent filter can return it.

Could we avoid prune-only transfer for Left and Right when self.fetch.is_some()? Mark joins should still be safe because they do not have this match-multiplicity issue. It would also be good to update the fetch test to compare supported and unsupported plans using the same fetch, with a failing-side key ordered first so this case is exercised directly.

[kosiew]

Once the outer-join fetch case is guarded, could we also add LeftMark and RightMark to a same-fetch differential test? Mark joins keep one output row per preserved-side input row, so this would document why prune-only transfer is still safe for mark joins when fetch is present.