feat: Determine ordering of file groups

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

@@ -643,17 +643,20 @@ impl TableProvider for ListingTable {
         }
 
         let output_ordering = self.try_create_output_ordering()?;
-        if let Some(new_groups) = output_ordering.first().and_then(|output_ordering| {
+        match output_ordering.first().map(|output_ordering| {
             FileScanConfig::sort_file_groups(
                 &self.table_schema,
                 &projected_schema,
                 &partitioned_file_lists,
                 output_ordering,
             )
-            .ok()
         }) {
-            partitioned_file_lists = new_groups;
-        }
+            Some(Err(e)) => log::debug!("failed to sort file groups: {e}"),
+            Some(Ok(new_groups)) => {
+                partitioned_file_lists = new_groups;
+            }
+            None => {} // no ordering required
+        };
 
         // extract types of partition columns
         let table_partition_cols = self
@@ -838,10 +841,11 @@ impl ListingTable {
         // collect the statistics if required by the config
         let files = file_list
             .map(|part_file| async {
-                let part_file = part_file?;
+                let mut part_file = part_file?;
                 if self.options.collect_stat {
                     let statistics =
                         self.do_collect_statistics(ctx, &store, &part_file).await?;
+                    part_file.statistics = Some(statistics.clone());
                     Ok((part_file, statistics)) as Result<(PartitionedFile, Statistics)>
                 } else {
                     Ok((part_file, Statistics::new_unknown(&self.file_schema)))

datafusion/core/src/datasource/physical_plan/file_scan_config.rs

@@ -230,7 +230,8 @@ impl FileScanConfig {
             table_schema,
             projected_schema,
             flattened_files.iter().copied(),
-        )?;
+        )
+        .map_err(|e| e.context("construct min/max statistics"))?;
 
         let indices_sorted_by_min = {
             let mut sort: Vec<_> = statistics.min.iter().enumerate().collect();

datafusion/core/src/datasource/physical_plan/mod.rs

@@ -564,25 +564,42 @@ impl MinMaxStatistics {
             .fields()
             .iter()
             .map(|field| {
-                let (min, max) = get_min_max(table_schema.index_of(field.name())?);
+                let (min, max) =
+                    get_min_max(table_schema.index_of(field.name()).map_err(|e| {
+                        DataFusionError::ArrowError(
+                            e,
+                            Some(format!("get min/max for field: '{}'", field.name())),
+                        )
+                    })?);
                 Ok((
                     ScalarValue::iter_to_array(min)?,
                     ScalarValue::iter_to_array(max)?,
                 ))
             })
-            .collect::<Result<Vec<_>>>()?
+            .collect::<Result<Vec<_>>>()
+            .map_err(|e| e.context("collect min/max values"))?
             .into_iter()
             .unzip();
 
         Self::new(
             sort_order,
-            RecordBatch::try_new(Arc::clone(projected_schema), min_values)?,
-            RecordBatch::try_new(Arc::clone(projected_schema), max_values)?,
+            table_schema,
+            RecordBatch::try_new(Arc::clone(projected_schema), min_values).map_err(
+                |e| {
+                    DataFusionError::ArrowError(e, Some("\ncreate min batch".to_string()))
+                },
+            )?,
+            RecordBatch::try_new(Arc::clone(projected_schema), max_values).map_err(
+                |e| {
+                    DataFusionError::ArrowError(e, Some("\ncreate max batch".to_string()))
+                },
+            )?,
         )
     }
 
     fn new(
         sort_order: &[PhysicalSortExpr],
+        table_schema: &SchemaRef,
         min_values: RecordBatch,
         max_values: RecordBatch,
     ) -> Result<Self> {
@@ -592,10 +609,11 @@ impl MinMaxStatistics {
             .iter()
             .map(|expr| {
                 expr.expr
-                    .data_type(&min_values.schema())
+                    .data_type(table_schema)
                     .map(|data_type| SortField::new_with_options(data_type, expr.options))
             })
-            .collect::<Result<Vec<_>>>()?;
+            .collect::<Result<Vec<_>>>()
+            .map_err(|e| e.context("create sort fields"))?;
         let converter = RowConverter::new(sort_fields)?;
 
         let [min, max] = [min_values, max_values].map(|values| {
@@ -627,7 +645,8 @@ impl MinMaxStatistics {
                         options: Some(sort_expr.options),
                     })
                 })
-                .collect::<Result<Vec<_>>>()?;
+                .collect::<Result<Vec<_>>>()
+                .map_err(|e| e.context("create sorting columns"))?;
             converter
                 .convert_columns(
                     &sorting_columns
@@ -641,8 +660,8 @@ impl MinMaxStatistics {
         });
 
         Ok(Self {
-            min: min?,
-            max: max?,
+            min: min.map_err(|e| e.context("build min rows"))?,
+            max: max.map_err(|e| e.context("build max rows"))?,
         })
     }
 

datafusion/sqllogictest/test_files/parquet.slt

@@ -21,6 +21,10 @@
 statement ok
 set datafusion.execution.target_partitions = 2;
 
+# Collect statistics -- used for sorting files
+statement ok
+set datafusion.execution.collect_statistics = true;
+
 # Create a table as a data source
 statement ok
 CREATE TABLE src_table (
@@ -132,8 +136,8 @@ STORED AS PARQUET;
 ----
 3
 
-# Check output plan again, expect no "output_ordering" clause in the physical_plan -> ParquetExec,
-# due to there being more files than partitions:
+# Check output plan again, expect an "output_ordering" clause in the physical_plan -> ParquetExec:
+# After https://github.com/apache/arrow-datafusion/pull/9593 this should not require a sort.
 query TT
 EXPLAIN SELECT int_col, string_col
 FROM test_table
@@ -144,9 +148,7 @@ Sort: test_table.string_col ASC NULLS LAST, test_table.int_col ASC NULLS LAST
 --TableScan: test_table projection=[int_col, string_col]
 physical_plan
 SortPreservingMergeExec: [string_col@1 ASC NULLS LAST,int_col@0 ASC NULLS LAST]
---SortExec: expr=[string_col@1 ASC NULLS LAST,int_col@0 ASC NULLS LAST]
-----ParquetExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/2.parquet]]}, projection=[int_col, string_col]
-
+--ParquetExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/2.parquet]]}, projection=[int_col, string_col], output_ordering=[string_col@1 ASC NULLS LAST, int_col@0 ASC NULLS LAST]
 
 # Perform queries using MIN and MAX
 query I
