diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff index 368eec91588..69f7297a004 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 64fb8182936..1da643efa7c 100644 --- a/dev/diffs/3.5.9.diff +++ b/dev/diffs/3.5.9.diff @@ -1220,7 +1220,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 @@ -1276,6 +1276,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 ac636fd808d..de3f897fc2b 100644 --- a/dev/diffs/4.0.4.diff +++ b/dev/diffs/4.0.4.diff @@ -1390,7 +1390,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 @@ -1436,7 +1436,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 563b03f64bd..d1fa6ed007b 100644 --- a/dev/diffs/4.1.3.diff +++ b/dev/diffs/4.1.3.diff @@ -1548,7 +1548,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 @@ -1594,7 +1594,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 03651433382..b2e4accf9ff 100644 --- a/dev/diffs/4.2.0.diff +++ b/dev/diffs/4.2.0.diff @@ -1597,7 +1597,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 @@ -1643,7 +1643,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 87268b5fc69..d73d238c9f8 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 @@ -2876,7 +2876,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 @@ -2909,7 +2910,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 } @@ -3006,7 +3006,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, _) => @@ -3043,14 +3043,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 } @@ -3427,7 +3427,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 @@ -3437,7 +3438,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 d0e1438ee4a..23c27333e5e 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala @@ -33,7 +33,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, SQLExecution, WholeStageCodegenExec} @@ -1717,6 +1717,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") @@ -2134,6 +2165,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)] = 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") + } + } + } +}