Support row group limit pruning

Author: xudong963

datafusion/datasource-parquet/src/metrics.rs

@@ -45,18 +45,12 @@ pub struct ParquetFileMetrics {
     pub files_ranges_pruned_statistics: PruningMetrics,
     /// Number of times the predicate could not be evaluated
     pub predicate_evaluation_errors: Count,
-    /// Number of row groups whose bloom filters were checked and matched (not pruned)
-    pub row_groups_matched_bloom_filter: Count,
     /// Number of row groups pruned by bloom filters
-    pub row_groups_pruned_bloom_filter: Count,
+    pub row_groups_pruned_bloom_filter: PruningMetrics,
     /// Number of row groups pruned due to limit pruning.
-    pub limit_pruned_row_groups: Count,
-    /// Number of row groups whose statistics were checked and fully matched
-    pub row_groups_fully_matched_statistics: Count,
-    /// Number of row groups whose statistics were checked and matched (not pruned)
-    pub row_groups_matched_statistics: Count,
+    pub limit_pruned_row_groups: PruningMetrics,
     /// Number of row groups pruned by statistics
-    pub row_groups_pruned_statistics: Count,
+    pub row_groups_pruned_statistics: PruningMetrics,
     /// Total number of bytes scanned
     pub bytes_scanned: Count,
     /// Total rows filtered out by predicates pushed into parquet scan
@@ -98,19 +92,13 @@ impl ParquetFileMetrics {
         // -----------------------
         let row_groups_pruned_bloom_filter = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
-            .counter("row_groups_pruned_bloom_filter", partition);
+            .with_type(MetricType::SUMMARY)
+            .pruning_metrics("row_groups_pruned_bloom_filter", partition);
 
         let limit_pruned_row_groups = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
-            .counter("limit_pruned_row_groups", partition);
-
-        let row_groups_fully_matched_statistics = MetricBuilder::new(metrics)
-            .with_new_label("filename", filename.to_string())
-            .counter("row_groups_fully_matched_statistics", partition);
-
-        let row_groups_matched_statistics = MetricBuilder::new(metrics)
-            .with_new_label("filename", filename.to_string())
-            .counter("row_groups_matched_statistics", partition);
+            .with_type(MetricType::SUMMARY)
+            .pruning_metrics("limit_pruned_row_groups", partition);
 
         let row_groups_pruned_statistics = MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
@@ -185,8 +173,6 @@ impl ParquetFileMetrics {
             files_ranges_pruned_statistics,
             predicate_evaluation_errors,
             row_groups_pruned_bloom_filter,
-            row_groups_fully_matched_statistics,
-            row_groups_matched_statistics,
             row_groups_pruned_statistics,
             limit_pruned_row_groups,
             bytes_scanned,

datafusion/datasource-parquet/src/row_group_filter.rs

@@ -111,7 +111,7 @@ impl RowGroupAccessPlanFilter {
             let new_num_accessible_row_groups = fully_matched_row_group_indexes.len();
             let pruned_count = original_num_accessible_row_groups
                 .saturating_sub(new_num_accessible_row_groups);
-            metrics.limit_pruned_row_groups.add(pruned_count);
+            metrics.limit_pruned_row_groups.add_pruned(pruned_count);
 
             let mut new_access_plan = ParquetAccessPlan::new_none(rg_metadata.len());
             for &idx in &fully_matched_row_group_indexes {
@@ -194,7 +194,6 @@ impl RowGroupAccessPlanFilter {
                     } else {
                         metrics.row_groups_pruned_statistics.add_matched(1);
                         fully_contained_candidates_original_idx.push(*idx);
-                        metrics.row_groups_matched_statistics.add(1);
                     }
                 }
 
@@ -231,7 +230,9 @@ impl RowGroupAccessPlanFilter {
                                 // it implies that *all* rows in this group satisfy the original predicate.
                                 if !inverted_values[i] {
                                     self.is_fully_matched[original_row_group_idx] = true;
-                                    metrics.row_groups_fully_matched_statistics.add(1);
+                                    metrics
+                                        .row_groups_pruned_statistics
+                                        .add_fully_matched(1);
                                 }
                             }
                         }

datafusion/physical-plan/src/metrics/value.rs

@@ -373,6 +373,7 @@ impl Drop for ScopedTimerGuard<'_> {
 pub struct PruningMetrics {
     pruned: Arc<AtomicUsize>,
     matched: Arc<AtomicUsize>,
+    fully_matched: Arc<AtomicUsize>,
 }
 
 impl Display for PruningMetrics {
@@ -401,6 +402,7 @@ impl PruningMetrics {
         Self {
             pruned: Arc::new(AtomicUsize::new(0)),
             matched: Arc::new(AtomicUsize::new(0)),
+            fully_matched: Arc::new(AtomicUsize::new(0)),
         }
     }
 
@@ -418,6 +420,13 @@ impl PruningMetrics {
         self.matched.fetch_add(n, Ordering::Relaxed);
     }
 
+    /// Add `n` to the metric's fully matched value
+    pub fn add_fully_matched(&self, n: usize) {
+        // relaxed ordering for operations on `value` poses no issues
+        // we're purely using atomic ops with no associated memory ops
+        self.fully_matched.fetch_add(n, Ordering::Relaxed);
+    }
+
     /// Subtract `n` to the metric's matched value.
     pub fn subtract_matched(&self, n: usize) {
         // relaxed ordering for operations on `value` poses no issues