feat: transfer parent and dynamic filters across HashJoinExec equi-join keys (inner and semi joins)

[jayzhan211]

Which issue does this PR close?

No dedicated issue yet. Related: #18290, #19858, #7955 (dynamic filter pushdown follow-ups).

Rationale for this change

What "transferring a filter across join keys" means

For an inner or semi join ON a.k = b.k, every output row has a.k = b.k. So a filter that only touches a.k (say a.k IN (1, 3)) is also true of b.k for every row that can match. Today HashJoinExec routes a parent filter only to the side that owns its columns, so that filter reaches a and never b. With this PR the join also pushes b.k IN (1, 3) to b.

Static predicates already get this at the logical level (equality inference in push_down_filter). Dynamic filters do not exist there, and they are the ones that matter: the dynamic filter of a join above is a parent filter for the joins below it.

Before, with dim (small, filtered) joined to mid, and mid joined to fact:

HashJoinExec on=[(d_key, m_key)]              build: dim, probe: (mid ⋈ fact)
├── DataSourceExec dim
└── HashJoinExec on=[(m_key, f_key)]          build: mid, probe: fact
    ├── DataSourceExec mid   predicate=DynamicFilter[d_key from dim]      <- direct: m_key is a mid column
    └── DataSourceExec fact  predicate=DynamicFilter[m_key from mid]      <- only the lower join's own filter

After:

HashJoinExec on=[(d_key, m_key)]
├── DataSourceExec dim
└── HashJoinExec on=[(m_key, f_key)]
    ├── DataSourceExec mid   predicate=DynamicFilter[d_key from dim]
    └── DataSourceExec fact  predicate=DynamicFilter[m_key from mid] AND DynamicFilter[d_key from dim, rewritten over f_key]
                                                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                                                                     transferred across m_key = f_key

The transferred copy is the same DynamicFilterPhysicalExpr with its key columns remapped, so it is populated when dim finishes building and prunes fact at the scan (file / row-group / page pruning always, row-level with pushdown_filters=true). Because fact is now pruned before the lower join builds, the lower join's own filter tightens too.

Why the lower join's own filter is not enough on its own:

• it is a hash-table lookup once the lower build side exceeds hash_join_inlist_pushdown_max_size / _max_distinct_values, which cannot prune files, row groups or pages, while the transferred filter carries the small table's min/max bounds and IN list;
• it is absent when the lower join creates none (Partitioned mode without routing, preserve_file_partitions, null-aware joins with nullable build keys, a build-side scan that does not accept filters);
• it can only be as tight as its own build input, whereas the transferred filter prunes that input first (see TPC-H Q5 below, where customer had no dynamic filter at all before).

Semi-join routing bug fixed on the way

The same mechanism replaces the semi-join special case in gather_filters_for_pushdown. That code added the key's output index to the non-output side's allowed set and then let FilterRemapper::try_remap map the column to that side by name. Two failure modes followed when the keys were named differently:

• the name does not exist on the other side: the remap fails and nothing is pushed there (a missed optimisation);
• the name exists on the other side but is not the key: the filter is pushed onto that unrelated column, and because a child accepted it the FilterExec above the join is removed. That is a wrong-result bug. Example: LeftSemi on left.k = right.j where right also has a non-key column k. A parent filter k = 'x' reached the right scan as k@2 = 'x'; with this PR it is j@0 = 'x'. The RightSemi branch had the mirror-image bug.

The existing test_hashjoin_parent_filter_pushdown_semi_anti_join did not catch either, because both keys in it are named k. The transfer rewrites over the key expression on the other side, so column names no longer matter.

Benchmarks

TL;DR: JOB total 3 % faster with row-level parquet pushdown (16a 2.3x, 16c/16d 1.6x, 33a/33c 1.25x), TPC-H SF10 Q5 1.3x, everything else within noise, no reproducible regression. Details below.

M4 Pro (12 cores / 24 GB), release binaries built in separate target dirs, sides alternated per round, per-query minimum over all iterations. default = stock parquet config (dynamic filters prune files / row groups / pages only); pushdown = datafusion.execution.parquet.pushdown_filters=true.

suite	mode	geomean ratio	total time
JOB (113 q), 2 x 5 iter	default	0.997	51.07 s -> 50.95 s
JOB (113 q), 2 x 5 iter	pushdown	0.982	26.48 s -> 25.66 s
TPC-H SF10 (22 q), 2 x 3 iter	default	1.004	5.19 s -> 5.24 s (no query outside 5 %)
TPC-H SF10 (22 q), 2 x 3 iter	pushdown	0.996	6.49 s -> 6.42 s

