(datafusion) 01/01: fix: SortPreservingMergeExec drops input orderings when merging partitions (#26654)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26654-cd17d8ea053af1d55536d640324ad9dc0a5c862c
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit ebf48581898530f01d36e5580a29c8b8df3012de
Author: Harshil Mital <89...@users.noreply.github.com>
AuthorDate: Fri Oct 23 11:05:51 2026 +0000

    fix: SortPreservingMergeExec drops input orderings when merging partitions (#26654)
    
    ## Which issue does this PR close?
    
    - Closes #26646.
    
    ## Rationale for this change
    
    When a sorted input with more than one partition goes through
    `SortPreservingMergeExec`, the planner treats every ordering of the
    input as still holding after the merge. Only the merge key ordering
    actually holds. The other orderings are true within each partition, but
    not across them.
    
    Because of this, the planner can remove a sort that is still needed, and
    the query returns rows out of order. With two files that are each sorted
    by `a` and `b` and `target_partitions = 2`:
    
    ```sql
    SELECT * FROM (SELECT a, b FROM t ORDER BY a LIMIT 10) ORDER BY b;
    ```
    
    the plan has no sort for the outer `ORDER BY b`, and the rows come back
    as `(1, 10), (1, 1), (2, 20), (3, 2)`.
    
    ## What changes are included in this PR?
    
    - `SortPreservingMergeExec::compute_properties` clears the input
    orderings when there is more than one input partition, before adding the
    merge key ordering. This matches what `CoalescePartitionsExec` and
    `RepartitionExec` already do. With a single input partition nothing is
    merged, so the input orderings are kept as before.
    
    The preserve-order `RepartitionExec` case mentioned in the issue is not
    covered here.
    
    Draft #26643 moves this logic into `EquivalenceProperties::merged`. If
    that lands first, I'll update this PR to use `.merged(input_partitions,
    false)` followed by `add_ordering`.
    
    ## What is the testing strategy for this PR?
    
    - New sqllogictest case at the end of `order.slt`, using the
    reproduction from the issue. It checks both the query result and the
    physical plan, which must keep the `SortExec` above the
    `SortPreservingMergeExec`. It fails on `main` with the wrong row order
    from the issue, and passes with this change.
    - The existing sqllogictests and the `datafusion-physical-plan` /
    `datafusion-physical-optimizer` unit tests pass with no changes to
    expected plans.
    
    ## Are there any user-facing changes?
    
    Queries like the one above now return correctly ordered results. No
    public API changes.
---
 .../src/sorts/sort_preserving_merge.rs             |  5 ++
 datafusion/sqllogictest/test_files/order.slt       | 57 ++++++++++++++++++++++
 2 files changed, 62 insertions(+)

diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
index e865fef98f..febc494bfd 100644
--- a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
+++ b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
@@ -172,6 +172,11 @@ impl SortPreservingMergeExec {
         };
 
         let mut eq_properties = input.equivalence_properties().clone();
+        // Merging multiple partitions only preserves the merge key ordering.
+        // Other input orderings hold only within each partition:
+        if input_partitions > 1 {
+            eq_properties.clear_orderings();
+        }
         eq_properties.clear_per_partition_constants();
         eq_properties.add_ordering(ordering);
         PlanProperties::new(
diff --git a/datafusion/sqllogictest/test_files/order.slt b/datafusion/sqllogictest/test_files/order.slt
index 3a2c7ccb84..0d48421a83 100644
--- a/datafusion/sqllogictest/test_files/order.slt
+++ b/datafusion/sqllogictest/test_files/order.slt
@@ -2276,3 +2276,60 @@ SELECT i, -i AS n FROM (
 -128 -128
 -128 -128
 NULL NULL
+
+# SortPreservingMergeExec only guarantees its merge key ordering. Each file
+# below is sorted by both `a` and `b`, but merging them on `a` does not keep
+# them sorted by `b`, so the outer sort must stay.
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+query I
+COPY (VALUES (1, 10), (2, 20))
+TO 'test_files/scratch/order/spm_orderings/1.csv'
+OPTIONS ('format.has_header' 'true');
+----
+2
+
+query I
+COPY (VALUES (1, 1), (3, 2))
+TO 'test_files/scratch/order/spm_orderings/2.csv'
+OPTIONS ('format.has_header' 'true');
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE spm_orderings (
+  a BIGINT,
+  b BIGINT
+)
+STORED AS CSV
+WITH ORDER (a ASC)
+WITH ORDER (b ASC)
+LOCATION 'test_files/scratch/order/spm_orderings/'
+OPTIONS ('format.has_header' 'true');
+
+query II
+SELECT * FROM (SELECT a, b FROM spm_orderings ORDER BY a LIMIT 10) ORDER BY b;
+----
+1 1
+3 2
+1 10
+2 20
+
+query TT
+EXPLAIN SELECT * FROM (SELECT a, b FROM spm_orderings ORDER BY a LIMIT 10) ORDER BY b;
+----
+logical_plan
+01)Sort: spm_orderings.b ASC NULLS LAST
+02)--Sort: spm_orderings.a ASC NULLS LAST, fetch=10
+03)----TableScan: spm_orderings projection=[a, b]
+physical_plan
+01)SortExec: expr=[b@1 ASC NULLS LAST], preserve_partitioning=[false]
+02)--SortPreservingMergeExec: [a@0 ASC NULLS LAST], fetch=10
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/order/spm_orderings/1.csv], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/order/spm_orderings/2.csv]]}, projection=[a, b], limit=10, output_orderings=[[a@0 ASC NULLS LAST], [b@1 ASC NULLS LAST]], file_type=csv, has_header=true
+
+statement ok
+DROP TABLE spm_orderings;
+
+statement ok
+set datafusion.execution.target_partitions = 4;


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org