Repository navigation
perf: evaluate Parquet Bloom filters incrementally during row group pruning - #25811
mohitgurav20 wants to merge 14 commits into
Conversation
github-actions
Bot
added
datasource
|
hi @mohitgurav20 |
|
@kosiew ohkk sir will work on it . |
…d short-circuit test Ensure Bloom filter pruning only records matches when filters are evaluated, respects row group should_scan checks, and removes unrelated join changes.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #25811 +/- ##
==========================================
- Coverage 82.94% 82.94% -0.01%
==========================================
Files 1155 1155
Lines 458401 458462 +61
Branches 458401 458462 +61
==========================================
+ Hits 380222 380264 +42
- Misses 55132 55146 +14
- Partials 23047 23052 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
Thanks for improving Parquet Bloom filter pruning. Incremental evaluation avoids unnecessary column reads, and the new regression test demonstrates the intended I/O reduction.
Moving pruning into the asynchronous loading phase is appropriate because evaluation results determine whether further reads are needed. Removing the separate pruning state simplifies the pipeline while preserving conservative pruning behavior and the existing access-plan ownership.
I'm requesting changes because repeatedly evaluating the full predicate introduces substantial redundant CPU work when row groups survive pruning. This is a demonstrated operation-count regression, not a measured latency regression, and the I/O savings may still outweigh it. Please address the repeated evaluations and add regression coverage and benchmarks for both rejecting and surviving workloads. The additional multi-row-group and missing-filter tests are non-blocking suggestions.
| ); | ||
|
|
||
| let _timer_guard = bloom_filter_eval_time.timer(); | ||
| match predicate.prune(&row_group_filters) { |
There was a problem hiding this comment.
Calling the full predicate.prune() after every column reloads previously checked guarantees. For N Bloom-filtered columns whose queried values are all present, membership checks grow from N to N(N+1)/2 (528 versus 32 for N=32), with additional statistics evaluation. Please evaluate only guarantees for the newly loaded column while preserving conservative missing-filter handling and early rejection. Add an operation-count regression for surviving groups and benchmark both rejecting and surviving workloads.
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn incremental_bloom_filter_pruning_short_circuits_remaining_columns() { |
There was a problem hiding this comment.
Could we extend this test with an early-pruned row group followed by one containing a matching row? With statistics pruning disabled and decoder filtering enabled, verify the matching row is returned and Bloom metrics report (pruned=1, matched=1). Here matched means Bloom pruning could not exclude the group.
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn incremental_bloom_filter_pruning_short_circuits_remaining_columns() { |
There was a problem hiding this comment.
Could we also test partial Bloom-filter availability through the opener? Enable filters only for the second predicate column and verify it can still prune despite the first being unavailable, while surviving rows remain readable. Check that a surviving group with an available filter increments matched once, and an all-filters-absent group increments neither outcome. This can share the multi-row-group test setup.
…valuation and extend tests The previous incremental pruning loop accumulated all loaded Bloom filter columns into a single BloomFilterStatistics object and re-ran predicate.prune() after every column load. For N predicate columns where all values survive pruning, this causes N*(N+1)/2 membership checks instead of N (e.g. 528 vs 32 for N=32 columns), as the reviewer correctly identified. Fix: evaluate the predicate against a *single-column* BloomFilterStatistics containing only the newly loaded column. Because PruningStatistics::contained returns None for columns absent from the statistics object, evaluating a single-column filter is semantically equivalent for both conjunctive and disjunctive predicates -- only the newly presented column can drive a pruning decision in that round. This keeps the short-circuit behaviour intact while reducing complexity from O(N^2) to O(N). Also replace the any_filter_loaded sentinel (previously row_group_filters .is_empty()) with a plain bool to remove the now-unnecessary accumulator allocation, and drop the stray _timer_guard that was placed outside the loop. Tests updated/added: - incremental_bloom_filter_pruning_short_circuits_remaining_columns: extended to two row groups; Query 3 (a=10) verifies the (pruned=1, matched=1) split and that the actual matching row is returned through decoder filtering. - bloom_filter_pruning_with_partial_column_availability: new test verifying that a Bloom filter available for only one predicate column (b, not a) can still prune a row group, and that matched/pruned metrics are correct when filters are partially absent.
|
There was a problem hiding this comment.
Thanks for the follow-up. The single-column Bloom filter approach eliminates repeated membership probes, and the additional tests improve coverage of short-circuiting, surviving row groups, and partially available filters.
Evaluating filters during asynchronous loading is the right architectural direction because pruning decisions must prevent unnecessary I/O. However, the incremental path also needs to avoid repeatedly performing expensive full-predicate evaluation when filters cannot reject early.
Requesting changes because comparative benchmarks now demonstrate a substantial CPU regression for non-rejecting workloads, reaching 43x at 32 columns with in-memory storage. The early-rejection gains are significant, reaching 30x with simulated storage latency, and should be preserved. Please eliminate the repeated full-evaluation overhead and include rejecting and surviving benchmark results. No additional correctness issues were identified in the follow-up.
| col_filter.insert(column_name, bf, *physical_type, *type_length); | ||
|
|
||
| let _timer_guard = bloom_filter_eval_time.timer(); | ||
| match predicate.prune(&col_filter) { |
There was a problem hiding this comment.
Each loaded column still runs the full predicate.prune(&col_filter) path, rebuilding statistics and evaluating the entire expression. My Criterion comparison against main shows up to 43x slower reject_last and 12x slower survive_all at 32 columns on in-memory storage, despite up to 30x faster early rejection with simulated latency. Please remove the repeated full-evaluation overhead while preserving early rejection, and include rejecting and surviving benchmarks in the PR.
There was a problem hiding this comment.
thanks for ur patience sir will work on it and follow ur steps
…ectly and add benchmarks - Directly check Guarantee::In literal guarantees on loaded column bloom filters via BloomFilterStatistics::is_known_not_present, preserving early rejection while removing repeated full-expression evaluation overhead - Factor column guarantees into BloomColumnGuarantee struct - Include Criterion benchmarks for reject_first, reject_last, and survive_all queries across memory and slow storage configurations - Add test coverage for all-filters-absent queries
|
@kosiew Thanks for the clear feedback and the reproducible Criterion benchmark. I have updated the implementation in commit
|
…eral negation filters
3 participants
Closes #25797
Rationale for this change
Parquet Bloom filter pruning currently loads filters for all relevant predicate columns before evaluating whether a row group can be excluded. For queries with multiple filtering conditions, this results in unnecessary I/O reads if an earlier filter already proves the group cannot match.
By incrementally evaluating Bloom filters and stopping reading additional filters for a row group as soon as a condition proves it cannot match, we can skip unnecessary reads and release memory eagerly. This brings a solid performance bump for observability workloads filtering across several columns over remote storage.
What changes are included in this PR?
Incremental Evaluation: Folded the Bloom filter evaluation directly into the load_bloom_filters async loop to evaluate immediately as each column is fetched.
Short-circuiting: As soon as predicate.prune(&row_group_filters) rules out a row group, it breaks out of the loop and skips reading any remaining columns for that group.
State Machine Optimization: Dropped the BloomFiltersLoadedParquetOpen and PruneWithBloomFilters states entirely. Because we evaluate eagerly during I/O, we no longer need to retain all of those filters in memory for a secondary pass. The state machine now transitions cleaner directly to BuildStream.
Access Plan Updates: Added a mutable accessor access_plan_mut to RowGroupAccessPlanFilter to support skipping row groups directly from within the loader.
Are these changes tested?
Yes, this relies on existing Parquet pruning test suites which will validate that correctness is maintained while avoiding unnecessary reads.