Dont call multiunzip when no stats (apache#9220)

* Dont call multiunzip when no stats

* Update docstring

datafusion/core/benches/sql_planner.rs

@@ -234,7 +234,7 @@ fn criterion_benchmark(c: &mut Criterion) {
         let sql = std::fs::read_to_string(format!("../../benchmarks/queries/{}.sql", q))
             .unwrap();
         c.bench_function(&format!("physical_plan_tpch_{}", q), |b| {
-            b.iter(|| logical_plan(&ctx, &sql))
+            b.iter(|| physical_plan(&ctx, &sql))
         });
     }
 

datafusion/core/src/datasource/listing/table.rs

@@ -880,8 +880,13 @@ impl ListingTable {
             .boxed()
             .buffered(ctx.config_options().execution.meta_fetch_concurrency);
 
-        let (files, statistics) =
-            get_statistics_with_limit(files, self.schema(), limit).await?;
+        let (files, statistics) = get_statistics_with_limit(
+            files,
+            self.schema(),
+            limit,
+            self.options.collect_stat,
+        )
+        .await?;
 
         Ok((
             split_files(files, self.options.target_partitions),

datafusion/core/src/datasource/statistics.rs

@@ -29,12 +29,15 @@ use itertools::izip;
 use itertools::multiunzip;
 
 /// Get all files as well as the file level summary statistics (no statistic for partition columns).
-/// If the optional `limit` is provided, includes only sufficient files.
-/// Needed to read up to `limit` number of rows.
+/// If the optional `limit` is provided, includes only sufficient files. Needed to read up to
+/// `limit` number of rows. `collect_stats` is passed down from the configuration parameter on
+/// `ListingTable`. If it is false we only construct bare statistics and skip a potentially expensive
+///  call to `multiunzip` for constructing file level summary statistics.
 pub async fn get_statistics_with_limit(
     all_files: impl Stream<Item = Result<(PartitionedFile, Statistics)>>,
     file_schema: SchemaRef,
     limit: Option<usize>,
+    collect_stats: bool,
 ) -> Result<(Vec<PartitionedFile>, Statistics)> {
     let mut result_files = vec![];
     // These statistics can be calculated as long as at least one file provides
@@ -78,6 +81,9 @@ pub async fn get_statistics_with_limit(
             while let Some(current) = all_files.next().await {
                 let (file, file_stats) = current?;
                 result_files.push(file);
+                if !collect_stats {
+                    continue;
+                }
 
                 // We accumulate the number of rows, total byte size and null
                 // counts across all the files in question. If any file does not