jayzhan211 opened a new issue, #26646:
URL: https://github.com/apache/datafusion/issues/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`:
```sql
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`):
```rust
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:
```rust
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.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
