[Parquet] Improve Auto RowSelection construction for scattered predicate results - #10852
Conversation
|
run benchmarks |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (d087f0e) to c62002b (merge-base) diff Run configurationrun benchmark sql_plannerBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench sql_planner File an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: `` Kubernetes messageFile an issue against this benchmark runner |
|
run benchmarks |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (d087f0e) to c62002b (merge-base) diff Run configurationrun benchmark sql_plannerBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench sql_planner File an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: `` Kubernetes messageFile an issue against this benchmark runner |
|
run benchmarks |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (d087f0e) to c62002b (merge-base) diff Run configurationrun benchmark sql_plannerBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench sql_planner File an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: `` Kubernetes messageFile an issue against this benchmark runner |
|
I'm not sure how to run whole bench in this PR, So I run them on my dev server, here's the main results: End-to-end benchmark resultsI ran three cyclic-order rounds against Arrow main using the same DataFusion integration harness. The first iteration of each query was excluded, and the table reports the cross-round median of steady-state query times.
Selected queries:
The candidate fixes the target Q25 regression, preserves the major pushdown wins such as ClickBench Q23, TPC-H Q18, and TPC-DS Q72, and reduces several other Main-ON regressions. It is not a pure monotonic improvement: regressions remain on queries such as ClickBench Q21 and TPC-DS Q14/Q50, and pushdown is still not universally faster than OFF. All query row counts matched across every arm and round. |
|
run benchmarks |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (4ea9518) to c62002b (merge-base) diff Run configurationrun benchmark sql_plannerBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench sql_planner File an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: `` Kubernetes messageFile an issue against this benchmark runner |
|
run benchmarks |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (4ea9518) to c62002b (merge-base) diff Run configurationrun benchmark sql_plannerBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench sql_planner File an issue against this benchmark runner |
|
Benchmark for this request failed before finishing (Kubernetes reason: Benchmarks requested: `` Kubernetes messageFile an issue against this benchmark runner |
|
run benchmark arrow_reader_clickbench |
|
🤖 Arrow criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/issue-10776-capped-auto-selection (4ea9518) to c62002b (merge-base) diff Run configurationrun benchmark arrow_reader_clickbenchBENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader_clickbench File an issue against this benchmark runner |
|
🤖 Arrow criterion benchmark completed (GKE) | trigger Instance: Comparing fix/issue-10776-capped-auto-selection (4ea9518) to c62002b (merge-base) diff Run configurationrun benchmark arrow_reader_clickbenchCPU Details (lscpu)Details
Resource Usagebase (merge-base)
branch
File an issue against this benchmark runner |
|
Thanks @hhhizzz, look good to me. This is a good improvement for the first-predicate path. One remaining limitation is that when the selection stays selector-backed, later predicates still fully materialize selectors via for the regression in TPC-DS Q14/Q50, maybe we can run the benchmark once again to check if this a noisy, because both Q14/Q50 are join-dominated, the pr should not affect so much. also i can help test, if you can share how to run the test . |
I have a bunch of local environment to compile the datafusion and arrow to do the bench. Let me run a few more time quickly. |
Thanks for pointing this out. I investigated the apparent regressions further. In a new four focused 10-iteration rounds, neither regression reproduced:
Profiles showed no new candidate hotspot, and the Q14/Q50 direct-output counters were zero. The earlier regressions were most likely caused by insufficient iterations and benchmark noise. I’ll treat that table as experimental context rather than PR-only evidence. |
|
Hi @alamb, Just wondering if you have time to take a look this week? |
|
Hi @hhhizzz -- I will try -- I am soryr I have been out last week |
Resolve the PageIndex API conflict by preserving both sets of test imports and passing the column count to the first-predicate page-skipping test. Validated with parquet all-targets/all-features Clippy, all-features library and arrow_reader tests (1572 passed), cargo fmt, and typos. Codex assisted with the test import and argument conflict resolution.
|
Hi @alamb, just following up on this PR. I've merged the latest main and resolved the conflicts, and the Parquet CI checks are passing. Would you have a chance to review it this week? Happy to provide any additional details or benchmarks. Thanks for your time! |
|
I am so sorry -- I will do it first thing tomorrow morning. Put it on my list |
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thank you @hhhizzz and @haohuaijin . I sincerely apologize for the delay in review -- I just needed to find enough contiguous time to review this carefully and such time is hard to come by when i was catching up from other things.
I went through this PR quite carefully and it makes sense to me, and I verified that it does indeed improve the performance of the reproducer in #10776
Here is my test script q24.sql
Details
set datafusion.execution.parquet.pushdown_filters = true;
CREATE EXTERNAL TABLE hits_raw STORED AS PARQUET LOCATION '/Users/andrewlamb/Software/datafusion/benchmarks/data/hits_partitioned/';
CREATE VIEW hits AS SELECT * EXCEPT ("EventDate"), CAST(CAST("EventDate" AS INTEGER) AS DATE) AS "EventDate" FROM hits_raw;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;Then I ran the query like this (5x, 75 runs total)
datafusion-cli -f q24.sql | grep ElapsedThe results look like this
| binary | min | max | avg | median |
|---|---|---|---|---|
| baseline (arrow-60, PR 25335) | 0.100 | 0.113 | 0.104 | 0.103 |
| adaptive-selector (+ this PR) | 0.089 | 0.107 | 0.093 | 0.092 |
Average change: -10.1% (negative = adaptive-selector faster). Median change: -10.7%.
Details
datafusion-cli-arrow-60 (baseline: apache/datafusion#25335, arrow/parquet 60.0.0)
| run | min | max | avg |
|---|---|---|---|
| 1 | 0.101 | 0.111 | 0.104 |
| 2 | 0.100 | 0.110 | 0.104 |
| 3 | 0.100 | 0.110 | 0.103 |
| 4 | 0.101 | 0.113 | 0.105 |
| 5 | 0.101 | 0.108 | 0.104 |
| all 5 runs (75 queries) | 0.100 | 0.113 | 0.104 |
datafusion-cli-adaptive-selector (baseline + #10852)
| run | min | max | avg |
|---|---|---|---|
| 1 | 0.090 | 0.099 | 0.094 |
| 2 | 0.090 | 0.107 | 0.094 |
| 3 | 0.090 | 0.102 | 0.092 |
| 4 | 0.090 | 0.100 | 0.094 |
| 5 | 0.089 | 0.099 | 0.093 |
| all 5 runs (75 queries) | 0.089 | 0.107 | 0.093 |
I left some suggested comments, but nothing I think is required
My agent also suggested we could use RowSelection::from_filters_auto in more places (example below). I will try this out locally and see if it makes any difference.
Details
diff --git a/parquet/src/arrow/arrow_reader/read_plan.rs b/parquet/src/arrow/arrow_reader/read_plan.rs
index 5ebfbc4890..2dda5ce618 100644
--- a/parquet/src/arrow/arrow_reader/read_plan.rs
+++ b/parquet/src/arrow/arrow_reader/read_plan.rs
@@ -273,12 +273,28 @@ impl ReadPlanBuilder {
return Ok(self);
}
let raw = match (self.selection.as_ref(), self.row_selection_policy) {
+ // A mask-backed prior selection stays mask-backed: `and_then` then
+ // takes the mask/mask path and never materializes selectors.
(Some(selection), _) if selection.as_mask().is_some() => {
RowSelection::from_filters_mask(&filters)
}
+ // Selector-backed prior selection (e.g. from page index pruning)
+ // under Auto: stop materializing selectors as soon as the predicate
+ // result is fragmented enough that Auto would choose a mask.
+ // `and_then` then streams the mask as runs, so the fragmented
+ // predicate result is never held as a second selector vector.
+ (Some(_selection), RowSelectionPolicy::Auto { threshold }) => {
+ RowSelection::from_filters_auto(&filters, threshold)
+ }
+ // No prior selection under Auto: the backing chosen here is the
+ // one `build` will resolve to, so later predicates and `build`
+ // never need to convert it.
(None, RowSelectionPolicy::Auto { threshold }) => {
RowSelection::from_filters_auto(&filters, threshold)
}
+ // The final backing is known to be a mask, so build it directly
+ // rather than materializing selectors that `build` would convert.
+ (None, RowSelectionPolicy::Mask) => RowSelection::from_filters_mask(&filters),
_ => RowSelection::from_filters(&filters),
};
self.selection = match self.selection.take() {| } | ||
| } | ||
|
|
||
| /// Returns the minimum normalized run count for which Auto prefers a mask. |
There was a problem hiding this comment.
| /// Returns the minimum normalized run count for which Auto prefers a mask. | |
| /// Returns the minimum normalized run count for which the Auto selection policy | |
| /// would prefer a mask to RLE. |
|
|
||
| /// Returns the minimum normalized run count for which Auto prefers a mask. | ||
| /// | ||
| /// This matches `total_rows < run_count.saturating_mul(threshold)`. For totals |
There was a problem hiding this comment.
What is the relevance of matching total_rows < run_count.saturating_mul(threshold)? It seems like this comment may be trying to explain how the code was changed in this PR but will not be relevant after this PR merges.
There was a problem hiding this comment.
Maybe it is trying to say that this implements the definition of the Auto policy threshold, namely that
"average selector length below which masks are preferred" and average run length below threshold means total_rows / run_count < threshold ?
| /// | ||
| /// This matches `total_rows < run_count.saturating_mul(threshold)`. For totals | ||
| /// below `usize::MAX`, the first matching run count is | ||
| /// `floor(total_rows / threshold) + 1`. `None` means no attainable run count |
There was a problem hiding this comment.
maybe this would be clearer if it said "Returns None when selectors are always a better choice"
|
Thanks @alamb for the careful review and for checking the performance improvement! I've applied the documentation suggestions in 8e1d1e4, clarifying the average-run-length threshold and when the policy prefers selectors. I'll try the additional construction-path changes you suggested shortly and verify their correctness and performance. |
FWIW I did not see any difference in my tests (I tried the clickbench queries locally with filter pushdown turned on( |
|
Thank you again @hhhizzz and @haohuaijin -- and I really apologize for the delay. I am slowly working through my backlog and I hope to be able to review #10859 later this afternoon |
…ate results (apache#10852) # Which issue does this PR close? - Closes apache#10776. # Rationale for this change For the first predicate, when there is no existing `RowSelection`, `ReadPlanBuilder::with_predicate_options` currently materializes all predicate results as RLE selectors. `RowSelectionPolicy::Auto` may then decide that the selection is too fragmented, convert those selectors back to a bitmap, and use the mask strategy. For scattered selections such as ClickBench Q25, this creates a large temporary `Vec<RowSelector>` even though the final strategy becomes certain as soon as the normalized run count crosses Auto's threshold. Keeping the first fragmented predicate mask-backed also benefits subsequent predicates: they use the existing mask construction path and mask conjunction instead of rebuilding another large selector vector. # What changes are included in this PR? - Add an internal `RowSelection::from_filters_auto` constructor that builds normalized selectors only while the selector strategy is still possible. - Share one `auto_min_mask_runs` helper between selector-backed strategy resolution, mask-backed strategy resolution, and capped construction. This keeps the strict comparison, threshold `0`/`1`, and saturating overflow behavior in one place. - Once the minimum mask run count is reached, drop the partial selector allocation before constructing the mask directly from the predicate `BooleanArray`s. - Fall back to the existing `from_filters` path when no attainable run count can select Mask, avoiding a per-selector cutoff check for threshold `0` and `1`. - Add `from_filters_mask` as the semantic mask constructor, including a single-filter fast path that reuses the filter's `BooleanBuffer`. - Use capped construction only for a first predicate with no pre-existing selection and `RowSelectionPolicy::Auto`. Existing selections and explicit Mask/Selectors policies retain their existing paths. - Preserve unresolved Auto in `prepare_selection_for_page_skipping` when there is no selection yet. There are no selection-driven pages to skip in that state, and resolving early would force Selectors before the first predicate result is available. # Are these changes tested? Yes. Correctness coverage includes: - deterministic `31`/`32` run-length threshold boundaries and shared-helper boundary checks; - cross-filter run merging, empty filters, and trailing skips; - eight edge row counts crossed with all eight thresholds and four deterministic selection patterns; - a separate 512-case fixed-seed matrix spanning eight thresholds, eight selectivities, four named run shapes, aligned and non-byte-aligned buffers, and multiple `BooleanArray` splits; - LIMIT/padding behavior and async page-skipping preparation. Focused construction benchmark over 4,194,304 rows, split into 512 `BooleanArray`s of 8,192 rows each (median of three Criterion rounds): | Shape | Previous Auto | Capped Auto | Change | |---|---:|---:|---:| | Q25-like 15% scattered | 14.675 ms | 506.9 µs | 28.95x faster | | Alternating run-1 | 48.595 ms | 513.8 µs | 94.57x faster | | Exact run-32 boundary | 431.4 µs | 445.9 µs | +3.35% | | Clustered run-128 | 165.8 µs | 154.5 µs | -6.77% | | Sparse run-32 | 66.2 µs | 65.5 µs | -1.04% | An unconditional mask-first implementation was rejected because it regressed selector-friendly run-128 by 30.7% and sparse input by 107.6%. Three paired async-reader rounds with PageIndex disabled showed: - Q25-like: -26.02% - alternating run-1: -48.91% - run-32: -1.87% - run-128: -0.29% - sparse: -0.30% - all-selected: +1.45% The focused construction harness was kept separate in accordance with the repository guidance for benchmark additions. These results isolate selection construction and async reader behavior; they are not presented as a full ClickBench Q25 wall-time measurement. Local validation: - `cargo fmt --all -- --check`: passed - `cargo test -p parquet --lib -- --skip file::writer::tests::test_int96_interop`: 1,305 passed, 0 failed, 1 filtered - `cargo test -p parquet --test arrow_reader --features async -- --skip bad_data::test_invalid_files`: 123 passed, 0 failed, 1 ignored, 1 filtered - `cargo clippy -p parquet --all-targets --all-features -- -D warnings`: passed The two filtered tests require fixtures absent from the local `parquet-testing` checkout (`int96_timestamp_order.parquet` and the `bad_data/variants` fixture). The PR's GitHub `parquet` test, compilation, and Clippy checks pass without filtering. # Are there any user-facing changes? No public interface changes. The new constructors and cutoff helper are crate-private, and logical row-selection semantics and explicit policy behavior are unchanged.
6 participants
Which issue does this PR close?
ReadPlanBuilder::with_predicate_options#10776.Rationale for this change
For the first predicate, when there is no existing
RowSelection,ReadPlanBuilder::with_predicate_optionscurrently materializes all predicate results as RLE selectors.RowSelectionPolicy::Automay then decide that the selection is too fragmented, convert those selectors back to a bitmap, and use the mask strategy.For scattered selections such as ClickBench Q25, this creates a large temporary
Vec<RowSelector>even though the final strategy becomes certain as soon as the normalized run count crosses Auto's threshold.Keeping the first fragmented predicate mask-backed also benefits subsequent predicates: they use the existing mask construction path and mask conjunction instead of rebuilding another large selector vector.
What changes are included in this PR?
RowSelection::from_filters_autoconstructor that builds normalized selectors only while the selector strategy is still possible.auto_min_mask_runshelper between selector-backed strategy resolution, mask-backed strategy resolution, and capped construction. This keeps the strict comparison, threshold0/1, and saturating overflow behavior in one place.BooleanArrays.from_filterspath when no attainable run count can select Mask, avoiding a per-selector cutoff check for threshold0and1.from_filters_maskas the semantic mask constructor, including a single-filter fast path that reuses the filter'sBooleanBuffer.RowSelectionPolicy::Auto. Existing selections and explicit Mask/Selectors policies retain their existing paths.prepare_selection_for_page_skippingwhen there is no selection yet. There are no selection-driven pages to skip in that state, and resolving early would force Selectors before the first predicate result is available.Are these changes tested?
Yes.
Correctness coverage includes:
31/32run-length threshold boundaries and shared-helper boundary checks;BooleanArraysplits;Focused construction benchmark over 4,194,304 rows, split into 512
BooleanArrays of 8,192 rows each (median of three Criterion rounds):An unconditional mask-first implementation was rejected because it regressed selector-friendly run-128 by 30.7% and sparse input by 107.6%.
Three paired async-reader rounds with PageIndex disabled showed:
The focused construction harness was kept separate in accordance with the repository guidance for benchmark additions. These results isolate selection construction and async reader behavior; they are not presented as a full ClickBench Q25 wall-time measurement.
Local validation:
cargo fmt --all -- --check: passedcargo test -p parquet --lib -- --skip file::writer::tests::test_int96_interop: 1,305 passed, 0 failed, 1 filteredcargo test -p parquet --test arrow_reader --features async -- --skip bad_data::test_invalid_files: 123 passed, 0 failed, 1 ignored, 1 filteredcargo clippy -p parquet --all-targets --all-features -- -D warnings: passedThe two filtered tests require fixtures absent from the local
parquet-testingcheckout (int96_timestamp_order.parquetand thebad_data/variantsfixture). The PR's GitHubparquettest, compilation, and Clippy checks pass without filtering.Are there any user-facing changes?
No public interface changes. The new constructors and cutoff helper are crate-private, and logical row-selection semantics and explicit policy behavior are unchanged.