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

datafusion/common/src/config.rs

@@ -1189,6 +1189,22 @@ 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 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
 

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()
@@ -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,
+            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,
@@ -608,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,

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

@@ -65,7 +65,9 @@ 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_with_max_in_list_size,
+};
 
 #[cfg(feature = "parquet_encryption")]
 use datafusion_common::config::EncryptionFactoryOptions;
@@ -289,6 +291,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.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 +458,7 @@ struct PreparedParquetOpen {
     expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
     predicate_creation_errors: Count,
     max_predicate_cache_size: Option<usize>,
+    pruning_max_in_list_size: usize,
     reverse_row_groups: bool,
     sort_order_for_reorder: Option<LexOrdering>,
     preserve_order: bool,
@@ -850,6 +858,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 +1061,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 +1478,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 +1643,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(
+    build_pruning_predicate_with_max_in_list_size(
         Arc::clone(predicate),
         file_schema,
         predicate_creation_errors,
+        max_in_list_size,
     )
 }
 
@@ -1720,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};
@@ -1752,6 +1766,7 @@ mod test {
         enable_row_group_stats_pruning: bool,
         coerce_int96: Option<TimeUnit>,
         max_predicate_cache_size: Option<usize>,
+        pruning_max_in_list_size: usize,
         reverse_row_groups: bool,
         preserve_order: bool,
     }
@@ -1860,6 +1875,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 +2053,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,

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<ParquetMetaData>,
         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]));

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()

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 {

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()

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__,

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")]

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)),

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")]