feat: reorder files in shared work queue by statistics for TopK

Add FileSource::reorder_files() trait method and ParquetSource
implementation that sorts files by column statistics before placing
them in the shared work queue. For DESC queries, files with the
highest min value come first; for ASC, lowest max first.

This ensures the first file read by any partition is the globally
optimal one for TopK threshold convergence, complementing the
intra-file RG reorder.

Author: zhuqi-lucas

datafusion/datasource-parquet/src/source.rs

@@ -485,6 +485,24 @@ impl ParquetSource {
     pub(crate) fn reverse_row_groups(&self) -> bool {
         self.reverse_row_groups
     }
+
+    /// Extract the sort key from a file's statistics for reordering.
+    ///
+    /// For DESC sorts, returns the column's min_value (we want highest min first).
+    /// For ASC sorts, returns the column's max_value (we want lowest max first).
+    fn sort_key_for_file(
+        file: &datafusion_datasource::PartitionedFile,
+        col_idx: usize,
+        descending: bool,
+    ) -> Option<datafusion_common::ScalarValue> {
+        let stats = file.statistics.as_ref()?;
+        let col_stats = stats.column_statistics.get(col_idx)?;
+        if descending {
+            col_stats.min_value.get_value().cloned()
+        } else {
+            col_stats.max_value.get_value().cloned()
+        }
+    }
 }
 
 /// Parses datafusion.common.config.ParquetOptions.coerce_int96 String to a arrow_schema.datatype.TimeUnit
@@ -587,6 +605,67 @@ impl FileSource for ParquetSource {
         }))
     }
 
+    fn reorder_files(
+        &self,
+        mut files: Vec<datafusion_datasource::PartitionedFile>,
+    ) -> Vec<datafusion_datasource::PartitionedFile> {
+        let sort_order = match &self.sort_order_for_reorder {
+            Some(order) if !order.is_empty() => order,
+            _ => return files,
+        };
+
+        // We only handle single-column sort for now
+        let first_expr = sort_order.first();
+
+        // The sort expression must be a Column so we can look up statistics
+        let col: &datafusion_physical_expr::expressions::Column =
+            match first_expr.expr.downcast_ref() {
+                Some(col) => col,
+                None => return files,
+            };
+
+        let col_name = col.name();
+        let descending = self.reverse_row_groups;
+
+        // Find the column index in the table schema
+        let table_schema = self.table_schema.table_schema();
+        let col_idx = match table_schema.index_of(col_name) {
+            Ok(idx) => idx,
+            Err(_) => return files,
+        };
+
+        // Stable sort: files with usable stats first, ordered by the
+        // relevant bound; files without stats go to the end.
+        files.sort_by(|a, b| {
+            let key_a = Self::sort_key_for_file(a, col_idx, descending);
+            let key_b = Self::sort_key_for_file(b, col_idx, descending);
+            match (key_a, key_b) {
+                (Some(va), Some(vb)) => {
+                    if descending {
+                        // DESC: highest min first (reverse order)
+                        vb.partial_cmp(&va).unwrap_or(std::cmp::Ordering::Equal)
+                    } else {
+                        // ASC: lowest max first
+                        va.partial_cmp(&vb).unwrap_or(std::cmp::Ordering::Equal)
+                    }
+                }
+                // Files without stats go to the end
+                (Some(_), None) => std::cmp::Ordering::Less,
+                (None, Some(_)) => std::cmp::Ordering::Greater,
+                (None, None) => std::cmp::Ordering::Equal,
+            }
+        });
+
+        log::debug!(
+            "Reordered {} files by {} {} for TopK optimization",
+            files.len(),
+            col_name,
+            if descending { "DESC" } else { "ASC" }
+        );
+
+        files
+    }
+
     fn table_schema(&self) -> &TableSchema {
         &self.table_schema
     }

datafusion/datasource/src/file.rs

@@ -280,6 +280,19 @@ pub trait FileSource: Any + Send + Sync {
         Ok(SortOrderPushdownResult::Unsupported)
     }
 
+    /// Reorder files in the shared work queue to optimize query performance.
+    ///
+    /// For example, TopK queries benefit from reading files with the best
+    /// statistics first, so the dynamic filter threshold tightens quickly.
+    ///
+    /// The default implementation returns files unchanged (no reordering).
+    fn reorder_files(
+        &self,
+        files: Vec<crate::PartitionedFile>,
+    ) -> Vec<crate::PartitionedFile> {
+        files
+    }
+
     /// Try to push down a projection into this FileSource.
     ///
     /// `FileSource` implementations that support projection pushdown should

datafusion/datasource/src/file_stream/work_source.rs

@@ -85,8 +85,18 @@ impl SharedWorkSource {
     }
 
     /// Create a shared work source for the unopened files in `config`.
+    ///
+    /// Files are reordered by the file source (e.g. by statistics for TopK)
+    /// before being placed in the shared queue.
     pub(crate) fn from_config(config: &FileScanConfig) -> Self {
-        Self::new(config.file_groups.iter().flat_map(FileGroup::iter).cloned())
+        let files: Vec<_> = config
+            .file_groups
+            .iter()
+            .flat_map(FileGroup::iter)
+            .cloned()
+            .collect();
+        let files = config.file_source.reorder_files(files);
+        Self::new(files)
     }
 
     /// Pop the next file from the shared work queue.