viirya commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4288819658
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2885,6 +2887,56 @@ trait CometHashJoin {
val leftKeys = join.leftKeys.map(exprToProto(_, join.left.output))
val rightKeys = join.rightKeys.map(exprToProto(_, join.right.output))
+ // Spark reports the streamed side's order as the join's, and drops a sort
above the join
+ // that it satisfies. DataFusion's hash join keeps the unmatched probe
rows of a right outer
+ // join in place only when the probe input declares an order, and
otherwise moves them after
+ // the matched rows of each batch. Comet builds on the build side, so a
left outer join built
+ // on the right and a right outer join built on the left both run as that
right outer join,
+ // and an input sorted outside the native plan, a relation sorted before
it was cached for
+ // instance, declares no order to DataFusion. Send the order so the native
join declares it
+ // on the probe input; see SortedInputExec. Every other join type keeps
the probe order as
+ // it is, except the null-aware anti join: Comet builds that one on its
streamed side, so its
+ // output is not in the order Spark reports for it, and the planner
declares no order above
+ // it.
+ //
+ // The order Spark reports can be false to begin with. A Comet join that
may move its
+ // streamed rows reports it all the same: the null-aware anti join, a
right outer join like
+ // this one whose probe input declared no order, and the nested loop join,
which keeps the
+ // order of neither input. A cache reports the order of the plan that
wrote it, so the rows
+ // of such a join can reach this one through a cache, as a JVM input that
the native planner
+ // cannot look behind. Declaring the order there would let an operator
above rely on it, an
+ // aggregate in sorted mode for instance, which then splits groups. So
nothing is declared
+ // when such a join feeds the streamed side with no sort in between, or
when a leaf reports
+ // an order that cannot be checked; see mayReportFalseOrder.
+ //
+ // Declaring any order makes DataFusion keep the probe rows in place, so
the longest prefix
+ // of the order that Comet can express is enough. A key counts as
expressible when the native
+ // sort accepts its type, as `supportedSortType` decides for a sort: a
collated string is
+ // ordered by its collation in Spark but would be declared in raw byte
order. Nothing here can
+ // tell whether an operator above relies on the order, so a join whose
first key Comet cannot
+ // express stays in Spark.
+ val (streamed, _) = streamedSide(join.buildSide, join.left, join.right)
+ val streamedSortOrders =
+ if (runsAsRightOuterJoin(join.joinType, join.buildSide) &&
+ join.outputOrdering.nonEmpty && !mayReportFalseOrder(streamed)) {
+ val orders = join.outputOrdering.iterator
+ .takeWhile(order => supportedScalarSortElementType(order.dataType))
Review Comment:
Thanks, the collation check reads right now, and the cache-based array and
struct test covers it.
--
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]
