comphead commented on code in PR #7256:
URL: https://github.com/apache/datafusion-comet/pull/7256#discussion_r4419580056
##########
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:
Thanks. The check now looks at `order.child.references`, since only the
child is serialized. Fixed in dc94ccf3d3. With `a` pruned, the lower window
declares `(k, b)` again.
The second query in `stacked native windows over a sorted cache with aliases
of a sort key` is this shape. A cache sorted by `(k, x)` exposes `x` as `a` and
`b`, the query drops `a`, and `sum(v) OVER (PARTITION BY k)` runs under `sum(s)
OVER (PARTITION BY k, b)`. It checks the rows, that both windows send `k, b`,
and that no sort is planned.
--
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]
