Add exact reverse scan with per-RG buffering and tests

When exact_reverse is enabled via with_exact_reverse(true) on ParquetSource:
- try_reverse_output returns Exact (Sort operator removed, fetch pushdown enabled)
- ReversedRowGroupStream buffers batches per row group, reverses batch order,
  and reverses rows within each batch. Memory: O(largest_row_group).
- Default (exact_reverse=false) returns Inexact (backward compatible)

Row reversal is done in ReversedRowGroupStream (per-RG buffer), NOT in the
per-batch map closure. This ensures correct ordering across batch boundaries
within a row group.

Tests added:
- test_exact_reverse_scan_per_rg_buffer: multi-RG, small batch_size, verifies [6,5,4,3,2,1]
- test_inexact_reverse_scan_preserves_row_order: verifies [4,5,6,1,2,3]
- test_reversed_row_group_stream_standalone: unit test for ReversedRowGroupStream
- test_exact_reverse_returns_exact: option returns Exact
- test_default_returns_inexact: default returns Inexact

Based on the approach from apache#18817.

Author: zhuqi-lucas

datafusion/datasource-parquet/src/opener.rs

@@ -29,7 +29,7 @@ use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
 use datafusion_physical_expr::projection::ProjectionExprs;
 use datafusion_physical_expr::utils::reassign_expr_columns;
 use datafusion_physical_expr_adapter::replace_columns_with_literals;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
@@ -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
@@ -555,6 +558,18 @@ impl FileOpener for ParquetOpener {
                 prepared_plan = prepared_plan.reverse(file_metadata.as_ref())?;
             }
 
+            // Collect per-RG row counts for exact reverse buffering
+            let rg_row_counts: Vec<usize> = if reverse_rows {
+                let rg_metadata = file_metadata.row_groups();
+                prepared_plan
+                    .row_group_indexes
+                    .iter()
+                    .map(|&idx| rg_metadata[idx].num_rows() as usize)
+                    .collect()
+            } else {
+                vec![]
+            };
+
             // Apply the prepared plan to the builder
             builder = prepared_plan.apply_to_builder(builder);
 
@@ -601,6 +616,9 @@ impl FileOpener for ParquetOpener {
                         &predicate_cache_inner_records,
                         &predicate_cache_records,
                     );
+                    // Note: per-batch row reversal is handled by ReversedRowGroupStream
+                    // (wraps the stream below), NOT here. Reversing per-batch here would
+                    // double-reverse when combined with the RG-level buffer+reverse.
                     b = projector.project_batch(&b)?;
                     if replace_schema {
                         // Ensure the output batch has the expected schema.
@@ -677,6 +695,15 @@ impl FileOpener for ParquetOpener {
                 })
             });
 
