fix(sort-pushdown): restore SortExec elimination after stats-based file reorder (apache#22493)

## Which issue does this PR close?

- Closes apache#22494


Regression introduced by apache#21956 — single-partition multi-file scans with
`WITH ORDER` (or inferred parquet `sorting_columns` metadata) plus files
listed out of order on disk no longer have their `SortExec` eliminated,
even when the stats-based file reorder produces a non-overlapping
layout.
This was the central case PR apache#21182 was designed to fix.

## Rationale for this change

In the Phase 2 sort pushdown design (apache#21182), when a file source
returned
`Unsupported` because `validated_output_ordering()` stripped its
declared
ordering, the outer `FileScanConfig::try_pushdown_sort` invoked a
fallback
(`try_sort_file_groups_by_statistics`) that reordered files by min/max
stats, re-validated the ordering against the new file groups, and could
upgrade the result back to `Exact` — dropping the outer `SortExec`.

PR apache#21956 added the `column_in_file_schema` signal so that plain-column
sort requests would return `Inexact` (with `sort_order_for_reorder` set)
instead of `Unsupported`. That enabled runtime per-RG reorder, but it
also pulled the typical wrong-file-order case out of the
`Unsupported`-fallback re-validation path. The Inexact branch of
`rebuild_with_source` always strips `output_ordering`, so even when the
post-sort file groups became non-overlapping and the declared ordering
would re-validate, the outer wrapper returned `Inexact` and PushdownSort
kept the `SortExec`.

The SLT comment on test 6.1 (`# … → SortExec eliminated`) still
describes
the pre-apache#21956 behaviour, but the recorded expectation was updated to
match the post-apache#21956 plan where SortExec stayed — a silent regression.

## What changes are included in this PR?

### `rebuild_with_source`

When `is_exact=false` but the stats-based file sort produced
`all_non_overlapping=true`, re-validate the declared `output_ordering`
against the new file groups. If `ordering_satisfy` passes, preserve
`output_ordering` instead of stripping it.

### Outer `FileScanConfig::try_pushdown_sort`

On the `Inexact` branch, inspect `config.output_ordering` after rebuild
and return `Exact` if it survived. This restores the
`Unsupported → upgrade Exact` semantics, but on the `Inexact` branch the
plain-column path now follows.

### Safety

- When the source has no declared ordering (no `WITH ORDER` and no
  parquet `sorting_columns` metadata), `self.output_ordering` is empty,
  the re-validate produces no orderings, `ordering_satisfy` returns
  `false`, and `SortExec` stays. min/max stats alone never trigger an
  upgrade.
- When `all_non_overlapping=false` (overlapping files post-sort),
  `output_ordering` is stripped exactly as before.

### Tests

`sort_pushdown.slt`:

- Tests **4.1** (parquet metadata), **6.1** (`WITH ORDER ASC`),
  **8.1** (`WITH ORDER DESC` with reverse), **G.1**, **G.2** (multi-
  partition SPM + BufferExec) — expectations updated to match the
  restored `SortExec`-eliminated plans (matches the PR apache#21182 era plans
  and the SLT comments that were already in the file).
- New **Test 5b** — files written without `ORDER BY` (no
  `sorting_columns` metadata) + external table without `WITH ORDER`:
  asserts that even when min/max stats happen to be non-overlapping,
  `SortExec` is kept.

## Are these changes tested?

Yes — `sort_pushdown.slt` covers the affected paths comprehensively.
Full `cargo test -p datafusion-sqllogictest --test sqllogictests` and
`cargo test -p datafusion-datasource --lib` / `-p
datafusion-datasource-parquet --lib` / `-p
datafusion-physical-optimizer`
pass. `cargo clippy -D warnings` clean.

## Are there any user-facing changes?

Yes — for the in-scope scenarios above, `EXPLAIN` plans now show the
`SortExec` removed and `output_ordering` set on `DataSourceExec`. Query
results are unaffected.

---------

Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

datafusion/datasource/src/file_scan_config/mod.rs

@@ -937,14 +937,19 @@ impl DataSource for FileScanConfig {
     ///   │     → SortExec removed, fetch (LIMIT) pushed to DataSourceExec
     ///   │
     ///   ├─► FileSource returns Inexact
-    ///   │     (reverse_row_groups=true)
-    ///   │     → SortExec kept, scan optimized
+    ///   │     (e.g. column_in_file_schema: opener will reorder RGs at runtime)
+    ///   │     → rebuild_with_source: sort files by stats; if the post-sort
+    ///   │       file groups are non-overlapping AND the request now validates
+    ///   │       AND no NULLs sit in the sort columns of non-last files,
+    ///   │       upgrade back to Exact (SortExec removed). Otherwise stays
+    ///   │       Inexact and SortExec is kept while the scan is still
+    ///   │       optimised via `sort_order_for_reorder` / `reverse_row_groups`.
     ///   │
     ///   └─► FileSource returns Unsupported
-    ///         (ordering stripped because files in wrong order)
+    ///         (e.g. expression sort key or partition column)
     ///         → try_sort_file_groups_by_statistics():
     ///           1. Sort files within each group by min/max statistics
-    ///           2. Re-check: non-overlapping + ordering valid?
+    ///           2. Re-check: non-overlapping + ordering valid + no NULLs?
     ///              YES → Exact → SortExec removed
     ///              NO  → Inexact (files reordered, Sort stays)
     /// ```
@@ -973,8 +978,42 @@ impl DataSource for FileScanConfig {
                 }
             }
             SortOrderPushdownResult::Inexact { inner } => {
-                Ok(SortOrderPushdownResult::Inexact {
-                    inner: Arc::new(self.rebuild_with_source(inner, false, order)?),
+                let mut config = self.rebuild_with_source(inner, false, order)?;
+                // `rebuild_with_source` reorders files by stats; if the
+                // post-sort files are non-overlapping AND the request now
+                // validates against the new file groups, `output_ordering`
+                // is preserved and we can upgrade back to Exact. This
+                // restores the sort-elimination behaviour that lived in
+                // the `Unsupported` → `try_sort_file_groups_by_statistics`
+                // path before #21956 routed `column_in_file_schema` cases
+                // here.
+                if config.output_ordering.is_empty() {
+                    return Ok(SortOrderPushdownResult::Inexact {
+                        inner: Arc::new(config),
+                    });
+                }
+                // Upgrading to Exact: the post-sort file groups are
+                // non-overlapping and each file's declared ordering
+                // re-validates, so reading the files in their natural
+                // (declared-sorted) order already yields the requested
+                // ordering — exactly like the `Unsupported` → Exact path,
+                // which reads files in natural order too.
+                //
+                // Drop the runtime row-group reorder hints the Inexact
+                // source carried (`sort_order_for_reorder` /
+                // `reverse_row_groups`) by restoring the original,
+                // hint-free source. With the `SortExec` removed those
+                // hints are not just redundant but unsafe: for a DESC
+                // request the opener sorts row groups ASC-by-min and then
+                // reverses them, which reorders two row groups within a
+                // single file that share the same `min` incorrectly
+                // (e.g. a file `[10,8,8,8]` whose row groups are
+                // `[10,8]` and `[8,8]` would stream as `8,8,10,8`).
+                // The `SortExec` used to mask this; once it is gone the
+                // reordered stream is the final, wrong answer.
+                config.file_source = Arc::clone(&self.file_source);
+                Ok(SortOrderPushdownResult::Exact {
+                    inner: Arc::new(config),
                 })
             }
             SortOrderPushdownResult::Unsupported => {

datafusion/datasource/src/file_scan_config/sort_pushdown.rs

@@ -138,31 +138,76 @@ impl FileScanConfig {
             false
         };
 
-        if is_exact && all_non_overlapping {
-            // Truly exact: within-file ordering guaranteed and files are non-overlapping.
-            // Keep output_ordering so SortExec can be eliminated for each partition.
-            //
-            // We intentionally do NOT redistribute files across groups here.
-            // The planning-phase bin-packing may interleave file ranges across groups:
-            //
-            //   Group 0: [f1(1-10), f3(21-30)]   ← interleaved with group 1
-            //   Group 1: [f2(11-20), f4(31-40)]
-            //
-            // This interleaving is actually beneficial because SPM pulls from both
-            // partitions concurrently, keeping parallel I/O active:
-            //
-            //   SPM: pull P0 [1-10] → pull P1 [11-20] → pull P0 [21-30] → pull P1 [31-40]
-            //        ^^^^^^^^^^^^     ^^^^^^^^^^^^
-            //        both partitions scanning files simultaneously
-            //
-            // If we were to redistribute files consecutively:
-            //   Group 0: [f1(1-10), f2(11-20)]   ← all values < group 1
-            //   Group 1: [f3(21-30), f4(31-40)]
+        // Decide whether to keep `output_ordering` (i.e. let the outer
+        // pushdown report `Exact` and drop `SortExec`).
+        //
+        // Two paths can produce a keep:
+        //
+        //   1. `is_exact && all_non_overlapping`: the source already had
+        //      validated ordering and the post-sort files still don't
+        //      overlap — Exact carries through unchanged.
+        //
+        //   2. `!is_exact && all_non_overlapping`: source returned
+        //      `Inexact` because pre-sort `validated_output_ordering()`
+        //      stripped the declaration (files were listed out of order
+        //      on disk). After our stats-based sort the files are now
+        //      non-overlapping — re-validate against the new file
+        //      groups and, if it passes, upgrade back to Exact so the
+        //      outer wrapper drops the `SortExec`. Without this, the
+        //      `Inexact` branch stayed Inexact even when reorder
+        //      restored a perfectly valid ordering, leaving an
+        //      unnecessary `SortExec` above the source (regression
+        //      after #21956's `column_in_file_schema` signal pushed
+        //      this scenario into the Inexact branch instead of the
+        //      `try_sort_file_groups_by_statistics` fallback).
+        //
+        // We intentionally do NOT redistribute files across groups here.
+        // The planning-phase bin-packing may interleave file ranges across groups:
+        //
+        //   Group 0: [f1(1-10), f3(21-30)]   ← interleaved with group 1
+        //   Group 1: [f2(11-20), f4(31-40)]
+        //
+        // This interleaving is actually beneficial because SPM pulls from both
+        // partitions concurrently, keeping parallel I/O active.
+        let keep_ordering = match (all_non_overlapping, is_exact) {
+            // Files still overlap after the stats sort — the combined
+            // stream isn't ordered, so `output_ordering` must be dropped.
+            (false, _) => false,
+            // Source already had validated ordering and the post-sort
+            // files still don't overlap — Exact carries through.
+            (true, true) => true,
+            // Source returned `Inexact`; re-validate against the
+            // reordered file groups to decide whether to upgrade.
             //
-            // SPM would read ALL of group 0 first (values always smaller), then group 1.
-            // This degrades to single-threaded sequential I/O — the other partition
-            // sits idle the entire time, losing the parallelism benefit.
-        } else {
+            // Same NULL guard as `try_sort_file_groups_by_statistics`:
+            // we cannot claim Exact if any non-last file contains
+            // NULLs in the sort columns. With NULLS LAST those
+            // NULLs sit after all non-null rows in the file, so
+            // when the next file's non-nulls are smaller than the
+            // previous file's max, they'd appear *after* the NULLs
+            // in the concatenated stream — breaking the ordering.
+            (true, false) => {
+                let projected_schema = new_config.projected_schema()?;
+                let projection_indices = new_config
+                    .file_source
+                    .projection()
+                    .as_ref()
+                    .and_then(|p| ordered_column_indices_from_projection(p));
+                if any_file_has_nulls_in_sort_columns(
+                    &new_config.file_groups,
+                    order,
+                    &projected_schema,
+                    projection_indices.as_deref(),
+                ) {
+                    false
+                } else {
+                    let new_eq_props = new_config.eq_properties();
+                    new_eq_props.ordering_satisfy(order.iter().cloned())?
+                }
+            }
+        };
+
+        if !keep_ordering {
             new_config.output_ordering = vec![];
         }