# Understanding Comet Plans This guide explains how to read a Spark query plan once Comet is enabled, what happens when parts of a plan fall back to Spark, and which configs to use to inspect that behavior. ## Overview When Comet is enabled, the `CometSparkSessionExtensions` rules walk the physical plan bottom-up and replace Spark operators with Comet equivalents where possible. Consecutive Rust-implemented operators are combined into a single block that is serialized as protobuf and executed by DataFusion on the executor. Operators that Comet does not support remain as their original Spark form. As a result, a plan can mix three kinds of nodes: - **`Comet*` nodes** that are implemented in Rust (for example `CometProject`, `CometHashAggregate`). - **`Comet*` nodes that run on the JVM** but are still part of the Comet pipeline (for example `CometBroadcastExchange`, `CometColumnarExchange`). - **Standard Spark nodes** (for example `Project`, `HashAggregate`) where Comet either does not support the operator or has fallen back due to an unsupported expression, data type, or configuration. Wherever data crosses between columnar and row-based execution, Comet inserts a transition node such as `CometColumnarToRow` or `CometSparkRowToColumnar`. ## Reading a Plan You can print a plan with `df.explain("formatted")` or `EXPLAIN FORMATTED `, and the same plan is shown in the Spark SQL UI. When reading a plan, look for: - **Node prefix.** `Comet*` nodes are accelerated by Comet. Anything without the prefix is unmodified Spark. - **Transitions.** `CometColumnarToRow`, `CometNativeColumnarToRow`, and `CometSparkRowToColumnar` mark boundaries between columnar Comet execution and row-based Spark execution. Frequent transitions usually indicate fallback inside the plan. - **Exchange type.** `CometExchange` is the native shuffle path, `CometColumnarExchange` is the JVM columnar shuffle path, and a plain `Exchange` means Spark shuffle. See [Shuffle Operators](#shuffle-operators) below. ## Fallback A "fallback" happens when Comet cannot run part of a plan in the Comet pipeline. Fallback can be partial (a subtree falls back to Spark while the rest stays in the Comet pipeline) or full (no Comet nodes appear). Common reasons: - The Spark operator is not supported by Comet. - An expression inside an otherwise supported operator is not supported, or is marked incompatible and the per-expression opt-in `spark.comet.expression..allowIncompatible=true` is not set. Operators have an equivalent `spark.comet.operator..allowIncompatible` opt-in. - A data type is not supported by the operator. - A configuration setting disables a specific operator or expression. See [Supported Spark Operators](operators.md) and [Supported Expressions](expressions.md) for current coverage, and the [Compatibility Guide](compatibility/index.md) for incompatibility details. ## Configs for Inspecting Plans and Fallback Comet provides five configs for understanding what is happening in a plan. They serve different purposes and produce output in different places. | Config | Output destination | What you see | | ------------------------------------------ | ---------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `spark.comet.explain.fallback.enabled` | Driver log (only when fallback) | A WARN with the list of reasons each query stage could not run in Comet. | | `spark.comet.explain.fallback.log.enabled` | Driver log | One WARN per fallback reason as it is encountered, without surrounding plan context. | | `spark.comet.explain.codegen.enabled` | `ExtendedExplainInfo` / SQL UI | A single `[COMET-INFO: JVM codegen dispatcher: , , ...]` segment naming each expression on the operator that was routed through the JVM codegen dispatcher. | | `spark.comet.explain.format` | Spark SQL UI (Spark 4.0 and newer) | Annotated plan or fallback-reason list, depending on `verbose` (default) or `fallback` value. | | `spark.comet.explain.native.enabled` | Executor logs, per task | The DataFusion plan with metrics, useful for inspecting Rust execution. | ### `spark.comet.explain.fallback.enabled` Logs a single WARN listing the reasons each query stage could not be executed in Comet. Nothing is logged when the entire stage runs in Comet. Useful as a low-noise check that fallback is or is not happening. ### `spark.comet.explain.fallback.log.enabled` Logs every fallback reason as it is encountered, one WARN per reason. Use this when you want to see all reasons, including ones that `spark.comet.explain.fallback.enabled` may aggregate or omit. The output does not include the surrounding plan, so it is best for accumulating diagnostics across many queries. ### `spark.comet.explain.codegen.enabled` Disabled by default. When enabled, Comet annotates the surrounding Comet operator (`CometProject`, `CometFilter`, etc.) with a single combined `[COMET-INFO: JVM codegen dispatcher: , , ...]` segment listing every expression on that operator that was routed through the JVM codegen dispatcher (Spark's own `doGenCode` running inside a Comet kernel), gated by `spark.comet.exec.scalaUDF.codegen.enabled=true`. The projection stays Comet-native — this is informational only, useful for distinguishing "runs natively in DataFusion" from "runs Spark's generated code inside a Comet kernel". The annotation only appears for expressions Comet actually routed through the dispatcher on this plan — either because no native DataFusion implementation exists (`CometCodegenDispatch` serdes such as `hypot`, `pmod`, `months_between`), or because a native path exists but declined this specific input (`CodegenDispatchFallback` mixins). It is not a general "this expression ran here" marker. Example: ```scala spark.conf.set("spark.comet.exec.scalaUDF.codegen.enabled", "true") spark.conf.set("spark.comet.explain.codegen.enabled", "true") val df = spark.sql("SELECT hypot(a, b), pmod(a, b) FROM t") println(new org.apache.comet.ExtendedExplainInfo() .generateExtendedInfo(df.queryExecution.executedPlan)) ``` Output: ``` CometColumnarToRow +- CometProject [COMET-INFO: JVM codegen dispatcher: hypot, pmod] +- CometNativeScan parquet spark_catalog.default.t Comet accelerated 2 out of 2 eligible operators (100%). Final plan contains 1 transitions between Spark and Comet. Accelerated expressions: 0 native, 2 codegen dispatch. ``` Note that the operator is still `CometProject` (Comet-accelerated); only the per-expression evaluation for `hypot` and `pmod` takes the JVM codegen path. ### `spark.comet.explain.format` This config is read by `org.apache.comet.ExtendedExplainInfo`, which Spark loads via the `spark.sql.extendedExplainProviders` mechanism added in Spark 4.0. Add the provider: ```shell --conf spark.sql.extendedExplainProviders=org.apache.comet.ExtendedExplainInfo ``` The Spark SQL UI then shows an additional section under the detailed plan. The format is controlled by `spark.comet.explain.format`: - `verbose` (default): the full plan annotated with fallback reasons, plus a summary of how much of the plan is accelerated. The summary reports operator coverage, the number of Spark/Comet transitions, and how many distinct expressions Comet accelerated - split into those lowered to native DataFusion expressions and those routed through the JVM codegen dispatcher. Expression names are counted once per plan, and structural nodes (attribute references, literals, aliases) are excluded. - `fallback`: a list of fallback reasons only. This is the most convenient option on Spark 4.0 because the output is shown inline in the UI. Earlier Spark versions do not have the `extendedExplainProviders` extension point, so this provider is not used and the config has no effect there. Not every node in the plan is an eligible operator. The following are excluded from both operator counts: - Transition nodes (`CometColumnarToRow`, `CometNativeColumnarToRow`, `CometSparkRowToColumnar`, `CometSparkColumnarToColumnar`, `ColumnarToRow`, `RowToColumnar`), which are reported separately as the transition count. `CometSparkRowToColumnar` and `CometSparkColumnarToColumnar` are the two names a single operator renders under, depending on whether its child already produces columnar data, and both are excluded. - Wrappers that do no work of their own: `AdaptiveSparkPlan`, `InputAdapter`, `WholeStageCodegen`, query stages, and `AQEShuffleRead`. - The reuse marker `ReusedSubquery`. The subquery it points at is counted where that subquery is shown, so the marker itself does not add to the totals. - The reuse marker `ReusedExchange`. Walking the plan replaces the marker with the exchange it reuses, so the reused subtree is shown in full at every reference. The exchange runs once, so the operators and transitions in that subtree are counted once for the whole plan, where the subtree is first shown. ### `spark.comet.explain.native.enabled` When enabled, each executor task logs the DataFusion plan it executes, along with metrics. This is verbose because there is one plan per task, but it is the only way to see the plan as DataFusion sees it (including how operators were arranged after Comet's serialization). See the [Metrics Guide](metrics.md) for details on the DataFusion metrics that appear in this output. ### `spark.comet.explain.planOnly.enabled` When enabled, Comet plans every query as it normally would, logs the resulting Comet plan and coverage summary to the driver log with the prefix `[Comet plan-only]`, and then executes the original plan on Spark. Use it to estimate how much of a workload Comet would accelerate without changing execution. The report is the same annotated plan that `spark.comet.explain.format=verbose` produces, and reflects Comet's post-columnar rules, so a stage Comet would revert to Spark for having too many transitions is reported as reverted. Requires `spark.comet.exec.enabled=true`. Keep the following in mind when reading the reports: - Spark plans scalar and dynamic partition pruning subqueries separately from the query that contains them, so each gets its own report alongside the outer query's. The outer report also counts its subqueries, so do not add the reports together. - Only the JVM side of planning runs. Anything that would fail when DataFusion builds the native plan still counts as accelerated, so treat the percentage as an upper bound. - Comet's split Iceberg V2 write (`spark.comet.write.iceberg.enabled`) is declined in plan-only mode, so such writes run on, and are reported as, Spark. - Under AQE the report describes the plan before any adaptive re-planning, so coverage of the plan that finally executes can differ. In particular, AQE plans subqueries into the outer query only after the report is produced, so the outer report counts their operators as Spark. For subquery-heavy queries, read the per-subquery reports or disable AQE for the evaluation run. ## Programmatic Access to Fallback Reasons The configs above route fallback reasons to logs or the SQL UI. If you want the reasons for a specific query as data — for example, to assert in a test or to drive a custom report — use `org.apache.comet.ExtendedExplainInfo` directly. It is the same class that backs `spark.comet.explain.format`, and it works on any Spark version because it does not rely on the Spark 4.0 `extendedExplainProviders` extension point. ```scala import org.apache.comet.ExtendedExplainInfo val df = spark.sql("SELECT ...") val plan = df.queryExecution.executedPlan val info = new ExtendedExplainInfo() // Sorted, deduplicated list of fallback reasons across the whole plan. val reasons: Seq[String] = info.getFallbackReasons(plan) // Sorted, deduplicated names of the expressions Comet accelerated, split by how // they run. Structural nodes (attribute references, literals, aliases) are not // reported. val native: Seq[String] = info.getNativeExpressions(plan) val dispatched: Seq[String] = info.getCodegenDispatchExpressions(plan) // Formatted string. Honors spark.comet.explain.format: // - "verbose" -> the full plan annotated with per-node fallback reasons // - "fallback" -> just the list of reasons val formatted: String = info.generateExtendedInfo(plan) ``` A name can appear in both lists: the same function may be lowered natively for one set of arguments and routed through the dispatcher for another elsewhere in the same plan. Example: ``` val df = spark.sql("SELECT from_unixtime(eventTime) FROM events") val plan = df.queryExecution.executedPlan val info = new org.apache.comet.ExtendedExplainInfo() println(info.generateExtendedInfo(plan)) ``` Output ``` Project [COMET: from_unixtime(eventTime#5L, yyyy-MM-dd HH:mm:ss, Some(GMT)) is not fully compatible with Spark. To enable it anyway, set spark.comet.expression.FromUnixTime.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).] +- CometColumnarToRow +- CometNativeScan parquet Comet accelerated 1 out of 2 eligible operators (50%). Final plan contains 1 transitions between Spark and Comet. Accelerated expressions: 0 native, 0 codegen dispatch. ``` ## Comet Operator Reference The following sections describe the Comet nodes you will see in plans, grouped by role. Names match what is shown in the plan output. ### Scans | Node | Description | | ------------------------ | ------------------------------------------------------------------------------------------------------------------ | | `CometBatchScan` | DataSource V2 scan, including Iceberg Parquet, that produces Arrow batches consumed by Comet. | | `CometNativeScan` | Parquet scan that runs entirely in Rust via DataFusion. | | `CometIcebergNativeScan` | Iceberg Parquet scan that runs entirely in Rust via DataFusion. | | `CometCsvNativeScan` | CSV scan that runs entirely in Rust via DataFusion (experimental). | | `CometInMemoryTableScan` | JVM-side scan of a table cached in Comet's Arrow format (experimental, disabled by default). | | `CometLocalTableScan` | JVM-side scan of a `LocalTableScanExec` (disabled by default). | | `CometEmptyRelation` | Empty native input that replaces `EmptyRelationExec` (Spark 4.0 and later). | | `CometRange` | Generates the rows of `spark.range` and SQL `range()` in native code, replacing `RangeExec` (disabled by default). | ### Native Rust Operators These are implemented in Rust and run in DataFusion. When several appear consecutively in a plan, they execute as a single fused block. The write operators (`CometWriteFiles`, `CometNativeWrite` and `CometIcebergWrite`) are the exception: each runs as its own native plan over its child's output. | Node | Spark equivalent | | ------------------------------ | --------------------------------------------------------------------------------- | | `CometProject` | `ProjectExec` | | `CometFilter` | `FilterExec` | | `CometSort` | `SortExec` | | `CometLocalLimit` | `LocalLimitExec` | | `CometGlobalLimit` | `GlobalLimitExec` | | `CometExpand` | `ExpandExec` | | `CometExplode` | `GenerateExec` (for `explode`, `explode_outer`, `posexplode`, `posexplode_outer`) | | `CometHashAggregate` | `HashAggregateExec`, `ObjectHashAggregateExec` | | `CometHashJoin` | `ShuffledHashJoinExec` | | `CometBroadcastHashJoin` | `BroadcastHashJoinExec` | | `CometBroadcastNestedLoopJoin` | `BroadcastNestedLoopJoinExec` | | `CometSortMergeJoin` | `SortMergeJoinExec` | | `CometMergeRows` | `MergeRowsExec` (Spark 3.5+; opt-in) | | `CometWindowExec` | `WindowExec` | | `CometWindowGroupLimitExec` | `WindowGroupLimitExec` (Spark 3.5 and later) | | `CometSample` | `SampleExec` (sampling without replacement) | | `CometTakeOrderedAndProject` | `TakeOrderedAndProjectExec` | | `CometWriteFiles` | `WriteFilesExec` (Spark 4.0 and later, experimental native Parquet writes) | | `CometNativeWrite` | `DataWritingCommandExec` (Spark 3.x, experimental native Parquet writes) | | `CometIcebergWrite` | `IcebergWrite` (native Iceberg data-file writes, opt-in) | ### JVM-Side Operators These keep their data on the JVM but participate in the Comet pipeline. | Node | Notes | | ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------- | | `CometUnion` | JVM-side union of Comet inputs. The Rust side reads each branch as a separate scan. | | `CometCoalesce` | JVM-side partition coalesce. | | `CometCollectLimit` | JVM-side collect limit, equivalent to `CollectLimitExec`. | | `CometBroadcastExchange` | Broadcast exchange producing serialized Arrow batches that the consumer can decode. | | `CometSubqueryBroadcast` | Companion to `CometBroadcastExchange` for dynamic partition pruning subqueries. | | `CometMapInBatch` | Runs `mapInArrow` / `mapInPandas` Python UDFs on Comet's Arrow batches (experimental). See [PyArrow UDF Acceleration](pyarrow-udfs.md). | | `IcebergWrite` | Executor-side Iceberg data-file write in Comet's split-operator Iceberg write plan. See [Iceberg Writes](iceberg.md#writes). | | `IcebergCommit` | Driver-side commit for Comet's split-operator Iceberg write plan. | ### Shuffle Operators Comet has two shuffle implementations and the plan tells you which one is in use: - **`CometExchange`** is the **native shuffle** path. The child must already be a Comet operator producing columnar Arrow batches; the node calls `executeColumnar()` on its child and the partition, encode, and compress steps run in Rust. Range partitioning **keys** must be scalar types. Hash partitioning keys must also be scalar types unless `spark.comet.shuffle.native.partitioning.hash.nested.enabled=true`, which admits struct, array, and (Spark 4.0 and later) map keys. The data columns themselves can include `StructType`, `ArrayType`, and `MapType` since batches are serialized via the Arrow IPC writer. - **`CometColumnarExchange`** is the **JVM columnar shuffle** path. It accepts either Spark row-based input or Comet columnar input, which makes it the fallback when the child is not a Comet operator or when a hash/range key type is not supported by native shuffle (for example, a struct or array hash key while nested hash keys are disabled). It is still preferred over Spark's native shuffle when Comet shuffle is enabled. Keys with a non-default string collation are supported by neither path and use Spark's shuffle. Both paths support the same set of partitioning schemes (`HashPartitioning`, `RangePartitioning`, `RoundRobinPartitioning`, `SinglePartition`) and both can carry complex types in data columns. Native round-robin partitioning is disabled by default; see [Round-Robin Partitioning](compatibility/operators.md#round-robin-partitioning). The choice between the two is automatic. See the [shuffle tuning guide](tuning/shuffle.md) for how to enable Comet shuffle and choose between the implementations. ### Columnar/Row Transitions Comet inserts these nodes wherever data has to cross the columnar/row boundary. Multiple implementations exist because the optimal strategy depends on what produced the columnar data. | Node | Direction | Notes | | ------------------------------ | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `CometColumnarToRow` | columnar → row | JVM-based row conversion. A fork of Spark's `ColumnarToRowExec` that includes the SPARK-50235 fix. This is the default. | | `CometNativeColumnarToRow` | columnar → row | Rust-based row conversion via `NativeColumnarToRowConverter`. Disabled by default; enable with `spark.comet.exec.columnarToRow.native.enabled=true`. It carries a fixed JNI cost per batch and is slower than the JVM conversion for small batches. | | `CometSparkColumnarToColumnar` | columnar → columnar | Converts a Spark columnar input (a non-Comet `ColumnarBatch`) into Comet's Arrow batches. | | `CometSparkRowToColumnar` | row → columnar | Converts a Spark row input into Comet's Arrow batches. | The two `CometSpark*` names come from a single `CometSparkToColumnarExec` operator that picks the node name based on whether its child supports columnar.