feat: reorder row groups by statistics for TopK queries

When a parquet file has multiple row groups with out-of-order or
overlapping statistics, TopK queries benefit from reading "best" row
groups first so the dynamic filter threshold tightens quickly.

This PR adds:

1. `reorder_by_statistics`: sorts row groups by min values (ASC) based
   on parquet column statistics. Direction (DESC) is handled by the
   existing `reverse()` applied after reorder. The two steps compose:
   - Sorted data: reorder is a no-op, reverse gives perfect DESC order
   - Unsorted data: reorder fixes the order, reverse flips for DESC

2. `AccessPlanOptimizer` trait: extensible interface for row group
   access plan optimizations (reorder, reverse) applied after pruning.

3. `DynamicFilterPhysicalExpr.sort_options/fetch`: SortExec now passes
   sort direction and fetch limit to the dynamic filter, enabling the
   parquet reader to determine reorder direction for any TopK query.

4. `FileSource::reorder_files`: file-level reordering in the shared
   work queue so multi-file TopK reads the most promising files first.

5. Fix `SortExec::with_fetch` ordering: fetch must be set before
   `create_filter()` so the DynamicFilter gets the correct K value.

Author: zhuqi-lucas

datafusion/core/tests/fuzz_cases/topk_filter_pushdown.rs

@@ -317,10 +317,23 @@ async fn test_fuzz_topk_filter_pushdown() {
                     .map(|col| orders.get(**col).unwrap())
                     .multi_cartesian_product()
                 {
+                    // Add remaining columns as tiebreakers (ASC NULLS LAST)
+                    // to ensure deterministic results when RG reorder changes
+                    // the read order of rows with equal sort key values.
+                    let tiebreakers: Vec<String> = ["id", "name", "department"]
+                        .iter()
+                        .filter(|c| {
+                            !order_columns
+                                .iter()
+                                .take(num_order_by_columns)
+                                .any(|oc| **oc == **c)
+                        })
+                        .map(|c| format!("{c} ASC NULLS LAST"))
+                        .collect();
+                    let all_orderings =
+                        orderings.into_iter().chain(tiebreakers.iter()).join(", ");
                     let query = format!(
-                        "SELECT * FROM test_table ORDER BY {} LIMIT {}",
-                        orderings.into_iter().join(", "),
-                        limit
+                        "SELECT * FROM test_table ORDER BY {all_orderings} LIMIT {limit}",
                     );
                     queries.push(query);
                 }

datafusion/datasource-parquet/src/access_plan.rs

@@ -16,7 +16,12 @@
 // under the License.
 
 use crate::sort::reverse_row_selection;
+use arrow::datatypes::Schema;
 use datafusion_common::{Result, assert_eq_or_internal_err};
+use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
+use log::debug;
+use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
 use parquet::arrow::arrow_reader::{RowSelection, RowSelector};
 use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData};
 
@@ -377,6 +382,98 @@ impl PreparedAccessPlan {
         })
     }
 
