tschwarzinger commented on PR #27170:
URL: https://github.com/apache/datafusion/pull/27170#issuecomment-6738869507
After spending more time with this particular part of the code, I still
think this is a good (but not the only) option for fixing the bug. `SortExec`
can return, IMO, a wrong output ordering if `preserve_partitioning` is false
until the optimizer changes the inner plan to return a single partition. During
this window `COPY` / `DataSinkExec` obtains the output ordering and holds on to
it.
The cause of the bug:
1. `SortExec` is being planned from the `Sort` logical node in the physical
planner (sort expressions: [id, ts])
2. At this point, `SortExec`s equivalence properties show and output
ordering of [ts] and a heterogenous constant for id. Each partition has a
different id and each partition is sorted on [ts]. `SortExec` does not change
the output ordering to `[id, ts]` even though this will eventually be the
output ordering after the optimizer has changes the input of `SortExec`.
`OutputRequirements::reorder` does not contain `id` because it's a heterogenous
constant.
3. `COPY` uses the output ordering of `SortExec` (`[ts]`) as required input
ordering during logical -> physical planning.
4. The optimizer satisfies the sink’s input ordering requirement (`[ts]`). A
`SortPreservingMergeExec: [ts]` over the union is enough, so the `SortExec` is
removed.
There are at least two appraoches on how we could fix this:
- Make the output ordering function of `SortExec` aware that it will be
changed by the optimizer. So even though `SortExec` still has multiple input
partitions, it reports its output ordering as if it would have only a single
input partition, if `preserve_partitioning` is false. This is what this PR is
trying to achieve and it's similar to how it's done with the output
partitioning where we also have a special casing based on
`preserve_partitioning`. See below.
- We could use the mechanism that is used for `OutputRequirementsExec`
(i.e., traversing the plan and downcasting `SortExec`) also for `COPY`.
However, this only works for `SortExec` and `SortPreservingMergeExec`.
#### Existing Special Case based on `preserve_partitioning`:
```rust
fn output_partitioning_helper(
input: &Arc<dyn ExecutionPlan>,
preserve_partitioning: bool,
) -> Partitioning {
// Get output partitioning:
if preserve_partitioning {
input.output_partitioning().clone()
} else {
Partitioning::UnknownPartitioning(1)
}
}
```
#### Logical to Phyiscal Planner for Writes:
```rust
// ...
LogicalPlan::Copy(CopyTo {
input,
output_url,
file_type,
partition_by,
options: source_option_tuples,
output_schema: _,
}) => {
// ...
let ordering =
input_exec.properties().output_ordering().cloned(); // <--- here the not yet
finalized output ordering is stored
sink_format
.create_writer_physical_plan(
input_exec,
session,
config,
ordering.map(Into::into),
)
.await?
// ...
}
```
#### Why only `COPY` and not every query?
The issue doesn't get raised with `OutputRequirementExec` because it uses a
custom traversal of the execution plan and uses the sort expressions of
`SortExec` by downcasting it. It even has a comment for constants in
`require_top_ordering_helper`.
```text
// In case of constant columns, output ordering of the `SortExec` would
// be an empty set. Therefore; we check the sort expression field to
// assign the requirements.
let req_dist = sort_exec
.input_distribution_requirements()
.into_per_child()
.swap_remove(0);
let req_ordering = sort_exec.expr();
let reqs = OrderingRequirements::from(req_ordering.clone());
```
`EXPLAIN VERBOSE` at main without `COPY`:
```text
physical_plan after OutputRequirements
// The order_by below looks good!
01)OutputRequirementExec: order_by=[(id@0, asc), (ts@1, asc)],
dist_by=SinglePartition
02)--SortExec: expr=[id@0 ASC NULLS LAST, ts@1 ASC NULLS LAST],
preserve_partitioning=[false]
03)----UnionExec
04)------ProjectionExec: expr=[b as id, ts@0 as ts, x@1 as x]
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/copy/union_const_order/sorted.parquet]]},
projection=[ts, x], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet
06)------ProjectionExec: expr=[a as id, ts@0 as ts, x@1 as x]
07)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/copy/union_const_order/sorted.parquet]]},
projection=[ts, x], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet
```
`EXPLAIN VERBOSE` at main with `COPY` and a change that shows the input
ordering requirements of `DataSinkExec`:
initial_physical_plan_with_stats
```text
// The `order=` loses the `id` which causes the observed bug
01)DataSinkExec: sink=ParquetSink(file_groups=[]) order=[ts@1 ASC NULLS
LAST], statistics=[Rows=Absent, Bytes=Absent, [(Col[0]:)]]
02)--SortExec: expr=[id@0 ASC NULLS LAST, ts@1 ASC NULLS LAST],
preserve_partitioning=[false], statistics=[Rows=Exact(6), Bytes=Inexact(96),
[(Col[0]: Min=Exact(Utf8("a")) Max=Exact(Utf8("b")) Null=Exact(0)
Distinct=Inexact(2)),(Col[1]: Min=Exact(Int64(1)) Max=Exact(Int64(3))
Null=Exact(0) ScanBytes=Exact(48)),(Col[2]: Min=Exact(Float64(10))
Max=Exact(Float64(30)) Null=Exact(0) ScanBytes=Exact(48))]]
03)----UnionExec, statistics=[Rows=Exact(6), Bytes=Inexact(96), [(Col[0]:
Min=Exact(Utf8("a")) Max=Exact(Utf8("b")) Null=Exact(0)
Distinct=Inexact(2)),(Col[1]: Min=Exact(Int64(1)) Max=Exact(Int64(3))
Null=Exact(0) ScanBytes=Exact(48)),(Col[2]: Min=Exact(Float64(10))
Max=Exact(Float64(30)) Null=Exact(0) ScanBytes=Exact(48))]]
04)------ProjectionExec: expr=[b as id, ts@0 as ts, x@1 as x],
statistics=[Rows=Exact(3), Bytes=Inexact(48), [(Col[0]: Min=Exact(Utf8("b"))
Max=Exact(Utf8("b")) Null=Exact(0) Distinct=Exact(1)),(Col[1]:
Min=Exact(Int64(1)) Max=Exact(Int64(3)) Null=Exact(0)
ScanBytes=Exact(24)),(Col[2]: Min=Exact(Float64(10)) Max=Exact(Float64(30))
Null=Exact(0) ScanBytes=Exact(24))]]
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/copy/union_const_order/sorted.parquet]]},
projection=[ts, x], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet,
statistics=[Rows=Exact(3), Bytes=Exact(48), [(Col[0]: Min=Exact(Int64(1))
Max=Exact(Int64(3)) Null=Exact(0) ScanBytes=Exact(24)),(Col[1]:
Min=Exact(Float64(10)) Max=Exact(Float64(30)) Null=Exact(0)
ScanBytes=Exact(24))]]
06)------ProjectionExec: expr=[a as id, ts@0 as ts, x@1 as x],
statistics=[Rows=Exact(3), Bytes=Inexact(48), [(Col[0]: Min=Exact(Utf8("a"))
Max=Exact(Utf8("a")) Null=Exact(0) Distinct=Exact(1)),(Col[1]:
Min=Exact(Int64(1)) Max=Exact(Int64(3)) Null=Exact(0)
ScanBytes=Exact(24)),(Col[2]: Min=Exact(Float64(10)) Max=Exact(Float64(30))
Null=Exact(0) ScanBytes=Exact(24))]]
07)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/copy/union_const_order/sorted.parquet]]},
projection=[ts, x], output_ordering=[ts@0 ASC NULLS LAST], file_type=parquet,
statistics=[Rows=Exact(3), Bytes=Exact(48), [(Col[0]: Min=Exact(Int64(1))
Max=Exact(Int64(3)) Null=Exact(0) ScanBytes=Exact(24)),(Col[1]:
Min=Exact(Float64(10)) Max=Exact(Float64(30)) Null=Exact(0)
ScanBytes=Exact(24))]]
```
--
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]
