feat(parquet): skip RowFilter evaluation for fully matched row groups

parquet/src/arrow/arrow_reader/mod.rs

@@ -139,6 +139,12 @@ pub struct ArrowReaderBuilder<T> {
     pub(crate) metrics: ArrowReaderMetrics,
 
     pub(crate) max_predicate_cache_size: usize,
+
+    /// Row groups where ALL rows are known to match the filter predicate.
+    ///
+    /// For these row groups, the [`RowFilter`] evaluation is skipped entirely
+    /// since the predicate is guaranteed to be true for every row.
+    pub(crate) fully_matched_row_groups: Option<Vec<usize>>,
 }
 
 impl<T: Debug> Debug for ArrowReaderBuilder<T> {
@@ -178,6 +184,7 @@ impl<T> ArrowReaderBuilder<T> {
             offset: None,
             metrics: ArrowReaderMetrics::Disabled,
             max_predicate_cache_size: 100 * 1024 * 1024, // 100MB default cache size
+            fully_matched_row_groups: None,
         }
     }
 
@@ -344,6 +351,28 @@ impl<T> ArrowReaderBuilder<T> {
         }
     }
 
+    /// Specify row groups where ALL rows are known to match the filter predicate.
+    ///
+    /// For these row groups, the [`RowFilter`] evaluation (set via
+    /// [`Self::with_row_filter`]) is skipped entirely since the predicate is
+    /// guaranteed to be `true` for every row. This avoids the cost of decoding
+    /// filter columns and evaluating the predicate expression for those row
+    /// groups.
+    ///
+    /// This is typically determined by evaluating row group statistics: if the
+    /// statistics prove that all rows satisfy the predicate, the row group is
+    /// "fully matched."
+    ///
+    /// The provided indices must be a subset of the row groups specified via
+    /// [`Self::with_row_groups`] (or all row groups if none were specified).
+    /// Indices not present in the row group list are ignored.
+    pub fn with_fully_matched_row_groups(self, row_groups: Vec<usize>) -> Self {
+        Self {
+            fully_matched_row_groups: Some(row_groups),
+            ..self
+        }
+    }
+
     /// Provide a limit to the number of rows to be read
     ///
     /// The limit will be applied after any [`Self::with_row_selection`] and [`Self::with_row_filter`]
@@ -1188,6 +1217,8 @@ impl<T: ChunkReader + 'static> ParquetRecordBatchReaderBuilder<T> {
             metrics,
             // Not used for the sync reader, see https://github.com/apache/arrow-rs/issues/8000
             max_predicate_cache_size: _,
+            // Not used for the sync reader (single row group per reader)
+            fully_matched_row_groups: _,
         } = self;
 
         // Try to avoid allocate large buffer

parquet/src/arrow/async_reader/mod.rs

@@ -497,6 +497,7 @@ impl<T: AsyncFileReader + Send + 'static> ParquetRecordBatchStreamBuilder<T> {
             offset,
             metrics,
             max_predicate_cache_size,
+            fully_matched_row_groups,
         } = self;
 
         // Ensure schema of ParquetRecordBatchStream respects projection, and does
@@ -522,6 +523,7 @@ impl<T: AsyncFileReader + Send + 'static> ParquetRecordBatchStreamBuilder<T> {
             offset,
             metrics,
             max_predicate_cache_size,
+            fully_matched_row_groups,
         }
         .build()?;
 

parquet/src/arrow/push_decoder/mod.rs

@@ -176,6 +176,7 @@ impl ParquetPushDecoderBuilder {
             metrics,
             row_selection_policy,
             max_predicate_cache_size,
+            fully_matched_row_groups,
         } = self;
 
         // If no row groups were specified, read all of them
@@ -197,6 +198,7 @@ impl ParquetPushDecoderBuilder {
             max_predicate_cache_size,
             buffers,
             row_selection_policy,
+            fully_matched_row_groups,
         );
 
         // Initialize the decoder with the configured options

parquet/src/arrow/push_decoder/reader_builder/mod.rs

@@ -38,6 +38,7 @@ use bytes::Bytes;
 use data::DataRequest;
 use filter::AdvanceResult;
 use filter::FilterInfo;
+use std::collections::HashSet;
 use std::ops::Range;
 use std::sync::{Arc, RwLock};
 
@@ -160,6 +161,10 @@ pub(crate) struct RowGroupReaderBuilder {
     /// Strategy for materialising row selections
     row_selection_policy: RowSelectionPolicy,
 
+    /// Row groups where ALL rows are known to match the filter predicate.
+    /// For these row groups, filter evaluation is skipped entirely.
+    fully_matched_row_groups: HashSet<usize>,
+
     /// Current state of the decoder.
     ///
     /// It is taken when processing, and must be put back before returning
@@ -185,6 +190,7 @@ impl RowGroupReaderBuilder {
         max_predicate_cache_size: usize,
         buffers: PushBuffers,
         row_selection_policy: RowSelectionPolicy,
+        fully_matched_row_groups: Option<Vec<usize>>,
     ) -> Self {
         Self {
             batch_size,
@@ -197,6 +203,9 @@ impl RowGroupReaderBuilder {
             metrics,
             max_predicate_cache_size,
             row_selection_policy,
+            fully_matched_row_groups: fully_matched_row_groups
+                .map(|v| v.into_iter().collect())
+                .unwrap_or_default(),
             state: Some(RowGroupDecoderState::Finished),
             buffers,
         }
@@ -328,6 +337,22 @@ impl RowGroupReaderBuilder {
                     }));
                 };
 
+                // Skip filter for fully matched row groups: all rows are known
+                // to satisfy the predicate based on row group statistics, so
+                // evaluating the filter would be wasted work.
+                if self
+                    .fully_matched_row_groups
+                    .contains(&row_group_info.row_group_idx)
+                {
+                    // Put the filter back for subsequent non-fully-matched row groups
+                    self.filter = Some(filter);
+                    return Ok(NextState::again(RowGroupDecoderState::StartData {
+                        row_group_info,
+                        column_chunks,
+                        cache_info: None,
+                    }));
+                }
+
                 // we have predicates to evaluate
                 let cache_projection =
                     self.compute_cache_projection(row_group_info.row_group_idx, &filter);