Queries outside +/-5 % (pushdown mode, ms):

query	before	after	ratio
JOB 16a	723	321	0.44
JOB 16d	722	439	0.61
JOB 16c	719	464	0.64
JOB 33a	134	105	0.78
JOB 33c	143	116	0.81
TPC-H Q5	416	324	0.78
JOB 18a / 17e / 17f	208-329	221-349	1.06-1.07 (within run-to-run spread)
JOB 16b	874	990	1.13 (not reproducible: identical scan predicates on both binaries; interleaved re-timing gives before 922-1642 ms vs after 874-1144 ms)

Where the time goes, from per-scan EXPLAIN ANALYZE metrics in pushdown mode.

JOB 16a: the top join builds on the filtered title (68 K rows) and its dynamic filter lands on ci.movie_id, the build-side key of the join below. Transferred across that join's keys it now also reaches two scans:

scan	dynamic filters before -> after	output rows before -> after
movie_companies	1 -> 2	1.15 M -> 7.6 K
cast_info	2 -> 3	6.38 M -> 45 K

TPC-H SF10 Q5: the 5-row nation filter reaches customer through the key equivalence, which previously received no dynamic filter, and the pruning cascades down the join chain:

scan	dynamic filters before -> after	output rows before -> after
customer	0 -> 1	1.50 M -> 300 K
orders	1 -> 1	2.28 M -> 457 K
lineitem	1 -> 1	9.10 M -> 1.83 M

Cost side: the transferred copy is rewritten over the target side's key expression, so where the join key is a CAST it pays a cast per row. The bounds builder also emits duplicated bound pairs when several key pairs share one column (pre-existing, visible in the before plans too, just more often now).

What changes are included in this PR?

• HashJoinExec::gather_filters_for_pushdown: after the plain column-based routing, every parent filter whose columns are all plain Column join keys of one side is rewritten over the other side's key expressions and marked supported for that child. Inner, LeftSemi and RightSemi only: outer, anti and mark joins also emit unmatched rows, so the transferred filter would not be exact there and if_any could wrongly drop the parent filter. Outer joins would need "prune-only" semantics and are left as a follow-up.
• A DynamicFilterPhysicalExpr is rewritten through with_new_children, so the transferred copy shares the original's state and keeps tracking the build side.
• Removed the name-based semi-join routing, now covered by the general transfer. This fixes the wrong-column pushdown described above.
• A transfer only targets a side that lr_is_preserved permits, the same gate as the plain column routing (a no-op for the three permitted join types today, but it couples the two functions). Join keys are debug_asserted to have equal data types, since the planner coerces them and try_new does not check.

What is the testing strategy for this PR?

• test_hashjoin_parent_filter_transferred_across_join_keys: key names differ; key filters land on both scans, a non-key filter stays on its side.
• test_hashjoin_parent_filter_transfer_semi_join_different_key_names and its RightSemi variant: the case the old name-based routing missed (nothing reached the non-output scan).
• test_hashjoin_parent_filter_transfer_semi_join_key_name_shadowed_by_non_key: regression test for the wrong-column pushdown, LeftSemi and RightSemi.
• test_hashjoin_parent_filter_transfer_uses_first_on_pair: ON k = x AND k = y transfers over the first pair.
• test_hashjoin_parent_filter_transfer_cast_key_with_projection: a CAST key with a projection that puts the other side's key at the same output index as the cast's inner column. The rewrite must not descend into the substituted expression, or it would loop forever.
• test_hashjoin_dynamic_filter_transferred_through_nested_join: an upper join's dynamic filter reaches the lower probe scan. The lower build scan rejects filters, so the lower join's own filter still lists all four keys and the two pruned rows are attributable to the transfer alone (checked via scan metrics).
• join_dynamic_filter_transfer.slt: SQL plan shape and results.
• Two existing snapshots changed only by a build-key filter now also appearing on the probe scan. Full sqllogictest suite, physical-plan join unit tests, cargo fmt and clippy -D warnings pass.

Are there any user-facing changes?

No new configuration. EXPLAIN may now show a parent or dynamic filter on both scans of an inner or semi join where it previously appeared on one.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.22222% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.91%. Comparing base (a407990) to head (e62f53f).
⚠️ Report is 18 commits behind head on main.

