Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/pr_build_linux.yml
Original file line number Diff line number Diff line change
Expand Up @@ -418,6 +418,7 @@ jobs:
org.apache.comet.rules.CometScanContribSuite
org.apache.comet.rules.CometScanSchemeFallbackSuite
org.apache.comet.rules.CometExecRuleSuite
org.apache.comet.rules.CometPlanOnlySuite
org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite
org.apache.spark.sql.CometTPCDSQuerySuite
org.apache.spark.sql.CometTPCDSQueryTestSuite
Expand Down
1 change: 1 addition & 0 deletions .github/workflows/pr_build_macos.yml
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,7 @@ jobs:
org.apache.comet.rules.CometScanContribSuite
org.apache.comet.rules.CometScanSchemeFallbackSuite
org.apache.comet.rules.CometExecRuleSuite
org.apache.comet.rules.CometPlanOnlySuite
org.apache.comet.rules.RevertNativeForTransitionHeavyStagesSuite
org.apache.spark.sql.CometTPCDSQuerySuite
org.apache.spark.sql.CometTPCDSQueryTestSuite
Expand Down
53 changes: 53 additions & 0 deletions docs/source/user-guide/latest/understanding-comet-plans.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
13 changes: 13 additions & 0 deletions spark/src/main/scala/org/apache/comet/CometConf.scala
Original file line number Diff line number Diff line change
Expand Up @@ -741,6 +741,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")
Expand Down
12 changes: 11 additions & 1 deletion spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
Original file line number Diff line number Diff line change
Expand Up @@ -672,6 +672,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"""
Expand All @@ -682,7 +692,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

Expand Down
Loading
Loading