jayzhan211 opened a new issue, #26745:
URL: https://github.com/apache/datafusion/issues/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`:
```sql
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
| Query | DataFusion returns `(a, b)` | Expected |
|---|---|---|
| `SELECT a, b, count(*) FROM w GROUP BY a, b ORDER BY b` | `(1, 1), (1,
10), (2, 20), (3, 2)` | `(1, 1), (3, 2), (1, 10), (2, 20)` |
`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.
--
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]
