sunchao commented on code in PR #7256:
URL: https://github.com/apache/datafusion-comet/pull/7256#discussion_r4419357111
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -81,10 +81,55 @@ object CometWindowExec extends
CometOperatorSerde[WindowExec] {
if (windowExprProto.forall(_.isDefined) &&
partitionExprs.forall(_.isDefined)
&& sortOrders.forall(_.isDefined)) {
+ // Spark plans no sort below the window when its input reports the order
the window needs,
+ // a sorted cache for instance, but rows from the JVM arrive through a
ScanExec that
+ // declares none, and the native window rejects them. Send the order for
the planner to
+ // declare (see SortedInputExec), unless a native sort below declares it
or the reported
+ // order may be false. A key Comet cannot express or a collated string
key, which the
+ // native side would order by raw bytes, sends the window back to Spark.
+ val requiredOrdering = op.requiredChildOrdering.head
+ val inputSortOrders =
+ if (requiredOrdering.nonEmpty && !op.child.isInstanceOf[CometSortExec]
&&
+ SortOrder.orderingSatisfies(op.child.outputOrdering,
requiredOrdering) &&
+ !CometHashJoinExec.mayReportFalseInputOrder(op.child)) {
+ val orders = requiredOrdering.map { order =>
+ if (hasNonDefaultStringCollation(order.dataType)) None
+ else exprToProto(order, op.child.output)
+ }
+ if (!orders.forall(_.isDefined)) {
+ withFallbackReason(
+ op,
+ "the order of the window's input, which Spark relies on, cannot
be expressed " +
+ "natively")
+ return None
+ }
+ // The keys the child reports past the required ones hold too, and
this window reports
+ // them, so a window above in the same native plan may rely on them.
Send them up to the
+ // first one that cannot be expressed, in a scope of their own, so
that a key left out
+ // records no fallback reason.
+ val rest = op.child.outputOrdering
+ .drop(requiredOrdering.length)
+ .iterator
+ .map { order =>
+ if (hasNonDefaultStringCollation(order.dataType) ||
+ !order.references.subsetOf(op.child.outputSet)) {
Review Comment:
There is also a projection-sensitive version of the alias case. On a cache
sorted by `(k, x)`, expose `x` twice as `a` and `b`, then run a window
partitioned by `k` under one partitioned by `(k, b)`. With both aliases in the
scan output, the lower window declares `(k, b)` and Comet returns all 24 rows.
When the query drops unused `a`, Spark still reports `b` as the suffix key and
lists `a` in `sameOrderExpressions`. `order.references` includes both names,
although only `b` is serialized. This check therefore drops the suffix, the
lower window declares just `k`, and the upper window fails with two partition
keys required and only one ordered. Spark 4.1.3 returned all 24 rows in both
projections.
--
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]
