feat: add support for comet cartesian join

docs/source/user-guide/latest/operators.md

@@ -53,7 +53,6 @@ The following operator families fall back to Spark and are not on the current ro
 omitted from the tables below and may be reconsidered based on demand:
 
 - **Structured Streaming operators** (`StateStoreSaveExec`, `StateStoreRestoreExec`, `StreamingSymmetricHashJoinExec`, and similar): Comet targets batch execution.
-- **Cartesian / cross joins** (`CartesianProductExec`): rare and expensive, with little acceleration benefit.
 - **Pickled (non-Arrow) Python UDFs** (`BatchEvalPythonExec`): Comet accelerates Arrow-based Python UDFs only ([#4234](https://github.com/apache/datafusion-comet/pull/4234)).
 - **Typed Dataset operators** (`DeserializeToObjectExec`, `SerializeFromObjectExec`, `MapElementsExec`, `MapPartitionsExec`, `MapGroupsExec`, `CoGroupExec`, `AppendColumnsExec`, and similar): produced by `map`, `mapPartitions`, `groupByKey`, `cogroup`, and other typed `Dataset` transformations. They exist to run user JVM functions on JVM objects, which Comet cannot do natively. The operators above them can still run natively: set `spark.comet.convert.typedDataset.enabled=true` and Comet converts the output of a typed operation to Arrow. This is disabled by default because it can be slower than Spark when the operators above do little work, such as an aggregate over a few groups, which Spark compiles together with the typed operation into one loop.
 
@@ -103,6 +102,7 @@ omitted from the tables below and may be reconsidered based on demand:
 | `ShuffledHashJoinExec`        | ✅     |                                                                                                                                                                               |
 | `SortMergeJoinExec`           | ✅     | Supports `BINARY` join keys. Nested-type (struct, array, map) and collated-string join keys fall back to Spark.                                                               |
 | `BroadcastNestedLoopJoinExec` | ✅     | Falls back to Spark when the preserved side is broadcast (for example LEFT OUTER with BROADCAST on the left) ([#4429](https://github.com/apache/datafusion-comet/pull/4429)). |
+| `CartesianProductExec`        | ⚠️     | Opt-in (`spark.comet.exec.cartesianProduct.enabled`). Runs natively only without a join condition and when each input reports exactly one partition.                          |
 
 `ExistenceJoin` (produced when `EXISTS` / `IN` is combined with another predicate via `OR`) runs
 natively on `BroadcastHashJoinExec` and `ShuffledHashJoinExec`. Existence sort-merge joins, residual

native/core/src/execution/jni_api.rs

@@ -498,6 +498,7 @@ fn op_name(op: &OpStruct) -> &'static str {
         OpStruct::CsvScan(_) => "CsvScan",
         OpStruct::ShuffleScan(_) => "ShuffleScan",
         OpStruct::BroadcastNestedLoopJoin(_) => "BroadcastNestedLoopJoin",
+        OpStruct::CrossJoin(_) => "CrossJoin",
         OpStruct::Sample(_) => "Sample",
         OpStruct::MergeRows(_) => "MergeRows",
         OpStruct::RangeScan(_) => "RangeScan",

native/core/src/execution/planner/join.rs

@@ -15,8 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Builders for the native join operators (hash join, sort-merge join, and broadcast nested loop
-//! join), and the join-key and join-filter planning they share.
+//! Builders for the native join operators (hash join, sort-merge join, broadcast nested loop
+//! join, and cross join), and the join-key and join-filter planning they share.
 
 use std::{collections::HashMap, sync::Arc};
 
@@ -37,7 +37,8 @@ use datafusion::{
     physical_plan::{
         filter::FilterExec,
         joins::{
-            utils::JoinFilter, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
+            utils::JoinFilter, CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode,
+            SortMergeJoinExec,
         },
         projection::ProjectionExec,
         sorts::sort::SortExec,
@@ -420,6 +421,48 @@ impl OperatorBuilder for BroadcastNestedLoopJoinBuilder {
     }
 }
 
+/// Builder for `CrossJoin` (Spark's `CartesianProductExec`).
+pub struct CrossJoinBuilder;
+
+impl OperatorBuilder for CrossJoinBuilder {
+    fn build(
+        &self,
+        spark_plan: &Operator,
+        inputs: &mut Vec<Arc<Global<JObject<'static>>>>,
+        partition_count: usize,
+        planner: &PhysicalPlanner,
+    ) -> PlanCreationResult {
+        let children = &spark_plan.children;
+
+        // No keys or condition. The join type only satisfies the helper; CrossJoinExec emits left
+        // then right columns, as Spark does.
+        let (join_params, scans, shuffle_scans) = planner.parse_join_parameters(
+            inputs,
+            children,
+            &[],
+            &[],
+            JoinType::Inner as i32,
+            &None,
+            partition_count,
+        )?;
+
+        let left = Arc::clone(&join_params.left.native_plan);
+        let right = Arc::clone(&join_params.right.native_plan);
+
+        let cross_join = Arc::new(CrossJoinExec::new(left, right));
+
+        Ok((
+            scans,
+            shuffle_scans,
+            Arc::new(SparkPlan::new(
+                spark_plan.plan_id,
+                cross_join,
+                vec![join_params.left, join_params.right],
+            )),
+        ))
+    }
+}
+
 impl PhysicalPlanner {
     /// Keep Spark metric identities for the small set of nodes whose children
     /// runtime-filter placement can replace. Stop at native/Spark tree boundaries.

native/core/src/execution/planner/operator_registry.rs

@@ -63,6 +63,7 @@ pub enum OperatorType {
     MergeRows,
     Explode,
     BroadcastNestedLoopJoin,
+    CrossJoin,
     WindowGroupLimit,
 }
 
@@ -151,7 +152,9 @@ impl OperatorRegistry {
 
     /// Register join operators
     fn register_join_operators(&mut self) {
-        use super::join::{BroadcastNestedLoopJoinBuilder, HashJoinBuilder, SortMergeJoinBuilder};
+        use super::join::{
+            BroadcastNestedLoopJoinBuilder, CrossJoinBuilder, HashJoinBuilder, SortMergeJoinBuilder,
+        };
 
         self.builders
             .insert(OperatorType::HashJoin, Box::new(HashJoinBuilder));
@@ -161,6 +164,8 @@ impl OperatorRegistry {
             OperatorType::BroadcastNestedLoopJoin,
             Box::new(BroadcastNestedLoopJoinBuilder),
         );
+        self.builders
+            .insert(OperatorType::CrossJoin, Box::new(CrossJoinBuilder));
     }
 
     /// Register window operators
@@ -236,6 +241,7 @@ fn get_operator_type(spark_operator: &Operator) -> Option<OperatorType> {
         OpStruct::CsvScan(_) => Some(OperatorType::CsvScan),
         OpStruct::ShuffleScan(_) => Some(OperatorType::ShuffleScan),
         OpStruct::BroadcastNestedLoopJoin(_) => Some(OperatorType::BroadcastNestedLoopJoin),
+        OpStruct::CrossJoin(_) => Some(OperatorType::CrossJoin),
         OpStruct::RangeScan(_) => None,
         OpStruct::Sample(_) => None, // Not yet in OperatorType enum
         OpStruct::MergeRows(_) => Some(OperatorType::MergeRows),

native/proto/src/proto/operator.proto

@@ -71,6 +71,7 @@ message Operator {
     RangeScan range_scan = 121;
     MergeRows merge_rows = 122;
     ArrowPythonUdf arrow_python_udf = 123;
+    CrossJoin cross_join = 124;
     // Extension point for optional, out-of-tree contrib scans (Delta, Lance, ...). The concrete
     // scan message (e.g. `DeltaScan`) is packed into this envelope on the JVM side and dispatched
     // by `type_url` on the native side. Using a single permanent field -- rather than a new oneof
@@ -1158,6 +1159,10 @@ message BroadcastNestedLoopJoin {
   optional spark.spark_expression.Expr condition = 3;
 }
 
+// Spark's CartesianProductExec without a condition (one with a condition falls back to Spark).
+message CrossJoin {
+}
+
 enum JoinType {
   Inner = 0;
   LeftOuter = 1;

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -372,6 +372,14 @@ object CometConf extends ShimCometConf {
     createExecEnabledConfig("broadcastNestedLoopJoin", defaultValue = true)
   val COMET_EXEC_SORT_MERGE_JOIN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("sortMergeJoin", defaultValue = true)
+  val COMET_EXEC_CARTESIAN_PRODUCT_ENABLED: ConfigEntry[Boolean] =
+    createExecEnabledConfig(
+      "cartesianProduct",
+      defaultValue = false,
+      notes = Some(
+        "Opt-in: DataFusion's CrossJoinExec buffers the left input and cannot spill. Runs " +
+          "natively only without a join condition and when each input reports exactly one " +
+          "partition"))
   val COMET_EXEC_SORT_MERGE_JOIN_RIGHT_KEY_REDUCTION_ENABLED: ConfigEntry[Boolean] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.sortMergeJoin.rightKeyReduction.enabled")
       .category(CATEGORY_EXEC)

spark/src/main/scala/org/apache/comet/rules/CometCoalesceShufflePartitions.scala

@@ -104,7 +104,8 @@ case object CometCoalesceShufflePartitions extends AQEShuffleReadRule {
 
   // A Comet operator whose Spark original's children Spark's rule coalesces one by one, with that
   // original. The class match mirrors Spark's, and Spark's rule decides, for its version, which
-  // of these it actually treats that way. Comet has no counterpart of `CartesianProductExec`.
+  // of these it actually treats that way. `CometCartesianProductExec` is left out because each of
+  // its children is a single partition, so there is nothing to coalesce.
   private object Replaced {
     def unapply(plan: SparkPlan): Option[SparkPlan] = plan match {
       case comet: CometExec if comet.outputPartitioning.isInstanceOf[UnknownPartitioning] =>

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -52,7 +52,7 @@ import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
 import org.apache.spark.sql.execution.datasources.v2.text.TextScan
 import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, Exchange, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike}
-import org.apache.spark.sql.execution.joins.{BaseJoinExec, BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BaseJoinExec, BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, CartesianProductExec, ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.window.WindowExec
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -120,6 +120,7 @@ object CometExecRule {
       classOf[BroadcastNestedLoopJoinExec] -> CometBroadcastNestedLoopJoinExec,
       classOf[ShuffledHashJoinExec] -> CometHashJoinExec,
       classOf[SortMergeJoinExec] -> CometSortMergeJoinExec,
+      classOf[CartesianProductExec] -> CometCartesianProductExec,
       classOf[SortExec] -> CometSortExec,
       classOf[LocalTableScanExec] -> CometLocalTableScanExec,
       classOf[RangeExec] -> CometRangeExec,

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -46,7 +46,7 @@ import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregat
 import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase
 import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, HashJoin, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, CartesianProductExec, HashJoin, ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, IntegralType, LongType, MapType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType}
@@ -3289,6 +3289,79 @@ object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNest
   }
 }
 
+case class CometCartesianProductExec(
+    override val nativeOp: Operator,
+    override val originalPlan: SparkPlan,
+    override val output: Seq[Attribute],
+    override val left: SparkPlan,
+    override val right: SparkPlan,
+    override val serializedPlanOpt: SerializedPlan)
+    extends CometBinaryExec {
+
+  // Inherits Spark's UnknownPartitioning from originalPlan, so exchange planning is unchanged.
+
+  // Each input must stay one partition. Without this, AQE's ValidateRequirements lets
+  // OptimizeSkewInRebalancePartitions split a REBALANCE(1) input after the product went native.
+  override def requiredChildDistribution: Seq[Distribution] = Seq(AllTuples, AllTuples)
+
+  override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
+    this.copy(left = newLeft, right = newRight)
+
+  override def stringArgs: Iterator[Any] = Iterator(left, right)
+
+  override def equals(obj: Any): Boolean = {
+    obj match {
+      case other: CometCartesianProductExec =>
+        this.output == other.output &&
+        this.left == other.left &&
+        this.right == other.right &&
+        this.serializedPlanOpt == other.serializedPlanOpt
+      case _ =>
+        false
+    }
+  }
+
+  override def hashCode(): Int = Objects.hashCode(output, left, right)
+
+  override lazy val metrics: Map[String, SQLMetric] =
+    CometMetricNode.joinMetrics(sparkContext)
+}
+
+object CometCartesianProductExec extends CometOperatorSerde[CartesianProductExec] {
+
+  override def enabledConfig: Option[ConfigEntry[Boolean]] =
+    Some(CometConf.COMET_EXEC_CARTESIAN_PRODUCT_ENABLED)
+
+  override def getSupportLevel(op: CartesianProductExec): SupportLevel = {
+    // CartesianProductExec carries its own condition, and CrossJoinExec has no post-product filter.
+    if (op.condition.isDefined) {
+      Unsupported(Some("CartesianProduct with a join condition is not supported natively"))
+    } else if (op.left.outputPartitioning.numPartitions != 1 ||
+      op.right.outputPartitioning.numPartitions != 1) {
+      // A native plan runs per partition and cannot add an exchange here, so a cross product is
+      // only correct when both inputs are a single partition.
+      Unsupported(
+        Some(
+          "Native CartesianProduct requires single-partition inputs (a multi-partition " +
+            "cross product cannot be expressed per-partition)"))
+    } else {
+      Compatible(None)
+    }
+  }
+
+  override def convert(
+      op: CartesianProductExec,
+      builder: Operator.Builder,
+      childOp: Operator*): Option[Operator] = {
+    val joinBuilder = OperatorOuterClass.CrossJoin.newBuilder()
+    Some(builder.setCrossJoin(joinBuilder).build())
+  }
+
+  override def createExec(nativeOp: Operator, op: CartesianProductExec): CometNativeExec = {
+    CometCartesianProductExec(nativeOp, op, op.output, op.left, op.right, SerializedPlan(None))
+  }
+}
+
 object CometBroadcastHashJoinExec extends CometOperatorSerde[HashJoin] with CometHashJoin {
 
   override def enabledConfig: Option[ConfigEntry[Boolean]] =

spark/src/test/resources/sql-tests/join/cartesian_product.sql

@@ -0,0 +1,86 @@
+-- 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.
+
+-- Tests for CartesianProductExec, which Spark plans for a cross join with no equi key when
+-- nothing is broadcast. Comet runs it natively only without a condition and when each input
+-- reports exactly one partition. A non-bucketed Parquet scan reports UnknownPartitioning(0), so
+-- each side is coalesced to one partition. Partition coalescing is off so AQE cannot merge the
+-- REPARTITION(2) case, and the tiny advisory size makes AQE try to split the REBALANCE(1) case.
+
+-- Config: spark.comet.exec.cartesianProduct.enabled=true
+-- Config: spark.sql.autoBroadcastJoinThreshold=-1
+-- Config: spark.sql.adaptive.autoBroadcastJoinThreshold=-1
+-- Config: spark.sql.adaptive.coalescePartitions.enabled=false
+-- Config: spark.sql.adaptive.advisoryPartitionSizeInBytes=1k
+-- ConfigMatrix: spark.sql.adaptive.enabled=true,false
+
+statement
+CREATE TABLE cp_left(k int, v int) USING parquet
+
+statement
+INSERT INTO cp_left VALUES (0, 10), (NULL, 11), (2, NULL), (NULL, NULL)
+
+statement
+CREATE TABLE cp_right(k int, v int) USING parquet
+
+statement
+INSERT INTO cp_right VALUES (0, 100), (NULL, NULL)
+
+statement
+CREATE TABLE cp_empty(k int, v int) USING parquet
+
+statement
+CREATE TABLE cp_large(k int, v int) USING parquet
+
+-- More rows than the default batch size of 8192.
+statement
+INSERT INTO cp_large SELECT id, id FROM range(10000)
+
+-- Native, with NULLs on both sides
+query
+SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM cp_left) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_right) r
+
+-- Native, empty left side (the side CrossJoinExec buffers)
+query
+SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM cp_empty) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_right) r
+
+-- Native, empty right side
+query
+SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM cp_left) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_empty) r
+
+-- Native, right side spanning several batches
+query
+SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM cp_left) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_large) r
+
+-- Native, AQE must not split the REBALANCE(1) input into several partitions
+query
+SELECT * FROM (SELECT /*+ REBALANCE(1) */ * FROM cp_large) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_right) r
+
+-- A join condition falls back to Spark
+query expect_fallback(CartesianProduct with a join condition is not supported natively)
+SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM cp_left) l
+JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_right) r ON l.v > r.v
+
+-- A multi-partition input falls back to Spark
+query expect_fallback(Native CartesianProduct requires single-partition inputs)
+SELECT * FROM (SELECT /*+ REPARTITION(2) */ * FROM cp_left) l
+CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM cp_right) r