feat: adaptive filter selectivity tracking for Parquet row filters

datafusion/common/src/config.rs

@@ -688,7 +688,7 @@ config_namespace! {
 
         /// (reading) If true, filter expressions are be applied during the parquet decoding operation to
         /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
-        pub pushdown_filters: bool, default = false
+        pub pushdown_filters: bool, default = true
 
         /// (reading) If true, filter expressions evaluated during the parquet decoding operation
         /// will be reordered heuristically to minimize the cost of evaluation. If false,
@@ -732,6 +732,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 = 1.0
+
         // 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()
@@ -464,6 +465,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,
         }
     }
 
@@ -578,6 +580,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

@@ -150,7 +150,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/src/datasource/view_test.rs

@@ -301,7 +301,11 @@ mod tests {
 
     #[tokio::test]
     async fn filter_pushdown_view() -> Result<()> {
-        let ctx = SessionContext::new();
+        // Disable parquet pushdown_filters to ensure filters stay as FilterExec nodes
+        // rather than being pushed into the Parquet reader
+        let config = SessionConfig::new()
+            .set_bool("datafusion.execution.parquet.pushdown_filters", false);
+        let ctx = SessionContext::new_with_config(config);
 
         ctx.register_parquet(
             "test",

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -621,10 +621,10 @@ fn get_value(metrics: &MetricsSet, metric_name: &str) -> usize {
 #[tokio::test]
 async fn predicate_cache_default() -> datafusion_common::Result<()> {
     let ctx = SessionContext::new();
-    // The cache is on by default, but not used unless filter pushdown is enabled
+    // The cache is on by default, and used since pushdown_filters is now true by default
     PredicateCacheTest {
-        expected_inner_records: 0,
-        expected_records: 0,
+        expected_inner_records: 8,
+        expected_records: 7, // reads more than necessary from the cache as then another bitmap is applied
     }
     .run(&ctx)
     .await

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -1721,6 +1721,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"
@@ -995,11 +995,7 @@ async fn parquet_recursive_projection_pushdown() -> Result<()> {
         @r"
     SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
       RecursiveQueryExec: name=number_series, is_distinct=false
-        CoalescePartitionsExec
-          ProjectionExec: expr=[id@0 as id, 1 as level]
-            FilterExec: id@0 = 1
-              RepartitionExec: partitioning=RoundRobinBatch(NUM_CORES), input_partitions=1
-                DataSourceExec: file_groups={1 group: [[TMP_DIR/hierarchy.parquet]]}, projection=[id], file_type=parquet, predicate=id@0 = 1, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1 AND 1 <= id_max@1, required_guarantees=[id in (1)]
+        DataSourceExec: file_groups={1 group: [[TMP_DIR/hierarchy.parquet]]}, projection=[id, 1 as level], file_type=parquet, predicate=id@0 = 1, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1 AND 1 <= id_max@1, required_guarantees=[id in (1)]
         CoalescePartitionsExec
           ProjectionExec: expr=[id@0 + 1 as ns.id + Int64(1), level@1 + 1 as ns.level + Int64(1)]
             FilterExec: id@0 < 10

datafusion/datasource-parquet/src/file_format.rs

@@ -456,6 +456,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);
         }
@@ -467,6 +473,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

@@ -76,6 +76,8 @@ pub struct ParquetFileMetrics {
     /// number of rows that were stored in the cache after evaluating predicates
     /// reused for the output.
     pub predicate_cache_records: Count,
+    //// Time spent applying filters
+    pub filter_apply_time: Time,
 }
 
 impl ParquetFileMetrics {
@@ -162,6 +164,10 @@ impl ParquetFileMetrics {
             .with_new_label("filename", filename.to_string())
             .counter("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,
@@ -179,6 +185,7 @@ impl ParquetFileMetrics {
             scan_efficiency_ratio,
             predicate_cache_inner_records,
             predicate_cache_records,
+            filter_apply_time,
         }
     }
 }

datafusion/datasource-parquet/src/mod.rs

@@ -30,6 +30,7 @@ mod page_filter;
 mod reader;
 mod row_filter;
 mod row_group_filter;
+mod selectivity;
 mod sort;
 pub mod source;
 mod supported_predicates;
@@ -40,7 +41,10 @@ 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 writer::plan_to_parquet;