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

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.
@@ -1137,6 +1203,11 @@ config_namespace! {
         /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
         pub pushdown_filters: bool, default = false
 
+        /// (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,
         /// the filters are applied in the same order as written in the query

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,

datafusion/datasource-parquet/src/source.rs

@@ -32,6 +32,7 @@ use arrow_schema::{DataType, Field};
 use datafusion_common::config::ConfigOptions;
 #[cfg(feature = "parquet_encryption")]
 use datafusion_common::config::EncryptionFactoryOptions;
+use datafusion_common::config::ParquetPushdownFilterMode;
 use datafusion_datasource::as_file_source;
 use datafusion_datasource::file_stream::FileOpener;
 use datafusion_datasource::morsel::Morselizer;
@@ -46,6 +47,7 @@ use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_functions::core::file_row_index::FileRowIndexFunc;
 use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking};
 use datafusion_physical_expr::projection::ProjectionExprs;
+use datafusion_physical_expr::utils::collect_columns;
 use datafusion_physical_expr::{EquivalenceProperties, conjunction};
 use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
 use datafusion_physical_expr_adapter::rewrite::{
@@ -827,7 +829,79 @@ impl FileSource for ParquetSource {
         // because even if scan pushdown is disabled we can still use the filters for stats pruning.
         let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
         let table_pushdown_enabled = self.pushdown_filters();
-        let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled;
+        let mut pushdown_filters = table_pushdown_enabled || config_pushdown_enabled;
+        // Narrow-projection gate (issue #3463, discussion on PR #23369):
+        // RowFilter has a fixed per-row machinery overhead that only pays
+        // for itself when the wide-column decode it lets us skip is
+        // meaningful. When the projection is narrow and the filter columns
+        // already cover most of it (typical for `GROUP BY col`-style
+        // queries such as ClickBench Q10/Q11/Q40), the overhead dominates
+        // and pushdown regresses. Decline pushdown for those scans; keep
+        // it for wider projections where the fast-path pays.
+        //
+        // Kept intentionally simple: a plan-time column-count check,
+        // used when `pushdown_filter_mode` is `auto` or `heuristic`. A
+        // future adaptive-placement pass ([#22883]) can supersede this
+        // with a runtime cost model behind the same `auto` mode.
+        //
+        // When declined: the filter stays above the scan in a
+        // `FilterExec` (correctness preserved) and the predicate is
+        // still injected into `ParquetSource` for stats / bloom /
+        // page-index pruning.
+        //
+        // Users who want the pre-existing "always push" behavior can
+        // set `pushdown_filter_mode = always`.
+        //
+        // [#22883]: https://github.com/apache/datafusion/issues/22883
+        const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3;
+        let mut narrow_projection_gate_declined = false;
+        // Never gate a scan whose predicate already contains a dynamic
+        // filter — either in the incoming `filters` parameter, or already
+        // installed on `self.predicate` by an earlier optimizer rule (this
+        // is how `TopK`'s heap-threshold expression arrives: the sort
+        // pushdown / TopK rule injects the `DynamicFilterPhysicalExpr`
+        // into the scan's predicate before filter pushdown runs). Dynamic
+        // filters rely on the scan-time RowFilter/RowGroupPruner cascade
+        // to prune data as the threshold tightens, so declining pushdown
+        // here would silently disable the entire dynamic-RG-prune path
+        // for narrow `ORDER BY ... LIMIT` queries.
+        let existing_predicate_has_dynamic = self.predicate.as_ref().is_some_and(|p| {
+            DynamicFilterTracking::classify(p).contains_dynamic_filter()
+        });
+        let incoming_filters_have_dynamic = filters
+            .iter()
+            .any(|f| DynamicFilterTracking::classify(f).contains_dynamic_filter());
+        let has_dynamic_filter =
+            existing_predicate_has_dynamic || incoming_filters_have_dynamic;
+        if pushdown_filters
+            && !has_dynamic_filter
+            && config.execution.parquet.pushdown_filter_mode
+                != ParquetPushdownFilterMode::Always
+        {
+            // `TableSchema` layout is `[file, partition, virtual]`; only
+            // file columns are actually decoded from parquet, so partition
+            // and virtual columns don't count toward the "wide-decode
+            // saving" the RowFilter fast-path buys us.
+            let file_col_count = self.table_schema.file_schema().fields().len();
+            let filter_col_indices: std::collections::HashSet<usize> = filters
+                .iter()
+                .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index()))
+                .filter(|idx| *idx < file_col_count)
+                .collect();
+            let non_filter_projected = self
+                .projection
+                .as_ref()
+                .iter()
+                .flat_map(|pe| collect_columns(&pe.expr))
+                .map(|c| c.index())
+                .filter(|idx| *idx < file_col_count && !filter_col_indices.contains(idx))
+                .collect::<std::collections::HashSet<_>>()
+                .len();
+            if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS {
+                pushdown_filters = false;
+                narrow_projection_gate_declined = true;
+            }
+        }
 
         let mut source = self.clone();
         let filters: Vec<PushedDownPredicate> = filters
