jayzhan211 opened a new pull request, #25255: URL: https://github.com/apache/datafusion/pull/25255
## Which issue does this PR close? No dedicated issue yet. This is a first step toward multi-join predicate transfer (Robust Predicate Transfer style pre-filtering) on top of the existing dynamic filter machinery. Related: #18290, #19858, #7955. ## Rationale for this change For an inner or semi join, every output row has equal join keys on both sides. So a filter that only references one side's join-key columns holds for the matching rows of the other side as well. `HashJoinExec` only routed parent filters by column ownership, so such a filter reached one input and never the other. This matters most for dynamic filters, which do not exist at the logical level where equality inference already happens. The dynamic filter of a join above is a parent filter for the joins below it. When it lands on the *build*-side key of a lower join, the lower join's *probe*-side scan never saw it. The lower join's own dynamic filter usually carries similar keys, but - 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 arrives with 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, or a build-side scan that does not accept filters); - it can only be as tight as its own build input, whereas the transferred filter tightens that input first (cascading pruning, see TPC-H Q5 below). The same mechanism replaces the semi-join special case, which pushed key filters to the non-output side by column *name* and therefore silently did nothing when the two keys were named differently. ### Benchmarks 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 | base | branch | 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 base 922-1642 ms vs branch 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 | dyn filters base -> branch | output rows base -> branch | |---|---|---| | 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 | dyn filters base -> branch | output rows base -> branch | |---|---|---| | 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 base 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. ## 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`: the case the old name-based routing missed. - `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. -- 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]
