Joins with join type Cross (CROSS JOIN, Dataset.crossJoin) fall back to Spark #6903

What is the problem the feature request solves?

Spark gives a join the type Cross when the query asks for a cross join explicitly, with CROSS JOIN in SQL or Dataset.crossJoin. The type survives optimization. When a filter supplies an equi key, PushPredicateThroughJoin moves it into the join condition and keeps Cross, and JoinSelection plans a sort-merge, shuffled hash or broadcast hash join of type Cross. Without an equi key, a broadcast side gives a BroadcastNestedLoopJoinExec of type Cross.

Spark's join operators treat Cross exactly like Inner (case _: InnerLike). Comet's join serdes match only Inner, so each of these joins falls back. At 033cb37:

All three mappings carry the comment "Spark doesn't support other join types", but Spark does support Cross. The Comet exec nodes already handle it: CometSortMergeJoinExec and CometHashJoinExec match case _: InnerLike in outputPartitioning.

We hit this in a production job on Spark 3.5. A cross join with an equi-key filter became a SortMergeJoin of type Cross that produced 110 billion rows. It stayed in Spark with Unsupported join type Cross, and its stage took two thirds of the job's task time while the rest of the plan ran natively.

No test covers it. The CometJoinSuite tests named "BroadcastNestedLoopJoin cross join ..." use FROM tbl_a, tbl_b, which Spark plans as Inner.

To reproduce, in a spark-shell with Comet and spark.comet.explain.fallback.enabled=true:

import org.apache.spark.sql.functions.broadcast

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
val a = spark.range(1000).selectExpr("id % 10 AS k", "id AS v")
val b = spark.range(1000).selectExpr("id % 10 AS k2", "id AS w")

a.crossJoin(b).where("k = k2").explain()            // SortMergeJoin [k], [k2], Cross: falls back
a.join(b, a("k") === b("k2")).explain()              // CometSortMergeJoin, Inner: native

a.crossJoin(broadcast(b)).where("k = k2").explain() // BroadcastHashJoin [k], [k2], Cross: falls back
a.crossJoin(broadcast(b)).explain()                  // BroadcastNestedLoopJoin BuildRight, Cross: falls back
a.join(broadcast(b)).explain()                       // CometBroadcastNestedLoopJoin, Inner: native

Describe the potential solution

Map Cross to JoinType.Inner in the three places above: case _: InnerLike => JoinType.Inner in the sort-merge and hash join serdes, and (_: InnerLike, _) in the nested loop join's support check and conversion. The keys and condition carry over unchanged, so the native plan is the same as for the equivalent inner join.

Tests: CROSS JOIN ... WHERE a.k = b.k under sort-merge, shuffled hash and broadcast hash joins, and CROSS JOIN with no condition under a broadcast hint. Each should check the answer and that the join runs natively.

Additional context

CartesianProductExec, the cross join without a broadcast side, is a separate gap: #199.