feat(parquet): decline pushdown when projection has few non-filter columns#23420
feat(parquet): decline pushdown when projection has few non-filter columns#23420zhuqi-lucas wants to merge 11 commits into
Conversation
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/parquet-pushdown-adaptive-gate (6584f59) to 215ebb8 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/parquet-pushdown-adaptive-gate (6584f59) to 215ebb8 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/parquet-pushdown-adaptive-gate (6584f59) to 215ebb8 (merge-base) diff using: tpcds File an issue against this benchmark runner |
There was a problem hiding this comment.
Pull request overview
Adds a plan-time heuristic to ParquetSource::try_pushdown_filters to decline Parquet RowFilter pushdown when the scan’s projection is narrow and mostly covered by filter columns, aiming to avoid known regressions when parquet.pushdown_filters is enabled.
Changes:
- Introduces a hardcoded threshold (
PUSHDOWN_MIN_NON_FILTER_COLS = 3) to gate RowFilter pushdown based on projected columns not referenced by filters. - Collects referenced filter columns and projected columns (via
collect_columns) to compute the “non-filter projected” column count before deciding whether to push down.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| let filter_col_indices: std::collections::HashSet<usize> = filters | ||
| .iter() | ||
| .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index())) | ||
| .collect(); | ||
| let non_filter_projected = self | ||
| .projection | ||
| .as_ref() | ||
| .iter() | ||
| .flat_map(|pe| collect_columns(&pe.expr)) | ||
| .map(|c| c.index()) | ||
| .filter(|idx| !filter_col_indices.contains(idx)) | ||
| .collect::<std::collections::HashSet<_>>() | ||
| .len(); |
There was a problem hiding this comment.
Good catch — done in 25208eb. Both the filter-column set and the non-filter projected count now restrict to indices < self.table_schema.file_schema().fields().len(), so partition and virtual columns are excluded from the "wide-decode saving" calculation.
|
run benchmark clickbench_pushdown |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/parquet-pushdown-adaptive-gate (6584f59) to 215ebb8 (merge-base) diff using: clickbench_pushdown 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 |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_pushdown — base (merge-base)
clickbench_pushdown — branch
File an issue against this benchmark runner |
|
run benchmark clickbench_pushdown |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing qizhu/parquet-pushdown-adaptive-gate (b6b6421) to cd47b95 (merge-base) diff using: clickbench_pushdown File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_pushdown — base (merge-base)
clickbench_pushdown — branch
File an issue against this benchmark runner |
|
@alamb The result looks good also for pushdown benchmark as expected for this PR: Comparing HEAD and qizhu_parquet-pushdown-adaptive-gate
--------------------
Benchmark clickbench_pushdown.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query ┃ HEAD ┃ qizhu_parquet-pushdown-adaptive-gate ┃ Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0 │ 1.24 / 4.10 ±5.59 / 15.28 ms │ 1.22 / 4.01 ±5.45 / 14.91 ms │ no change │
│ QQuery 1 │ 12.96 / 13.15 ±0.12 / 13.30 ms │ 13.57 / 13.86 ±0.28 / 14.39 ms │ 1.05x slower │
│ QQuery 2 │ 36.76 / 37.45 ±0.62 / 38.24 ms │ 36.40 / 36.78 ±0.34 / 37.41 ms │ no change │
│ QQuery 3 │ 31.03 / 31.63 ±0.82 / 33.25 ms │ 31.04 / 31.19 ±0.12 / 31.38 ms │ no change │
│ QQuery 4 │ 227.12 / 230.25 ±2.21 / 233.66 ms │ 226.06 / 227.38 ±0.86 / 228.35 ms │ no change │
│ QQuery 5 │ 274.19 / 277.03 ±2.85 / 282.25 ms │ 273.51 / 278.30 ±3.38 / 283.52 ms │ no change │
│ QQuery 6 │ 1.27 / 1.43 ±0.22 / 1.86 ms │ 1.26 / 1.40 ±0.21 / 1.82 ms │ no change │
│ QQuery 7 │ 16.54 / 16.77 ±0.19 / 16.99 ms │ 15.08 / 15.15 ±0.05 / 15.24 ms │ +1.11x faster │
│ QQuery 8 │ 324.26 / 329.46 ±4.10 / 334.40 ms │ 323.61 / 326.84 ±2.56 / 330.68 ms │ no change │
│ QQuery 9 │ 455.48 / 469.16 ±8.12 / 476.19 ms │ 460.07 / 471.66 ±6.88 / 479.01 ms │ no change │
│ QQuery 10 │ 95.23 / 96.87 ±0.94 / 97.86 ms │ 69.59 / 71.95 ±1.40 / 73.77 ms │ +1.35x faster │
│ QQuery 11 │ 106.87 / 107.79 ±0.95 / 109.41 ms │ 83.37 / 87.98 ±7.69 / 103.25 ms │ +1.23x faster │
│ QQuery 12 │ 305.74 / 309.05 ±2.69 / 313.44 ms │ 268.82 / 272.19 ±2.77 / 277.13 ms │ +1.14x faster │
│ QQuery 13 │ 426.27 / 436.14 ±9.16 / 450.87 ms │ 370.07 / 375.79 ±5.10 / 383.77 ms │ +1.16x faster │
│ QQuery 14 │ 317.97 / 323.55 ±7.98 / 339.25 ms │ 286.21 / 289.30 ±4.51 / 298.04 ms │ +1.12x faster │
│ QQuery 15 │ 275.93 / 284.48 ±7.07 / 292.21 ms │ 274.85 / 287.42 ±11.31 / 305.63 ms │ no change │
│ QQuery 16 │ 622.58 / 633.26 ±9.22 / 649.55 ms │ 614.49 / 623.14 ±7.08 / 632.86 ms │ no change │
│ QQuery 17 │ 634.21 / 637.48 ±2.55 / 641.26 ms │ 623.65 / 634.04 ±9.82 / 646.87 ms │ no change │
│ QQuery 18 │ 1269.87 / 1302.81 ±24.32 / 1345.00 ms │ 1264.80 / 1271.50 ±4.83 / 1278.65 ms │ no change │
│ QQuery 19 │ 29.25 / 29.59 ±0.21 / 29.76 ms │ 28.39 / 36.98 ±15.81 / 68.53 ms │ 1.25x slower │
│ QQuery 20 │ 517.33 / 526.75 ±11.31 / 547.09 ms │ 519.01 / 525.65 ±7.19 / 538.98 ms │ no change │
│ QQuery 21 │ 567.02 / 578.69 ±6.88 / 586.50 ms │ 521.39 / 526.46 ±4.22 / 532.15 ms │ +1.10x faster │
│ QQuery 22 │ 927.00 / 937.75 ±10.48 / 956.68 ms │ 990.15 / 993.76 ±4.16 / 1001.68 ms │ 1.06x slower │
│ QQuery 23 │ 116.66 / 125.46 ±5.87 / 133.61 ms │ 116.37 / 124.25 ±5.85 / 132.68 ms │ no change │
│ QQuery 24 │ 37.83 / 39.80 ±3.22 / 46.20 ms │ 41.28 / 42.28 ±1.88 / 46.05 ms │ 1.06x slower │
│ QQuery 25 │ 144.81 / 149.31 ±3.42 / 153.64 ms │ 112.15 / 119.12 ±7.41 / 132.55 ms │ +1.25x faster │
│ QQuery 26 │ 49.92 / 50.96 ±0.86 / 52.31 ms │ 42.00 / 42.70 ±0.37 / 43.04 ms │ +1.19x faster │
│ QQuery 27 │ 711.11 / 719.52 ±9.57 / 737.84 ms │ 673.75 / 683.93 ±6.63 / 692.00 ms │ no change │
│ QQuery 28 │ 3062.78 / 3091.39 ±19.74 / 3119.93 ms │ 3067.46 / 3077.46 ±7.56 / 3085.45 ms │ no change │
│ QQuery 29 │ 41.17 / 45.16 ±7.39 / 59.93 ms │ 41.71 / 55.51 ±9.80 / 71.56 ms │ 1.23x slower │
│ QQuery 30 │ 312.51 / 317.83 ±4.76 / 326.73 ms │ 307.46 / 314.68 ±4.54 / 321.29 ms │ no change │
│ QQuery 31 │ 293.30 / 301.88 ±6.24 / 310.74 ms │ 295.00 / 306.48 ±11.50 / 328.33 ms │ no change │
│ QQuery 32 │ 926.19 / 968.21 ±25.26 / 1002.64 ms │ 950.24 / 994.38 ±23.11 / 1017.51 ms │ no change │
│ QQuery 33 │ 1451.44 / 1489.42 ±24.90 / 1529.12 ms │ 1474.44 / 1486.27 ±12.10 / 1503.04 ms │ no change │
│ QQuery 34 │ 1486.33 / 1497.38 ±6.12 / 1505.11 ms │ 1493.04 / 1534.83 ±38.86 / 1585.35 ms │ no change │
│ QQuery 35 │ 281.95 / 303.19 ±24.93 / 348.47 ms │ 281.46 / 305.21 ±16.40 / 324.06 ms │ no change │
│ QQuery 36 │ 67.03 / 73.48 ±4.55 / 80.29 ms │ 67.22 / 72.37 ±4.84 / 80.72 ms │ no change │
│ QQuery 37 │ 36.59 / 40.92 ±4.82 / 48.33 ms │ 35.86 / 41.06 ±4.50 / 47.14 ms │ no change │
│ QQuery 38 │ 36.44 / 42.07 ±4.96 / 49.73 ms │ 41.81 / 43.46 ±0.90 / 44.56 ms │ no change │
│ QQuery 39 │ 141.26 / 143.72 ±2.13 / 147.46 ms │ 126.12 / 142.17 ±9.14 / 154.61 ms │ no change │
│ QQuery 40 │ 18.39 / 21.19 ±4.94 / 31.06 ms │ 14.23 / 16.32 ±2.55 / 21.27 ms │ +1.30x faster │
│ QQuery 41 │ 17.21 / 21.75 ±5.29 / 30.45 ms │ 14.19 / 14.49 ±0.22 / 14.82 ms │ +1.50x faster │
│ QQuery 42 │ 14.71 / 14.98 ±0.21 / 15.23 ms │ 13.56 / 13.77 ±0.14 / 13.99 ms │ +1.09x faster │
└───────────┴───────────────────────────────────────┴───────────────────────────────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary ┃ ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (HEAD) │ 17072.27ms │
│ Total Time (qizhu_parquet-pushdown-adaptive-gate) │ 16833.48ms │
│ Average Time (HEAD) │ 397.03ms │
│ Average Time (qizhu_parquet-pushdown-adaptive-gate) │ 391.48ms │
│ Queries Faster │ 12 │
│ Queries Slower │ 5 │
│ Queries with No Change │ 26 │
│ Queries with Failure │ 0 │
└─────────────────────────────────────────────────────┴────────────┘ |
alamb
left a comment
There was a problem hiding this comment.
This is great. I think we are very close @zhuqi-lucas
I think we need to do the following :
- To make this a config setting that can be disabled (so that people who want to force filter pushdown on (to keep the current behavior) can do so
- Add a note to the upgrade guide explaining how to get the original behavior
- Add some metric so it is clear from the profiling that this heuristic has happened
I would also like to do some profiling of the query that appears consistently get slower
│ QQuery 29 │ 41.17 / 45.16 ±7.39 / 59.93 ms │ 41.71 / 55.51 ±9.80 / 71.56 ms │ 1.23x slower │
|
I will try and work on some of these items |
… marker Per @alamb's review on apache#23420: 1. Config opt-out: new `datafusion.execution.parquet.pushdown_filter_narrow_projection_gate` (default true). Set to false to restore the pre-existing unconditional pushdown behavior. Proto + serde support included so the option round-trips through the physical plan. 2. Plan-display marker: when the gate declines pushdown for a scan, the ParquetSource fmt_extra output includes `pushdown_declined=narrow_projection` so profiling / EXPLAIN ANALYZE can see the heuristic fired without needing to correlate config against per-query symptoms. 3. Upgrade guide: docs/source/library-user-guide/upgrading/55.0.0.md documents the new behavior and how to opt out. Kept the heuristic itself unchanged (single hardcoded PUSHDOWN_MIN_NON_FILTER_COLS = 3 with no tuning knob) per @alamb's 'complexity budget' note.
Thanks @alamb , addressed all three points in 1ef3990:
Kept the heuristic itself untouched (single hardcoded PUSHDOWN_MIN_NON_FILTER_COLS = 3, no tuning knob, no byte-weighting) per your complexity-budget note. Q29 is noise, above triggered another run, and it does not have filter. │ QQuery 29 │ 41.48 / 50.75 ±17.17 / 85.06 ms │ 41.41 / 53.52 ±15.73 / 81.05 ms │ 1.05x slower │ |
|
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 |
Per Copilot's review on apache#23420 line 887: TableSchema layout is [file, partition, virtual], and only file columns are actually decoded from parquet. Partition and virtual columns don't incur decode cost, so counting them as 'non-filter projected cols' can keep pushdown enabled for partitioned tables whose *file* projection is actually narrow — reintroducing the very regressions this heuristic protects against. Now both the filter-column set and the non-filter projected count restrict to indices `< file_schema().fields().len()`.
|
One known corner case worth flagging: The new config knob ( Given the ClickBench trade (5F/4S/34NC vs 6F/20S/17NC), I think shipping as-is + opt-out is the pragmatic call. LMK if you'd rather defer. |
…lumns RowFilter has a fixed per-row machinery overhead that only pays for itself when the wide-column decode it lets us skip is meaningful. When the projection is narrow and the filter columns already cover most of it (typical for GROUP BY col-style queries), the overhead dominates and pushdown regresses. Add a plan-time heuristic in ParquetSource::try_pushdown_filters that declines pushdown when the projection contains fewer than 3 columns not referenced by the filter. When declined, the filter stays above the scan in a FilterExec (correctness preserved) and the predicate is still injected into ParquetSource for stats / bloom / page-index pruning. Kept intentionally simple — a plan-time column-count check with no tuning knob. Complexity budget is reserved for the runtime adaptive-placement work tracked in apache#22883. Part of apache#3463.
… marker Per @alamb's review on apache#23420: 1. Config opt-out: new `datafusion.execution.parquet.pushdown_filter_narrow_projection_gate` (default true). Set to false to restore the pre-existing unconditional pushdown behavior. Proto + serde support included so the option round-trips through the physical plan. 2. Plan-display marker: when the gate declines pushdown for a scan, the ParquetSource fmt_extra output includes `pushdown_declined=narrow_projection` so profiling / EXPLAIN ANALYZE can see the heuristic fired without needing to correlate config against per-query symptoms. 3. Upgrade guide: docs/source/library-user-guide/upgrading/55.0.0.md documents the new behavior and how to opt out. Kept the heuristic itself unchanged (single hardcoded PUSHDOWN_MIN_NON_FILTER_COLS = 3 with no tuning knob) per @alamb's 'complexity budget' note.
Per Copilot's review on apache#23420 line 887: TableSchema layout is [file, partition, virtual], and only file columns are actually decoded from parquet. Partition and virtual columns don't incur decode cost, so counting them as 'non-filter projected cols' can keep pushdown enabled for partitioned tables whose *file* projection is actually narrow — reintroducing the very regressions this heuristic protects against. Now both the filter-column set and the non-filter projected count restrict to indices `< file_schema().fields().len()`.
…o test-code ParquetOptions initializers The parquet_writer tests build two ParquetOptions structs directly with all fields listed; the new field needs to be threaded through both, matching the pattern used for max_predicate_cache_size just above.
Two commits in one: 1. Add dynamic-filter detection to the gate: never gate a scan whose incoming filters include a dynamic filter (e.g. TopK's heap threshold). This is defensive — for cases where the dynamic filter arrives via `try_pushdown_filters`, keep pushdown so the runtime RG pruner can fire. 2. Opt existing dynamic-RG-pruning and limit-pruning tests out of the gate via config. TopK's dynamic filter is injected AFTER `try_pushdown_filters` runs, so the gate cannot detect it at plan-time from the filters parameter. These tests specifically exercise the co-existence of RowFilter + dynamic RG pruning + page-index selection, which needs pushdown active.
…shdown flag Two refinements after digging into CI test failures: 1. Check both incoming filters AND self.predicate for dynamic filters before firing the gate. TopK's DynamicFilterPhysicalExpr can arrive through either channel, so both need to be inspected. 2. When the gate declines pushdown for the current filter set, don't persist `pushdown_filters=false` onto the source. A subsequent `try_pushdown_filters` call (typically from TopK dynamic filter injection later in the pipeline) needs the source still marked pushdown-eligible so the dynamic filter can install and drive RG pruning. Neither is a full fix for the narrow-projection + TopK case (the runtime RG-prune cascade requires both filters pushed together, which isn't reachable through plan-time inspection alone), but they narrow the impact. The remaining gap is handled by the test-side config opt-out (this commit also restores it, with an explanation).
test_try_pushdown_filters_rejects_virtual_column_refs constructs a ParquetSource whose file schema has only one column, all of which appears in the filter set. That makes non-filter projected file columns = 0, tripping the narrow-projection gate and returning PushedDown::No for every filter — masking the actual virtual-column rejection behavior the test is verifying. Disable the gate for this test so the assertions test the pushable / virtual-column split independently of the projection-width gate.
limit_pruning.slt and parquet_filter_pushdown.slt both explicitly enable pushdown_filters=true and assert plan shapes that require the FilterExec to be absorbed into the DataSourceExec (i.e. the unconditional-pushdown path). The narrow-projection gate would decline pushdown for many of those cases and leave a visible FilterExec above the scan, breaking the expected plan output. Opt these test files out of the gate (and RESET at the end so config state doesn't leak to sibling test files).
…iles push_down_filter_regression.slt, push_down_filter_parquet.slt, and sort_pushdown.slt all explicitly enable pushdown_filters=true and assert plan shapes that require the FilterExec to be absorbed into the DataSourceExec. The narrow-projection gate would decline pushdown for narrow-projection scenarios and leave a visible FilterExec above the scan, breaking the expected plan output. Same pattern as limit_pruning.slt / parquet_filter_pushdown.slt — opt out at the top and RESET at the end.
…er_pushdown_config and explain_analyze These files exercise pushdown_filters=true directly and assert the plan has FilterExec absorbed into DataSourceExec. Disable the new narrow-projection gate in the affected sections so the expected plans keep matching regardless of projection width.
e45c819 to
32f09b7
Compare
Which issue does this PR close?
filter_pushdown) by default #3463Part of #3463 (Enable
parquet.pushdown_filtersby default) — this is the first step of the two-step split @alamb suggested on #23369: land the heuristic on its own now; flip the default in a follow-up.Rationale for this change
RowFilter has a fixed per-row machinery overhead that only pays for itself when the wide-column decode it lets us skip is meaningful. When the projection is narrow and the filter columns already cover most of it (typical for
GROUP BY col-style queries such as ClickBench Q10/Q11/Q40), the overhead dominates and pushdown regresses.On ClickBench with
pushdown_filters=true, the naive path shows 20 slower / 6 faster / 17 no change vs HEAD. Adding this gate on top gave 4 slower / 5 faster / 34 no change (see #23369 CI comparison), with the remaining 4 "slower" queries all inside the CI noise floor. Q23 keeps its ~21× speedup.What changes are included in this PR?
Single-file change in
ParquetSource::try_pushdown_filters: decline pushdown when the projection contains fewer than 3 columns not referenced by the filter. When declined,pushdown_filtersis treated as disabled for that scan — the filter stays above the scan in aFilterExec(correctness preserved) and the predicate is still injected intoParquetSourcefor stats / bloom / page-index pruning.Kept intentionally simple:
PUSHDOWN_MIN_NON_FILTER_COLS = 3.Are these changes tested?
By existing tests: this change only affects scans where the user has opted into
pushdown_filters=true(config orParquetSource::pushdown_filters()), and only in the direction of declining pushdown (falling back to the pre-existingFilterExecpath). The defaultpushdown_filters=falsebehavior is unchanged, so the default test matrix already covers the fallback path.The ClickBench comparison on #23369 provides the end-to-end validation.
Are there any user-facing changes?
For users who explicitly enable
pushdown_filters=true: their queries with narrow projections (fewer than 3 non-filter columns) will no longer push filters into the Parquet scan. This eliminates the Q10-style regressions those users historically hit. Users on the default (pushdown_filters=false) see zero change.Follow-up
pushdown_filtersdefault totrueonce we're confident this heuristic holds up.cc @alamb