fix: report fully matched page-index skips

Author: xudong963

Cargo.lock

@@ -203,6 +203,11 @@ impl ParquetAccessPlan {
         self.should_scan(idx) && self.fully_matched[idx]
     }
 
+    /// Returns the fully matched row group flags.
+    pub(crate) fn fully_matched(&self) -> &Vec<bool> {
+        &self.fully_matched
+    }
+
     /// Return true if any scanned row group is fully matched.
     fn has_fully_matched(&self) -> bool {
         self.row_group_index_iter()

datafusion/datasource-parquet/src/access_plan.rs

@@ -19,6 +19,7 @@ use datafusion_physical_plan::metrics::{
     Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, MetricCategory, MetricType,
     PruningMetrics, RatioMergeStrategy, RatioMetrics, Time,
 };
+use std::sync::{Arc, OnceLock};
 
 /// Stores metrics about the parquet execution for a particular parquet file.
 ///
@@ -67,6 +68,11 @@ pub struct ParquetFileMetrics {
     pub page_index_rows_pruned: PruningMetrics,
     /// Total pages filtered or matched by parquet page index
     pub page_index_pages_pruned: PruningMetrics,
+    /// Lazily registered counter for pages whose page-index pruning was skipped
+    /// because the containing row group was fully matched by row-group statistics.
+    ///
+    /// These pages are still scanned; only page-index predicate evaluation is skipped.
+    page_index_pages_skipped_by_fully_matched: LazyParquetSummaryCount,
     /// Total time spent evaluating parquet page index filters
     pub page_index_eval_time: Time,
     /// Total time spent reading and parsing metadata from the footer
@@ -204,6 +210,12 @@ impl ParquetFileMetrics {
             row_pushdown_eval_time,
             page_index_rows_pruned,
             page_index_pages_pruned,
+            page_index_pages_skipped_by_fully_matched: LazyParquetSummaryCount::new(
+                "page_index_pages_skipped_by_fully_matched",
+                partition,
+                filename,
+                metrics,
+            ),
             statistics_eval_time,
             bloom_filter_eval_time,
             page_index_eval_time,
@@ -213,4 +225,58 @@ impl ParquetFileMetrics {
             predicate_cache_records,
         }
     }
+
+    /// Record pages whose page-index pruning was skipped because the containing
+    /// row group was fully matched by row-group statistics.
+    pub(crate) fn add_page_index_pages_skipped_by_fully_matched(&self, n: usize) {
+        if n == 0 {
+            return;
+        }
+
+        self.page_index_pages_skipped_by_fully_matched.add(n);
+    }
+}
+
+#[derive(Debug, Clone)]
+struct LazyParquetSummaryCount {
+    inner: Arc<LazyParquetSummaryCountInner>,
+}
+
+#[derive(Debug)]
+struct LazyParquetSummaryCountInner {
+    count: OnceLock<Count>,
+    name: &'static str,
+    partition: usize,
+    filename: String,
+    metrics: ExecutionPlanMetricsSet,
+}
+
+impl LazyParquetSummaryCount {
+    fn new(
+        name: &'static str,
+        partition: usize,
+        filename: &str,
+        metrics: &ExecutionPlanMetricsSet,
+    ) -> Self {
+        Self {
+            inner: Arc::new(LazyParquetSummaryCountInner {
+                count: OnceLock::new(),
+                name,
+                partition,
+                filename: filename.to_string(),
+                metrics: metrics.clone(),
+            }),
+        }
+    }
+
+    fn add(&self, n: usize) {
+        let count = self.inner.count.get_or_init(|| {
+            MetricBuilder::new(&self.inner.metrics)
+                .with_new_label("filename", self.inner.filename.clone())
+                .with_type(MetricType::Summary)
+                .with_category(MetricCategory::Rows)
+                .counter(self.inner.name, self.inner.partition)
+        });
+        count.add(n);
+    }
 }

datafusion/datasource-parquet/src/metrics.rs

@@ -35,6 +35,7 @@ use datafusion_pruning::PruningPredicate;
 
 use log::{debug, trace};
 use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
+use parquet::arrow::parquet_column;
 use parquet::file::metadata::{ParquetColumnIndex, ParquetOffsetIndex};
 use parquet::file::page_index::offset_index::PageLocation;
 use parquet::schema::types::SchemaDescriptor;
@@ -193,21 +194,24 @@ impl PagePruningAccessPlanFilter {
         let mut total_pages_skip = 0;
         // track the total number of pages that should not be skipped
         let mut total_pages_select = 0;
-        // track rows that were already proven fully matched at row group
-        // level and therefore did not need page-index predicate evaluation
-        let mut total_rows_fully_matched = 0;
+        // track pages for which page-index pruning was skipped because the
+        // containing row group was already proven fully matched by statistics
+        let mut total_pages_skipped_by_fully_matched = 0;
 
         // for each row group specified in the access plan
         let row_group_indexes = access_plan.row_group_indexes();
         for row_group_index in row_group_indexes {
             // Skip page pruning for fully matched row groups: all rows are
             // known to satisfy the predicate, so page-level pruning is wasted work.
             if access_plan.is_fully_matched(row_group_index) {
-                // Page metrics count evaluated page-index pruning work; this
-                // branch only records rows already proven fully matched.
-                let row_count = groups[row_group_index].num_rows() as usize;
-                total_select += row_count;
-                total_rows_fully_matched += row_count;
+                let page_count = fully_matched_page_count(
+                    row_group_index,
+                    page_index_predicates,
+                    arrow_schema,
+                    parquet_schema,
+                    parquet_metadata,
+                );
+                total_pages_skipped_by_fully_matched += page_count;
 
                 continue;
             }
@@ -218,10 +222,13 @@ impl PagePruningAccessPlanFilter {
             let mut matched_pages_in_group: Option<HashSet<usize>> = None;
 
             for predicate in page_index_predicates {
-                let column = predicate
-                    .required_columns()
-                    .single_column()
-                    .expect("Page pruning requires single column predicates");
+                let Some(column) = predicate.required_columns().single_column() else {
+                    debug!(
+                        "Ignoring multi-column page pruning predicate: {:?}",
+                        predicate.predicate_expr()
+                    );
+                    continue;
+                };
 
                 let converter = StatisticsConverter::try_new(
                     column.name(),
@@ -318,15 +325,15 @@ impl PagePruningAccessPlanFilter {
         file_metrics
             .page_index_rows_pruned
             .add_matched(total_select);
-        file_metrics
-            .page_index_rows_pruned
-            .add_fully_matched(total_rows_fully_matched);
         file_metrics
             .page_index_pages_pruned
             .add_pruned(total_pages_skip);
         file_metrics
             .page_index_pages_pruned
             .add_matched(total_pages_select);
+        file_metrics.add_page_index_pages_skipped_by_fully_matched(
+            total_pages_skipped_by_fully_matched,
+        );
         access_plan
     }
 
@@ -346,6 +353,45 @@ fn update_selection(
     }
 }
 
+/// Returns the number of pages for which page-index pruning is skipped because
+/// the containing row group is fully matched by row-group statistics.
+fn fully_matched_page_count(
+    row_group_index: usize,
+    page_index_predicates: &[PruningPredicate],
+    arrow_schema: &Schema,
+    parquet_schema: &SchemaDescriptor,
+    parquet_metadata: &ParquetMetaData,
+) -> usize {
+    let Some(offset_index) = parquet_metadata.offset_index() else {
+        return 0;
+    };
+
+    let Some(row_group_offsets) = offset_index.get(row_group_index) else {
+        return 0;
+    };
+
+    for predicate in page_index_predicates {
+        let Some(column) = predicate.required_columns().single_column() else {
+            continue;
+        };
+
+        let Some((parquet_column_index, _)) =
+            parquet_column(parquet_schema, arrow_schema, column.name())
+        else {
+            continue;
+        };
+
+        let Some(offset_index_metadata) = row_group_offsets.get(parquet_column_index)
+        else {
+            continue;
+        };
+
+        return offset_index_metadata.page_locations().len();
+    }
+
+    0
+}
+
 /// Returns a [`RowSelection`] for the rows in this row group to scan, in addition to a vec of
 /// booleans that state if each page was matched (true) or not (false).
 ///

datafusion/datasource-parquet/src/page_filter.rs

@@ -74,6 +74,11 @@ impl RowGroupAccessPlanFilter {
         self.access_plan
     }
 
+    /// Returns the is_fully_matched vector.
+    pub fn is_fully_matched(&self) -> &Vec<bool> {
+        self.access_plan.fully_matched()
+    }
+
     /// Prunes the access plan based on the limit and fully contained row groups.
     ///
     /// The pruning works by leveraging the concept of fully matched row groups. Consider a query like:

datafusion/datasource-parquet/src/row_group_filter.rs

@@ -1010,7 +1010,10 @@ impl MetricValue {
             Self::SpilledBytes(_) => 11,
             Self::SpilledRows(_) => 12,
             Self::CurrentMemoryUsage(_) => 13,
-            Self::Count { .. } => 14,
+            Self::Count { name, .. } => match name.as_ref() {
+                "page_index_pages_skipped_by_fully_matched" => 8,
+                _ => 14,
+            },
             Self::Gauge { .. } => 15,
             Self::Time { .. } => 16,
             Self::Ratio { .. } => 17,

datafusion/physical-expr-common/src/metrics/value.rs

@@ -104,7 +104,7 @@ Plan with Metrics
 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, <slt:ignore>]
 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=<slt:ignore>, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=18.31% (210/1.15 K)]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=<slt:ignore>, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=18.31% (210/1.15 K)]
 
 statement ok
 set datafusion.explain.analyze_level = dev;

datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt

@@ -247,7 +247,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
 ----
 Plan with Metrics
 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=22.13% (521/2.35 K)]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=22.13% (521/2.35 K)]
 
 statement ok
 reset datafusion.explain.analyze_categories;
@@ -262,7 +262,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
 ----
 Plan with Metrics
 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
 
 statement ok
 reset datafusion.explain.analyze_categories;
@@ -277,7 +277,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
 ----
 Plan with Metrics
 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
 
 statement ok
 reset datafusion.explain.analyze_categories;