andygrove opened a new issue, #6922:
URL: https://github.com/apache/datafusion-comet/issues/6922
### Describe the bug
#6835 fixes #6787 by declaring the order Spark reports for a native outer
hash join's streamed side on the join's probe input, so DataFusion keeps the
unmatched probe rows in place. It declares the order only when it can show that
the order holds, and two shapes where it does hold still get no declaration.
The outer join then appends its unmatched rows after the matched ones of each
batch, and a sort that Spark dropped above it is lost, the symptom of #6787.
The first shape is a cache written by an outer join whose probe input is a
native operator over the sorted cache, such as a filter, a projection or the
semi join of an `IN` subquery. The planner does declare the order for that
first join, so the cache is sorted. But the serde counts an outer join as
declared only when its probe input is a scan in its own native plan, so an
outer join that reads the cache sends no order.
The second is an RDD checkpoint of a sorted relation
(`Dataset.localCheckpoint`). The checkpoint reports the order of the plan it
was taken from, and that plan is gone, so the serde treats the order as one it
cannot check.
### Steps to reproduce
With `spark.comet.exec.inMemoryCache.enabled=true`,
`spark.comet.batchSize=8`, `spark.sql.adaptive.enabled=false` and
`spark.sql.shuffle.partitions=2`, Parquet tables `big(k, v)` with 2000 rows
where `k` is `(i % 40) * 1000` and `v` is `i`, and `small(k, w)` with a row for
every even `i % 40`:
```scala
spark.table("big").repartition(2, $"k").sortWithinPartitions("k").cache()
.createOrReplaceTempView("cached")
spark.table("cached").count()
sql("""SELECT /*+ BROADCAST(b) */ a.k, a.v, b.w
|FROM (SELECT * FROM cached WHERE v >= 0) a LEFT JOIN small b ON a.k =
b.k""".stripMargin)
.cache().createOrReplaceTempView("joined")
spark.table("joined").count()
sql("""SELECT /*+ BROADCAST(c) */ j.k, j.v, c.w
|FROM joined j LEFT JOIN small c ON j.k = c.k""".stripMargin)
.sortWithinPartitions("k")
```
The plan has no sort, and the partitions are not sorted by `k`. Taking
`spark.table("cached").localCheckpoint()` in place of the second cache gives
the same result for the second shape.
### Expected behavior
Every partition sorted by `k`, as in Spark.
### Additional context
Both shapes behave as on main, so #6835 does not make them worse. The first
shape could be fixed by letting the serde follow the same filters, projections
and joins as the native walk, or by having the outer join report no order
unless it declared one, so that a cache it writes reports none either. The
second of these would also cover an outer join over an operator the planner
cannot walk through, a native sample for one: `SampleExec` does not report that
it keeps its input's order, so the planner declares nothing through it, and the
cache such a join writes reports an order the join did not keep.
--
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]
