andygrove opened a new issue, #6770:
URL: https://github.com/apache/datafusion-comet/issues/6770
### Describe the bug
With `spark.comet.exec.forceShuffledHashJoin=true`, a local sort above a
sort-merge join is lost when `RewriteJoin` converts that join to a hash join.
Spark's `RemoveRedundantSorts` runs before Comet's rules and deletes the
`SortExec`, because the sort-merge join's output ordering already satisfies it.
`RewriteJoin` then replaces the join with a `ShuffledHashJoinExec` and drops
its input sorts, so the ordering is gone and nothing puts it back. It only
happens when the join's ordering satisfies the sort, which means the sort keys
are the join keys or a leading prefix of them.
Two things break:
- `sortWithinPartitions(k)` and `SORT BY k` return partitions that are not
sorted. The rows are right, and nothing reports the problem.
- A partitioned insert whose partition column is the join key fails. Spark's
planned write (`V1Writes`) adds a local sort on the partition columns, and that
is the sort that gets removed. Spark's single dynamic-partition writer expects
sorted input. When a partition value comes back, it opens the same file name
again and the task fails with `FileAlreadyExistsException`.
### Steps to reproduce
Parquet tables `big` as `(i % 100, i)` for 10000 rows and `small` as `(i *
10, i)` for 10 rows, with:
```
spark.comet.exec.forceShuffledHashJoin=true
spark.sql.autoBroadcastJoinThreshold=-1
spark.sql.adaptive.autoBroadcastJoinThreshold=-1
spark.sql.shuffle.partitions=2
```
Unsorted partitions, with AQE on or off:
```scala
val df = spark
.sql("SELECT big._1 AS k, big._2 AS v FROM big JOIN small ON big._1 =
small._1")
.sortWithinPartitions("k")
df.rdd
.mapPartitions { it =>
val ks = it.map(_.getInt(0)).toArray
Iterator(ks.sameElements(ks.sorted))
}
.collect()
// Array(false, false) here, Array(true, true) with
forceShuffledHashJoin=false
```
Appending `SORT BY k` to the query gives the same result. The plan has
`CometHashJoin` directly under the final `CometProject`, with no sort above it.
Failed insert, with AQE on or off:
```sql
CREATE TABLE out (v INT, w INT, k INT) USING parquet PARTITIONED BY (k);
INSERT OVERWRITE TABLE out
SELECT big._2, small._2, big._1 FROM big JOIN small ON big._1 = small._1;
```
```
org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while
writing rows to file:/.../out. SQLSTATE: 58030
...
Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: File already
exists: file:/.../out/_temporary/.../k=10/part-00000-<job
id>.c000.snappy.parquet
```
With `forceShuffledHashJoin=false`, the insert writes one file per partition
and all 1000 rows. That plan has `WriteFiles` directly over
`CometSortMergeJoin` with no sort between them, because Spark already removed
the write's sort as redundant. With the force config on, a `CometHashJoin` sits
in the same place.
Reproduced on `main` with Spark 4.1.
### Expected behavior
The same result as Spark: every partition sorted by `k`, and the insert
succeeds.
### Additional context
Setting `spark.sql.execution.removeRedundantSorts=false` works around both
cases. Spark then keeps the `SortExec`, the partitions come back sorted and the
insert succeeds.
#6384 adds a pass after the rewrite that restores any ordering an operator
above still requires. That fixes a kept sort-merge join above a rewritten one
(#6673) and a window over one, but it can't fix this. Nothing left in the plan
requires the ordering, because `WriteFilesExec` declares no required child
ordering. `sortWithinPartitions` output is still unsorted on that branch.
`branch-1.1` has the same rewrite, so 1.1.0 is affected too.
One way to fix it is to choose the hash join before `EnsureRequirements` and
`RemoveRedundantSorts` run, for example in a planner strategy, so that Spark
places and keeps the sorts itself.
--
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]
