From 6728b22c64c9015c53238ef68d18467f7a7c3f5b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 12 Oct 2026 12:24:30 -0600 Subject: [PATCH 1/2] feat: run joins of type Cross natively Spark gives a join the type Cross when the query asks for a cross join explicitly, with CROSS JOIN in SQL or Dataset.crossJoin, and its join operators run Cross exactly as they run Inner. Comet's sort-merge, hash and nested loop join serdes matched only Inner, so these joins fell back with "Unsupported join type Cross". Map InnerLike to the native Inner join type in all three. Spark's SPARK-26138 test takes LIMIT 3 of a cross join with no ORDER BY. That join now runs as a native nested loop join, which emits rows in a different order than Spark, so ignore the test under Comet in every Spark diff. Closes #6903. --- dev/diffs/3.4.3.diff | 12 +++- dev/diffs/3.5.9.diff | 12 +++- dev/diffs/4.0.4.diff | 14 ++++- dev/diffs/4.1.3.diff | 14 ++++- dev/diffs/4.2.0.diff | 14 ++++- .../apache/spark/sql/comet/operators.scala | 14 ++--- .../apache/comet/exec/CometJoinSuite.scala | 58 ++++++++++++++++++- 7 files changed, 122 insertions(+), 16 deletions(-) diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff index 1caa75a571a..829e9fb95c5 100644 --- a/dev/diffs/3.4.3.diff +++ b/dev/diffs/3.4.3.diff @@ -1047,7 +1047,7 @@ index b5b34922694..a72403780c4 100644 protected val baseResourcePath = { // use the same way as `SQLQueryTestSuite` to get the resource path diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala -index 525d97e4998..aded8906d75 100644 +index 525d97e4998..2e96cd4aa73 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala @@ -1508,7 +1508,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark @@ -1087,6 +1087,16 @@ index 525d97e4998..aded8906d75 100644 val sc = spark.sparkContext val hiveVersion = "2.3.9" // transitive=false, only download specified jar +@@ -3989,7 +3998,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark + } + } + +- test("SPARK-26138 Pushdown limit through InnerLike when condition is empty") { ++ test("SPARK-26138 Pushdown limit through InnerLike when condition is empty", ++ IgnoreComet("Comet's nested loop join emits rows in a different order than Spark")) { + withTable("t1", "t2") { + spark.range(5).repartition(1).write.saveAsTable("t1") + spark.range(5).repartition(1).write.saveAsTable("t2") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala index 2dabcf01be7..8fcec0d1ce4 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala diff --git a/dev/diffs/3.5.9.diff b/dev/diffs/3.5.9.diff index 67e59a4a323..7b794d72d9e 100644 --- a/dev/diffs/3.5.9.diff +++ b/dev/diffs/3.5.9.diff @@ -1218,7 +1218,7 @@ index c26757c9cff..d55775f09d7 100644 protected val baseResourcePath = { // use the same way as `SQLQueryTestSuite` to get the resource path diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala -index 3cf2bfd17ab..11d3ca4d5a9 100644 +index 3cf2bfd17ab..0230a8e0b24 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala @@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, NestedCo @@ -1274,6 +1274,16 @@ index 3cf2bfd17ab..11d3ca4d5a9 100644 val sc = spark.sparkContext val hiveVersion = "2.3.9" // transitive=false, only download specified jar +@@ -4019,7 +4030,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark + } + } + +- test("SPARK-26138 Pushdown limit through InnerLike when condition is empty") { ++ test("SPARK-26138 Pushdown limit through InnerLike when condition is empty", ++ IgnoreComet("Comet's nested loop join emits rows in a different order than Spark")) { + withTable("t1", "t2") { + spark.range(5).repartition(1).write.saveAsTable("t1") + spark.range(5).repartition(1).write.saveAsTable("t2") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala index 71af1fd69c3..81a04c93c9c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff index 83a19b27250..f657408c885 100644 --- a/dev/diffs/4.0.4.diff +++ b/dev/diffs/4.0.4.diff @@ -1388,7 +1388,7 @@ index ad424b3a7cc..4ece0117a34 100644 protected val baseResourcePath = { // use the same way as `SQLQueryTestSuite` to get the resource path diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala -index f294ff81021..f019c6de8e9 100644 +index f294ff81021..3a019cb3b99 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala @@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, NestedCo @@ -1434,7 +1434,17 @@ index f294ff81021..f019c6de8e9 100644 } assert(inMemoryTableScan.size == 2) checkAnswer(queryDf, Row(0, 1) :: Row(1, 2) :: Row(2, 3) :: Row(3, 4) :: Row(4, 5) :: Nil) -@@ -4744,7 +4754,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark +@@ -4009,7 +4019,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark + } + } + +- test("SPARK-26138 Pushdown limit through InnerLike when condition is empty") { ++ test("SPARK-26138 Pushdown limit through InnerLike when condition is empty", ++ IgnoreComet("Comet's nested loop join emits rows in a different order than Spark")) { + withTable("t1", "t2") { + spark.range(5).repartition(1).write.saveAsTable("t1") + spark.range(5).repartition(1).write.saveAsTable("t2") +@@ -4744,7 +4755,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark ) } diff --git a/dev/diffs/4.1.3.diff b/dev/diffs/4.1.3.diff index c1714a7357f..1f1e168b6f4 100644 --- a/dev/diffs/4.1.3.diff +++ b/dev/diffs/4.1.3.diff @@ -1546,7 +1546,7 @@ index cb9d0909554..084d6515e8b 100644 } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala -index 74cdee49e55..da1e9bd966a 100644 +index 74cdee49e55..54dae1aeb77 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala @@ -36,6 +36,7 @@ import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, NestedCo @@ -1592,7 +1592,17 @@ index 74cdee49e55..da1e9bd966a 100644 } assert(inMemoryTableScan.size == 2) checkAnswer(queryDf, Row(0, 1) :: Row(1, 2) :: Row(2, 3) :: Row(3, 4) :: Row(4, 5) :: Nil) -@@ -4741,7 +4751,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark +@@ -4006,7 +4016,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark + } + } + +- test("SPARK-26138 Pushdown limit through InnerLike when condition is empty") { ++ test("SPARK-26138 Pushdown limit through InnerLike when condition is empty", ++ IgnoreComet("Comet's nested loop join emits rows in a different order than Spark")) { + withTable("t1", "t2") { + spark.range(5).repartition(1).write.saveAsTable("t1") + spark.range(5).repartition(1).write.saveAsTable("t2") +@@ -4741,7 +4752,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark ) } diff --git a/dev/diffs/4.2.0.diff b/dev/diffs/4.2.0.diff index 549799d941a..998e09b2362 100644 --- a/dev/diffs/4.2.0.diff +++ b/dev/diffs/4.2.0.diff @@ -1595,7 +1595,7 @@ index 291aa7cab72..7783c37683e 100644 super.test(testName, testTags: _*) { withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala -index da6f6aca2ad..64c981ad3d7 100644 +index da6f6aca2ad..d178253dc98 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala @@ -40,6 +40,7 @@ import org.apache.spark.sql.catalyst.parser.ParseException @@ -1641,7 +1641,17 @@ index da6f6aca2ad..64c981ad3d7 100644 } assert(inMemoryTableScan.size == 2) checkAnswer(queryDf, Row(0, 1) :: Row(1, 2) :: Row(2, 3) :: Row(3, 4) :: Row(4, 5) :: Nil) -@@ -4755,7 +4765,8 @@ class SQLQuerySuite extends SharedSparkSession with AdaptiveSparkPlanHelper +@@ -4020,7 +4030,8 @@ class SQLQuerySuite extends SharedSparkSession with AdaptiveSparkPlanHelper + } + } + +- test("SPARK-26138 Pushdown limit through InnerLike when condition is empty") { ++ test("SPARK-26138 Pushdown limit through InnerLike when condition is empty", ++ IgnoreComet("Comet's nested loop join emits rows in a different order than Spark")) { + withTable("t1", "t2") { + spark.range(5).repartition(1).write.saveAsTable("t1") + spark.range(5).repartition(1).write.saveAsTable("t2") +@@ -4755,7 +4766,8 @@ class SQLQuerySuite extends SharedSparkSession with AdaptiveSparkPlanHelper ) } 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 de354502c59..c7c1cf62f7b 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 @@ -2843,7 +2843,8 @@ trait CometHashJoin { val joinType = { import OperatorOuterClass.JoinType join.joinType match { - case Inner => JoinType.Inner + // Cross, the type of an explicit cross join, runs as an inner join in Spark. + case _: InnerLike => JoinType.Inner case LeftOuter => JoinType.LeftOuter case RightOuter => JoinType.RightOuter case FullOuter => JoinType.FullOuter @@ -2876,7 +2877,6 @@ trait CometHashJoin { s"${CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key}=true to enable it") return None case _ => - // Spark doesn't support other join types withFallbackReason(join, s"Unsupported join type ${join.joinType}") return None } @@ -2973,7 +2973,7 @@ object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNest override def getSupportLevel(op: BroadcastNestedLoopJoinExec): SupportLevel = (op.joinType, op.buildSide) match { - case (Inner, _) | (LeftOuter, BuildRight) | (RightOuter, BuildLeft) | + case (_: InnerLike, _) | (LeftOuter, BuildRight) | (RightOuter, BuildLeft) | (LeftSemi, BuildRight) | (LeftAnti, BuildRight) => Compatible(None) case (LeftOuter | RightOuter | FullOuter | LeftSemi | LeftAnti, _) => @@ -3010,14 +3010,14 @@ object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNest val joinType = { import OperatorOuterClass.JoinType join match { - case Inner => JoinType.Inner + // Cross, the type of an explicit cross join, runs as an inner join in Spark. + case _: InnerLike => JoinType.Inner case LeftOuter => JoinType.LeftOuter case RightOuter => JoinType.RightOuter case FullOuter => JoinType.FullOuter case LeftSemi => JoinType.LeftSemi case LeftAnti => JoinType.LeftAnti case _ => - // Spark doesn't support other join types withFallbackReason(op, s"Unsupported join type $join") return None } @@ -3394,7 +3394,8 @@ object CometSortMergeJoinExec extends CometOperatorSerde[SortMergeJoinExec] { val joinType = { import OperatorOuterClass.JoinType join.joinType match { - case Inner => JoinType.Inner + // Cross, the type of an explicit cross join, runs as an inner join in Spark. + case _: InnerLike => JoinType.Inner case LeftOuter => JoinType.LeftOuter case RightOuter => JoinType.RightOuter case FullOuter => JoinType.FullOuter @@ -3404,7 +3405,6 @@ object CometSortMergeJoinExec extends CometOperatorSerde[SortMergeJoinExec] { // before emitting, risking OOM on large equal-key groups. Tracked in // https://github.com/apache/datafusion-comet/issues/6514. case _ => - // Spark doesn't support other join types withFallbackReason(join, s"Unsupported join type ${join.joinType}") return None } 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 36e7c5c4f4a..fb346003af9 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala @@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, DynamicPruningExpression, IsNotNull} import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight} -import org.apache.spark.sql.catalyst.plans.Inner +import org.apache.spark.sql.catalyst.plans.{Cross, Inner} import org.apache.spark.sql.catalyst.plans.logical.Join import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec} import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, WholeStageCodegenExec} @@ -1541,6 +1541,37 @@ class CometJoinSuite extends CometTestBase { } } + // An explicit cross join has the join type Cross, which Spark's join operators run as Inner. + // The optimizer moves an equi-key filter into the join condition and keeps Cross, so each hint + // plans an equi-join of type Cross. + for ((hint, joinClass) <- Seq( + "SHUFFLE_HASH" -> classOf[CometHashJoinExec], + "MERGE" -> classOf[CometSortMergeJoinExec], + "BROADCAST" -> classOf[CometBroadcastHashJoinExec])) { + test(s"CROSS JOIN with an equi-key filter runs natively: $hint") { + withSQLConf( + SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + withParquetTable((0 until 100).map(i => (i, i % 5)), "tbl_a") { + withParquetTable((0 until 50).map(i => (i % 10, i + 2)), "tbl_b") { + for (residual <- Seq("", " AND tbl_a._1 > tbl_b._2")) { + val df = sql( + s"SELECT /*+ $hint(tbl_b) */ * FROM tbl_a CROSS JOIN tbl_b " + + s"WHERE tbl_a._2 = tbl_b._1$residual") + val (_, cometPlan) = checkSparkAnswerAndOperator(df, Seq(joinClass)) + val joinTypes = collect(cometPlan) { + case j: CometHashJoinExec => j.joinType + case j: CometSortMergeJoinExec => j.joinType + case j: CometBroadcastHashJoinExec => j.joinType + } + assert(joinTypes == Seq(Cross), cometPlan) + } + } + } + } + } + } + test("full outer join") { withTempView("`left`", "`right`", "allNulls") { allNulls.createOrReplaceTempView("allNulls") @@ -1930,6 +1961,31 @@ class CometJoinSuite extends CometTestBase { } } + test("BroadcastNestedLoopJoin of join type Cross runs natively") { + withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + val left: Seq[(Integer, Int)] = + (0 until 5).map(i => ((i: Integer), i * 10)) ++ Seq[(Integer, Int)]((null, 99)) + val right: Seq[(Integer, String)] = + (0 until 4).map(i => ((i: Integer), s"v_$i")) ++ Seq[(Integer, String)]((null, "v_null")) + withParquetTable(left, "tbl_a") { + withParquetTable(right, "tbl_b") { + // The tests above join with FROM tbl_a, tbl_b, which gives an Inner join. CROSS JOIN + // gives Cross, which can broadcast either side, as Inner can. + for (build <- Seq("tbl_a", "tbl_b"); filter <- Seq("", " WHERE tbl_a._1 > tbl_b._1")) { + val df = sql( + s"SELECT /*+ BROADCAST($build) */ tbl_a._1, tbl_b._2 " + + s"FROM tbl_a CROSS JOIN tbl_b$filter") + val (_, cometPlan) = checkSparkAnswerAndOperator( + df, + Seq(classOf[CometBroadcastExchangeExec], classOf[CometBroadcastNestedLoopJoinExec])) + val joins = collect(cometPlan) { case j: CometBroadcastNestedLoopJoinExec => j } + assert(joins.map(_.joinType) == Seq(Cross), cometPlan) + } + } + } + } + } + test("BroadcastNestedLoopJoin LEFT OUTER without condition") { withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { val left: Seq[(Integer, Int)] = From 98244b0dd076fbdb0ccd346374d9bfa1ab3e86df Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 12 Oct 2026 13:39:04 -0600 Subject: [PATCH 2/2] test: add a benchmark for joins of type Cross Times Spark, Comet with the join left in Spark, and Comet for sort-merge, shuffled hash, broadcast hash and broadcast nested loop joins of type Cross, at 10 and 100 matches per probe row. Before timing, each case checks that every arm planned the join it claims, that the Comet arm is fully native, and that the arms return the same sum. --- .../benchmark/CometCrossJoinBenchmark.scala | 241 ++++++++++++++++++ 1 file changed, 241 insertions(+) create mode 100644 spark/src/test/scala/org/apache/spark/sql/benchmark/CometCrossJoinBenchmark.scala diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCrossJoinBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCrossJoinBenchmark.scala new file mode 100644 index 00000000000..36459877eab --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometCrossJoinBenchmark.scala @@ -0,0 +1,241 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.benchmark + +import org.apache.spark.SparkConf +import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.{DataFrame, SparkSession} +import org.apache.spark.sql.catalyst.plans.Cross +import org.apache.spark.sql.comet.{CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometExec, CometHashJoinExec, CometPlan, CometSortMergeJoinExec} +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.adaptive.{AQEShuffleReadExec, QueryStageExec} +import org.apache.spark.sql.execution.joins.{BaseJoinExec, BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec} +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.{CometConf, CometSparkSessionExtensions, ConfigEntry} + +/** + * Benchmark to measure performance of joins of type Cross, which Spark plans for an explicit + * `CROSS JOIN` or `Dataset.crossJoin`. A filter on an equi key turns the cross join into a + * sort-merge, shuffled hash or broadcast hash join of type Cross, and a broadcast side with no + * equi key gives a broadcast nested loop join of type Cross. + * + * Each case runs Spark, Comet with the join left in Spark, which is how Comet ran these joins + * before it converted the join type Cross, and Comet. To run this benchmark: + * {{{ + * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometCrossJoinBenchmark + * }}} + * Results will be written to "spark/benchmarks/CometCrossJoinBenchmark-**results.txt". + */ +object CometCrossJoinBenchmark extends CometBenchmarkBase { + + override def getSparkSession: SparkSession = { + val conf = new SparkConf() + .setAppName("CometCrossJoinBenchmark") + .set("spark.master", "local[5]") + .setIfMissing("spark.driver.memory", "3g") + .setIfMissing("spark.executor.memory", "3g") + .set( + "spark.shuffle.manager", + "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") + // Off-heap memory is disabled, so Comet runs in on-heap mode, which must be enabled + // explicitly or Comet stays disabled and every case measures Spark. + .set("spark.comet.exec.onHeap.enabled", "true") + + val sparkSession = SparkSession + .builder() + .config(conf) + .withExtensions(new CometSparkSessionExtensions) + .getOrCreate() + + sparkSession.conf.set(CometConf.COMET_ENABLED.key, "false") + sparkSession.conf.set(CometConf.COMET_EXEC_ENABLED.key, "false") + sparkSession.conf.set(SQLConf.ANSI_ENABLED.key, "false") + sparkSession.conf.set("spark.sql.shuffle.partitions", "2") + // The hints choose each join's strategy, so nothing is broadcast by size. + sparkSession.conf.set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key, "-1") + sparkSession.conf.set(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key, "-1") + + sparkSession + } + + /** + * @param joinEnabled + * Comet's switch for the join operator. Turning it off leaves the join in Spark and the + * operators around it as they were, which is the plan Comet made when it rejected the join + * type Cross. + */ + private case class CrossJoinCase( + name: String, + query: String, + joinEnabled: ConfigEntry[Boolean], + sparkJoin: Class[_ <: SparkPlan], + cometJoin: Class[_ <: SparkPlan]) + + private val SparkCaseName = "Spark" + private val JoinInSparkCaseName = "Comet, join in Spark" + private val CometCaseName = "Comet" + + private val sparkConfigs = Seq(CometConf.COMET_ENABLED.key -> "false") + + private val cometConfigs = + Seq(CometConf.COMET_ENABLED.key -> "true", CometConf.COMET_EXEC_ENABLED.key -> "true") + + private def joinInSparkConfigs(c: CrossJoinCase): Seq[(String, String)] = + cometConfigs :+ (c.joinEnabled.key -> "false") + + override def runCometBenchmark(mainArgs: Array[String]): Unit = { + val probeRows = 1024 * 1024L + val keys = 1024L + + // The sum reads a column from each side, so the join has to produce both, and only its one + // row leaves the native plan. + def equiJoin(hint: String): String = + s"SELECT /*+ $hint(b) */ sum(p.v + b.w) FROM probe p CROSS JOIN build b WHERE p.k = b.k" + + val cases = Seq( + CrossJoinCase( + "SortMergeJoin", + equiJoin("MERGE"), + CometConf.COMET_EXEC_SORT_MERGE_JOIN_ENABLED, + classOf[SortMergeJoinExec], + classOf[CometSortMergeJoinExec]), + CrossJoinCase( + "ShuffledHashJoin", + equiJoin("SHUFFLE_HASH"), + CometConf.COMET_EXEC_HASH_JOIN_ENABLED, + classOf[ShuffledHashJoinExec], + classOf[CometHashJoinExec]), + CrossJoinCase( + "BroadcastHashJoin", + equiJoin("BROADCAST"), + CometConf.COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED, + classOf[BroadcastHashJoinExec], + classOf[CometBroadcastHashJoinExec]), + CrossJoinCase( + "BroadcastNestedLoopJoin", + "SELECT /*+ BROADCAST(t) */ sum(p.v + t.w) FROM probe p CROSS JOIN tiny t", + CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED, + classOf[BroadcastNestedLoopJoinExec], + classOf[CometBroadcastNestedLoopJoinExec])) + + withTempPath { dir => + withTempTable("probe", "build", "tiny") { + def writeTable(view: String, path: String, df: DataFrame): Unit = { + df.write.parquet(s"${dir.getAbsolutePath}/$path") + spark.read.parquet(s"${dir.getAbsolutePath}/$path").createOrReplaceTempView(view) + } + + writeTable( + "probe", + "probe", + spark.range(probeRows).selectExpr(s"id % $keys AS k", "id AS v")) + + // Every probe row meets fanOut rows on the other side, so every case joins + // probeRows * fanOut rows. The larger fan-out leaves less of the time to the scans and + // shuffles around the join. + Seq(10L, 100L).foreach { fanOut => + val joinedRows = probeRows * fanOut + writeTable( + "build", + s"build_$fanOut", + spark.range(keys * fanOut).selectExpr(s"id % $keys AS k", "id AS w")) + writeTable("tiny", s"tiny_$fanOut", spark.range(fanOut).selectExpr("id AS w")) + + cases.foreach { c => + val name = s"Cross join - ${c.name}, fan-out $fanOut" + runBenchmark(name) { + verify(name, c, joinedRows) + val benchmark = new Benchmark(name, joinedRows, output = output) + benchmark.addCase(SparkCaseName)(_ => runQuery(c.query, sparkConfigs)) + benchmark.addCase(JoinInSparkCaseName)(_ => + runQuery(c.query, joinInSparkConfigs(c))) + benchmark.addCase(CometCaseName)(_ => runQuery(c.query, cometConfigs)) + benchmark.run() + } + } + } + } + } + } + + private def runQuery(query: String, configs: Seq[(String, String)]): Unit = + withSQLConf(configs: _*) { + spark.sql(query).noop() + } + + /** + * Fails unless every arm runs the join its name claims, with the join type Cross, the Spark + * join produces `joinedRows` rows, and the three arms return the same sum. + */ + private def verify(name: String, c: CrossJoinCase, joinedRows: Long): Unit = { + val arms = Seq( + SparkCaseName -> sparkConfigs, + JoinInSparkCaseName -> joinInSparkConfigs(c), + CometCaseName -> cometConfigs) + val sums = arms.map { case (arm, configs) => + // Assigned to a local rather than returned from the block: Spark 3.4 and 3.5 declare + // `SQLHelper.withSQLConf` as returning `Unit`. + var sum: Any = null + withSQLConf(configs: _*) { + val df = spark.sql(c.query) + sum = df.collect().head.get(0) + val plan = stripAQEPlan(df.queryExecution.executedPlan) + def fail(problem: String): Nothing = + throw new AssertionError(s"$name, $arm: $problem\n${plan.treeString}") + + // Each join with the Spark join it is or replaced, which carries the join type. + val joins = collect(plan) { + case j: BaseJoinExec => j -> j + case j: CometExec if j.originalPlan.isInstanceOf[BaseJoinExec] => + j -> j.originalPlan.asInstanceOf[BaseJoinExec] + } + val expected = if (arm == CometCaseName) c.cometJoin else c.sparkJoin + joins match { + case Seq((node, join)) if expected.isInstance(node) && join.joinType == Cross => + if (arm == SparkCaseName && join.metrics("numOutputRows").value != joinedRows) { + fail(s"the join produced ${join.metrics("numOutputRows").value} rows") + } + case _ => fail(s"expected one ${expected.getSimpleName} of join type Cross") + } + if (arm == CometCaseName) { + // Query stages are leaves to the checker, so check the plan inside each one as well. + val stagePlans = collect(plan) { case s: QueryStageExec => s.plan } + val wrappers = Seq(classOf[QueryStageExec], classOf[AQEShuffleReadExec]) + (plan +: stagePlans) + .flatMap(findFirstNonCometOperator(_, wrappers: _*)) + .headOption + .foreach(op => + fail(s"the plan is not fully native, first non-Comet operator: ${op.nodeName}")) + } + if (arm == JoinInSparkCaseName && collect(plan) { case p: CometPlan => p }.isEmpty) { + fail("Comet runs no operator") + } + } + arm -> sum + } + val (_, sparkSum) = sums.head + sums.tail.foreach { case (arm, sum) => + if (sum != sparkSum) { + throw new AssertionError(s"$name: Spark returned $sparkSum, $arm returned $sum") + } + } + } +}