Wrong results: SortPreservingMergeExec keeps orderings of its input that do not hold after the merge #26646

Describe the bug

SortPreservingMergeExec merges several sorted partitions into one, ordered by its merge key. Its output properties keep every ordering its input had, and add the merge key. But an input ordering other than the merge key holds only inside each input partition, not across the merged output. The planner then believes the output is sorted by that other ordering and removes a sort that is needed, so the query returns rows in the wrong order.

To Reproduce

Two CSV files in one directory t/, each sorted by a and also by b:

t/f1.csv

a,b
1,10
2,20

t/f2.csv

a,b
1,1
3,2

In datafusion-cli:

SET datafusion.execution.target_partitions = 2;

CREATE EXTERNAL TABLE t (a INT, b INT) STORED AS CSV
WITH ORDER (a ASC) WITH ORDER (b ASC)
LOCATION 't/' OPTIONS ('format.has_header' 'true');

SELECT * FROM (SELECT a, b FROM t ORDER BY a LIMIT 10) ORDER BY b;

Result:

+---+----+
| a | b  |
+---+----+
| 1 | 10 |
| 1 | 1  |
| 2 | 20 |
| 3 | 2  |
+---+----+

The plan has no sort for the outer ORDER BY b:

SortPreservingMergeExec: [a@0 ASC NULLS LAST], fetch=10
  DataSourceExec: file_groups={2 groups: [[.../f1.csv], [.../f2.csv]]}, projection=[a, b], output_orderings=[[a@0 ASC NULLS LAST], [b@1 ASC NULLS LAST]], file_type=csv, has_header=true

Expected behavior

The rows are sorted by b: (1, 1), (3, 2), (1, 10), (2, 20). The plan sorts by b after the merge.

Additional context

The output properties are computed in SortPreservingMergeExec::compute_properties (datafusion/physical-plan/src/sorts/sort_preserving_merge.rs):

let mut eq_properties = input.equivalence_properties().clone();
eq_properties.clear_per_partition_constants();
eq_properties.add_ordering(ordering);

With more than one input partition, only the merge key's ordering holds after the merge, so the input orderings should be cleared before add_ordering. With one input partition nothing is merged and the input orderings still hold.

I tried that change on main:

let mut eq_properties = input.equivalence_properties().clone();
if input_partitions > 1 {
    eq_properties.clear_orderings();
}
eq_properties.clear_per_partition_constants();
eq_properties.add_ordering(ordering);

The query above then sorts by b and returns the expected rows, and all sqllogictests, the physical-plan sort tests and the physical optimizer integration tests pass without changes. A fix should add a sqllogictest for the query above.

#26643 (draft) moves this code to EquivalenceProperties::merged; a fix on top of it would be
.merged(input_partitions, false) followed by add_ordering.

A preserve-order RepartitionExec has the same problem in a different form (it merges on all orderings joined together but keeps each one separately). That case is harder and is not part of this issue.

Willingness to contribute

I cannot contribute a fix for this bug at this time. (Leaving it for another contributor.)