viirya commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4286263738
##########
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:
`supportedScalarSortElementType` rejects every array and struct, so a join
whose streamed order starts with one now stays in Spark. Before this commit
those keys went through `exprToProto`, which takes them: `serializeDataType`
covers arrays and structs, and `CometSortOrder` only declines a nested key
under a non-default null order. On `branch-1.1` the join runs natively whatever
its streamed order. One way to hit it is `sortWithinPartitions` on an array
column, which `supportedSortType` lets `CometSortExec` run natively, followed
by a broadcast left outer join. The probe input then already declares the
order, so the join needs nothing from the serde, but it now falls back.
The case this check is for is the collated string. Would
`hasNonDefaultStringCollation(order.dataType)` cover it here? That is the check
`supportedSortType` makes, and it also catches a collated string inside a
struct or array, while the nested null-order case stays with `CometSortOrder`
as before. The comment on the new key-type test says struct, array and map keys
are out of scope because `exprToProto` rejects them, which isn't the case. An
array key in that test would pin this down.
--
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]
