Wrong results: order-preserving RepartitionExec keeps orderings of its input that do not hold after the merge #26745

Describe the bug

An order-preserving RepartitionExec (preserve_order=true) whose input has more than one ordering merges its input partitions on all of those orderings joined into one key. Its output properties, however, still claim each input ordering on its own, and those no longer hold after the merge. The planner then removes a sort that is needed, so ORDER BY returns rows in the wrong order.

This is the RepartitionExec form of #26646, which #26654 fixed for SortPreservingMergeExec.

To Reproduce

In datafusion-cli:

COPY (VALUES (1, 10), (2, 20)) TO 'w/f1.csv' STORED AS CSV OPTIONS ('format.has_header' 'true');
COPY (VALUES (1, 1), (3, 2)) TO 'w/f2.csv' STORED AS CSV OPTIONS ('format.has_header' 'true');

SET datafusion.execution.target_partitions = 2;
SET datafusion.optimizer.prefer_existing_sort = true;
SET datafusion.optimizer.repartition_file_scans = false;

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

SELECT a, b, count(*) FROM w GROUP BY a, b ORDER BY b;

Each file is sorted by a and also by b, as the table declares. prefer_existing_sort makes the planner keep the order through the repartition instead of sorting, and repartition_file_scans = false keeps one file per partition.

Expected behavior

count(*) is 1 on every row. DuckDB returns the expected order.

Additional context

The plan has no sort for ORDER BY b:

SortPreservingMergeExec: [b@1 ASC NULLS LAST]
  ProjectionExec: expr=[a@0 as a, b@1 as b, count(Int64(1))@2 as count(*)]
    AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[count(Int64(1))], group_clustering_mode=Full
      RepartitionExec: partitioning=Hash([a@0, b@1], 2), input_partitions=2, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST, b@1 ASC NULLS LAST
        AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b], aggr=[count(Int64(1))], group_clustering_mode=Full
          DataSourceExec: file_groups={2 groups: [[.../w/f1.csv], [.../w/f2.csv]]}, projection=[a, b], output_orderings=[[a@0 ASC NULLS LAST], [b@1 ASC NULLS LAST]], file_type=csv, has_header=true

• RepartitionExec merges on sort_exprs, which is self.input.output_ordering() (datafusion/physical-plan/src/repartition/mod.rs). EquivalenceProperties::output_ordering joins all the input's orderings into one, so here the merge key is [a, b], and each output partition is sorted by [a, b], not by b.
• RepartitionExec::eq_properties_helper keeps every input ordering when preserve_order is set, so the output still claims [b]. The sort for ORDER BY b is removed, and the final SortPreservingMergeExec merges partitions that are not sorted by b.

A fix direction: when preserve_order is set and there is more than one input partition, keep only the merge key's ordering (clear the orderings, then add sort_exprs), as #26654 did for SortPreservingMergeExec. I have not tried it yet. A fix should add a sqllogictest for the query above.

Willingness to contribute

Anyone is welcome to pick this up. Feel free to ping me with questions about the cause or when you'd like a review.