feat: add support for comet cartesian join

native/core/src/execution/jni_api.rs

@@ -490,6 +490,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.rs

@@ -129,7 +129,7 @@ use arrow::buffer::{BooleanBuffer, NullBuffer, OffsetBuffer};
 use datafusion::common::utils::SingleRowListArrayBuilder;
 use datafusion::common::{NullHandling, UnnestOptions};
 use datafusion::physical_plan::filter::FilterExec;
-use datafusion::physical_plan::joins::NestedLoopJoinExec;
+use datafusion::physical_plan::joins::{CrossJoinExec, NestedLoopJoinExec};
 use datafusion::physical_plan::limit::GlobalLimitExec;
 use datafusion::physical_plan::unnest::ListUnnest;
 use datafusion_comet_proto::spark_expression::ListLiteral;
@@ -1989,6 +1989,36 @@ impl PhysicalPlanner {
                     ))
                 }
             }
+            OpStruct::CrossJoin(_) => {
+                // Spark's CartesianProductExec, delegated to DataFusion's CrossJoinExec. A pure
+                // cross product has no join keys and no condition; the Inner join type here only
+                // satisfies the shared helper and is unused by CrossJoinExec. CrossJoinExec emits
+                // left columns followed by right columns, matching Spark's output order.
+                let (join_params, scans, shuffle_scans) = self.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],
+                    )),
+                ))
+            }
             OpStruct::Limit(limit) => {
                 assert_eq!(children.len(), 1);
                 let num = limit.limit;

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

@@ -177,6 +177,7 @@ fn get_operator_type(spark_operator: &Operator) -> Option<OperatorType> {
         OpStruct::CsvScan(_) => Some(OperatorType::CsvScan),
         OpStruct::ShuffleScan(_) => Some(OperatorType::ShuffleScan),
         OpStruct::BroadcastNestedLoopJoin(_) => None,
+        OpStruct::CrossJoin(_) => None, // Not yet in OperatorType enum
         OpStruct::RangeScan(_) => None,
         OpStruct::Sample(_) => None,    // Not yet in OperatorType enum
         OpStruct::MergeRows(_) => None, // Not yet in OperatorType enum

native/proto/src/proto/operator.proto

@@ -70,6 +70,7 @@ message Operator {
     IcebergWrite iceberg_write = 120;
     RangeScan range_scan = 121;
     MergeRows merge_rows = 122;
+    CrossJoin cross_join = 123;
     // 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
@@ -1112,6 +1113,12 @@ message BroadcastNestedLoopJoin {
   optional spark.spark_expression.Expr condition = 3;
 }
 
+// Spark's CartesianProductExec, delegated to DataFusion's CrossJoinExec. A pure cross product has
+// no join keys and no condition, so the message carries nothing beyond the two child operators
+// (Spark folds any cross-join predicate into a separate FilterExec above this node).
+message CrossJoin {
+}
+
 enum JoinType {
   Inner = 0;
   LeftOuter = 1;

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

@@ -319,6 +319,15 @@ 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 entire left input in memory and " +
+          "cannot spill, so a large left input can exhaust memory where Spark's spillable " +
+          "CartesianProduct succeeds. Only single-partition, condition-free cross joins are " +
+          "eligible; everything else falls back to Spark"))
   val COMET_EXEC_EXISTENCE_JOIN_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.exec.existenceJoin.enabled")
       .category(CATEGORY_ENABLE_EXEC)

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

@@ -49,7 +49,7 @@ import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
 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.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, Exchange, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike}
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{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._
@@ -117,6 +117,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

@@ -43,7 +43,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive.{AQEShuffleReadExec, BroadcastQueryStageExec, QueryStageExec, ShuffleQueryStageExec}
 import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
 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, LongType, MapType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType}
@@ -3060,6 +3060,81 @@ 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 {
+
+  // outputPartitioning is inherited from the base exec (originalPlan.outputPartitioning), which
+  // matches Spark's CartesianProductExec (UnknownPartitioning), so downstream exchange planning
+  // is unchanged.
+
+  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 = {
+    // Spark folds a cross-join predicate into a separate FilterExec above the product, so the
+    // common CartesianProductExec has no condition. DataFusion's CrossJoinExec has no post-product
+    // filter, so fall back to Spark when a condition is attached.
+    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) {
+      // Comet runs a native plan per Spark partition and cannot inject an exchange here
+      // (EnsureRequirements has already run). A cross product is cross-partition, so it is only
+      // correct natively when both inputs are a single partition; otherwise fall back to Spark's
+      // CartesianProductExec, which crosses every left x right partition pair via CartesianRDD.
+      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/scala/org/apache/comet/exec/CometJoinSuite.scala

@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, Dynam
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
 import org.apache.spark.sql.catalyst.plans.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.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometCartesianProductExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec}
 import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, WholeStageCodegenExec}
 import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, QueryStageExec}
 import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException
@@ -150,6 +150,60 @@ class CometJoinSuite extends CometTestBase {
     }
   }
 
+  test("cross join (CartesianProduct) runs natively") {
+    withSQLConf(
+      CometConf.COMET_EXEC_CARTESIAN_PRODUCT_ENABLED.key -> "true",
+      // Disable broadcast so Spark plans a CartesianProductExec rather than a
+      // BroadcastNestedLoopJoinExec for the keyless join.
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      withParquetTable(Seq((0, 10), (1, 11), (2, 12)), "l") {
+        withParquetTable(Seq((0, 100), (1, 101)), "r") {
+          checkSparkAnswerAndOperator(
+            sql(
+              "SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM l) l " +
+                "CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM r) r"),
+            Seq(classOf[CometCartesianProductExec]))
+        }
+      }
+    }
+  }
+
+  test("cross join with a join condition falls back to Spark") {
+    withSQLConf(
+      CometConf.COMET_EXEC_CARTESIAN_PRODUCT_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      withParquetTable(Seq((0, 10), (1, 11), (2, 12)), "l") {
+        withParquetTable(Seq((0, 100), (1, 101)), "r") {
+          // A non-equi inner join plans a CartesianProductExec that carries the predicate; Comet
+          // has no native post-product filter, so results must still match Spark via fallback.
+          checkSparkAnswer(sql("SELECT * FROM l JOIN r ON l._2 > r._2"))
+        }
+      }
+    }
+  }
+
+  test("cross join stays on Spark by default (native path is opt-in)") {
+    // Native CrossJoinExec can't spill, so it's opt-in; by default the cross join stays on Spark.
+    withSQLConf(
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      withParquetTable(Seq((0, 10), (1, 11), (2, 12)), "l") {
+        withParquetTable(Seq((0, 100), (1, 101)), "r") {
+          val df = sql(
+            "SELECT * FROM (SELECT /*+ COALESCE(1) */ * FROM l) l " +
+              "CROSS JOIN (SELECT /*+ COALESCE(1) */ * FROM r) r")
+          checkSparkAnswer(df)
+          assert(
+            !df.queryExecution.executedPlan.exists(_.isInstanceOf[CometCartesianProductExec]),
+            "cross join must stay on Spark when cartesianProduct.enabled is false")
+        }
+      }
+    }
+  }
+
   test("join - self join") {
     val df1 = testData.select(testData("key")).as("df1")
     val df2 = testData.select(testData("key")).as("df2")