feat: add spark.comet.explain.planOnly.enabled, reported per query

.github/workflows/pr_build_linux.yml

@@ -545,6 +545,7 @@ jobs:
               org.apache.comet.rules.CometScanContribSuite
               org.apache.comet.rules.CometScanSchemeFallbackSuite
               org.apache.comet.rules.CometExecRuleSuite
+              org.apache.comet.rules.CometPlanOnlySuite
               org.apache.spark.sql.comet.CometPlanEqualitySuite
               org.apache.comet.rules.CometEmptyRelationExecRuleSuite
               org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite

.github/workflows/pr_build_macos.yml

@@ -193,6 +193,7 @@ jobs:
               org.apache.comet.rules.CometScanContribSuite
               org.apache.comet.rules.CometScanSchemeFallbackSuite
               org.apache.comet.rules.CometExecRuleSuite
+              org.apache.comet.rules.CometPlanOnlySuite
               org.apache.spark.sql.comet.CometPlanEqualitySuite
               org.apache.comet.rules.CometEmptyRelationExecRuleSuite
               org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite

docs/source/user-guide/latest/understanding-comet-plans.md

@@ -199,6 +199,59 @@ from both operator counts:
   Counting reused exchanges once is tracked as item 3 of
   [#5203](https://github.com/apache/datafusion-comet/issues/5203).
 
+### `spark.comet.explain.planOnly.enabled`
+
+When enabled, Comet leaves every query for Spark to execute and, once the query
+finishes, logs the Comet plan it would have executed along with a coverage
+summary. Use this to evaluate how much of a workload Comet would accelerate
+without changing how that workload runs.
+
+Comet's conversion rules return the plan untouched while this is on, so the plan
+Spark executes is the plan it would have built with Comet switched off. The
+report is built afterwards from the plan that ran, and then discarded.
+
+The log line is prefixed with `[Comet plan-only]` and carries the same annotated
+plan and summary that `spark.comet.explain.format=verbose` produces for a real
+Comet plan. It is written once per SQL execution, so a plan that is built but
+never executed — `df.explain()`, or reading `queryExecution.executedPlan` — is
+not reported, and a DataFrame collected twice is reported twice.
+
+"Per SQL execution" rather than "per action" is a real distinction for RDD work.
+The report comes from a `QueryExecutionListener`, which Spark drives from the
+Dataset action path, so actions taken on `df.rdd` are outside it. On Spark 4.0
+and later, obtaining `df.rdd` runs a query of its own and is reported once at
+that point; later actions on the resulting RDD add nothing, because no new SQL
+execution starts. On Spark 3.4 and 3.5 obtaining `df.rdd` is not reported at
+all. Either way the RDD's own actions are never counted individually, so treat
+an RDD-heavy workload's report as covering the plans it built, not the jobs it
+ran.
+
+The preview goes through the whole Comet planning sequence rather than operator
+conversion alone: Spark's columnar transitions are inserted and Comet's
+post-columnar rules (`RevertNativeForTransitionHeavyStages`,
+`EliminateRedundantTransitions`) are applied, so a stage Comet would have handed
+back to Spark for having too many transitions is reported as handed back.
+
+Two things to keep in mind when reading the percentage:
+
+- The estimate reflects Scala-side conversion only. The plan is never handed to
+  DataFusion, so anything that would have failed in DataFusion's `create_plan`
+  still counts as accelerated. Treat the percentage as an upper bound.
+- Under AQE the report describes the plan AQE settled on, not the query as
+  written, because that is the plan Comet would have been asked to run. If a
+  stage materialized empty and AQE replaced the query with an empty relation,
+  that is what gets reported. The transition count can also be slightly lower
+  than a real Comet run's, because Spark inserts transitions one query stage at
+  a time whereas the report is produced from the whole plan in one pass. The
+  operator counts behind the percentage are not affected.
+
+A reused exchange is expanded in the report, so its subtree appears once per
+reference. That matches how coverage is counted for a real Comet plan; see the
+note under `spark.comet.explain.format` above.
+
+The config requires `spark.comet.exec.enabled=true`. With Comet exec disabled
+the rule that arranges the report does not run.
+
 ### `spark.comet.explain.native.enabled`
 
 When enabled, each executor task logs the DataFusion plan it executes,

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -756,6 +756,19 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_EXPLAIN_PLAN_ONLY_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.explain.planOnly.enabled")
+      .category(CATEGORY_EXEC_EXPLAIN)
+      .doc("When enabled, Comet leaves the query for Spark to execute and afterwards logs the " +
+        "Comet plan it would have executed, with a coverage summary. Use this to evaluate how " +
+        "much of a workload Comet would accelerate without changing execution. The estimate is " +
+        "Scala-side only; the plan is never handed to DataFusion, so native planning failures " +
+        "are not surfaced and the acceleration percentage can be optimistic. Reported once per " +
+        "action, so a plan built but never executed is not reported. Requires " +
+        "`spark.comet.exec.enabled=true`. Disabled by default.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_EXPLAIN_FALLBACK_LOG_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.explain.fallback.log.enabled")
       .withAlternative("spark.comet.logFallbackReasons.enabled")

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -675,6 +675,16 @@ case class CometExecRule(session: SparkSession)
   }
 
   override def apply(plan: SparkPlan): SparkPlan = {
+    // Plan-only mode: leave the plan alone, so Spark executes exactly what it would with Comet
+    // off, and arrange for the Comet plan to be reported once the query is over. See
+    // `CometPlanOnly` for why the report is not built here.
+    if (CometConf.COMET_EXPLAIN_PLAN_ONLY_ENABLED.get()) {
+      CometPlanOnly.register(session)
+      // Snapshot the settings with the plan: the report is delivered asynchronously, long after
+      // a `withSQLConf` block around the action may have restored them.
+      return CometPlanOnly.tagSettings(session, plan)
+    }
+
     val newPlan = _apply(plan)
     if (showTransformations && !newPlan.fastEquals(plan)) {
       logInfo(s"""
@@ -685,7 +695,7 @@ case class CometExecRule(session: SparkSession)
     newPlan
   }
 
-  private def _apply(plan: SparkPlan): SparkPlan = {
+  private[rules] def _apply(plan: SparkPlan): SparkPlan = {
     // We shouldn't transform Spark query plan if Comet is not loaded.
     if (!isCometLoaded(conf)) return plan