You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/23 11:56:29 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-7126-eefa4dddb394534f78f3cc63701a0a44d7dce149 created (now c56f6a0209)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7126-eefa4dddb394534f78f3cc63701a0a44d7dce149
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at c56f6a0209 fix: fall back when an outer hash join's streamed order cannot be checked (#7126)
This branch includes the following new commits:
new c56f6a0209 fix: fall back when an outer hash join's streamed order cannot be checked (#7126)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: fall back when an outer hash join's streamed order cannot be checked (#7126)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7126-eefa4dddb394534f78f3cc63701a0a44d7dce149
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit c56f6a0209dba0600103a37fac7ca9b006639857
Author: dustin <dw...@users.noreply.github.com>
AuthorDate: Fri Oct 23 10:17:58 2026 +0000
fix: fall back when an outer hash join's streamed order cannot be checked (#7126)
* fix: fall back when an outer hash join's streamed order cannot be checked
An outer hash join that runs as DataFusion's right outer join keeps its
unmatched probe rows in place only when the probe input declares an
order. When the streamed side's reported order could not be checked, the
serde declared none but still converted the join, so a sort Spark had
already dropped above it was lost. This happened over a sorted
localCheckpoint and over some caches written by an outer hash join.
The join now falls back to Spark when that order reaches it through an
input from the JVM or through a nested loop join. Over native operators
in the same plan it stays native, since DataFusion keeps the order it
can see there.
* test: cover a right outer join over a cache of a right outer join
The cache test now runs both shapes that run as DataFusion's right outer
join, LeftOuter built on the right and RightOuter built on the left, in
both cache formats. The nested loop join case notes that it only gives up
the native join for nothing until #6898 is fixed.
* fix: say that an outer hash join with an unchecked order stays in Spark
The fallback reason still described the old behavior, where the join
stayed native and declared no order. It now says the join stays in
Spark, and the tests share it through one constant. The compatibility
guide leads with when the join falls back and notes that a checkpoint
reaches a native join only with spark.comet.convert.rdd.enabled. The two
checkpoint tests share their setup.
* docs: keep only the comments that explain when an outer join's order is trusted
Drop comments that restate the code or a test's name, and shorten the
rest to why an order from the JVM or a nested loop join can't be used.
---
.../contributor-guide/adding_a_new_operator.md | 5 +-
.../user-guide/latest/compatibility/operators.md | 19 ++-
.../org/apache/spark/sql/comet/operators.scala | 58 +++++--
.../comet/exec/CometInMemoryCacheSuite.scala | 98 ++++++++++-
.../org/apache/comet/exec/CometJoinSuite.scala | 183 ++++++++++++++++++++-
5 files changed, 336 insertions(+), 27 deletions(-)
diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md
index e142aaa499..0c36921a94 100644
--- a/docs/source/contributor-guide/adding_a_new_operator.md
+++ b/docs/source/contributor-guide/adding_a_new_operator.md
@@ -372,7 +372,10 @@ ordering must keep it natively, and the native operator must say so by returning
`maintains_input_order` for that child, as DataFusion's filter and projection do. The planner
declares the order through an operator only when the operator says it keeps it. An operator that
cannot keep the order natively must not report it, or has to be added to `mayReportFalseOrder` in
-`operators.scala`, which makes the hash join serde send no order over it.
+`operators.scala`. The hash join serde then sends no order over it, and an outer hash join above it
+that has to keep that order falls back to Spark. Comet's hash joins are the exception within one
+native plan: an outer hash join over one of them stays native, even over one that declared no
+order, since DataFusion keeps the order it can see there.
A leaf that reports an ordering is where the order starts, so the planner has to know that the
native operator produces its rows in that order. It knows that of an input from the JVM and of a
diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 01d2a2ef67..8785874c6f 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -132,12 +132,19 @@ 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)).
+An outer join falls back to Spark when Comet cannot check the order Spark reports and that order
+reaches the join through an input from the JVM or through a nested loop join. Spark keeps the rows
+in the order they arrive. One such input is an RDD checkpoint. It reaches a native join only with
+`spark.comet.convert.rdd.enabled=true`, which defaults to `false`. Without that setting the
+checkpoint stays a Spark input and the join stays in Spark anyway. Another is a cache written by an
+outer hash join whose own probe input was not a scan
+([#6922](https://github.com/apache/datafusion-comet/issues/6922)). 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)).
+
+When the order Comet cannot check comes only from another native hash join in the same plan, the
+join does not fall back. It runs natively and declares no order, since DataFusion keeps the order
+it can see there.
## Window Functions
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index ee5c5358e7..da26c91f9c 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -2816,6 +2816,13 @@ case class CometSortAggregateExec(
this.copy(child = newChild)
}
+object CometHashJoin {
+
+ private[apache] val uncheckedStreamedOrderReason: String =
+ "the streamed side's sort order, which this outer join has to keep, cannot be checked, " +
+ "so the join stays in Spark"
+}
+
trait CometHashJoin {
// Only BroadcastHashJoinExec can be null-aware (NOT IN subqueries).
@@ -2943,7 +2950,8 @@ trait CometHashJoin {
// 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.
+ // an order that cannot be checked; see mayReportFalseOrder. When it comes from the JVM or a
+ // nested loop join, the join stays in Spark, as Spark may have dropped a sort above it.
//
// 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 that exprToProto takes counts as
@@ -2952,9 +2960,14 @@ 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, CometHashJoin.uncheckedStreamedOrderReason)
+ return None
+ }
val streamedSortOrders =
- if (runsAsRightOuterJoin(join.joinType, join.buildSide) &&
- join.outputOrdering.nonEmpty && !mayReportFalseOrder(streamed)) {
+ if (keepsStreamedOrder && !mayReportFalseOrder(streamed, countInPlan = true)) {
val orders = join.outputOrdering.iterator
.takeWhile(order => !hasNonDefaultStringCollation(order.dataType))
.map(exprToProto(_, streamed.output))
@@ -3007,11 +3020,16 @@ trait CometHashJoin {
// order, but the scan reads the final plan behind it once there is one. While CometExecRule
// converts a plan, a native plan's JVM inputs sit behind a CometScanWrapper, a leaf, so the
// walk looks through that too.
- private def mayReportFalseOrder(plan: SparkPlan): Boolean = plan match {
+ //
+ // With `countInPlan` off, a Comet hash join in the same native plan does not count: DataFusion
+ // sees its probe input's order itself. Past an input from the JVM, every join counts again.
+ private def mayReportFalseOrder(plan: SparkPlan, countInPlan: Boolean): Boolean = plan match {
case adaptive: AdaptiveSparkPlanExec if adaptive.isFinalPlan =>
- mayReportFalseOrder(adaptive.executedPlan)
+ mayReportFalseOrder(adaptive.executedPlan, countInPlan)
case _ if plan.outputOrdering.isEmpty => false
case _: SortExec | _: CometSortExec => false
+ // Its rows stay in order only over a one-row build side, and a native join above would move
+ // them. Over more build rows they are already out of order (#6898); drop this once fixed.
case _: CometBroadcastNestedLoopJoinExec => true
case join: CometBroadcastHashJoinExec =>
hashJoinMayReportFalseOrder(
@@ -3019,40 +3037,46 @@ trait CometHashJoin {
join.joinType,
join.buildSide,
join.left,
- join.right)
+ join.right,
+ countInPlan)
case join: CometHashJoinExec =>
hashJoinMayReportFalseOrder(
join.nativeOp,
join.joinType,
join.buildSide,
join.left,
- join.right)
+ join.right,
+ countInPlan)
case join: HashJoin =>
- mayReportFalseOrder(streamedSide(join.buildSide, join.left, join.right)._1)
- case wrapper: CometScanWrapper => mayReportFalseOrder(wrapper.originalPlan)
+ mayReportFalseOrder(streamedSide(join.buildSide, join.left, join.right)._1, countInPlan)
+ case wrapper: CometScanWrapper =>
+ mayReportFalseOrder(wrapper.originalPlan, countInPlan = true)
case scan: CometInMemoryTableScanExec =>
- mayReportFalseOrder(scan.originalPlan.relation.cachedPlan)
- case scan: InMemoryTableScanExec => mayReportFalseOrder(scan.relation.cachedPlan)
- case stage: QueryStageExec => mayReportFalseOrder(stage.plan)
+ mayReportFalseOrder(scan.originalPlan.relation.cachedPlan, countInPlan = true)
+ case scan: InMemoryTableScanExec =>
+ mayReportFalseOrder(scan.relation.cachedPlan, countInPlan = true)
+ case stage: QueryStageExec => mayReportFalseOrder(stage.plan, countInPlan)
// A scan reports an order that Spark already relies on.
case _: DataSourceScanExec | _: DataSourceV2ScanExecBase | _: RangeExec | _: CometLeafExec =>
false
case _ if plan.children.isEmpty => true
- case _ => plan.children.exists(mayReportFalseOrder)
+ case _ => plan.children.exists(mayReportFalseOrder(_, countInPlan))
}
// Whether a Comet hash join that reports an order may not keep it. Only the streamed side's
// order reaches the join's output, so a false order on the build side is of no concern, and an
- // outer join of the two shapes has to have declared the order on its probe input as well.
+ // outer join of the two shapes has to have declared the order on its probe input if counted.
private def hashJoinMayReportFalseOrder(
nativeOp: Operator,
joinType: JoinType,
buildSide: BuildSide,
left: SparkPlan,
- right: SparkPlan): Boolean = {
+ right: SparkPlan,
+ countInPlan: Boolean): Boolean = {
val (streamed, streamedIndex) = streamedSide(buildSide, left, right)
- mayReportFalseOrder(streamed) ||
- (runsAsRightOuterJoin(joinType, buildSide) && !declaresStreamedOrder(nativeOp, streamedIndex))
+ mayReportFalseOrder(streamed, countInPlan) ||
+ (countInPlan && runsAsRightOuterJoin(joinType, buildSide) &&
+ !declaresStreamedOrder(nativeOp, streamedIndex))
}
// Whether the native planner declares the streamed order on the probe input of the join that
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
index 3273b2af7b..b0edda4233 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
@@ -38,9 +38,9 @@ import org.apache.spark.CometDriverPlugin
import org.apache.spark.sql.{CometTestBase, DataFrame, Observation, QueryTest, Row}
import org.apache.spark.sql.catalyst.expressions.{And, Ascending, Attribute, AttributeReference, EqualTo, Expression, GreaterThan, GreaterThanOrEqual, LessThan, Literal}
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
-import org.apache.spark.sql.catalyst.plans.LeftOuter
+import org.apache.spark.sql.catalyst.plans.{LeftOuter, RightOuter}
import org.apache.spark.sql.columnar.{CachedBatch, SimpleMetricsCachedBatch}
-import org.apache.spark.sql.comet.{CometBroadcastHashJoinExec, CometHashAggregateExec, CometHashJoinExec, CometInMemoryTableScanExec, CometSortAggregateExec, CometSortExec, CometSortMergeJoinExec}
+import org.apache.spark.sql.comet.{CometBroadcastHashJoinExec, CometHashAggregateExec, CometHashJoin, CometHashJoinExec, CometInMemoryTableScanExec, CometSortAggregateExec, CometSortExec, CometSortMergeJoinExec}
import org.apache.spark.sql.comet.execution.arrow.{ArrowCachedBatchSerializer, CometCachedBatchHelper}
import org.apache.spark.sql.comet.execution.shuffle.CometCelebornShuffleManager
import org.apache.spark.sql.comet.util.Utils
@@ -3547,4 +3547,98 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
}
}
+
+ test("an outer hash join over a cache of an outer hash join over a native filter keeps order") {
+ // The filter keeps the serde from seeing that the first join declared its order, so the
+ // second join, reading that join's cache from the JVM, cannot check the order.
+ for {
+ rightOuter <- Seq(false, true)
+ nativeCache <- Seq(true, false)
+ adaptive <- Seq(false, true)
+ } {
+ val (joinType, buildSide) =
+ if (rightOuter) (RightOuter, BuildLeft) else (LeftOuter, BuildRight)
+ def outerJoin(streamed: String, small: String, on: String): String =
+ if (rightOuter) s"$small RIGHT JOIN $streamed ON $on"
+ else s"$streamed LEFT JOIN $small ON $on"
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ SQLConf.COLUMN_BATCH_SIZE.key -> "8",
+ CometConf.COMET_BATCH_SIZE.key -> "8",
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> nativeCache.toString,
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED.key -> (!nativeCache).toString) {
+ withParquetTable((0 until 2000).map(i => ((i % 40) * 1000, i)), "big") {
+ withParquetTable((0 until 40 by 2).map(m => (m * 1000, m)), "small") {
+ withTempView("cached", "joined") {
+ try {
+ spark
+ .table("big")
+ .toDF("k", "v")
+ .repartition(2, $"k")
+ .sortWithinPartitions("k")
+ .cache()
+ .createOrReplaceTempView("cached")
+ spark.table("cached").count()
+ sql("SELECT /*+ BROADCAST(b) */ a.k, a.v, b._2 AS w FROM " +
+ outerJoin("(SELECT * FROM cached WHERE v >= 0) a", "small b", "a.k = b._1"))
+ .cache()
+ .createOrReplaceTempView("joined")
+ spark.table("joined").count()
+ val cachedPlan = cachedPlanOf(spark.table("joined"))
+ val context = s"$joinType nativeCache=$nativeCache AQE=$adaptive"
+ assert(
+ find(cachedPlan) {
+ case join: CometBroadcastHashJoinExec =>
+ join.joinType == joinType && join.buildSide == buildSide &&
+ join.nativeOp.getHashJoin.getStreamedSortOrdersCount > 0
+ case _ => false
+ }.isDefined,
+ s"$context: the cache was not written by a native outer join that sent the " +
+ s"order:\n$cachedPlan")
+ assert(
+ partitionsInOrder(spark.table("joined"), keyOrder(Ascending)),
+ s"$context: the cached outer join rows are out of order")
+
+ def df: DataFrame = sql("SELECT /*+ BROADCAST(c) */ j.k, j.v, c._2 FROM " +
+ outerJoin("joined j", "small c", "j.k = c._1"))
+ .sortWithinPartitions("k")
+ val (_, plan) =
+ checkSparkAnswerAndFallbackReason(
+ df,
+ CometHashJoin.uncheckedStreamedOrderReason)
+ val planContext = s"$context\n$plan"
+ if (nativeCache) {
+ assert(
+ collect(plan) { case s: CometInMemoryTableScanExec => s }.size == 1,
+ planContext)
+ } else {
+ assert(
+ collect(plan) { case s: InMemoryTableScanExec => s }.size == 1,
+ planContext)
+ assert(
+ collect(plan) { case s: CometInMemoryTableScanExec => s }.isEmpty,
+ planContext)
+ }
+ assert(
+ collect(plan) { case j: BroadcastHashJoinExec =>
+ (j.joinType, j.buildSide)
+ } == Seq((joinType, buildSide)),
+ planContext)
+ assert(
+ collect(plan) { case j: CometBroadcastHashJoinExec => j }.isEmpty,
+ planContext)
+ assert(
+ collect(plan) { case s @ (_: CometSortExec | _: SortExec) => s }.isEmpty,
+ planContext)
+ assert(partitionsInOrder(df, keyOrder(Ascending)), planContext)
+ } finally {
+ spark.catalog.clearCache()
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
index 2d5a522825..b39d16f123 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.expressions.{And, Ascending, AttributeRefer
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
import org.apache.spark.sql.catalyst.plans.{Inner, LeftOuter, LeftSemi}
import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometRangeExec, CometSampleExec, CometSortExec, CometSortMergeJoinExec, CometSparkToColumnarExec, CometUnionExec, CometWindowExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoin, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometRangeExec, CometSampleExec, CometSortExec, CometSortMergeJoinExec, CometSparkToColumnarExec, CometUnionExec, CometWindowExec}
import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, SQLExecution, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, QueryStageExec}
import org.apache.spark.sql.execution.aggregate.SortAggregateExec
@@ -2712,6 +2712,127 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Takes the checkpoint with AQE off, as an adaptive plan reports no order.
+ private def withOrderedCheckpoint(f: DataFrame => Unit): Unit =
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_RANGE_ENABLED.key -> "true") {
+ f(
+ spark
+ .range(2000)
+ .selectExpr("id % 100 AS k", "id AS v")
+ .repartition(2, $"k")
+ .sortWithinPartitions("k", "v")
+ .localCheckpoint(eager = true))
+ }
+
+ test("outer hash joins over a checkpointed streamed side fall back and keep its order") {
+ // Spark drops the sort above the join for the checkpoint's order, which Comet cannot check.
+ withOrderedCheckpoint { streamed =>
+ val build = spark.range(10).selectExpr("id * 10 AS k", "id AS w")
+ val order =
+ (0 to 1).map(i => SortOrder(BoundReference(i, LongType, nullable = true), Ascending))
+ for (adaptive <- Seq(false, true))
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString) {
+ for {
+ (hint, sparkJoin) <- Seq(
+ "shuffle_hash" -> classOf[ShuffledHashJoinExec],
+ "broadcast" -> classOf[BroadcastHashJoinExec])
+ joinType <- Seq("left_outer", "right_outer")
+ } {
+ def df: DataFrame = {
+ val on = streamed("k") === build("k")
+ val joined = if (joinType == "left_outer") {
+ streamed.join(build.hint(hint), on, joinType)
+ } else {
+ build.hint(hint).join(streamed, on, joinType)
+ }
+ joined
+ .select(streamed("k"), streamed("v"), build("w"))
+ .sortWithinPartitions("k", "v")
+ }
+ val (_, plan) =
+ checkSparkAnswerAndFallbackReason(df, CometHashJoin.uncheckedStreamedOrderReason)
+ val context = s"$hint $joinType AQE=$adaptive\n$plan"
+ assert(collect(plan) { case j if j.getClass == sparkJoin => j }.size == 1, context)
+ assert(
+ collect(plan) {
+ case j: CometHashJoinExec => j
+ case j: CometBroadcastHashJoinExec => j
+ }.isEmpty,
+ context)
+ assert(
+ collect(plan) { case s @ (_: CometSortExec | _: SortExec) => s }.isEmpty,
+ context)
+ assert(partitionsInOrder(df, order: _*), context)
+ }
+ }
+ }
+ }
+
+ test("an outer hash join over a native inner join over a checkpoint falls back") {
+ // The order walk reaches the checkpoint through the inner join's streamed side.
+ withOrderedCheckpoint { streamed =>
+ val d1 = spark.range(100).where("id % 4 != 3").selectExpr("id AS k", "id AS a")
+ val d2 = spark.range(10).selectExpr("id * 10 AS k", "id AS w")
+ val order =
+ (0 to 1).map(i => SortOrder(BoundReference(i, LongType, nullable = true), Ascending))
+ def df: DataFrame = streamed
+ .join(d1.hint("broadcast"), streamed("k") === d1("k"), "inner")
+ .join(d2.hint("broadcast"), streamed("k") === d2("k"), "left_outer")
+ .select(streamed("k"), streamed("v"), d1("a"), d2("w"))
+ .sortWithinPartitions("k", "v")
+ val (_, plan) =
+ checkSparkAnswerAndFallbackReason(df, CometHashJoin.uncheckedStreamedOrderReason)
+ val inner = collect(plan) { case j: CometBroadcastHashJoinExec => j.joinType }
+ assert(inner == Seq(Inner), plan)
+ val outer = collect(plan) { case j: BroadcastHashJoinExec => j.joinType }
+ assert(outer == Seq(LeftOuter), plan)
+ assert(collect(plan) { case s @ (_: CometSortExec | _: SortExec) => s }.isEmpty, plan)
+ assert(partitionsInOrder(df, order: _*), plan)
+ }
+ }
+
+ test("chained outer hash joins over a native window keep its order natively") {
+ // The window and both joins share one native plan, where DataFusion sees the order itself.
+ withParquetTable((0 until 2000).map(i => (i % 40, i)), "t") {
+ withParquetTable((0 until 40 by 2).map(k => (k, k * 10)), "d1") {
+ withParquetTable((0 until 40 by 3).map(k => (k, k * 100)), "d2") {
+ val order = (0 to 1).map { i =>
+ SortOrder(BoundReference(i, IntegerType, nullable = true), Ascending)
+ }
+ for (adaptive <- Seq(false, true)) {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString) {
+ def df: DataFrame = sql("""SELECT w.k, w.v, w.rn, d1._2 AS x, d2._2 AS y
+ |FROM (SELECT k, v, row_number() OVER (PARTITION BY k ORDER BY v) AS rn
+ | FROM (SELECT _1 AS k, _2 AS v FROM t)) w
+ |LEFT JOIN d1 ON w.k = d1._1
+ |LEFT JOIN d2 ON w.k = d2._1""".stripMargin)
+ .sortWithinPartitions("k", "v")
+ val (_, plan) = checkSparkAnswer(df)
+ val context = s"AQE=$adaptive\n$plan"
+ assert(new ExtendedExplainInfo().getFallbackReasons(plan).isEmpty, context)
+ val joins = collect(plan) { case j: CometBroadcastHashJoinExec => j }
+ assert(joins.size == 2 && joins.forall(_.joinType == LeftOuter), context)
+ assert(collect(plan) { case w: CometWindowExec => w }.size == 1, context)
+ assert(
+ collect(plan) {
+ case s @ (_: CometSortExec | _: SortExec)
+ if s.find(_.isInstanceOf[CometBroadcastHashJoinExec]).isDefined =>
+ s
+ }.isEmpty,
+ context)
+ assert(partitionsInOrder(df, order: _*), context)
+ }
+ }
+ }
+ }
+ }
+ }
+
// Compares the rows of a fresh `df` with the ones Spark returns for it, as a multiset unless
// `isSorted`. The rows of a key tie under a sort, so their order within it can differ between
// the two.
@@ -3186,6 +3307,66 @@ class CometJoinSuite extends CometTestBase {
}
}
+ test("chained outer hash joins over a native filter over a sorted cache stay native") {
+ // DataFusion sees the first join's order, in the same native plan, so the second sends none.
+ withSortedCache() {
+ def df: DataFrame = sql("""SELECT /*+ BROADCAST(b), BROADCAST(c) */ a.k, a.v, b._2, c._2
+ |FROM (SELECT * FROM cached WHERE v >= 0) a
+ |LEFT JOIN small b ON a.k = b._1
+ |LEFT JOIN small c ON a.k = c._1""".stripMargin)
+ .sortWithinPartitions("k")
+ val (_, plan) = checkSparkAnswer(df)
+ // Spark's cache format adds a reason of its own to the cache scan.
+ assert(
+ !new ExtendedExplainInfo().getFallbackReasons(plan).exists(_.contains("sort order")),
+ plan)
+ val joins = collect(plan) { case j: CometBroadcastHashJoinExec => j }
+ assert(joins.size == 2 && joins.forall(_.joinType == LeftOuter), plan)
+ val (first, second) =
+ joins.partition(_.left.find(_.isInstanceOf[CometBroadcastHashJoinExec]).isEmpty)
+ assert(first.size == 1 && second.size == 1, plan)
+ assert(collect(first.head.left) { case f: CometFilterExec => f }.size == 1, plan)
+ assert(first.head.nativeOp.getHashJoin.getStreamedSortOrdersCount > 0, plan)
+ assert(second.head.nativeOp.getHashJoin.getStreamedSortOrdersCount == 0, plan)
+ assert(collect(plan) { case s @ (_: CometSortExec | _: SortExec) => s }.isEmpty, plan)
+ assert(partitionsInOrder(df, keyOrder(Ascending)), plan)
+ }
+ }
+
+ test("an outer hash join over a native nested loop join falls back") {
+ // Over two build rows the nested loop join emits rows out of order (#6898), so they are
+ // compared as a multiset. Over one build row they stay in order, and Spark's join keeps it.
+ withSortedCache() {
+ withParquetTable(Seq(Tuple1(0), Tuple1(1)), "two") {
+ withParquetTable(Seq(Tuple1(1)), "one") {
+ def query(nlBuild: String): String =
+ s"""SELECT /*+ BROADCAST(n), BROADCAST(b) */ a.k, a.v, n._1 AS z, b._2 AS w
+ |FROM cached a JOIN $nlBuild n ON a.v % 2 <= n._1
+ |LEFT JOIN small b ON a.k = b._1""".stripMargin
+ val reason = CometHashJoin.uncheckedStreamedOrderReason
+ def checkFallback(df: => DataFrame): SparkPlan = {
+ sameRowsAsSpark(df)
+ val plan = df.queryExecution.executedPlan
+ val reasons = new ExtendedExplainInfo().getFallbackReasons(plan)
+ assert(reasons.exists(_.contains(reason)), plan)
+ val nestedLoopJoins = collect(plan) { case j: CometBroadcastNestedLoopJoinExec => j }
+ assert(nestedLoopJoins.size == 1, plan)
+ val outer = collect(plan) { case j: BroadcastHashJoinExec => j.joinType }
+ assert(outer == Seq(LeftOuter), plan)
+ plan
+ }
+
+ checkFallback(sql(query("two")))
+
+ def ordered: DataFrame = sql(query("one")).sortWithinPartitions("k")
+ val plan = checkFallback(ordered)
+ assert(collect(plan) { case s @ (_: CometSortExec | _: SortExec) => s }.isEmpty, plan)
+ assert(partitionsInOrder(ordered, keyOrder(Ascending)), plan)
+ }
+ }
+ }
+ }
+
test("an outer hash join declares the streamed order over a semi join") {
// `k IN (SELECT ...)` runs as a left semi join, which emits its probe rows in ascending
// order whether or not the probe input declares an order. The outer join above it therefore
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org