From 9e13ebd528aa57677152061f819117f1ab350b4c Mon Sep 17 00:00:00 2001 From: Bhargava Vadlamani Date: Mon, 12 Oct 2026 21:38:16 -0700 Subject: [PATCH 1/2] add support for comet cartesian join --- native/core/src/execution/jni_api.rs | 1 + native/core/src/execution/planner.rs | 32 +++++++- .../execution/planner/operator_registry.rs | 1 + native/proto/src/proto/operator.proto | 7 ++ .../scala/org/apache/comet/CometConf.scala | 2 + .../apache/comet/rules/CometExecRule.scala | 3 +- .../apache/spark/sql/comet/operators.scala | 77 ++++++++++++++++++- .../apache/comet/exec/CometJoinSuite.scala | 36 ++++++++- 8 files changed, 155 insertions(+), 4 deletions(-) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 0812b9f7087..f7e783ea344 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -489,6 +489,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", diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 96802ca805d..5bc522ee096 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -125,7 +125,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; @@ -1777,6 +1777,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; diff --git a/native/core/src/execution/planner/operator_registry.rs b/native/core/src/execution/planner/operator_registry.rs index 70ed309ebeb..cd66c2ef424 100644 --- a/native/core/src/execution/planner/operator_registry.rs +++ b/native/core/src/execution/planner/operator_registry.rs @@ -177,6 +177,7 @@ fn get_operator_type(spark_operator: &Operator) -> Option { 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 diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 97c65f0da36..5e1bd7de0ae 100644 --- a/native/proto/src/proto/operator.proto +++ b/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 @@ -1042,6 +1043,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; diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 4aab7c72a11..a11ab8db01f 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -319,6 +319,8 @@ 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 = true) val COMET_EXEC_EXISTENCE_JOIN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.exec.existenceJoin.enabled") .category(CATEGORY_ENABLE_EXEC) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index a47f1c2a6b4..5abf6ee7ff2 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala @@ -50,7 +50,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._ @@ -118,6 +118,7 @@ object CometExecRule { classOf[BroadcastNestedLoopJoinExec] -> CometBroadcastNestedLoopJoinExec, classOf[ShuffledHashJoinExec] -> CometHashJoinExec, classOf[SortMergeJoinExec] -> CometSortMergeJoinExec, + classOf[CartesianProductExec] -> CometCartesianProductExec, classOf[SortExec] -> CometSortExec, classOf[LocalTableScanExec] -> CometLocalTableScanExec, classOf[RangeExec] -> CometRangeExec, diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index bcae614556b..85c83f85508 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/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} @@ -3038,6 +3038,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]] = diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala index 369c2517636..faf14fade43 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/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, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec} +import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometCartesianProductExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, 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,40 @@ 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("join - self join") { val df1 = testData.select(testData("key")).as("df1") val df2 = testData.select(testData("key")).as("df2") From f9a39feb1f8543c5fc686ed51ea31b0c1a67329e Mon Sep 17 00:00:00 2001 From: Bhargava Vadlamani Date: Thu, 15 Oct 2026 11:16:02 -0700 Subject: [PATCH 2/2] address review: make native cartesian product opt-in (default false) DataFusion's CrossJoinExec buffers the whole left input and cannot spill, so a large left can OOM where Spark's spillable CartesianProduct succeeds (sunchao's P2). Default the flag to false until the native path can spill; add a regression test that a cross join stays on Spark by default. Part of #199. --- .../scala/org/apache/comet/CometConf.scala | 9 ++++++++- .../apache/comet/exec/CometJoinSuite.scala | 20 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 4a5986f59bd..b8cfd0907de 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -320,7 +320,14 @@ object CometConf extends ShimCometConf { val COMET_EXEC_SORT_MERGE_JOIN_ENABLED: ConfigEntry[Boolean] = createExecEnabledConfig("sortMergeJoin", defaultValue = true) val COMET_EXEC_CARTESIAN_PRODUCT_ENABLED: ConfigEntry[Boolean] = - createExecEnabledConfig("cartesianProduct", defaultValue = true) + 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) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala index 50b2c6dd158..618bd87749e 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala @@ -184,6 +184,26 @@ class CometJoinSuite extends CometTestBase { } } + 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")