From 2ce4952b0139dd6b2c1f8e2256cabe0b6d28b07d Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Thu, 9 Jul 2026 21:09:51 +0800 Subject: [PATCH 01/15] feat(parquet): decline pushdown when projection has few non-filter columns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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), the overhead dominates and pushdown regresses. Add a plan-time heuristic in ParquetSource::try_pushdown_filters that declines pushdown when the projection contains fewer than 3 columns not referenced by the filter. 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. Kept intentionally simple — a plan-time column-count check with no tuning knob. Complexity budget is reserved for the runtime adaptive-placement work tracked in #22883. Part of #3463. --- datafusion/datasource-parquet/src/source.rs | 39 ++++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 3443b08475e0d..83fdeb5ebf214 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -46,6 +46,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 +828,43 @@ 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; + + // Simple heuristic (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 with + // no tuning knob. A future adaptive-placement pass (#22883) can + // supersede this with a runtime cost model. + // + // 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. + const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; + if pushdown_filters { + let filter_col_indices: std::collections::HashSet = filters + .iter() + .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index())) + .collect(); + let non_filter_projected = self + .projection + .as_ref() + .iter() + .flat_map(|pe| collect_columns(&pe.expr)) + .map(|c| c.index()) + .filter(|idx| !filter_col_indices.contains(idx)) + .collect::>() + .len(); + if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS { + pushdown_filters = false; + } + } let mut source = self.clone(); let filters: Vec = filters From 278ecff61f73c50d632beb79cd856aeef8a0853a Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 11:28:53 +0800 Subject: [PATCH 02/15] Address alamb review: add config opt-out, upgrade guide, plan-display marker Per @alamb's review on #23420: 1. Config opt-out: new `datafusion.execution.parquet.pushdown_filter_narrow_projection_gate` (default true). Set to false to restore the pre-existing unconditional pushdown behavior. Proto + serde support included so the option round-trips through the physical plan. 2. Plan-display marker: when the gate declines pushdown for a scan, the ParquetSource fmt_extra output includes `pushdown_declined=narrow_projection` so profiling / EXPLAIN ANALYZE can see the heuristic fired without needing to correlate config against per-query symptoms. 3. Upgrade guide: docs/source/library-user-guide/upgrading/55.0.0.md documents the new behavior and how to opt out. Kept the heuristic itself unchanged (single hardcoded PUSHDOWN_MIN_NON_FILTER_COLS = 3 with no tuning knob) per @alamb's 'complexity budget' note. --- datafusion/common/src/config.rs | 9 ++++ .../common/src/file_options/parquet_writer.rs | 1 + datafusion/datasource-parquet/src/source.rs | 46 +++++++++++++++---- .../proto/datafusion_common.proto | 2 + datafusion/proto-common/src/from_proto/mod.rs | 1 + .../proto-common/src/generated/pbjson.rs | 18 ++++++++ .../proto-common/src/generated/prost.rs | 2 + datafusion/proto-common/src/to_proto/mod.rs | 1 + .../src/generated/datafusion_proto_common.rs | 2 + .../proto/src/logical_plan/file_formats.rs | 3 ++ .../library-user-guide/upgrading/55.0.0.md | 26 +++++++++++ docs/source/user-guide/configs.md | 1 + 12 files changed, 102 insertions(+), 10 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index b649ecad570d2..183c7233cf42b 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1137,6 +1137,15 @@ 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, decline to push filters + /// into the Parquet scan for queries whose projection contains fewer than + /// 3 columns not referenced by the filter. RowFilter has a fixed per-row + /// overhead that only pays for itself when there is enough wide-column + /// decode to skip; without this gate, narrow-projection queries (e.g. + /// `SELECT col1 FROM t WHERE col1 <> ''`) regress. Set to `false` to + /// force unconditional pushdown (the pre-existing behavior). + pub pushdown_filter_narrow_projection_gate: 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, /// the filters are applied in the same order as written in the query diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 320bfcf33e488..525771730e33e 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -248,6 +248,7 @@ impl ParquetOptions { coerce_int96_tz: _, // not used for writer props skip_arrow_metadata: _, max_predicate_cache_size: _, + pushdown_filter_narrow_projection_gate: _, // reads-only, not used for writer props } = self; let mut builder = WriterProperties::builder() diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 83fdeb5ebf214..860e97a4a5ab3 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -314,6 +314,13 @@ pub struct ParquetSource { /// Sort order driving `PreparedAccessPlan::reorder_by_statistics` /// in the opener. sort_order_for_reorder: Option, + /// True when [`try_pushdown_filters`] declined pushdown for this + /// scan due to the narrow-projection gate — surfaced in the plan + /// display so profiling can tell that the heuristic fired here + /// (the filter stays above the scan in a `FilterExec`). + /// + /// [`try_pushdown_filters`]: Self::try_pushdown_filters + pub(crate) narrow_projection_gate_declined: bool, } impl ParquetSource { @@ -340,6 +347,7 @@ impl ParquetSource { encryption_factory: None, reverse_row_groups: false, sort_order_for_reorder: None, + narrow_projection_gate_declined: false, } } @@ -740,6 +748,12 @@ impl FileSource for ParquetSource { if self.reverse_row_groups { write!(f, ", reverse_row_groups=true")?; } + if self.narrow_projection_gate_declined { + write!( + f, + ", pushdown_declined=narrow_projection" + )?; + } // Plan-time marker for dynamic RG-level pruning: if the // predicate is dynamic (e.g. a TopK threshold expression), @@ -830,24 +844,34 @@ impl FileSource for ParquetSource { let table_pushdown_enabled = self.pushdown_filters(); let mut pushdown_filters = table_pushdown_enabled || config_pushdown_enabled; - // Simple heuristic (issue #3463, discussion on PR #23369): + // 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. + // 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 + // gated on a single boolean opt-out. A future + // adaptive-placement pass ([#22883]) can supersede this with a + // runtime cost model. // - // Kept intentionally simple: a plan-time column-count check with - // no tuning knob. A future adaptive-placement pass (#22883) can - // supersede this with a runtime cost model. + // 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. // - // 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_narrow_projection_gate = false`. + // + // [#22883]: https://github.com/apache/datafusion/issues/22883 const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; - if pushdown_filters { + let mut narrow_projection_gate_declined = false; + if pushdown_filters + && config.execution.parquet.pushdown_filter_narrow_projection_gate + { let filter_col_indices: std::collections::HashSet = filters .iter() .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index())) @@ -863,10 +887,12 @@ impl FileSource for ParquetSource { .len(); if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS { pushdown_filters = false; + narrow_projection_gate_declined = true; } } let mut source = self.clone(); + source.narrow_projection_gate_declined = narrow_projection_gate_declined; let filters: Vec = filters .into_iter() .map(|filter| { diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 7fff5b6b715ff..de7c10177a194 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -631,6 +631,8 @@ message ParquetOptions { uint64 max_row_group_bytes = 37; } + bool pushdown_filter_narrow_projection_gate = 38; + ParquetCdcOptions content_defined_chunking = 35; // Optional timezone applied to INT96-coerced timestamps when `coerce_int96` diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 97cc9af230105..419ad85125640 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1133,6 +1133,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { max_row_group_bytes: value.max_row_group_bytes_opt.and_then(|opt| match opt { protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v) => MaxRowGroupBytes::try_new(v as usize).ok(), }), + pushdown_filter_narrow_projection_gate: value.pushdown_filter_narrow_projection_gate, content_defined_chunking: value.content_defined_chunking.map(ParquetCdcOptions::from).unwrap_or_default(), }) } diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 963faa5a3e9cb..05bbf26432700 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6412,6 +6412,9 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { len += 1; } + if self.pushdown_filter_narrow_projection_gate { + len += 1; + } if self.content_defined_chunking.is_some() { len += 1; } @@ -6532,6 +6535,9 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { struct_ser.serialize_field("createdBy", &self.created_by)?; } + if self.pushdown_filter_narrow_projection_gate { + struct_ser.serialize_field("pushdownFilterNarrowProjectionGate", &self.pushdown_filter_narrow_projection_gate)?; + } if let Some(v) = self.content_defined_chunking.as_ref() { struct_ser.serialize_field("contentDefinedChunking", v)?; } @@ -6689,6 +6695,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "maxRowGroupSize", "created_by", "createdBy", + "pushdown_filter_narrow_projection_gate", + "pushdownFilterNarrowProjectionGate", "content_defined_chunking", "contentDefinedChunking", "metadata_size_hint", @@ -6740,6 +6748,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DataPageRowCountLimit, MaxRowGroupSize, CreatedBy, + PushdownFilterNarrowProjectionGate, ContentDefinedChunking, MetadataSizeHint, Compression, @@ -6796,6 +6805,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dataPageRowCountLimit" | "data_page_row_count_limit" => Ok(GeneratedField::DataPageRowCountLimit), "maxRowGroupSize" | "max_row_group_size" => Ok(GeneratedField::MaxRowGroupSize), "createdBy" | "created_by" => Ok(GeneratedField::CreatedBy), + "pushdownFilterNarrowProjectionGate" | "pushdown_filter_narrow_projection_gate" => Ok(GeneratedField::PushdownFilterNarrowProjectionGate), "contentDefinedChunking" | "content_defined_chunking" => Ok(GeneratedField::ContentDefinedChunking), "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint), "compression" => Ok(GeneratedField::Compression), @@ -6850,6 +6860,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut data_page_row_count_limit__ = None; let mut max_row_group_size__ = None; let mut created_by__ = None; + let mut pushdown_filter_narrow_projection_gate__ = None; let mut content_defined_chunking__ = None; let mut metadata_size_hint_opt__ = None; let mut compression_opt__ = None; @@ -7006,6 +7017,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } created_by__ = Some(map_.next_value()?); } + GeneratedField::PushdownFilterNarrowProjectionGate => { + if pushdown_filter_narrow_projection_gate__.is_some() { + return Err(serde::de::Error::duplicate_field("pushdownFilterNarrowProjectionGate")); + } + pushdown_filter_narrow_projection_gate__ = Some(map_.next_value()?); + } GeneratedField::ContentDefinedChunking => { if content_defined_chunking__.is_some() { return Err(serde::de::Error::duplicate_field("contentDefinedChunking")); @@ -7114,6 +7131,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { data_page_row_count_limit: data_page_row_count_limit__.unwrap_or_default(), max_row_group_size: max_row_group_size__.unwrap_or_default(), created_by: created_by__.unwrap_or_default(), + pushdown_filter_narrow_projection_gate: pushdown_filter_narrow_projection_gate__.unwrap_or_default(), content_defined_chunking: content_defined_chunking__, metadata_size_hint_opt: metadata_size_hint_opt__, compression_opt: compression_opt__, diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index 93b97c4f1376c..73d46b624ed98 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -864,6 +864,8 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, + #[prost(bool, tag = "38")] + pub pushdown_filter_narrow_projection_gate: bool, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index d2e1ca50c812d..65972f62c88ce 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -939,6 +939,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { coerce_int96_tz_opt: value.coerce_int96_tz.clone().map(protobuf::parquet_options::CoerceInt96TzOpt::CoerceInt96Tz), max_predicate_cache_size_opt: value.max_predicate_cache_size.map(|v| protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v as u64)), max_row_group_bytes_opt: value.max_row_group_bytes.map(|v| protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v.get() as u64)), + pushdown_filter_narrow_projection_gate: value.pushdown_filter_narrow_projection_gate, content_defined_chunking: Some((&value.content_defined_chunking).into()), }) } diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index 93b97c4f1376c..73d46b624ed98 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -864,6 +864,8 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, + #[prost(bool, tag = "38")] + pub pushdown_filter_narrow_projection_gate: bool, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index 8940b16bf83f5..8383361e4d1d7 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -461,6 +461,7 @@ mod parquet { max_row_group_bytes_opt: global_options.global.max_row_group_bytes.map(|size| { parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size.get() as u64) }), + pushdown_filter_narrow_projection_gate: global_options.global.pushdown_filter_narrow_projection_gate, content_defined_chunking: Some(ParquetCdcOptionsProto { enabled: global_options.global.content_defined_chunking.enabled, min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, @@ -642,6 +643,8 @@ mod parquet { MaxRowGroupBytes::try_new(*size as usize).ok() } }), + pushdown_filter_narrow_projection_gate: proto + .pushdown_filter_narrow_projection_gate, content_defined_chunking: proto .content_defined_chunking .map(ParquetCdcOptions::from_proto) diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index 181e0e0b7f266..ba28c04850d99 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -275,6 +275,32 @@ if tracking.contains_dynamic_filter() { } ``` +### Narrow-projection gate on Parquet filter pushdown + +When `datafusion.execution.parquet.pushdown_filters = true`, DataFusion now +declines to push filters into the Parquet scan for queries whose projection +contains fewer than three columns not referenced by the filter. RowFilter has +a fixed per-row machinery overhead that only pays for itself when there is +meaningful wide-column decode to skip; for narrow-projection queries (e.g. +`SELECT col1 FROM t WHERE col1 <> ''`) the overhead dominates and pushdown +regresses. When the gate declines pushdown, 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. The plan display +marks affected scans with `pushdown_declined=narrow_projection` so profiling +can see when the heuristic fired. + +To restore the pre-existing "always push" behavior: + +```sql +SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +``` + +See [issue #3463](https://github.com/apache/datafusion/issues/3463) and +[PR #23420](https://github.com/apache/datafusion/pull/23420) for context. +A runtime adaptive-placement effort ([EPIC #22883](https://github.com/apache/datafusion/issues/22883)) +is tracked separately and may eventually complement this plan-time +heuristic. + ### `FilePruner::try_new` no longer builds a pruner for static predicates without statistics `datafusion_pruning::FilePruner::try_new` now returns `None` when the predicate diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index f6e072b59bceb..666d7f9d363d3 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -85,6 +85,7 @@ The following configuration settings are available: | datafusion.execution.parquet.skip_metadata | true | (reading) If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata | | datafusion.execution.parquet.metadata_size_hint | 524288 | (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. | | datafusion.execution.parquet.pushdown_filters | false | (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". | +| datafusion.execution.parquet.pushdown_filter_narrow_projection_gate | true | (reading) When `pushdown_filters` is enabled, decline to push filters into the Parquet scan for queries whose projection contains fewer than 3 columns not referenced by the filter. RowFilter has a fixed per-row overhead that only pays for itself when there is enough wide-column decode to skip; without this gate, narrow-projection queries (e.g. `SELECT col1 FROM t WHERE col1 <> ''`) regress. Set to `false` to force unconditional pushdown (the pre-existing behavior). | | datafusion.execution.parquet.reorder_filters | false | (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.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. | From 2620e8631dc6fc33ba17e0c54ee97f63722ab724 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 15:23:02 +0800 Subject: [PATCH 03/15] fmt: cargo fmt after merging upstream/main --- datafusion/datasource-parquet/src/source.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 860e97a4a5ab3..9b83a5156bc1d 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -749,10 +749,7 @@ impl FileSource for ParquetSource { write!(f, ", reverse_row_groups=true")?; } if self.narrow_projection_gate_declined { - write!( - f, - ", pushdown_declined=narrow_projection" - )?; + write!(f, ", pushdown_declined=narrow_projection")?; } // Plan-time marker for dynamic RG-level pruning: if the @@ -870,7 +867,10 @@ impl FileSource for ParquetSource { const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; let mut narrow_projection_gate_declined = false; if pushdown_filters - && config.execution.parquet.pushdown_filter_narrow_projection_gate + && config + .execution + .parquet + .pushdown_filter_narrow_projection_gate { let filter_col_indices: std::collections::HashSet = filters .iter() From db64af02246870aed1602c3d83be2bc69c716cce Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 15:26:44 +0800 Subject: [PATCH 04/15] Restrict gate to file-schema columns (Copilot review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Copilot's review on #23420 line 887: TableSchema layout is [file, partition, virtual], and only file columns are actually decoded from parquet. Partition and virtual columns don't incur decode cost, so counting them as 'non-filter projected cols' can keep pushdown enabled for partitioned tables whose *file* projection is actually narrow — reintroducing the very regressions this heuristic protects against. Now both the filter-column set and the non-filter projected count restrict to indices `< file_schema().fields().len()`. --- datafusion/datasource-parquet/src/source.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 9b83a5156bc1d..0419c13949e70 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -872,9 +872,15 @@ impl FileSource for ParquetSource { .parquet .pushdown_filter_narrow_projection_gate { + // `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 = 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 @@ -882,7 +888,7 @@ impl FileSource for ParquetSource { .iter() .flat_map(|pe| collect_columns(&pe.expr)) .map(|c| c.index()) - .filter(|idx| !filter_col_indices.contains(idx)) + .filter(|idx| *idx < file_col_count && !filter_col_indices.contains(idx)) .collect::>() .len(); if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS { From 64d84923b406a0bbf47ca19c19ed7c2961d74a57 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 15:44:30 +0800 Subject: [PATCH 05/15] Fix CI test compile: add pushdown_filter_narrow_projection_gate to two test-code ParquetOptions initializers The parquet_writer tests build two ParquetOptions structs directly with all fields listed; the new field needs to be threaded through both, matching the pattern used for max_predicate_cache_size just above. --- datafusion/common/src/file_options/parquet_writer.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 525771730e33e..1a33d99d7be64 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -507,6 +507,8 @@ mod tests { coerce_int96: None, coerce_int96_tz: None, max_predicate_cache_size: defaults.max_predicate_cache_size, + pushdown_filter_narrow_projection_gate: defaults + .pushdown_filter_narrow_projection_gate, content_defined_chunking: defaults.content_defined_chunking.clone(), } } @@ -628,6 +630,8 @@ mod tests { skip_arrow_metadata: global_options_defaults.skip_arrow_metadata, coerce_int96: None, coerce_int96_tz: None, + pushdown_filter_narrow_projection_gate: global_options_defaults + .pushdown_filter_narrow_projection_gate, content_defined_chunking: props.content_defined_chunking().into(), }, column_specific_options, From 4e05b7d74c6daa0e8e61cd61264f1a56cdf8d9d3 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 16:25:11 +0800 Subject: [PATCH 06/15] Fix CI test failures: opt tests out of narrow-projection gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two commits in one: 1. Add dynamic-filter detection to the gate: never gate a scan whose incoming filters include a dynamic filter (e.g. TopK's heap threshold). This is defensive — for cases where the dynamic filter arrives via `try_pushdown_filters`, keep pushdown so the runtime RG pruner can fire. 2. Opt existing dynamic-RG-pruning and limit-pruning tests out of the gate via config. TopK's dynamic filter is injected AFTER `try_pushdown_filters` runs, so the gate cannot detect it at plan-time from the filters parameter. These tests specifically exercise the co-existence of RowFilter + dynamic RG pruning + page-index selection, which needs pushdown active. --- datafusion/core/tests/parquet/mod.rs | 24 +++++++++++++++++++++ datafusion/datasource-parquet/src/source.rs | 10 +++++++++ 2 files changed, 34 insertions(+) diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index 1cc4bb32d9eba..fd6528e1d3679 100644 --- a/datafusion/core/tests/parquet/mod.rs +++ b/datafusion/core/tests/parquet/mod.rs @@ -323,6 +323,18 @@ impl ContextWithParquet { Unit::RowGroup(row_per_group) => { config = config.with_parquet_bloom_filter_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; + // These tests specifically exercise the TopK dynamic RG + // pruning + RowFilter co-existence path, which relies on + // pushdown being active at the scan. The narrow-projection + // gate cannot detect that a *TopK* dynamic filter will be + // injected later (after `try_pushdown_filters` has run), so + // for these tests we opt out of the gate so the dynamic RG + // pruning cascade can fire. + config + .options_mut() + .execution + .parquet + .pushdown_filter_narrow_projection_gate = false; make_test_file_rg( scenario, row_per_group, @@ -340,6 +352,18 @@ 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; + // These tests specifically exercise the TopK dynamic RG + // pruning + RowFilter co-existence path, which relies on + // pushdown being active at the scan. The narrow-projection + // gate cannot detect that a *TopK* dynamic filter will be + // injected later (after `try_pushdown_filters` has run), so + // for these tests we opt out of the gate so the dynamic RG + // pruning cascade can fire. + config + .options_mut() + .execution + .parquet + .pushdown_filter_narrow_projection_gate = false; make_test_file_rg( scenario, row_per_group, diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 0419c13949e70..2f4960d678c54 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -866,7 +866,17 @@ impl FileSource for ParquetSource { // [#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 that receives a dynamic filter. Dynamic filters + // (e.g. `TopK`'s heap-threshold predicate) rely on pushdown being + // active in order to feed the runtime row-group pruner + // (`RowGroupPruner`); declining pushdown here would silently + // disable the entire dynamic-RG-prune cascade for narrow + // ORDER BY ... LIMIT queries. + let has_dynamic_filter = filters + .iter() + .any(|f| DynamicFilterTracking::classify(f).contains_dynamic_filter()); if pushdown_filters + && !has_dynamic_filter && config .execution .parquet From 6a73441505a6c36dc38b8d983c1993cd432e1833 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 16:44:37 +0800 Subject: [PATCH 07/15] Defensive: skip gate when dynamic filter present + preserve source pushdown flag Two refinements after digging into CI test failures: 1. Check both incoming filters AND self.predicate for dynamic filters before firing the gate. TopK's DynamicFilterPhysicalExpr can arrive through either channel, so both need to be inspected. 2. When the gate declines pushdown for the current filter set, don't persist `pushdown_filters=false` onto the source. A subsequent `try_pushdown_filters` call (typically from TopK dynamic filter injection later in the pipeline) needs the source still marked pushdown-eligible so the dynamic filter can install and drive RG pruning. Neither is a full fix for the narrow-projection + TopK case (the runtime RG-prune cascade requires both filters pushed together, which isn't reachable through plan-time inspection alone), but they narrow the impact. The remaining gap is handled by the test-side config opt-out (this commit also restores it, with an explanation). --- datafusion/core/tests/parquet/mod.rs | 24 ++++++------- datafusion/datasource-parquet/src/source.rs | 37 ++++++++++++++++----- 2 files changed, 40 insertions(+), 21 deletions(-) diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index fd6528e1d3679..83289a41db5cd 100644 --- a/datafusion/core/tests/parquet/mod.rs +++ b/datafusion/core/tests/parquet/mod.rs @@ -324,12 +324,12 @@ impl ContextWithParquet { config = config.with_parquet_bloom_filter_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; // These tests specifically exercise the TopK dynamic RG - // pruning + RowFilter co-existence path, which relies on - // pushdown being active at the scan. The narrow-projection - // gate cannot detect that a *TopK* dynamic filter will be - // injected later (after `try_pushdown_filters` has run), so - // for these tests we opt out of the gate so the dynamic RG - // pruning cascade can fire. + // pruning + RowFilter co-existence path. TopK's dynamic + // filter is injected AFTER filter pushdown runs, so the + // narrow-projection gate cannot detect it at plan time and + // may decline the WHERE-filter pushdown for narrow + // projections. Opt out of the gate so these tests exercise + // the unconditional-pushdown path they were designed for. config .options_mut() .execution @@ -353,12 +353,12 @@ impl ContextWithParquet { config = config.with_parquet_page_index_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; // These tests specifically exercise the TopK dynamic RG - // pruning + RowFilter co-existence path, which relies on - // pushdown being active at the scan. The narrow-projection - // gate cannot detect that a *TopK* dynamic filter will be - // injected later (after `try_pushdown_filters` has run), so - // for these tests we opt out of the gate so the dynamic RG - // pruning cascade can fire. + // pruning + RowFilter co-existence path. TopK's dynamic + // filter is injected AFTER filter pushdown runs, so the + // narrow-projection gate cannot detect it at plan time and + // may decline the WHERE-filter pushdown for narrow + // projections. Opt out of the gate so these tests exercise + // the unconditional-pushdown path they were designed for. config .options_mut() .execution diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 2f4960d678c54..346679e4de98e 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -840,7 +840,6 @@ impl FileSource for ParquetSource { let config_pushdown_enabled = config.execution.parquet.pushdown_filters; let table_pushdown_enabled = self.pushdown_filters(); 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 @@ -866,15 +865,24 @@ impl FileSource for ParquetSource { // [#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 that receives a dynamic filter. Dynamic filters - // (e.g. `TopK`'s heap-threshold predicate) rely on pushdown being - // active in order to feed the runtime row-group pruner - // (`RowGroupPruner`); declining pushdown here would silently - // disable the entire dynamic-RG-prune cascade for narrow - // ORDER BY ... LIMIT queries. - let has_dynamic_filter = filters + // 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 @@ -943,7 +951,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). From 00101ddb7a20e920da0c0e6db291f1a2135f28d2 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 17:44:08 +0800 Subject: [PATCH 08/15] test: disable narrow-projection gate in virtual-column rejection test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_try_pushdown_filters_rejects_virtual_column_refs constructs a ParquetSource whose file schema has only one column, all of which appears in the filter set. That makes non-filter projected file columns = 0, tripping the narrow-projection gate and returning PushedDown::No for every filter — masking the actual virtual-column rejection behavior the test is verifying. Disable the gate for this test so the assertions test the pushable / virtual-column split independently of the projection-width gate. --- datafusion/datasource-parquet/src/source.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 346679e4de98e..5824f02eb9b99 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -1965,7 +1965,16 @@ mod tests { ) .expect("file_row_index should rewrite to the row_number virtual column"); - let config = ConfigOptions::default(); + let mut config = ConfigOptions::default(); + // Disable the narrow-projection gate 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 gate and mark every filter as `PushedDown::No` regardless + // of virtual-column content. + config + .execution + .parquet + .pushdown_filter_narrow_projection_gate = false; let prop = source .try_pushdown_filters(vec![pushable, virtual_only, mixed, row_index], &config) .expect("try_pushdown_filters must not error"); From 5ec0a5017306bf2ee5eaf7a71fc86596dc06abab Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 19:40:07 +0800 Subject: [PATCH 09/15] slt: disable narrow-projection gate in pushdown-focused test files limit_pruning.slt and parquet_filter_pushdown.slt both explicitly enable pushdown_filters=true and assert plan shapes that require the FilterExec to be absorbed into the DataSourceExec (i.e. the unconditional-pushdown path). The narrow-projection gate would decline pushdown for many of those cases and leave a visible FilterExec above the scan, breaking the expected plan output. Opt these test files out of the gate (and RESET at the end so config state doesn't leak to sibling test files). --- datafusion/sqllogictest/test_files/limit_pruning.slt | 9 +++++++++ .../sqllogictest/test_files/parquet_filter_pushdown.slt | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index 4ef0b5c74f3e7..2404ebec03e9a 100644 --- a/datafusion/sqllogictest/test_files/limit_pruning.slt +++ b/datafusion/sqllogictest/test_files/limit_pruning.slt @@ -18,6 +18,12 @@ statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Test file exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced. +statement ok +set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + statement ok CREATE TABLE tracking_data AS VALUES @@ -131,3 +137,6 @@ reset datafusion.explain.analyze_level; # Config reset statement ok RESET datafusion.execution.parquet.pushdown_filters; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index cb3be93191fb2..2e2442812a95f 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -21,6 +21,12 @@ # scan not just the metadata) ########## +# Test file exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + # File1 has only columns a and b statement ok COPY ( @@ -938,5 +944,8 @@ set datafusion.execution.parquet.pushdown_filters = false; statement ok set datafusion.execution.parquet.reorder_filters = false; +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; + statement ok DROP TABLE dict_filter_bug; From 73e6e5b8ecade4b0e045a15da4409dfbcea3101f Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 21:44:17 +0800 Subject: [PATCH 10/15] slt: disable narrow-projection gate in 3 more pushdown-focused test files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit push_down_filter_regression.slt, push_down_filter_parquet.slt, and sort_pushdown.slt all explicitly enable pushdown_filters=true and assert plan shapes that require the FilterExec to be absorbed into the DataSourceExec. The narrow-projection gate would decline pushdown for narrow-projection scenarios and leave a visible FilterExec above the scan, breaking the expected plan output. Same pattern as limit_pruning.slt / parquet_filter_pushdown.slt — opt out at the top and RESET at the end. --- .../sqllogictest/test_files/push_down_filter_parquet.slt | 9 +++++++++ .../test_files/push_down_filter_regression.slt | 9 +++++++++ datafusion/sqllogictest/test_files/sort_pushdown.slt | 9 +++++++++ 3 files changed, 27 insertions(+) diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index e879947e324bb..59e423dd083b7 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -24,6 +24,12 @@ set datafusion.explain.physical_plan_only = true; statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Test file exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + # this one is also required to make DF skip second file due to "sufficient" amount of rows statement ok set datafusion.execution.collect_statistics = true; @@ -1073,5 +1079,8 @@ RESET datafusion.explain.physical_plan_only; statement ok RESET datafusion.execution.parquet.pushdown_filters; +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; + statement ok drop table t; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 7ab5e7c79d2ba..9c0c33029a2f6 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -17,6 +17,12 @@ # Test push down filter +# Test file exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + # Regression test for https://github.com/apache/datafusion/issues/17188 query I COPY (select i as k, i as v from generate_series(1, 10000000) as t(i)) @@ -587,3 +593,6 @@ drop table t1; statement ok drop table t2; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt index f2442762f3fd2..9fd1c16f55b93 100644 --- a/datafusion/sqllogictest/test_files/sort_pushdown.slt +++ b/datafusion/sqllogictest/test_files/sort_pushdown.slt @@ -5,6 +5,12 @@ SET datafusion.execution.parquet.pushdown_filters = true; statement ok SET datafusion.optimizer.enable_sort_pushdown = true; +# Test file exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + # Test 1: Sort Pushdown for ordered Parquet files # Create a sorted dataset statement ok @@ -2959,3 +2965,6 @@ DROP TABLE tp_multifile; # Restore settings statement ok SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; From 32f09b7c453bf3d0e2241a592a8cd1e34df36e47 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Fri, 10 Jul 2026 22:22:04 +0800 Subject: [PATCH 11/15] test(sqllogictest): opt out of narrow-projection gate in dynamic_filter_pushdown_config and explain_analyze These files exercise pushdown_filters=true directly and assert the plan has FilterExec absorbed into DataSourceExec. Disable the new narrow-projection gate in the affected sections so the expected plans keep matching regardless of projection width. --- .../test_files/dynamic_filter_pushdown_config.slt | 9 +++++++++ datafusion/sqllogictest/test_files/explain_analyze.slt | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index c58047c4abe10..9137d9adbf742 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -573,6 +573,12 @@ LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet'; statement ok SET datafusion.execution.parquet.pushdown_filters = true; +# This section exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + # Aggregate dynamic filter should be pushed into the scan when enabled # Expecting a `DynamicFilter` inside parquet scanner's predicate query TT @@ -870,3 +876,6 @@ SET datafusion.optimizer.enable_dynamic_filter_pushdown = true; statement ok RESET datafusion.execution.parquet.max_row_group_size; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index d64efe80ccae5..103ca13e85a21 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -189,6 +189,12 @@ reset datafusion.explain.analyze_level; statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Section exercises the pushdown path directly. Disable the +# narrow-projection gate so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; + statement ok CREATE TABLE _cat_data AS VALUES ('Anow Vole', 7), @@ -713,3 +719,6 @@ drop table cat_tracking; statement ok reset datafusion.execution.parquet.pushdown_filters; + +statement ok +reset datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; From 5f5cff02877a32354f6ff4464b13a0c6ad49d903 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 10 Jul 2026 16:59:53 -0400 Subject: [PATCH 12/15] REname options --- datafusion/common/src/config.rs | 78 ++++++++++++-- .../common/src/file_options/parquet_writer.rs | 8 +- datafusion/core/tests/parquet/mod.rs | 33 ++---- datafusion/datasource-parquet/src/source.rs | 28 +++-- .../proto/datafusion_common.proto | 11 +- datafusion/proto-common/src/from_proto/mod.rs | 25 ++++- .../proto-common/src/generated/pbjson.rs | 102 +++++++++++++++--- .../proto-common/src/generated/prost.rs | 46 +++++++- datafusion/proto-common/src/to_proto/mod.rs | 4 +- .../src/generated/datafusion_proto_common.rs | 46 +++++++- .../proto/src/logical_plan/file_formats.rs | 27 ++++- .../dynamic_filter_pushdown_config.slt | 8 +- .../test_files/explain_analyze.slt | 8 +- .../test_files/information_schema.slt | 2 + .../sqllogictest/test_files/limit_pruning.slt | 8 +- .../test_files/parquet_filter_pushdown.slt | 8 +- .../test_files/push_down_filter_parquet.slt | 8 +- .../push_down_filter_regression.slt | 8 +- .../sqllogictest/test_files/sort_pushdown.slt | 8 +- .../library-user-guide/upgrading/55.0.0.md | 36 +++---- docs/source/user-guide/configs.md | 2 +- 21 files changed, 378 insertions(+), 126 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 183c7233cf42b..28fc9706d84aa 100644 --- a/datafusion/common/src/config.rs +++ b/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 ), 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 { + 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(&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,14 +1203,10 @@ 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, decline to push filters - /// into the Parquet scan for queries whose projection contains fewer than - /// 3 columns not referenced by the filter. RowFilter has a fixed per-row - /// overhead that only pays for itself when there is enough wide-column - /// decode to skip; without this gate, narrow-projection queries (e.g. - /// `SELECT col1 FROM t WHERE col1 <> ''`) regress. Set to `false` to - /// force unconditional pushdown (the pre-existing behavior). - pub pushdown_filter_narrow_projection_gate: 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, diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 1a33d99d7be64..7c5b9000109ce 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -248,7 +248,7 @@ impl ParquetOptions { coerce_int96_tz: _, // not used for writer props skip_arrow_metadata: _, max_predicate_cache_size: _, - pushdown_filter_narrow_projection_gate: _, // reads-only, not used for writer props + pushdown_filter_mode: _, // reads-only, not used for writer props } = self; let mut builder = WriterProperties::builder() @@ -507,8 +507,7 @@ mod tests { coerce_int96: None, coerce_int96_tz: None, max_predicate_cache_size: defaults.max_predicate_cache_size, - pushdown_filter_narrow_projection_gate: defaults - .pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: defaults.pushdown_filter_mode, content_defined_chunking: defaults.content_defined_chunking.clone(), } } @@ -630,8 +629,7 @@ mod tests { skip_arrow_metadata: global_options_defaults.skip_arrow_metadata, coerce_int96: None, coerce_int96_tz: None, - pushdown_filter_narrow_projection_gate: global_options_defaults - .pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: global_options_defaults.pushdown_filter_mode, content_defined_chunking: props.content_defined_chunking().into(), }, column_specific_options, diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index 83289a41db5cd..64f85badae9d3 100644 --- a/datafusion/core/tests/parquet/mod.rs +++ b/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,18 +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; - // These tests specifically exercise the TopK dynamic RG - // pruning + RowFilter co-existence path. TopK's dynamic - // filter is injected AFTER filter pushdown runs, so the - // narrow-projection gate cannot detect it at plan time and - // may decline the WHERE-filter pushdown for narrow - // projections. Opt out of the gate so these tests exercise - // the unconditional-pushdown path they were designed for. - config - .options_mut() - .execution - .parquet - .pushdown_filter_narrow_projection_gate = false; + // 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, @@ -352,18 +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; - // These tests specifically exercise the TopK dynamic RG - // pruning + RowFilter co-existence path. TopK's dynamic - // filter is injected AFTER filter pushdown runs, so the - // narrow-projection gate cannot detect it at plan time and - // may decline the WHERE-filter pushdown for narrow - // projections. Opt out of the gate so these tests exercise - // the unconditional-pushdown path they were designed for. - config - .options_mut() - .execution - .parquet - .pushdown_filter_narrow_projection_gate = false; + // 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, diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 5824f02eb9b99..87dcd22824165 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/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; @@ -849,10 +850,10 @@ impl FileSource for ParquetSource { // 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 - // gated on a single boolean opt-out. A future - // adaptive-placement pass ([#22883]) can supersede this with a - // runtime cost model. + // 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 @@ -860,7 +861,7 @@ impl FileSource for ParquetSource { // page-index pruning. // // Users who want the pre-existing "always push" behavior can - // set `pushdown_filter_narrow_projection_gate = false`. + // set `pushdown_filter_mode = always`. // // [#22883]: https://github.com/apache/datafusion/issues/22883 const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; @@ -885,10 +886,8 @@ impl FileSource for ParquetSource { existing_predicate_has_dynamic || incoming_filters_have_dynamic; if pushdown_filters && !has_dynamic_filter - && config - .execution - .parquet - .pushdown_filter_narrow_projection_gate + && config.execution.parquet.pushdown_filter_mode + != ParquetPushdownFilterMode::Always { // `TableSchema` layout is `[file, partition, virtual]`; only // file columns are actually decoded from parquet, so partition @@ -1966,15 +1965,12 @@ mod tests { .expect("file_row_index should rewrite to the row_number virtual column"); let mut config = ConfigOptions::default(); - // Disable the narrow-projection gate so this test can exercise the + // 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 gate and mark every filter as `PushedDown::No` regardless - // of virtual-column content. - config - .execution - .parquet - .pushdown_filter_narrow_projection_gate = false; + // 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"); diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index de7c10177a194..4406d22a2c3aa 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -631,7 +631,16 @@ message ParquetOptions { uint64 max_row_group_bytes = 37; } - bool pushdown_filter_narrow_projection_gate = 38; + // 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; ParquetCdcOptions content_defined_chunking = 35; diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 419ad85125640..15b7af9f72205 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/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 for protobuf::CompressionTypeVariant { } } +impl From 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 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 for datafusion_common::parsers::CsvQuoteStyle { fn from(value: protobuf::CsvQuoteStyle) -> Self { match value { @@ -1133,7 +1154,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { max_row_group_bytes: value.max_row_group_bytes_opt.and_then(|opt| match opt { protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v) => MaxRowGroupBytes::try_new(v as usize).ok(), }), - pushdown_filter_narrow_projection_gate: value.pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: value.pushdown_filter_mode().into(), content_defined_chunking: value.content_defined_chunking.map(ParquetCdcOptions::from).unwrap_or_default(), }) } diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 05bbf26432700..5e37378946ac6 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6412,7 +6412,7 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { len += 1; } - if self.pushdown_filter_narrow_projection_gate { + if self.pushdown_filter_mode != 0 { len += 1; } if self.content_defined_chunking.is_some() { @@ -6535,8 +6535,10 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { struct_ser.serialize_field("createdBy", &self.created_by)?; } - if self.pushdown_filter_narrow_projection_gate { - struct_ser.serialize_field("pushdownFilterNarrowProjectionGate", &self.pushdown_filter_narrow_projection_gate)?; + if self.pushdown_filter_mode != 0 { + let v = parquet_options::PushdownFilterMode::try_from(self.pushdown_filter_mode) + .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.pushdown_filter_mode)))?; + struct_ser.serialize_field("pushdownFilterMode", &v)?; } if let Some(v) = self.content_defined_chunking.as_ref() { struct_ser.serialize_field("contentDefinedChunking", v)?; @@ -6695,8 +6697,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "maxRowGroupSize", "created_by", "createdBy", - "pushdown_filter_narrow_projection_gate", - "pushdownFilterNarrowProjectionGate", + "pushdown_filter_mode", + "pushdownFilterMode", "content_defined_chunking", "contentDefinedChunking", "metadata_size_hint", @@ -6748,7 +6750,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DataPageRowCountLimit, MaxRowGroupSize, CreatedBy, - PushdownFilterNarrowProjectionGate, + PushdownFilterMode, ContentDefinedChunking, MetadataSizeHint, Compression, @@ -6805,7 +6807,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dataPageRowCountLimit" | "data_page_row_count_limit" => Ok(GeneratedField::DataPageRowCountLimit), "maxRowGroupSize" | "max_row_group_size" => Ok(GeneratedField::MaxRowGroupSize), "createdBy" | "created_by" => Ok(GeneratedField::CreatedBy), - "pushdownFilterNarrowProjectionGate" | "pushdown_filter_narrow_projection_gate" => Ok(GeneratedField::PushdownFilterNarrowProjectionGate), + "pushdownFilterMode" | "pushdown_filter_mode" => Ok(GeneratedField::PushdownFilterMode), "contentDefinedChunking" | "content_defined_chunking" => Ok(GeneratedField::ContentDefinedChunking), "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint), "compression" => Ok(GeneratedField::Compression), @@ -6860,7 +6862,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut data_page_row_count_limit__ = None; let mut max_row_group_size__ = None; let mut created_by__ = None; - let mut pushdown_filter_narrow_projection_gate__ = None; + let mut pushdown_filter_mode__ = None; let mut content_defined_chunking__ = None; let mut metadata_size_hint_opt__ = None; let mut compression_opt__ = None; @@ -7017,11 +7019,11 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } created_by__ = Some(map_.next_value()?); } - GeneratedField::PushdownFilterNarrowProjectionGate => { - if pushdown_filter_narrow_projection_gate__.is_some() { - return Err(serde::de::Error::duplicate_field("pushdownFilterNarrowProjectionGate")); + GeneratedField::PushdownFilterMode => { + if pushdown_filter_mode__.is_some() { + return Err(serde::de::Error::duplicate_field("pushdownFilterMode")); } - pushdown_filter_narrow_projection_gate__ = Some(map_.next_value()?); + pushdown_filter_mode__ = Some(map_.next_value::()? as i32); } GeneratedField::ContentDefinedChunking => { if content_defined_chunking__.is_some() { @@ -7131,7 +7133,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { data_page_row_count_limit: data_page_row_count_limit__.unwrap_or_default(), max_row_group_size: max_row_group_size__.unwrap_or_default(), created_by: created_by__.unwrap_or_default(), - pushdown_filter_narrow_projection_gate: pushdown_filter_narrow_projection_gate__.unwrap_or_default(), + pushdown_filter_mode: pushdown_filter_mode__.unwrap_or_default(), content_defined_chunking: content_defined_chunking__, metadata_size_hint_opt: metadata_size_hint_opt__, compression_opt: compression_opt__, @@ -7152,6 +7154,80 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { deserializer.deserialize_struct("datafusion_common.ParquetOptions", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for parquet_options::PushdownFilterMode { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + let variant = match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + }; + serializer.serialize_str(variant) + } +} +impl<'de> serde::Deserialize<'de> for parquet_options::PushdownFilterMode { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "AUTO", + "ALWAYS", + "HEURISTIC", + ]; + + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = parquet_options::PushdownFilterMode; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + fn visit_i64(self, v: i64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self) + }) + } + + fn visit_u64(self, v: u64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self) + }) + } + + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "AUTO" => Ok(parquet_options::PushdownFilterMode::Auto), + "ALWAYS" => Ok(parquet_options::PushdownFilterMode::Always), + "HEURISTIC" => Ok(parquet_options::PushdownFilterMode::Heuristic), + _ => Err(serde::de::Error::unknown_variant(value, FIELDS)), + } + } + } + deserializer.deserialize_any(GeneratedVisitor) + } +} impl serde::Serialize for Precision { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index 73d46b624ed98..5f0c9c842fa8a 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -864,8 +864,8 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, - #[prost(bool, tag = "38")] - pub pushdown_filter_narrow_projection_gate: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] + pub pushdown_filter_mode: i32, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] @@ -915,6 +915,48 @@ pub struct ParquetOptions { } /// Nested message and enum types in `ParquetOptions`. pub mod parquet_options { + /// 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. + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum PushdownFilterMode { + Auto = 0, + Always = 1, + Heuristic = 2, + } + impl PushdownFilterMode { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "AUTO" => Some(Self::Auto), + "ALWAYS" => Some(Self::Always), + "HEURISTIC" => Some(Self::Heuristic), + _ => None, + } + } + } #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)] pub enum MetadataSizeHintOpt { #[prost(uint64, tag = "4")] diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 65972f62c88ce..31d872ec3475b 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -939,7 +939,9 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { coerce_int96_tz_opt: value.coerce_int96_tz.clone().map(protobuf::parquet_options::CoerceInt96TzOpt::CoerceInt96Tz), max_predicate_cache_size_opt: value.max_predicate_cache_size.map(|v| protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v as u64)), max_row_group_bytes_opt: value.max_row_group_bytes.map(|v| protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v.get() as u64)), - pushdown_filter_narrow_projection_gate: value.pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: protobuf::parquet_options::PushdownFilterMode::from( + value.pushdown_filter_mode, + ) as i32, content_defined_chunking: Some((&value.content_defined_chunking).into()), }) } diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index 73d46b624ed98..5f0c9c842fa8a 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -864,8 +864,8 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, - #[prost(bool, tag = "38")] - pub pushdown_filter_narrow_projection_gate: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] + pub pushdown_filter_mode: i32, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] @@ -915,6 +915,48 @@ pub struct ParquetOptions { } /// Nested message and enum types in `ParquetOptions`. pub mod parquet_options { + /// 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. + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum PushdownFilterMode { + Auto = 0, + Always = 1, + Heuristic = 2, + } + impl PushdownFilterMode { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "AUTO" => Some(Self::Auto), + "ALWAYS" => Some(Self::Always), + "HEURISTIC" => Some(Self::Heuristic), + _ => None, + } + } + } #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)] pub enum MetadataSizeHintOpt { #[prost(uint64, tag = "4")] diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index 8383361e4d1d7..fde4e35aa1ae4 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -386,7 +386,7 @@ mod parquet { }; use datafusion_common::config::{ MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, ParquetOptions, - TableParquetOptions, + ParquetPushdownFilterMode, TableParquetOptions, }; use datafusion_datasource_parquet::file_format::ParquetFormatFactory; @@ -461,7 +461,17 @@ mod parquet { max_row_group_bytes_opt: global_options.global.max_row_group_bytes.map(|size| { parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size.get() as u64) }), - pushdown_filter_narrow_projection_gate: global_options.global.pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: match global_options.global.pushdown_filter_mode { + ParquetPushdownFilterMode::Auto => { + parquet_options::PushdownFilterMode::Auto + } + ParquetPushdownFilterMode::Always => { + parquet_options::PushdownFilterMode::Always + } + ParquetPushdownFilterMode::Heuristic => { + parquet_options::PushdownFilterMode::Heuristic + } + } as i32, content_defined_chunking: Some(ParquetCdcOptionsProto { enabled: global_options.global.content_defined_chunking.enabled, min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, @@ -643,8 +653,17 @@ mod parquet { MaxRowGroupBytes::try_new(*size as usize).ok() } }), - pushdown_filter_narrow_projection_gate: proto - .pushdown_filter_narrow_projection_gate, + pushdown_filter_mode: match proto.pushdown_filter_mode() { + parquet_options::PushdownFilterMode::Auto => { + ParquetPushdownFilterMode::Auto + } + parquet_options::PushdownFilterMode::Always => { + ParquetPushdownFilterMode::Always + } + parquet_options::PushdownFilterMode::Heuristic => { + ParquetPushdownFilterMode::Heuristic + } + }, content_defined_chunking: proto .content_defined_chunking .map(ParquetCdcOptions::from_proto) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 9137d9adbf742..d6fee253cba21 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -573,11 +573,11 @@ LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet'; statement ok SET datafusion.execution.parquet.pushdown_filters = true; -# This section exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# This section exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +SET datafusion.execution.parquet.pushdown_filter_mode = always; # Aggregate dynamic filter should be pushed into the scan when enabled # Expecting a `DynamicFilter` inside parquet scanner's predicate @@ -878,4 +878,4 @@ statement ok RESET datafusion.execution.parquet.max_row_group_size; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index 103ca13e85a21..bc95e3f9b162c 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -189,11 +189,11 @@ reset datafusion.explain.analyze_level; statement ok set datafusion.execution.parquet.pushdown_filters = true; -# Section exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Section exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +set datafusion.execution.parquet.pushdown_filter_mode = always; statement ok CREATE TABLE _cat_data AS VALUES @@ -721,4 +721,4 @@ statement ok reset datafusion.execution.parquet.pushdown_filters; statement ok -reset datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +reset datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 1adf98f67ff99..5ebce0ab83dde 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -260,6 +260,7 @@ datafusion.execution.parquet.maximum_buffered_record_batches_per_stream 2 datafusion.execution.parquet.maximum_parallel_row_group_writers 1 datafusion.execution.parquet.metadata_size_hint 524288 datafusion.execution.parquet.pruning true +datafusion.execution.parquet.pushdown_filter_mode auto datafusion.execution.parquet.pushdown_filters false datafusion.execution.parquet.reorder_filters false datafusion.execution.parquet.schema_force_view_types true @@ -419,6 +420,7 @@ datafusion.execution.parquet.maximum_buffered_record_batches_per_stream 2 (writi datafusion.execution.parquet.maximum_parallel_row_group_writers 1 (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. datafusion.execution.parquet.metadata_size_hint 524288 (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. datafusion.execution.parquet.pruning true (reading) If true, the parquet reader attempts to skip entire row groups based on the predicate in the query and the metadata (min/max values) stored in the parquet file +datafusion.execution.parquet.pushdown_filter_mode auto (reading) When `pushdown_filters` is enabled, selects how DataFusion decides whether to actually push each filter into the Parquet scan as a `RowFilter`. See [`ParquetPushdownFilterMode`] for the available strategies. Defaults to `auto`, which today applies a plan-time heuristic that declines pushdown for narrow-projection queries where the `RowFilter` overhead tends to outweigh the decode it saves. Set to `always` to unconditionally push filters down (the behavior before the heuristic was introduced). datafusion.execution.parquet.pushdown_filters false (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". datafusion.execution.parquet.reorder_filters false (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.execution.parquet.schema_force_view_types true (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index 2404ebec03e9a..09fe5cd879430 100644 --- a/datafusion/sqllogictest/test_files/limit_pruning.slt +++ b/datafusion/sqllogictest/test_files/limit_pruning.slt @@ -18,11 +18,11 @@ statement ok set datafusion.execution.parquet.pushdown_filters = true; -# Test file exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced. statement ok -set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +set datafusion.execution.parquet.pushdown_filter_mode = always; statement ok @@ -139,4 +139,4 @@ statement ok RESET datafusion.execution.parquet.pushdown_filters; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index 2e2442812a95f..e729b7b7a020c 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -21,11 +21,11 @@ # scan not just the metadata) ########## -# Test file exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +set datafusion.execution.parquet.pushdown_filter_mode = always; # File1 has only columns a and b statement ok @@ -945,7 +945,7 @@ statement ok set datafusion.execution.parquet.reorder_filters = false; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; statement ok DROP TABLE dict_filter_bug; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 59e423dd083b7..5bb22083f4c29 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -24,11 +24,11 @@ set datafusion.explain.physical_plan_only = true; statement ok set datafusion.execution.parquet.pushdown_filters = true; -# Test file exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +set datafusion.execution.parquet.pushdown_filter_mode = always; # this one is also required to make DF skip second file due to "sufficient" amount of rows statement ok @@ -1080,7 +1080,7 @@ statement ok RESET datafusion.execution.parquet.pushdown_filters; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; statement ok drop table t; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 9c0c33029a2f6..186fa926c354f 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -17,11 +17,11 @@ # Test push down filter -# Test file exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -set datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +set datafusion.execution.parquet.pushdown_filter_mode = always; # Regression test for https://github.com/apache/datafusion/issues/17188 query I @@ -595,4 +595,4 @@ statement ok drop table t2; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt index 9fd1c16f55b93..93b5f75f97fd9 100644 --- a/datafusion/sqllogictest/test_files/sort_pushdown.slt +++ b/datafusion/sqllogictest/test_files/sort_pushdown.slt @@ -5,11 +5,11 @@ SET datafusion.execution.parquet.pushdown_filters = true; statement ok SET datafusion.optimizer.enable_sort_pushdown = true; -# Test file exercises the pushdown path directly. Disable the -# narrow-projection gate so the expected plans (FilterExec absorbed +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed # into DataSourceExec) are reproduced regardless of projection width. statement ok -SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +SET datafusion.execution.parquet.pushdown_filter_mode = always; # Test 1: Sort Pushdown for ordered Parquet files # Create a sorted dataset @@ -2967,4 +2967,4 @@ statement ok SET datafusion.execution.target_partitions = 4; statement ok -RESET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate; +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index ba28c04850d99..f1aadd324061b 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -275,31 +275,29 @@ if tracking.contains_dynamic_filter() { } ``` -### Narrow-projection gate on Parquet filter pushdown - -When `datafusion.execution.parquet.pushdown_filters = true`, DataFusion now -declines to push filters into the Parquet scan for queries whose projection -contains fewer than three columns not referenced by the filter. RowFilter has -a fixed per-row machinery overhead that only pays for itself when there is -meaningful wide-column decode to skip; for narrow-projection queries (e.g. -`SELECT col1 FROM t WHERE col1 <> ''`) the overhead dominates and pushdown -regresses. When the gate declines pushdown, 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. The plan display -marks affected scans with `pushdown_declined=narrow_projection` so profiling -can see when the heuristic fired. - -To restore the pre-existing "always push" behavior: +### New `pushdown_filter_mode` option for Parquet filter pushdown + +DataFusion 55 adds a new option, +`datafusion.execution.parquet.pushdown_filter_mode`, that controls how filters +are pushed into Parquet scans when `pushdown_filters` is enabled. It accepts: + +Because the default is `auto`, enabling `pushdown_filters` no longer guarantees +that **every** filter is pushed into the scan as was the case in DataFusion 54 +and earlier. To get the old behavior, also set `pushdown_filter_mode = always`: + +| DataFusion 54 settings | Equivalent DataFusion 55 settings | +| ---------------------------------------------------- | ------------------------------------------------------------------------------------- | +| `pushdown_filters = true`, `reorder_filters = true` | `pushdown_filters = true`, `reorder_filters = true`, `pushdown_filter_mode = always` | +| `pushdown_filters = true`, `reorder_filters = false` | `pushdown_filters = true`, `reorder_filters = false`, `pushdown_filter_mode = always` | + +For example: ```sql -SET datafusion.execution.parquet.pushdown_filter_narrow_projection_gate = false; +SET datafusion.execution.parquet.pushdown_filter_mode = always; ``` See [issue #3463](https://github.com/apache/datafusion/issues/3463) and [PR #23420](https://github.com/apache/datafusion/pull/23420) for context. -A runtime adaptive-placement effort ([EPIC #22883](https://github.com/apache/datafusion/issues/22883)) -is tracked separately and may eventually complement this plan-time -heuristic. ### `FilePruner::try_new` no longer builds a pruner for static predicates without statistics diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 666d7f9d363d3..ce54c295a52b4 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -85,7 +85,7 @@ The following configuration settings are available: | datafusion.execution.parquet.skip_metadata | true | (reading) If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata | | datafusion.execution.parquet.metadata_size_hint | 524288 | (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. | | datafusion.execution.parquet.pushdown_filters | false | (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". | -| datafusion.execution.parquet.pushdown_filter_narrow_projection_gate | true | (reading) When `pushdown_filters` is enabled, decline to push filters into the Parquet scan for queries whose projection contains fewer than 3 columns not referenced by the filter. RowFilter has a fixed per-row overhead that only pays for itself when there is enough wide-column decode to skip; without this gate, narrow-projection queries (e.g. `SELECT col1 FROM t WHERE col1 <> ''`) regress. Set to `false` to force unconditional pushdown (the pre-existing behavior). | +| datafusion.execution.parquet.pushdown_filter_mode | auto | (reading) When `pushdown_filters` is enabled, selects how DataFusion decides whether to actually push each filter into the Parquet scan as a `RowFilter`. See [`ParquetPushdownFilterMode`] for the available strategies. Defaults to `auto`, which today applies a plan-time heuristic that declines pushdown for narrow-projection queries where the `RowFilter` overhead tends to outweigh the decode it saves. Set to `always` to unconditionally push filters down (the behavior before the heuristic was introduced). | | datafusion.execution.parquet.reorder_filters | false | (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.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. | From 4721666faeb1af19d732e09e6c3d29db130524f2 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 10 Jul 2026 17:03:35 -0400 Subject: [PATCH 13/15] reorder for consistency --- .../common/src/file_options/parquet_writer.rs | 6 +-- .../proto/datafusion_common.proto | 20 ++++----- datafusion/proto-common/src/from_proto/mod.rs | 2 +- datafusion/proto-common/src/to_proto/mod.rs | 6 +-- .../proto/src/logical_plan/file_formats.rs | 44 +++++++++---------- .../test_files/information_schema.slt | 2 +- 6 files changed, 39 insertions(+), 41 deletions(-) diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 7c5b9000109ce..933c7404c3413 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/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: _, @@ -248,7 +249,6 @@ impl ParquetOptions { coerce_int96_tz: _, // not used for writer props skip_arrow_metadata: _, max_predicate_cache_size: _, - pushdown_filter_mode: _, // reads-only, not used for writer props } = self; let mut builder = WriterProperties::builder() @@ -493,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, @@ -507,7 +508,6 @@ mod tests { coerce_int96: None, coerce_int96_tz: None, max_predicate_cache_size: defaults.max_predicate_cache_size, - pushdown_filter_mode: defaults.pushdown_filter_mode, content_defined_chunking: defaults.content_defined_chunking.clone(), } } @@ -613,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 @@ -629,7 +630,6 @@ mod tests { skip_arrow_metadata: global_options_defaults.skip_arrow_metadata, coerce_int96: None, coerce_int96_tz: None, - pushdown_filter_mode: global_options_defaults.pushdown_filter_mode, content_defined_chunking: props.content_defined_chunking().into(), }, column_specific_options, diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 4406d22a2c3aa..c14635fe2f35b 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/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 @@ -631,17 +640,6 @@ message ParquetOptions { uint64 max_row_group_bytes = 37; } - // 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; - ParquetCdcOptions content_defined_chunking = 35; // Optional timezone applied to INT96-coerced timestamps when `coerce_int96` diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 15b7af9f72205..6b3ffc442b6df 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1082,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, @@ -1154,7 +1155,6 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { max_row_group_bytes: value.max_row_group_bytes_opt.and_then(|opt| match opt { protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v) => MaxRowGroupBytes::try_new(v as usize).ok(), }), - pushdown_filter_mode: value.pushdown_filter_mode().into(), content_defined_chunking: value.content_defined_chunking.map(ParquetCdcOptions::from).unwrap_or_default(), }) } diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 31d872ec3475b..0191a5cea1686 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -910,6 +910,9 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { skip_metadata: value.skip_metadata, metadata_size_hint_opt: value.metadata_size_hint.map(|v| protobuf::parquet_options::MetadataSizeHintOpt::MetadataSizeHint(v as u64)), pushdown_filters: value.pushdown_filters, + pushdown_filter_mode: protobuf::parquet_options::PushdownFilterMode::from( + value.pushdown_filter_mode, + ) as i32, reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, data_pagesize_limit: value.data_pagesize_limit as u64, @@ -939,9 +942,6 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { coerce_int96_tz_opt: value.coerce_int96_tz.clone().map(protobuf::parquet_options::CoerceInt96TzOpt::CoerceInt96Tz), max_predicate_cache_size_opt: value.max_predicate_cache_size.map(|v| protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v as u64)), max_row_group_bytes_opt: value.max_row_group_bytes.map(|v| protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v.get() as u64)), - pushdown_filter_mode: protobuf::parquet_options::PushdownFilterMode::from( - value.pushdown_filter_mode, - ) as i32, content_defined_chunking: Some((&value.content_defined_chunking).into()), }) } diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index fde4e35aa1ae4..d10ed738df288 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -408,6 +408,17 @@ mod parquet { parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64) }), pushdown_filters: global_options.global.pushdown_filters, + pushdown_filter_mode: match global_options.global.pushdown_filter_mode { + ParquetPushdownFilterMode::Auto => { + parquet_options::PushdownFilterMode::Auto + } + ParquetPushdownFilterMode::Always => { + parquet_options::PushdownFilterMode::Always + } + ParquetPushdownFilterMode::Heuristic => { + parquet_options::PushdownFilterMode::Heuristic + } + } as i32, reorder_filters: global_options.global.reorder_filters, force_filter_selections: global_options.global.force_filter_selections, data_pagesize_limit: global_options.global.data_pagesize_limit as u64, @@ -461,17 +472,6 @@ mod parquet { max_row_group_bytes_opt: global_options.global.max_row_group_bytes.map(|size| { parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size.get() as u64) }), - pushdown_filter_mode: match global_options.global.pushdown_filter_mode { - ParquetPushdownFilterMode::Auto => { - parquet_options::PushdownFilterMode::Auto - } - ParquetPushdownFilterMode::Always => { - parquet_options::PushdownFilterMode::Always - } - ParquetPushdownFilterMode::Heuristic => { - parquet_options::PushdownFilterMode::Heuristic - } - } as i32, content_defined_chunking: Some(ParquetCdcOptionsProto { enabled: global_options.global.content_defined_chunking.enabled, min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, @@ -555,6 +555,17 @@ mod parquet { } }), pushdown_filters: proto.pushdown_filters, + pushdown_filter_mode: match proto.pushdown_filter_mode() { + parquet_options::PushdownFilterMode::Auto => { + ParquetPushdownFilterMode::Auto + } + parquet_options::PushdownFilterMode::Always => { + ParquetPushdownFilterMode::Always + } + parquet_options::PushdownFilterMode::Heuristic => { + ParquetPushdownFilterMode::Heuristic + } + }, reorder_filters: proto.reorder_filters, force_filter_selections: proto.force_filter_selections, data_pagesize_limit: proto.data_pagesize_limit as usize, @@ -653,17 +664,6 @@ mod parquet { MaxRowGroupBytes::try_new(*size as usize).ok() } }), - pushdown_filter_mode: match proto.pushdown_filter_mode() { - parquet_options::PushdownFilterMode::Auto => { - ParquetPushdownFilterMode::Auto - } - parquet_options::PushdownFilterMode::Always => { - ParquetPushdownFilterMode::Always - } - parquet_options::PushdownFilterMode::Heuristic => { - ParquetPushdownFilterMode::Heuristic - } - }, content_defined_chunking: proto .content_defined_chunking .map(ParquetCdcOptions::from_proto) diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 5ebce0ab83dde..14b270214ada8 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -420,7 +420,7 @@ datafusion.execution.parquet.maximum_buffered_record_batches_per_stream 2 (writi datafusion.execution.parquet.maximum_parallel_row_group_writers 1 (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. datafusion.execution.parquet.metadata_size_hint 524288 (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. datafusion.execution.parquet.pruning true (reading) If true, the parquet reader attempts to skip entire row groups based on the predicate in the query and the metadata (min/max values) stored in the parquet file -datafusion.execution.parquet.pushdown_filter_mode auto (reading) When `pushdown_filters` is enabled, selects how DataFusion decides whether to actually push each filter into the Parquet scan as a `RowFilter`. See [`ParquetPushdownFilterMode`] for the available strategies. Defaults to `auto`, which today applies a plan-time heuristic that declines pushdown for narrow-projection queries where the `RowFilter` overhead tends to outweigh the decode it saves. Set to `always` to unconditionally push filters down (the behavior before the heuristic was introduced). +datafusion.execution.parquet.pushdown_filter_mode auto (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). datafusion.execution.parquet.pushdown_filters false (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". datafusion.execution.parquet.reorder_filters false (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.execution.parquet.schema_force_view_types true (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. From 5dee386ed792a9c15ede8051ffa1965579dbd1e4 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 10 Jul 2026 17:14:59 -0400 Subject: [PATCH 14/15] reorder --- .../proto-common/src/generated/pbjson.rs | 40 +++++++++---------- .../proto-common/src/generated/prost.rs | 4 +- .../src/generated/datafusion_proto_common.rs | 4 +- docs/source/user-guide/configs.md | 2 +- 4 files changed, 25 insertions(+), 25 deletions(-) diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 5e37378946ac6..1bc05790dc768 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6361,6 +6361,9 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { len += 1; } + if self.pushdown_filter_mode != 0 { + len += 1; + } if self.reorder_filters { len += 1; } @@ -6412,9 +6415,6 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { len += 1; } - if self.pushdown_filter_mode != 0 { - len += 1; - } if self.content_defined_chunking.is_some() { len += 1; } @@ -6470,6 +6470,11 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { struct_ser.serialize_field("pushdownFilters", &self.pushdown_filters)?; } + if self.pushdown_filter_mode != 0 { + let v = parquet_options::PushdownFilterMode::try_from(self.pushdown_filter_mode) + .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.pushdown_filter_mode)))?; + struct_ser.serialize_field("pushdownFilterMode", &v)?; + } if self.reorder_filters { struct_ser.serialize_field("reorderFilters", &self.reorder_filters)?; } @@ -6535,11 +6540,6 @@ impl serde::Serialize for ParquetOptions { if !self.created_by.is_empty() { struct_ser.serialize_field("createdBy", &self.created_by)?; } - if self.pushdown_filter_mode != 0 { - let v = parquet_options::PushdownFilterMode::try_from(self.pushdown_filter_mode) - .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.pushdown_filter_mode)))?; - struct_ser.serialize_field("pushdownFilterMode", &v)?; - } if let Some(v) = self.content_defined_chunking.as_ref() { struct_ser.serialize_field("contentDefinedChunking", v)?; } @@ -6663,6 +6663,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "skipMetadata", "pushdown_filters", "pushdownFilters", + "pushdown_filter_mode", + "pushdownFilterMode", "reorder_filters", "reorderFilters", "force_filter_selections", @@ -6697,8 +6699,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "maxRowGroupSize", "created_by", "createdBy", - "pushdown_filter_mode", - "pushdownFilterMode", "content_defined_chunking", "contentDefinedChunking", "metadata_size_hint", @@ -6733,6 +6733,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Pruning, SkipMetadata, PushdownFilters, + PushdownFilterMode, ReorderFilters, ForceFilterSelections, DataPagesizeLimit, @@ -6750,7 +6751,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DataPageRowCountLimit, MaxRowGroupSize, CreatedBy, - PushdownFilterMode, ContentDefinedChunking, MetadataSizeHint, Compression, @@ -6790,6 +6790,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "pruning" => Ok(GeneratedField::Pruning), "skipMetadata" | "skip_metadata" => Ok(GeneratedField::SkipMetadata), "pushdownFilters" | "pushdown_filters" => Ok(GeneratedField::PushdownFilters), + "pushdownFilterMode" | "pushdown_filter_mode" => Ok(GeneratedField::PushdownFilterMode), "reorderFilters" | "reorder_filters" => Ok(GeneratedField::ReorderFilters), "forceFilterSelections" | "force_filter_selections" => Ok(GeneratedField::ForceFilterSelections), "dataPagesizeLimit" | "data_pagesize_limit" => Ok(GeneratedField::DataPagesizeLimit), @@ -6807,7 +6808,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dataPageRowCountLimit" | "data_page_row_count_limit" => Ok(GeneratedField::DataPageRowCountLimit), "maxRowGroupSize" | "max_row_group_size" => Ok(GeneratedField::MaxRowGroupSize), "createdBy" | "created_by" => Ok(GeneratedField::CreatedBy), - "pushdownFilterMode" | "pushdown_filter_mode" => Ok(GeneratedField::PushdownFilterMode), "contentDefinedChunking" | "content_defined_chunking" => Ok(GeneratedField::ContentDefinedChunking), "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint), "compression" => Ok(GeneratedField::Compression), @@ -6845,6 +6845,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut pruning__ = None; let mut skip_metadata__ = None; let mut pushdown_filters__ = None; + let mut pushdown_filter_mode__ = None; let mut reorder_filters__ = None; let mut force_filter_selections__ = None; let mut data_pagesize_limit__ = None; @@ -6862,7 +6863,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut data_page_row_count_limit__ = None; let mut max_row_group_size__ = None; let mut created_by__ = None; - let mut pushdown_filter_mode__ = None; let mut content_defined_chunking__ = None; let mut metadata_size_hint_opt__ = None; let mut compression_opt__ = None; @@ -6903,6 +6903,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } pushdown_filters__ = Some(map_.next_value()?); } + GeneratedField::PushdownFilterMode => { + if pushdown_filter_mode__.is_some() { + return Err(serde::de::Error::duplicate_field("pushdownFilterMode")); + } + pushdown_filter_mode__ = Some(map_.next_value::()? as i32); + } GeneratedField::ReorderFilters => { if reorder_filters__.is_some() { return Err(serde::de::Error::duplicate_field("reorderFilters")); @@ -7019,12 +7025,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } created_by__ = Some(map_.next_value()?); } - GeneratedField::PushdownFilterMode => { - if pushdown_filter_mode__.is_some() { - return Err(serde::de::Error::duplicate_field("pushdownFilterMode")); - } - pushdown_filter_mode__ = Some(map_.next_value::()? as i32); - } GeneratedField::ContentDefinedChunking => { if content_defined_chunking__.is_some() { return Err(serde::de::Error::duplicate_field("contentDefinedChunking")); @@ -7116,6 +7116,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { pruning: pruning__.unwrap_or_default(), skip_metadata: skip_metadata__.unwrap_or_default(), pushdown_filters: pushdown_filters__.unwrap_or_default(), + pushdown_filter_mode: pushdown_filter_mode__.unwrap_or_default(), reorder_filters: reorder_filters__.unwrap_or_default(), force_filter_selections: force_filter_selections__.unwrap_or_default(), data_pagesize_limit: data_pagesize_limit__.unwrap_or_default(), @@ -7133,7 +7134,6 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { data_page_row_count_limit: data_page_row_count_limit__.unwrap_or_default(), max_row_group_size: max_row_group_size__.unwrap_or_default(), created_by: created_by__.unwrap_or_default(), - pushdown_filter_mode: pushdown_filter_mode__.unwrap_or_default(), content_defined_chunking: content_defined_chunking__, metadata_size_hint_opt: metadata_size_hint_opt__, compression_opt: compression_opt__, diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index 5f0c9c842fa8a..04250ffb46903 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -815,6 +815,8 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] + pub pushdown_filter_mode: i32, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, @@ -864,8 +866,6 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, - #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] - pub pushdown_filter_mode: i32, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index 5f0c9c842fa8a..04250ffb46903 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -815,6 +815,8 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] + pub pushdown_filter_mode: i32, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, @@ -864,8 +866,6 @@ pub struct ParquetOptions { pub max_row_group_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, - #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "38")] - pub pushdown_filter_mode: i32, #[prost(message, optional, tag = "35")] pub content_defined_chunking: ::core::option::Option, #[prost(oneof = "parquet_options::MetadataSizeHintOpt", tags = "4")] diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index ce54c295a52b4..c52e6f55d2b8f 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -85,7 +85,7 @@ The following configuration settings are available: | datafusion.execution.parquet.skip_metadata | true | (reading) If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata | | datafusion.execution.parquet.metadata_size_hint | 524288 | (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. | | datafusion.execution.parquet.pushdown_filters | false | (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". | -| datafusion.execution.parquet.pushdown_filter_mode | auto | (reading) When `pushdown_filters` is enabled, selects how DataFusion decides whether to actually push each filter into the Parquet scan as a `RowFilter`. See [`ParquetPushdownFilterMode`] for the available strategies. Defaults to `auto`, which today applies a plan-time heuristic that declines pushdown for narrow-projection queries where the `RowFilter` overhead tends to outweigh the decode it saves. Set to `always` to unconditionally push filters down (the behavior before the heuristic was introduced). | +| datafusion.execution.parquet.pushdown_filter_mode | auto | (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). | | datafusion.execution.parquet.reorder_filters | false | (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.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. | From f4f8666546c027dc4299ec78e0d1cf43553b1e82 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 10 Jul 2026 17:38:49 -0400 Subject: [PATCH 15/15] clean --- datafusion/datasource-parquet/src/source.rs | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 87dcd22824165..b9f898ef05272 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -315,13 +315,6 @@ pub struct ParquetSource { /// Sort order driving `PreparedAccessPlan::reorder_by_statistics` /// in the opener. sort_order_for_reorder: Option, - /// True when [`try_pushdown_filters`] declined pushdown for this - /// scan due to the narrow-projection gate — surfaced in the plan - /// display so profiling can tell that the heuristic fired here - /// (the filter stays above the scan in a `FilterExec`). - /// - /// [`try_pushdown_filters`]: Self::try_pushdown_filters - pub(crate) narrow_projection_gate_declined: bool, } impl ParquetSource { @@ -348,7 +341,6 @@ impl ParquetSource { encryption_factory: None, reverse_row_groups: false, sort_order_for_reorder: None, - narrow_projection_gate_declined: false, } } @@ -749,9 +741,6 @@ impl FileSource for ParquetSource { if self.reverse_row_groups { write!(f, ", reverse_row_groups=true")?; } - if self.narrow_projection_gate_declined { - write!(f, ", pushdown_declined=narrow_projection")?; - } // Plan-time marker for dynamic RG-level pruning: if the // predicate is dynamic (e.g. a TopK threshold expression), @@ -915,7 +904,6 @@ impl FileSource for ParquetSource { } let mut source = self.clone(); - source.narrow_projection_gate_declined = narrow_projection_gate_declined; let filters: Vec = filters .into_iter() .map(|filter| {