diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index b649ecad570d2..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,6 +1203,11 @@ config_namespace! { /// reduce the number of rows decoded. This optimization is sometimes called "late materialization". pub pushdown_filters: bool, default = false + /// (reading) When `pushdown_filters` is enabled, determines how DataFusion + /// pushes each filter into the Parquet scan. Options are `auto` (the default) + /// `always`, and `heurstic` (plan time heuristic). + pub pushdown_filter_mode: ParquetPushdownFilterMode, default = ParquetPushdownFilterMode::Auto + /// (reading) If true, filter expressions evaluated during the parquet decoding operation /// will be reordered heuristically to minimize the cost of evaluation. If false, /// the filters are applied in the same order as written in the query diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 320bfcf33e488..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: _, @@ -492,6 +493,7 @@ mod tests { skip_metadata: defaults.skip_metadata, metadata_size_hint: defaults.metadata_size_hint, pushdown_filters: defaults.pushdown_filters, + pushdown_filter_mode: defaults.pushdown_filter_mode, reorder_filters: defaults.reorder_filters, force_filter_selections: defaults.force_filter_selections, allow_single_file_parallelism: defaults.allow_single_file_parallelism, @@ -611,6 +613,7 @@ mod tests { skip_metadata: global_options_defaults.skip_metadata, metadata_size_hint: global_options_defaults.metadata_size_hint, pushdown_filters: global_options_defaults.pushdown_filters, + pushdown_filter_mode: global_options_defaults.pushdown_filter_mode, reorder_filters: global_options_defaults.reorder_filters, force_filter_selections: global_options_defaults.force_filter_selections, allow_single_file_parallelism: global_options_defaults diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index 1cc4bb32d9eba..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,6 +324,10 @@ impl ContextWithParquet { Unit::RowGroup(row_per_group) => { config = config.with_parquet_bloom_filter_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; + // force unconditional pushdown to test so the filters are + // applied for TopK dynamic RG pruning + config.options_mut().execution.parquet.pushdown_filter_mode = + ParquetPushdownFilterMode::Always; make_test_file_rg( scenario, row_per_group, @@ -340,6 +345,10 @@ impl ContextWithParquet { config = config.with_parquet_bloom_filter_pruning(true); config = config.with_parquet_page_index_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; + // force unconditional pushdown to test so the filters are + // applied for TopK dynamic RG pruning + config.options_mut().execution.parquet.pushdown_filter_mode = + ParquetPushdownFilterMode::Always; make_test_file_rg( scenario, row_per_group, diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 3443b08475e0d..b9f898ef05272 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; @@ -46,6 +47,7 @@ use datafusion_datasource::file_scan_config::FileScanConfig; use datafusion_functions::core::file_row_index::FileRowIndexFunc; use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking}; use datafusion_physical_expr::projection::ProjectionExprs; +use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr::{EquivalenceProperties, conjunction}; use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory; use datafusion_physical_expr_adapter::rewrite::{ @@ -827,7 +829,79 @@ impl FileSource for ParquetSource { // because even if scan pushdown is disabled we can still use the filters for stats pruning. let config_pushdown_enabled = config.execution.parquet.pushdown_filters; let table_pushdown_enabled = self.pushdown_filters(); - let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled; + let mut pushdown_filters = table_pushdown_enabled || config_pushdown_enabled; + // Narrow-projection gate (issue #3463, discussion on PR #23369): + // RowFilter has a fixed per-row machinery overhead that only pays + // for itself when the wide-column decode it lets us skip is + // meaningful. When the projection is narrow and the filter columns + // already cover most of it (typical for `GROUP BY col`-style + // queries such as ClickBench Q10/Q11/Q40), the overhead dominates + // and pushdown regresses. Decline pushdown for those scans; keep + // it for wider projections where the fast-path pays. + // + // Kept intentionally simple: a plan-time column-count check, + // used when `pushdown_filter_mode` is `auto` or `heuristic`. A + // future adaptive-placement pass ([#22883]) can supersede this + // with a runtime cost model behind the same `auto` mode. + // + // When declined: the filter stays above the scan in a + // `FilterExec` (correctness preserved) and the predicate is + // still injected into `ParquetSource` for stats / bloom / + // page-index pruning. + // + // Users who want the pre-existing "always push" behavior can + // set `pushdown_filter_mode = always`. + // + // [#22883]: https://github.com/apache/datafusion/issues/22883 + const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; + let mut narrow_projection_gate_declined = false; + // Never gate a scan whose predicate already contains a dynamic + // filter — either in the incoming `filters` parameter, or already + // installed on `self.predicate` by an earlier optimizer rule (this + // is how `TopK`'s heap-threshold expression arrives: the sort + // pushdown / TopK rule injects the `DynamicFilterPhysicalExpr` + // into the scan's predicate before filter pushdown runs). Dynamic + // filters rely on the scan-time RowFilter/RowGroupPruner cascade + // to prune data as the threshold tightens, so declining pushdown + // here would silently disable the entire dynamic-RG-prune path + // for narrow `ORDER BY ... LIMIT` queries. + let existing_predicate_has_dynamic = self.predicate.as_ref().is_some_and(|p| { + DynamicFilterTracking::classify(p).contains_dynamic_filter() + }); + let incoming_filters_have_dynamic = filters + .iter() + .any(|f| DynamicFilterTracking::classify(f).contains_dynamic_filter()); + let has_dynamic_filter = + existing_predicate_has_dynamic || incoming_filters_have_dynamic; + if pushdown_filters + && !has_dynamic_filter + && config.execution.parquet.pushdown_filter_mode + != ParquetPushdownFilterMode::Always + { + // `TableSchema` layout is `[file, partition, virtual]`; only + // file columns are actually decoded from parquet, so partition + // and virtual columns don't count toward the "wide-decode + // saving" the RowFilter fast-path buys us. + let file_col_count = self.table_schema.file_schema().fields().len(); + let filter_col_indices: std::collections::HashSet = filters + .iter() + .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index())) + .filter(|idx| *idx < file_col_count) + .collect(); + let non_filter_projected = self + .projection + .as_ref() + .iter() + .flat_map(|pe| collect_columns(&pe.expr)) + .map(|c| c.index()) + .filter(|idx| *idx < file_col_count && !filter_col_indices.contains(idx)) + .collect::>() + .len(); + if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS { + pushdown_filters = false; + narrow_projection_gate_declined = true; + } + } let mut source = self.clone(); let filters: Vec = filters @@ -864,7 +938,18 @@ impl FileSource for ParquetSource { None => conjunction(allowed_filters), }; source.predicate = Some(predicate); - source = source.with_pushdown_filters(pushdown_filters); + // Only persist the pushdown_filters bit on the source when this is + // a "real" config-driven decision. When the gate declined, we + // deliberately do NOT flip the source's pushdown flag to false — + // a later `try_pushdown_filters` call (e.g. from TopK's dynamic + // filter injection) needs the source to still be pushdown-eligible + // so that dynamic filter can install its RowFilter and drive the + // runtime RG-prune cascade. The current call still returns + // `PushedDown::No` for its incoming filters, so the FilterExec + // above the scan stays and correctness is preserved. + if !narrow_projection_gate_declined { + source = source.with_pushdown_filters(pushdown_filters); + } let source = Arc::new(source); // If pushdown_filters is false we tell our parents that they still have to handle the filters, // even if we updated the predicate to include the filters (they will only be used for stats pruning). @@ -1867,7 +1952,13 @@ mod tests { ) .expect("file_row_index should rewrite to the row_number virtual column"); - let config = ConfigOptions::default(); + let mut config = ConfigOptions::default(); + // Force unconditional pushdown so this test can exercise the + // virtual-column rejection path independently. The scan projection + // here has 0 non-filter file columns, which would otherwise trip + // the narrow-projection heuristic and mark every filter as + // `PushedDown::No` regardless of virtual-column content. + config.execution.parquet.pushdown_filter_mode = ParquetPushdownFilterMode::Always; let prop = source .try_pushdown_filters(vec![pushable, virtual_only, mixed, row_index], &config) .expect("try_pushdown_filters must not error"); diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 7fff5b6b715ff..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 diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 97cc9af230105..6b3ffc442b6df 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 { @@ -1061,6 +1082,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { }) .unwrap_or(None), pushdown_filters: value.pushdown_filters, + pushdown_filter_mode: value.pushdown_filter_mode().into(), reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, data_pagesize_limit: value.data_pagesize_limit as usize, diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 963faa5a3e9cb..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; } @@ -6467,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)?; } @@ -6655,6 +6663,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "skipMetadata", "pushdown_filters", "pushdownFilters", + "pushdown_filter_mode", + "pushdownFilterMode", "reorder_filters", "reorderFilters", "force_filter_selections", @@ -6723,6 +6733,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Pruning, SkipMetadata, PushdownFilters, + PushdownFilterMode, ReorderFilters, ForceFilterSelections, DataPagesizeLimit, @@ -6779,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), @@ -6833,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; @@ -6890,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")); @@ -7097,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(), @@ -7134,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 93b97c4f1376c..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, @@ -913,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 d2e1ca50c812d..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, diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index 93b97c4f1376c..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, @@ -913,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 8940b16bf83f5..d10ed738df288 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; @@ -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, @@ -544,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, diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index c58047c4abe10..d6fee253cba21 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. 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_mode = always; + # 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_mode; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index d64efe80ccae5..bc95e3f9b162c 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. 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_mode = always; + 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_mode; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 1adf98f67ff99..14b270214ada8 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, 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`. diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index 4ef0b5c74f3e7..09fe5cd879430 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. 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_mode = always; + 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_mode; diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index cb3be93191fb2..e729b7b7a020c 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. 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_mode = always; + # 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_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 e879947e324bb..5bb22083f4c29 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. 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_mode = always; + # 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_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 7ab5e7c79d2ba..186fa926c354f 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. 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_mode = always; + # 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_mode; diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt index f2442762f3fd2..93b5f75f97fd9 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. 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_mode = always; + # 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_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 181e0e0b7f266..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,6 +275,30 @@ if tracking.contains_dynamic_filter() { } ``` +### 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_mode = always; +``` + +See [issue #3463](https://github.com/apache/datafusion/issues/3463) and +[PR #23420](https://github.com/apache/datafusion/pull/23420) for context. + ### `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..c52e6f55d2b8f 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_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`. |