feat: run joins of type Cross natively

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

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

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
      )
    }
 

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
      )
    }
 

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
      )
    }
 

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
       }

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)] =