Add exact reverse scan: reverse rows within each batch for Exact sort pushdown

When try_reverse_output is called, enable both reverse_row_groups (reverse
RG read order) and reverse_rows (reverse rows within each batch). This gives
globally sorted output, allowing the Sort operator to be removed entirely
and fetch to be pushed down to the scan for file-level early termination.

Previously, only reverse_row_groups was set (Inexact), which kept TopK in
the plan and prevented fetch pushdown — all files had to be opened even
when only 1 row was needed.

Author: zhuqi-lucas

datafusion/core/tests/physical_optimizer/pushdown_sort.rs

@@ -84,8 +84,7 @@ fn test_sort_pushdown_basic_phase1() {
         -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -113,8 +112,7 @@ fn test_sort_with_limit_phase1() {
         -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: TopK(fetch=10), expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -144,8 +142,7 @@ fn test_sort_multiple_columns_phase1() {
         -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 ASC, b@1 DESC NULLS LAST], preserve_partitioning=[false]
-          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -179,8 +176,7 @@ fn test_prefix_match_single_column() {
         -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
-          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -213,8 +209,7 @@ fn test_prefix_match_with_limit() {
         -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC, b@1 DESC NULLS LAST, c@2 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: TopK(fetch=100), expr=[a@0 DESC NULLS LAST, b@1 ASC], preserve_partitioning=[false]
-          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC, b@1 DESC NULLS LAST, c@2 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -249,10 +244,9 @@ fn test_prefix_match_through_transparent_nodes() {
         -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC, c@2 DESC NULLS LAST], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
-          -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
-          -     CoalesceBatchesExec: target_batch_size=1024
-          -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+          -   CoalesceBatchesExec: target_batch_size=1024
+          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC, c@2 DESC NULLS LAST], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -344,9 +338,8 @@ fn test_sort_through_coalesce_batches() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   CoalesceBatchesExec: target_batch_size=1024
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - CoalesceBatchesExec: target_batch_size=1024
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -373,9 +366,8 @@ fn test_sort_through_repartition() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -406,8 +398,7 @@ fn test_nested_sorts() {
       output:
         Ok:
           - SortExec: expr=[b@1 ASC], preserve_partitioning=[false]
-          -   SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -469,8 +460,8 @@ fn test_sort_through_coalesce_partitions() {
         Ok:
           - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
           -   CoalescePartitionsExec
-          -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
-          -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+          -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -503,9 +494,9 @@ fn test_complex_plan_with_multiple_operators() {
         Ok:
           - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
           -   CoalescePartitionsExec
-          -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
+          -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
           -       CoalesceBatchesExec: target_batch_size=1024
-          -         DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          -         DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -538,8 +529,7 @@ fn test_multiple_sorts_different_columns() {
       output:
         Ok:
           - SortExec: expr=[c@2 ASC], preserve_partitioning=[false]
-          -   SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -667,8 +657,7 @@ fn test_pushdown_through_blocking_node() {
         Ok:
           - SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
           -   AggregateExec: mode=Final, gby=[a@0 as a], aggr=[COUNT(b)], ordering_mode=Sorted
-          -     SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -704,9 +693,8 @@ fn test_sort_pushdown_through_simple_projection() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[a@0 as a, b@1 as b]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[a@0 as a, b@1 as b]
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -739,9 +727,8 @@ fn test_sort_pushdown_through_projection_with_alias() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[id@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[a@0 as id, b@1 as value]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[a@0 as id, b@1 as value]
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -828,9 +815,8 @@ fn test_sort_pushdown_projection_reordered_columns() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@2 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[c@2 as c, b@1 as b, a@0 as a]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[c@2 as c, b@1 as b, a@0 as a]
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -862,9 +848,8 @@ fn test_sort_pushdown_projection_with_limit() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: TopK(fetch=10), expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[a@0 as a, b@1 as b]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[a@0 as a, b@1 as b]
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -899,10 +884,9 @@ fn test_sort_pushdown_through_projection_and_coalesce() {
         -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[a@0 as a, b@1 as b]
-          -     CoalesceBatchesExec: target_batch_size=1024
-          -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[a@0 as a, b@1 as b]
+          -   CoalesceBatchesExec: target_batch_size=1024
+          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }
@@ -935,9 +919,8 @@ fn test_sort_pushdown_projection_subset_of_columns() {
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC, b@1 ASC], file_type=parquet
       output:
         Ok:
-          - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-          -   ProjectionExec: expr=[a@0 as a]
-          -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, reverse_row_groups=true
+          - ProjectionExec: expr=[a@0 as a]
+          -   DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC, b@1 ASC], file_type=parquet, reverse_row_groups=true
     "
     );
 }

datafusion/datasource-parquet/src/opener.rs

@@ -118,6 +118,8 @@ pub(super) struct ParquetOpener {
     pub max_predicate_cache_size: Option<usize>,
     /// Whether to read row groups in reverse order
     pub reverse_row_groups: bool,
+    /// Whether to reverse rows within each batch (for Exact reverse scan)
+    pub reverse_rows: bool,
 }
 
 /// Represents a prepared access plan with optional row selection
@@ -274,6 +276,7 @@ impl FileOpener for ParquetOpener {
         let max_predicate_cache_size = self.max_predicate_cache_size;
 
         let reverse_row_groups = self.reverse_row_groups;
+        let reverse_rows = self.reverse_rows;
         Ok(Box::pin(async move {
             #[cfg(feature = "parquet_encryption")]
             let file_decryption_properties = encryption_context
@@ -601,6 +604,16 @@ impl FileOpener for ParquetOpener {
                         &predicate_cache_inner_records,
                         &predicate_cache_records,
                     );
+                    // When exact reverse scanning, reverse the rows within each
+                    // batch so that the output is in exact descending order.
+                    // Combined with reversed row group order, this gives globally
+                    // sorted output (Exact), enabling fetch pushdown to the scan.
+                    if reverse_rows && b.num_rows() > 1 {
+                        let indices = arrow::array::UInt32Array::from_iter_values(
+                            (0..b.num_rows() as u32).rev(),
+                        );
+                        b = arrow::compute::take_record_batch(&b, &indices)?;
+                    }
                     b = projector.project_batch(&b)?;
                     if replace_schema {
                         // Ensure the output batch has the expected schema.
@@ -1076,6 +1089,7 @@ mod test {
         coerce_int96: Option<arrow::datatypes::TimeUnit>,
         max_predicate_cache_size: Option<usize>,
         reverse_row_groups: bool,
+        reverse_rows: bool,
     }
 
     impl ParquetOpenerBuilder {
@@ -1101,6 +1115,7 @@ mod test {
                 coerce_int96: None,
                 max_predicate_cache_size: None,
                 reverse_row_groups: false,
+                reverse_rows: false,
             }
         }
 
@@ -1208,6 +1223,7 @@ mod test {
                 encryption_factory: None,
                 max_predicate_cache_size: self.max_predicate_cache_size,
                 reverse_row_groups: self.reverse_row_groups,
+                reverse_rows: self.reverse_rows,
             }
         }
     }

datafusion/datasource-parquet/src/source.rs

@@ -288,11 +288,14 @@ pub struct ParquetSource {
     pub(crate) projection: ProjectionExprs,
     #[cfg(feature = "parquet_encryption")]
     pub(crate) encryption_factory: Option<Arc<dyn EncryptionFactory>>,
-    /// If true, read files in reverse order and reverse row groups within files.
-    /// But it's not guaranteed that rows within row groups are in reverse order,
-    /// so we still need to sort them after reading, so the reverse scan is inexact.
-    /// Used to optimize ORDER BY ... DESC on sorted data.
+    /// If true, read row groups in reverse order within each file.
+    /// Combined with `reverse_rows`, controls the sort pushdown behavior:
+    /// - `reverse_row_groups=true, reverse_rows=false`: Inexact (RGs reversed, rows within RG not)
+    /// - `reverse_row_groups=true, reverse_rows=true`: Exact (both RGs and rows reversed)
     reverse_row_groups: bool,
+    /// If true, reverse the row order within each batch after reading.
+    /// This gives exact descending order when combined with `reverse_row_groups`.
+    reverse_rows: bool,
 }
 
 impl ParquetSource {
@@ -318,6 +321,7 @@ impl ParquetSource {
             #[cfg(feature = "parquet_encryption")]
             encryption_factory: None,
             reverse_row_groups: false,
+            reverse_rows: false,
         }
     }
 
@@ -567,6 +571,7 @@ impl FileSource for ParquetSource {
             encryption_factory: self.get_encryption_factory_with_config(),
             max_predicate_cache_size: self.max_predicate_cache_size(),
             reverse_row_groups: self.reverse_row_groups,
+            reverse_rows: self.reverse_rows,
         });
         Ok(opener)
     }
@@ -803,10 +808,13 @@ impl FileSource for ParquetSource {
             return Ok(SortOrderPushdownResult::Unsupported);
         }
 
-        // Return Inexact because we're only reversing row group order,
-        // not guaranteeing perfect row-level ordering
-        let new_source = self.clone().with_reverse_row_groups(true);
-        Ok(SortOrderPushdownResult::Inexact {
+        // Return Exact because the opener reverses both row group order
+        // AND rows within each batch, giving globally sorted output.
+        // This allows the Sort operator to be removed entirely and
+        // fetch to be pushed down to the scan for early termination.
+        let mut new_source = self.clone().with_reverse_row_groups(true);
+        new_source.reverse_rows = true;
+        Ok(SortOrderPushdownResult::Exact {
             inner: Arc::new(new_source) as Arc<dyn FileSource>,
         })