You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/02 19:30:55 UTC
(datafusion-comet) 02/02: docs: explain which Spark operators Comet leaves in place (#6551)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 0fdfaad708d5c82195269134e0621dd959e2c0c9
Author: Andy Grove <ag...@apache.org>
AuthorDate: Fri Oct 2 18:45:27 2026 +0000
docs: explain which Spark operators Comet leaves in place (#6551)
* docs: explain which Spark operators Comet leaves in place
Add a section to the operator contributor guide on plan nodes Comet should
not convert: wrappers and scheduling nodes (including ResultQueryStageExec),
typed Dataset operators, and driver-side commands, plus what to do when a new
Spark version adds a wrapper node.
In the user guide's operator page, explain that wrapper nodes in a plan are
not fallbacks, and list the typed Dataset operators under "Not currently
planned".
* docs: say what Comet's Iceberg write row on the operators page covers
Name Iceberg in the operator column, describe the split-operator and native
writer settings and the plan nodes each produces, list the supported
statements, and note that merge-on-read writes fall back.
---
.../contributor-guide/adding_a_new_operator.md | 38 ++++++++++++++++++++++
docs/source/user-guide/latest/operators.md | 20 +++++++++---
2 files changed, 53 insertions(+), 5 deletions(-)
diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md
index 7f95cb4bd2..20d526b150 100644
--- a/docs/source/contributor-guide/adding_a_new_operator.md
+++ b/docs/source/contributor-guide/adding_a_new_operator.md
@@ -118,6 +118,44 @@ type, they are converted to `ScanExec` in the native plan. This allows them to s
execution blocks. The original Spark operator is wrapped with `CometScanWrapper` or `CometSinkPlaceHolder` which
manages the boundary between JVM and native execution.
+### Operators That Should Not Be Converted
+
+Before adding an operator, check that converting it would speed anything up. Comet deliberately
+leaves three kinds of Spark plan nodes in place.
+
+**Wrappers and scheduling nodes.** `AdaptiveSparkPlanExec`, the AQE query stages
+(`ShuffleQueryStageExec`, `BroadcastQueryStageExec`, `TableCacheQueryStageExec`, and, on Spark 4.0
+and later, `ResultQueryStageExec`), `AQEShuffleReadExec`, `InputAdapter`, `WholeStageCodegenExec`,
+`ReusedExchangeExec`, and `ReusedSubqueryExec` do no data processing of their own. They schedule
+stages, mark whole-stage code generation boundaries, choose which shuffle blocks each task reads,
+or point at a plan that runs elsewhere. AQE creates the query stages itself, after Comet's rules
+have run on the plan inside them, and depends on their exact class. For example, it casts the root
+of the final plan to `ResultQueryStageExec`. When a query stage wraps a Comet shuffle, broadcast,
+or cached relation, `CometExecRule` reads from it as a native input through `CometExchangeSink`
+and leaves the stage itself in place.
+
+**Operators that run user JVM code on JVM objects.** The typed `Dataset` API plans
+`DeserializeToObjectExec`, `SerializeFromObjectExec`, `MapElementsExec`, `MapPartitionsExec`,
+`AppendColumnsExec`, `AppendColumnsWithObjectExec`, `MapGroupsExec`, and `CoGroupExec`. They
+convert rows to JVM objects, run an arbitrary user function on those objects, or convert them back,
+and most of them pass the objects to the next operator as an `ObjectType` column, which has no
+Arrow representation. None of this can run natively. Per-row operators can still stay inside a
+Comet plan: `MapElementsExec`, the operator behind `Dataset.map`, generates its call to the user
+function as a Catalyst `Invoke` expression, so the deserializer, the call, and the serializer could
+run together as one projection in the JVM codegen dispatcher. `mapPartitions`, `mapGroups`, and
+`cogroup` pass the user function an iterator or a whole group, so there is no per-row expression to
+build. A typed `filter` is planned as an ordinary `FilterExec`, not as one of these operators.
+
+**Driver-side commands.** `ExecutedCommandExec` runs a `RunnableCommand`, such as DDL or `SET`, on
+the driver, so there is no data path for Comet to accelerate.
+
+If a new Spark version adds a wrapper node, do not write a serde for it. Add it to the nodes that
+`ExtendedExplainInfo.generateTreeString` skips when counting operators, and to the wrapper list in
+[Understanding Comet Plans](../user-guide/latest/understanding-comet-plans.md), so the coverage
+summary does not count it as a Spark operator. If `CometExecRule` visits the node, also add it to
+the operators it leaves in place without recording a fallback reason. `ExtendedExplainInfo`
+already skips every `QueryStageExec`, so a new query stage type needs no change there.
+
## Implementing a Native Operator
This section focuses on adding a native operator, which is the most common and complex case.
diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md
index a47055df7e..88fdc30ddc 100644
--- a/docs/source/user-guide/latest/operators.md
+++ b/docs/source/user-guide/latest/operators.md
@@ -37,6 +37,15 @@ all native execution can be turned off with `spark.comet.exec.enabled=false`. Se
| ⚠️ Supported (caveats) | Experimental or disabled by default, or accelerates only a limited subset. See the [Compatibility Guide](compatibility/index.md). |
| 🔜 Planned | Intended; tracked by an open issue or pull request. |
+## Wrapper nodes
+
+Some nodes in a Spark plan do no work of their own: `AdaptiveSparkPlan`, the AQE query stages
+(including `ResultQueryStage` on Spark 4.0 and later), `AQEShuffleRead`, `InputAdapter`,
+`WholeStageCodegen`, and the reuse markers `ReusedExchange` and `ReusedSubquery`. Comet leaves
+them in place around the operators it converts, so this page does not list them, and seeing one in
+a plan does not mean that part of the query fell back to Spark. The coverage summary described in
+[Understanding Comet Plans](understanding-comet-plans.md) does not count them.
+
## Not currently planned
The following operator families fall back to Spark and are not on the current roadmap. They are
@@ -46,6 +55,7 @@ omitted from the tables below and may be reconsidered based on demand:
- **Cartesian / cross joins** (`CartesianProductExec`): rare and expensive, with little acceleration benefit.
- **Range generation** (`RangeExec`): niche leaf operator.
- **Pickled (non-Arrow) Python UDFs** (`BatchEvalPythonExec`): Comet accelerates Arrow-based Python UDFs only ([#4234](https://github.com/apache/datafusion-comet/pull/4234)).
+- **Typed Dataset operators** (`DeserializeToObjectExec`, `SerializeFromObjectExec`, `MapElementsExec`, `MapPartitionsExec`, `MapGroupsExec`, `CoGroupExec`, `AppendColumnsExec`, and similar): produced by `map`, `mapPartitions`, `groupByKey`, `cogroup`, and other typed `Dataset` transformations. They exist to run user JVM functions on JVM objects, which Comet cannot do natively.
## Scans
@@ -121,11 +131,11 @@ natively on `BroadcastHashJoinExec` and `ShuffledHashJoinExec`. Existence sort-m
## Writes
-| Operator | Status | Notes |
-| -------------------------------------------------------------------------------------------------- | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. |
-| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. |
-| `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Apache Iceberg tables only. Experimental, disabled by default. See [Iceberg Writes](iceberg-writes.md). |
+| Operator | Status | Notes [...]
+| ------------------------------------------------------------------------------------------------------------------ | ------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
+| `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. [...]
+| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. [...]
+| Iceberg writes: `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Experimental, disabled by default. `spark.comet.write.iceberg.splitOperator.enabled=true` plans the write as `IcebergWrite` and `IcebergCommit`, and Iceberg's Java writer still writes the data files. `spark.comet.write.iceberg.enabled=true` then writes eligible data files natively (`CometIcebergWrite`). Covers `INSERT INTO`, `INSERT OVERWRITE`, and copy-on-write [...]
## Python and UDF
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org