dwsmith1983 opened a new issue, #7223:
URL: https://github.com/apache/datafusion-comet/issues/7223
### Describe the bug
Spark drops the sort below a window when its input already reports the order
the window needs.
A relation sorted before it was cached reports that order through
`InMemoryTableScanExec`, so a
window partitioned and ordered by the cache's sort keys gets no sort of its
own. When Comet reads
the cache from the JVM (`spark.comet.convert.inMemoryCache.enabled=true` with
`spark.comet.exec.inMemoryCache.enabled=false`), the rows enter the native
plan through a
`ScanExec`, which declares no order to DataFusion. The native
`BoundedWindowAggExec` runs in
`InputOrderMode::Sorted` (`native/core/src/execution/planner/window.rs`),
checks its input's
declared order and fails the task:
```
org.apache.comet.CometNativeException: Expects PARTITION BY expression to be
ordered
```
The rows are in fact sorted. Only the declaration is missing.
With #6785, an inner hash join between the cache and the window no longer
hits this: the join
sends the streamed side's order to the native planner, which declares it on
the probe input. A
window directly over the cache has no such operator in between. A native
window directly over a
sorted bucketed table read by the native scan (one file per bucket, with
`spark.sql.legacy.bucketedTableScan.outputOrdering=true`) fails on main with
the same error.
### Steps to reproduce
On Spark 4.1.3:
```scala
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.shuffle.partitions", "2")
spark.conf.set("spark.comet.exec.range.enabled", "true")
spark.conf.set("spark.comet.exec.inMemoryCache.enabled", "false")
spark.conf.set("spark.comet.convert.inMemoryCache.enabled", "true")
val cached = spark
.range(2000)
.selectExpr("id % 20 AS k", "id AS v")
.repartition(2, col("k"))
.sortWithinPartitions("k", "v")
.cache()
cached.count()
cached.selectExpr("k", "v", "dense_rank() OVER (PARTITION BY k ORDER BY v)
AS r").collect()
```
### Expected behavior
The 2,000 ranked rows Spark returns.
### Additional context
One fix is to carry the order Spark reports for the window's input to the
native side, for
example an `input_sort_orders` field on the `Window` proto that the planner
declares on the
window's input with `with_declared_order`, guarded the same way
`CometHashJoin` guards
`streamedSortOrders` (`mayReportFalseOrder`). Another is to keep a sort
below the native window
when its input reads from the JVM and reports an order.
Found in #6785
(https://github.com/apache/datafusion-comet/pull/6785#discussion_r4368087548).
--
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]
