feat: adaptive filter selectivity tracking for Parquet row filters

datafusion/common/src/config.rs

@@ -751,6 +751,16 @@ 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
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -209,6 +209,7 @@ impl ParquetOptions {
             coerce_int96: _,     // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
+            filter_effectiveness_threshold: _, // not used for writer props
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -460,6 +461,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,
         }
     }
 
@@ -574,6 +576,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,
             },
             column_specific_options,
             key_value_metadata,

datafusion/core/src/dataframe/parquet.rs

@@ -152,7 +152,14 @@ mod tests {
         let plan = df.explain(false, false)?.collect().await?;
         // Filters all the way to Parquet
         let formatted = pretty::pretty_format_batches(&plan)?.to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"), "{formatted}");
+        let data_source_exec_row = formatted
+            .lines()
+            .find(|line| line.contains("DataSourceExec:"))
+            .unwrap();
+        assert!(
+            data_source_exec_row.contains("predicate=id@0 = 1"),
+            "{formatted}"
+        );
 
         Ok(())
     }

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

@@ -169,7 +169,8 @@ mod tests {
             if self.pushdown_predicate {
                 source = source
                     .with_pushdown_filters(true)
-                    .with_reorder_filters(true);
+                    .with_reorder_filters(true)
+                    .with_filter_effectiveness_threshold(0.0);
             } else {
                 source = source.with_pushdown_filters(false);
             }

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1731,6 +1731,11 @@ async fn test_topk_dynamic_filter_pushdown_integration() {
     let mut cfg = SessionConfig::new();
     cfg.options_mut().execution.parquet.pushdown_filters = true;
     cfg.options_mut().execution.parquet.max_row_group_size = 128;
+    // Always pushdown filters into row filters for this test
+    cfg.options_mut()
+        .execution
+        .parquet
+        .filter_effectiveness_threshold = 0.0;
     let ctx = SessionContext::new_with_config(cfg);
     ctx.register_object_store(
         ObjectStoreUrl::parse("memory://").unwrap().as_ref(),

datafusion/core/tests/sql/explain_analyze.rs

@@ -862,7 +862,7 @@ async fn parquet_explain_analyze() {
         .to_string();
 
     // should contain aggregated stats
-    assert_contains!(&formatted, "output_rows=8");
+    assert_contains!(&formatted, "output_rows=5");
     assert_contains!(
         &formatted,
         "row_groups_pruned_bloom_filter=1 total \u{2192} 1 matched"

datafusion/datasource-parquet/benches/parquet_nested_filter_pushdown.rs

@@ -24,7 +24,9 @@ use arrow::array::{
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use criterion::{Criterion, Throughput, criterion_group, criterion_main};
 use datafusion_common::ScalarValue;
-use datafusion_datasource_parquet::{ParquetFileMetrics, build_row_filter};
+use datafusion_datasource_parquet::{
+    ParquetFileMetrics, SelectivityTracker, build_row_filter,
+};
 use datafusion_expr::{Expr, col};
 use datafusion_functions_nested::expr_fn::array_has;
 use datafusion_physical_expr::planner::logical2physical;
@@ -115,9 +117,14 @@ fn scan_with_predicate(
     let file_metrics = ParquetFileMetrics::new(0, &path.display().to_string(), &metrics);
 
     let builder = if pushdown {
-        if let Some(row_filter) =
-            build_row_filter(predicate, file_schema, &metadata, false, &file_metrics)?
-        {
+        if let Some(row_filter) = build_row_filter(
+            predicate,
+            file_schema,
+            &metadata,
+            false,
+            &file_metrics,
+            &SelectivityTracker::default(),
+        )? {
             builder.with_row_filter(row_filter)
         } else {
             builder

datafusion/datasource-parquet/src/file_format.rs

@@ -507,6 +507,12 @@ impl FileFormat for ParquetFormat {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let mut metadata_size_hint = None;
 
+        let filter_effectiveness_threshold = state
+            .config_options()
+            .execution
+            .parquet
+            .filter_effectiveness_threshold;
+
         if let Some(metadata) = self.metadata_size_hint() {
             metadata_size_hint = Some(metadata);
         }
@@ -518,6 +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);
 
         // Use the CachedParquetFileReaderFactory
         let metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();

datafusion/datasource-parquet/src/metrics.rs

@@ -91,6 +91,8 @@ pub struct ParquetFileMetrics {
     /// number of rows that were stored in the cache after evaluating predicates
     /// reused for the output.
     pub predicate_cache_records: Gauge,
+    //// Time spent applying filters
+    pub filter_apply_time: Time,
 }
 
 impl ParquetFileMetrics {
@@ -186,6 +188,10 @@ impl ParquetFileMetrics {
             .with_new_label("filename", filename.to_string())
             .gauge("predicate_cache_records", partition);
 
+        let filter_apply_time = MetricBuilder::new(metrics)
+            .with_new_label("filename", filename.to_string())
+            .subset_time("filter_apply_time", partition);
+
         Self {
             files_ranges_pruned_statistics,
             predicate_evaluation_errors,
@@ -205,6 +211,7 @@ impl ParquetFileMetrics {
             scan_efficiency_ratio,
             predicate_cache_inner_records,
             predicate_cache_records,
+            filter_apply_time,
         }
     }
 }

datafusion/datasource-parquet/src/mod.rs

@@ -29,6 +29,7 @@ mod page_filter;
 mod reader;
 mod row_filter;
 mod row_group_filter;
+mod selectivity;
 mod sort;
 pub mod source;
 mod supported_predicates;
@@ -39,7 +40,11 @@ pub use file_format::*;
 pub use metrics::ParquetFileMetrics;
 pub use page_filter::PagePruningAccessPlanFilter;
 pub use reader::*; // Expose so downstream crates can use it
+pub use row_filter::FilterMetrics;
+pub use row_filter::RowFilterWithMetrics;
 pub use row_filter::build_row_filter;
+pub use row_filter::build_row_filter_with_metrics;
 pub use row_filter::can_expr_be_pushed_down_with_schemas;
 pub use row_group_filter::RowGroupAccessPlanFilter;
+pub use selectivity::SelectivityTracker;
 pub use writer::plan_to_parquet;