perf(parquet): fuse consecutive filters on the same projection - #10859
Open
haohuaijin wants to merge 13 commits into
Open
haohuaijin wants to merge 13 commits into
haohuaijin wants to merge 13 commits into
Conversation
Contributor
Author
|
Replace the dedicated `with_same_projection_predicates` loop and the predicate-group bookkeeping in the push decoder state machine with a crate-private `FusedPredicate` that implements `ArrowPredicate`. `RowFilter::fuse_same_projection` groups consecutive predicates that share a single-leaf projection once at build time, so `ReadPlanBuilder`, `FilterInfo` and the LIMIT short-circuit work unchanged and the sync reader gets fusion for free. Within a batch the accepted rows are tracked as a `RowSelection` whose mask/selector backing follows the reader's `RowSelectionPolicy`, via the new `RowSelectionPolicy::resolve` and `RowSelection::into_boolean_buffer`.
Only the accumulated selection drives the composition algorithm in `FusedPredicate::evaluate`, so adapt it once, right before it is used as the left operand of `and_then`. The per-predicate mask is composed as-is and the last predicate's result is no longer converted, which removes one strategy scan per rejecting predicate and a mask-to-selectors-to-mask round trip on the final predicate. Interleaved benchmark runs on the int64 cases show a geometric-mean speedup of about 2%, and 9-15% on eight-predicate fragmented chains with 99% survivors, with no regressions.
Contributor
Author
|
the benchmark result of
|
haohuaijin
marked this pull request as ready for review
September 6, 2026 15:05
haohuaijin
force-pushed
the
parquet-same-projection-filter-fusion
branch
from
September 6, 2026 15:54
f78e300 to
9093f55
Compare
Contributor
Author
|
Hi @alamb, would you have time to review this? The idea is that consecutive RowFilter predicates on the same column are evaluated together on one decoded batch, instead of each predicate decoding the column again or replaying it from the predicate cache. The core change is in I'd like to know whether you think this is a good direction. |
alamb
pushed a commit
that referenced
this pull request
Sep 19, 2026
# Which issue does this PR close? Related to #10926. # Rationale for this change `RowFilter` evaluates each `ArrowPredicate` separately, so consecutive predicates on the same projection decode that column, or replay it from the predicate cache, once per predicate. #10859 fuses such chains. Per the contributing guide, the benchmark is submitted separately so it can run on the automated runner and serve as the baseline for that change. # What changes are included in this PR? A criterion benchmark, `parquet/benches/arrow_reader_predicate_fusion.rs`, that scans an in-memory Snappy Parquet file of 262,144 rows through the async reader. Case names are `type/layout/cache/predicates/profile`: - `4/all99` chains across both column types (`int64`, `string`), layouts (`fragmented`, `clustered`), and cache modes (`cached`, `uncached`): 8 cases - fragmented `1/all99` controls, `2/all99`, and `2/all50` chains across both types and cache modes: 12 cases - clustered uncached `2/all50` and `4/all50` chains across both types, retaining cases where fusion has shown regressions: 4 cases - `int64/fragmented/4` with `early1` and `late1` across both cache modes, covering predicate ordering: 4 cases - a `selection_boundary` group with run lengths 16 and 64 on either side of the default row selection policy threshold: 2 cases 30 cases in total. Each case is validated once outside measurement for the expected row count and predicate cache use. Run-length cases also validate the predicate mask run lengths. # Are these changes tested? The benchmark builds and all 30 cases pass with `cargo bench -p parquet --bench arrow_reader_predicate_fusion --features "arrow async snap" --locked -- --test`. # Are there any user-facing changes? No.
Contributor
|
run benchmark arrow_reader_predicate_fusion |
1 similar comment
Contributor
|
run benchmark arrow_reader_predicate_fusion |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-same-projection-filter-fusion (916b65c) to a7b89dd (merge-base) diff Run configurationrun benchmark arrow_reader_predicate_fusionBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader_predicate_fusion File an issue against this benchmark runner |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing parquet-same-projection-filter-fusion (916b65c) to a7b89dd (merge-base) diff Run configurationrun benchmark arrow_reader_predicate_fusionBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader_predicate_fusion File an issue against this benchmark runner |
|
🤖 Arrow criterion benchmark completed (GKE) | trigger Instance: Comparing parquet-same-projection-filter-fusion (916b65c) to a7b89dd (merge-base) diff Run configurationrun benchmark arrow_reader_predicate_fusionCPU Details (lscpu)Details
Resource Usagebase (merge-base)
branch
File an issue against this benchmark runner |
|
🤖 Arrow criterion benchmark completed (GKE) | trigger Instance: Comparing parquet-same-projection-filter-fusion (916b65c) to a7b89dd (merge-base) diff Run configurationrun benchmark arrow_reader_predicate_fusionCPU Details (lscpu)Details
Resource Usagebase (merge-base)
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?
Closes Parquet: support same-projection
RowFilterfusion #10926.Related to Improve performance of CachedArrayreader filter+concat #10774 and parquet: Improve performance of mask/selection construction in
ReadPlanBuilder::with_predicate_options#10776.bench(parquet): add same-projection predicate chain benchmark #11007 adds a benchmark for same-projection predicate chains.
Rationale for this change
Consecutive same-projection predicates can repeatedly decode a column or replay it from the predicate cache. The push decoder now wraps eligible groups in a
FusedPredicateand evaluates them from one decoded stream. Predicates keep their order, and later predicates receive only surviving rows.Fusion is limited to a single top-level, non-repeated leaf. Contiguous survivors use zero-copy slices; fragmented survivors use
filter_record_batch. Intermediate selections follow the reader'sRowSelectionPolicy, so the defaultAutopolicy switches betweenRowSelectorruns and bitmaps by run density.The synchronous reader is unchanged and is left as a follow-up: it has no predicate cache, so it would benefit at least as much, but this PR keeps fusion inside the push decoder.
There is no option to disable fusion. The measured regressions are bounded (see below) and only appear where compaction of ~50% survivors costs more than the saved decode.
What changes are included?
RowFilterdocs.Performance
ClickBench Q25 (
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10) has two predicates onSearchPhraseonce the TopK dynamic filter is pushed down, so the pair is fused.Measured with DataFusion
eea8c0961dfbench clickbenchon the partitioned 100-file dataset, 12 partitions, batch size 8192,[patch.crates-io]pointing the arrow crates at local checkouts. DataFusion builds against the released 59.x API, somainis the59.3.0tag andfusionis59.3.0plus this PR's diff. Three interleaved rounds of 40 iterations per variant, 120 samples each:Fusion removes the 21% cost that enabling pushdown adds on
mainfor this query: fusion/on is 18.6% faster than main/on and 1.1% faster than main/off.Testing
User-facing changes
No public API changes.
RowFilterdocs now describe when consecutive predicates share one decode.