You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/29 07:28:37 UTC
(datafusion-comet) 01/01: docs: explain operator plan identity and exchange reuse (#6352)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6352-d8a7a4d999d660ca1c37674e5c4301c9ed5943fc
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 0d9a8f9de229bd8f05d527d1420baf145350aefb
Author: KUAN-HAO HUANG <10...@users.noreply.github.com>
AuthorDate: Mon Sep 28 23:39:41 2026 +0000
docs: explain operator plan identity and exchange reuse (#6352)
* docs: explain operator plan identity and exchange reuse
* docs: explain operator plan identity and exchange reuse
---
.../contributor-guide/adding_a_new_operator.md | 86 +++++++++++++++++++++-
1 file changed, 84 insertions(+), 2 deletions(-)
diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md
index c0a20f40aa..ac97c83d81 100644
--- a/docs/source/contributor-guide/adding_a_new_operator.md
+++ b/docs/source/contributor-guide/adding_a_new_operator.md
@@ -171,6 +171,8 @@ The validation workflow in `CometExecRule.isOperatorEnabled`:
#### Simple Example (Filter)
```scala
+import com.google.common.base.Objects
+
object CometFilterExec extends CometOperatorSerde[FilterExec] {
override def enabledConfig: Option[ConfigEntry[Boolean]] =
@@ -188,7 +190,6 @@ object CometFilterExec extends CometOperatorSerde[FilterExec] {
.setPredicate(cond.get)
Some(builder.setFilter(filterBuilder).build())
} else {
- withInfo(op, op.condition, op.child)
None
}
}
@@ -213,12 +214,30 @@ case class CometFilterExec(
override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
this.copy(child = newChild)
+
+ override def stringArgs: Iterator[Any] =
+ Iterator(output, condition, child)
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometFilterExec =>
+ this.output == other.output &&
+ this.condition == other.condition && this.child == other.child &&
+ this.serializedPlanOpt == other.serializedPlanOpt
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int = Objects.hashCode(output, condition, child)
}
```
#### More Complex Example (Project)
```scala
+import com.google.common.base.Objects
+
object CometProjectExec extends CometOperatorSerde[ProjectExec] {
override def enabledConfig: Option[ConfigEntry[Boolean]] =
@@ -236,7 +255,6 @@ object CometProjectExec extends CometOperatorSerde[ProjectExec] {
.addAllProjectList(exprs.map(_.get).asJava)
Some(builder.setProjection(projectBuilder).build())
} else {
- withInfo(op, op.projectList: _*)
None
}
}
@@ -260,9 +278,58 @@ case class CometProjectExec(
override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
this.copy(child = newChild)
+
+ override def stringArgs: Iterator[Any] = Iterator(output, projectList, child)
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometProjectExec =>
+ this.output == other.output &&
+ this.projectList == other.projectList &&
+ this.child == other.child &&
+ this.serializedPlanOpt == other.serializedPlanOpt
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int = Objects.hashCode(output, projectList, child)
+
+ override protected def outputExpressions: Seq[NamedExpression] = projectList
}
```
+#### Plan Identity and Exchange Reuse
+
+Spark's `ReuseExchangeAndSubquery` identifies equivalent plans through canonicalization and may
+reuse one exchange for both branches. If `equals` omits a parameter that changes results, different
+operators can appear equivalent and silently return the wrong rows. If it includes execution state,
+equivalent operators can fail to reuse an exchange.
+
+For operators such as Filter and Project, compare the children, output and every parameter that
+affects results in `equals`, and include those semantic fields in `hashCode`. Capture semantic flags
+on the Comet operator itself: a value stored only in the protobuf or original Spark plan cannot
+participate in this field-based identity. Keep `equals` and `hashCode` consistent: equal operators
+must have equal hashes. Do not rely on the default case-class implementations.
+
+The examples above follow the implementations in
+[`operators.scala`](https://github.com/apache/datafusion-comet/blob/main/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala):
+
+- `nativeOp` is the protobuf representation and `originalPlan` is the source Spark plan; neither is
+ compared by these operators. `CometNativeExec.canonicalizePlans` clears the non-child Spark plan
+ references, including their `originalPlan`.
+- `serializedPlanOpt` holds the bytes for a native execution block. These operators compare it in
+ `equals`, but omit it from `hashCode`; `CometNativeExec.doCanonicalize` clears the block's serialized
+ plan. The bytes therefore do not distinguish canonicalized plans.
+- `stringArgs` controls the plan's displayed arguments. Show the semantic fields and children
+ rather than serialization state; this method does not define equality.
+
+Some operators use a different convention. `CometNativeScanExec` compares `originalPlan` to retain
+scan identity, and its `doCanonicalize` canonicalizes that plan while removing unused dynamic
+pruning filters. `CometBroadcastExchangeExec` also compares `originalPlan` before canonicalization,
+but its `doCanonicalize` clears that reference and retains the canonicalized child. Follow each
+operator's equality and canonicalization together rather than copying an exclusion in isolation.
+
#### Using getSupportLevel
Override `getSupportLevel` to control operator support based on specific conditions:
@@ -411,6 +478,21 @@ The `checkSparkAnswerAndOperator` helper verifies:
1. Results match Spark's native execution
2. Your operator is actually being used (not falling back)
+#### Plan Identity Regression Tests
+
+If the operator has a result-affecting parameter beyond its children and output, add an
+exchange-reuse regression. Build two branches that differ in that parameter, verify that both
+execute with Comet, and assert the expected distinct results and that `sameResult` is false.
+Also test equivalent branches with fresh expression IDs or aliases: `sameResult` should be true,
+`semanticHash` values should match, and the executed plan should contain a reused exchange. Cover AQE both enabled and disabled when applicable.
+
+Use the "aggregate canonicalization preserves result expressions and equivalent reuse" tests in
+[`CometAggregateSuite`](https://github.com/apache/datafusion-comet/blob/main/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala)
+as a model. Inspect the executed plans: an optimizer rewrite can make the branches differ for an
+unrelated reason, allowing a test to pass even when the intended field is missing from `equals`.
+Choose inputs and query shapes that preserve the parameter difference, and traverse adaptive and
+query-stage wrappers when checking the native operator and reused exchange.
+
#### Rust Unit Tests
Add unit tests in your Rust implementation file:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org