apache / apache/datafusion-comet
AQE + DPP + spark.comet.exec.transitionRevert.enabled fails with "SubqueryAdaptiveBroadcastExec does not support the execute() code path"
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 6h
- Merged PRs (30d)
- 190
Description
### Describe the bug
With AQE enabled, dynamic partition pruning in play, and `spark.comet.exec.transitionRevert.enabled=true`, a broadcast join against a partitioned fact table fails with:
```
org.apache.spark.SparkUnsupportedOperationException: SubqueryAdaptiveBroadcastExec does not support the execute() code path.
at org.apache.spark.sql.execution.SubqueryAdaptiveBroadcastExec.doExecute(SubqueryAdaptiveBroadcastExec.scala:44)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeRDD$1(SparkPlan.scala:188)
```
`SubqueryAdaptiveBroadcastExec` is a placeholder that `PlanAdaptiveDynamicPruningFilters` is supposed to replace with a real `SubqueryBroadcastExec` during query stage optimization. Reaching `doExecute` means the replacement never happened, so something in the reversion is leaving the plan in a state where that rule no longer matches.
The failure needs `transitionRevert` on; it does not depend on `spark.comet.exec.project.enabled`.
### Steps to reproduce
Against a partitioned fact table and a small dimension table:
```scala
(0 until 400).map(i => (i, i % 10, s"f$i"))
.toDF("fact_id", "fact_key", "fact_str")
.write.partitionBy("fact_key").parquet(factPath)
(0 until 10).map(i => (i, i, s"d$i"))
.toDF("dim_id", "dim_key", "dim_str")
.write.parquet(dimPath)
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "true",
CometConf.COMET_EXEC_TRANSITION_REVERT_MAX_TRANSITIONS.key -> "0") {
spark.sql(
"""SELECT f.fact_id, f.fact_str, d.dim_str
|FROM fact f JOIN dim d ON f.fact_key = d.dim_key
|WHERE d.dim_id < 10""".stripMargin).collect()
}
```
The full matrix on Spark 4.1.3, commit a28ac348f:
| AQE | `transitionRevert.enabled` | result |
| --- | --- | --- |
| on | on | throws |
| on | off | 400 rows |
| off | on | 400 rows |
| off | off | 400 rows |
### Expected behavior
The query returns 400 rows regardless of whether `transitionRevert` is enabled.
### Additional context
`transitionRevert` is currently off by default, so this is not hit on a default configuration. It matters for #5207, which proposes turning whole-stage revert on by default.
Possibly related: #4145, which is also about `SubqueryAdaptiveBroadcastExec` wrapping being skipped, though that one is specific to V2 scans wrapped in `CometSparkToColumnarExec`. This reproduction uses the V1 scan path.
Found while reviewing #5394; the bug is independent of that PR and reproduces on its base commit.
Contributor guide
Research direction
Reproduce the query with AQE, dynamic partition pruning, and transitionRevert enabled, then trace PlanAdaptiveDynamicPruningFilters and SubqueryAdaptiveBroadcastExec through transition reversion. Done means the partitioned fact-table join returns 400 rows without the execute() exception, while the other configuration combinations remain passing.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- data-engineering, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100