comphead commented on code in PR #6938:
URL: https://github.com/apache/datafusion-comet/pull/6938#discussion_r4407318646
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -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 ||
Review Comment:
As far as I can tell, no plain file scan passes this check, since
non-bucketed scans report `UnknownPartitioning(0)`. Inputs qualify only after
`coalesce(1)`, `repartition(1)`, a single-slice `range` or a `SinglePartition`
exchange, which is what every benchmark case forces. Under AQE this rule runs
before shuffle partitions are coalesced, so a shuffle that later shrinks to one
partition is probably still rejected, though I have not run that. Could the
config note and the `operators.md` entry say this, so users know when enabling
it helps? Separately, for the large-left, small-right case, DataFusion 55.2
`CrossJoinExec::build_batch` emits one batch per left row, sized by the right
batch, so 262144 x 64 produces 262144 batches of 64 rows. That probably
explains why that case gains 3.1X against 18.6X for the square one. I read this
from the source and did not measure it.
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -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"))
Review Comment:
This test cannot fail on the condition guard. Both sides are plain parquet
scans, which report `UnknownPartitioning(0)`
(`CometNativeScanExec.outputPartitioning` and `FileSourceScanExec` for
non-bucketed scans), so the single-partition check rejects the product even if
the `op.condition.isDefined` branch were removed. Could both sides use `/*+
COALESCE(1) */` as in the native test, so the condition is the only reason
left? A `sql-tests/join/cartesian_product.sql` fixture with `-- Config:
spark.comet.exec.cartesianProduct.enabled=true` and both broadcast thresholds
at `-1` could then hold this as `query expect_fallback(CartesianProduct with a
join condition is not supported natively)`, next to the native case and the
multi-partition, NULL, empty-side and over-8192-row cases andygrove listed. The
default-off check needs the plan and can stay in Scala. I have not run the
fixture.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
