apache / apache/datafusion-comet

AQE + DPP + spark.comet.exec.transitionRevert.enabled fails with "SubqueryAdaptiveBroadcastExec does not support the execute() code path"

Open
#5,486 0 comments 0 reactions 0 assignees View on GitHub
bug priority:medium
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.