comphead commented on code in PR #7126:
URL: https://github.com/apache/datafusion-comet/pull/7126#discussion_r4332014426
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2952,9 +2954,17 @@ trait CometHashJoin {
// check `supportedSortType` makes for a sort. 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 keepsStreamedOrder =
+ runsAsRightOuterJoin(join.joinType, join.buildSide) &&
join.outputOrdering.nonEmpty
+ if (keepsStreamedOrder && mayReportFalseOrder(streamed, countInPlan =
false)) {
+ withFallbackReason(
+ join,
+ "the streamed side's sort order, which this outer join has to keep,
cannot be " +
+ "checked, so it is not declared natively")
+ return None
Review Comment:
This reason is shown in the explain output for a join that now stays in
Spark, but "so it is not declared natively" describes the old behavior, where
the join stayed native and sent no order. Could it say what happens now, for
example "cannot be checked, so the join stays in Spark"? The same literal is
repeated in four tests (`CometInMemoryCacheSuite` and three places in
`CometJoinSuite`), so a shared constant would keep them in step.
##########
docs/source/user-guide/latest/compatibility/operators.md:
##########
@@ -129,12 +129,15 @@ express falls back to Spark and records a fallback reason
(see
[Understanding Comet Plans](../understanding-comet-plans.md)), because Comet
cannot tell whether
an operator above the join relies on the order.
-Where the order Spark reports might not hold, Comet declares nothing and an
outer join can return
-its unmatched rows out of order. The output of a nested loop join
-([#6898](https://github.com/apache/datafusion-comet/issues/6898)) is one such
case, and so is a
-cache or checkpoint that one wrote. Comet also declares nothing when it cannot
follow the
-order back to its source, as with an RDD checkpoint, and over some caches
written by an outer hash
-join whose order does hold
([#6922](https://github.com/apache/datafusion-comet/issues/6922)).
+Where Comet cannot check the order Spark reports, it declares nothing. If that
order reaches the
+outer join through an input from the JVM, such as an RDD checkpoint or some
caches written by an
+outer hash join whose own probe input was not a scan
+([#6922](https://github.com/apache/datafusion-comet/issues/6922)), or through
a nested
+loop join, the join falls back to Spark, which keeps the rows in the order
they arrive. The output
+of a nested loop join can already be out of order when it reaches the join
+([#6898](https://github.com/apache/datafusion-comet/issues/6898)). A join over
other native
+operators in the same plan does not fall back for its order, since DataFusion
keeps the order it
+can see there.
Review Comment:
This paragraph opens with "it declares nothing" and only later says the join
falls back, so a reader cannot tell when "declares nothing" means native and
when it means a fallback. Leading with the fallback rule and then giving the
in-plan exception would be clearer. The RDD checkpoint example also reaches a
native join only with `spark.comet.convert.rdd.enabled=true`, which defaults to
`false`. Without it the checkpoint is a plain Spark child and the join stays in
Spark with or without this change. The tests set the config, so the doc could
say so.
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -2712,6 +2712,141 @@ class CometJoinSuite extends CometTestBase {
}
}
+ test("outer hash joins over a checkpointed streamed side fall back and keep
its order") {
Review Comment:
`outer hash joins over a checkpointed streamed side fall back and keep its
order` and `an outer hash join over a native inner join over a checkpoint falls
back` (line 2775) repeat the same conf block and the same
`range(2000)...localCheckpoint` setup. A small `withOrderedCheckpoint {
streamed => ... }` helper would remove about 20 lines.
--
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]
