You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "haohuaijin (via GitHub)" <gi...@apache.org> on 2026/08/26 15:23:24 UTC
[PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
haohuaijin opened a new pull request, #10859:
URL: https://github.com/apache/arrow-rs/pull/10859
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax.
-->
- Closes #NNN.
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
If this PR claims a performance improvement, please include evidence such as benchmark results.
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4111100827
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> Or maybe we should change the ArrowPredicate API to return a RowSelection (with a default implementation that returns the BooleanArray 🤔 )
I just tested it, and the change is a bit too big. and I don’t find much difference in performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968, the #10968 optimizer need the outer selection > 75%).
but i agree use `RowSelection` inside a `ArrowPredicate` is really strange.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740894857
🤖 Arrow criterion benchmark running (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880198)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5740880198-2482-6mbxg 6.12.94+ #1 SMP Tue Aug 4 08:44:15 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing parquet-same-projection-filter-fusion (916b65c70ff1cef85ebb490f7e21a960d2f903a3) to a7b89dd (merge-base) [diff](https://github.com/apache/arrow-rs/compare/a7b89dd34eb6bd17e99bdc61a60b37a7cc08f166..916b65c70ff1cef85ebb490f7e21a960d2f903a3)
<details><summary>Run configuration</summary>
```yaml
run benchmark arrow_reader_predicate_fusion
```
</details>
BENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader_predicate_fusion
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4129245889
##########
parquet/src/arrow/arrow_reader/read_plan.rs:
##########
@@ -157,17 +157,7 @@ impl ReadPlanBuilder {
///
/// Guarantees to return either `Selectors` or `Mask`, never `Auto`.
pub(crate) fn resolve_selection_strategy(&self) -> RowSelectionStrategy {
- match self.row_selection_policy {
- RowSelectionPolicy::Selectors => RowSelectionStrategy::Selectors,
- RowSelectionPolicy::Mask => RowSelectionStrategy::Mask,
- RowSelectionPolicy::Auto { threshold, .. } => {
- let Some(selection) = self.selection.as_ref() else {
- return RowSelectionStrategy::Selectors;
- };
-
- selection.auto_selection_strategy(threshold)
- }
- }
+ self.row_selection_policy.resolve(self.selection.as_ref())
Review Comment:
sure, i will do this as a follow up
##########
parquet/src/arrow/arrow_reader/read_plan.rs:
##########
@@ -157,17 +157,7 @@ impl ReadPlanBuilder {
///
/// Guarantees to return either `Selectors` or `Mask`, never `Auto`.
pub(crate) fn resolve_selection_strategy(&self) -> RowSelectionStrategy {
- match self.row_selection_policy {
- RowSelectionPolicy::Selectors => RowSelectionStrategy::Selectors,
- RowSelectionPolicy::Mask => RowSelectionStrategy::Mask,
- RowSelectionPolicy::Auto { threshold, .. } => {
- let Some(selection) = self.selection.as_ref() else {
- return RowSelectionStrategy::Selectors;
- };
-
- selection.auto_selection_strategy(threshold)
- }
- }
+ self.row_selection_policy.resolve(self.selection.as_ref())
Review Comment:
okay, i will do this as a follow up
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5882906989
Thanks again @alamb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5892109281
Thanks @haohuaijin
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5892107366
I ran this and confirmed that for Q25 the performance of pushdown vs not pushdown is now the same (maybe even slightly faster) 🎉
ClickBench q25 on `hits_partitioned`, local run, 10 iterations per phase (ms):
| Binary | Pushdown | Min | Max | Avg |
|---|---|---:|---:|---:|
| merge-base (991fd23) | off | 76 | 85 | 78.5 |
| merge-base (991fd23) | on | 100 | 115 | 106.4 |
| PR 10859 | off | 77 | 86 | 80.3 |
| PR 10859 | on | 76 | 82 | 78.3 |
<details>
Test script: [q25.sql](https://github.com/user-attachments/files/32807242/q25.sql)
Run with this PR (via https://github.com/apache/datafusion/pull/25832)
```shell
andrewlamb@Andrews-MacBook-Pro-3:~/Downloads/10859$ ../datafusion-cli-test-arrow-pr-10859 -f q25.sql | grep Elapsed
Elapsed 0.001 seconds.
Elapsed 0.035 seconds.
Elapsed 0.001 seconds.
| ===== Elapsed lines below: PUSHDOWN OFF (q25) ===== |
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.083 seconds.
Elapsed 0.079 seconds.
Elapsed 0.077 seconds.
Elapsed 0.085 seconds.
Elapsed 0.086 seconds.
Elapsed 0.080 seconds.
Elapsed 0.078 seconds.
Elapsed 0.078 seconds.
Elapsed 0.079 seconds.
Elapsed 0.078 seconds.
| ===== Elapsed lines below: PUSHDOWN ON (q25) ===== |
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.077 seconds.
Elapsed 0.076 seconds.
Elapsed 0.077 seconds.
Elapsed 0.078 seconds.
Elapsed 0.080 seconds.
Elapsed 0.077 seconds.
Elapsed 0.078 seconds.
Elapsed 0.078 seconds.
Elapsed 0.080 seconds.
Elapsed 0.082 seconds.
```
Run with merge-base of the PR
```shell
andrewlamb@Andrews-MacBook-Pro-3:~/Downloads/10859$ ../datafusion-cli.991fd23dd0be0046af5945b8c3905612859b657a -f q25.sql | grep Elapsed
Elapsed 0.001 seconds.
Elapsed 0.035 seconds.
Elapsed 0.001 seconds.
| ===== Elapsed lines below: PUSHDOWN OFF (q25) ===== |
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.082 seconds.
Elapsed 0.085 seconds.
Elapsed 0.079 seconds.
Elapsed 0.077 seconds.
Elapsed 0.077 seconds.
Elapsed 0.077 seconds.
Elapsed 0.077 seconds.
Elapsed 0.077 seconds.
Elapsed 0.078 seconds.
Elapsed 0.076 seconds.
| ===== Elapsed lines below: PUSHDOWN ON (q25) ===== |
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.000 seconds.
Elapsed 0.100 seconds.
Elapsed 0.100 seconds.
Elapsed 0.102 seconds.
Elapsed 0.103 seconds.
Elapsed 0.110 seconds.
Elapsed 0.108 seconds.
Elapsed 0.115 seconds.
Elapsed 0.108 seconds.
Elapsed 0.112 seconds.
Elapsed 0.106 seconds.
```
</details>
Also I looked at q35,q36 and q39 that the bot may have shown slightly slower, but could not reproduce it locally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740947504
🤖 Arrow criterion benchmark completed (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880642)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing parquet-same-projection-filter-fusion (916b65c70ff1cef85ebb490f7e21a960d2f903a3) to a7b89dd (merge-base) [diff](https://github.com/apache/arrow-rs/compare/a7b89dd34eb6bd17e99bdc61a60b37a7cc08f166..916b65c70ff1cef85ebb490f7e21a960d2f903a3)
<details><summary>Run configuration</summary>
```yaml
run benchmark arrow_reader_predicate_fusion
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
group main parquet-same-projection-filter-fusion
----- ---- -------------------------------------
same_projection_filter/int64/clustered/cached/4/all99 1.13 2.4±0.01ms 106.3 MElem/sec 1.00 2.1±0.01ms 120.6 MElem/sec
same_projection_filter/int64/clustered/uncached/2/all50 1.00 2.8±0.01ms 88.0 MElem/sec 1.00 2.9±0.01ms 87.7 MElem/sec
same_projection_filter/int64/clustered/uncached/4/all50 1.00 3.1±0.01ms 81.5 MElem/sec 1.01 3.1±0.01ms 81.0 MElem/sec
same_projection_filter/int64/clustered/uncached/4/all99 1.08 4.3±0.02ms 58.5 MElem/sec 1.00 4.0±0.02ms 63.3 MElem/sec
same_projection_filter/int64/fragmented/cached/1/all99 1.01 3.0±0.02ms 84.4 MElem/sec 1.00 2.9±0.02ms 85.4 MElem/sec
same_projection_filter/int64/fragmented/cached/2/all50 1.06 4.5±0.04ms 56.2 MElem/sec 1.00 4.2±0.03ms 59.7 MElem/sec
same_projection_filter/int64/fragmented/cached/2/all99 1.05 3.9±0.03ms 64.4 MElem/sec 1.00 3.7±0.04ms 67.5 MElem/sec
same_projection_filter/int64/fragmented/cached/4/all99 1.00 5.9±0.06ms 42.3 MElem/sec 1.02 6.1±0.05ms 41.3 MElem/sec
same_projection_filter/int64/fragmented/cached/4/early1 1.12 3.4±0.02ms 73.7 MElem/sec 1.00 3.0±0.01ms 82.6 MElem/sec
same_projection_filter/int64/fragmented/cached/4/late1 1.03 5.5±0.03ms 45.1 MElem/sec 1.00 5.4±0.03ms 46.6 MElem/sec
same_projection_filter/int64/fragmented/uncached/1/all99 1.00 5.0±0.03ms 49.7 MElem/sec 1.01 5.1±0.03ms 49.3 MElem/sec
same_projection_filter/int64/fragmented/uncached/2/all50 1.36 8.3±0.04ms 30.2 MElem/sec 1.00 6.1±0.07ms 41.2 MElem/sec
same_projection_filter/int64/fragmented/uncached/2/all99 1.41 7.9±0.03ms 31.7 MElem/sec 1.00 5.6±0.03ms 44.5 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/all99 1.73 13.9±0.06ms 18.0 MElem/sec 1.00 8.0±0.04ms 31.1 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/early1 2.23 10.6±0.06ms 23.7 MElem/sec 1.00 4.7±0.03ms 52.9 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/late1 1.90 13.3±0.08ms 18.7 MElem/sec 1.00 7.0±0.04ms 35.5 MElem/sec
same_projection_filter/selection_boundary/int64/uncached/4/run16 2.20 11.9±0.06ms 21.0 MElem/sec 1.00 5.4±0.04ms 46.1 MElem/sec
same_projection_filter/selection_boundary/int64/uncached/4/run64 2.08 10.7±0.05ms 23.3 MElem/sec 1.00 5.2±0.04ms 48.4 MElem/sec
same_projection_filter/string/clustered/cached/4/all99 1.34 8.2±0.28ms 30.3 MElem/sec 1.00 6.2±0.11ms 40.6 MElem/sec
same_projection_filter/string/clustered/uncached/2/all50 1.00 4.9±0.02ms 51.2 MElem/sec 1.07 5.2±0.03ms 47.7 MElem/sec
same_projection_filter/string/clustered/uncached/4/all50 1.00 5.9±0.02ms 42.7 MElem/sec 1.03 6.0±0.04ms 41.6 MElem/sec
same_projection_filter/string/clustered/uncached/4/all99 1.36 9.3±0.02ms 26.8 MElem/sec 1.00 6.8±0.02ms 36.6 MElem/sec
same_projection_filter/string/fragmented/cached/1/all99 1.00 35.4±0.54ms 7.1 MElem/sec 1.04 36.9±0.73ms 6.8 MElem/sec
same_projection_filter/string/fragmented/cached/2/all50 1.08 38.6±0.85ms 6.5 MElem/sec 1.00 35.8±4.76ms 7.0 MElem/sec
same_projection_filter/string/fragmented/cached/2/all99 1.00 37.4±0.85ms 6.7 MElem/sec 1.03 38.6±0.28ms 6.5 MElem/sec
same_projection_filter/string/fragmented/cached/4/all99 1.09 42.1±0.89ms 5.9 MElem/sec 1.00 38.7±3.81ms 6.5 MElem/sec
same_projection_filter/string/fragmented/uncached/1/all99 1.01 26.6±0.16ms 9.4 MElem/sec 1.00 26.5±0.08ms 9.4 MElem/sec
same_projection_filter/string/fragmented/uncached/2/all50 1.84 51.7±0.55ms 4.8 MElem/sec 1.00 28.1±0.13ms 8.9 MElem/sec
same_projection_filter/string/fragmented/uncached/2/all99 1.79 49.9±0.33ms 5.0 MElem/sec 1.00 27.8±0.13ms 9.0 MElem/sec
same_projection_filter/string/fragmented/uncached/4/all99 3.05 98.8±0.59ms 2.5 MElem/sec 1.00 32.5±0.20ms 7.7 MElem/sec
```
</p>
</details>
<details><summary>Resource Usage</summary>
**base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 310.1s |
| Peak memory | 50.3 MiB |
| Avg memory | 37.6 MiB |
| CPU user | 293.2s |
| CPU sys | 11.9s |
| Peak spill | 0 B |
**branch**
| Metric | Value |
|--------|-------|
| Wall time | 305.1s |
| Peak memory | 51.9 MiB |
| Avg memory | 37.8 MiB |
| CPU user | 289.0s |
| CPU sys | 11.4s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968, because the outer selection should > 75%).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4104718800
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -153,7 +158,10 @@ where
/// This design has a couple of implications:
///
/// * [`RowFilter`] can be used to skip entire pages, and thus IO, in addition to CPU decode overheads
-/// * Columns may be decoded multiple times if they appear in multiple [`ProjectionMask`]
+/// * Columns may be decoded multiple times if they appear in multiple [`ProjectionMask`].
Review Comment:
👍
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
I don't fully understand the need for a row_selection_policy here. I was expecting that when a predicate was fused, it would basically evaluate `pred1 AND pred2 AND ...`, combining the results with `and` -- or more likely the `&=` operator that knows how to reuse the bitmaps
It is really strange to me to try and use the RowSelection API *inside* a `ArrowPredicate` whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
Or maybe we should change the ArrowPredicate API to return a RowSelection (with a default implementation that returns the BooleanArray 🤔 )
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
+ let filter = evaluate_predicate(predicate.as_mut(), filtered_batch.clone())?;
+ // No mapping is needed if all preceding predicates accepted every row.
+ if idx == last_predicate_idx && selection.is_none() {
+ return Ok(filter);
+ }
+ let true_count = filter.true_count();
+ if true_count == 0 {
+ return Ok(BooleanArray::new(BooleanBuffer::new_unset(num_rows), None));
+ }
+ if true_count == filter.len() {
+ continue;
+ }
+
+ let predicate_selection = RowSelection::from_boolean_buffer(filter.values().clone());
+ selection = Some(match selection.take() {
+ // Only the accumulated selection drives the composition
+ // algorithm, so adapt it once, right before it is used.
+ Some(prev) => adapt_fusion_selection(prev, self.row_selection_policy)
+ .and_then(&predicate_selection),
+ None => predicate_selection,
+ });
+ if idx != last_predicate_idx {
+ filtered_batch = narrow_batch(&filtered_batch, &filter, true_count)?;
+ }
+ }
+
+ let mask = match selection {
+ Some(selection) => selection.into_boolean_buffer(),
+ None => BooleanBuffer::new_set(num_rows),
+ };
+ debug_assert_eq!(mask.len(), num_rows);
+ Ok(BooleanArray::new(mask, None))
+ }
+}
+
+/// Apply the reader's row selection policy to a fusion selection.
+fn adapt_fusion_selection(selection: RowSelection, policy: RowSelectionPolicy) -> RowSelection {
Review Comment:
This feels very similar to [ReadPlanBuilder::resolve_selection_strategy](https://github.com/apache/arrow-rs/blob/d506e145f42169e9c188b627f341da499bf09859/parquet/src/arrow/arrow_reader/read_plan.rs#L159-L171) and [build_cursor](https://github.com/apache/arrow-rs/blob/d506e145f42169e9c188b627f341da499bf09859/parquet/src/arrow/arrow_reader/read_plan.rs#L320-L339)
would it be possible to re-use that logic here? For example by making `RowSelectionStrategy::build_cursor(selection)` or adding some way to apply a row selection to an array 🤔
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
Review Comment:
we could potentially scan through the predicates first and find if any needed to be fused before allocating, though I am not sure that matters
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740947315
🤖 Arrow criterion benchmark completed (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880198)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing parquet-same-projection-filter-fusion (916b65c70ff1cef85ebb490f7e21a960d2f903a3) to a7b89dd (merge-base) [diff](https://github.com/apache/arrow-rs/compare/a7b89dd34eb6bd17e99bdc61a60b37a7cc08f166..916b65c70ff1cef85ebb490f7e21a960d2f903a3)
<details><summary>Run configuration</summary>
```yaml
run benchmark arrow_reader_predicate_fusion
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
group main parquet-same-projection-filter-fusion
----- ---- -------------------------------------
same_projection_filter/int64/clustered/cached/4/all99 1.15 2.4±0.00ms 105.4 MElem/sec 1.00 2.1±0.00ms 121.0 MElem/sec
same_projection_filter/int64/clustered/uncached/2/all50 1.00 2.8±0.02ms 89.0 MElem/sec 1.02 2.9±0.01ms 87.3 MElem/sec
same_projection_filter/int64/clustered/uncached/4/all50 1.00 3.1±0.01ms 81.3 MElem/sec 1.00 3.1±0.01ms 81.7 MElem/sec
same_projection_filter/int64/clustered/uncached/4/all99 1.06 4.2±0.01ms 59.5 MElem/sec 1.00 3.9±0.01ms 63.3 MElem/sec
same_projection_filter/int64/fragmented/cached/1/all99 1.00 2.9±0.00ms 86.6 MElem/sec 1.01 2.9±0.01ms 86.0 MElem/sec
same_projection_filter/int64/fragmented/cached/2/all50 1.07 4.4±0.02ms 56.8 MElem/sec 1.00 4.1±0.01ms 60.8 MElem/sec
same_projection_filter/int64/fragmented/cached/2/all99 1.04 3.8±0.02ms 65.1 MElem/sec 1.00 3.7±0.02ms 67.8 MElem/sec
same_projection_filter/int64/fragmented/cached/4/all99 1.00 5.8±0.01ms 42.9 MElem/sec 1.02 5.9±0.01ms 42.1 MElem/sec
same_projection_filter/int64/fragmented/cached/4/early1 1.12 3.3±0.01ms 74.9 MElem/sec 1.00 3.0±0.02ms 83.8 MElem/sec
same_projection_filter/int64/fragmented/cached/4/late1 1.04 5.5±0.02ms 45.1 MElem/sec 1.00 5.3±0.01ms 47.1 MElem/sec
same_projection_filter/int64/fragmented/uncached/1/all99 1.00 5.0±0.03ms 49.8 MElem/sec 1.00 5.0±0.03ms 50.0 MElem/sec
same_projection_filter/int64/fragmented/uncached/2/all50 1.36 8.3±0.04ms 30.3 MElem/sec 1.00 6.1±0.02ms 41.3 MElem/sec
same_projection_filter/int64/fragmented/uncached/2/all99 1.40 7.9±0.03ms 31.7 MElem/sec 1.00 5.6±0.01ms 44.5 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/all99 1.76 13.7±0.06ms 18.3 MElem/sec 1.00 7.8±0.01ms 32.1 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/early1 2.21 10.6±0.04ms 23.7 MElem/sec 1.00 4.8±0.01ms 52.2 MElem/sec
same_projection_filter/int64/fragmented/uncached/4/late1 1.88 13.3±0.02ms 18.8 MElem/sec 1.00 7.1±0.02ms 35.4 MElem/sec
same_projection_filter/selection_boundary/int64/uncached/4/run16 2.17 11.7±0.06ms 21.3 MElem/sec 1.00 5.4±0.01ms 46.2 MElem/sec
same_projection_filter/selection_boundary/int64/uncached/4/run64 2.07 10.6±0.02ms 23.7 MElem/sec 1.00 5.1±0.02ms 49.1 MElem/sec
same_projection_filter/string/clustered/cached/4/all99 1.29 7.8±0.13ms 32.0 MElem/sec 1.00 6.0±0.04ms 41.4 MElem/sec
same_projection_filter/string/clustered/uncached/2/all50 1.00 4.9±0.01ms 51.2 MElem/sec 1.04 5.1±0.03ms 49.1 MElem/sec
same_projection_filter/string/clustered/uncached/4/all50 1.00 5.9±0.02ms 42.5 MElem/sec 1.01 6.0±0.04ms 42.0 MElem/sec
same_projection_filter/string/clustered/uncached/4/all99 1.40 9.3±0.03ms 26.7 MElem/sec 1.00 6.7±0.01ms 37.4 MElem/sec
same_projection_filter/string/fragmented/cached/1/all99 1.00 33.5±0.65ms 7.5 MElem/sec 1.02 34.3±0.86ms 7.3 MElem/sec
same_projection_filter/string/fragmented/cached/2/all50 1.23 37.5±0.92ms 6.7 MElem/sec 1.00 30.6±3.05ms 8.2 MElem/sec
same_projection_filter/string/fragmented/cached/2/all99 1.00 36.1±0.81ms 6.9 MElem/sec 1.06 38.2±0.22ms 6.5 MElem/sec
same_projection_filter/string/fragmented/cached/4/all99 1.08 40.1±0.87ms 6.2 MElem/sec 1.00 37.0±4.47ms 6.8 MElem/sec
same_projection_filter/string/fragmented/uncached/1/all99 1.00 26.0±0.15ms 9.6 MElem/sec 1.00 26.0±0.15ms 9.6 MElem/sec
same_projection_filter/string/fragmented/uncached/2/all50 1.81 50.5±0.24ms 5.0 MElem/sec 1.00 27.8±0.16ms 9.0 MElem/sec
same_projection_filter/string/fragmented/uncached/2/all99 1.79 49.2±0.28ms 5.1 MElem/sec 1.00 27.5±0.11ms 9.1 MElem/sec
same_projection_filter/string/fragmented/uncached/4/all99 3.11 98.2±0.44ms 2.5 MElem/sec 1.00 31.6±0.16ms 7.9 MElem/sec
```
</p>
</details>
<details><summary>Resource Usage</summary>
**base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 315.1s |
| Peak memory | 50.9 MiB |
| Avg memory | 37.8 MiB |
| CPU user | 301.2s |
| CPU sys | 10.9s |
| Peak spill | 0 B |
**branch**
| Metric | Value |
|--------|-------|
| Wall time | 300.1s |
| Peak memory | 50.6 MiB |
| Avg memory | 39.3 MiB |
| CPU user | 287.4s |
| CPU sys | 9.0s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5731725301
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 `parquet/src/arrow/arrow_reader/filter.rs`.
I'd like to know whether you think this is a good direction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880642
run benchmark arrow_reader_predicate_fusion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968, the #10968 optimizer need the outer selection > 75%).
but i agree use `RowSelection` inside a `ArrowPredicate` is really strange. If we can further optimize `and_then_mask` to avoid these regressions, we could just use `BooleanArray` throughout the fused predicate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4111100827
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> Or maybe we should change the ArrowPredicate API to return a RowSelection (with a default implementation that returns the BooleanArray 🤔 )
I just tested it, and the change is a bit too big. But I don’t find much difference in performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110694724
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> I don't fully understand the need for a row_selection_policy here. I was expecting that when a predicate was fused, it would basically evaluate pred1 AND pred2 AND ..., combining the results with and -- or more likely the &= operator that knows how to reuse the bitmaps
there two reason:
1. it is necessary to preserve `RowFilter's` original layer-by-layer filtering semantics: subsequent predicates process only the rows remaining after previous filtering. For example, if the first predicate is `a > 0`, and the second contains a `1 / a `expression that may error on zero values, and both predicates are evaluated on the full batch and the results are then ANDed, the second predicate may error on zero values that it would otherwise never receive.
2. Avoid evaluating expensive downstream filters when the first filter in a fused filter chain is selective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4123679478
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +205,477 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(self, parquet_schema: &SchemaDescriptor) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ /// Chooses how intermediate selections are composed. The output is always
+ /// a [`BooleanArray`], so this is independent of the reader's
+ /// [`RowSelectionPolicy`], which governs decoding.
+ composition_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(predicates: Vec<Box<dyn ArrowPredicate>>) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ composition_policy: RowSelectionPolicy::default(),
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
+ let filter = evaluate_predicate(predicate.as_mut(), filtered_batch.clone())?;
+ // No mapping is needed if all preceding predicates accepted every row.
+ if idx == last_predicate_idx && selection.is_none() {
+ return Ok(filter);
+ }
+ let true_count = filter.true_count();
+ if true_count == 0 {
+ return Ok(BooleanArray::new(BooleanBuffer::new_unset(num_rows), None));
+ }
+ if true_count == filter.len() {
+ continue;
+ }
+
+ let predicate_selection = RowSelection::from_boolean_buffer(filter.values().clone());
+ selection = Some(match selection.take() {
+ // Only the accumulated selection drives the composition
+ // algorithm, so adapt it once, right before it is used.
+ Some(prev) => self
+ .composition_policy
+ .apply(prev)
+ .and_then(&predicate_selection),
+ None => predicate_selection,
+ });
+ if idx != last_predicate_idx {
+ filtered_batch = narrow_batch(&filtered_batch, &filter, true_count)?;
+ }
+ }
+
+ let mask = match selection {
+ Some(selection) => selection.into_boolean_buffer(),
+ None => BooleanBuffer::new_set(num_rows),
+ };
+ debug_assert_eq!(mask.len(), num_rows);
+ Ok(BooleanArray::new(mask, None))
+ }
+}
+
+/// Validate the predicate result length and treat nulls as false.
+fn evaluate_predicate(
+ predicate: &mut dyn ArrowPredicate,
+ batch: RecordBatch,
+) -> Result<BooleanArray, ArrowError> {
+ let input_rows = batch.num_rows();
+ let filter = predicate.evaluate(batch)?;
+ if filter.len() != input_rows {
+ return Err(ArrowError::InvalidArgumentError(format!(
+ "ArrowPredicate predicate returned {} rows, expected {input_rows}",
+ filter.len()
+ )));
+ }
+ Ok(match filter.null_count() {
+ 0 => filter,
+ _ => prep_null_mask_filter(&filter),
+ })
+}
+
+/// Restrict `batch` to the rows `filter` accepts, slicing zero-copy when they
+/// form one contiguous range and compacting otherwise.
+///
+/// `filter` must be null-free and accept some but not all rows.
+fn narrow_batch(
+ batch: &RecordBatch,
+ filter: &BooleanArray,
+ true_count: usize,
+) -> Result<RecordBatch, ArrowError> {
+ let mut slices = SlicesIterator::new(filter);
Review Comment:
I think we could potentially add this optimization to `filter` and then just use `filter_record_batch`:
- Filed https://github.com/apache/arrow-rs/issues/11255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5560107625
the benchmark result of `arrow_reader_predicate_fusion` from #11007, `main` dbd24cc95 vs this PR rebased on it:
- `uncached` chains gain the most, up to 3.6x (`string/fragmented/uncached/4/*`), since `main` decodes the filter column once per predicate.
- `cached` chains gain 3% to 46%, where `main` only replays the predicate cache.
- Three cases with 50% survivors are 3% to 6% slower, where compacting survivors costs more than the saved decode.
<img width="1526" height="1822" alt="benchmark result" src="https://github.com/user-attachments/assets/70a918a9-fe50-4003-8110-d8367c71469b" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4114360714
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
Updated in [2f81d70](https://github.com/apache/arrow-rs/pull/10859/commits/2f81d70b4f3208c78424246ecbfeff76e28426b2): `FusedPredicate` no longer uses the reader's `RowSelectionPolicy`; it always uses the default `Auto` heuristic, since the policy only affects how intermediate selections are composed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110204003
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
+ let filter = evaluate_predicate(predicate.as_mut(), filtered_batch.clone())?;
+ // No mapping is needed if all preceding predicates accepted every row.
+ if idx == last_predicate_idx && selection.is_none() {
+ return Ok(filter);
+ }
+ let true_count = filter.true_count();
+ if true_count == 0 {
+ return Ok(BooleanArray::new(BooleanBuffer::new_unset(num_rows), None));
+ }
+ if true_count == filter.len() {
+ continue;
+ }
+
+ let predicate_selection = RowSelection::from_boolean_buffer(filter.values().clone());
+ selection = Some(match selection.take() {
+ // Only the accumulated selection drives the composition
+ // algorithm, so adapt it once, right before it is used.
+ Some(prev) => adapt_fusion_selection(prev, self.row_selection_policy)
+ .and_then(&predicate_selection),
+ None => predicate_selection,
+ });
+ if idx != last_predicate_idx {
+ filtered_batch = narrow_batch(&filtered_batch, &filter, true_count)?;
+ }
+ }
+
+ let mask = match selection {
+ Some(selection) => selection.into_boolean_buffer(),
+ None => BooleanBuffer::new_set(num_rows),
+ };
+ debug_assert_eq!(mask.len(), num_rows);
+ Ok(BooleanArray::new(mask, None))
+ }
+}
+
+/// Apply the reader's row selection policy to a fusion selection.
+fn adapt_fusion_selection(selection: RowSelection, policy: RowSelectionPolicy) -> RowSelection {
Review Comment:
update in [6984860](https://github.com/apache/arrow-rs/pull/10859/commits/6984860b5dc44453043189d00d49d4fe239ff82c)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968, the #10968 optimizer need the outer selection should > 75%).
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968, the #10968 optimizer need the outer selection > 75%).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4129245889
##########
parquet/src/arrow/arrow_reader/read_plan.rs:
##########
@@ -157,17 +157,7 @@ impl ReadPlanBuilder {
///
/// Guarantees to return either `Selectors` or `Mask`, never `Auto`.
pub(crate) fn resolve_selection_strategy(&self) -> RowSelectionStrategy {
- match self.row_selection_policy {
- RowSelectionPolicy::Selectors => RowSelectionStrategy::Selectors,
- RowSelectionPolicy::Mask => RowSelectionStrategy::Mask,
- RowSelectionPolicy::Auto { threshold, .. } => {
- let Some(selection) = self.selection.as_ref() else {
- return RowSelectionStrategy::Selectors;
- };
-
- selection.auto_selection_strategy(threshold)
- }
- }
+ self.row_selection_policy.resolve(self.selection.as_ref())
Review Comment:
yes, i will do this as a follow up
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5476524815
- the related issue https://github.com/apache/arrow-rs/pull/10859
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4111102595
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
Review Comment:
i also test this, do not have much different in e2e performance
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
Review Comment:
i also tested this, do not have much different in e2e performance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4110704262
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> It is really strange to me to try and use the RowSelection API inside a ArrowPredicate whose external API is in terms of BooleanArray (because then we have to turn the result back to BooleanArray, to turn it right back into a RowSeelction...)
> Is the idea to avoid evaluating expensive subsequent filters when the first filter in a fused filter was selective? I wonder if we should look into doing this without RowSelection if possible 🤔
I started with only use `BooleanArray`, but profiling (https://github.com/apache/arrow-rs/issues/10926#issuecomment-5527179934) showed that selectors work better when the surviving rows are grouped together. Even with #10968, using only `BooleanArray` was still about 15% slower on clustered data with four filters that each keep roughly half the rows(this case is not optimizer by #10968).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740892444
🤖 Arrow criterion benchmark running (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880642)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5740880642-2483-4zwlm 6.12.94+ #1 SMP Tue Aug 4 08:44:15 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing parquet-same-projection-filter-fusion (916b65c70ff1cef85ebb490f7e21a960d2f903a3) to a7b89dd (merge-base) [diff](https://github.com/apache/arrow-rs/compare/a7b89dd34eb6bd17e99bdc61a60b37a7cc08f166..916b65c70ff1cef85ebb490f7e21a960d2f903a3)
<details><summary>Run configuration</summary>
```yaml
run benchmark arrow_reader_predicate_fusion
```
</details>
BENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader_predicate_fusion
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#issuecomment-5740880198
run benchmark arrow_reader_predicate_fusion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859#discussion_r4123380960
##########
parquet/src/arrow/arrow_reader/selection/cursor.rs:
##########
@@ -51,6 +50,38 @@ impl Default for RowSelectionPolicy {
}
}
+impl RowSelectionPolicy {
+ /// Resolve this policy for a selection without changing its backing.
Review Comment:
This API is becoming quite nice 👌
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
> I just tested it, and the change is a bit too big. and I don’t find much difference in performance.
Yeah, I agree this would be a larger change
It would only matter if we have predicates that could natively create RowSelections backed by `Vec<...>` -- but by definition no such predicates exist at the moment (or at least they aren't wired in)
##########
parquet/src/arrow/arrow_reader/read_plan.rs:
##########
@@ -157,17 +157,7 @@ impl ReadPlanBuilder {
///
/// Guarantees to return either `Selectors` or `Mask`, never `Auto`.
pub(crate) fn resolve_selection_strategy(&self) -> RowSelectionStrategy {
- match self.row_selection_policy {
- RowSelectionPolicy::Selectors => RowSelectionStrategy::Selectors,
- RowSelectionPolicy::Mask => RowSelectionStrategy::Mask,
- RowSelectionPolicy::Auto { threshold, .. } => {
- let Some(selection) = self.selection.as_ref() else {
- return RowSelectionStrategy::Selectors;
- };
-
- selection.auto_selection_strategy(threshold)
- }
- }
+ self.row_selection_policy.resolve(self.selection.as_ref())
Review Comment:
As a follow on PR we could probably just remove this level of indirection as call `self.row_selection_policy.resolve(self.selection.as_ref())` directly
##########
parquet/src/arrow/arrow_reader/filter.rs:
##########
@@ -198,4 +206,540 @@ impl RowFilter {
pub fn into_predicates(self) -> Vec<Box<dyn ArrowPredicate>> {
self.predicates
}
+
+ /// Fuse consecutive predicates on the same single top-level, non-repeated leaf.
+ /// This avoids repeated decoding or predicate-cache replay of that column.
+ pub(crate) fn fuse_same_projection(
+ self,
+ parquet_schema: &SchemaDescriptor,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> =
+ Vec::with_capacity(self.predicates.len());
+ let mut group: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ let mut flush_group = |group: &mut Vec<Box<dyn ArrowPredicate>>| {
+ if group.len() > 1 && can_fuse_projection(group[0].projection(), parquet_schema) {
+ let group = std::mem::take(group);
+ predicates.push(Box::new(FusedPredicate::new(group, row_selection_policy)));
+ } else {
+ predicates.append(group);
+ }
+ };
+
+ for predicate in self.predicates {
+ if group
+ .last()
+ .is_some_and(|last| last.projection() != predicate.projection())
+ {
+ flush_group(&mut group);
+ }
+ group.push(predicate);
+ }
+ flush_group(&mut group);
+
+ Self { predicates }
+ }
+}
+
+/// Restrict fusion to one top-level, non-repeated leaf to limit compaction costs.
+fn can_fuse_projection(projection: &ProjectionMask, parquet_schema: &SchemaDescriptor) -> bool {
+ let mut leaf_indices =
+ (0..parquet_schema.num_columns()).filter(|idx| projection.leaf_included(*idx));
+ let Some(leaf_idx) = leaf_indices.next() else {
+ return false;
+ };
+ if leaf_indices.next().is_some() {
+ return false;
+ }
+
+ let column = parquet_schema.column(leaf_idx);
+ column.path().parts().len() == 1 && column.max_rep_level() == 0
+}
+
+/// Evaluate same-projection predicates in order on one decoded batch.
+/// Later predicates see only surviving rows, sliced when contiguous or compacted
+/// otherwise. Selections follow the reader's row selection policy.
+struct FusedPredicate {
+ /// At least two predicates, all with the same projection.
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+}
+
+impl FusedPredicate {
+ /// Create a fused predicate from at least two predicates that share one
+ /// projection.
+ fn new(
+ predicates: Vec<Box<dyn ArrowPredicate>>,
+ row_selection_policy: RowSelectionPolicy,
+ ) -> Self {
+ debug_assert!(predicates.len() > 1);
+ debug_assert!(
+ predicates
+ .windows(2)
+ .all(|pair| pair[0].projection() == pair[1].projection())
+ );
+ Self {
+ predicates,
+ row_selection_policy,
+ }
+ }
+}
+
+impl Debug for FusedPredicate {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "FusedPredicate {{ {} predicates }}",
+ self.predicates.len()
+ )
+ }
+}
+
+impl ArrowPredicate for FusedPredicate {
+ fn projection(&self) -> &ProjectionMask {
+ self.predicates[0].projection()
+ }
+
+ fn evaluate(&mut self, batch: RecordBatch) -> Result<BooleanArray, ArrowError> {
+ let num_rows = batch.num_rows();
+ // Positions in the original batch; None until a predicate rejects rows.
+ let mut selection: Option<RowSelection> = None;
+ let mut filtered_batch = batch;
+ let last_predicate_idx = self.predicates.len() - 1;
+
+ for (idx, predicate) in self.predicates.iter_mut().enumerate() {
Review Comment:
This is looking really nice now -- thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(parquet): fuse consecutive filters on the same projection [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #10859:
URL: https://github.com/apache/arrow-rs/pull/10859
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org