fix: compose reorder and reverse as sequential steps instead of mutually exclusive

Previously reorder_by_statistics and reverse_row_groups were mutually
exclusive (else-if). This meant DESC queries on unsorted data could
only get one optimization. Now they compose: reorder always sorts RGs
by min ASC, then reverse flips for DESC. This ensures correct results
for both sorted and unsorted inputs without regression.

Also removes prepare_with_optimizer in favor of calling optimize()
directly on each optimizer, and simplifies reorder_by_statistics to
always use min ASC (direction handled by reverse).

Author: zhuqi-lucas

datafusion/datasource-parquet/src/access_plan.rs

@@ -353,20 +353,6 @@ impl ParquetAccessPlan {
         let row_selection = self.into_overall_row_selection(row_group_meta_data)?;
         PreparedAccessPlan::new(row_group_indexes, row_selection)
     }
-
-    /// Like [`prepare`](Self::prepare), but also applies an
-    /// `AccessPlanOptimizer` to reorder/reverse row groups after
-    /// preparing the plan.
-    pub(crate) fn prepare_with_optimizer(
-        self,
-        row_group_meta_data: &[RowGroupMetaData],
-        file_metadata: &ParquetMetaData,
-        arrow_schema: &Schema,
-        optimizer: &dyn crate::access_plan_optimizer::AccessPlanOptimizer,
-    ) -> Result<PreparedAccessPlan> {
-        let plan = self.prepare(row_group_meta_data)?;
-        optimizer.optimize(plan, file_metadata, arrow_schema)
-    }
 }
 
 /// Represents a prepared, fully resolved [`ParquetAccessPlan`]
@@ -436,8 +422,6 @@ impl PreparedAccessPlan {
             }
         };
 
-        let descending = first_sort_expr.options.descending;
-
         // Build statistics converter for this column
         let converter = match StatisticsConverter::try_new(
             column.name(),
@@ -451,40 +435,31 @@ impl PreparedAccessPlan {
             }
         };
 
-        // Get the relevant statistics for the selected row groups.
-        // For ASC sort: use min values — we want the RG with the smallest min
-        //   to come first (best candidate for "smallest values").
-        // For DESC sort: use max values — we want the RG with the largest max
-        //   to come first (best candidate for "largest values"). Using min for
-        //   DESC can pick a worse first RG when ranges overlap (e.g., RG0 50-60
-        //   vs RG1 40-100 — RG1 has larger values but smaller min).
+        // Always sort by min values in ASC order to align row groups with
+        // the file's declared output ordering. Direction (DESC) is handled
+        // separately by ReverseRowGroups which is applied AFTER reorder.
+        //
+        // This composable design avoids the problem where reorder(DESC)
+        // followed by reverse would double-flip the order, and ensures
+        // that for already-sorted data, reorder is a no-op and reverse
+        // gives the correct DESC order (including placing small tail RGs first).
         let rg_metadata: Vec<&RowGroupMetaData> = self
             .row_group_indexes
             .iter()
             .map(|&idx| file_metadata.row_group(idx))
             .collect();
 
-        let stat_values = if descending {
-            match converter.row_group_maxes(rg_metadata.iter().copied()) {
-                Ok(vals) => vals,
-                Err(e) => {
-                    debug!("Skipping RG reorder: cannot get max values: {e}");
-                    return Ok(self);
-                }
-            }
-        } else {
-            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 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);
             }
         };
 
-        // Sort indices by statistic values (min for ASC, max for DESC)
+        // Always sort ASC by min values — direction is handled by reverse
         let sort_options = arrow::compute::SortOptions {
-            descending,
+            descending: false,
             nulls_first: first_sort_expr.options.nulls_first,
         };
         let sorted_indices =
@@ -836,7 +811,11 @@ mod test {
     }
 
     #[test]
