add % of dataset trigger

Author: adriangb

datafusion/common/src/config.rs

@@ -776,6 +776,14 @@ config_namespace! {
         /// Set to 0 to disable the collection phase entirely.
         pub filter_statistics_collection_min_rows: u64, default = 10_000
 
+        /// (reading) Fraction of total dataset rows to use for the statistics
+        /// collection phase. When > 0 and the dataset row count is known, the
+        /// effective collection threshold is max(min_rows, fraction * total_rows).
+        /// 0.0 (default) = disabled, use filter_statistics_collection_min_rows only.
+        /// 0.05 = collect stats on at least 5% of the dataset.
+        /// Must be in [0.0, 1.0].
+        pub filter_statistics_collection_fraction: f64, default = 0.0
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -212,6 +212,7 @@ impl ParquetOptions {
             filter_pushdown_min_bytes_per_sec: _, // not used for writer props
             filter_correlation_threshold: _,   // not used for writer props
             filter_statistics_collection_min_rows: _, // not used for writer props
+            filter_statistics_collection_fraction: _, // not used for writer props
         } = self;
 
         let mut builder = WriterProperties::builder()

datafusion/datasource-parquet/src/opener.rs

@@ -1499,7 +1499,7 @@ mod test {
                 // (skipping adaptive logic), and min_rows_for_collection=0 to
                 // disable the collection phase.
                 selectivity_tracker: Arc::new(parking_lot::RwLock::new(
-                    SelectivityTracker::new_with_config(0.0, 1.5, 0),
+                    SelectivityTracker::new_with_config(0.0, 1.5, 0, 0.0),
                 )),
             }
         }

datafusion/datasource-parquet/src/selectivity.rs

@@ -253,6 +253,13 @@ pub struct SelectivityTracker {
     /// During collection, all filters go to post-scan for accurate measurement.
     /// Default: 10_000
     min_rows_for_collection: u64,
+    /// Fraction of total dataset rows for collection phase (0.0 = disabled).
+    /// When > 0 and dataset size is known, effective threshold =
+    /// max(min_rows_for_collection, (fraction * total_rows) as u64).
+    collection_fraction: f64,
+    /// Resolved minimum rows after notify_dataset_rows() is called.
+    /// None = not yet resolved (use min_rows_for_collection as-is).
+    resolved_min_rows: Option<u64>,
 }
 
 impl Default for SelectivityTracker {
@@ -275,6 +282,8 @@ impl SelectivityTracker {
             min_bytes_per_sec,
             correlation_threshold: 1.5,
             min_rows_for_collection: 10_000,
+            collection_fraction: 0.0,
+            resolved_min_rows: None,
         }
     }
 
@@ -283,13 +292,16 @@ impl SelectivityTracker {
         min_bytes_per_sec: f64,
         correlation_threshold: f64,
         min_rows_for_collection: u64,
+        collection_fraction: f64,
     ) -> Self {
         Self {
             stats: HashMap::new(),
             correlations: HashMap::new(),
             min_bytes_per_sec,
             correlation_threshold,
             min_rows_for_collection,
+            collection_fraction,
+            resolved_min_rows: None,
         }
     }
 
@@ -298,6 +310,25 @@ impl SelectivityTracker {
         self.min_bytes_per_sec
     }
 
