Repository navigation
feat: support pushdown-aware dynamic filter - #23701
zhuqi-lucas wants to merge 12 commits into
Conversation
|
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 |
|
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 |
|
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 |
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.
5950f2c to
bf00cbf
Compare
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:
|
|
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 |
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).
|
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
|
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 |
|
@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 ( |
|
Good point on splitting @adriangb — I agree the shape change and the config gate are logically independent and shouldn't both stack on #22384. Two clarifications on ordering:
If your adaptive work (#22234 / #22237) lands before step 3, A + B can be superseded — but they're cheap enough (~50 lines total) that landing them independently doesn't block anything. Happy to split as described. Want me to close this PR and open A + B fresh against main? |
|
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days. |
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.