feat(parquet): Enable Parquet filter_pushdown by default, with heurstic fallback when projecting few non-filter columns

datafusion-examples/examples/data_io/json_shredding.rs

@@ -23,6 +23,7 @@ use arrow::array::{RecordBatch, StringArray};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 
 use datafusion::assert_batches_eq;
+use datafusion::common::config::ParquetPushdownFilterMode;
 use datafusion::common::tree_node::{
     Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
 };
@@ -92,6 +93,11 @@ pub async fn json_shredding() -> Result<()> {
     // Set up query execution
     let mut cfg = SessionConfig::new();
     cfg.options_mut().execution.parquet.pushdown_filters = true;
+    // This example needs the filter pushed into the scan so the JSON
+    // shredding rewriter can rewrite it into direct shredded-column
+    // access. Force pushdown regardless of the projection width.
+    cfg.options_mut().execution.parquet.pushdown_filter_mode =
+        ParquetPushdownFilterMode::Always;
     let ctx = SessionContext::new_with_config(cfg);
     ctx.runtime_env().register_object_store(
         ObjectStoreUrl::parse("memory://")?.as_ref(),

datafusion/common/src/config.rs

@@ -583,6 +583,72 @@ impl Display for SpillCompression {
     }
 }
 
+/// Strategy for filter pushdown in Parquet scan when
+/// `datafusion.execution.parquet.pushdown_filters`
+/// (*[`ParquetOptions::pushdown_filters`]) is enabled
+///
+/// Different strategies are better depending on how data is stored in the
+/// Parquet files and what rows predicates select (e.g. their selectivity and
+/// how many contiguous rows they select).
+///
+/// Note: This option has no effect unless `pushdown_filters` is also enabled.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum ParquetPushdownFilterMode {
+    /// Let DataFusion pick the best available strategy.
+    ///
+    /// Note: currently  identical to [`Self::Heuristic`], but as we implement
+    /// more sophisticated pushdown strategies (e.g. runtime-adaptive placement
+    /// in <https://github.com/apache/datafusion/issues/22883>), this may change.
+    #[default]
+    Auto,
+    /// Always push filters into the scan.
+    Always,
+    /// Use plan-time heuristics to decide which filters to push.
+    ///
+    /// The current heuristic skips pushdown when the projection contains fewer
+    /// than 3 non-filter columns, which avoid narrow-projection queries such as
+    /// `SELECT col2 FROM t WHERE col1 <> ''`, where `RowFilter` overhead
+    /// tends to dominate the decode it would save.
+    Heuristic,
+}
+
+impl FromStr for ParquetPushdownFilterMode {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_ascii_lowercase().as_str() {
+            "auto" | "" => Ok(Self::Auto),
+            "always" => Ok(Self::Always),
+            "heuristic" => Ok(Self::Heuristic),
+            other => Err(DataFusionError::Configuration(format!(
+                "Invalid pushdown filter mode: {other}. Expected one of: auto, always, heuristic"
+            ))),
+        }
+    }
+}
+
+impl ConfigField for ParquetPushdownFilterMode {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = ParquetPushdownFilterMode::from_str(value)?;
+        Ok(())
+    }
+}
+
+impl Display for ParquetPushdownFilterMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let str = match self {
+            Self::Auto => "auto",
+            Self::Always => "always",
+            Self::Heuristic => "heuristic",
+        };
+        write!(f, "{str}")
+    }
+}
+
 /// A `usize` configuration value that rejects zero when set from strings.
 ///
 /// Use this for options where zero is never a meaningful runtime value.
@@ -1135,7 +1201,12 @@ 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) When `pushdown_filters` is enabled, determines how DataFusion
+        /// pushes each filter into the Parquet scan. Options are `auto` (the default)
+        ///  `always`, and `heurstic` (plan time heuristic).
+        pub pushdown_filter_mode: ParquetPushdownFilterMode, default = ParquetPushdownFilterMode::Auto
 
         /// (reading) If true, filter expressions evaluated during the parquet decoding operation
         /// will be reordered heuristically to minimize the cost of evaluation. If false,

datafusion/common/src/file_options/parquet_writer.rs

@@ -236,6 +236,7 @@ impl ParquetOptions {
             skip_metadata: _,
             metadata_size_hint: _,
             pushdown_filters: _,
+            pushdown_filter_mode: _, // reads-only, not used for writer props
             reorder_filters: _,
             force_filter_selections: _, // not used for writer props
             allow_single_file_parallelism: _,
@@ -492,6 +493,7 @@ mod tests {
             skip_metadata: defaults.skip_metadata,
             metadata_size_hint: defaults.metadata_size_hint,
             pushdown_filters: defaults.pushdown_filters,
+            pushdown_filter_mode: defaults.pushdown_filter_mode,
             reorder_filters: defaults.reorder_filters,
             force_filter_selections: defaults.force_filter_selections,
             allow_single_file_parallelism: defaults.allow_single_file_parallelism,
@@ -611,6 +613,7 @@ mod tests {
                 skip_metadata: global_options_defaults.skip_metadata,
                 metadata_size_hint: global_options_defaults.metadata_size_hint,
                 pushdown_filters: global_options_defaults.pushdown_filters,
+                pushdown_filter_mode: global_options_defaults.pushdown_filter_mode,
                 reorder_filters: global_options_defaults.reorder_filters,
                 force_filter_selections: global_options_defaults.force_filter_selections,
                 allow_single_file_parallelism: global_options_defaults

datafusion/core/tests/parquet/mod.rs

@@ -37,6 +37,7 @@ use datafusion::{
     physical_plan::metrics::MetricsSet,
     prelude::{ParquetReadOptions, SessionConfig, SessionContext},
 };
+use datafusion_common::config::ParquetPushdownFilterMode;
 use datafusion_expr::{Expr, LogicalPlan, LogicalPlanBuilder};
 use datafusion_physical_plan::metrics::MetricValue;
 use parquet::arrow::ArrowWriter;
@@ -323,6 +324,10 @@ impl ContextWithParquet {
             Unit::RowGroup(row_per_group) => {
                 config = config.with_parquet_bloom_filter_pruning(true);
                 config.options_mut().execution.parquet.pushdown_filters = true;
+                // force unconditional pushdown to test so the filters are
+                // applied for TopK dynamic RG pruning
+                config.options_mut().execution.parquet.pushdown_filter_mode =
+                    ParquetPushdownFilterMode::Always;
                 make_test_file_rg(
                     scenario,
                     row_per_group,
@@ -340,6 +345,10 @@ impl ContextWithParquet {
                 config = config.with_parquet_bloom_filter_pruning(true);
                 config = config.with_parquet_page_index_pruning(true);
                 config.options_mut().execution.parquet.pushdown_filters = true;
+                // force unconditional pushdown to test so the filters are
+                // applied for TopK dynamic RG pruning
+                config.options_mut().execution.parquet.pushdown_filter_mode =
+                    ParquetPushdownFilterMode::Always;
                 make_test_file_rg(
                     scenario,
                     row_per_group,