Remove redundant collect_stat and target_partitions on ListingOptions (#22969)

## Which issue does this PR close?

- Closes #.

## Rationale for this change

Something that was spotted during the review of:
- #22657

`ListingOptions::target_partitions` and `ListingOptions::collect_stat`
duplicate `SessionConfig`'s `execution.target_partitions` and
`execution.collect_statistics`.

After some investigation, I think they only live on `ListingOptions` for
historical reasons: when the struct was added (#1010 5 years ago),
`TableProvider::scan` had no access to the session, so the values had to
be copied onto the table at build time. Once #2660 passed `SessionState`
into `scan`, the fields became redundant (and had already drifted —
`scan` read them from the session config while `list_files_for_scan`
read the stale copy). This PR makes `SessionConfig` the single source of
truth.

## What changes are included in this PR?

- Remove `target_partitions`/`collect_stat` fields, their builders, and
`with_session_config_options` from `ListingOptions`.
- `ListingTable` now reads both values from the session config at scan
time.
- Reserve proto tags 8/9 in `ListingTableScanNode` and drop the related
(de)serialization.
- Update benchmarks, factory, and test call sites.

## Are these changes tested?

Yes, by existing tests

## Are there any user-facing changes?

Yes, breaking: the removed fields/builders require configuring
`SessionConfig` instead, and the two proto fields no longer round-trip.

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: gabotechs

benchmarks/src/bin/external_aggr.rs

@@ -318,9 +318,7 @@ impl ExternalAggrConfig {
         );
         let extension = DEFAULT_PARQUET_EXTENSION;
 
-        let options = ListingOptions::new(format)
-            .with_file_extension(extension)
-            .with_collect_stat(state.config().collect_statistics());
+        let options = ListingOptions::new(format).with_file_extension(extension);
 
         let table_path = ListingTableUrl::parse(path)?;
         let config = ListingTableConfig::new(table_path).with_listing_options(options);

benchmarks/src/imdb/run.rs

@@ -425,7 +425,6 @@ impl RunOpt {
         let table_format = self.file_format.as_str();
 
         // Obtain a snapshot of the SessionState
-        let state = ctx.state();
         let (format, path, extension): (Arc<dyn FileFormat>, String, &'static str) =
             match table_format {
                 // dbgen creates .tbl ('|' delimited) files without header
@@ -458,9 +457,7 @@ impl RunOpt {
                 }
             };
 
-        let options = ListingOptions::new(format)
-            .with_file_extension(extension)
-            .with_collect_stat(state.config().collect_statistics());
+        let options = ListingOptions::new(format).with_file_extension(extension);
 
         let table_path = ListingTableUrl::parse(path)?;
         let config = ListingTableConfig::new(table_path).with_listing_options(options);

benchmarks/src/sort_pushdown.rs

@@ -162,7 +162,8 @@ impl RunOpt {
         let config = self.common.config()?;
         let rt = self.common.build_runtime()?;
         let state = SessionStateBuilder::new()
-            .with_config(config)
+            // Always collect statistics for sort pushdown
+            .with_config(config.with_collect_statistics(true))
             .with_runtime_env(rt)
             .with_default_features()
             .build();
@@ -255,9 +256,7 @@ impl RunOpt {
         );
         let extension = DEFAULT_PARQUET_EXTENSION;
 
-        let options = ListingOptions::new(format)
-            .with_file_extension(extension)
-            .with_collect_stat(true); // Always collect statistics for sort pushdown
+        let options = ListingOptions::new(format).with_file_extension(extension);
 
         let table_path = ListingTableUrl::parse(path)?;
         let schema = options.infer_schema(&state, &table_path).await?;

benchmarks/src/sort_tpch.rs

@@ -333,9 +333,7 @@ impl RunOpt {
         );
         let extension = DEFAULT_PARQUET_EXTENSION;
 
-        let options = ListingOptions::new(format)
-            .with_file_extension(extension)
-            .with_collect_stat(state.config().collect_statistics());
+        let options = ListingOptions::new(format).with_file_extension(extension);
 
         let table_path = ListingTableUrl::parse(path)?;
         let schema = options.infer_schema(&state, &table_path).await?;

benchmarks/src/tpcds/run.rs

@@ -364,7 +364,6 @@ impl RunOpt {
         table: &str,
     ) -> Result<Arc<dyn TableProvider>> {
         let path = self.path.to_str().unwrap();
-        let target_partitions = self.partitions();
 
         // Obtain a snapshot of the SessionState
         let state = ctx.state();
@@ -380,9 +379,7 @@ impl RunOpt {
 
         let table_path = ListingTableUrl::parse(path)?;
         let options = ListingOptions::new(Arc::new(format))
-            .with_file_extension(DEFAULT_PARQUET_EXTENSION)
-            .with_target_partitions(target_partitions)
-            .with_collect_stat(state.config().collect_statistics());
+            .with_file_extension(DEFAULT_PARQUET_EXTENSION);
 
         let schema = options.infer_schema(&state, &table_path).await?;
         let constraints = table_constraints(table, schema.as_ref());

benchmarks/src/tpch/run.rs

@@ -283,7 +283,6 @@ impl RunOpt {
     ) -> Result<Arc<dyn TableProvider>> {
         let path = self.path.to_str().unwrap();
         let table_format = self.file_format.as_str();
-        let target_partitions = self.partitions();
 
         // Obtain a snapshot of the SessionState
         let state = ctx.state();
@@ -320,10 +319,7 @@ impl RunOpt {
             };
 
         let table_path = ListingTableUrl::parse(path)?;
-        let options = ListingOptions::new(format)
-            .with_file_extension(extension)
-            .with_target_partitions(target_partitions)
-            .with_collect_stat(state.config().collect_statistics());
+        let options = ListingOptions::new(format).with_file_extension(extension);
 
         let schema = match table_format {
             "parquet" => options.infer_schema(&state, &table_path).await?,

datafusion/catalog-listing/src/config.rs

@@ -152,8 +152,7 @@ impl ListingTableConfig {
     /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
     /// # let table_paths = ListingTableUrl::parse("file:///path/to/data").unwrap();
     /// let options = ListingOptions::new(Arc::new(ParquetFormat::default()))
-    ///     .with_file_extension(".parquet")
-    ///     .with_collect_stat(true);
+    ///     .with_file_extension(".parquet");
     ///
     /// let config = ListingTableConfig::new(table_paths).with_listing_options(options);
     /// // Configure file format and options

datafusion/catalog-listing/src/options.rs

@@ -20,7 +20,6 @@ use datafusion_catalog::Session;
 use datafusion_common::plan_err;
 use datafusion_datasource::ListingTableUrl;
 use datafusion_datasource::file_format::FileFormat;
-use datafusion_execution::config::SessionConfig;
 use datafusion_expr::SortExpr;
 use futures::StreamExt;
 use futures::TryStreamExt;
@@ -38,13 +37,6 @@ pub struct ListingOptions {
     /// The expected partition column names in the folder structure.
     /// See [Self::with_table_partition_cols] for details
     pub table_partition_cols: Vec<(String, DataType)>,
-    /// Set true to try to guess statistics from the files.
-    /// This can add a lot of overhead as it will usually require files
-    /// to be opened and at least partially parsed.
-    pub collect_stat: bool,
-    /// Group files to avoid that the number of partitions exceeds
-    /// this limit
-    pub target_partitions: usize,
     /// Optional pre-known sort order(s). Must be `SortExpr`s.
     ///
     /// DataFusion may take advantage of this ordering to omit sorts
@@ -68,30 +60,15 @@ impl ListingOptions {
     /// Default values:
     /// - use default file extension filter
     /// - no input partition to discover
-    /// - one target partition
-    /// - do not collect statistics
     pub fn new(format: Arc<dyn FileFormat>) -> Self {
         Self {
             file_extension: format.get_ext(),
             format,
             table_partition_cols: vec![],
-            collect_stat: false,
-            target_partitions: 1,
             file_sort_order: vec![],
         }
     }
 
-    /// Set options from [`SessionConfig`] and returns self.
-    ///
-    /// Currently this sets `target_partitions` and `collect_stat`
-    /// but if more options are added in the future that need to be coordinated
-    /// they will be synchronized through this method.
-    pub fn with_session_config_options(mut self, config: &SessionConfig) -> Self {
-        self = self.with_target_partitions(config.target_partitions());
-        self = self.with_collect_stat(config.collect_statistics());
-        self
-    }
-
     /// Set file extension on [`ListingOptions`] and returns self.
     ///
     /// # Example
@@ -205,40 +182,6 @@ impl ListingOptions {
         self
     }
 
-    /// Set stat collection on [`ListingOptions`] and returns self.
-    ///
-    /// ```
-    /// # use std::sync::Arc;
-    /// # use datafusion_catalog_listing::ListingOptions;
-    /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
-    ///
-    /// let listing_options =
-    ///     ListingOptions::new(Arc::new(ParquetFormat::default())).with_collect_stat(true);
-    ///
-    /// assert_eq!(listing_options.collect_stat, true);
-    /// ```
-    pub fn with_collect_stat(mut self, collect_stat: bool) -> Self {
-        self.collect_stat = collect_stat;
-        self
-    }
-
-    /// Set number of target partitions on [`ListingOptions`] and returns self.
-    ///
-    /// ```
-    /// # use std::sync::Arc;
-    /// # use datafusion_catalog_listing::ListingOptions;
-    /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
-    ///
-    /// let listing_options =
-    ///     ListingOptions::new(Arc::new(ParquetFormat::default())).with_target_partitions(8);
-    ///
-    /// assert_eq!(listing_options.target_partitions, 8);
-    /// ```
-    pub fn with_target_partitions(mut self, target_partitions: usize) -> Self {
-        self.target_partitions = target_partitions;
-        self
-    }
-
     /// Set file sort order on [`ListingOptions`] and returns self.
     ///
     /// ```

datafusion/catalog-listing/src/table.rs

@@ -533,15 +533,15 @@ impl TableProvider for ListingTable {
                         &self.table_schema,
                         &partitioned_file_lists,
                         output_ordering,
-                        self.options.target_partitions,
+                        state.config().target_partitions(),
                     )
                 })
             })
             .flatten()
         {
             Some(Err(e)) => log::debug!("failed to split file groups by statistics: {e}"),
             Some(Ok(new_groups)) => {
-                if new_groups.len() <= self.options.target_partitions {
+                if new_groups.len() <= state.config().target_partitions() {
                     partitioned_file_lists = new_groups;
                 } else {
                     log::debug!(
@@ -724,7 +724,7 @@ impl ListingTable {
         let files = file_list
             .map(|part_file| async {
                 let part_file = part_file?;
-                let (statistics, ordering) = if self.options.collect_stat {
+                let (statistics, ordering) = if ctx.config().collect_statistics() {
                     self.do_collect_statistics_and_ordering(ctx, &store, &part_file)
                         .await?
                 } else {
@@ -738,7 +738,7 @@ impl ListingTable {
             .buffer_unordered(ctx.config_options().execution.meta_fetch_concurrency);
 
         let (file_group, inexact_stats) =
-            get_files_with_limit(files, limit, self.options.collect_stat).await?;
+            get_files_with_limit(files, limit, ctx.config().collect_statistics()).await?;
 
         // Threshold: 0 = disabled, N > 0 = enabled when distinct_keys >= N
         //
@@ -747,32 +747,32 @@ impl ListingTable {
         // hash repartitioning for aggregates and joins on partition columns.
         let threshold = ctx.config_options().optimizer.preserve_file_partitions;
 
-        let (file_groups, grouped_by_partition) = if threshold > 0
-            && !self.options.table_partition_cols.is_empty()
-        {
-            let grouped =
-                file_group.group_by_partition_values(self.options.target_partitions);
-            if grouped.len() >= threshold {
-                (grouped, true)
+        let (file_groups, grouped_by_partition) =
+            if threshold > 0 && !self.options.table_partition_cols.is_empty() {
+                let grouped = file_group
+                    .group_by_partition_values(ctx.config().target_partitions());
+                if grouped.len() >= threshold {
+                    (grouped, true)
+                } else {
+                    let all_files: Vec<_> =
+                        grouped.into_iter().flat_map(|g| g.into_inner()).collect();
+                    (
+                        FileGroup::new(all_files)
+                            .split_files(ctx.config().target_partitions()),
+                        false,
+                    )
+                }
             } else {
-                let all_files: Vec<_> =
-                    grouped.into_iter().flat_map(|g| g.into_inner()).collect();
                 (
-                    FileGroup::new(all_files).split_files(self.options.target_partitions),
+                    file_group.split_files(ctx.config().target_partitions()),
                     false,
                 )
-            }
-        } else {
-            (
-                file_group.split_files(self.options.target_partitions),
-                false,
-            )
-        };
+            };
 
         let (file_groups, stats) = compute_all_files_statistics(
             file_groups,
             self.schema(),
-            self.options.collect_stat,
+            ctx.config().collect_statistics(),
             inexact_stats,
         )?;
 

datafusion/common/src/config.rs

@@ -677,8 +677,7 @@ config_namespace! {
         pub coalesce_batches: bool, default = true
 
         /// Should DataFusion collect statistics when first creating a table.
-        /// Has no effect after the table is created. Applies to the default
-        /// `ListingTableProvider` in DataFusion. Defaults to true.
+        /// Has no effect after the table is created. Defaults to true.
         pub collect_statistics: bool, default = true
 
         /// Number of partitions for query execution. Increasing partitions can increase