feat: add support for comet cartesian join

native/core/src/execution/jni_api.rs

@@ -497,6 +497,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,
         ChildrenPropertiesMode, ExecutionPlan, ExecutionPlanProperties, ReplaceChildrenOptions,
@@ -344,6 +345,49 @@ impl OperatorBuilder for BroadcastNestedLoopJoinBuilder {
     }
 }
 
+/// Builder for `CrossJoin`: Spark's CartesianProductExec, delegated to DataFusion's CrossJoinExec.
+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;
+
+        // 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) = 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
@@ -1156,6 +1157,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

@@ -361,6 +361,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

@@ -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.{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._
@@ -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

@@ -45,7 +45,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, LongType, MapType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType}
@@ -3273,6 +3273,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

@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.expressions.{And, Ascending, AttributeRefer
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide}
 import org.apache.spark.sql.catalyst.plans.{Inner, LeftOuter, LeftSemi}
 import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoin, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometRangeExec, CometSampleExec, CometSortExec, CometSortMergeJoinExec, CometSparkToColumnarExec, CometUnionExec, CometWindowExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometCartesianProductExec, CometFilterExec, CometHashJoin, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometRangeExec, CometSampleExec, CometSortExec, CometSortMergeJoinExec, CometSparkToColumnarExec, CometUnionExec, CometWindowExec}
 import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, SQLExecution, WholeStageCodegenExec}
 import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, QueryStageExec}
 import org.apache.spark.sql.execution.aggregate.SortAggregateExec
@@ -154,6 +154,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")