-    fn test_reorder_by_statistics_desc() {
+    fn test_reorder_by_statistics_desc_sorts_asc() {
+        // reorder_by_statistics always sorts by min ASC regardless of sort
+        // direction. DESC is handled separately by ReverseRowGroups which
+        // is applied after reorder in the optimizer pipeline.
+        //
         // RGs: [50-99, 200-299, 1-30]
         let metadata = make_metadata_with_stats(&[(50, 99), (200, 299), (1, 30)]);
         let schema = make_arrow_schema();
@@ -847,8 +826,9 @@ mod test {
             .reorder_by_statistics(&sort_order, &metadata, &schema)
             .unwrap();
 
-        // DESC: largest max first: RG1(max=299), RG0(max=99), RG2(max=30)
-        assert_eq!(plan.row_group_indexes, vec![1, 0, 2]);
+        // Always ASC by min: RG2(min=1), RG0(min=50), RG1(min=200)
+        // Reverse is applied separately for DESC queries.
+        assert_eq!(plan.row_group_indexes, vec![2, 0, 1]);
     }
 
     #[test]
@@ -949,17 +929,14 @@ mod test {
     }
 
     #[test]
-    fn test_reorder_by_statistics_desc_uses_max_for_overlapping_rgs() {
-        // Overlapping ranges where min DESC would pick worse RG than max DESC:
-        //   RG0: 50-60 (small range, moderate max)
-        //   RG1: 40-100 (wide range, high max but lower min)
-        //   RG2: 20-30 (low max)
-        //
-        // For ORDER BY DESC LIMIT N:
-        //   Using min DESC: [RG0(50), RG1(40), RG2(20)] → reads RG0 first (max=60 only)
-        //   Using max DESC: [RG1(100), RG0(60), RG2(30)] → reads RG1 first (max=100)
+    fn test_reorder_by_statistics_overlapping_rgs_sorts_asc() {
+        // Overlapping ranges — reorder always uses min ASC:
+        //   RG0: 50-60
+        //   RG1: 40-100 (lower min, wider range)
+        //   RG2: 20-30 (lowest min)
         //
-        // RG1 is the better first choice for DESC because it contains the largest values.
+        // Sorted by min ASC: [RG2(20), RG1(40), RG0(50)]
+        // For DESC queries, ReverseRowGroups is applied after to flip order.
         let metadata = make_metadata_with_stats(&[(50, 60), (40, 100), (20, 30)]);
         let schema = make_arrow_schema();
         let sort_order = make_sort_order_desc();
@@ -969,8 +946,8 @@ mod test {
             .reorder_by_statistics(&sort_order, &metadata, &schema)
             .unwrap();
 
-        // Expected: RG1 (max=100) first, then RG0 (max=60), then RG2 (max=30)
-        assert_eq!(plan.row_group_indexes, vec![1, 0, 2]);
+        // Always ASC by min: RG2(min=20), RG1(min=40), RG0(min=50)
+        assert_eq!(plan.row_group_indexes, vec![2, 1, 0]);
     }
 
     #[test]

datafusion/datasource-parquet/src/opener.rs

@@ -1130,35 +1130,53 @@ impl RowGroupsPrunedParquetOpen {
             );
         }
 
-        // Build the access plan optimizer from sort pushdown hints.
-        // ReorderByStatistics is preferred (handles both ASC and DESC via
-        // min/max stats). ReverseRowGroups is a fallback when no statistics
-        // are available on the sort column.
-        let optimizer: Option<
+        // Row group ordering optimization (two composable steps):
+        //
+        // 1. reorder_by_statistics: sort RGs by min values (ASC) to align
+        //    with the file's declared output ordering. This fixes out-of-order
+        //    RGs (e.g., from append-heavy workloads) without changing direction.
+        //    Skipped gracefully when statistics are unavailable.
+        //
+        // 2. reverse: flip the order for DESC queries. Applied AFTER reorder
+        //    so the reversed order is correct whether or not reorder changed
+        //    anything. Also handles row_selection remapping.
+        //
+        // For sorted data: reorder is a no-op, reverse gives perfect DESC.
+        // For unsorted data: reorder fixes the order, reverse flips for DESC.
+        let reorder_optimizer: Option<
             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(),
-                ),
-            ))
-        } else if prepared.reverse_row_groups {
+        > = 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>
+        });
+
+        let reverse_optimizer: Option<
+            Box<dyn crate::access_plan_optimizer::AccessPlanOptimizer>,
+        > = if prepared.reverse_row_groups {
             Some(Box::new(crate::access_plan_optimizer::ReverseRowGroups))
         } else {
             None
         };
 
-        // Prepare the access plan and apply row group optimizer if configured.
-        let prepared_plan = if let Some(opt) = &optimizer {
-            access_plan.prepare_with_optimizer(
-                rg_metadata,
+        // Prepare the access plan and apply optimizers in order:
+        // 1. reorder (fix out-of-order RGs to match declared ordering)
+        // 2. reverse (flip for DESC queries)
+        let mut prepared_plan = access_plan.prepare(rg_metadata)?;
+        if let Some(opt) = &reorder_optimizer {
+            prepared_plan = opt.optimize(
+                prepared_plan,
                 file_metadata.as_ref(),
                 &prepared.physical_file_schema,
-                opt.as_ref(),
-            )?
-        } else {
-            access_plan.prepare(rg_metadata)?
-        };
+            )?;
+        }
+        if let Some(opt) = &reverse_optimizer {
+            prepared_plan = opt.optimize(
+                prepared_plan,
+                file_metadata.as_ref(),
+                &prepared.physical_file_schema,
+            )?;
+        }
 
         let arrow_reader_metrics = ArrowReaderMetrics::enabled();
         let read_plan = build_projection_read_plan(