Repository navigation
Description
Describe the bug
When CometExecRule converts a BroadcastExchangeExec to CometBroadcastExchangeExec, any ReusedExchangeExec nodes that referenced the original BroadcastExchangeExec are not updated. This breaks exchange reuse, resulting in duplicate broadcast execution and an extra project node in the plan.
This is caused by rule ordering: ReuseExchangeAndSubquery runs as a physical preparation rule and sets up ReusedExchangeExec → BroadcastExchangeExec. Then CometExecRule runs as a columnar rule (after preparation) and converts the join's BroadcastExchangeExec to CometBroadcastExchangeExec. The SubqueryBroadcastExec inside the DPP filter still holds a reference to the original BroadcastExchangeExec. The two exchanges have different canonicalized types and hashes, so they can't be matched for reuse.
Returning the wrapped exchange's canonicalized form from CometBroadcastExchangeExec.doCanonicalize() would be incorrect because the two exchanges produce different output formats (Arrow columnar vs Spark HashedRelation). They are not interchangeable at runtime.
This was exposed by adding non-AQE DPP support to CometNativeScanExec. Previously the DPP scan fell back to FileSourceScanExec, so the join wasn't fully native and CometExecRule didn't convert the broadcast exchange. With DPP working natively, the full join tree converts to Comet operators and the reuse breaks.
Steps to reproduce
test("DPP join exchange reuse") {
withTempDir { dir =>
val path = s"${dir.getAbsolutePath}/data"
spark.range(100).selectExpr(
"id % 10 as key", "cast(id * 2 as int) as a",
"cast(id * 3 as int) as b", "cast(id as string) as c",
"array(id, id + 1, id + 3) as d")
.write.partitionBy("key").parquet(path)
spark.read.parquet(path).createOrReplaceTempView("testView")
withSQLConf(
SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val query =
"""select * from (select key, a, c, b from testView) as t1
|join (select key, a, b, c from testView) as t2
|on t1.key = t2.key where t2.a > 50""".stripMargin
val df = sql(query)
df.collect()
val plan = df.queryExecution.executedPlan
val reused = collectWithSubqueries(plan) {
case e: ReusedExchangeExec => e
}
assert(reused.nonEmpty, "Expected exchange reuse")
}
}
}Expected behavior
The plan should contain a ReusedExchangeExec on the broadcast side of the join, reusing the broadcast from the DPP SubqueryBroadcastExec:
CometBroadcastHashJoin
:- CometProject [key, a, c, b]
: +- CometNativeScan [a, b, c, key] (DPP scan)
: +- SubqueryBroadcast
: +- BroadcastExchange (original)
+- ReusedExchange (reuses BroadcastExchange)
Additional context
Actual plan (no reuse, extra project):
CometBroadcastHashJoin
:- CometProject [key, a, c, b]
: +- CometNativeScan [a, b, c, key] (DPP scan)
: +- SubqueryBroadcast
: +- BroadcastExchange (Spark exchange, in subquery)
+- CometBroadcastExchange (Comet exchange, NOT reused)
+- CometProject [key, a, b, c] (extra project)
+- CometFilter
+- CometNativeScan
Canonicalization output confirms the mismatch:
CometBroadcastExchangeExec canonicalized: CometBroadcastExchangeExec, hash=-356598217
BroadcastExchangeExec canonicalized: BroadcastExchangeExec, hash=-555436138
On main (where DPP falls back to Spark), the join stays as BroadcastHashJoinExec with ReusedExchangeExec and only 2 project nodes.
This surfaces as a failure in Spark's RemoveRedundantProjectsSuite "join with ordering requirement" test, which expects 2 project nodes but finds 3.
Activity
mbutrovich commentedon Apr 21, 2026
I have fixed this in #4011 and will push a commit with new golden plans shortly.