feat: exact reverse scan for parquet sort pushdown

datafusion/common/src/config.rs

@@ -1367,6 +1367,16 @@ config_namespace! {
         /// pattern of selected rows.
         pub force_filter_selections: bool, default = false
 
+        /// (reading) If true, a scan whose requested ordering is the reverse of
+        /// the files' declared ordering reads each row group into memory and
+        /// emits its rows in reverse, so the scan is exactly ordered and the
+        /// `SortExec` above it can be removed. Peak memory per scanned file is
+        /// one decoded row group (after projection and filtering). If false,
+        /// only the row group order is reversed and the sort is kept. The value
+        /// is captured when the table is created, so a later `SET` does not
+        /// affect existing tables.
+        pub enable_exact_reverse_scan: bool, default = false
+
         /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
         /// and `Binary/BinaryLarge` with `BinaryView`.
         ///

datafusion/common/src/file_options/parquet_writer.rs

@@ -226,6 +226,7 @@ impl ParquetOptions {
             pushdown_filters: _,
             reorder_filters: _,
             force_filter_selections: _, // not used for writer props
+            enable_exact_reverse_scan: _, // not used for writer props
             allow_single_file_parallelism: _,
             maximum_parallel_row_group_writers: _,
             maximum_buffered_record_batches_per_stream: _,
@@ -391,6 +392,7 @@ mod tests {
             pushdown_filters: defaults.pushdown_filters,
             reorder_filters: defaults.reorder_filters,
             force_filter_selections: defaults.force_filter_selections,
+            enable_exact_reverse_scan: defaults.enable_exact_reverse_scan,
             allow_single_file_parallelism: defaults.allow_single_file_parallelism,
             maximum_parallel_row_group_writers: defaults
                 .maximum_parallel_row_group_writers,
@@ -522,6 +524,8 @@ mod tests {
                 pushdown_filters: global_options_defaults.pushdown_filters,
                 reorder_filters: global_options_defaults.reorder_filters,
                 force_filter_selections: global_options_defaults.force_filter_selections,
+                enable_exact_reverse_scan: global_options_defaults
+                    .enable_exact_reverse_scan,
                 allow_single_file_parallelism: global_options_defaults
                     .allow_single_file_parallelism,
                 maximum_parallel_row_group_writers: global_options_defaults

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -440,8 +440,10 @@ fn test_filter_collapse_outer_fetch_preserved() {
 }
 
 #[test]
-fn test_filter_collapse_inner_fetch_preserved() {
-    // When the inner filter has fetch and outer does not, the merged filter should preserve fetch
+fn test_filter_not_collapsed_into_inner_fetch() {
+    // The inner filter keeps its first 5 matching rows before the outer filter
+    // applies, so merging the outer predicate into it would change which rows
+    // are kept. The filters stay separate.
     let scan = TestScanBuilder::new(schema()).with_support(false).build();
     let predicate1 = col_lit_predicate("a", "foo", &schema());
     let filter1 = Arc::new(
@@ -463,16 +465,17 @@ fn test_filter_collapse_inner_fetch_preserved() {
         -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
       output:
         Ok:
-          - FilterExec: b@1 = bar AND a@0 = foo, fetch=5
-          -   DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+          - FilterExec: b@1 = bar
+          -   FilterExec: a@0 = foo, fetch=5
+          -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
     "
     );
 }
 
 #[test]
-fn test_filter_collapse_both_fetch_uses_minimum() {
-    // When both filters have fetch, the merged filter should use the smaller (tighter) fetch.
-    // Inner fetch=5 is tighter than outer fetch=10, so the result should be fetch=5.
+fn test_filter_with_fetch_not_collapsed_into_inner_fetch() {
+    // Same as above with a fetch on the outer filter too: the inner filter's
+    // fetch still picks its rows first, so the filters stay separate.
     let scan = TestScanBuilder::new(schema()).with_support(false).build();
     let predicate1 = col_lit_predicate("a", "foo", &schema());
     let filter1 = Arc::new(
@@ -499,8 +502,9 @@ fn test_filter_collapse_both_fetch_uses_minimum() {
         -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
       output:
         Ok:
-          - FilterExec: b@1 = bar AND a@0 = foo, fetch=5
-          -   DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+          - FilterExec: b@1 = bar, fetch=10
+          -   FilterExec: a@0 = foo, fetch=5
+          -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
     "
     );
 }

datafusion/datasource-parquet/src/access_plan.rs

@@ -879,6 +879,50 @@ impl PreparedAccessPlan {
         self
     }
 
+    /// For a scan that emits each row group's rows in reverse, select only
+    /// the rows a `limit` can reach, so the rows ahead of them are never
+    /// decoded.
+    ///
+    /// Row groups are walked in scan order. One whose output row count is
+    /// known up front (every row passes: `all_rows_match`, or statistics
+    /// proved it `fully_matched`) and that has no selection of its own
+    /// contributes its last `min(remaining, num_rows)` rows. The walk stops
+    /// at the first row group whose output is not known, since it may
+    /// produce fewer rows than it holds, and once `limit` rows are covered:
+    /// the scan stops reading before it reaches the row groups after that.
+    pub(crate) fn select_tail_for_limit(
+        mut self,
+        limit: usize,
+        all_rows_match: bool,
+        row_group_meta_data: &[RowGroupMetaData],
+    ) -> Self {
+        let mut remaining = limit;
+        for row_group in &mut self.row_groups {
+            if remaining == 0
+                || !(all_rows_match || row_group.fully_matched)
+                || row_group.selection.selection().is_some()
+            {
+                break;
+            }
+            let index = row_group.selection.row_group_index();
+            let num_rows = usize::try_from(row_group_meta_data[index].num_rows())
+                .unwrap_or(usize::MAX);
+            let take = remaining.min(num_rows);
+            if take < num_rows {
+                row_group.selection = RowGroupSelection::new(
+                    index,
+                    Some(RowSelection::from(vec![
+                        RowSelector::skip(num_rows - take),
+                        RowSelector::select(take),
+                    ])),
+                );
+                self.has_row_selection = true;
+            }
+            remaining -= take;
+        }
+        self
+    }
+
     #[cfg(test)]
     pub(crate) fn row_group_indexes(&self) -> Vec<usize> {
         self.row_groups
@@ -1144,6 +1188,110 @@ mod test {
         }
     }
 
+    /// Row groups of 10, 20, 30, 40 rows read back to front (3, 2, 1, 0).
+    fn reversed_plan(plan: ParquetAccessPlan) -> PreparedAccessPlan {
+        plan.prepare(&ROW_GROUP_METADATA).unwrap().reverse()
+    }
+
+    fn selected_rows(plan: &PreparedAccessPlan) -> Vec<Option<Vec<RowSelector>>> {
+        plan.row_groups
+            .iter()
+            .map(|rg| {
+                rg.selection
+                    .selection()
+                    .map(|selection| selection.iter().copied().collect())
+            })
+            .collect()
+    }
+
+    /// Without a predicate every row passes, so each row group's output is
+    /// known: the first one read (40 rows) is cut to its last 25 rows, and
+    /// the walk stops once the limit is covered.
+    #[test]
+    fn select_tail_for_limit_selects_last_rows() {
+        let plan = reversed_plan(ParquetAccessPlan::new_all(4)).select_tail_for_limit(
+            25,
+            true,
+            &ROW_GROUP_METADATA,
+        );
+
+        assert_eq!(
+            selected_rows(&plan),
+            vec![
+                Some(vec![RowSelector::skip(15), RowSelector::select(25)]),
+                None,
+                None,
+                None,
+            ]
+        );
+        assert!(plan.has_row_selection);
+    }
+
+    /// A limit larger than a row group takes it whole and the rest from the
+    /// next one.
+    #[test]
+    fn select_tail_for_limit_spans_row_groups() {
+        let plan = reversed_plan(ParquetAccessPlan::new_all(4)).select_tail_for_limit(
+            50,
+            true,
+            &ROW_GROUP_METADATA,
+        );
+
+        assert_eq!(
+            selected_rows(&plan),
+            vec![
+                None,
+                Some(vec![RowSelector::skip(20), RowSelector::select(10)]),
+                None,
+                None,
+            ]
+        );
+    }
+
+    /// With a predicate, only row groups that statistics proved fully
+    /// matched have a known output; the walk stops at the first one that
+    /// was not.
+    #[test]
+    fn select_tail_for_limit_stops_at_unknown_output() {
+        let mut access_plan = ParquetAccessPlan::new_all(4);
+        access_plan.mark_fully_matched(3);
+        let plan = reversed_plan(access_plan).select_tail_for_limit(
+            50,
+            false,
+            &ROW_GROUP_METADATA,
+        );
+
+        // Row group 3 is taken whole (40 < 50), row group 2 is not fully
+        // matched, so neither it nor anything after it is cut.
+        assert_eq!(selected_rows(&plan), vec![None, None, None, None]);
+        assert!(!plan.has_row_selection);
+    }
+
+    /// A row group that already has a selection (e.g. from the page index)
+    /// has an unknown output too.
+    #[test]
+    fn select_tail_for_limit_stops_at_existing_selection() {
+        let page_selection =
+            RowSelection::from(vec![RowSelector::skip(10), RowSelector::select(30)]);
+        let mut access_plan = ParquetAccessPlan::new_all(4);
+        access_plan.scan_selection(3, page_selection.clone());
+        let plan = reversed_plan(access_plan).select_tail_for_limit(
+            5,
+            true,
+            &ROW_GROUP_METADATA,
+        );
+
+        assert_eq!(
+            selected_rows(&plan),
+            vec![
+                Some(page_selection.iter().copied().collect()),
+                None,
+                None,
+                None
+            ]
+        );
+    }
+
     #[test]
     fn test_new_from_overall_row_selection() {
         let row_selection = RowSelection::from(vec![

datafusion/datasource-parquet/src/file_format.rs

@@ -813,6 +813,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
             pushdown_filters: global_options.global.pushdown_filters,
             reorder_filters: global_options.global.reorder_filters,
             force_filter_selections: global_options.global.force_filter_selections,
+            enable_exact_reverse_scan: global_options.global.enable_exact_reverse_scan,
             data_pagesize_limit: global_options.global.data_pagesize_limit as u64,
             write_batch_size: global_options.global.write_batch_size as u64,
             writer_version: global_options.global.writer_version.to_string(),