feat: enable file reorder and RG reorder for all TopK queries

Previously file reorder and RG reorder only worked with sort pushdown
(Inexact path, WITH ORDER). Now they extract sort info from
DynamicFilterPhysicalExpr.sort_options in the predicate, which is set
by SortExec for ALL TopK queries regardless of WITH ORDER.

This means ORDER BY col DESC LIMIT K on any parquet table benefits
from file reorder (best file first in shared queue), RG reorder
(best RG first within file), and stats init (threshold before I/O).

Author: zhuqi-lucas

datafusion/datasource-parquet/src/opener.rs

@@ -1165,17 +1165,41 @@ impl RowGroupsPrunedParquetOpen {
         //
         // For sorted data: reorder is a no-op, reverse gives perfect DESC.
         // For unsorted data: reorder fixes the order, reverse flips for DESC.
+        // Build reorder optimizer from sort_order_for_reorder (Inexact path)
+        // or from DynamicFilterPhysicalExpr sort_options (any TopK query).
         let reorder_optimizer: Option<
             Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>,
-        > = prepared.sort_order_for_reorder.as_ref().map(|sort_order| {
-            Box::new(crate::access_plan_optimizer::ReorderByStatistics::new(
-                sort_order.clone(),
-            )) as Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>
-        });
+        > = if let Some(sort_order) = &prepared.sort_order_for_reorder {
+            Some(
+                Box::new(crate::access_plan_optimizer::ReorderByStatistics::new(
+                    sort_order.clone(),
+                ))
+                    as Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>,
+            )
+        } else if let Some(predicate) = &prepared.predicate {
+            // For non-Inexact TopK queries: extract sort info from the
+            // DynamicFilterPhysicalExpr to enable RG reorder.
+            extract_sort_order_from_predicate(predicate).map(|sort_order| {
+                Box::new(crate::access_plan_optimizer::ReorderByStatistics::new(
+                    sort_order,
+                ))
+                    as Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>
+            })
+        } else {
+            None
+        };
 
+        // Reverse for DESC queries. Triggered by sort pushdown (reverse_row_groups)
+        // or by DynamicFilterPhysicalExpr sort_options indicating DESC.
+        let is_descending = prepared.reverse_row_groups
+            || prepared
+                .predicate
+                .as_ref()
+                .and_then(extract_descending_from_predicate)
+                .unwrap_or(false);
         let reverse_optimizer: Option<
             Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>,
-        > = if prepared.reverse_row_groups {
+        > = if is_descending {
             Some(Box::new(crate::access_plan_optimizer::ReverseRowGroups))
         } else {
             None
@@ -1480,6 +1504,46 @@ fn compute_best_threshold_from_stats(
     Ok(best)
 }
 
+/// Extract a [`LexOrdering`] from a [`DynamicFilterPhysicalExpr`] in the predicate.
+///
+/// Returns a single-column ASC sort order if a DynamicFilterPhysicalExpr with
+/// sort_options is found. The sort direction in the returned LexOrdering is
+/// always ASC because `ReorderByStatistics` always sorts by min ASC; the
+/// caller handles DESC via `ReverseRowGroups`.
+fn extract_sort_order_from_predicate(
+    predicate: &Arc<dyn PhysicalExpr>,
+) -> Option<LexOrdering> {
+    let df = find_dynamic_filter(predicate)?;
+    let sort_options = df.sort_options()?;
+    if sort_options.len() != 1 {
+        return None;
+    }
+    let children = df.children();
+    if children.is_empty() {
+        return None;
+    }
+    // Build ASC sort order (reorder always sorts ASC, reverse handles DESC)
+    let sort_expr = datafusion_physical_expr_common::sort_expr::PhysicalSortExpr {
+        expr: Arc::clone(children[0]),
+        options: arrow::compute::SortOptions {
+            descending: false,
+            nulls_first: sort_options[0].nulls_first,
+        },
+    };
+    LexOrdering::new(vec![sort_expr])
+}
+
+/// Check if a [`DynamicFilterPhysicalExpr`] in the predicate indicates DESC sort.
+fn extract_descending_from_predicate(predicate: &Arc<dyn PhysicalExpr>) -> Option<bool> {
+    let df = find_dynamic_filter(predicate)?;
+    let sort_options = df.sort_options()?;
+    if sort_options.len() == 1 {
+        Some(sort_options[0].descending)
+    } else {
+        None
+    }
+}
+
 /// State for a stream that decodes a single Parquet file using a push-based decoder.
 ///
 /// The [`transition`](Self::transition) method drives the decoder in a loop: it requests

datafusion/datasource-parquet/src/source.rs

@@ -486,6 +486,70 @@ impl ParquetSource {
         self.reverse_row_groups
     }
 
+    /// Extract TopK sort column name and direction from available sources.
+    ///
+    /// Tries two sources in order:
+    /// 1. DynamicFilterPhysicalExpr in the predicate (works for ALL TopK queries)
+    /// 2. sort_order_for_reorder (only set in Inexact sort pushdown path)
+    fn extract_topk_sort_info(&self) -> Option<(String, bool)> {
+        // Try 1: from predicate's DynamicFilterPhysicalExpr
+        if let Some(predicate) = &self.predicate
+            && let Some(info) = Self::sort_info_from_dynamic_filter(predicate)
+        {
+            return Some(info);
+        }
+
+        // Try 2: fallback to sort_order_for_reorder (Inexact path)
+        if let Some(sort_order) = &self.sort_order_for_reorder
+            && !sort_order.is_empty()
+        {
+            let first = sort_order.first();
+            if let Some(col) = first
+                .expr
+                .downcast_ref::<datafusion_physical_expr::expressions::Column>()
+            {
+                return Some((col.name().to_string(), self.reverse_row_groups));
+            }
+        }
+
+        None
+    }
+
+    /// Try to extract sort column name and direction from a DynamicFilterPhysicalExpr
+    /// in the predicate tree.
+    fn sort_info_from_dynamic_filter(
+        expr: &Arc<dyn PhysicalExpr>,
+    ) -> Option<(String, bool)> {
+        // Try downcast to DynamicFilterPhysicalExpr
+        let cloned = Arc::clone(expr);
+        let any_arc: Arc<dyn std::any::Any + Send + Sync> = cloned;
+        if let Ok(df) = Arc::downcast::<
+            datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr,
+        >(any_arc)
+        {
+            let sort_options = df.sort_options()?;
+            if sort_options.len() != 1 {
+                return None;
+            }
+            let children = df.children();
+            if children.is_empty() {
+                return None;
+            }
+            let col = children[0]
+                .downcast_ref::<datafusion_physical_expr::expressions::Column>()?;
+            return Some((col.name().to_string(), sort_options[0].descending));
+        }
+
+        // Recursively check children
+        for child in expr.children() {
+            if let Some(info) = Self::sort_info_from_dynamic_filter(child) {
+                return Some(info);
+            }
+        }
+
+        None
+    }
+
     /// 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).
@@ -609,27 +673,17 @@ impl FileSource for ParquetSource {
         &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,
+        // Extract sort column and direction from either:
+        // 1. The DynamicFilterPhysicalExpr in the predicate (works for ALL TopK)
+        // 2. Fallback to sort_order_for_reorder (Inexact path only)
+        let (col_name, descending) = match self.extract_topk_sort_info() {
+            Some(info) => info,
+            None => 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) {
+        let col_idx = match table_schema.index_of(&col_name) {
             Ok(idx) => idx,
             Err(_) => return files,
         };