@@ -169,6 +171,38 @@ SELECT min(date_col) FROM test_table;
 ----
 1970-01-02
 
+# Clean up
+statement ok
+DROP TABLE test_table;
+
+# Do one more test, but order by numeric columns:
+# This is to exercise file group sorting, which uses file-level statistics
+# DataFusion doesn't currently support string column statistics
+statement ok
+CREATE EXTERNAL TABLE test_table (
+  int_col INT NOT NULL,
+  string_col TEXT NOT NULL,
+  bigint_col BIGINT NOT NULL,
+  date_col DATE NOT NULL
+)
+STORED AS PARQUET
+WITH HEADER ROW
+WITH ORDER (int_col ASC NULLS LAST, bigint_col ASC NULLS LAST)
+LOCATION 'test_files/scratch/parquet/test_table';
+
+# Check output plan again, expect an "output_ordering" clause in the physical_plan -> ParquetExec:
+# After https://github.com/apache/arrow-datafusion/pull/9593 this should not require a sort.
+query TT
+EXPLAIN SELECT int_col, bigint_col
+FROM test_table
+ORDER BY int_col, bigint_col;
+----
+logical_plan
+Sort: test_table.int_col ASC NULLS LAST, test_table.bigint_col ASC NULLS LAST
+--TableScan: test_table projection=[int_col, bigint_col]
+physical_plan ParquetExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet/test_table/2.parquet]]}, projection=[int_col, bigint_col], output_ordering=[int_col@0 ASC NULLS LAST, bigint_col@1 ASC NULLS LAST]
+
+
 # Clean up
 statement ok
 DROP TABLE test_table;