adriangb opened a new pull request, #25713: URL: https://github.com/apache/datafusion/pull/25713
## Which issue does this PR close? - Part of #22883 (experimental "optional filters" stack, design notes: https://claude.ai/artifact/SSz7t6hPyhFWp1MDPecVqt). Builds on the prototype #24235. - **Depends on #25673, #25681, #25674, #25682 and #25677.** Review only the top two commits (the change and the benchmark). ## Rationale for this change A partitioned hash join pushes one dynamic filter into the probe side scan: ``` DynamicFilter [ CASE hash_repartition % N WHEN i THEN bounds_i AND membership_i ... END ] ``` Statistics pruning cannot use a `CASE` expression. Thus in partitioned mode the scan cannot use the join key bounds to skip row groups, even when the build side matches only a small, ordered slice of the probe key (for example one day of time-ordered events). The whole probe table is read. Also, the bounds and the membership check have very different cost and selectivity, but a consumer can only keep or pause the combined filter. Example (`hj_ordered_subset` Q01: `events` has 20M rows sorted by `event_id` in 200 row groups; the build side matches one day, 1% of the key range; partitioned join): | | Probe row groups read | Bytes scanned | Join probe rows | Time (min) | |---|---|---|---|---| | before | 200 of 200 | 134 MB | 15.3 M | 257–271 ms | | after | 2 of 200 | 1.45 MB | 200 K | 19–20 ms | ## What changes are included in this PR? `HashJoinExec` pushes two optional dynamic filters instead of one: ``` Optional(DynamicFilter [ event_id >= 8400000 AND event_id <= 8599999 ]) -- bounds AND Optional(DynamicFilter [ CASE hash_repartition % 12 WHEN 0 THEN hash_lookup ... END ]) -- membership ``` - **Bounds:** the union of the bounds of all build partitions, one range per key column. Statistics pruning can use it in all join modes. For range partitioning, the disjoint ranges of the partitions are kept (up to 8 per column). - **Membership:** the routed `CASE` without the per-partition bounds (they never reject a row that the membership check of that partition accepts), or the collapsed `IN` list from #25677. - **CollectLeft** is split the same way, for consistency. - Each filter has its own expression id, gets `update()` / `mark_complete()`, is transferred across join keys as an optional view, and round trips through proto. A proto plan without the new field decodes to the old single filter. - Fallbacks: with a canceled partition or no usable bounds, the bounds filter stays `true` and the bounds stay in the `CASE`. An empty build side sets both filters to `false`. The null-equal / null-aware `IS NULL OR` escape wraps each filter. - Because each part is a separate optional filter, the Parquet gate (#25682) measures and pauses each part on its own. On TPC-H Q3 the bounds reject 0.01% of rows and are paused, while the membership check (95.5% rejected) stays on. New benchmark suite `hj_ordered_subset` (`./bench.sh run hj_ordered_subset`): time-ordered probe table, build side matching 1 day, 10 days, or a scattered 1% of keys (control), in partitioned and CollectLeft mode. ## What is the testing strategy for this PR? - 12 new unit tests for the split filters (canceled and empty partitions, range partitioning, null-equal, bounds only, `mark_complete` on both) and 11 for the bounds union. - The proto round trip checks that both filters stay linked to their probe side copies. - `hash_join` tests (also with `force_hash_collisions`), `physical-plan` lib tests, `filter_pushdown`, proto and `parquet_integration` tests pass. Snapshot diffs change only the dynamic filter text and derived pruning predicates. - The benchmark suite checks the join mode with `expect_plan` and asserts that the join count equals the build count. Results are identical before and after in all combinations. ### Benchmark `hj_ordered_subset`, `optional_filter_mode = adaptive`, min time in ms, two runs (A: n=15, B: n=24), base = the same stack without this PR. The machine was busy (load average 40–58 on 12 cores): a no-join control query varied up to ±25%, so only large differences are significant. | pushdown_filters | Join mode | Query | Base A / B | This PR A / B | |---|---|---|---|---| | true | Partitioned | Q01 ordered, 1 day | 271 / 257 | **19 / 20** | | true | Partitioned | Q02 ordered, 10 days | 384 / 332 | **188 / 134** | | true | Partitioned | Q03 scattered (control) | 298 / 273 | 244 / 238 | | true | CollectLeft | Q04 / Q05 / Q06 | 13 / 102 / 253 | 17 / 140 / 229 (noise) | | false | Partitioned | Q01 ordered, 1 day | 177 / 193 | **15 / 13** | | false | Partitioned | Q02 ordered, 10 days | 263 / 355 | **102 / 170** | | false | Partitioned | Q03 scattered (control) | 254 / 270 | 289 / 246 | | false | CollectLeft | Q04 / Q05 / Q06 | 12 / 106 / 200 | 14 / 132 / 257 (noise) | The win does not depend on `pushdown_filters`: it comes from statistics pruning. CollectLeft already pruned before this PR (its filter is a plain conjunction), so it does not change. On TPC-H and TPC-DS SF1 this PR is neutral within noise. Known limitations: - A paused optional filter is still a separate row filter stage in the scan, so splitting adds a small cost when the bounds are useless (for example `predicate_cache_records` doubles). A follow-up can drop paused filters from the `RowFilter` at the next row group. - In the scattered CollectLeft control with pushdown on, the gate also paused the membership part for part of the scan (the join probed 447K rows instead of 226K). The time effect was within noise, but the gate may pause too early there. - Q01 needs the mode forced to Partitioned (`hash_join_single_partition_threshold[_rows] = 0`); with the defaults the 1-day build side plans as CollectLeft. Numbers should be confirmed on a quiet machine or the benchmark bot. ## Are there any user-facing changes? The text of hash join dynamic filters in `EXPLAIN` changes (two filters instead of one). New proto field for the second filter (backward compatible for decoding). New benchmark suite. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