+            // When exact reverse is enabled, wrap the stream to buffer
+            // and reverse rows per row group. Memory cost: O(largest_RG).
+            let stream: futures::stream::BoxStream<'static, Result<RecordBatch>> =
+                if reverse_rows {
+                    ReversedRowGroupStream::new(stream, rg_row_counts).boxed()
+                } else {
+                    stream.boxed()
+                };
+
             if let Some(file_pruner) = file_pruner {
                 Ok(EarlyStoppingStream::new(
                     stream,
@@ -691,6 +718,127 @@ impl FileOpener for ParquetOpener {
     }
 }
 
+/// Buffers batches per row group, then emits them in reversed order with
+/// reversed rows within each batch. Memory: O(largest row group).
+///
+/// The input stream has row groups already in reversed order (via
+/// `PreparedAccessPlan::reverse`). This stream reverses the row order
+/// *within* each row group so the final output is in exact descending order.
+struct ReversedRowGroupStream<S> {
+    inner: S,
+    /// Number of rows in each row group (in read order, already reversed)
+    rg_row_counts: Vec<usize>,
+    /// Index of the current row group being buffered
+    current_rg: usize,
+    /// Rows remaining in the current row group
+    rows_remaining_in_rg: usize,
+    /// Buffered batches for the current row group
+    buffer: Vec<RecordBatch>,
+    /// Reversed batches ready to emit
+    output_buffer: VecDeque<RecordBatch>,
+    /// Whether the inner stream is exhausted
+    done: bool,
+}
+
+impl<S> ReversedRowGroupStream<S> {
+    fn new(inner: S, rg_row_counts: Vec<usize>) -> Self {
+        let rows_remaining = rg_row_counts.first().copied().unwrap_or(0);
+        Self {
+            inner,
+            rg_row_counts,
+            current_rg: 0,
+            rows_remaining_in_rg: rows_remaining,
+            buffer: Vec::new(),
+            output_buffer: VecDeque::new(),
+            done: false,
+        }
+    }
+
+    /// Reverse the buffered batches: reverse batch order, reverse rows
+    /// within each batch, and move them to output_buffer.
+    fn flush_buffer(&mut self) -> Result<()> {
+        let batches = std::mem::take(&mut self.buffer);
+        for batch in batches.into_iter().rev() {
+            if batch.num_rows() <= 1 {
+                self.output_buffer.push_back(batch);
+                continue;
+            }
+            let indices = arrow::array::UInt32Array::from_iter_values(
+                (0..batch.num_rows() as u32).rev(),
+            );
+            let reversed = arrow::compute::take_record_batch(&batch, &indices)?;
+            self.output_buffer.push_back(reversed);
+        }
+        // Advance to next row group
+        self.current_rg += 1;
+        self.rows_remaining_in_rg = self
+            .rg_row_counts
+            .get(self.current_rg)
+            .copied()
+            .unwrap_or(0);
+        Ok(())
+    }
+}
+
+impl<S> Stream for ReversedRowGroupStream<S>
+where
+    S: Stream<Item = Result<RecordBatch>> + Unpin,
+{
+    type Item = Result<RecordBatch>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        use Poll;
+
+        // First, emit any already-reversed batches
+        if let Some(batch) = self.output_buffer.pop_front() {
+            return Poll::Ready(Some(Ok(batch)));
+        }
+
+        if self.done {
+            return Poll::Ready(None);
+        }
+
+        // Pull batches from the inner stream until we complete a row group
+        loop {
+            match ready!(self.inner.poll_next_unpin(cx)) {
+                Some(Ok(batch)) => {
+                    let num_rows = batch.num_rows();
+                    self.buffer.push(batch);
+                    self.rows_remaining_in_rg =
+                        self.rows_remaining_in_rg.saturating_sub(num_rows);
+
+                    if self.rows_remaining_in_rg == 0 {
+                        // Row group complete — flush buffer
+                        if let Err(e) = self.flush_buffer() {
+                            return Poll::Ready(Some(Err(e)));
+                        }
+                        if let Some(batch) = self.output_buffer.pop_front() {
+                            return Poll::Ready(Some(Ok(batch)));
+                        }
+                    }
+                }
+                Some(Err(e)) => return Poll::Ready(Some(Err(e))),
+                None => {
+                    self.done = true;
+                    // Flush any remaining buffered batches
+                    if !self.buffer.is_empty()
+                        && let Err(e) = self.flush_buffer()
+                    {
+                        return Poll::Ready(Some(Err(e)));
+                    }
+                    if let Some(batch) = self.output_buffer.pop_front() {
+                        return Poll::Ready(Some(Ok(batch)));
+                    }
+                    return Poll::Ready(None);
+                }
+            }
+        }
+    }
+}
+
 /// Copies metrics from ArrowReaderMetrics (the metrics collected by the
 /// arrow-rs parquet reader) to the parquet file metrics for DataFusion
 fn copy_arrow_reader_metrics(
@@ -1027,6 +1175,7 @@ fn should_enable_page_index(
 
 #[cfg(test)]
 mod test {
+    use std::pin::Pin;
     use std::sync::Arc;
 
     use super::{ConstantColumns, constant_columns_from_stats};
@@ -1076,6 +1225,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 +1251,7 @@ mod test {
                 coerce_int96: None,
                 max_predicate_cache_size: None,
                 reverse_row_groups: false,
+                reverse_rows: false,
             }
         }
 
@@ -1208,6 +1359,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,
             }
         }
     }
@@ -1296,7 +1448,7 @@ mod test {
     }
 
     async fn count_batches_and_rows(
-        mut stream: std::pin::Pin<
+        mut stream: Pin<
             Box<
                 dyn Stream<Item = Result<arrow::array::RecordBatch, DataFusionError>>
                     + Send,
@@ -1314,7 +1466,7 @@ mod test {
 
     /// Helper to collect all int32 values from the first column of batches
     async fn collect_int32_values(
-        mut stream: std::pin::Pin<
+        mut stream: Pin<
             Box<
                 dyn Stream<Item = Result<arrow::array::RecordBatch, DataFusionError>>
                     + Send,