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

[andygrove]

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:

(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 a28ac34:

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.