[POC] Add statistics-based thresholds for deferring RowFilter selection in ReadPlanBuilder

parquet/benches/arrow_reader_clickbench.rs

@@ -716,15 +716,15 @@ impl ReadTest {
         };
 
         // setup the reader
-        let mut stream = ParquetRecordBatchStreamBuilder::new_with_metadata(
+        let builder = ParquetRecordBatchStreamBuilder::new_with_metadata(
             parquet_file,
             self.arrow_reader_metadata.clone(),
         )
         .with_batch_size(8192)
         .with_projection(self.projection_mask.clone())
         .with_row_filter(self.row_filter())
-        .build()
-        .unwrap();
+        .with_scatter_threshold(Some(0.75));
+        let mut stream = builder.build().unwrap();
 
         // run the stream to its end
         let mut row_count = 0;
@@ -747,15 +747,15 @@ impl ReadTest {
         let reader = ParquetObjectReader::new(store, location);
 
         // setup the reader
-        let mut stream = ParquetRecordBatchStreamBuilder::new_with_metadata(
+        let builder = ParquetRecordBatchStreamBuilder::new_with_metadata(
             reader,
             self.arrow_reader_metadata.clone(),
         )
         .with_batch_size(8192)
         .with_projection(self.projection_mask.clone())
         .with_row_filter(self.row_filter())
-        .build()
-        .unwrap();
+        .with_scatter_threshold(Some(0.75));
+        let mut stream = builder.build().unwrap();
 
         // run the stream to its end
         let mut row_count = 0;
@@ -774,15 +774,15 @@ impl ReadTest {
         };
 
         // setup the reader
-        let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(
+        let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(
             parquet_file,
             self.arrow_reader_metadata.clone(),
         )
         .with_batch_size(8192)
         .with_projection(self.projection_mask.clone())
         .with_row_filter(self.row_filter())
-        .build()
-        .unwrap();
+        .with_scatter_threshold(Some(0.75));
+        let reader = builder.build().unwrap();
 
         // run the stream to its end
         let mut row_count = 0;

parquet/src/arrow/arrow_reader/metrics.rs

@@ -18,8 +18,62 @@
 //! [ArrowReaderMetrics] for collecting metrics about the Arrow reader
 
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::sync::atomic::AtomicUsize;
 
+/// Why a predicate was applied or deferred at the read-plan stage.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FilterDeferralDecisionReason {
+    /// Predicate selected all rows with no existing selection, so no selection
+    /// structure was materialized.
+    AllSelectedFastPath,
+    /// Deferral threshold is not configured.
+    ThresholdDisabled,
+    /// Row count was zero.
+    ZeroRowCount,
+    /// Predicate did not increase selector fragmentation.
+    FragmentationNotIncreased,
+    /// Predicate passed non-deferral gates and was kept applied.
+    GatesPassed,
+    /// Predicate failed one or more non-deferral gates and was deferred.
+    GatesFailedDeferred,
+}
+
+/// Per-filter stats captured during read-plan predicate evaluation.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FilterSelectivityStat {
+    /// Zero-based predicate evaluation index within this read-plan build.
+    pub predicate_index: usize,
+    /// Number of rows considered by the predicate decision.
+    pub row_count: usize,
+    /// Selector count before applying this predicate.
+    pub current_selector_count: usize,
+    /// Selector count after applying this predicate.
+    pub absolute_selector_count: usize,
+    /// Skipped rows before applying this predicate.
+    pub current_skipped_rows: usize,
+    /// Skipped rows after applying this predicate.
+    pub absolute_skipped_rows: usize,
+    /// Long skipped rows before applying this predicate.
+    pub current_long_skip_rows: usize,
+    /// Long skipped rows after applying this predicate.
+    pub absolute_long_skip_rows: usize,
+    /// Absolute skipped/rows ratio.
+    pub absolute_skip_selectivity: f64,
+    /// Absolute long-skipped/skipped ratio.
+    pub absolute_long_skip_share: f64,
+    /// Incremental skipped/rows ratio contributed by this predicate.
+    pub delta_skip_selectivity: f64,
+    /// Incremental long-skipped/skipped ratio contributed by this predicate.
+    pub delta_long_skip_share: f64,
+    /// Threshold supplied via `with_scatter_threshold`.
+    pub long_skip_share_threshold: Option<f64>,
+    /// Whether this predicate result was deferred.
+    pub deferred: bool,
+    /// Why this predicate was applied or deferred.
+    pub decision_reason: FilterDeferralDecisionReason,
+}
+
 /// This enum represents the state of Arrow reader metrics collection.
 ///
 /// The inner metrics are stored in an `Arc<ArrowReaderMetricsInner>`
@@ -90,6 +144,22 @@ impl ArrowReaderMetrics {
         }
     }
 
+    /// Per-filter selectivity/deferral stats captured during planning.
+    ///
+    /// Returns `None` if metrics are disabled.
+    pub fn filter_selectivity_stats(&self) -> Option<Vec<FilterSelectivityStat>> {
+        match self {
+            Self::Disabled => None,
+            Self::Enabled(inner) => {
+                let stats = match inner.filter_selectivity_stats.lock() {
+                    Ok(stats) => stats,
+                    Err(poisoned) => poisoned.into_inner(),
+                };
+                Some(stats.clone())
+            }
+        }
+    }
+
     /// Increments the count of records read from the inner reader
     pub(crate) fn increment_inner_reads(&self, count: usize) {
         let Self::Enabled(inner) = self else {
@@ -110,6 +180,19 @@ impl ArrowReaderMetrics {
             .records_read_from_cache
             .fetch_add(count, std::sync::atomic::Ordering::Relaxed);
     }
+
+    /// Records a per-filter selectivity stat.
+    pub(crate) fn record_filter_selectivity_stat(&self, stat: FilterSelectivityStat) {
+        let Self::Enabled(inner) = self else {
+            return;
+        };
+
+        let mut stats = match inner.filter_selectivity_stats.lock() {
+            Ok(stats) => stats,
+            Err(poisoned) => poisoned.into_inner(),
+        };
+        stats.push(stat);
+    }
 }
 
 /// Holds the actual metrics for the Arrow reader.
@@ -122,6 +205,8 @@ pub struct ArrowReaderMetricsInner {
     records_read_from_inner: AtomicUsize,
     /// Total number of records read from previously cached pages
     records_read_from_cache: AtomicUsize,
+    /// Per-filter selectivity stats captured during read planning.
+    filter_selectivity_stats: Mutex<Vec<FilterSelectivityStat>>,
 }
 
 impl ArrowReaderMetricsInner {
@@ -130,6 +215,49 @@ impl ArrowReaderMetricsInner {
         Self {
             records_read_from_inner: AtomicUsize::new(0),
             records_read_from_cache: AtomicUsize::new(0),
+            filter_selectivity_stats: Mutex::new(vec![]),
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_filter_selectivity_stats_disabled() {
+        let metrics = ArrowReaderMetrics::disabled();
+        assert!(metrics.filter_selectivity_stats().is_none());
+    }
+
+    #[test]
+    fn test_filter_selectivity_stats_enabled() {
+        let metrics = ArrowReaderMetrics::enabled();
+        metrics.record_filter_selectivity_stat(FilterSelectivityStat {
+            predicate_index: 0,
+            row_count: 100,
+            current_selector_count: 1,
+            absolute_selector_count: 3,
+            current_skipped_rows: 10,
+            absolute_skipped_rows: 20,
+            current_long_skip_rows: 10,
+            absolute_long_skip_rows: 15,
+            absolute_skip_selectivity: 0.2,
+            absolute_long_skip_share: 0.75,
+            delta_skip_selectivity: 0.1,
+            delta_long_skip_share: 0.5,
+            long_skip_share_threshold: Some(0.75),
+            deferred: true,
+            decision_reason: FilterDeferralDecisionReason::GatesFailedDeferred,
+        });
+
+        let stats = metrics.filter_selectivity_stats().expect("metrics enabled");
+        assert_eq!(stats.len(), 1);
+        assert_eq!(stats[0].predicate_index, 0);
+        assert!(stats[0].deferred);
+        assert_eq!(
+            stats[0].decision_reason,
+            FilterDeferralDecisionReason::GatesFailedDeferred
+        );
+    }
+}

parquet/src/arrow/arrow_reader/mod.rs

@@ -139,6 +139,8 @@ pub struct ArrowReaderBuilder<T> {
     pub(crate) metrics: ArrowReaderMetrics,
 
     pub(crate) max_predicate_cache_size: usize,
+
+    pub(crate) scatter_threshold: Option<f64>,
 }
 
 impl<T: Debug> Debug for ArrowReaderBuilder<T> {
@@ -157,6 +159,7 @@ impl<T: Debug> Debug for ArrowReaderBuilder<T> {
             .field("limit", &self.limit)
             .field("offset", &self.offset)
             .field("metrics", &self.metrics)
+            .field("scatter_threshold", &self.scatter_threshold)
             .finish()
     }
 }
@@ -178,6 +181,7 @@ impl<T> ArrowReaderBuilder<T> {
             offset: None,
             metrics: ArrowReaderMetrics::Disabled,
             max_predicate_cache_size: 100 * 1024 * 1024, // 100MB default cache size
+            scatter_threshold: None,
         }
     }
 
@@ -430,6 +434,37 @@ impl<T> ArrowReaderBuilder<T> {
             ..self
         }
     }
+
+    /// Set a scatter threshold for filter deferral.
+    ///
+    /// Deferral is considered only when a predicate increases selector
+    /// fragmentation. In that case, the result is deferred unless:
+    ///
+    /// 1. absolute skip selectivity (`skipped_rows / total_rows`) is at least
+    ///    10%,
+    /// 2. absolute long-skip share (`long_skip_rows / skipped_rows`) is at
+    ///    least this threshold,
+    /// 3. incremental skip selectivity added by the predicate is at least 2%,
+    ///    and
+    /// 4. incremental long-skip share added by the predicate is at least this
+    ///    threshold.
+    ///
+    /// For example, `0.75` means at least 75% of skipped rows must be in long
+    /// skip runs to avoid deferral.
+    ///
+    /// The deferred results are applied at the end via
+    /// [`RowSelection::intersection`], so correctness is preserved.
+    ///
+    /// `None` disables deferral (the default).
+    ///
+    /// [`RowFilter`]: crate::arrow::arrow_reader::RowFilter
+    /// [`RowSelection::intersection`]: crate::arrow::arrow_reader::RowSelection::intersection
+    pub fn with_scatter_threshold(self, threshold: Option<f64>) -> Self {
+        Self {
+            scatter_threshold: threshold,
+            ..self
+        }
+    }
 }
 
 /// Options that control how [`ParquetMetaData`] is read when constructing
@@ -1188,6 +1223,7 @@ 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: _,
+            scatter_threshold,
         } = self;
 
         // Try to avoid allocate large buffer
@@ -1203,7 +1239,9 @@ impl<T: ChunkReader + 'static> ParquetRecordBatchReaderBuilder<T> {
 
         let mut plan_builder = ReadPlanBuilder::new(batch_size)
             .with_selection(selection)
-            .with_row_selection_policy(row_selection_policy);
+            .with_metrics(metrics.clone())
+            .with_row_selection_policy(row_selection_policy)
+            .with_scatter_threshold(scatter_threshold);
 
         // Update selection based on any filters
         if let Some(filter) = filter.as_mut() {
@@ -1217,7 +1255,13 @@ impl<T: ChunkReader + 'static> ParquetRecordBatchReaderBuilder<T> {
                     .with_parquet_metadata(&reader.metadata)
                     .build_array_reader(fields.as_deref(), predicate.projection())?;
 
-                plan_builder = plan_builder.with_predicate(array_reader, predicate.as_mut())?;
+                let row_count: usize = reader
+                    .row_groups
+                    .iter()
+                    .map(|&i| reader.metadata.row_group(i).num_rows() as usize)
+                    .sum();
+                plan_builder =
+                    plan_builder.with_predicate(array_reader, predicate.as_mut(), row_count)?;
             }
         }