feat: support pushdown-aware dynamic filter#23701
Draft
zhuqi-lucas wants to merge 12 commits into
Draft
Conversation
github-actions
Bot
added
documentation
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (82ea18c) to 67947b6 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (82ea18c) to 67947b6 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (82ea18c) to 67947b6 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (5a8ca02) to 67947b6 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (5a8ca02) to 67947b6 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (5a8ca02) to 67947b6 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (836f46e) to 67947b6 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (836f46e) to 67947b6 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (836f46e) to 67947b6 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
zhuqi-lucas
changed the title
…d conjuncts post-scan
Rebased onto main. The original commit 1 of this PR ("extract
DecoderProjection from build_stream") landed independently on main as
current `decoder_projection` / single-decoder + `rg_plan` model.
Two changes, both applied inside the parquet scan so the parent
`FilterExec` can be removed unconditionally for pushable filters:
1. Never drop conjuncts the `RowFilter` cannot place.
`build_row_filter` previously `.flatten()`-ed away conjuncts that
`FilterCandidateBuilder::build` rejected (whole-struct references,
per-file physical-schema mismatches) and swallowed whole-build
errors. By the time it runs, `try_pushdown_filters` has already
removed the `FilterExec`, so those conjuncts were applied nowhere —
wrong results. `build_row_filter` now returns
`(Option<RowFilter>, Vec<rejected>)`, `RowFilterGenerator` exposes
`rejected_conjuncts()`, and a whole-file build error routes every
conjunct to the rejected list rather than relaxing the predicate.
2. Always accept pushable filters and run the remainder post-scan.
`try_pushdown_filters` reports each pushable filter as accepted so the
`FilterExec` is always removed; the scan owns the predicate. The
opener routes conjuncts to two places, applying every one:
- pushdown_filters=true -> row-filterable conjuncts via the parquet
`RowFilter`; rejected conjuncts via the in-scan post-scan filter.
- pushdown_filters=false -> the whole predicate runs as a post-scan
filter on decoded batches (behaviorally identical to `FilterExec`).
Implementation:
- `DecoderProjection` (main's `decoder_projection` module) grows a
`post_scan_conjuncts` parameter: it widens the decoder mask over
(user projection ∪ post-scan filter columns), rebases the conjuncts
onto the stream schema, and returns a `PostScanFilter` applied to
every decoded batch with SQL `WHERE` semantics. Virtual-column
conjuncts are stripped from the read-plan mask (they aren't file
columns) but kept in the post-scan predicate, which sees the
reader-appended virtual columns.
- `PushDecoderStreamState` applies the post-scan filter in the decoded-
batch arm, skips empty batches, and re-introduces a stream-level
`remaining_limit` (main enforces LIMIT decoder-locally, which is
unsafe once a post-scan filter can reject rows). The opener routes the
limit to `remaining_limit` iff a post-scan filter is present.
- New `post_scan_rows_pruned` / `post_scan_rows_matched` counters and
`post_scan_filter_eval_time` on `ParquetFileMetrics`.
Tests:
- `build_row_filter_surfaces_rejected_struct_conjunct` (row_filter.rs)
asserts the rejected struct conjunct is returned, not dropped.
- `rejected_struct_conjunct_runs_post_scan_not_dropped` (opener) is
end-to-end: `s IS NOT NULL` over a struct column with pushdown on
returns 2 (was 3 before the fix).
- Parquet `.slt` files regenerated: `FilterExec` above `DataSourceExec`
gone, predicate on the scan, `post_scan_rows_*` metrics on EXPLAIN
ANALYZE. Opener / core insta / page_pruning assertions updated for the
now-applied predicate.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UYxLsgHvtixT2Y9JYHaEsU
Tactical fix on top of apache#22384 to address the benchmark regressions that paper reported (TPCH SF1: +27% total, Q17 2.09x slower, Q3/5/7/8/9/12/ 13/14/18/20 all 1.24-1.67x slower). Approach was suggested by @adriangb in the apache#23420 discussion: "splitting out the min/max range dynamic filters that HashJoinExec pushes down from the hash table ones and then we could turn off the hash table ones by default". Why: HashJoin's build-side dynamic filter today publishes a combined `bounds AND membership` expression to the probe scan. - Bounds (`col >= min AND col <= max`) is 2 comparisons per row (~2ns) and drives the RG-level statistics pruning that is by far the largest contribution. - Membership (`InListExpr` over the build keys, or a hash-table lookup for large builds) is a per-row hash-set / hash-table probe (~50-100ns). apache#22384's contract change (`try_pushdown_filters` always accepts pushable filters, PostScanFilter picks up whatever the RowFilter cannot place) means the combined expression now runs on every scanned batch even with `pushdown_filters = false`, where previously it was silently propagated through source.predicate but never row-evaluated. On multi-join queries with high match rate, the membership check pays the hash cost twice (once in the scan, once inside HashJoin) with no selectivity win — that's exactly the "not earning their keep" case @adriangb described. What: a new config knob `datafusion.optimizer.enable_hash_join_dynamic_membership_filter` (default `false`) gates the membership creation. When off, `SharedBuildAccumulator` skips `create_membership_predicate` in both the CollectLeft and Partitioned finalize paths and publishes only the bounds portion. RG pruning is unaffected. Highly-selective joins with big build sides that used to see 2-3x wins from membership pruning can restore the historical behavior by flipping the knob to `true`. Tests: two new unit tests in `shared_bounds.rs`: - `collect_left_with_gate_off_publishes_bounds_only` drives an accumulator with the gate off, asserts the published expression contains no `InListExpr` and its top op is `AND` (bounds). - `collect_left_with_gate_on_publishes_bounds_and_membership` the inverse, guards against accidentally regressing the wiring. All 398 pre-existing `joins::hash_join` tests still pass. All 1559 `datafusion-physical-plan` lib tests pass. Full `information_schema.slt` passes with the new option listed. Draft while we run benchmarks to quantify how much of the apache#22384 regression this closes. Companion to apache#22384 (adriangb's foundation), follow-up to apache#23532 (DynamicFilter cache — Layer 1 of the regression fix).
…filter The pushdown=false path in the parquet opener split the whole predicate into 'post_scan_conjuncts' — a per-batch FilterExec-equivalent — which included any dynamic filter conjuncts (HashJoin bounds, TopK threshold, aggregate dynamic filter). For join-heavy TPC-H / TPC-DS this dominates cost: HashJoin's Partitioned- mode dynamic filter is a 'CASE hash(col) % N WHEN pid THEN bounds ELSE lit(false) END' — per-row hash + modulo + CASE branch — and it prunes almost nothing on high-match-rate joins where the downstream hash lookup would eliminate the same rows anyway. Local TPC-H SF1 Q9 profile showed 1.1% self-time in 'expressions::case::PartialResultIndex::merge_n' and 1.3% in 'arrow_select::filter::filter_native' on the PR, both at 0% on main — driving Q9 from 40ms → 80ms (2.09x on CI, 1.79x locally). This commit filters DynamicFilterPhysicalExpr-containing conjuncts out of 'post_scan_conjuncts'. Effects: - RowGroupPruner (added by apache#22450) still sees the full predicate via prepared.predicate, so RG-level dynamic pruning continues to fire on bounds/threshold updates. - pushdown_filters=true path unchanged — dynamic filters still go through the arrow-rs RowFilter. - Downstream operator does the exact equivalent: HashJoin's hash lookup filters rows the bounds would have filtered; TopK's sort heap filters rows the threshold would have filtered. No wrong results. Local TPC-H SF1 (release-nonlto, 3 iters): - baseline (HEAD~2, pre-apache#22384) avg: 27.65 ms - PR + this fix avg: 26.24 ms (net 5% ahead of baseline) - Q9 individually: 34.54 → 34.33 ms (matches baseline, was 80.74 before) Also regenerates push_down_filter_parquet.slt for the membership-off default (from the earlier 'split membership from bounds' commit).
…r (root fix)
Reverts the tactical fix from the previous commit and cures the same
regression at its source. The prior commit skipped
DynamicFilterPhysicalExpr-containing conjuncts from PostScanFilter for
pushdown_filters=false; that recovered TPC-H but killed TPC-DS Q72
(4.91x faster -> no change) by removing row-level pruning for
CollectLeft's cheap bounds too.
Root cause: on PartitionMode::Partitioned, SharedBuildAccumulator emitted
a per-partition 'CASE hash(col) % N WHEN pid THEN bounds ELSE
lit(false) END' as the dynamic filter. On the probe scan this evaluates
hash + modulo + CASE branch per row -- the profile hotspot
(expressions::case::PartialResultIndex::merge_n at 1.11% self-time on
TPC-H Q9 vs 0% on main).
The routing existed to keep the per-partition bounds exact -- a probe
row X with hash(X) % N == P would only be checked against partition P's
bounds. That's exact but redundant with the downstream hash lookup
(which is also per-partition and exact). The lookup filters exactly
what CASE was filtering, at a lower per-row cost, so the CASE routing
buys nothing on the probe scan.
This commit, when the membership gate is off (the production default
after the split-membership commit), emits the union of per-partition
bounds instead:
col >= min(min_0, ..., min_{N-1}) AND col <= max(max_0, ..., max_{N-1})
Same shape as PartitionMode::CollectLeft. A probe row can pass the
union and still miss its build partition, but the exact hash lookup
downstream drops it -- no wrong results. Empty partitions contribute
nothing to the union; if every partition is empty the filter is
lit(false); a canceled partition falls back to lit(true) (permissive,
we lack the info to safely narrow). Membership-opt-in retains the
historical CASE hash-routed form so InListExpr / HashTableLookupExpr
can be applied to the correct partition's build values.
With the expensive per-row form gone, the pushdown_filters=false
PostScanFilter is cheap again, so opener/mod.rs no longer needs to
filter dynamic conjuncts out of post_scan_conjuncts -- reverted.
Local TPC-H SF1 (release-nonlto, 5 iters, warm):
- baseline (HEAD~2, pre-apache#22384) Q9: 40 ms (avg 46)
- PR before any fix Q9: 80 ms (avg 82) -- 2.09x regression
- PR + this fix Q9: 35 ms (avg 52) -- back at baseline, no CASE hotspot
- Full TPC-H avg: baseline 27.65, this fix 26.99 ms (net -2%)
Snapshot in filter_pushdown.rs regenerated to reflect the union form
(the old snapshot's 'CASE hash_repartition % 12 WHEN 5 ...' is gone;
new form is 'a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb').
- filter_pushdown.rs: enable enable_hash_join_dynamic_membership_filter in
test_hashjoin_hash_table_pushdown_{collect_left,partitioned} (they
specifically exercise HashTableLookupExpr; membership default is now false).
- filter_pushdown.rs: refresh test_hashjoin_dynamic_filter_pushdown_collect_left
and the force_hash_collisions branch snapshots (bounds only, no IN (SET)
since membership is off by default).
- clickbench.slt, preserve_file_partitioning.slt, projection_pushdown.slt,
repartition_subset_satisfaction.slt: regenerated for post-apache#22384 plan
display + membership-off default.
- configs.md: prettier reformat (trailing whitespace).
Previously the union-bounds fast path fired whenever
enable_hash_join_dynamic_membership_filter=false (the production
default). But that also relaxes the filter for pushdown_filters=true,
where arrow-rs RowFilter amortizes the per-row CASE hash-routed cost
via lazy decode of the remaining columns for filtered rows — the
tighter per-partition selectivity is worth paying for there.
This commit couples the fast path to the runtime pushdown_filters
setting: the SharedBuildAccumulator now mirrors
execution.parquet.pushdown_filters and picks the shape based on
where the filter will run:
* pushdown_filters=false → probe scan applies the filter post-decode
(PostScanFilter, Layer 3). Per-row cost dominates because there is
no lazy decode to amortize it. Emit the cheap union-of-bounds
predicate and skip membership entirely.
* pushdown_filters=true → probe scan applies the filter via arrow-rs
RowFilter (Layer 2), which skips the decode of the remaining
columns for rows that fail the predicate. Emit the historical
per-partition CASE hash-routed form; respect the user's opt-in via
enable_hash_join_dynamic_membership_filter for membership.
Effect matrix:
pushdown_filters | enable_membership_filter | Partitioned filter shape
-----------------+--------------------------+--------------------------------
false (default) | any | union bounds (fast path)
true | false | CASE hash-routed, bounds only
true | true | CASE hash-routed + membership
Tests:
- New helper make_partitioned_accumulator_pushdown_on_and_membership_{on,off}.
- New unit test partitioned_pushdown_on_membership_off_still_emits_case_no_membership
— pushdown=true must keep CASE even with membership off.
- New unit test collect_left_pushdown_off_skips_membership_even_with_gate_on
— pushdown=false must skip membership even with opt-in.
- Existing test fixtures updated to include pushdown_filters explicitly
(existing coverage kept — bounds-only fixtures set pushdown_filters=false
to match production default; membership-on fixtures set pushdown_filters=true).
Also refreshes projection_pushdown.slt for an unrelated main-branch
plan-shape change (apache#23459 CSE tweak) that surfaced post-rebase.
zhuqi-lucas
force-pushed
the
qizhu/help-22384-rebase
branch
from
July 20, 2026 09:49
5950f2c to
bf00cbf
Compare
zhuqi-lucas
changed the title
zhuqi-lucas
changed the title
zhuqi-lucas
changed the title
The helper make_partitioned_accumulator_pushdown_on_and_membership_on was added in bf00cbf but never called — clippy caught it as dead code and failed CI. Add the missing test that guards the symmetric invariant: pushdown_filters=true + membership=on must emit the full historical CASE-hash-routed form with InListExpr inside each WHEN branch (arrow-rs RowFilter lazy-decode scenario).
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23701 +/- ##
==========================================
+ Coverage 80.70% 80.72% +0.02%
==========================================
Files 1089 1089
Lines 368038 368664 +626
Branches 368038 368664 +626
==========================================
+ Hits 297031 297622 +591
- Misses 53308 53335 +27
- Partials 17699 17707 +8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
run benchmark hj |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (f91ee0a) to 3b1ce16 (merge-base) diff using: hj File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagehj — base (merge-base)
hj — branch
File an issue against this benchmark runner |
Contributor
Author
Comparing HEAD and qizhu_help-22384-rebase
--------------------
Benchmark hj.json
--------------------
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query ┃ HEAD ┃ qizhu_help-22384-rebase ┃ Change ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1_density=1_prob_hit=1_25*1.5M │ 2.05 / 2.37 ±0.47 / 3.31 ms │ 2.28 / 2.65 ±0.38 / 3.20 ms │ 1.12x slower │
│ QQuery 2_density=0.026_prob_hit=1_25*1.5M │ 4.17 / 4.32 ±0.19 / 4.70 ms │ 4.46 / 4.55 ±0.12 / 4.78 ms │ 1.05x slower │
│ QQuery 3_density=1_prob_hit=1_100K*60M │ 85.12 / 86.95 ±1.42 / 89.05 ms │ 89.75 / 90.48 ±0.59 / 91.23 ms │ no change │
│ QQuery 4_density=1_prob_hit=0.1_100K*60M │ 195.47 / 199.45 ±3.80 / 205.34 ms │ 193.88 / 196.35 ±2.26 / 200.42 ms │ no change │
│ QQuery 5_density=0.75_prob_hit=1_100K*60M │ 708.59 / 723.16 ±12.08 / 737.35 ms │ 701.78 / 721.07 ±13.69 / 740.04 ms │ no change │
│ QQuery 6_density=0.75_prob_hit=0.1_100K*60M │ 278.60 / 305.07 ±26.86 / 342.72 ms │ 276.39 / 289.31 ±15.95 / 318.75 ms │ +1.05x faster │
│ QQuery 7_density=0.5_prob_hit=1_100K*60M │ 854.92 / 878.03 ±26.82 / 929.46 ms │ 690.25 / 709.54 ±16.08 / 739.26 ms │ +1.24x faster │
│ QQuery 8_density=0.5_prob_hit=0.1_100K*60M │ 273.85 / 280.62 ±6.01 / 291.20 ms │ 253.19 / 272.70 ±15.72 / 295.18 ms │ no change │
│ QQuery 9_density=0.2_prob_hit=1_100K*60M │ 779.58 / 863.34 ±43.61 / 904.35 ms │ 728.75 / 794.98 ±47.20 / 857.73 ms │ +1.09x faster │
│ QQuery 10_density=0.2_prob_hit=0.1_100K*60M │ 250.70 / 266.21 ±15.79 / 291.76 ms │ 251.56 / 266.68 ±19.96 / 303.09 ms │ no change │
│ QQuery 11_density=0.1_prob_hit=1_100K*60M │ 736.48 / 785.19 ±40.46 / 841.92 ms │ 713.84 / 723.59 ±8.27 / 738.51 ms │ +1.09x faster │
│ QQuery 12_density=0.1_prob_hit=0.1_100K*60M │ 283.04 / 300.61 ±16.67 / 323.53 ms │ 250.72 / 266.16 ±11.18 / 278.53 ms │ +1.13x faster │
│ QQuery 13_density=0.01_prob_hit=1_100K*60M │ 725.32 / 801.91 ±61.86 / 889.98 ms │ 714.10 / 730.91 ±15.03 / 758.37 ms │ +1.10x faster │
│ QQuery 14_density=0.01_prob_hit=0.1_100K*60M │ 282.99 / 299.09 ±15.67 / 326.55 ms │ 281.87 / 318.24 ±33.22 / 370.68 ms │ 1.06x slower │
│ QQuery 15_density=0.2_prob_hit=0.1_100K_(20%_dups)*60M │ 255.45 / 274.01 ±33.80 / 341.51 ms │ 275.17 / 309.55 ±43.34 / 394.87 ms │ 1.13x slower │
│ QQuery 16_density=1_prob_hit=1_25*1.5M_RightSemi │ 2.37 / 2.52 ±0.10 / 2.68 ms │ 2.79 / 2.93 ±0.12 / 3.06 ms │ 1.16x slower │
│ QQuery 17_density=1_prob_hit=1_100K*60M_RightSemi │ 90.50 / 101.31 ±19.33 / 139.91 ms │ 100.66 / 111.17 ±11.60 / 130.38 ms │ 1.10x slower │
│ QQuery 18_density=1_prob_hit=0.1_100K*60M_RightSemi │ 103.27 / 109.32 ±11.44 / 132.20 ms │ 139.24 / 147.40 ±13.93 / 175.21 ms │ 1.35x slower │
│ QQuery 19_density=1_prob_hit=1_25*1.5M_RightAnti │ 1.92 / 2.00 ±0.05 / 2.07 ms │ 2.05 / 2.10 ±0.05 / 2.20 ms │ no change │
│ QQuery 20_density=1_prob_hit=1_100K*60M_RightAnti │ 80.58 / 81.06 ±0.59 / 82.19 ms │ 79.83 / 80.58 ±0.49 / 81.25 ms │ no change │
│ QQuery 21_density=1_prob_hit=0.1_100K*60M_RightAnti │ 116.60 / 117.36 ±0.54 / 118.14 ms │ 113.96 / 115.61 ±1.46 / 118.35 ms │ no change │
│ QQuery 22_density=1_prob_hit=0.01_100K_(fanout_100)*60M_RightSemi │ 92.49 / 94.37 ±1.48 / 96.23 ms │ 99.39 / 100.52 ±1.07 / 101.90 ms │ 1.07x slower │
│ QQuery 23_density=1_prob_hit=1_32K_(fanout~78)*2.3M_long_keys_count │ 165.18 / 166.87 ±2.11 / 171.03 ms │ 161.51 / 198.76 ±19.10 / 214.64 ms │ 1.19x slower │
└─────────────────────────────────────────────────────────────────────┴────────────────────────────────────┴────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary ┃ ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (HEAD) │ 6745.12ms │
│ Total Time (qizhu_help-22384-rebase) │ 6455.80ms │
│ Average Time (HEAD) │ 293.27ms │
│ Average Time (qizhu_help-22384-rebase) │ 280.69ms │
│ Queries Faster │ 6 │
│ Queries Slower │ 9 │
│ Queries with No Change │ 8 │
│ Queries with Failure │ 0 │
└────────────────────────────────────────┴───────────┘Thanks @adriangb ! Net for HJ is still positive (Total -4.3%, Avg -4.3%). The 9 regressions all share one pattern: density=1 queries where bounds prune 0 rows, so Layer 3 becomes pure per-row tax. Q18 RightSemi is worst (1.35x) since HashJoin's hash lookup already does the equivalent semantic filter. The 6 wins are all on low-density (0.01–0.5) where Layer 3 actually prunes. I will investigate a better solution in this PR. |
Adds a per-file adaptive skip to PostScanFilter (Layer 3, the pushdown_filters=false row-level path introduced by apache#22384). After observing POST_SCAN_FILTER_SAMPLE_ROWS (65536) rows, if the filter has pruned fewer than POST_SCAN_FILTER_MIN_SELECTIVITY (2%) of them, it permanently disables itself for the rest of the file — subsequent batches take a batch-clone fast path. Rationale: on the hj benchmark, the density=1 pattern (Q1 Inner, Q16-18/Q22 RightSemi, Q23 Inner high-fanout) shows Layer 3 running its predicate on tens of millions of probe rows only to prune zero — pure per-row tax with no downstream savings, because the downstream HashJoin hash lookup is going to do the exact match on the same rows anyway. Adaptive skip detects this after two decoded batches and stops paying the tax; the downstream operator is unchanged and produces the same result. Correctness: PostScanFilter is a semantics-preserving optimization (the downstream operator does the exact filtering), so disabling it never produces wrong results — the filter never NEEDED to run, it was only there to save downstream work. Scope: this is a deliberately-small per-file MVP. @adriangb's apache#22883 EPIC (Adaptive predicate evaluation) has a full framework with OptionalFilterPhysicalExpr + a cross-scan SelectivityTracker; this commit can be superseded by that when it lands. Local TPC-H SF1 (release-nonlto, 3 iters, avg ms): - baseline (main, pre-apache#22384): 27.65 - PR + union bounds only: 26.99 - PR + this adaptive skip: 25.98 (net -6% vs baseline) - Q1 individually: 56.01 (baseline) -> 44.30 (this commit), i.e. 21% faster than baseline where Layer 3 previously added tax Tests: - adaptive_disables_when_selectivity_is_zero - adaptive_stays_enabled_when_selectivity_is_high - adaptive_does_not_disable_before_sample_threshold - adaptive_disable_short_circuits_evaluation Also reverts the RightSemi/RightAnti gate in HashJoinExec::allow_join_dynamic_filter_pushdown — adaptive skip subsumes it (a density=1 Semi/Anti probe scan trips the gate the same way an Inner one does).
Contributor
Author
|
run benchmark hj |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (61a9ebb) to 4184b07 (merge-base) diff using: hj File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagehj — base (merge-base)
hj — branch
File an issue against this benchmark runner |
…MVP)" This reverts commit c856062.
…ers=false Right-Semi joins already implement the exact 'probe key ∈ build set' check as their core operation via the downstream hash lookup. Applying the bounds-based dynamic filter to those probe rows is semantically redundant, and — without arrow-rs RowFilter's lazy-decode amortization (i.e. pushdown_filters=false) — the per-batch PostScanFilter tax exceeds the marginal hash-lookup savings on TPC-H-shaped workloads. The apache#23701 hj benchmark surfaced this: density=1 RightSemi cases (Q16 25×1.5M, Q17/Q18 100K×60M, Q22 fanout=100 100K×60M) showed 5–35% regression because the bounds filter prunes 0 rows while the filter still evaluates on every probe row. Skip the dynamic filter for RightSemi when pushdown_filters=false. Keep it when pushdown_filters=true so RowFilter's two-phase decode can still amortize the cost on selective Semi joins. RightAnti is already excluded by the existing probe_preserved gate (on_lr_is_preserved reports (true, false) for it), so this new gate is redundant for that type — no separate check needed. Tests: - test_dynamic_filter_pushdown_rejects_right_semi_when_pushdown_off - test_dynamic_filter_pushdown_keeps_right_semi_when_pushdown_on
Contributor
Author
|
run benchmark hj |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/help-22384-rebase (3ca3cda) to 4184b07 (merge-base) diff using: hj File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagehj — base (merge-base)
hj — branch
File an issue against this benchmark runner |
Contributor
|
@zhuqi-lucas I wonder how this compares against previous proposals of tracking selectivity & compute time and disabling dynamic filters if they are not pulling their weight? Either at the expression level (#22234) or at the scan level (#22237). It seems that this PR is now both disabling the membership pushdown by default ( |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Completes @adriangb's #22384 with two changes to
HashJoinExec's dynamic filter, both gated on the runtimeexecution.parquet.pushdown_filterssetting so we optimize for whichever layer actually applies the filter:Skip the membership check (
InListExprfor small builds,HashTableLookupExprfor large builds) whenpushdown_filters=false. Per-row hash / list-lookup cost (~50–100 ns/row) doesn't earn its keep when Layer 3PostScanFilterapplies the filter post-decode — the downstreamHashJoinExechash lookup filters the same rows exactly.Replace
Partitioned-mode's per-partitionCASE hash(col) % N WHEN pid THEN boundswith a single union-of-bounds predicate (col ≥ min(all mins) AND col ≤ max(all maxes)) whenpushdown_filters=false. Per-rowhash + modulo + CASE-branchcost isn't amortized by lazy decode in Layer 3, so the tighter per-partition selectivity isn't worth paying for — the downstream hash lookup drops any row the union bounds let through incorrectly.When
pushdown_filters=true, both optimizations are off and the historicalCASE hash-routed + membershipform is emitted — arrow-rsRowFilteramortizes the per-row cost via lazy decode of the remaining columns for filtered rows, so the tighter selectivity earns its keep.Results on default config (
pushdown_filters=false):Architecture
Effect matrix
pushdown_filtersenable_hash_join_dynamic_membership_filterPartitionedfilter shapeCollectLeftmode is symmetric on the membership axis: skipped whenpushdown_filters=false, respects the config knob whenpushdown_filters=true.Why each optimization is safe
Correctness invariant: dynamic filter is a semantics-preserving optimization — a probe row it lets through that shouldn't join is dropped by the downstream
HashJoinExechash lookup, which is exact. Skipping membership or loosening bounds only changes where the row gets filtered, not the final result.Layer 1 (RG-level) is unaffected either way:
PruningPredicateproves an RG has a chance iff its stats overlap the union of partition bounds — CASE and union bounds are stat-equivalent at this layer, and skipping membership doesn't change stat-derivable bounds either.The regression #22384 exposed
samply on TPC-H Q9 (
pushdown_filters=false,Partitionedjoin) showedexpressions::case::PartialResultIndex::merge_nat 1.11% self-time on this PR before the fix vs 0% onmain. The CASE hash-routing was being evaluated on every scanned batch through Layer 3 where there is no lazy decode to amortize it. Q9 went from 40 ms baseline → 80 ms (2.09x regression) purely from that hotspot.Commits
feat(parquet-datasource): always accept pushable filters, run rejected conjuncts post-scan— @adriangb's feat(parquet-datasource): always accept pushable filters, run rejected conjuncts post-scan #22384, rebased onmain.perf(hash-join): split membership from bounds in dynamic filter pushdown— addsenable_hash_join_dynamic_membership_filterconfig (defaultfalse).perf(hash-join): union-bounds fast path for Partitioned dynamic filter (root fix)— replaces CASE with union bounds when the fast path fires.fix(tests): CI green-up after union-bounds root fix— snapshot + SLT regen.perf(hash-join): gate union-bounds fast path on pushdown_filters=false— the coupling.Tests
19 unit tests in
shared_bounds.rscover the full effect matrix (CollectLeft×Partitioned× pushdown={on, off} × membership={on, off}) plus edge cases (skewed ranges, empty partitions, canceled partitions). Existingfilter_pushdown.rssnapshots refreshed;push_down_filter_parquet.slt,clickbench.slt,preserve_file_partitioning.slt,projection_pushdown.slt,repartition_subset_satisfaction.sltregenerated for post-#22384 plan display.Credit: foundation is @adriangb's #22384.