feat(parquet-datasource): always accept pushable filters, run rejected conjuncts post-scan - #22384
Draft
adriangb wants to merge 5 commits into
Draft
feat(parquet-datasource): always accept pushable filters, run rejected conjuncts post-scan#22384adriangb wants to merge 5 commits into
adriangb wants to merge 5 commits into
Conversation
github-actions
Bot
added
core
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (7dd85fc) to c8b784a (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (7dd85fc) to c8b784a (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (7dd85fc) to c8b784a (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 |
adriangb
force-pushed
the
parquet-post-scan-filter
branch
from
May 20, 2026 08:06
7dd85fc to
2fffad2
Compare
|
🤖 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 |
3 tasks
adriangb
force-pushed
the
parquet-post-scan-filter
branch
from
July 2, 2026 23:10
1404755 to
cca69df
Compare
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (cca69df) to ad7d6ea (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (cca69df) to ad7d6ea (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (cca69df) to ad7d6ea (merge-base) diff using: clickbench_partitioned 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 |
This was referenced Jul 13, 2026
zhuqi-lucas
added a commit
to zhuqi-lucas/arrow-datafusion
that referenced
this pull request
Jul 19, 2026
…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).
zhuqi-lucas
added a commit
to zhuqi-lucas/arrow-datafusion
that referenced
this pull request
Jul 20, 2026
…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').
zhuqi-lucas
added a commit
to zhuqi-lucas/arrow-datafusion
that referenced
this pull request
Jul 20, 2026
- 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).
zhuqi-lucas
added a commit
to zhuqi-lucas/arrow-datafusion
that referenced
this pull request
Jul 20, 2026
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).
zhuqi-lucas
added a commit
to zhuqi-lucas/arrow-datafusion
that referenced
this pull request
Jul 20, 2026
…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).
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
Contributor
Author
|
show benchmark queue |
|
Hi @adriangb, you asked to view the benchmark queue (#22384 (comment)).
File an issue against this benchmark runner |
Contributor
Author
|
run benchmarks env: |
Contributor
Author
|
show benchmark queue |
|
Hi @adriangb, you asked to view the benchmark queue (#22384 (comment)).
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpchCPU 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 parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
Benchmark for this request failed. Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Last 20 lines of output: Click to expandFile an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #22384 +/- ##
========================================
Coverage 80.99% 80.99%
========================================
Files 1106 1106
Lines 383158 383334 +176
Branches 383158 383334 +176
========================================
+ Hits 310331 310475 +144
- Misses 54513 54531 +18
- Partials 18314 18328 +14 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpcds env: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (e743ae8) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
…ltering `PostScanFilter::filter` ran `filter_record_batch` over the whole decoded batch and only then applied the projector, so every column the decoder mask was widened for — filter-only columns like TPC-H's `l_shipdate` — was pushed through the filter kernel just to be discarded immediately afterwards. `FilterExec::filter_and_project` does it the other way round: project first (a cheap `Arc` reslice), then filter only what survives into the output. Match that here. `PostScanFilter::evaluate` now returns the selection mask instead of an already-filtered batch, and `DecoderProjection::narrow` drops the filter-only columns before the caller applies it. The projector is rebased onto the narrowed schema at construction time, so the runtime cost is one `RecordBatch::project` per batch. Null mask entries are normalized with `prep_null_mask_filter` so the rows-matched / rows-pruned metrics still count a NULL predicate result as pruned, matching `filter_record_batch`'s own treatment. Narrowing is skipped entirely when the projector already reads every stream column, which is always the case without a post-scan filter — that path keeps its zero-extra-work guarantee. Measured on TPC-H SF=1 q3 (lineitem scan, pushdown_filters=false, dynamic filter pushdown off to isolate this): 141ms -> 126ms of post-scan filter time, matching main's FilterExec. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…tch_size The in-scan post-scan filter emitted one batch per decoded batch. With a selective predicate that means slivers: TPC-H SF=1 q3's lineitem scan produced 30.52K rows spread over 742 batches — about 41 rows each — and every operator above the scan paid per-batch overhead on all 742. `FilterExec` runs its output through a coalescer for exactly this reason. Do the same here: survivors are pushed into an `arrow` `BatchCoalescer` built with the scan's `batch_size`, and the stream hands out batches only once the coalescer has assembled a full one, flushing the remainder at end of input. The coalescer buffers *narrowed, pre-projection* batches, so the projection expressions are also evaluated once per full-size batch instead of once per sliver. The coalescer is installed only when the file has a post-scan filter. Without one the decoder's batches are already the right shape and routing them through a coalescer would add a copy for nothing. The stream-level LIMIT moves to the coalescer's output, where it still counts rows actually emitted. When the limit is exhausted any buffered remainder is dropped rather than flushed. One `.slt` metric moves: a 10-row result's `output_bytes` goes 80.0 B -> 64.0 KB, because the coalescer allocates at the target batch size. That is not new behaviour — main's `FilterExec` reports 64.0 KB for a 27-row result through the same mechanism. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…filter `PostScanFilter` conjoined its conjuncts into one `BinaryExpr` `AND` and evaluated it over the whole decoded batch. A fused `AND` never compacts between conjuncts, so every conjunct ran on ~every decoded row — and with `pushdown_filters = false` the *whole* predicate lands here, so an expensive dynamic filter (a `CASE` over per-partition hash-table probes) was evaluated on all 6.00 M decoded lineitem rows in TPC-H q3 even though `l_shipdate > '1995-03-15'` had already rejected 46% of them. Keep the conjuncts split and run them through a compact-once loop instead: evaluate a conjunct, `AND` its mask into the accumulator, and physically compact the working batch to the survivors once a conjunct proves selective enough. This is the post-scan equivalent of what arrow-rs already does for the `RowFilter` path, where each conjunct is its own `ArrowPredicate` applied against an accumulating `RowSelection`. `DecoderProjection::narrow` used to run before the filter, which is no longer safe: with several conjuncts, a later one may need a column the projector does not read. `evaluate` now takes the batch by value and returns the working batch plus a residual mask, so the caller narrows after the loop and before applying that mask — the filter kernel still never touches a column that is about to be dropped, and the last conjunct never compacts (its mask goes to the caller instead). Metrics and semantics are unchanged: `post_scan_rows_pruned` stays "rows in minus final survivors", and a NULL predicate result still drops the row. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…ost-scan predicate At `pushdown_filters = false` the scan routed the entire predicate into the post-scan filter, dynamic filters included. That is a bad trade for the canonical dynamic filter shape — a `CASE` over per-partition hash-table probes — because the operator that produced it re-checks the same rows anyway, so evaluating it in the scan is doing the join's work twice, on more rows. In TPC-H q3 it was 52% of total CPU. A dynamic filter earns its keep at the row-group / page level, where one evaluation against statistics can skip millions of rows, so it still joins the scan's predicate and `dynamic_rg_pruning=eligible` is unchanged. It is only kept out of the *row-level* filter, and only at this setting: with `pushdown_filters = true` arrow-rs evaluates it against an accumulating `RowSelection`, so it sees only rows the cheaper conjuncts already kept and the rows it rejects need never be decoded. Dropping a conjunct from the post-scan filter is only sound if nothing was relying on the scan to enforce it, so `try_pushdown_filters` is the other half of the change: a dynamic filter is reported as *not* pushed down at this setting, exactly as on `main`, so a parent that was enforcing one keeps doing so. This is not a dead path — a join's dynamic filter reaches the scan as a parent filter when it is pushed through an intervening operator, and a plain `SortExec` materialises a `FilterExec` for the conjuncts that come back unsupported. The `dynamic_filter_pushdown_config.slt` expectation is restored to the plan `main` produces for exactly that case. Static conjuncts, and conjuncts the `RowFilter` machinery rejected, still run post-scan — that is what the in-scan filter is for and it is unchanged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (bd6ee28) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (bd6ee28) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-post-scan-filter (bd6ee28) to 2bfdd4a (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (bd6ee28) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing parquet-post-scan-filter (bd6ee28) to 2bfdd4a (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
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
3 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.
Which issue does this PR close?
Rationale for this change
The Parquet scan today gives the predicate to the source for row-group / page
/ bloom pruning, but only applies it row-level via
RowFilterwhenpushdown_filters=true. With pushdown off, aFilterExecis left above thescan to do row-level filtering. This is the substrate the adaptive-filter
work (#22237 / #22144) builds on, but it also has a real correctness bug
on
mainthat's worth fixing on its own.build_row_filter(row_filter.rs:994-1083, see its own doc comment at1009-1014) silently drops conjuncts thatFilterCandidateBuilder::buildreturns
Ok(None)for, andRowFilterGenerator::buildswallows whole-builderrors. By the time
build_row_filterruns,ParquetSource::try_pushdown_filtershas already accepted the filter and the parent
FilterExechas been removed— so those dropped conjuncts are never applied anywhere and the query
returns wrong results. The most reproducible trigger is the per-file expr
adapter rewriting a predicate that was pushable at table schema time into
something
PushdownCheckerrejects at physical file schema time (schemaevolution / coercion, whole-struct refs introduced by the rewrite, etc.).
This PR makes the Parquet scan always own its pushable filters and
guarantees every accepted conjunct is applied — either by the parquet
RowFilteror by a new in-scan post-scan filter evaluated on decodedbatches (the in-scan equivalent of a
FilterExec). Nothing is silentlydropped.
What changes are included in this PR?
row_filter.rs— never drop conjuncts.build_row_filternow returnsResult<(Option<RowFilter>, Vec<Arc<dyn PhysicalExpr>>)>— the secondelement is the conjuncts it could not place.
RowFilterGeneratorexposesthem via
rejected_conjuncts(); on whole-file build errors it routesevery conjunct through that list (no silent error swallowing).
post_scan_filter.rs(new module) — encapsulates the projectionwidening + rebasing + filter evaluation behind a small API:
PostScanFilter— evaluates a predicate on decoded batches; SQLWHEREsemantics (
NULLdrops the row); records rows-pruned / matched / time.DecoderProjection::build(projection, post_scan_conjuncts, schemas, …)— widens the decoder projection over (user projection ∪ post-scan
conjunct columns), rebases the projection and conjuncts onto the
decoder's stream schema, and returns the
ProjectionMask,Projector,replace_schemaflag, and the rebasedPostScanFilter. Empty conjunctslist = the prior projection-only behaviour, so the opener routes every
file through this one call.
ParquetSource::try_pushdown_filters— always returns the per-filterYes/Nodiscriminant based oncan_expr_be_pushed_down_with_schemas,regardless of the
pushdown_filtersconfig. The flag still records whetherthe
RowFilter(vs. post-scan) path is used downstream.opener/mod.rs::build_stream— orchestrates: builds theRowFilterGeneratoronly whenpushdown_filters=true; computespost_scan_conjuncts(rejected conjuncts when pushdown is on, fullsplit-conjunction of the predicate when off); calls
DecoderProjection::build;routes the
LIMITtoremaining_limitinstead of a decoder limit wheneverthe post-scan filter is present (decoder-local limit + post-scan filter is
unsafe — the decoder would stop before the post-scan rejected enough rows).
The prior inline
build_projection_read_plan/reassign_expr_columns/make_projectorblock is replaced by the singleDecoderProjection::buildcall — net simplification.
push_decoder.rs—PushDecoderStreamStatecarries anOption<PostScanFilter>; in theDecodeResult::Dataarm it applies thefilter, skips empty batches, then enforces
remaining_limitand projects.DecoderBuilderConfigis fedprojection_mask: &ProjectionMaskdirectly(no longer the full
ParquetReadPlan).metrics.rs— newpost_scan_rows_pruned/post_scan_rows_matchedcounters and
post_scan_filter_eval_timeTime, mirroring the existingpushdown_rows_*/row_pushdown_eval_timesoEXPLAIN ANALYZEkeepssurfacing filter cost once the
FilterExecis gone.The adaptive-filter machinery from #22237 / #22144 (
SelectivityTracker,FilterIdtagging, per-conjunct pruning stats,StrategySwapmid-streamswaps,
OptionalFilterPhysicalExpr, customarrow-rsbranch, the threefilter_pushdown_*config knobs) is intentionally not included — thisPR is the standalone substrate they would build on.
Are these changes tested?
Yes.
build_row_filter_surfaces_rejected_struct_conjunct(row_filter.rs)asserts the new API contract directly —
build_row_filterno longerdrops the rejected conjunct.
rejected_struct_conjunct_runs_post_scan_not_dropped(opener/mod.rs)is an end-to-end test: with
pushdown_filters=trueand as IS NOT NULLpredicate over a struct column where row 1 is NULL,
mainreturns 3rows (conjunct silently dropped, predicate relaxed) and this PR returns
the correct 2.
filter-pushdown tests, and sqllogictest all pass.
pruning" behaviour (e.g. `a = 1` over data `[1, 2, 3]` should still
return 3 rows because the row group wasn't stats-pruned) are updated to
reflect the new behaviour — the scan now applies the predicate row-level
via the post-scan filter, so they return only the matching row.
.sltfiles are regenerated (clickbench,push_down_filter_parquet, projection_pushdown, parquet*, etc.) — the
FilterExecabove parquet scans is gone from those plans. Spuriouswhitespace-only churn from
--completewas reverted.Are there any user-facing changes?
references, certain schema-evolution edge cases) now return correct
results with
pushdown_filters=true. Before this PR they were silentlyrelaxed.
FilterExecno longer appears above aDataSourceExecforpushable filters on a parquet source. The predicate appears as
predicate=…on the
DataSourceExec. Query results are unchanged.ParquetFileMetrics—post_scan_rows_pruned,post_scan_rows_matched,post_scan_filter_eval_time— appear in
EXPLAIN ANALYZEoutput for parquet scans.build_row_filterreturn type changes fromResult<Option<RowFilter>>toResult<(Option<RowFilter>, Vec<Arc<dyn PhysicalExpr>>)>; callers mustapply the rejected conjuncts (or they'll have the same drop-on-floor bug
this PR fixes).
Draft. Happy to split into a stack (refactor → row_filter fix → opener
orchestration + tests) if reviewers prefer.
🤖 Generated with Claude Code