feat: sort file groups by statistics during sort pushdown (Sort pushdown phase 2) (#21182)

## Which issue does this PR close?

- Closes #17348
- Closes #19329

## Rationale for this change

When a partition (file group) contains multiple files in wrong order,
`validated_output_ordering()` strips the ordering and `EnforceSorting`
inserts an unnecessary `SortExec` — even though the files are
non-overlapping and internally sorted.

This PR fixes it by **sorting files within each group by min/max
statistics** during sort pushdown. After sorting, the file order matches
the sort key order, the ordering becomes valid, and `SortExec` can be
eliminated. This works for both single-partition and multi-partition
plans with multi-file groups.

## What changes are included in this PR?

### Core optimization

```text
Files in wrong order within a partition:     After statistics-based sorting:
  [a_high(400k+), b_mid(200k), c_low(1)]     [c_low(1), b_mid(200k), a_high(400k+)]
  → ordering stripped                          → ordering valid, non-overlapping
  → SortExec stays                             → SortExec eliminated
```

When `PushdownSort` finds a `SortExec` above a file-based
`DataSourceExec`:

1. **FileSource returns Exact** (natural ordering satisfies request):
   - Sort files within each group by statistics, verify non-overlapping
   - SortExec removed, fetch (LIMIT) pushed to DataSourceExec

2. **FileSource returns Unsupported** (ordering stripped due to wrong
file order):
   - Sort files within each group by statistics
- Re-check: if files are now non-overlapping and ordering is valid →
upgrade to Exact
   - SortExec eliminated + fetch pushed down

3. **FileSource returns Inexact** (reverse scan):
   - SortExec kept, scan optimized with reverse_row_groups

### Files Changed

| File | Change |
|------|--------|
| `datasource-parquet/src/source.rs` | ParquetSource returns `Exact`
when natural ordering satisfies request |
| `datasource/src/file_scan_config.rs` | Statistics-based file sorting,
non-overlapping re-check, Unsupported→Exact upgrade |
| `physical-optimizer/src/pushdown_sort.rs` | Preserve fetch (LIMIT)
when eliminating SortExec, module doc update |
| `core/tests/physical_optimizer/pushdown_sort.rs` | Updated prefix
match test |
| `sqllogictest/test_files/sort_pushdown.slt` | Updated existing tests +
5 new test groups (A-E) |

## Benchmark Results

Local release build, `--partitions 1`, 3 non-overlapping files with
reversed naming (6M rows):

| Query | Description | Main (ms) | PR (ms) | Speedup |
|-------|-------------|-----------|---------|---------|
| Q1 | `ORDER BY ASC` (full scan) | 259 | 122 | **2.1x faster** |
| Q2 | `ORDER BY ASC LIMIT 100` | 80 | 3 | **27x faster** |
| Q3 | `SELECT * ORDER BY ASC` | 700 | 313 | **2.2x faster** |
| Q4 | `SELECT * LIMIT 100` | 342 | 7 | **49x faster** |

LIMIT queries benefit most because sort elimination + limit pushdown
means only the first ~100 rows are read.

## Tests

- 13 new unit tests covering all sort pushdown paths
- 5 new SLT integration test groups (sort elimination, overlapping
files, LIMIT, multi-partition, inferred ordering)
- All existing tests pass with no regressions

## Test plan

- [x] `cargo test -p datafusion-datasource` — all tests pass
- [x] `cargo test -p datafusion-datasource-parquet` — all tests pass
- [x] `cargo test -p datafusion-physical-optimizer` — all tests pass
- [x] `cargo test -p datafusion --test core_integration` — all tests
pass
- [x] SLT sort/order/topk/window/union/joins tests pass (no regressions)
- [x] `cargo clippy` — 0 warnings

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: zhuqi-lucas

datafusion/core/tests/physical_optimizer/pushdown_sort.rs

@@ -255,8 +255,9 @@ fn test_prefix_match_through_transparent_nodes() {
 }
 
 #[test]
-fn test_no_prefix_match_wrong_direction() {
-    // Test that prefix matching does NOT work if the direction is wrong
+fn test_exact_prefix_match_same_direction() {
+    // Test that when the requested sort [a DESC] matches a prefix of the source's
+    // natural ordering [a DESC, b ASC], the Sort is eliminated (Exact pushdown).
     let schema = schema();
 
     // Source has [a DESC, b ASC] ordering
@@ -265,7 +266,7 @@ fn test_no_prefix_match_wrong_direction() {
     let source_ordering = LexOrdering::new(vec![a.clone().reverse(), b]).unwrap();
     let source = parquet_exec_with_sort(schema.clone(), vec![source_ordering]);
 
-    // Request [a DESC] - same direction as source, NOT a reverse prefix
+    // Request [a DESC] - same direction as source prefix, Sort should be eliminated
     let same_direction = LexOrdering::new(vec![a.clone().reverse()]).unwrap();
     let plan = sort_exec(same_direction, source);
 
@@ -278,8 +279,7 @@ fn test_no_prefix_match_wrong_direction() {
         -   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 DESC NULLS LAST], preserve_partitioning=[false]
-          -   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
+          - 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
     "
     );
 }

datafusion/datasource-parquet/src/source.rs

@@ -743,19 +743,17 @@ impl FileSource for ParquetSource {
     ///
     /// With both pieces of information, ParquetSource can decide what optimizations to apply.
     ///
-    /// # Phase 1 Behavior (Current)
-    /// Returns `Inexact` when reversing the row group scan order would help satisfy the
-    /// requested ordering. We still need a Sort operator at a higher level because:
-    /// - We only reverse row group read order, not rows within row groups
-    /// - This provides approximate ordering that benefits limit pushdown
-    ///
-    /// # Phase 2 (Future)
-    /// Could return `Exact` when we can guarantee perfect ordering through techniques like:
-    /// - File reordering based on statistics
-    /// - Detecting already-sorted data
-    ///   This would allow removing the Sort operator entirely.
+    /// # Behavior
+    /// - Returns `Exact` when the file's natural ordering (from Parquet metadata) already
+    ///   satisfies the requested ordering. This allows the Sort operator to be eliminated
+    ///   if the files within each group are also non-overlapping (checked by FileScanConfig).
+    /// - Returns `Inexact` when reversing the row group scan order would help satisfy the
+    ///   requested ordering. We still need a Sort operator at a higher level because:
+    ///   - We only reverse row group read order, not rows within row groups
+    ///   - This provides approximate ordering that benefits limit pushdown
     ///
     /// # Returns
+    /// - `Exact`: The file's natural ordering satisfies the request (within-file ordering guaranteed)
     /// - `Inexact`: Created an optimized source (e.g., reversed scan) that approximates the order
     /// - `Unsupported`: Cannot optimize for this ordering
     fn try_pushdown_sort(
@@ -767,6 +765,16 @@ impl FileSource for ParquetSource {
             return Ok(SortOrderPushdownResult::Unsupported);
         }
 
+        // Check if the natural (non-reversed) ordering already satisfies the request.
+        // Parquet metadata guarantees within-file ordering, so if the ordering matches
+        // we can return Exact. FileScanConfig will verify that files within each group
+        // are non-overlapping before declaring the entire scan as Exact.
+        if eq_properties.ordering_satisfy(order.iter().cloned())? {
+            return Ok(SortOrderPushdownResult::Exact {
+                inner: Arc::new(self.clone()) as Arc<dyn FileSource>,
+            });
+        }
+
         // Build new equivalence properties with the reversed ordering.
         // This allows us to check if the reversed ordering satisfies the request
         // by leveraging:
@@ -811,11 +819,6 @@ impl FileSource for ParquetSource {
         Ok(SortOrderPushdownResult::Inexact {
             inner: Arc::new(new_source) as Arc<dyn FileSource>,
         })
-
-        // TODO Phase 2: Add support for other optimizations:
-        // - File reordering based on min/max statistics
-        // - Detection of exact ordering (return Exact to remove Sort operator)
-        // - Partial sort pushdown for prefix matches
     }
 
     fn apply_expressions(