From 98e7a5ac5bd832a61c15181fe6852712b3d1b0d6 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Mon, 3 Aug 2026 21:02:40 +0800 Subject: [PATCH 1/5] feat(pruning): expose IN-list rewrite size cap as a config option MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue: apache/datafusion#24059 `PruningPredicate` rewrites `col IN (v1..vn)` into a chain of per-value min/max checks (via `build_predicate_expression`), but only when `n` is below a hardcoded `MAX_LIST_VALUE_SIZE_REWRITE = 20`. Beyond that, the IN branch falls through to `unhandled_hook`, which by default returns `TRUE` — so row-group / file-range statistics pruning does not fire at all for IN lists longer than 20. This is problematic for query patterns that pass a batch of identifiers as `col IN (...)` (REST endpoints filtering by a page of ~25-100 values, ORM-generated `WHERE id IN (25 items)` queries, batched crawlers). On a table sorted by `col`, the reader is forced to materialize the filter column across every row group instead of skipping row groups whose stats disagree with the IN set. ## Changes - Add `datafusion.execution.parquet.pruning_max_in_list_size: usize` (default `20`, preserving existing behaviour) to `TableParquetOptions` next to `max_predicate_cache_size`. - Add `PredicateRewriter::with_max_in_list_size(usize) -> Self` builder, mirroring the existing `with_unhandled_hook`. - Add `PruningPredicate::try_new_with_max_in_list_size` variant. - Add `build_pruning_predicate_with_max_in_list_size` variant of the public helper. - Make `MAX_LIST_VALUE_SIZE_REWRITE` `pub const` so callers can reference the historical default explicitly. - Wire the value through `datasource-parquet`: - `ParquetSource::pruning_max_in_list_size()` reads from `TableParquetOptions.global`. - `ParquetMorselizer` / `PreparedParquetOpen` / `RowGroupPruner` carry the value alongside `max_predicate_cache_size`. - `build_pruning_predicates` (opener) accepts the size and forwards to `build_pruning_predicate_with_max_in_list_size`. ## Backward compatibility - Public `PruningPredicate::try_new` and `build_pruning_predicate` are preserved as thin wrappers passing the historical default. - Internal `build_predicate_expression` takes a new `usize` parameter (crate-private). - Default value of the config option is `20`, so behaviour is unchanged unless the option is set explicitly. ## Tests - `row_group_predicate_in_list_rewritten_at_raised_cap`: `PredicateRewriter::with_max_in_list_size(32)` rewrites a 25-item IN into per-value min/max checks instead of falling through to `true`. - `row_group_predicate_in_list_disabled_at_zero_cap`: cap = 0 skips the IN rewrite even for small lists (opt-out path). - Existing `row_group_predicate_in_list_to_many_values` continues to pass, guarding the default-20 behaviour. --- datafusion/common/src/config.rs | 16 ++ .../common/src/file_options/parquet_writer.rs | 1 + .../datasource-parquet/src/opener/mod.rs | 22 ++- .../datasource-parquet/src/push_decoder.rs | 18 +- datafusion/datasource-parquet/src/source.rs | 10 ++ datafusion/pruning/src/lib.rs | 5 +- datafusion/pruning/src/pruning_predicate.rs | 154 ++++++++++++++++-- 7 files changed, 204 insertions(+), 22 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 81f573fc2a23e..41d5c84656059 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1189,6 +1189,22 @@ 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 the + /// pruning predicate will rewrite the list into a chain of per-value + /// statistics checks. Lists longer than this fall back to the + /// unhandled-predicate hook (defaulting to "keep the container"), + /// which effectively skips container-level pruning for large IN + /// lists. + /// + /// Higher values keep row-group / file-range statistics pruning + /// effective for larger IN lists (for example, REST endpoints that + /// filter by a batch of ~25-100 identifiers), at the cost of a + /// larger rewritten predicate expression evaluated for every + /// container. Set to 0 to disable the rewrite path entirely. + /// + /// The default of 20 preserves the previous hardcoded behaviour. + pub pruning_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..d2feeb7dc98d4 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: _, + pruning_max_in_list_size: _, } = self; let mut builder = WriterProperties::builder() diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index af97a192fa7ce..3bd820ecb4981 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -65,7 +65,10 @@ 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, build_pruning_predicate, + build_pruning_predicate_with_max_in_list_size, +}; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; @@ -289,6 +292,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.pruning_max_in_list_size`. + pub pruning_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 +459,7 @@ struct PreparedParquetOpen { expr_adapter_factory: Arc, predicate_creation_errors: Count, max_predicate_cache_size: Option, + pruning_max_in_list_size: usize, reverse_row_groups: bool, sort_order_for_reorder: Option, preserve_order: bool, @@ -850,6 +859,7 @@ impl ParquetMorselizer { expr_adapter_factory: Arc::clone(&self.expr_adapter_factory), predicate_creation_errors, max_predicate_cache_size: self.max_predicate_cache_size, + pruning_max_in_list_size: self.pruning_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 +1062,7 @@ impl MetadataLoadedParquetOpen { prepared.predicate.as_ref(), &physical_file_schema, &prepared.predicate_creation_errors, + prepared.pruning_max_in_list_size, ); // Only build page pruning predicate if page index is enabled @@ -1468,6 +1479,7 @@ impl RowGroupsPrunedParquetOpen { Arc::clone(reader_metadata.metadata()), prepared.predicate_creation_errors.clone(), prepared.file_metrics.predicate_evaluation_errors.clone(), + prepared.pruning_max_in_list_size, )) } _ => None, @@ -1632,12 +1644,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( + build_pruning_predicate_with_max_in_list_size( Arc::clone(predicate), file_schema, predicate_creation_errors, + max_in_list_size, ) } @@ -1696,6 +1710,7 @@ mod test { CachedParquetFileReaderFactory, DefaultParquetFileReaderFactory, ParquetFileReaderFactory, ParquetRowSelection, RowGroupAccess, }; + use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; use arrow::array::{RecordBatch, record_batch}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use bytes::{BufMut, BytesMut}; @@ -1752,6 +1767,7 @@ mod test { enable_row_group_stats_pruning: bool, coerce_int96: Option, max_predicate_cache_size: Option, + pruning_max_in_list_size: usize, reverse_row_groups: bool, preserve_order: bool, } @@ -1860,6 +1876,7 @@ mod test { enable_row_group_stats_pruning: false, coerce_int96: None, max_predicate_cache_size: None, + pruning_max_in_list_size: MAX_LIST_VALUE_SIZE_REWRITE, reverse_row_groups: false, preserve_order: false, } @@ -2037,6 +2054,7 @@ mod test { #[cfg(feature = "parquet_encryption")] encryption_factory: None, max_predicate_cache_size: self.max_predicate_cache_size, + pruning_max_in_list_size: self.pruning_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..5ad91cf5bf547 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -57,7 +57,9 @@ 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, build_pruning_predicate_with_max_in_list_size, +}; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; @@ -142,6 +144,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.pruning_max_in_list_size`. + pruning_max_in_list_size: usize, } impl RowGroupPruner { @@ -151,6 +158,7 @@ impl RowGroupPruner { parquet_metadata: Arc, predicate_creation_errors: Count, predicate_evaluation_errors: Count, + pruning_max_in_list_size: usize, ) -> Self { let tracking = DynamicFilterTracking::classify(&predicate); Self { @@ -162,6 +170,7 @@ impl RowGroupPruner { pruning_predicate: None, predicate_creation_errors, predicate_evaluation_errors, + pruning_max_in_list_size, } } @@ -186,10 +195,11 @@ impl RowGroupPruner { .watcher() .is_some_and(|tracker| tracker.changed()); if self.needs_initial_build || dynamic_changed { - self.pruning_predicate = build_pruning_predicate( + self.pruning_predicate = build_pruning_predicate_with_max_in_list_size( Arc::clone(&self.predicate), &self.arrow_schema, &self.predicate_creation_errors, + self.pruning_max_in_list_size, ); self.needs_initial_build = false; } @@ -436,6 +446,7 @@ mod tests { BinaryExpr, Column, DynamicFilterPhysicalExpr, lit, }; use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder}; + use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; use parquet::arrow::ArrowWriter; use parquet::file::metadata::ParquetMetaDataPushDecoder; use parquet::file::properties::WriterProperties; @@ -514,6 +525,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_LIST_VALUE_SIZE_REWRITE, ); // RG0 (0..1000) is entirely below threshold → fully prunable. @@ -545,6 +557,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_LIST_VALUE_SIZE_REWRITE, ); // Initial threshold 500 → only the lower half of RG0 fails, so RG0 @@ -590,6 +603,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, + MAX_LIST_VALUE_SIZE_REWRITE, ); // 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..4a81963873786 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.pruning_max_in_list_size`. + pub fn pruning_max_in_list_size(&self) -> usize { + self.table_parquet_options.global.pruning_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(), + pruning_max_in_list_size: self.pruning_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.pruning_max_in_list_size(), ) { let mut guarantees = pruning_predicate .literal_guarantees() diff --git a/datafusion/pruning/src/lib.rs b/datafusion/pruning/src/lib.rs index be17f29eaafa0..1aea8e8b6df49 100644 --- a/datafusion/pruning/src/lib.rs +++ b/datafusion/pruning/src/lib.rs @@ -22,6 +22,7 @@ mod pruning_predicate; pub use file_pruner::FilePruner; pub use pruning_predicate::{ - PredicateRewriter, PruningPredicate, PruningStatistics, RequiredColumns, - UnhandledPredicateHook, build_pruning_predicate, + MAX_LIST_VALUE_SIZE_REWRITE, PredicateRewriter, PruningPredicate, PruningStatistics, + RequiredColumns, UnhandledPredicateHook, build_pruning_predicate, + build_pruning_predicate_with_max_in_list_size, }; diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index bacdd7032ead2..1e5f42f4b8084 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -388,7 +388,29 @@ pub fn build_pruning_predicate( file_schema: &SchemaRef, predicate_creation_errors: &Count, ) -> Option> { - match PruningPredicate::try_new(predicate, Arc::clone(file_schema)) { + build_pruning_predicate_with_max_in_list_size( + predicate, + file_schema, + predicate_creation_errors, + MAX_LIST_VALUE_SIZE_REWRITE, + ) +} + +/// Same as [`build_pruning_predicate`] but with an explicit cap on the size +/// of `IN (...)` lists rewritten into per-value statistics checks. Query +/// engines typically pass `datafusion.execution.pruning_max_in_list_size` +/// here. +pub fn build_pruning_predicate_with_max_in_list_size( + predicate: Arc, + file_schema: &SchemaRef, + predicate_creation_errors: &Count, + max_in_list_size: usize, +) -> Option> { + match PruningPredicate::try_new_with_max_in_list_size( + predicate, + Arc::clone(file_schema), + max_in_list_size, + ) { Ok(pruning_predicate) => { if !pruning_predicate.always_true() { return Some(Arc::new(pruning_predicate)); @@ -461,7 +483,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_with_max_in_list_size(expr, schema, MAX_LIST_VALUE_SIZE_REWRITE) + } + + /// Same as [`PruningPredicate::try_new`] but with an explicit cap on the + /// size of `IN (...)` lists rewritten into per-value statistics checks. + /// Query engines typically pass + /// `datafusion.execution.pruning_max_in_list_size` here. + pub fn try_new_with_max_in_list_size( + 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 +521,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 +1395,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.pruning_max_in_list_size` config option. +pub const MAX_LIST_VALUE_SIZE_REWRITE: 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_LIST_VALUE_SIZE_REWRITE, } } } @@ -1386,10 +1427,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_LIST_VALUE_SIZE_REWRITE`]) preserves the + /// historical behaviour. Callers wiring config through can override via + /// `datafusion.execution.pruning_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 +1455,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_LIST_VALUE_SIZE_REWRITE`]) fall back to calling `unhandled_hook`. pub fn rewrite_predicate_to_statistics_predicate( &self, expr: &Arc, @@ -1412,6 +1468,7 @@ impl PredicateRewriter { &Arc::new(schema.clone()), &mut required_columns, &self.unhandled_hook, + self.max_in_list_size, ) } } @@ -1424,12 +1481,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 +1524,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 +1552,7 @@ fn build_predicate_expression( schema, required_columns, unhandled_hook, + max_in_list_size, ); } else { return unhandled_hook.handle(expr); @@ -1528,10 +1587,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) @@ -3326,6 +3395,58 @@ 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(()) + } + #[test] fn row_group_predicate_cast_int_int() -> Result<()> { let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]); @@ -5760,6 +5881,7 @@ mod tests { &Arc::new(schema.clone()), required_columns, &unhandled_hook, + MAX_LIST_VALUE_SIZE_REWRITE, ) } From 5f4cfa1940a99db2b3cb83f31c1603cabfc9790a Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Tue, 4 Aug 2026 13:03:53 +0800 Subject: [PATCH 2/5] Thread pruning_max_in_list_size through proto + fix fmt/unused import The new ParquetOptions::pruning_max_in_list_size field was not carried through the proto layer, so every explicit ParquetOptions initializer in datafusion-proto-common and datafusion-proto failed to compile (E0063), which also cascaded into the MSRV job. - Add uint64 pruning_max_in_list_size = 38 to the ParquetOptions proto message and regenerate proto-common (prost.rs, pbjson.rs) and proto-models (datafusion_proto_common.rs). - Map the field in proto-common from_proto/to_proto and in the proto crate's file_formats TryFromProto/IntoProto for TableParquetOptions. - Reorder the datafusion_pruning import (cargo fmt) and drop the now unused build_pruning_predicate import. --- .../common/src/file_options/parquet_writer.rs | 3 +++ .../datasource-parquet/src/opener/mod.rs | 5 ++--- .../proto/datafusion_common.proto | 2 ++ datafusion/proto-common/src/from_proto/mod.rs | 1 + .../proto-common/src/generated/pbjson.rs | 22 +++++++++++++++++++ .../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 | 2 ++ 9 files changed, 37 insertions(+), 3 deletions(-) diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index d2feeb7dc98d4..b30bee6a6cb2b 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -490,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, + pruning_max_in_list_size: defaults.pruning_max_in_list_size, skip_metadata: defaults.skip_metadata, metadata_size_hint: defaults.metadata_size_hint, pushdown_filters: defaults.pushdown_filters, @@ -609,6 +610,8 @@ mod tests { // not in WriterProperties enable_page_index: global_options_defaults.enable_page_index, pruning: global_options_defaults.pruning, + pruning_max_in_list_size: global_options_defaults + .pruning_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 3bd820ecb4981..0477e2e756f89 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -66,8 +66,7 @@ use datafusion_physical_plan::metrics::{ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory, }; use datafusion_pruning::{ - FilePruner, PruningPredicate, build_pruning_predicate, - build_pruning_predicate_with_max_in_list_size, + FilePruner, PruningPredicate, build_pruning_predicate_with_max_in_list_size, }; #[cfg(feature = "parquet_encryption")] @@ -1710,7 +1709,6 @@ mod test { CachedParquetFileReaderFactory, DefaultParquetFileReaderFactory, ParquetFileReaderFactory, ParquetRowSelection, RowGroupAccess, }; - use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; use arrow::array::{RecordBatch, record_batch}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use bytes::{BufMut, BytesMut}; @@ -1735,6 +1733,7 @@ mod test { DefaultPhysicalExprAdapterFactory, replace_columns_with_literals, }; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; + use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; use futures::StreamExt; use futures::stream::BoxStream; use object_store::{ObjectStore, ObjectStoreExt, memory::InMemory, path::Path}; diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 7fff5b6b715ff..c0ea72a4a7610 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 pruning_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..9cdb0145373c8 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, + pruning_max_in_list_size: value.pruning_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..66019df619684 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.pruning_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.pruning_max_in_list_size != 0 { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("pruningMaxInListSize", ToString::to_string(&self.pruning_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", + "pruning_max_in_list_size", + "pruningMaxInListSize", "created_by", "createdBy", "content_defined_chunking", @@ -6739,6 +6749,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DictionaryPageSizeLimit, DataPageRowCountLimit, MaxRowGroupSize, + PruningMaxInListSize, 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), + "pruningMaxInListSize" | "pruning_max_in_list_size" => Ok(GeneratedField::PruningMaxInListSize), "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 pruning_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::PruningMaxInListSize => { + if pruning_max_in_list_size__.is_some() { + return Err(serde::de::Error::duplicate_field("pruningMaxInListSize")); + } + pruning_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(), + pruning_max_in_list_size: pruning_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..220cb0db0cdbd 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 pruning_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..d61c89d2c2c22 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, + pruning_max_in_list_size: value.pruning_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..220cb0db0cdbd 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 pruning_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..4733c8e814b6d 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, + pruning_max_in_list_size: global_options.global.pruning_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, + pruning_max_in_list_size: proto.pruning_max_in_list_size as usize, created_by: proto.created_by.clone(), column_index_truncate_length: proto .column_index_truncate_length_opt From 4ad8d50c7c04f53c096b2f2e1cdedb559940fe53 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Tue, 4 Aug 2026 16:34:45 +0800 Subject: [PATCH 3/5] Fix doc link, regenerate configs.md + information_schema.slt - cargo doc: the public MAX_LIST_VALUE_SIZE_REWRITE doc linked the private build_predicate_expression via an intra-doc link; demote it to a code span and correct the config path to datafusion.execution.parquet.pruning_max_in_list_size. - Regenerate configs.md for the new pruning_max_in_list_size option. - Add the two pruning_max_in_list_size rows to information_schema.slt (SHOW ALL and the df_settings description listing). --- datafusion/pruning/src/pruning_predicate.rs | 4 ++-- datafusion/sqllogictest/test_files/information_schema.slt | 2 ++ docs/source/user-guide/configs.md | 1 + 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index 1e5f42f4b8084..8b158098ebfe9 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -1397,10 +1397,10 @@ fn build_is_null_column_expr( /// 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`] +/// `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.pruning_max_in_list_size` config option. +/// `datafusion.execution.parquet.pruning_max_in_list_size` config option. pub const MAX_LIST_VALUE_SIZE_REWRITE: usize = 20; /// Rewrite a predicate expression in terms of statistics (min/max/null_counts) diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 77acaa4747f9d..fc8eaa8da00e6 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.pruning_max_in_list_size 20 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.pruning_max_in_list_size 20 Maximum number of values in an `IN (...)` list for which the pruning predicate will rewrite the list into a chain of per-value statistics checks. Lists longer than this fall back to the unhandled-predicate hook (defaulting to "keep the container"), which effectively skips container-level pruning for large IN lists. Higher values keep row-group / file-range statistics pruning effective for larger IN lists (for example, REST endpoints that filter by a batch of ~25-100 identifiers), at the cost of a larger rewritten predicate expression evaluated for every container. Set to 0 to disable the rewrite path entirely. The default of 20 preserves the previous hardcoded behaviour. 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/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index e01af3476b94c..04ebf0b8fbb1b 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.pruning_max_in_list_size | 20 | Maximum number of values in an `IN (...)` list for which the pruning predicate will rewrite the list into a chain of per-value statistics checks. Lists longer than this fall back to the unhandled-predicate hook (defaulting to "keep the container"), which effectively skips container-level pruning for large IN lists. Higher values keep row-group / file-range statistics pruning effective for larger IN lists (for example, REST endpoints that filter by a batch of ~25-100 identifiers), at the cost of a larger rewritten predicate expression evaluated for every container. Set to 0 to disable the rewrite path entirely. The default of 20 preserves the previous hardcoded behaviour. | | 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" | From 10669da86cf5bc9fec22c992f361e7126e365734 Mon Sep 17 00:00:00 2001 From: Qi Zhu Date: Wed, 5 Aug 2026 11:40:46 +0800 Subject: [PATCH 4/5] Update datafusion/common/src/config.rs Co-authored-by: Andrew Lamb --- datafusion/common/src/config.rs | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 41d5c84656059..0aff2f5a40e1f 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1189,20 +1189,15 @@ 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 the - /// pruning predicate will rewrite the list into a chain of per-value - /// statistics checks. Lists longer than this fall back to the - /// unhandled-predicate hook (defaulting to "keep the container"), - /// which effectively skips container-level pruning for large IN - /// lists. + /// 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 keep row-group / file-range statistics pruning - /// effective for larger IN lists (for example, REST endpoints that - /// filter by a batch of ~25-100 identifiers), at the cost of a - /// larger rewritten predicate expression evaluated for every - /// container. Set to 0 to disable the rewrite path entirely. + /// Higher values help in cases such as a list of + /// of ~25-100 identifiers, but also makes the predicate + /// more expensive to evaluate. Set to 0 to disable the IN (..) list pruning entirely /// - /// The default of 20 preserves the previous hardcoded behaviour. + /// Defaults to 20. pub pruning_max_in_list_size: usize, default = 20 // The following options affect writing to parquet files From 92b7e69177b064037efae51d3caa4994a16741e0 Mon Sep 17 00:00:00 2001 From: Qi Zhu <821684824@qq.com> Date: Wed, 5 Aug 2026 16:02:27 +0800 Subject: [PATCH 5/5] Address review: builder API + naming + doc regen - Introduce PruningPredicateBuilder per alamb's suggestion, replacing the ad-hoc try_new_with_max_in_list_size / build_pruning_predicate_with_max_in_list_size helpers with a single builder that also carries the error counter. - Rename config option pruning_max_in_list_size -> max_in_list_size to match the max_predicate_cache_size naming style (per alamb). - Rename const MAX_LIST_VALUE_SIZE_REWRITE -> MAX_IN_LIST_SIZE so the field name and the const name are consistent (per alamb's opener/mod.rs comment). - Trim the config-option doc (removed accidentally-duplicated 'of' word, tightened wording). - Regenerate docs/source/user-guide/configs.md and datafusion/sqllogictest/test_files/information_schema.slt. - Add PruningPredicateBuilder unit test verifying max_in_list_size is threaded end-to-end (default -> 'true'; raised cap -> real statistics predicate). --- datafusion/common/src/config.rs | 10 +- .../common/src/file_options/parquet_writer.rs | 7 +- .../datasource-parquet/src/opener/mod.rs | 35 ++-- .../datasource-parquet/src/push_decoder.rs | 31 ++- datafusion/datasource-parquet/src/source.rs | 10 +- .../proto/datafusion_common.proto | 2 +- datafusion/proto-common/src/from_proto/mod.rs | 2 +- .../proto-common/src/generated/pbjson.rs | 26 +-- .../proto-common/src/generated/prost.rs | 2 +- datafusion/proto-common/src/to_proto/mod.rs | 2 +- .../src/generated/datafusion_proto_common.rs | 2 +- .../proto/src/logical_plan/file_formats.rs | 4 +- datafusion/pruning/src/lib.rs | 5 +- datafusion/pruning/src/pruning_predicate.rs | 196 ++++++++++++++---- .../test_files/information_schema.slt | 34 ++- docs/source/user-guide/configs.md | 2 +- 16 files changed, 232 insertions(+), 138 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 0aff2f5a40e1f..f0be10bc6c797 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1191,14 +1191,14 @@ config_namespace! { /// 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. + /// data pages. /// - /// Higher values help in cases such as a list of - /// of ~25-100 identifiers, but also makes the predicate - /// more expensive to evaluate. Set to 0 to disable the IN (..) list pruning entirely + /// 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 pruning_max_in_list_size: usize, default = 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 b30bee6a6cb2b..c539245764d45 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: _, - pruning_max_in_list_size: _, + max_in_list_size: _, } = self; let mut builder = WriterProperties::builder() @@ -490,7 +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, - pruning_max_in_list_size: defaults.pruning_max_in_list_size, + 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, @@ -610,8 +610,7 @@ mod tests { // not in WriterProperties enable_page_index: global_options_defaults.enable_page_index, pruning: global_options_defaults.pruning, - pruning_max_in_list_size: global_options_defaults - .pruning_max_in_list_size, + 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 0477e2e756f89..d67d7c0caf923 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -65,9 +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_with_max_in_list_size, -}; +use datafusion_pruning::{FilePruner, PruningPredicate, PruningPredicateBuilder}; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; @@ -294,8 +292,8 @@ pub(super) struct ParquetMorselizer { /// 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.pruning_max_in_list_size`. - pub pruning_max_in_list_size: usize, + /// `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. @@ -458,7 +456,7 @@ struct PreparedParquetOpen { expr_adapter_factory: Arc, predicate_creation_errors: Count, max_predicate_cache_size: Option, - pruning_max_in_list_size: usize, + max_in_list_size: usize, reverse_row_groups: bool, sort_order_for_reorder: Option, preserve_order: bool, @@ -858,7 +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, - pruning_max_in_list_size: self.pruning_max_in_list_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, @@ -1061,7 +1059,7 @@ impl MetadataLoadedParquetOpen { prepared.predicate.as_ref(), &physical_file_schema, &prepared.predicate_creation_errors, - prepared.pruning_max_in_list_size, + prepared.max_in_list_size, ); // Only build page pruning predicate if page index is enabled @@ -1478,7 +1476,7 @@ impl RowGroupsPrunedParquetOpen { Arc::clone(reader_metadata.metadata()), prepared.predicate_creation_errors.clone(), prepared.file_metrics.predicate_evaluation_errors.clone(), - prepared.pruning_max_in_list_size, + prepared.max_in_list_size, )) } _ => None, @@ -1646,12 +1644,11 @@ pub(crate) fn build_pruning_predicates( max_in_list_size: usize, ) -> Option> { let predicate = predicate.as_ref()?; - build_pruning_predicate_with_max_in_list_size( - Arc::clone(predicate), - file_schema, - predicate_creation_errors, - max_in_list_size, - ) + 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. @@ -1733,7 +1730,7 @@ mod test { DefaultPhysicalExprAdapterFactory, replace_columns_with_literals, }; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; - use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; + use datafusion_pruning::MAX_IN_LIST_SIZE; use futures::StreamExt; use futures::stream::BoxStream; use object_store::{ObjectStore, ObjectStoreExt, memory::InMemory, path::Path}; @@ -1766,7 +1763,7 @@ mod test { enable_row_group_stats_pruning: bool, coerce_int96: Option, max_predicate_cache_size: Option, - pruning_max_in_list_size: usize, + max_in_list_size: usize, reverse_row_groups: bool, preserve_order: bool, } @@ -1875,7 +1872,7 @@ mod test { enable_row_group_stats_pruning: false, coerce_int96: None, max_predicate_cache_size: None, - pruning_max_in_list_size: MAX_LIST_VALUE_SIZE_REWRITE, + max_in_list_size: MAX_IN_LIST_SIZE, reverse_row_groups: false, preserve_order: false, } @@ -2053,7 +2050,7 @@ mod test { #[cfg(feature = "parquet_encryption")] encryption_factory: None, max_predicate_cache_size: self.max_predicate_cache_size, - pruning_max_in_list_size: self.pruning_max_in_list_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 5ad91cf5bf547..14904bada2cfc 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -57,9 +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_with_max_in_list_size, -}; +use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder}; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; @@ -147,8 +145,8 @@ pub(crate) struct RowGroupPruner { /// 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.pruning_max_in_list_size`. - pruning_max_in_list_size: usize, + /// `datafusion.execution.parquet.max_in_list_size`. + max_in_list_size: usize, } impl RowGroupPruner { @@ -158,7 +156,7 @@ impl RowGroupPruner { parquet_metadata: Arc, predicate_creation_errors: Count, predicate_evaluation_errors: Count, - pruning_max_in_list_size: usize, + max_in_list_size: usize, ) -> Self { let tracking = DynamicFilterTracking::classify(&predicate); Self { @@ -170,7 +168,7 @@ impl RowGroupPruner { pruning_predicate: None, predicate_creation_errors, predicate_evaluation_errors, - pruning_max_in_list_size, + max_in_list_size, } } @@ -195,12 +193,11 @@ impl RowGroupPruner { .watcher() .is_some_and(|tracker| tracker.changed()); if self.needs_initial_build || dynamic_changed { - self.pruning_predicate = build_pruning_predicate_with_max_in_list_size( - Arc::clone(&self.predicate), - &self.arrow_schema, - &self.predicate_creation_errors, - self.pruning_max_in_list_size, - ); + 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; } @@ -446,7 +443,7 @@ mod tests { BinaryExpr, Column, DynamicFilterPhysicalExpr, lit, }; use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder}; - use datafusion_pruning::MAX_LIST_VALUE_SIZE_REWRITE; + use datafusion_pruning::MAX_IN_LIST_SIZE; use parquet::arrow::ArrowWriter; use parquet::file::metadata::ParquetMetaDataPushDecoder; use parquet::file::properties::WriterProperties; @@ -525,7 +522,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, - MAX_LIST_VALUE_SIZE_REWRITE, + MAX_IN_LIST_SIZE, ); // RG0 (0..1000) is entirely below threshold → fully prunable. @@ -557,7 +554,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, - MAX_LIST_VALUE_SIZE_REWRITE, + MAX_IN_LIST_SIZE, ); // Initial threshold 500 → only the lower half of RG0 fails, so RG0 @@ -603,7 +600,7 @@ mod tests { Arc::clone(&meta), creation, evaluation, - MAX_LIST_VALUE_SIZE_REWRITE, + 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 4a81963873786..ddcf8db20c8ef 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -488,9 +488,9 @@ impl ParquetSource { /// 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.pruning_max_in_list_size`. - pub fn pruning_max_in_list_size(&self) -> usize { - self.table_parquet_options.global.pruning_max_in_list_size + /// `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")] @@ -655,7 +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(), - pruning_max_in_list_size: self.pruning_max_in_list_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, @@ -790,7 +790,7 @@ impl FileSource for ParquetSource { Some(predicate), self.table_schema.table_schema(), &predicate_creation_errors, - self.pruning_max_in_list_size(), + 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 c0ea72a4a7610..27d1101036d9b 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -617,7 +617,7 @@ message ParquetOptions { uint64 max_row_group_size = 15; - uint64 pruning_max_in_list_size = 38; + uint64 max_in_list_size = 38; string created_by = 16; diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 9cdb0145373c8..56c3be5ca6299 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1081,7 +1081,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { }) .unwrap_or(None), max_row_group_size: value.max_row_group_size as usize, - pruning_max_in_list_size: value.pruning_max_in_list_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 66019df619684..c222cd1cb8687 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6409,7 +6409,7 @@ impl serde::Serialize for ParquetOptions { if self.max_row_group_size != 0 { len += 1; } - if self.pruning_max_in_list_size != 0 { + if self.max_in_list_size != 0 { len += 1; } if !self.created_by.is_empty() { @@ -6532,10 +6532,10 @@ 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.pruning_max_in_list_size != 0 { + if self.max_in_list_size != 0 { #[allow(clippy::needless_borrow)] #[allow(clippy::needless_borrows_for_generic_args)] - struct_ser.serialize_field("pruningMaxInListSize", ToString::to_string(&self.pruning_max_in_list_size).as_str())?; + 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)?; @@ -6695,8 +6695,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "dataPageRowCountLimit", "max_row_group_size", "maxRowGroupSize", - "pruning_max_in_list_size", - "pruningMaxInListSize", + "max_in_list_size", + "maxInListSize", "created_by", "createdBy", "content_defined_chunking", @@ -6749,7 +6749,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { DictionaryPageSizeLimit, DataPageRowCountLimit, MaxRowGroupSize, - PruningMaxInListSize, + MaxInListSize, CreatedBy, ContentDefinedChunking, MetadataSizeHint, @@ -6806,7 +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), - "pruningMaxInListSize" | "pruning_max_in_list_size" => Ok(GeneratedField::PruningMaxInListSize), + "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), @@ -6861,7 +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 pruning_max_in_list_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; @@ -7013,11 +7013,11 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0) ; } - GeneratedField::PruningMaxInListSize => { - if pruning_max_in_list_size__.is_some() { - return Err(serde::de::Error::duplicate_field("pruningMaxInListSize")); + GeneratedField::MaxInListSize => { + if max_in_list_size__.is_some() { + return Err(serde::de::Error::duplicate_field("maxInListSize")); } - pruning_max_in_list_size__ = + max_in_list_size__ = Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0) ; } @@ -7134,7 +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(), - pruning_max_in_list_size: pruning_max_in_list_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 220cb0db0cdbd..bdbe38538e1d7 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -863,7 +863,7 @@ pub struct ParquetOptions { #[prost(uint64, tag = "15")] pub max_row_group_size: u64, #[prost(uint64, tag = "38")] - pub pruning_max_in_list_size: u64, + 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 d61c89d2c2c22..f0c7ff945eaa4 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -920,7 +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, - pruning_max_in_list_size: value.pruning_max_in_list_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 220cb0db0cdbd..bdbe38538e1d7 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -863,7 +863,7 @@ pub struct ParquetOptions { #[prost(uint64, tag = "15")] pub max_row_group_size: u64, #[prost(uint64, tag = "38")] - pub pruning_max_in_list_size: u64, + 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 4733c8e814b6d..373b592837e55 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -424,7 +424,7 @@ mod parquet { parquet_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) }), max_row_group_size: global_options.global.max_row_group_size as u64, - pruning_max_in_list_size: global_options.global.pruning_max_in_list_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) @@ -571,7 +571,7 @@ mod parquet { }, ), max_row_group_size: proto.max_row_group_size as usize, - pruning_max_in_list_size: proto.pruning_max_in_list_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 1aea8e8b6df49..2b334d2847980 100644 --- a/datafusion/pruning/src/lib.rs +++ b/datafusion/pruning/src/lib.rs @@ -22,7 +22,6 @@ mod pruning_predicate; pub use file_pruner::FilePruner; pub use pruning_predicate::{ - MAX_LIST_VALUE_SIZE_REWRITE, PredicateRewriter, PruningPredicate, PruningStatistics, - RequiredColumns, UnhandledPredicateHook, build_pruning_predicate, - build_pruning_predicate_with_max_in_list_size, + 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 8b158098ebfe9..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,40 +390,107 @@ pub fn build_pruning_predicate( file_schema: &SchemaRef, predicate_creation_errors: &Count, ) -> Option> { - build_pruning_predicate_with_max_in_list_size( - predicate, - file_schema, - predicate_creation_errors, - MAX_LIST_VALUE_SIZE_REWRITE, - ) + PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(file_schema)) + .with_error_counter(predicate_creation_errors) + .build(predicate) } -/// Same as [`build_pruning_predicate`] but with an explicit cap on the size -/// of `IN (...)` lists rewritten into per-value statistics checks. Query -/// engines typically pass `datafusion.execution.pruning_max_in_list_size` -/// here. -pub fn build_pruning_predicate_with_max_in_list_size( - predicate: Arc, - file_schema: &SchemaRef, - predicate_creation_errors: &Count, +/// 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, -) -> Option> { - match PruningPredicate::try_new_with_max_in_list_size( - predicate, - Arc::clone(file_schema), - max_in_list_size, - ) { - Ok(pruning_predicate) => { - if !pruning_predicate.always_true() { - return Some(Arc::new(pruning_predicate)); - } +} + +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 @@ -484,14 +553,14 @@ impl PruningPredicate { /// 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(expr: Arc, schema: SchemaRef) -> Result { - Self::try_new_with_max_in_list_size(expr, schema, MAX_LIST_VALUE_SIZE_REWRITE) + Self::try_new_inner(expr, schema, MAX_IN_LIST_SIZE) } - /// Same as [`PruningPredicate::try_new`] but with an explicit cap on the - /// size of `IN (...)` lists rewritten into per-value statistics checks. - /// Query engines typically pass - /// `datafusion.execution.pruning_max_in_list_size` here. - pub fn try_new_with_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, @@ -1400,8 +1469,8 @@ fn build_is_null_column_expr( /// `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.pruning_max_in_list_size` config option. -pub const MAX_LIST_VALUE_SIZE_REWRITE: usize = 20; +/// `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`]. @@ -1414,7 +1483,7 @@ impl Default for PredicateRewriter { fn default() -> Self { Self { unhandled_hook: Arc::new(ConstantUnhandledPredicateHook::default()), - max_in_list_size: MAX_LIST_VALUE_SIZE_REWRITE, + max_in_list_size: MAX_IN_LIST_SIZE, } } } @@ -1439,9 +1508,9 @@ impl PredicateRewriter { /// to the unhandled-predicate hook (typically "keep the container"), /// effectively skipping container-level pruning for large IN lists. /// - /// The default (see [`MAX_LIST_VALUE_SIZE_REWRITE`]) preserves the + /// The default (see [`MAX_IN_LIST_SIZE`]) preserves the /// historical behaviour. Callers wiring config through can override via - /// `datafusion.execution.pruning_max_in_list_size`. + /// `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 @@ -1456,7 +1525,7 @@ impl PredicateRewriter { /// Returns the pruning predicate as an [`PhysicalExpr`] /// /// Notice: `IN (...)` lists longer than `max_in_list_size` (default - /// [`MAX_LIST_VALUE_SIZE_REWRITE`]) fall back to calling `unhandled_hook`. + /// [`MAX_IN_LIST_SIZE`]) fall back to calling `unhandled_hook`. pub fn rewrite_predicate_to_statistics_predicate( &self, expr: &Arc, @@ -3383,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); @@ -3447,6 +3516,47 @@ mod tests { 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)]); @@ -5881,7 +5991,7 @@ mod tests { &Arc::new(schema.clone()), required_columns, &unhandled_hook, - MAX_LIST_VALUE_SIZE_REWRITE, + MAX_IN_LIST_SIZE, ) } diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index fc8eaa8da00e6..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 @@ -260,7 +261,6 @@ 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.pruning_max_in_list_size 20 datafusion.execution.parquet.pushdown_filters false datafusion.execution.parquet.reorder_filters false datafusion.execution.parquet.schema_force_view_types true @@ -413,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. @@ -420,7 +421,6 @@ 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.pruning_max_in_list_size 20 Maximum number of values in an `IN (...)` list for which the pruning predicate will rewrite the list into a chain of per-value statistics checks. Lists longer than this fall back to the unhandled-predicate hook (defaulting to "keep the container"), which effectively skips container-level pruning for large IN lists. Higher values keep row-group / file-range statistics pruning effective for larger IN lists (for example, REST endpoints that filter by a batch of ~25-100 identifiers), at the cost of a larger rewritten predicate expression evaluated for every container. Set to 0 to disable the rewrite path entirely. The default of 20 preserves the previous hardcoded behaviour. 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`. @@ -774,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 04ebf0b8fbb1b..860884e11fbf1 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -93,7 +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.pruning_max_in_list_size | 20 | Maximum number of values in an `IN (...)` list for which the pruning predicate will rewrite the list into a chain of per-value statistics checks. Lists longer than this fall back to the unhandled-predicate hook (defaulting to "keep the container"), which effectively skips container-level pruning for large IN lists. Higher values keep row-group / file-range statistics pruning effective for larger IN lists (for example, REST endpoints that filter by a batch of ~25-100 identifiers), at the cost of a larger rewritten predicate expression evaluated for every container. Set to 0 to disable the rewrite path entirely. The default of 20 preserves the previous hardcoded behaviour. | +| 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" |