Merge pull request #2 from adriangb/pr22493-rg-reorder-fix

fix(sort-pushdown): drop runtime row-group reorder hints on Inexact→Exact upgrade (correctness)

Author: zhuqi-lucas

datafusion/datasource/src/file_scan_config/mod.rs

@@ -978,7 +978,7 @@ impl DataSource for FileScanConfig {
                 }
             }
             SortOrderPushdownResult::Inexact { inner } => {
-                let config = 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`
@@ -992,6 +992,26 @@ impl DataSource for FileScanConfig {
                         inner: Arc::new(config),
                     })
                 } else {
+                    // 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 mis-orders two row groups within a
+                    // single file that share the same `min` (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),
                     })

datafusion/datasource/src/file_scan_config/sort_pushdown.rs

@@ -169,20 +169,24 @@ impl FileScanConfig {
         //
         // This interleaving is actually beneficial because SPM pulls from both
         // partitions concurrently, keeping parallel I/O active.
-        let keep_ordering = if all_non_overlapping {
-            if is_exact {
-                true
-            } else {
-                // Re-validate: now that files are sorted, can the
-                // request be satisfied?
-                //
-                // 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.
+        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.
+            //
+            // 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
@@ -201,8 +205,6 @@ impl FileScanConfig {
                     new_eq_props.ordering_satisfy(order.iter().cloned())?
                 }
             }
-        } else {
-            false
         };
 
         if !keep_ordering {

datafusion/sqllogictest/test_files/sort_pushdown.slt

@@ -1110,7 +1110,7 @@ EXPLAIN SELECT * FROM reversed_parquet ORDER BY id ASC;
 logical_plan
 01)Sort: reversed_parquet.id ASC NULLS LAST
 02)--TableScan: reversed_parquet projection=[id, value]
-physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/c_low.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/a_high.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, sort_order_for_reorder=[id@0 ASC NULLS LAST]
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/c_low.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/a_high.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 # Test 4.2: Results must be correct
 query II
@@ -1336,7 +1336,7 @@ EXPLAIN SELECT * FROM reversed_with_order_parquet ORDER BY id ASC;
 logical_plan
 01)Sort: reversed_with_order_parquet.id ASC NULLS LAST
 02)--TableScan: reversed_with_order_parquet projection=[id, value]
-physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/c_low.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/a_high.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, sort_order_for_reorder=[id@0 ASC NULLS LAST]
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/c_low.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/reversed/a_high.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 # Test 6.2: Results must be correct
 query II
@@ -1473,7 +1473,7 @@ EXPLAIN SELECT * FROM desc_reversed_parquet ORDER BY id DESC;
 logical_plan
 01)Sort: desc_reversed_parquet.id DESC NULLS FIRST
 02)--TableScan: desc_reversed_parquet projection=[id, value]
-physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/desc_reversed/b_high.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/desc_reversed/a_low.parquet]]}, projection=[id, value], output_ordering=[id@0 DESC], file_type=parquet, sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/desc_reversed/b_high.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/desc_reversed/a_low.parquet]]}, projection=[id, value], output_ordering=[id@0 DESC], file_type=parquet
 
 # Test 8.2: Results must be correct
 query II
@@ -1486,6 +1486,78 @@ SELECT * FROM desc_reversed_parquet ORDER BY id DESC;
 2 200
 1 100
 
+# Test 8b: DESC with multiple row groups per file sharing a min value.
+# Regression test for the Inexact→Exact upgrade: when SortExec is eliminated
+# the files must be read in natural order. The opener's runtime row-group
+# reorder (sort ASC-by-min then reverse) mis-orders two row groups in one file
+# that share the same min — so the upgrade must NOT leave those hints active.
+#
+# File b_high is DESC-sorted [10,8,8,8] written with 2 rows per row group:
+#   RG0 = [10, 8] (min 8, max 10)
+#   RG1 = [ 8, 8] (min 8, max 8)
+# Both row groups have min=8. Naively reordering RGs ASC-by-min then reversing
+# yields [RG1, RG0] → 8,8,10,8 (wrong). Natural order [RG0, RG1] is correct.
+
+statement ok
+CREATE TABLE rg_desc_high(id INT, value INT) AS VALUES (10, 100), (8, 801), (8, 802), (8, 803);
+
+statement ok
+CREATE TABLE rg_desc_low(id INT, value INT) AS VALUES (3, 300), (2, 200), (1, 100);
+
+query I
+COPY (SELECT * FROM rg_desc_high ORDER BY id DESC)
+TO 'test_files/scratch/sort_pushdown/rg_desc/b_high.parquet'
+OPTIONS ('format.max_row_group_size' '2');
+----
+4
+
+query I
+COPY (SELECT * FROM rg_desc_low ORDER BY id DESC)
+TO 'test_files/scratch/sort_pushdown/rg_desc/a_low.parquet'
+OPTIONS ('format.max_row_group_size' '2');
+----
+3
+
+# Files named so filesystem order [a_low, b_high] is wrong for DESC → the
+# Inexact path fires, stats reorder makes file groups [b_high, a_low]
+# non-overlapping, and the upgrade eliminates SortExec.
+statement ok
+CREATE EXTERNAL TABLE rg_desc_parquet(id INT, value INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/sort_pushdown/rg_desc/'
+WITH ORDER (id DESC);
+
+# SortExec eliminated, files reordered, NO sort_order_for_reorder /
+# reverse_row_groups (natural read is correct after the upgrade).
+query TT
+EXPLAIN SELECT id FROM rg_desc_parquet ORDER BY id DESC;
+----
+logical_plan
+01)Sort: rg_desc_parquet.id DESC NULLS FIRST
+02)--TableScan: rg_desc_parquet projection=[id]
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/rg_desc/b_high.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/rg_desc/a_low.parquet]]}, projection=[id], output_ordering=[id@0 DESC], file_type=parquet
+
+# Results must be in DESC order — id=10 first.
+query I
+SELECT id FROM rg_desc_parquet ORDER BY id DESC;
+----
+10
+8
+8
+8
+3
+2
+1
+
+statement ok
+DROP TABLE rg_desc_parquet;
+
+statement ok
+DROP TABLE rg_desc_high;
+
+statement ok
+DROP TABLE rg_desc_low;
+
 # Test 9: Multi-column sort key validation
 # Files have (category, id) ordering. Files share a boundary value on category='B'
 # so column-level min/max statistics overlap on the primary key column.
@@ -2366,7 +2438,7 @@ logical_plan
 physical_plan
 01)SortPreservingMergeExec: [id@0 ASC NULLS LAST]
 02)--BufferExec: capacity=1073741824
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, sort_order_for_reorder=[id@0 ASC NULLS LAST]
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 # Verify correctness
 query II
@@ -2393,7 +2465,7 @@ logical_plan
 physical_plan
 01)SortPreservingMergeExec: [id@0 ASC NULLS LAST], fetch=3
 02)--BufferExec: capacity=1073741824
-03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], limit=3, output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, sort_order_for_reorder=[id@0 ASC NULLS LAST]
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/b_mid.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/a_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tg_buffer/c_low.parquet]]}, projection=[id, value], limit=3, output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
 
 query II
 SELECT * FROM tg_buffer ORDER BY id ASC LIMIT 3;