add correlation, bytes/sec

Author: adriangb

datafusion/common/src/config.rs

@@ -751,15 +751,30 @@ config_namespace! {
         /// parquet reader setting. 0 means no caching.
         pub max_predicate_cache_size: Option<usize>, default = None
 
-        /// (reading) Minimum filter effectiveness threshold for adaptive filter
-        /// pushdown.
-        /// Only filters that filter out at least this fraction of rows will be
-        /// promoted to row filters during adaptive filter pushdown.
-        /// A value of 1.0 means only filters that filter out all rows will be
-        /// promoted. A value of 0.0 means all filters will be promoted.
-        /// Because there can be a high I/O cost to pushing down ineffective filters,
-        /// recommended values are in the range [0.8, 0.95], depending on random I/0 costs.
-        pub filter_effectiveness_threshold: f64, default = 0.8
+        /// (reading) Minimum bytes/sec throughput for adaptive filter pushdown.
+        /// Filters that achieve at least this throughput (bytes_saved / eval_time)
+        /// are promoted to row filters.
+        /// f64::INFINITY (default) = no filters promoted (feature disabled).
+        /// 0.0 = all filters pushed as row filters (no adaptive logic).
+        pub filter_pushdown_min_bytes_per_sec: f64, default = f64::INFINITY
+
+        /// (reading) Correlation ratio threshold for grouping filters.
+        /// The ratio is P(A ∧ B) / (P(A) * P(B)):
+        ///   1.0 = independent (keep separate for late materialization benefit)
+        ///   1.5 = filters co-pass 50% more often than chance (default threshold)
+        ///   2.0 = filters co-pass twice as often as chance (conservative)
+        /// Higher values = less grouping = more late materialization, more overhead.
+        /// Lower values = more grouping = less overhead, less late materialization.
+        /// Set to f64::MAX to disable grouping entirely.
+        pub filter_correlation_threshold: f64, default = 1.5
+
+        /// (reading) Minimum rows of post-scan evaluation before statistics-based
+        /// optimization activates. During collection, all filters are evaluated
+        /// as post-scan to gather accurate marginal and joint selectivity statistics.
+        /// Used for BOTH individual filter effectiveness decisions AND correlation-
+        /// based grouping. Larger values = more accurate estimates, longer collection.
+        /// Set to 0 to disable the collection phase entirely.
+        pub filter_statistics_collection_min_rows: u64, default = 10_000
 
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties

datafusion/common/src/file_options/parquet_writer.rs

@@ -209,7 +209,9 @@ impl ParquetOptions {
             coerce_int96: _,     // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
-            filter_effectiveness_threshold: _, // not used for writer props
+            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
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -461,7 +463,7 @@ mod tests {
             skip_arrow_metadata: defaults.skip_arrow_metadata,
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
-            filter_effectiveness_threshold: defaults.filter_effectiveness_threshold,
+            filter_pushdown_min_bytes_per_sec: defaults.filter_pushdown_min_bytes_per_sec,
         }
     }
 
@@ -576,8 +578,8 @@ mod tests {
                 binary_as_string: global_options_defaults.binary_as_string,
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
-                filter_effectiveness_threshold: global_options_defaults
-                    .filter_effectiveness_threshold,
+                filter_pushdown_min_bytes_per_sec: global_options_defaults
+                    .filter_pushdown_min_bytes_per_sec,
             },
             column_specific_options,
             key_value_metadata,

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -170,7 +170,7 @@ mod tests {
                 source = source
                     .with_pushdown_filters(true)
                     .with_reorder_filters(true)
-                    .with_filter_effectiveness_threshold(0.0);
+                    .with_filter_pushdown_min_bytes_per_sec(0.0);
             } else {
                 source = source.with_pushdown_filters(false);
             }

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1735,7 +1735,7 @@ async fn test_topk_dynamic_filter_pushdown_integration() {
     cfg.options_mut()
         .execution
         .parquet
-        .filter_effectiveness_threshold = 0.0;
+        .filter_pushdown_min_bytes_per_sec = 0.0;
     let ctx = SessionContext::new_with_config(cfg);
     ctx.register_object_store(
         ObjectStoreUrl::parse("memory://").unwrap().as_ref(),

datafusion/datasource-parquet/src/file_format.rs

@@ -507,11 +507,11 @@ impl FileFormat for ParquetFormat {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let mut metadata_size_hint = None;
 
-        let filter_effectiveness_threshold = state
+        let filter_pushdown_min_bytes_per_sec = state
             .config_options()
             .execution
             .parquet
-            .filter_effectiveness_threshold;
+            .filter_pushdown_min_bytes_per_sec;
 
         if let Some(metadata) = self.metadata_size_hint() {
             metadata_size_hint = Some(metadata);
@@ -524,7 +524,7 @@ impl FileFormat for ParquetFormat {
             .cloned()
             .ok_or_else(|| internal_datafusion_err!("Expected ParquetSource"))?;
         source = source.with_table_parquet_options(self.options.clone());
-        source = source.with_filter_pushdown_selectivity(filter_effectiveness_threshold);
+        source = source.with_filter_pushdown_min_bytes_per_sec(filter_pushdown_min_bytes_per_sec);
 
         // Use the CachedParquetFileReaderFactory
         let metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();