feat(pruning): expose pruning predicate IN-list rewrite size cap as a config option

datafusion/common/src/config.rs

@@ -1189,6 +1189,17 @@ config_namespace! {
         /// parquet reader setting. 0 means no caching.
         pub max_predicate_cache_size: Option<usize>, default = None
 
+        /// Maximum number of values in an `IN (...)` list for which pruning will
+        /// occur. Longer lists will not be used to prune files, row groups, or
+        /// data pages.
+        ///
+        /// Higher values help in cases such as filtering on a list of
+        /// ~25-100 identifiers, but also make the predicate more expensive to
+        /// evaluate. Set to 0 to disable `IN (...)` list pruning entirely.
+        ///
+        /// Defaults to 20.
+        pub max_in_list_size: usize, default = 20
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -248,6 +248,7 @@ impl ParquetOptions {
             coerce_int96_tz: _,  // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
+            max_in_list_size: _,
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -489,6 +490,7 @@ mod tests {
             // not in WriterProperties, but itemizing here to not skip newly added props
             enable_page_index: defaults.enable_page_index,
             pruning: defaults.pruning,
+            max_in_list_size: defaults.max_in_list_size,
             skip_metadata: defaults.skip_metadata,
             metadata_size_hint: defaults.metadata_size_hint,
             pushdown_filters: defaults.pushdown_filters,
@@ -608,6 +610,7 @@ mod tests {
                 // not in WriterProperties
                 enable_page_index: global_options_defaults.enable_page_index,
                 pruning: global_options_defaults.pruning,
+                max_in_list_size: global_options_defaults.max_in_list_size,
                 skip_metadata: global_options_defaults.skip_metadata,
                 metadata_size_hint: global_options_defaults.metadata_size_hint,
                 pushdown_filters: global_options_defaults.pushdown_filters,

datafusion/datasource-parquet/src/opener/mod.rs

@@ -65,7 +65,7 @@ use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::metrics::{
     BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory,
 };
-use datafusion_pruning::{FilePruner, PruningPredicate, build_pruning_predicate};
+use datafusion_pruning::{FilePruner, PruningPredicate, PruningPredicateBuilder};
 
 #[cfg(feature = "parquet_encryption")]
 use datafusion_common::config::EncryptionFactoryOptions;
@@ -289,6 +289,11 @@ pub(super) struct ParquetMorselizer {
     /// Maximum size of the predicate cache, in bytes. If none, uses
     /// the arrow-rs default.
     pub max_predicate_cache_size: Option<usize>,
+    /// Maximum `IN (...)` list size that the pruning predicate will rewrite
+    /// into per-value statistics checks. Lists longer than this skip
+    /// container-level pruning. Sourced from
+    /// `datafusion.execution.parquet.max_in_list_size`.
+    pub max_in_list_size: usize,
     /// Whether to read row groups in reverse order
     pub reverse_row_groups: bool,
     /// Optional sort order used to reorder row groups by their min/max statistics.
@@ -451,6 +456,7 @@ struct PreparedParquetOpen {
     expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
     predicate_creation_errors: Count,
     max_predicate_cache_size: Option<usize>,
+    max_in_list_size: usize,
     reverse_row_groups: bool,
     sort_order_for_reorder: Option<LexOrdering>,
     preserve_order: bool,
@@ -850,6 +856,7 @@ impl ParquetMorselizer {
             expr_adapter_factory: Arc::clone(&self.expr_adapter_factory),
             predicate_creation_errors,
             max_predicate_cache_size: self.max_predicate_cache_size,
+            max_in_list_size: self.max_in_list_size,
             reverse_row_groups: self.reverse_row_groups,
             sort_order_for_reorder: self.sort_order_for_reorder.clone(),
             preserve_order: self.preserve_order,
@@ -1052,6 +1059,7 @@ impl MetadataLoadedParquetOpen {
             prepared.predicate.as_ref(),
             &physical_file_schema,
             &prepared.predicate_creation_errors,
+            prepared.max_in_list_size,
         );
 
         // Only build page pruning predicate if page index is enabled
@@ -1468,6 +1476,7 @@ impl RowGroupsPrunedParquetOpen {
                     Arc::clone(reader_metadata.metadata()),
                     prepared.predicate_creation_errors.clone(),
                     prepared.file_metrics.predicate_evaluation_errors.clone(),
+                    prepared.max_in_list_size,
                 ))
             }
             _ => None,
@@ -1632,13 +1641,14 @@ pub(crate) fn build_pruning_predicates(
     predicate: Option<&Arc<dyn PhysicalExpr>>,
     file_schema: &SchemaRef,
     predicate_creation_errors: &Count,
+    max_in_list_size: usize,
 ) -> Option<Arc<PruningPredicate>> {
     let predicate = predicate.as_ref()?;
-    build_pruning_predicate(
-        Arc::clone(predicate),
-        file_schema,
-        predicate_creation_errors,
-    )
+    PruningPredicateBuilder::new()
+        .with_file_schema(Arc::clone(file_schema))
+        .with_error_counter(predicate_creation_errors)
+        .with_max_in_list_size(max_in_list_size)
+        .build(Arc::clone(predicate))
 }
 
 /// Returns true if the page index must be loaded for page-level pruning.
@@ -1720,6 +1730,7 @@ mod test {
         DefaultPhysicalExprAdapterFactory, replace_columns_with_literals,
     };
     use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
+    use datafusion_pruning::MAX_IN_LIST_SIZE;
     use futures::StreamExt;
     use futures::stream::BoxStream;
     use object_store::{ObjectStore, ObjectStoreExt, memory::InMemory, path::Path};
@@ -1752,6 +1763,7 @@ mod test {
         enable_row_group_stats_pruning: bool,
         coerce_int96: Option<TimeUnit>,
         max_predicate_cache_size: Option<usize>,
+        max_in_list_size: usize,
         reverse_row_groups: bool,
         preserve_order: bool,
     }
@@ -1860,6 +1872,7 @@ mod test {
                 enable_row_group_stats_pruning: false,
                 coerce_int96: None,
                 max_predicate_cache_size: None,
+                max_in_list_size: MAX_IN_LIST_SIZE,
                 reverse_row_groups: false,
                 preserve_order: false,
             }
@@ -2037,6 +2050,7 @@ mod test {
                 #[cfg(feature = "parquet_encryption")]
                 encryption_factory: None,
                 max_predicate_cache_size: self.max_predicate_cache_size,
+                max_in_list_size: self.max_in_list_size,
                 reverse_row_groups: self.reverse_row_groups,
                 sort_order_for_reorder: None,
                 virtual_state,

datafusion/datasource-parquet/src/push_decoder.rs

@@ -57,7 +57,7 @@ use datafusion_common::{DataFusionError, Result};
 use datafusion_physical_expr::expressions::DynamicFilterTracking;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_plan::metrics::{BaselineMetrics, Count, Gauge};
-use datafusion_pruning::{PruningPredicate, build_pruning_predicate};
+use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder};
 
 use crate::access_plan::PreparedAccessPlan;
 use crate::decoder_projection::DecoderProjection;
@@ -142,6 +142,11 @@ pub(crate) struct RowGroupPruner {
     /// Metric for `PruningPredicate::prune` failures (evaluating an
     /// already-built predicate against row-group statistics).
     predicate_evaluation_errors: Count,
+    /// Cap on the `IN (...)` list size that the pruning predicate will
+    /// rewrite into per-value statistics checks. Longer lists skip
+    /// container-level pruning. Sourced from
+    /// `datafusion.execution.parquet.max_in_list_size`.
+    max_in_list_size: usize,
 }
 
 impl RowGroupPruner {
@@ -151,6 +156,7 @@ impl RowGroupPruner {
         parquet_metadata: Arc<ParquetMetaData>,
         predicate_creation_errors: Count,
         predicate_evaluation_errors: Count,
+        max_in_list_size: usize,
     ) -> Self {
         let tracking = DynamicFilterTracking::classify(&predicate);
         Self {
@@ -162,6 +168,7 @@ impl RowGroupPruner {
             pruning_predicate: None,
             predicate_creation_errors,
             predicate_evaluation_errors,
+            max_in_list_size,
         }
     }
 
@@ -186,11 +193,11 @@ impl RowGroupPruner {
             .watcher()
             .is_some_and(|tracker| tracker.changed());
         if self.needs_initial_build || dynamic_changed {
-            self.pruning_predicate = build_pruning_predicate(
-                Arc::clone(&self.predicate),
-                &self.arrow_schema,
-                &self.predicate_creation_errors,
-            );
+            self.pruning_predicate = PruningPredicateBuilder::new()
+                .with_file_schema(Arc::clone(&self.arrow_schema))
+                .with_error_counter(&self.predicate_creation_errors)
+                .with_max_in_list_size(self.max_in_list_size)
+                .build(Arc::clone(&self.predicate));
             self.needs_initial_build = false;
         }
 
@@ -436,6 +443,7 @@ mod tests {
         BinaryExpr, Column, DynamicFilterPhysicalExpr, lit,
     };
     use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder};
+    use datafusion_pruning::MAX_IN_LIST_SIZE;
     use parquet::arrow::ArrowWriter;
     use parquet::file::metadata::ParquetMetaDataPushDecoder;
     use parquet::file::properties::WriterProperties;
@@ -514,6 +522,7 @@ mod tests {
             Arc::clone(&meta),
             creation,
             evaluation,
+            MAX_IN_LIST_SIZE,
         );
 
         // RG0 (0..1000) is entirely below threshold → fully prunable.
@@ -545,6 +554,7 @@ mod tests {
             Arc::clone(&meta),
             creation,
             evaluation,
+            MAX_IN_LIST_SIZE,
         );
 
         // Initial threshold 500 → only the lower half of RG0 fails, so RG0
@@ -590,6 +600,7 @@ mod tests {
             Arc::clone(&meta),
             creation,
             evaluation,
+            MAX_IN_LIST_SIZE,
         );
         // No pruning predicate could be built → conservatively keep RGs.
         assert!(!pruner.should_prune(&[0]));

datafusion/datasource-parquet/src/source.rs

@@ -485,6 +485,14 @@ impl ParquetSource {
         self.table_parquet_options.global.max_predicate_cache_size
     }
 
+    /// Return the maximum size of an `IN (...)` list that the pruning
+    /// predicate will rewrite into per-value statistics checks. Lists
+    /// longer than this skip container-level pruning. Reads from
+    /// `datafusion.execution.parquet.max_in_list_size`.
+    pub fn max_in_list_size(&self) -> usize {
+        self.table_parquet_options.global.max_in_list_size
+    }
+
     #[cfg(feature = "parquet_encryption")]
     fn get_encryption_factory_with_config(
         &self,
@@ -647,6 +655,7 @@ impl FileSource for ParquetSource {
             #[cfg(feature = "parquet_encryption")]
             encryption_factory: self.get_encryption_factory_with_config(),
             max_predicate_cache_size: self.max_predicate_cache_size(),
+            max_in_list_size: self.max_in_list_size(),
             reverse_row_groups: self.reverse_row_groups,
             sort_order_for_reorder: self.sort_order_for_reorder.clone(),
             virtual_state,
@@ -781,6 +790,7 @@ impl FileSource for ParquetSource {
                         Some(predicate),
                         self.table_schema.table_schema(),
                         &predicate_creation_errors,
+                        self.max_in_list_size(),
                     ) {
                         let mut guarantees = pruning_predicate
                             .literal_guarantees()

datafusion/proto-common/proto/datafusion_common.proto

@@ -617,6 +617,8 @@ message ParquetOptions {
 
   uint64 max_row_group_size = 15;
 
+  uint64 max_in_list_size = 38;
+
   string created_by = 16;
 
   oneof coerce_int96_opt {

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

@@ -1081,6 +1081,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
                 })
                 .unwrap_or(None),
             max_row_group_size: value.max_row_group_size as usize,
+            max_in_list_size: value.max_in_list_size as usize,
             created_by: value.created_by.clone(),
             column_index_truncate_length: value
                 .column_index_truncate_length_opt.as_ref()

datafusion/proto-common/src/generated/pbjson.rs

@@ -6409,6 +6409,9 @@ impl serde::Serialize for ParquetOptions {
         if self.max_row_group_size != 0 {
             len += 1;
         }
+        if self.max_in_list_size != 0 {
+            len += 1;
+        }
         if !self.created_by.is_empty() {
             len += 1;
         }
@@ -6529,6 +6532,11 @@ impl serde::Serialize for ParquetOptions {
             #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("maxRowGroupSize", ToString::to_string(&self.max_row_group_size).as_str())?;
         }
+        if self.max_in_list_size != 0 {
+            #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
+            struct_ser.serialize_field("maxInListSize", ToString::to_string(&self.max_in_list_size).as_str())?;
+        }
         if !self.created_by.is_empty() {
             struct_ser.serialize_field("createdBy", &self.created_by)?;
         }
@@ -6687,6 +6695,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
             "dataPageRowCountLimit",
             "max_row_group_size",
             "maxRowGroupSize",
+            "max_in_list_size",
+            "maxInListSize",
             "created_by",
             "createdBy",
             "content_defined_chunking",
@@ -6739,6 +6749,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
             DictionaryPageSizeLimit,
             DataPageRowCountLimit,
             MaxRowGroupSize,
+            MaxInListSize,
             CreatedBy,
             ContentDefinedChunking,
             MetadataSizeHint,
@@ -6795,6 +6806,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
                             "dictionaryPageSizeLimit" | "dictionary_page_size_limit" => Ok(GeneratedField::DictionaryPageSizeLimit),
                             "dataPageRowCountLimit" | "data_page_row_count_limit" => Ok(GeneratedField::DataPageRowCountLimit),
                             "maxRowGroupSize" | "max_row_group_size" => Ok(GeneratedField::MaxRowGroupSize),
+                            "maxInListSize" | "max_in_list_size" => Ok(GeneratedField::MaxInListSize),
                             "createdBy" | "created_by" => Ok(GeneratedField::CreatedBy),
                             "contentDefinedChunking" | "content_defined_chunking" => Ok(GeneratedField::ContentDefinedChunking),
                             "metadataSizeHint" | "metadata_size_hint" => Ok(GeneratedField::MetadataSizeHint),
@@ -6849,6 +6861,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
                 let mut dictionary_page_size_limit__ = None;
                 let mut data_page_row_count_limit__ = None;
                 let mut max_row_group_size__ = None;
+                let mut max_in_list_size__ = None;
                 let mut created_by__ = None;
                 let mut content_defined_chunking__ = None;
                 let mut metadata_size_hint_opt__ = None;
@@ -7000,6 +7013,14 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
                                 Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0)
                             ;
                         }
+                        GeneratedField::MaxInListSize => {
+                            if max_in_list_size__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("maxInListSize"));
+                            }
+                            max_in_list_size__ = 
+                                Some(map_.next_value::<::pbjson::private::NumberDeserialize<_>>()?.0)
+                            ;
+                        }
                         GeneratedField::CreatedBy => {
                             if created_by__.is_some() {
                                 return Err(serde::de::Error::duplicate_field("createdBy"));
@@ -7113,6 +7134,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions {
                     dictionary_page_size_limit: dictionary_page_size_limit__.unwrap_or_default(),
                     data_page_row_count_limit: data_page_row_count_limit__.unwrap_or_default(),
                     max_row_group_size: max_row_group_size__.unwrap_or_default(),
+                    max_in_list_size: max_in_list_size__.unwrap_or_default(),
                     created_by: created_by__.unwrap_or_default(),
                     content_defined_chunking: content_defined_chunking__,
                     metadata_size_hint_opt: metadata_size_hint_opt__,

datafusion/proto-common/src/generated/prost.rs

@@ -862,6 +862,8 @@ pub struct ParquetOptions {
     pub data_page_row_count_limit: u64,
     #[prost(uint64, tag = "15")]
     pub max_row_group_size: u64,
+    #[prost(uint64, tag = "38")]
+    pub max_in_list_size: u64,
     #[prost(string, tag = "16")]
     pub created_by: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "35")]

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

@@ -912,6 +912,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions {
             dictionary_page_size_limit: value.dictionary_page_size_limit as u64,
             statistics_enabled_opt: value.statistics_enabled.clone().map(protobuf::parquet_options::StatisticsEnabledOpt::StatisticsEnabled),
             max_row_group_size: value.max_row_group_size as u64,
+            max_in_list_size: value.max_in_list_size as u64,
             created_by: value.created_by.clone(),
             column_index_truncate_length_opt: value.column_index_truncate_length.map(|v| protobuf::parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(v as u64)),
             statistics_truncate_length_opt: value.statistics_truncate_length.map(|v| protobuf::parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(v as u64)),

datafusion/proto-models/src/generated/datafusion_proto_common.rs

@@ -862,6 +862,8 @@ pub struct ParquetOptions {
     pub data_page_row_count_limit: u64,
     #[prost(uint64, tag = "15")]
     pub max_row_group_size: u64,
+    #[prost(uint64, tag = "38")]
+    pub max_in_list_size: u64,
     #[prost(string, tag = "16")]
     pub created_by: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "35")]