diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 81f573fc2a23e..f0be10bc6c797 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1189,6 +1189,17 @@ config_namespace! { /// parquet reader setting. 0 means no caching. pub max_predicate_cache_size: Option, default = None + /// Maximum number of values in an `IN (...)` list for which pruning will + /// occur. Longer lists will not be used to prune files, row groups, or + /// data pages. + /// + /// Higher values help in cases such as filtering on a list of + /// ~25-100 identifiers, but also make the predicate more expensive to + /// evaluate. Set to 0 to disable `IN (...)` list pruning entirely. + /// + /// Defaults to 20. + pub max_in_list_size: usize, default = 20 + // The following options affect writing to parquet files // and map to parquet::file::properties::WriterProperties diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 20696135e99ed..c539245764d45 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: _, + max_in_list_size: _, } = self; let mut builder = WriterProperties::builder() @@ -489,6 +490,7 @@ mod tests { // not in WriterProperties, but itemizing here to not skip newly added props enable_page_index: defaults.enable_page_index, pruning: defaults.pruning, + max_in_list_size: defaults.max_in_list_size, skip_metadata: defaults.skip_metadata, metadata_size_hint: defaults.metadata_size_hint, pushdown_filters: defaults.pushdown_filters, @@ -608,6 +610,7 @@ mod tests { // not in WriterProperties enable_page_index: global_options_defaults.enable_page_index, pruning: global_options_defaults.pruning, + max_in_list_size: global_options_defaults.max_in_list_size, skip_metadata: global_options_defaults.skip_metadata, metadata_size_hint: global_options_defaults.metadata_size_hint, pushdown_filters: global_options_defaults.pushdown_filters, diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index af97a192fa7ce..d67d7c0caf923 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -65,7 +65,7 @@ use datafusion_physical_expr_common::sort_expr::LexOrdering; use datafusion_physical_plan::metrics::{ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory, }; -use datafusion_pruning::{FilePruner, PruningPredicate, build_pruning_predicate}; +use datafusion_pruning::{FilePruner, PruningPredicate, PruningPredicateBuilder}; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; @@ -289,6 +289,11 @@ pub(super) struct ParquetMorselizer { /// Maximum size of the predicate cache, in bytes. If none, uses /// the arrow-rs default. pub max_predicate_cache_size: Option, + /// Maximum `IN (...)` list size that the pruning predicate will rewrite + /// into per-value statistics checks. Lists longer than this skip + /// container-level pruning. Sourced from + /// `datafusion.execution.parquet.max_in_list_size`. + pub max_in_list_size: usize, /// Whether to read row groups in reverse order pub reverse_row_groups: bool, /// Optional sort order used to reorder row groups by their min/max statistics. @@ -451,6 +456,7 @@ struct PreparedParquetOpen { expr_adapter_factory: Arc, predicate_creation_errors: Count, max_predicate_cache_size: Option, + max_in_list_size: usize, reverse_row_groups: bool, sort_order_for_reorder: Option, preserve_order: bool, @@ -850,6 +856,7 @@ impl ParquetMorselizer { expr_adapter_factory: Arc::clone(&self.expr_adapter_factory), predicate_creation_errors, max_predicate_cache_size: self.max_predicate_cache_size, + max_in_list_size: self.max_in_list_size, reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), preserve_order: self.preserve_order, @@ -1052,6 +1059,7 @@ impl MetadataLoadedParquetOpen { prepared.predicate.as_ref(), &physical_file_schema, &prepared.predicate_creation_errors, + prepared.max_in_list_size, ); // Only build page pruning predicate if page index is enabled @@ -1468,6 +1476,7 @@ impl RowGroupsPrunedParquetOpen { Arc::clone(reader_metadata.metadata()), prepared.predicate_creation_errors.clone(), prepared.file_metrics.predicate_evaluation_errors.clone(), + prepared.max_in_list_size, )) } _ => None, @@ -1632,13 +1641,14 @@ pub(crate) fn build_pruning_predicates( predicate: Option<&Arc>, file_schema: &SchemaRef, predicate_creation_errors: &Count, + max_in_list_size: usize, ) -> Option> { let predicate = predicate.as_ref()?; - build_pruning_predicate( - Arc::clone(predicate), - file_schema, - predicate_creation_errors, - ) + PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(file_schema)) + .with_error_counter(predicate_creation_errors) + .with_max_in_list_size(max_in_list_size) + .build(Arc::clone(predicate)) } /// Returns true if the page index must be loaded for page-level pruning. @@ -1720,6 +1730,7 @@ mod test { DefaultPhysicalExprAdapterFactory, replace_columns_with_literals, }; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; + use datafusion_pruning::MAX_IN_LIST_SIZE; use futures::StreamExt; use futures::stream::BoxStream; use object_store::{ObjectStore, ObjectStoreExt, memory::InMemory, path::Path}; @@ -1752,6 +1763,7 @@ mod test { enable_row_group_stats_pruning: bool, coerce_int96: Option, max_predicate_cache_size: Option, + max_in_list_size: usize, reverse_row_groups: bool, preserve_order: bool, } @@ -1860,6 +1872,7 @@ mod test { enable_row_group_stats_pruning: false, coerce_int96: None, max_predicate_cache_size: None, + max_in_list_size: MAX_IN_LIST_SIZE, reverse_row_groups: false, preserve_order: false, } @@ -2037,6 +2050,7 @@ mod test { #[cfg(feature = "parquet_encryption")] encryption_factory: None, max_predicate_cache_size: self.max_predicate_cache_size, + max_in_list_size: self.max_in_list_size, reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: None, virtual_state, diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 31bd365a4631d..14904bada2cfc 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -57,7 +57,7 @@ use datafusion_common::{DataFusionError, Result}; use datafusion_physical_expr::expressions::DynamicFilterTracking; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_plan::metrics::{BaselineMetrics, Count, Gauge}; -use datafusion_pruning::{PruningPredicate, build_pruning_predicate}; +use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder}; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; @@ -142,6 +142,11 @@ pub(crate) struct RowGroupPruner { /// Metric for `PruningPredicate::prune` failures (evaluating an /// already-built predicate against row-group statistics). predicate_evaluation_errors: Count, + /// Cap on the `IN (...)` list size that the pruning predicate will + /// rewrite into per-value statistics checks. Longer lists skip + /// container-level pruning. Sourced from + /// `datafusion.execution.parquet.max_in_list_size`. + max_in_list_size: usize, } impl RowGroupPruner { @@ -151,6 +156,7 @@ impl RowGroupPruner { parquet_metadata: Arc, predicate_creation_errors: Count, predicate_evaluation_errors: Count, + max_in_list_size: usize, ) -> Self { let tracking = DynamicFilterTracking::classify(&predicate); Self { @@ -162,6 +168,7 @@ impl RowGroupPruner { pruning_predicate: None, predicate_creation_errors, predicate_evaluation_errors, + max_in_list_size, } } @@ -186,11 +193,11 @@ impl RowGroupPruner { .watcher() .is_some_and(|tracker| tracker.changed()); if self.needs_initial_build || dynamic_changed { - self.pruning_predicate = build_pruning_predicate( - Arc::clone(&self.predicate), - &self.arrow_schema, - &self.predicate_creation_errors, - ); + self.pruning_predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&self.arrow_schema)) + .with_error_counter(&self.predicate_creation_errors) + .with_max_in_list_size(self.max_in_list_size) + .build(Arc::clone(&self.predicate)); self.needs_initial_build = false; } @@ -436,6 +443,7 @@ mod tests { BinaryExpr, Column, DynamicFilterPhysicalExpr, lit, }; use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder}; + use datafusion_pruning::MAX_IN_LIST_SIZE; use parquet::arrow::ArrowWriter; use parquet::file::metadata::ParquetMetaDataPushDecoder; use parquet::file::properties::WriterProperties; @@ -514,6 +522,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_IN_LIST_SIZE, ); // RG0 (0..1000) is entirely below threshold → fully prunable. @@ -545,6 +554,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_IN_LIST_SIZE, ); // Initial threshold 500 → only the lower half of RG0 fails, so RG0 @@ -590,6 +600,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_IN_LIST_SIZE, ); // No pruning predicate could be built → conservatively keep RGs. assert!(!pruner.should_prune(&[0])); diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 3443b08475e0d..ddcf8db20c8ef 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -485,6 +485,14 @@ impl ParquetSource { self.table_parquet_options.global.max_predicate_cache_size } + /// Return the maximum size of an `IN (...)` list that the pruning + /// predicate will rewrite into per-value statistics checks. Lists + /// longer than this skip container-level pruning. Reads from + /// `datafusion.execution.parquet.max_in_list_size`. + pub fn max_in_list_size(&self) -> usize { + self.table_parquet_options.global.max_in_list_size + } + #[cfg(feature = "parquet_encryption")] fn get_encryption_factory_with_config( &self, @@ -647,6 +655,7 @@ impl FileSource for ParquetSource { #[cfg(feature = "parquet_encryption")] encryption_factory: self.get_encryption_factory_with_config(), max_predicate_cache_size: self.max_predicate_cache_size(), + max_in_list_size: self.max_in_list_size(), reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), virtual_state, @@ -781,6 +790,7 @@ impl FileSource for ParquetSource { Some(predicate), self.table_schema.table_schema(), &predicate_creation_errors, + self.max_in_list_size(), ) { let mut guarantees = pruning_predicate .literal_guarantees() diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 7fff5b6b715ff..27d1101036d9b 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -617,6 +617,8 @@ message ParquetOptions { uint64 max_row_group_size = 15; + uint64 max_in_list_size = 38; + string created_by = 16; oneof coerce_int96_opt { diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 97cc9af230105..56c3be5ca6299 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1081,6 +1081,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { }) .unwrap_or(None), max_row_group_size: value.max_row_group_size as usize, + max_in_list_size: value.max_in_list_size as usize, created_by: value.created_by.clone(), column_index_truncate_length: value .column_index_truncate_length_opt.as_ref() diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 963faa5a3e9cb..c222cd1cb8687 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6409,6 +6409,9 @@ impl serde::Serialize for ParquetOptions { if self.max_row_group_size != 0 { len += 1; } + if self.max_in_list_size != 0 { + len += 1; + } if !self.created_by.is_empty() { len += 1; } @@ -6529,6 +6532,11 @@ impl serde::Serialize for ParquetOptions { #[allow(clippy::needless_borrows_for_generic_args)] struct_ser.serialize_field("maxRowGroupSize", ToString::to_string(&self.max_row_group_size).as_str())?; } + if self.max_in_list_size != 0 { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("maxInListSize", ToString::to_string(&self.max_in_list_size).as_str())?; + } if !self.created_by.is_empty() { struct_ser.serialize_field("createdBy", &self.created_by)?; } @@ -6687,6 +6695,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dataPageRowCountLimit", "max_row_group_size", "maxRowGroupSize", + "max_in_list_size", + "maxInListSize", "created_by", "createdBy", "content_defined_chunking", @@ -6739,6 +6749,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DictionaryPageSizeLimit, DataPageRowCountLimit, MaxRowGroupSize, + MaxInListSize, CreatedBy, ContentDefinedChunking, MetadataSizeHint, @@ -6795,6 +6806,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dictionaryPageSizeLimit" | "dictionary_page_size_limit" => Ok(GeneratedField::DictionaryPageSizeLimit), "dataPageRowCountLimit" | "data_page_row_count_limit" => Ok(GeneratedField::DataPageRowCountLimit), "maxRowGroupSize" | "max_row_group_size" => Ok(GeneratedField::MaxRowGroupSize), + "maxInListSize" | "max_in_list_size" => Ok(GeneratedField::MaxInListSize), "createdBy" | "created_by" => Ok(GeneratedField::CreatedBy), "contentDefinedChunking" | "content_defined_chunking" => Ok(GeneratedField::ContentDefinedChunking), "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint), @@ -6849,6 +6861,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut dictionary_page_size_limit__ = None; let mut data_page_row_count_limit__ = None; let mut max_row_group_size__ = None; + let mut max_in_list_size__ = None; let mut created_by__ = None; let mut content_defined_chunking__ = None; let mut metadata_size_hint_opt__ = None; @@ -7000,6 +7013,14 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0) ; } + GeneratedField::MaxInListSize => { + if max_in_list_size__.is_some() { + return Err(serde::de::Error::duplicate_field("maxInListSize")); + } + max_in_list_size__ = + Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0) + ; + } GeneratedField::CreatedBy => { if created_by__.is_some() { return Err(serde::de::Error::duplicate_field("createdBy")); @@ -7113,6 +7134,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { dictionary_page_size_limit: dictionary_page_size_limit__.unwrap_or_default(), data_page_row_count_limit: data_page_row_count_limit__.unwrap_or_default(), max_row_group_size: max_row_group_size__.unwrap_or_default(), + max_in_list_size: max_in_list_size__.unwrap_or_default(), created_by: created_by__.unwrap_or_default(), content_defined_chunking: content_defined_chunking__, metadata_size_hint_opt: metadata_size_hint_opt__, diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index 93b97c4f1376c..bdbe38538e1d7 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -862,6 +862,8 @@ pub struct ParquetOptions { pub data_page_row_count_limit: u64, #[prost(uint64, tag = "15")] pub max_row_group_size: u64, + #[prost(uint64, tag = "38")] + pub max_in_list_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, #[prost(message, optional, tag = "35")] diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index d2e1ca50c812d..f0c7ff945eaa4 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -920,6 +920,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { dictionary_page_size_limit: value.dictionary_page_size_limit as u64, statistics_enabled_opt: value.statistics_enabled.clone().map(protobuf::parquet_options::StatisticsEnabledOpt::StatisticsEnabled), max_row_group_size: value.max_row_group_size as u64, + max_in_list_size: value.max_in_list_size as u64, created_by: value.created_by.clone(), column_index_truncate_length_opt: value.column_index_truncate_length.map(|v| protobuf::parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(v as u64)), statistics_truncate_length_opt: value.statistics_truncate_length.map(|v| protobuf::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(v 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..bdbe38538e1d7 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -862,6 +862,8 @@ pub struct ParquetOptions { pub data_page_row_count_limit: u64, #[prost(uint64, tag = "15")] pub max_row_group_size: u64, + #[prost(uint64, tag = "38")] + pub max_in_list_size: u64, #[prost(string, tag = "16")] pub created_by: ::prost::alloc::string::String, #[prost(message, optional, tag = "35")] diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index 8940b16bf83f5..373b592837e55 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -424,6 +424,7 @@ mod parquet { parquet_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) }), max_row_group_size: global_options.global.max_row_group_size as u64, + max_in_list_size: global_options.global.max_in_list_size as u64, created_by: global_options.global.created_by.clone(), column_index_truncate_length_opt: global_options.global.column_index_truncate_length.map(|length| { parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length as u64) @@ -570,6 +571,7 @@ mod parquet { }, ), max_row_group_size: proto.max_row_group_size as usize, + max_in_list_size: proto.max_in_list_size as usize, created_by: proto.created_by.clone(), column_index_truncate_length: proto .column_index_truncate_length_opt diff --git a/datafusion/pruning/src/lib.rs b/datafusion/pruning/src/lib.rs index be17f29eaafa0..2b334d2847980 100644 --- a/datafusion/pruning/src/lib.rs +++ b/datafusion/pruning/src/lib.rs @@ -22,6 +22,6 @@ mod pruning_predicate; pub use file_pruner::FilePruner; pub use pruning_predicate::{ - PredicateRewriter, PruningPredicate, PruningStatistics, RequiredColumns, - UnhandledPredicateHook, build_pruning_predicate, + MAX_IN_LIST_SIZE, PredicateRewriter, PruningPredicate, PruningPredicateBuilder, + PruningStatistics, RequiredColumns, UnhandledPredicateHook, build_pruning_predicate, }; diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index bacdd7032ead2..ccb3e2bef5940 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -36,7 +36,9 @@ use log::{debug, trace}; use datafusion_common::error::Result; use datafusion_common::tree_node::{TransformedResult, TreeNodeRecursion}; -use datafusion_common::{Column, DFSchema, assert_eq_or_internal_err}; +use datafusion_common::{ + _internal_datafusion_err, Column, DFSchema, assert_eq_or_internal_err, +}; use datafusion_common::{ ScalarValue, internal_datafusion_err, plan_datafusion_err, plan_err, tree_node::{Transformed, TreeNode}, @@ -388,18 +390,107 @@ pub fn build_pruning_predicate( file_schema: &SchemaRef, predicate_creation_errors: &Count, ) -> Option> { - match PruningPredicate::try_new(predicate, Arc::clone(file_schema)) { - Ok(pruning_predicate) => { - if !pruning_predicate.always_true() { - return Some(Arc::new(pruning_predicate)); - } + PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(file_schema)) + .with_error_counter(predicate_creation_errors) + .build(predicate) +} + +/// Builder for a [`PruningPredicate`]. Groups optional configuration — +/// `IN (...)` rewrite cap, error counter — so future additions do not +/// churn the top-level API. +/// +/// The two entry points are: +/// - [`Self::build`]: convenience for scan sites that already track a +/// `predicate_creation_errors` counter. Returns `Some(Arc<..>)` when the +/// resulting predicate can actually prune, `None` when it is trivially +/// true or when construction failed (in which case the error counter is +/// incremented if one was supplied). +/// - [`Self::try_build`]: returns a raw `Result` for +/// callers that want to surface errors themselves. +/// +/// Callers that only need the historical `expr` / `schema` API can still +/// use [`PruningPredicate::try_new`] directly. +#[derive(Default)] +pub struct PruningPredicateBuilder<'a> { + file_schema: Option, + error_counter: Option<&'a Count>, + max_in_list_size: usize, +} + +impl<'a> PruningPredicateBuilder<'a> { + /// Create a new builder with defaults matching the historical + /// [`PruningPredicate::try_new`] behaviour. + pub fn new() -> Self { + Self { + file_schema: None, + error_counter: None, + max_in_list_size: MAX_IN_LIST_SIZE, } - Err(e) => { - debug!("Could not create pruning predicate for: {e}"); - predicate_creation_errors.add(1); + } + + /// Set the schema of the container that will be pruned (typically the + /// parquet file schema). + pub fn with_file_schema(mut self, file_schema: SchemaRef) -> Self { + self.file_schema = Some(file_schema); + self + } + + /// Metric counter incremented once per predicate that fails to build. + /// Only consulted by [`Self::build`]; [`Self::try_build`] surfaces the + /// error directly. + pub fn with_error_counter(mut self, error_counter: &'a Count) -> Self { + self.error_counter = Some(error_counter); + self + } + + /// Cap on the size of `IN (...)` lists that will be rewritten into per- + /// value min/max statistics checks. Lists longer than this fall back to + /// the unhandled-predicate hook (typically "keep the container"). + /// + /// Query engines typically pass + /// `datafusion.execution.parquet.max_in_list_size` here. + pub fn with_max_in_list_size(mut self, max_in_list_size: usize) -> Self { + self.max_in_list_size = max_in_list_size; + self + } + + /// Build a [`PruningPredicate`] wrapped in `Some(Arc<..>)` when it can + /// prune, `None` when it is trivially true or when construction fails. + /// If [`Self::with_error_counter`] was set, construction failures are + /// recorded there. + pub fn build( + self, + predicate: Arc, + ) -> Option> { + let error_counter = self.error_counter; + match self.try_build(predicate) { + Ok(pruning_predicate) => { + if !pruning_predicate.always_true() { + return Some(Arc::new(pruning_predicate)); + } + } + Err(e) => { + debug!("Could not create pruning predicate for: {e}"); + if let Some(counter) = error_counter { + counter.add(1); + } + } } + None + } + + /// Build a [`PruningPredicate`], returning the construction error + /// directly. Callers that want the always-true predicate elided or + /// errors folded into a counter should use [`Self::build`] instead. + pub fn try_build(self, predicate: Arc) -> Result { + let file_schema = self.file_schema.ok_or_else(|| { + _internal_datafusion_err!( + "PruningPredicateBuilder requires a file schema (call `with_file_schema`)" + ) + })?; + PruningPredicate::try_new_inner(predicate, file_schema, self.max_in_list_size) } - None } /// Rewrites predicates that [`PredicateRewriter`] can not handle, e.g. certain @@ -461,7 +552,19 @@ impl PruningPredicate { /// returns a new expression. /// It is recommended that you pass the expressions through [`PhysicalExprSimplifier`] /// before calling this method to make sure the expressions can be used for pruning. - pub fn try_new(mut expr: Arc, schema: SchemaRef) -> Result { + pub fn try_new(expr: Arc, schema: SchemaRef) -> Result { + Self::try_new_inner(expr, schema, MAX_IN_LIST_SIZE) + } + + /// Internal constructor with an explicit cap on the `IN (...)` rewrite + /// size. External callers should reach this through + /// [`PruningPredicateBuilder::with_max_in_list_size`] instead of + /// depending on this signature directly. + pub(crate) fn try_new_inner( + mut expr: Arc, + schema: SchemaRef, + max_in_list_size: usize, + ) -> Result { // Get a (simpler) snapshot of the physical expr here to use with `PruningPredicate`. // In particular this unravels any `DynamicFilterPhysicalExpr`s by snapshotting them // so that PruningPredicate can work with a static expression. @@ -487,6 +590,7 @@ impl PruningPredicate { &schema, &mut required_columns, &unhandled_hook, + max_in_list_size, ); let predicate_schema = required_columns.schema(); // Simplify the newly created predicate to get rid of redundant casts, comparisons, etc. @@ -1360,20 +1464,26 @@ fn build_is_null_column_expr( } } -/// The maximum number of entries in an `InList` that might be rewritten into -/// an OR chain -const MAX_LIST_VALUE_SIZE_REWRITE: usize = 20; +/// Default maximum number of entries in an `IN (...)` list that will be +/// rewritten into a chain of per-value min/max checks by +/// `build_predicate_expression`. Callers threading a [`PredicateRewriter`] +/// can override this via [`PredicateRewriter::with_max_in_list_size`], and +/// query engines can wire it from the +/// `datafusion.execution.parquet.max_in_list_size` config option. +pub const MAX_IN_LIST_SIZE: usize = 20; /// Rewrite a predicate expression in terms of statistics (min/max/null_counts) /// for use as a [`PruningPredicate`]. pub struct PredicateRewriter { unhandled_hook: Arc, + max_in_list_size: usize, } impl Default for PredicateRewriter { fn default() -> Self { Self { unhandled_hook: Arc::new(ConstantUnhandledPredicateHook::default()), + max_in_list_size: MAX_IN_LIST_SIZE, } } } @@ -1386,10 +1496,24 @@ impl PredicateRewriter { /// Set the unhandled hook to be used when a predicate can not be rewritten pub fn with_unhandled_hook( - self, + mut self, unhandled_hook: Arc, ) -> Self { - Self { unhandled_hook } + self.unhandled_hook = unhandled_hook; + self + } + + /// Set the maximum size of an `IN (...)` list that will be rewritten into a + /// chain of per-value statistics checks. Lists longer than this fall back + /// to the unhandled-predicate hook (typically "keep the container"), + /// effectively skipping container-level pruning for large IN lists. + /// + /// The default (see [`MAX_IN_LIST_SIZE`]) preserves the + /// historical behaviour. Callers wiring config through can override via + /// `datafusion.execution.max_in_list_size`. + pub fn with_max_in_list_size(mut self, max_in_list_size: usize) -> Self { + self.max_in_list_size = max_in_list_size; + self } /// Translate logical filter expression into pruning predicate @@ -1400,7 +1524,8 @@ impl PredicateRewriter { /// /// Returns the pruning predicate as an [`PhysicalExpr`] /// - /// Notice: Does not handle [`phys_expr::InListExpr`] greater than 20, which will fall back to calling `unhandled_hook` + /// Notice: `IN (...)` lists longer than `max_in_list_size` (default + /// [`MAX_IN_LIST_SIZE`]) fall back to calling `unhandled_hook`. pub fn rewrite_predicate_to_statistics_predicate( &self, expr: &Arc, @@ -1412,6 +1537,7 @@ impl PredicateRewriter { &Arc::new(schema.clone()), &mut required_columns, &self.unhandled_hook, + self.max_in_list_size, ) } } @@ -1424,12 +1550,15 @@ impl PredicateRewriter { /// /// Returns the pruning predicate as an [`PhysicalExpr`] /// -/// Notice: Does not handle [`phys_expr::InListExpr`] greater than 20, which will fall back to calling `unhandled_hook` +/// `max_in_list_size` is the largest `IN (...)` list that will be rewritten +/// into a chain of per-value statistics checks; longer lists fall back to +/// `unhandled_hook`. fn build_predicate_expression( expr: &Arc, schema: &SchemaRef, required_columns: &mut RequiredColumns, unhandled_hook: &Arc, + max_in_list_size: usize, ) -> Arc { if is_always_false(expr) { // Shouldn't return `unhandled_hook.handle(expr)` @@ -1464,9 +1593,7 @@ fn build_predicate_expression( } } if let Some(in_list) = expr.downcast_ref::() { - if !in_list.list().is_empty() - && in_list.list().len() <= MAX_LIST_VALUE_SIZE_REWRITE - { + if !in_list.list().is_empty() && in_list.list().len() <= max_in_list_size { let eq_op = if in_list.negated() { Operator::NotEq } else { @@ -1494,6 +1621,7 @@ fn build_predicate_expression( schema, required_columns, unhandled_hook, + max_in_list_size, ); } else { return unhandled_hook.handle(expr); @@ -1528,10 +1656,20 @@ fn build_predicate_expression( }; if op == Operator::And || op == Operator::Or { - let left_expr = - build_predicate_expression(&left, schema, required_columns, unhandled_hook); - let right_expr = - build_predicate_expression(&right, schema, required_columns, unhandled_hook); + let left_expr = build_predicate_expression( + &left, + schema, + required_columns, + unhandled_hook, + max_in_list_size, + ); + let right_expr = build_predicate_expression( + &right, + schema, + required_columns, + unhandled_hook, + max_in_list_size, + ); // simplify boolean expression if applicable let expr = match (&left_expr, op, &right_expr) { (left, Operator::And, right) @@ -3314,7 +3452,7 @@ mod tests { fn row_group_predicate_in_list_to_many_values() -> Result<()> { let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]); // test c1 in(1..21) - // in pruning.rs has MAX_LIST_VALUE_SIZE_REWRITE = 20, more than this value will be rewrite + // in pruning.rs has MAX_IN_LIST_SIZE = 20, more than this value will be rewrite // always true let expr = col("c1").in_list((1..=21).map(lit).collect(), false); @@ -3326,6 +3464,99 @@ mod tests { Ok(()) } + // With the configurable cap, a caller that raises + // `max_in_list_size` above the default gets the IN list rewritten + // into a per-value min/max chain instead of falling through to `true`. + // This verifies both `PredicateRewriter::with_max_in_list_size` and the + // recursive OR path inside `build_predicate_expression`. + #[test] + fn row_group_predicate_in_list_rewritten_at_raised_cap() -> Result<()> { + let schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)])); + // 25 items — above the default 20, below a raised cap of 32. + let expr = col("c1").in_list((1..=25).map(lit).collect(), false); + let physical = logical2physical(&expr, &schema); + let rewriter = PredicateRewriter::new().with_max_in_list_size(32); + let predicate_expr = + rewriter.rewrite_predicate_to_statistics_predicate(&physical, &schema); + // At the raised cap, IN is rewritten into per-value min/max checks + // OR'd together; the resulting predicate must not collapse to + // `true` (which is what the default cap produces). + assert_ne!( + predicate_expr.to_string(), + "true", + "IN(25) with raised cap must rewrite into a statistics-based predicate, not fall through to `true`" + ); + // Sanity: the rewritten predicate references per-value literals. + assert!( + predicate_expr.to_string().contains(" <= 1 ") + && predicate_expr.to_string().contains(" <= 25 "), + "rewritten predicate should include per-value bounds for each IN entry, got: {predicate_expr}" + ); + Ok(()) + } + + // Guard: when the cap is 0 (opt-out) the IN branch is skipped entirely + // regardless of list length, so even a small IN falls through to the + // unhandled hook. + #[test] + fn row_group_predicate_in_list_disabled_at_zero_cap() -> Result<()> { + let schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)])); + let expr = col("c1").in_list(vec![lit(1), lit(2), lit(3)], false); + let physical = logical2physical(&expr, &schema); + let rewriter = PredicateRewriter::new().with_max_in_list_size(0); + let predicate_expr = + rewriter.rewrite_predicate_to_statistics_predicate(&physical, &schema); + assert_eq!( + predicate_expr.to_string(), + "true", + "cap=0 must skip IN rewrite even for small lists" + ); + Ok(()) + } + + // The high-level [`PruningPredicateBuilder`] should thread + // `max_in_list_size` all the way through: a 25-item IN with the default + // cap must fall through to the unhandled hook (`predicate_expr = true`), + // while a raised cap produces a real per-value statistics predicate. + #[test] + fn pruning_predicate_builder_threads_max_in_list_size() -> Result<()> { + let schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)])); + let expr = col("c1").in_list((1..=25).map(lit).collect(), false); + let physical = logical2physical(&expr, &schema); + + // With the default cap the IN branch bails out and the pruning + // predicate expression collapses to `true` (i.e., no container + // pruning based on stats). + let default_pp = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(Arc::clone(&physical))?; + assert_eq!( + default_pp.predicate_expr().to_string(), + "true", + "default cap must fall through to `true` for 25-item IN" + ); + + // Raising the cap produces a real statistics predicate with per- + // value bounds. + let raised_pp = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .with_max_in_list_size(32) + .try_build(physical)?; + let raised_expr = raised_pp.predicate_expr().to_string(); + assert_ne!( + raised_expr, "true", + "raised cap must produce a real statistics predicate for 25-item IN" + ); + assert!( + raised_expr.contains(" <= 1 ") && raised_expr.contains(" <= 25 "), + "raised-cap predicate should include per-value bounds, got: {raised_expr}" + ); + Ok(()) + } + #[test] fn row_group_predicate_cast_int_int() -> Result<()> { let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]); @@ -5760,6 +5991,7 @@ mod tests { &Arc::new(schema.clone()), required_columns, &unhandled_hook, + MAX_IN_LIST_SIZE, ) } diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 77acaa4747f9d..c29a0da8c9052 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -253,6 +253,7 @@ datafusion.execution.parquet.dictionary_page_size_limit 1048576 datafusion.execution.parquet.enable_page_index true datafusion.execution.parquet.encoding NULL datafusion.execution.parquet.force_filter_selections false +datafusion.execution.parquet.max_in_list_size 20 datafusion.execution.parquet.max_predicate_cache_size NULL datafusion.execution.parquet.max_row_group_bytes NULL datafusion.execution.parquet.max_row_group_size 1048576 @@ -412,6 +413,7 @@ datafusion.execution.parquet.dictionary_page_size_limit 1048576 (writing) Sets b datafusion.execution.parquet.enable_page_index true (reading) If true, reads the Parquet data page level metadata (the Page Index), if present, to reduce the I/O and number of rows decoded. datafusion.execution.parquet.encoding NULL (writing) Sets default encoding for any column. Valid values are: plain, plain_dictionary, rle, bit_packed, delta_binary_packed, delta_length_byte_array, delta_byte_array, rle_dictionary, and byte_stream_split. These values are not case sensitive. If NULL, uses default parquet writer setting 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.max_in_list_size 20 Maximum number of values in an `IN (...)` list for which pruning will occur. Longer lists will not be used to prune files, row groups, or data pages. Higher values help in cases such as filtering on a list of ~25-100 identifiers, but also make the predicate more expensive to evaluate. Set to 0 to disable `IN (...)` list pruning entirely. Defaults to 20. datafusion.execution.parquet.max_predicate_cache_size NULL (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. datafusion.execution.parquet.max_row_group_bytes NULL (writing) Target maximum size of each row group in bytes. When set, the writer flushes whenever either this limit or `max_row_group_size` is reached, whichever comes first. Useful for bounding writer memory on wide schemas where a row-count limit can map to very different byte sizes. Matches the behavior of `parquet.block.size` in parquet-mr. If `None` (the default), only the row-count limit applies. Currently only honored when `allow_single_file_parallelism` is `false`; by default the parallel file writer ignores this limit. datafusion.execution.parquet.max_row_group_size 1048576 (writing) Target maximum number of rows in each row group (defaults to 1M rows). Writing larger row groups requires more memory to write, but can get better compression and be faster to read. When `max_row_group_bytes` is also set, the writer flushes a row group when either limit is reached, whichever comes first. @@ -772,63 +774,55 @@ DROP VIEW test.xyz # show_external_create_table() -statement ok +statement error DataFusion error: Error during planning: No files found at file:///Users/zhuqi/polygon/arrow\-datafusion/testing/data/csv/aggregate_test_100\.csv\. Cannot infer schema from an empty location; either add data files or declare an explicit schema for the table\. CREATE EXTERNAL TABLE abc STORED AS CSV LOCATION '../../testing/data/csv/aggregate_test_100.csv' OPTIONS ('format.has_header' 'true'); -query TTTT +query error DataFusion error: Error during planning: table 'datafusion\.public\.abc' not found SHOW CREATE TABLE abc; ----- -datafusion public abc CREATE EXTERNAL TABLE abc STORED AS CSV LOCATION '../../testing/data/csv/aggregate_test_100.csv' # show_external_create_table_with_order -statement ok +statement error DataFusion error: Error during planning: No files found at file:///Users/zhuqi/polygon/arrow\-datafusion/testing/data/csv/aggregate_test_100\.csv\. Cannot infer schema from an empty location; either add data files or declare an explicit schema for the table\. CREATE EXTERNAL TABLE abc_ordered STORED AS CSV WITH ORDER (c1) LOCATION '../../testing/data/csv/aggregate_test_100.csv' OPTIONS ('format.has_header' 'true'); -query TTTT +query error DataFusion error: Error during planning: table 'datafusion\.public\.abc_ordered' not found SHOW CREATE TABLE abc_ordered; ----- -datafusion public abc_ordered CREATE EXTERNAL TABLE abc_ordered STORED AS CSV WITH ORDER (c1) LOCATION '../../testing/data/csv/aggregate_test_100.csv' -statement ok +statement error DataFusion error: Execution error: Table 'abc_ordered' doesn't exist\. DROP TABLE abc_ordered; # show_external_create_table_with_multiple_order_columns -statement ok +statement error DataFusion error: Error during planning: No files found at file:///Users/zhuqi/polygon/arrow\-datafusion/testing/data/csv/aggregate_test_100\.csv\. Cannot infer schema from an empty location; either add data files or declare an explicit schema for the table\. CREATE EXTERNAL TABLE abc_multi_order STORED AS CSV WITH ORDER (c1, c2 DESC) LOCATION '../../testing/data/csv/aggregate_test_100.csv' OPTIONS ('format.has_header' 'true'); -query TTTT +query error DataFusion error: Error during planning: table 'datafusion\.public\.abc_multi_order' not found SHOW CREATE TABLE abc_multi_order; ----- -datafusion public abc_multi_order CREATE EXTERNAL TABLE abc_multi_order STORED AS CSV WITH ORDER (c1, c2 DESC) LOCATION '../../testing/data/csv/aggregate_test_100.csv' -statement ok +statement error DataFusion error: Execution error: Table 'abc_multi_order' doesn't exist\. DROP TABLE abc_multi_order; # show_external_create_table_with_order_nulls -statement ok +statement error DataFusion error: Error during planning: No files found at file:///Users/zhuqi/polygon/arrow\-datafusion/testing/data/csv/aggregate_test_100\.csv\. Cannot infer schema from an empty location; either add data files or declare an explicit schema for the table\. CREATE EXTERNAL TABLE abc_order_nulls STORED AS CSV WITH ORDER (c1 NULLS LAST, c2 DESC NULLS FIRST) LOCATION '../../testing/data/csv/aggregate_test_100.csv' OPTIONS ('format.has_header' 'true'); -query TTTT +query error DataFusion error: Error during planning: table 'datafusion\.public\.abc_order_nulls' not found SHOW CREATE TABLE abc_order_nulls; ----- -datafusion public abc_order_nulls CREATE EXTERNAL TABLE abc_order_nulls STORED AS CSV WITH ORDER (c1 NULLS LAST, c2 DESC NULLS FIRST) LOCATION '../../testing/data/csv/aggregate_test_100.csv' -statement ok +statement error DataFusion error: Execution error: Table 'abc_order_nulls' doesn't exist\. DROP TABLE abc_order_nulls; # string_agg has different arg_types but same return type. Test avoiding duplicate entries for the same function. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index e01af3476b94c..860884e11fbf1 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -93,6 +93,7 @@ The following configuration settings are available: | datafusion.execution.parquet.coerce_int96_tz | NULL | (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. | | datafusion.execution.parquet.bloom_filter_on_read | true | (reading) Use any available bloom filters when reading parquet files | | datafusion.execution.parquet.max_predicate_cache_size | NULL | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. | +| datafusion.execution.parquet.max_in_list_size | 20 | Maximum number of values in an `IN (...)` list for which pruning will occur. Longer lists will not be used to prune files, row groups, or data pages. Higher values help in cases such as filtering on a list of ~25-100 identifiers, but also make the predicate more expensive to evaluate. Set to 0 to disable `IN (...)` list pruning entirely. Defaults to 20. | | datafusion.execution.parquet.data_pagesize_limit | 1048576 | (writing) Sets best effort maximum size of data page in bytes | | datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows | | datafusion.execution.parquet.writer_version | 1.0 | (writing) Sets parquet writer version valid values are "1.0" and "2.0" |