+    /// Reorder row groups by their min statistics for the given sort order.
+    ///
+    /// This helps TopK queries find optimal values first. Row groups are
+    /// always sorted by min values in ASC order — direction (DESC) is
+    /// handled separately by `reverse()` which is applied after reorder.
+    ///
+    /// Gracefully skips reordering when:
+    /// - There is a row_selection (too complex to remap)
+    /// - 0 or 1 row groups (nothing to reorder)
+    /// - Sort expression is not a simple column reference
+    /// - Statistics are unavailable
+    pub(crate) fn reorder_by_statistics(
+        mut self,
+        sort_order: &LexOrdering,
+        file_metadata: &ParquetMetaData,
+        arrow_schema: &Schema,
+    ) -> Result<Self> {
+        // Skip if row_selection present (too complex to remap)
+        if self.row_selection.is_some() {
+            debug!("Skipping RG reorder: row_selection present");
+            return Ok(self);
+        }
+
+        // Nothing to reorder
+        if self.row_group_indexes.len() <= 1 {
+            return Ok(self);
+        }
+
+        let first_sort_expr = sort_order.first();
+
+        // Extract column name from sort expression
+        let column: &Column = match first_sort_expr.expr.downcast_ref::<Column>() {
+            Some(col) => col,
+            None => {
+                debug!("Skipping RG reorder: sort expr is not a simple column");
+                return Ok(self);
+            }
+        };
+
+        // Build statistics converter for this column
+        let converter = match StatisticsConverter::try_new(
+            column.name(),
+            arrow_schema,
+            file_metadata.file_metadata().schema_descr(),
+        ) {
+            Ok(c) => c,
+            Err(e) => {
+                debug!("Skipping RG reorder: cannot create stats converter: {e}");
+                return Ok(self);
+            }
+        };
+
+        // Always sort ASC by min values — direction is handled by reverse
+        let rg_metadata: Vec<&RowGroupMetaData> = self
+            .row_group_indexes
+            .iter()
+            .map(|&idx| file_metadata.row_group(idx))
+            .collect();
+
+        let stat_values = match converter.row_group_mins(rg_metadata.iter().copied()) {
+            Ok(vals) => vals,
+            Err(e) => {
+                debug!("Skipping RG reorder: cannot get min values: {e}");
+                return Ok(self);
+            }
+        };
+
+        let sort_options = arrow::compute::SortOptions {
+            descending: false,
+            nulls_first: first_sort_expr.options.nulls_first,
+        };
+        let sorted_indices =
+            match arrow::compute::sort_to_indices(&stat_values, Some(sort_options), None)
+            {
+                Ok(indices) => indices,
+                Err(e) => {
+                    debug!("Skipping RG reorder: sort failed: {e}");
+                    return Ok(self);
+                }
+            };
+
+        // Apply the reordering
+        let original_indexes = self.row_group_indexes.clone();
+        self.row_group_indexes = sorted_indices
+            .values()
+            .iter()
+            .map(|&i| original_indexes[i as usize])
+            .collect();
+
+        Ok(self)
+    }
+
     /// Reverse the access plan for reverse scanning
     pub(crate) fn reverse(mut self, file_metadata: &ParquetMetaData) -> Result<Self> {
         // Get the row group indexes before reversing

datafusion/datasource-parquet/src/access_plan_optimizer.rs

@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! [`AccessPlanOptimizer`] trait and implementations for optimizing
+//! row group access order during parquet scans.
+//!
+//! Applied after row group pruning but before building the decoder,
+//! these optimizers reorder (or reverse) the row groups to improve
+//! query performance — e.g., placing the "best" row groups first
+//! so TopK's dynamic filter threshold tightens quickly.
+
+use crate::access_plan::PreparedAccessPlan;
+use arrow::datatypes::Schema;
+use datafusion_common::Result;
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
+use parquet::file::metadata::ParquetMetaData;
+use std::fmt::Debug;
+
+/// Optimizes the row group access order for a prepared access plan.
+///
+/// Implementations can reorder, reverse, or otherwise transform the
+/// row group read order to improve scan performance. The optimizer
+/// is applied once per file, after all pruning passes are complete.
+///
+/// # Examples
+///
+/// - [`ReverseRowGroups`]: simple O(n) reversal for DESC on ASC-sorted data
+/// - [`ReorderByStatistics`]: sort row groups by min/max statistics
+///   so TopK queries find optimal values first
+pub(crate) trait AccessPlanOptimizer: Send + Sync + Debug {
+    /// Transform the prepared access plan.
+    ///
+    /// Implementations should return the plan unchanged if they cannot
+    /// apply their optimization (e.g., missing statistics).
+    fn optimize(
+        &self,
+        plan: PreparedAccessPlan,
+        file_metadata: &ParquetMetaData,
+        arrow_schema: &Schema,
+    ) -> Result<PreparedAccessPlan>;
+}
+
+/// Reverse the row group order — simple O(n) reversal.
+///
+/// Used as a fallback when the sort column has no statistics available.
+/// For ASC-sorted files with a DESC query, reversing row groups places
+/// the highest-value row groups first.
+#[derive(Debug)]
+pub(crate) struct ReverseRowGroups;
+
+impl AccessPlanOptimizer for ReverseRowGroups {
+    fn optimize(
+        &self,
+        plan: PreparedAccessPlan,
+        file_metadata: &ParquetMetaData,
+        _arrow_schema: &Schema,
+    ) -> Result<PreparedAccessPlan> {
+        plan.reverse(file_metadata)
+    }
+}
+
+/// Reorder row groups by min/max statistics of the sort column.
+///
+/// Row groups are always sorted by min values in ASC order. Direction
+/// (DESC) is handled separately by [`ReverseRowGroups`] applied after.
+///
+/// This is more effective than [`ReverseRowGroups`] alone when row groups
+/// are out of order (e.g., append-heavy workloads), because it uses
+/// actual statistics rather than assuming the original order is sorted.
+///
+/// Gracefully falls back to the original order when statistics are
+/// unavailable, the sort expression is not a simple column, etc.
+#[derive(Debug)]
+pub(crate) struct ReorderByStatistics {
+    sort_order: LexOrdering,
+}
+
+impl ReorderByStatistics {
+    pub(crate) fn new(sort_order: LexOrdering) -> Self {
+        Self { sort_order }
+    }
+}
+
+impl AccessPlanOptimizer for ReorderByStatistics {
+    fn optimize(
+        &self,
+        plan: PreparedAccessPlan,
+        file_metadata: &ParquetMetaData,
+        arrow_schema: &Schema,
+    ) -> Result<PreparedAccessPlan> {
+        plan.reorder_by_statistics(&self.sort_order, file_metadata, arrow_schema)
+    }
+}

datafusion/datasource-parquet/src/mod.rs

@@ -25,6 +25,7 @@
 #![cfg_attr(test, allow(clippy::needless_pass_by_value))]
 
 pub mod access_plan;
+mod access_plan_optimizer;
 pub mod file_format;
 pub mod metadata;
 mod metrics;