feat: Add trail mode config for Comet evaluation without native execution

[coderfender]

Which issue does this PR close?

Closes #5335

Rationale for this change

One of the feedbacks / growing patterns we observe in the industry is to evaluate comet to see how many operators are supported without actually executing anything in the native side. This should help the data engineering teams to better evaluate thair plans and potentially make changes to make use of full native execution. spark.comet.trial.enabled" unlocks that config. All the plans are annotated Comet[Trail] to let users know (through the driver logs) that comet is only printing the plan while the execution is still native

What changes are included in this PR?

1. Added a new config to enable trail mode which prints out the plan with comet trail annotations while still running things on JVM

How are these changes tested?

Unit tests to grep the plans and make sure no operators are native when the trail mode is enabled

[coderfender]

I will post screenshots of the plan shortly

PLAN_CAPTURE_AFTER_BEGIN
HashAggregate(keys=[_2#7], functions=[count(1)], output=[_2#7, count(1)#11L])
+- Exchange hashpartitioning(_2#7, 10), ENSURE_REQUIREMENTS, [plan_id=149]
   +- HashAggregate(keys=[_2#7], functions=[partial_count(1)], output=[_2#7, count#17L])
      +- FileScan parquet [_2#7] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tf/g3z1_9sd2gz35kvld2np8pfr0000gn/T/spark-5c..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<_2:int>

PLAN_CAPTURE_AFTER_END
26/08/18 16:28:15 INFO core/src/execution/jni_api.rs: Comet tokio runtime: using spark.executor.cores=5 worker threads
PLAN_CAPTURE_BEFORE_BEGIN
*(1) CometColumnarToRow
+- CometHashAggregate [_2#7, count#31L], [Final], [_2#7], [count(1)]
   +- AQEShuffleRead coalesced
      +- ShuffleQueryStage 0
         +- CometExchange hashpartitioning(_2#7, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=179]
            +- CometHashAggregate [_2#7], [Partial], [_2#7], [partial_count(1)]
               +- CometNativeScan parquet [_2#7] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tf/g3z1_9sd2gz35kvld2np8pfr0000gn/T/spark-5c..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<_2:int>

PLAN_CAPTURE_BEFORE_END

[andygrove]

Heads up, I used an LLM to help pull this review together. All the concerns below are ones I want to see addressed, but the reading of the code is machine-assisted, so push back if any of it looks off.

Big picture, I really like the direction and the revert-the-scans approach is nice and simple. One thing I would like to work through before merge: this doesn't compose with the reporting we already have. CometMetricsListener calls CometCoverageStats.forPlan(qe.executedPlan) and feeds CometSource (that's where comet.acceleration.ratio comes from), and ExtendedExplainInfo does the same for the Spark 4.0 UI panel via spark.comet.explain.format. Both read the executed plan, which in trial mode has no Comet operators, so a user with spark.comet.metrics.enabled set and Grafana watching the ratio would see 0% and draw exactly the wrong conclusion. #5335 called that out specifically as a goal. Could we stash the stats computed on the pre-revert plan and have the listener and explain provider read them when trial mode is on, rather than only writing to the driver log?

edit: this previous paragraph is nonsense, mostly

On the revert itself, I don't think the .copy(runtimeFilters = s.runtimeFilters) on line 630 is doing what it looks like. CometScanRule builds the node with runtimeFilters = scanExec.runtimeFilters, and PlanAdaptiveDynamicPruningFilters hasn't run by the time this rule sees the plan, so the two should already agree. More worrying, a case class copy returns a fresh instance with an empty tags map, and logicalLink lives in SparkPlan.LOGICAL_PLAN_TAG, so this hands AQE a scan with no logical link. The V1 branch above just uses s.wrapped, which is also what line 272 does when native conversion fails. Any reason the V2 case shouldn't do the same?

Related, plan.transformUp walks children only, so it doesn't descend into a SubqueryExec inside a ScalarSubquery or the SubqueryBroadcastExec that PlanDynamicPruningFilters creates. CometCoverageStats.forPlan does descend into those (there's a case for SubqueryBroadcast in CometCoverageStatsSuite), so the report and the revert are walking different trees. My guess is this works out because Spark prepares subquery plans through their own pipeline, which reruns CometExecRule, but CometScanExec.doExecute throws by design, so if one ever survives the revert the query dies instead of falling back. Could we add a test with a scalar subquery and one with DPP to pin this down?

AQE behavior isn't addressed and it was one of the open questions in #5335. The rule runs once as a queryStagePrepRule against the whole initial plan, and then again from preColumnarTransitions for each stage covering only that stage's operators. One query produces several different percentages at WARN level and nothing tells the reader which one is the whole-query answer. Can we either emit the report once per query, or label each line so it's clear what it covers?

On the config, every other config in CATEGORY_EXEC_EXPLAIN is named spark.comet.explain.*, so spark.comet.trial.enabled sits a bit oddly there. #5335 suggested spark.comet.planOnly.enabled, which also reads less like a licensing trial. Since config names are public API and painful to change after a release, could we settle on spark.comet.explain.planOnly.enabled or similar now? And could the doc string mention that the report goes to the driver log and that this needs spark.comet.exec.enabled=true? The check at line 624 is inside the else branch of _apply, so with exec disabled and shuffle enabled we return applyCometShuffle(plan) and Comet columnar shuffle still runs natively despite trial mode being on. With shuffle also disabled the user gets no report at all.

Docs: docs/source/user-guide/latest/understanding-comet-plans.md already has sections on the coverage summary and spark.comet.explain.format, so that seems like the natural home for a short trial-mode section. Worth noting there that the estimate is Scala-side only. The native plan is never handed to DataFusion, so anything that would fail in create_plan still counts as accelerated and the number can be optimistic. configs.md is generated so no need to touch that.

On the test: good that it asserts both directions. A few things I'd like to see added: a DSv2 case so the CometBatchScanExec branch is actually exercised (that's the trickier of the two), the same query with AQE on and off given the rule runs at different points, and something asserting the report itself. Right now the feature's entire user-visible output is untested. Also checkSparkAnswer(df) inside the trial-mode block is comparing Spark against Spark, since Comet isn't executing anything. Comparing against the Comet-enabled result would be a stronger check, or the assertion could just be dropped in favor of the plan check.

Last thing: the description says "trail mode" throughout, the config in the body doesn't match spark.comet.trial.enabled in the code, and "All the plans are annotated Comet[Trail]" doesn't match what the code does (it logs the coverage stats and extended explain without adding any annotation). Also "comet is only printing the plan while the execution is still native" reads backwards - I think you mean execution stays on the JVM. Could you tidy this up? It'll end up in the changelog.

[andygrove]

I created an alternative PR #5394 - could you take a look @coderfender?