+    /// Returns the effective minimum rows for collection, taking into account
+    /// the fraction-based threshold if it has been resolved.
+    fn effective_min_rows(&self) -> u64 {
+        self.resolved_min_rows.unwrap_or(self.min_rows_for_collection)
+    }
+
+    /// Notify the tracker of the total dataset row count so the fraction-based
+    /// threshold can be resolved.
+    ///
+    /// When `collection_fraction > 0`, computes:
+    /// `resolved_min_rows = max(min_rows_for_collection, (fraction * total_rows) as u64)`
+    pub fn notify_dataset_rows(&mut self, total_rows: u64) {
+        if self.collection_fraction > 0.0 {
+            let fraction_rows = (self.collection_fraction * total_rows as f64) as u64;
+            self.resolved_min_rows =
+                Some(self.min_rows_for_collection.max(fraction_rows));
+        }
+    }
+
     /// Get the effectiveness for a filter expression, if known.
     pub fn get_effectiveness(&self, expr: &Arc<dyn PhysicalExpr>) -> Option<f64> {
         let key = ExprKey::new(expr);
@@ -314,17 +345,16 @@ impl SelectivityTracker {
     /// Returns false if no stats exist yet (no filters registered) or if
     /// min_rows_for_collection is 0 (collection disabled).
     pub fn in_collection_phase(&self) -> bool {
-        if self.min_rows_for_collection == 0 {
+        let min_rows = self.effective_min_rows();
+        if min_rows == 0 {
             return false;
         }
         if self.stats.is_empty() {
             // No filters registered yet - treat as collection phase
             // so the first file's filters go to post-scan for measurement
             return true;
         }
-        self.stats
-            .values()
-            .any(|s| s.rows_total < self.min_rows_for_collection)
+        self.stats.values().any(|s| s.rows_total < min_rows)
     }
 
     /// Partition filters into row_filters and post_scan based on bytes/sec throughput.
@@ -498,16 +528,15 @@ impl SelectivityTracker {
         let stats_b = self.stats.get(&key_b)?;
 
         // Need sufficient data
-        if stats_a.rows_total < self.min_rows_for_collection
-            || stats_b.rows_total < self.min_rows_for_collection
-        {
+        let min_rows = self.effective_min_rows();
+        if stats_a.rows_total < min_rows || stats_b.rows_total < min_rows {
             return None;
         }
 
         let pair_key = PairKey::new(&key_a, &key_b);
         let pair_stats = self.correlations.get(&pair_key)?;
 
-        if pair_stats.rows_total < self.min_rows_for_collection {
+        if pair_stats.rows_total < min_rows {
             return None;
         }
 
@@ -890,7 +919,7 @@ mod tests {
 
     #[test]
     fn test_correlation_stats_update() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -916,7 +945,7 @@ mod tests {
 
     #[test]
     fn test_correlation_ratio_independent() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -932,7 +961,7 @@ mod tests {
 
     #[test]
     fn test_correlation_ratio_insufficient_data() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 1000);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 1000, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -947,7 +976,7 @@ mod tests {
 
     #[test]
     fn test_in_collection_phase() {
-        let mut tracker = SelectivityTracker::new_with_config(100.0, 1.5, 1000);
+        let mut tracker = SelectivityTracker::new_with_config(100.0, 1.5, 1000, 0.0);
 
         // No stats yet - in collection phase
         assert!(tracker.in_collection_phase());
@@ -966,15 +995,15 @@ mod tests {
 
     #[test]
     fn test_in_collection_phase_disabled() {
-        let tracker = SelectivityTracker::new_with_config(100.0, 1.5, 0);
+        let tracker = SelectivityTracker::new_with_config(100.0, 1.5, 0, 0.0);
 
         // min_rows = 0 means collection is disabled
         assert!(!tracker.in_collection_phase());
     }
 
     #[test]
     fn test_partition_filters_grouped_collection_phase() {
-        let tracker = SelectivityTracker::new_with_config(100.0, 1.5, 10_000);
+        let tracker = SelectivityTracker::new_with_config(100.0, 1.5, 10_000, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -991,7 +1020,7 @@ mod tests {
 
     #[test]
     fn test_partition_filters_grouped_all_independent() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -1018,7 +1047,7 @@ mod tests {
 
     #[test]
     fn test_partition_filters_grouped_correlated() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -1043,7 +1072,7 @@ mod tests {
 
     #[test]
     fn test_partition_filters_grouped_mixed() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -1096,7 +1125,7 @@ mod tests {
 
     #[test]
     fn test_partition_filters_grouped_single_filter() {
-        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         tracker.update(&filter_a, 10, 100, 0);
@@ -1113,7 +1142,7 @@ mod tests {
     #[test]
     fn test_partition_filters_grouped_with_low_throughput() {
         // Use a bytes/sec threshold: 100 bytes/sec
-        let mut tracker = SelectivityTracker::new_with_config(100.0, 1.5, 100);
+        let mut tracker = SelectivityTracker::new_with_config(100.0, 1.5, 100, 0.0);
 
         let filter_a = make_filter("a", 5);
         let filter_b = make_filter("a", 10);
@@ -1136,4 +1165,53 @@ mod tests {
         assert_eq!(result.row_filter_groups[0].len(), 1);
         assert_eq!(result.post_scan.len(), 1);
     }
+
+    #[test]
+    fn test_notify_dataset_rows_resolves_fraction() {
+        let mut tracker =
+            SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.05);
+        // Before notify, effective_min_rows = min_rows_for_collection
+        assert_eq!(tracker.effective_min_rows(), 100);
+
+        // 5% of 10_000 = 500, which is > 100
+        tracker.notify_dataset_rows(10_000);
+        assert_eq!(tracker.effective_min_rows(), 500);
+    }
+
+    #[test]
+    fn test_notify_dataset_rows_floor_behavior() {
+        let mut tracker =
+            SelectivityTracker::new_with_config(0.0, 1.5, 1000, 0.05);
+        // 5% of 10_000 = 500, but min_rows = 1000 is larger
+        tracker.notify_dataset_rows(10_000);
+        assert_eq!(tracker.effective_min_rows(), 1000);
+    }
+
+    #[test]
+    fn test_notify_dataset_rows_fraction_disabled() {
+        let mut tracker =
+            SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.0);
+        tracker.notify_dataset_rows(1_000_000);
+        // fraction = 0.0, so resolved_min_rows stays None
+        assert_eq!(tracker.effective_min_rows(), 100);
+    }
+
+    #[test]
+    fn test_collection_phase_with_fraction() {
+        let filter = make_filter("a", 5);
+
+        let mut tracker =
+            SelectivityTracker::new_with_config(0.0, 1.5, 100, 0.05);
+        // 5% of 100_000 = 5000
+        tracker.notify_dataset_rows(100_000);
+        assert_eq!(tracker.effective_min_rows(), 5000);
+
+        // Record 200 rows — still in collection (200 < 5000)
+        tracker.update(&filter, 80, 200, 10_000_000);
+        assert!(tracker.in_collection_phase());
+
+        // Record enough to pass threshold
+        tracker.update(&filter, 1000, 5000, 50_000_000);
+        assert!(!tracker.in_collection_phase());
+    }
 }

datafusion/datasource-parquet/src/source.rs

@@ -355,6 +355,7 @@ impl ParquetSource {
                 opts.filter_pushdown_min_bytes_per_sec,
                 opts.filter_correlation_threshold,
                 opts.filter_statistics_collection_min_rows,
+                opts.filter_statistics_collection_fraction,
             ),
         ));
         self.table_parquet_options = table_parquet_options;
@@ -511,6 +512,7 @@ impl ParquetSource {
                 min_bytes_per_sec,
                 opts.filter_correlation_threshold,
                 opts.filter_statistics_collection_min_rows,
+                opts.filter_statistics_collection_fraction,
             ),
         ));
         self
@@ -633,6 +635,16 @@ impl FileSource for ParquetSource {
             reverse_row_groups: self.reverse_row_groups,
             selectivity_tracker: Arc::clone(&self.selectivity_tracker),
         });
+
+        // Notify the selectivity tracker of the total dataset row count
+        // so fraction-based collection thresholds can be resolved.
+        if let Some(&total_rows) = base_config.statistics().num_rows.get_value()
+        {
+            self.selectivity_tracker
+                .write()
+                .notify_dataset_rows(total_rows as u64);
+        }
+
         Ok(opener)
     }
 

datafusion/proto-common/proto/datafusion_common.proto

@@ -607,6 +607,18 @@ message ParquetOptions {
   oneof filter_pushdown_min_bytes_per_sec_opt {
     double filter_pushdown_min_bytes_per_sec = 35;
   }
+
+  oneof filter_correlation_threshold_opt {
+    double filter_correlation_threshold = 36;
+  }
+
+  oneof filter_statistics_collection_min_rows_opt {
+    uint64 filter_statistics_collection_min_rows = 37;
+  }
+
+  oneof filter_statistics_collection_fraction_opt {
+    double filter_statistics_collection_fraction = 38;
+  }
 }
 
 enum JoinSide {

datafusion/proto-common/src/from_proto/mod.rs

@@ -1093,8 +1093,15 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
             filter_pushdown_min_bytes_per_sec: value.filter_pushdown_min_bytes_per_sec_opt.map(|opt| match opt {
                 protobuf::parquet_options::FilterPushdownMinBytesPerSecOpt::FilterPushdownMinBytesPerSec(v) => v,
             }).unwrap_or(f64::INFINITY),
-            filter_correlation_threshold: 1.5,
-            filter_statistics_collection_min_rows: 10_000,
+            filter_correlation_threshold: value.filter_correlation_threshold_opt.map(|opt| match opt {
+                protobuf::parquet_options::FilterCorrelationThresholdOpt::FilterCorrelationThreshold(v) => v,
+            }).unwrap_or(1.5),
+            filter_statistics_collection_min_rows: value.filter_statistics_collection_min_rows_opt.map(|opt| match opt {
+                protobuf::parquet_options::FilterStatisticsCollectionMinRowsOpt::FilterStatisticsCollectionMinRows(v) => v,
+            }).unwrap_or(10_000),
+            filter_statistics_collection_fraction: value.filter_statistics_collection_fraction_opt.map(|opt| match opt {
+                protobuf::parquet_options::FilterStatisticsCollectionFractionOpt::FilterStatisticsCollectionFraction(v) => v,
+            }).unwrap_or(0.0),
         })
     }
 }