Files with missing lines	Patch %	Lines
...tafusion/physical-plan/src/joins/hash_join/exec.rs	92.22%	2 Missing and 5 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25255      +/-   ##
==========================================
- Coverage   81.95%   81.91%   -0.04%     
==========================================
  Files        1133     1134       +1     
  Lines      423828   425683    +1855     
  Branches   423828   425683    +1855     
==========================================
+ Hits       347344   348694    +1350     
- Misses      55890    56304     +414     
- Partials    20594    20685      +91     

☔ 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.

[adriangb]

I reviewed the code, the tests and the benchmark data. I also ran a differential test and mutation tests against the new code. This review uses Simplified Technical English.

Summary

The transfer logic is correct. I did not find a regression.

• For Inner, LeftSemi and RightSemi joins, each output row has equal keys on the two sides. Thus a filter over one key gives the same result over the other key. The if_any propagation stays exact, also when only the transferred side accepts the filter.
• with_new_children keeps the original children and only replaces remapped_children. Thus the transfer composes with a projection remap that occurred before it.
• HashExpr, HashTableLookupExpr and RangeExpr build again through with_new_children. Thus the CASE hash(keys) % n routing filter stays correct on the other side.
• The proto encoder accepts remapped children that are not a Column.
• The optimizer removes volatile predicates before gather_filters_for_pushdown runs. Thus the transfer cannot copy a volatile predicate.

Tests that I ran

• cargo fmt, cargo clippy -D warnings, the filter_pushdown integration tests, the hash_join unit tests, the proto round-trip tests and the full sqllogictest suite pass.
• A differential harness ran 67 curated queries and 300 random queries in 12 configurations. The configurations toggle dynamic filters, parquet row-level pushdown, CollectLeft and forced Partitioned mode, bounds-only filters, 1 and 4 partitions, and the logical push_down_filter rule. The results were the same in all configurations.
• The queries include CAST keys, IS NOT DISTINCT FROM, NULL keys, empty build sides, outer, anti and mark joins, projections that remove or duplicate keys, four-level nesting, recursive CTEs and TopK filters over joins.

Findings

1. This PR corrects a wrong-result bug on main, but the description does not say so. The removed semi-join code pushed a key filter to the other side by column name. See the inline comment on the semi-join test. Please add a regression test and a note in the description.
2. Mutation tests show branches that no test covers: the Jump recursion control, the RightSemi branch, the PushedDown::Yes guard, the any_column condition, the JoinSide::None arm and the first-pair tie-break. See the inline comments. The Jump case is the important one, because it prevents an infinite loop.
3. lr_is_preserved does not gate the transfer. See the inline comment.

Related bug that this PR does not cause

FilterRemapper::try_remap maps a column to the first field with the same name. When a child schema has two fields with the same name, a filter over the second field moves to the first field. This gives wrong results with the default configuration. Example: a TopK filter over a non-key column with a duplicated name, above two joins where the planner swaps the lower join. This PR removes one caller of that remap and does not make the bug worse. This bug needs a separate issue with a reproducer.

[adriangb]

transfer_key_filters writes into the child descriptions after the lr_is_preserved gate. It also overwrites all_unsupported entries.

For the three permitted join types, both sides are preserved. Thus the gate has no effect today. If a future change adds a join type to supports_key_transfer that is not preserved on one side, the transfer pushes filters to that side without a warning. A mutant that returns (true, false) for semi joins passes all tests.

Please transfer only into a side that lr_is_preserved permits, or add an assertion that couples the two functions.

[adriangb]

This guard cannot trigger. to_right contains only left-side output indices, and to_left contains only right-side output indices. A filter is directly pushable to a child only when all its columns are on that side. Thus a filter cannot be both directly pushable and transferable to the same child.

A mutant that removes the guard passes all tests. Please remove the guard, or add a comment that says it is a safety check.

[adriangb]

The any_column condition is redundant. try_remap already marks a filter with no columns as supported for both children, as the comment above says.

A mutant that removes this condition passes all tests.

[adriangb]

This comment describes a rule that the code does not enforce. See my comment on the transfer_key_filters calls above.

[jayzhan211]

Thanks @adriangb for the detailed review and helpful suggestions! I've addressed everything.

[adriangb]

Thanks @jayzhan211 ! I'll let @kosiew review this but from my perspective this is g2g. Nice change!