@@ -864,7 +938,18 @@ impl FileSource for ParquetSource {
             None => conjunction(allowed_filters),
         };
         source.predicate = Some(predicate);
-        source = source.with_pushdown_filters(pushdown_filters);
+        // Only persist the pushdown_filters bit on the source when this is
+        // a "real" config-driven decision. When the gate declined, we
+        // deliberately do NOT flip the source's pushdown flag to false —
+        // a later `try_pushdown_filters` call (e.g. from TopK's dynamic
+        // filter injection) needs the source to still be pushdown-eligible
+        // so that dynamic filter can install its RowFilter and drive the
+        // runtime RG-prune cascade. The current call still returns
+        // `PushedDown::No` for its incoming filters, so the FilterExec
+        // above the scan stays and correctness is preserved.
+        if !narrow_projection_gate_declined {
+            source = source.with_pushdown_filters(pushdown_filters);
+        }
         let source = Arc::new(source);
         // If pushdown_filters is false we tell our parents that they still have to handle the filters,
         // even if we updated the predicate to include the filters (they will only be used for stats pruning).
@@ -1867,7 +1952,13 @@ mod tests {
         )
         .expect("file_row_index should rewrite to the row_number virtual column");
 
-        let config = ConfigOptions::default();
+        let mut config = ConfigOptions::default();
+        // Force unconditional pushdown so this test can exercise the
+        // virtual-column rejection path independently. The scan projection
+        // here has 0 non-filter file columns, which would otherwise trip
+        // the narrow-projection heuristic and mark every filter as
+        // `PushedDown::No` regardless of virtual-column content.
+        config.execution.parquet.pushdown_filter_mode = ParquetPushdownFilterMode::Always;
         let prop = source
             .try_pushdown_filters(vec![pushable, virtual_only, mixed, row_index], &config)
             .expect("try_pushdown_filters must not error");

datafusion/proto-common/proto/datafusion_common.proto

@@ -558,6 +558,15 @@ message ParquetOptions {
   bool pruning = 2; // default = true
   bool skip_metadata = 3; // default = true
   bool pushdown_filters = 5; // default = false
+  // Strategy for deciding whether to push filters into the parquet scan when
+  // `pushdown_filters` is enabled. Nested so its `ALWAYS` value does not
+  // collide with other file-level enum values.
+  enum PushdownFilterMode {
+    AUTO = 0;
+    ALWAYS = 1;
+    HEURISTIC = 2;
+  }
+  PushdownFilterMode pushdown_filter_mode = 38;
   bool reorder_filters = 6; // default = false
   bool force_filter_selections = 34; // default = false
   uint64 data_pagesize_limit = 7; // default = 1024 * 1024

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

@@ -40,7 +40,8 @@ use datafusion_common::{
     arrow_datafusion_err,
     config::{
         CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions,
-        ParquetColumnOptions, ParquetOptions, TableParquetOptions,
+        ParquetColumnOptions, ParquetOptions, ParquetPushdownFilterMode,
+        TableParquetOptions,
     },
     file_options::{csv_writer::CsvWriterOptions, json_writer::JsonWriterOptions},
     parsers::CompressionTypeVariant,
@@ -964,6 +965,26 @@ impl From<CompressionTypeVariant> for protobuf::CompressionTypeVariant {
     }
 }
 
+impl From<protobuf::parquet_options::PushdownFilterMode> for ParquetPushdownFilterMode {
+    fn from(value: protobuf::parquet_options::PushdownFilterMode) -> Self {
+        match value {
+            protobuf::parquet_options::PushdownFilterMode::Auto => Self::Auto,
+            protobuf::parquet_options::PushdownFilterMode::Always => Self::Always,
+            protobuf::parquet_options::PushdownFilterMode::Heuristic => Self::Heuristic,
+        }
+    }
+}
+
+impl From<ParquetPushdownFilterMode> for protobuf::parquet_options::PushdownFilterMode {
+    fn from(value: ParquetPushdownFilterMode) -> Self {
+        match value {
+            ParquetPushdownFilterMode::Auto => Self::Auto,
+            ParquetPushdownFilterMode::Always => Self::Always,
+            ParquetPushdownFilterMode::Heuristic => Self::Heuristic,
+        }
+    }
+}
+
 impl From<protobuf::CsvQuoteStyle> for datafusion_common::parsers::CsvQuoteStyle {
     fn from(value: protobuf::CsvQuoteStyle) -> Self {
         match value {
@@ -1061,6 +1082,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
                 })
                 .unwrap_or(None),
             pushdown_filters: value.pushdown_filters,
+            pushdown_filter_mode: value.pushdown_filter_mode().into(),
             reorder_filters: value.reorder_filters,
             force_filter_selections: value.force_filter_selections,
             data_pagesize_limit: value.data_pagesize_limit as usize,