From 25b5dba6dd9302c6a489ff5b574eb3b7ff6ea8cf Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 13 Oct 2026 18:50:47 -0600 Subject: [PATCH] feat: make native Iceberg writes opt-in for 1.2.0 spark.comet.write.iceberg.enabled now defaults to false again, so 1.2.0 ships the native Iceberg writer as a supported, opt-in feature instead of turning it on for every eligible write. The rest of #6664 stays: the setting is in the query execution category, it plans the split operator and the native writer by itself, merge-on-read writes keep Spark's operator, and spark.comet.exec.enabled=false keeps Spark's write operator. The Iceberg Spark test diffs and both Iceberg write benchmarks set the setting themselves, so they still run the native writer. CometIcebergWriteActionSuite now checks that an application that sets neither write flag gets Spark's own operator, and that the write flag alone writes an eligible write natively under the split plan. The merge-on-read and transition-reversion tests set the write flag themselves. The user guide, FAQ, Gluten comparison, contributor guide, roadmap and Iceberg write review skill describe the native writer as opt-in, and the 1.2.0 upgrade-guide entry covers the move from 1.1.0's testing settings to the one supported setting. --- .../review-comet-iceberg-write-pr/SKILL.md | 10 +-- docs/source/about/faq.md | 6 +- docs/source/about/gluten_comparison.md | 8 +-- .../contributor-guide/iceberg-spark-tests.md | 3 +- .../contributor-guide/iceberg-writes.md | 10 +-- docs/source/contributor-guide/roadmap.md | 16 ++--- docs/source/user-guide/latest/datasources.md | 4 +- .../user-guide/latest/iceberg-writes.md | 35 +++++----- docs/source/user-guide/latest/iceberg.md | 7 +- .../user-guide/latest/migration-guide.md | 53 +++++--------- docs/source/user-guide/latest/operators.md | 12 ++-- .../scala/org/apache/comet/CometConf.scala | 15 ++-- .../comet/CometIcebergWriteActionSuite.scala | 70 ++++++++++++++----- 13 files changed, 135 insertions(+), 114 deletions(-) diff --git a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md index 7fe363c81a9..f5c832ec898 100644 --- a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md +++ b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md @@ -49,11 +49,11 @@ not, inherits it. There is no query that fails to tell you. | Native writer | `CometIcebergNativeWrite` (gate and serde), `IcebergWriteProtoTranslation`, `CometIcebergWriteExec`, `iceberg_write.rs` | | Shared with the scan | `iceberg_common.rs` (`load_file_io`, `storage_factory_for`, `scheme_of`), `IcebergReflection`, `NativeConfig` S3 translation | -`spark.comet.write.iceberg.enabled` switches both write layers and is on by default, so a change to -the split plan affects every Iceberg write, including writes that never reach the native writer, and -a change to the native writer affects every eligible one. The testing-only -`spark.comet.write.iceberg.splitOperator.enabled` plans the split operator with the native writer -off. A change to a shared file affects the native Iceberg scan too. +`spark.comet.write.iceberg.enabled` switches both write layers. It is off by default, but once a +user turns it on, a change to the split plan affects every Iceberg write, including writes that +never reach the native writer, and a change to the native writer affects every eligible one. The +testing-only `spark.comet.write.iceberg.splitOperator.enabled` plans the split operator with the +native writer off. A change to a shared file affects the native Iceberg scan too. ## 2. Direction of the Gate Change diff --git a/docs/source/about/faq.md b/docs/source/about/faq.md index d92976a1e19..8e411caa693 100644 --- a/docs/source/about/faq.md +++ b/docs/source/about/faq.md @@ -141,9 +141,9 @@ Common causes of a small speedup are: ### Does Comet support Delta Lake, Apache Hudi, or Apache Paimon tables? Not yet. Comet accelerates [Apache Iceberg](../user-guide/latest/iceberg.md) tables, with a native -reader and [native writes](../user-guide/latest/iceberg-writes.md) that are both enabled by default. -It does not accelerate scans of Delta Lake, Hudi, or Paimon tables, so Spark reads those. Native -Delta Lake reads are in development; see the +reader that is enabled by default and opt-in +[native writes](../user-guide/latest/iceberg-writes.md). It does not accelerate scans of Delta Lake, +Hudi, or Paimon tables, so Spark reads those. Native Delta Lake reads are in development; see the [roadmap](../contributor-guide/roadmap.md#delta-lake-support). Hudi and Paimon are not on the roadmap. diff --git a/docs/source/about/gluten_comparison.md b/docs/source/about/gluten_comparison.md index fbf1a6b7faf..4065d19ef83 100644 --- a/docs/source/about/gluten_comparison.md +++ b/docs/source/about/gluten_comparison.md @@ -118,10 +118,10 @@ covers a broader set of table formats overall. Comet provides a native Iceberg scan built on iceberg-rust. It has been tested with Iceberg 1.5 and 1.8 through 1.11 and supports Iceberg spec v1, v2, and v3, schema evolution, time travel and branch reads, positional and equality deletes and deletion vectors, encrypted v3 tables, REST catalogs, and S3-compatible object storage. For Iceberg -writes, Comet writes the data files of eligible writes natively through iceberg-rust by default, and iceberg-java -writes the rest and commits every write (see [Comet Iceberg writes]). Comet does not currently provide native -integrations for Delta Lake, Hudi, or Paimon. See the [Comet Iceberg guide] for the full list of supported features -and known limitations. +writes, Comet can write the data files of eligible writes natively through iceberg-rust, an opt-in feature, and +iceberg-java writes the rest and commits every write (see [Comet Iceberg writes]). Comet does not currently provide +native integrations for Delta Lake, Hudi, or Paimon. See the [Comet Iceberg guide] for the full list of supported +features and known limitations. [Comet Iceberg guide]: /user-guide/latest/iceberg.md [Comet Iceberg writes]: /user-guide/latest/iceberg-writes.md diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md b/docs/source/contributor-guide/iceberg-spark-tests.md index 83724feb6b8..dcee0903436 100644 --- a/docs/source/contributor-guide/iceberg-spark-tests.md +++ b/docs/source/contributor-guide/iceberg-spark-tests.md @@ -34,7 +34,8 @@ Here is an overview of the changes that the diffs make to Iceberg: - Enable Comet's Iceberg write path, the split-operator plan (`IcebergCommit -> IcebergWrite`) and the native (iceberg-rust) Parquet writer, alongside the native scan in every Comet-configured session, so that Iceberg's write, commit, and row-level-operation tests run through it. `spark.comet.write.iceberg.enabled` switches both - and is on by default; the diffs set it explicitly, together with the testing-only + and is off by default for users, so these suites are where it runs against Iceberg's own tests. The diffs set + it explicitly, together with the testing-only `spark.comet.write.iceberg.splitOperator.enabled`. That flag is redundant for the normal write path, but it is still what plans merge-on-read (`WriteDelta`) writes as the split plan on Spark 3.5+, so Iceberg's row-level-operation tests exercise that path. See [#5259] diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index 376988d8b75..05653338c9c 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -37,11 +37,11 @@ Two layers, the second built on the first, both switched by `spark.comet.write.i | The split-operator plan | The plan shape. Spark's single V2 write operator becomes `IcebergCommit` over `IcebergWrite`, and iceberg-java writes the data files inside it. | | The native writer | Who writes the data files. An eligible `IcebergWrite` becomes `CometIcebergWrite`, which writes Parquet with iceberg-rust. | -The flag defaults to `true` since Comet 1.2.0, so a change to the split plan reaches every Iceberg -write and a change to the native writer reaches every eligible one. Set to `false`, it plans Spark's -own write operator. The testing-only `spark.comet.write.iceberg.splitOperator.enabled` plans the -split operator with the native writer off, which the suites use to compare the two writers under -the same plan. The rollout and its criteria are tracked in +The flag defaults to `false`, which plans Spark's own write operator. With it on, a change to the +split plan reaches every Iceberg write and a change to the native writer reaches every eligible one. +The testing-only `spark.comet.write.iceberg.splitOperator.enabled` plans the split operator with the +native writer off, which the suites use to compare the two writers under the same plan. The +criteria for turning the flag on by default are tracked in [#5644](https://github.com/apache/datafusion-comet/issues/5644) under the epic [#5649](https://github.com/apache/datafusion-comet/issues/5649). diff --git a/docs/source/contributor-guide/roadmap.md b/docs/source/contributor-guide/roadmap.md index aca1712afdd..7913230f10e 100644 --- a/docs/source/contributor-guide/roadmap.md +++ b/docs/source/contributor-guide/roadmap.md @@ -137,14 +137,14 @@ enabled by default ([#1625]). ## Iceberg Table Writes -Since Comet 1.2.0, Comet writes Iceberg tables natively by default, controlled by one setting, -`spark.comet.write.iceberg.enabled` ([#5644]). Comet splits Spark's Iceberg V2 write operator into separate writer -and committer operators, so the query feeding the write becomes visible to AQE and to Comet's columnar rules -([#4658]), and delegates each task's Parquet write to `iceberg-rust` when the write passes an eligibility check -([#5361]); writes that don't pass keep Iceberg Java's writer. Merge-on-read writes keep Spark's write operator -until the native writer can write them ([#6240]). The remaining work toward the original goal of [#4322], an ETL job that runs end to end in native code, is -tracked in [#5649]: correctness fixes, failure handling that matches Iceberg Java, and broader coverage. See -[Iceberg Writes](iceberg-writes.md) for how the write path works. +Comet can write Iceberg tables natively, controlled by one setting, `spark.comet.write.iceberg.enabled`, which is +off by default. Comet splits Spark's Iceberg V2 write operator into separate writer and committer operators, so the +query feeding the write becomes visible to AQE and to Comet's columnar rules ([#4658]), and delegates each task's +Parquet write to `iceberg-rust` when the write passes an eligibility check ([#5361]); writes that don't pass keep +Iceberg Java's writer. Merge-on-read writes keep Spark's write operator until the native writer can write them +([#6240]). The remaining work toward the original goal of [#4322], an ETL job that runs end to end in native code, +is tracked in [#5649]: correctness fixes, failure handling that matches Iceberg Java, broader coverage, and enabling +the setting by default ([#5644]). See [Iceberg Writes](iceberg-writes.md) for how the write path works. [#4322]: https://github.com/apache/datafusion-comet/issues/4322 [#4658]: https://github.com/apache/datafusion-comet/pull/4658 diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 6f96171f244..58b68e2410a 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -29,8 +29,8 @@ Arrow format, allowing the Comet pipeline to take over after that, but the proce ### Apache Iceberg -Comet accelerates Iceberg scans of Parquet files and, since Comet 1.2.0, writes the data files of -eligible Iceberg writes natively. See the [Iceberg Guide] and [Iceberg Writes](iceberg-writes.md) +Comet accelerates Iceberg scans of Parquet files and can write the data files of eligible Iceberg +writes natively, an opt-in feature. See the [Iceberg Guide] and [Iceberg Writes](iceberg-writes.md) for more information. [iceberg guide]: iceberg.md diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index c314ba1da37..5d50d34cc18 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -19,13 +19,15 @@ # Iceberg Writes: Comet's Split-Operator Plan and Native Writer -Since Comet 1.2.0, Comet plans Iceberg writes with the split-operator plan described below, and -writes the data files of each eligible write natively with iceberg-rust. A write that is not -eligible falls back to iceberg-java's writer. Natively written files differ from iceberg-java's in -the ways listed under [Accepted divergences](#accepted-divergences). +With `spark.comet.write.iceberg.enabled=true`, Comet plans Iceberg writes with the split-operator +plan described below, and writes the data files of each eligible write natively with iceberg-rust. +A write that is not eligible falls back to iceberg-java's writer. Natively written files differ +from iceberg-java's in the ways listed under [Accepted divergences](#accepted-divergences). -`spark.comet.write.iceberg.enabled` switches both. Set it to `false` to plan Spark's own write -operator instead, as Comet 1.1.0 did, so that iceberg-java writes every data file. +The setting is off by default, so Spark plans its own write operator and iceberg-java writes every +data file. Validate the native writer against your own workloads before enabling it, including +the costs described under [Native Parquet write eligibility](#native-parquet-write-eligibility) +and [Failure handling](#failure-handling). ## Overview @@ -37,8 +39,8 @@ data-file writing cannot be re-planned in response to how its input ran. And bec writing is bundled with the metadata and commit steps, there is no separate step for Comet to replace. -When `spark.comet.write.iceberg.enabled=true`, the default, Comet rewrites eligible Iceberg writes -into two operators: +When `spark.comet.write.iceberg.enabled=true`, Comet rewrites eligible Iceberg writes into two +operators: 1. **`IcebergWrite`** — writes the data files on the executors, exactly as iceberg-java does today, and returns each task's serialized commit message. This operator and the sub-query @@ -86,9 +88,8 @@ already has, even if they do not fill its first page. ## Configuration -Iceberg writes need only the standard Comet and Iceberg setup (see [`iceberg.md`](iceberg.md)). -`spark.comet.write.iceberg.enabled` is on by default; set it to `false` to plan Spark's own write -operator instead. +Standard Comet + Iceberg setup (see [`iceberg.md`](iceberg.md)) plus the write setting, which is +off by default: ``` # Standard Comet / Iceberg wiring @@ -98,6 +99,9 @@ spark.sql.catalog.=org.apache.iceberg.spark.SparkCatalog spark.sql.catalog..type=hadoop # or hive / glue / rest / ... spark.sql.catalog..warehouse=... +# Split-operator plan and native Parquet writer +spark.comet.write.iceberg.enabled=true + # Lets writes whose input is a local relation (INSERT ... VALUES, a local DataFrame) use the # native writer; see "Native Parquet write eligibility" below spark.comet.exec.localTableScan.enabled=true @@ -139,7 +143,7 @@ changes. The rewrite is skipped — and the write runs through Spark's stock combined operator — when: -- `spark.comet.write.iceberg.enabled` is set to `false` and the testing setting +- `spark.comet.write.iceberg.enabled` is `false` (the default) and the testing setting `spark.comet.write.iceberg.splitOperator.enabled`, which Comet 1.1.0 documented as a user setting, is not on; - Comet is disabled (`spark.comet.enabled=false`), or its native execution is @@ -160,7 +164,7 @@ trade-off, only no plan change. ## Native Parquet write eligibility -When `spark.comet.write.iceberg.enabled=true`, the default +When `spark.comet.write.iceberg.enabled=true` ([#5361](https://github.com/apache/datafusion-comet/pull/5361)), the `IcebergWrite` operator's per-task Parquet write is delegated to [iceberg-rust](https://github.com/apache/iceberg-rust). The native writer must produce the same outcome as iceberg-java — the same Parquet features, @@ -376,9 +380,8 @@ unchanged. ## Accepted divergences Some differences between parquet-mr and the pinned parquet-rs / iceberg-rust are unconditional — -they apply to every native write and cannot be configured away. Leaving -`spark.comet.write.iceberg.enabled` at its default of `true` accepts them, and setting it to -`false` avoids them. They fall into three classes with very +they apply to every native write and cannot be configured away. Enabling +`spark.comet.write.iceberg.enabled` accepts them. They fall into three classes with very different blast radius: differences confined to the physical bytes of a data file (cosmetic — no reader decision is based on them), differences visible in manifest metadata (these outlive the write and feed later readers' pruning decisions, so each one is analyzed individually diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index af9d3d61a13..81fbac586c6 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -198,9 +198,10 @@ The following scenarios will fall back to the JVM Iceberg reader: change. The check uses the table's schema history, so the fallback stays after those files are rewritten -Writes are not covered by this list. By default Comet plans an Iceberg write with its split-operator -plan and writes eligible data files natively; see [Iceberg Writes](iceberg-writes.md) for the plan, -the native writer, and when each applies. +Writes are not covered by this list. By default Iceberg writes use Spark's own writer. With +`spark.comet.write.iceberg.enabled=true`, Comet plans an Iceberg write with its split-operator plan +and writes eligible data files natively; see [Iceberg Writes](iceberg-writes.md) for the plan, the +native writer, and when each applies. ### Iceberg UDFs diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md index 8b983d02118..a3de441b39d 100644 --- a/docs/source/user-guide/latest/migration-guide.md +++ b/docs/source/user-guide/latest/migration-guide.md @@ -113,42 +113,23 @@ operators, such as the scan of a Data Source V2 connector. ### Iceberg Writes -`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`, -`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite` -under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of -each eligible write natively with iceberg-rust. A write that is not eligible still uses -iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows, -but natively written files differ from iceberg-java's in the ways listed under -[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI -show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator, -as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the -testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the -new default unless it sets `spark.comet.write.iceberg.enabled=false`. -`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to -`false` does not turn the split operator off. Comet 1.1.0 documented it as the switch for the split -plan, so a deployment that still sets it to `true` keeps the split plan when -`spark.comet.write.iceberg.enabled=false`, with iceberg-java writing the data files, and on Spark -3.5+ it also plans merge-on-read writes as the split plan. Unset that setting as well as setting -`spark.comet.write.iceberg.enabled=false` to plan Spark's own operator. See -[Iceberg Writes](iceberg-writes.md). - -To compute the Iceberg metrics of a natively written file, Comet reads the file's Parquet footer -back from storage, where iceberg-java's writer takes them from memory. On S3 or GCS that is two GET -requests per file, which adds up for a write that produces many small files. See -[Native Parquet write eligibility](iceberg-writes.md#native-parquet-write-eligibility). - -The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers -sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and -each open file holds the row group it is writing, so a task that writes to many partitions can need -more memory than the pool grants it. The task then writes out and closes the partitions holding the -most memory, whose next rows open new files, so the write finishes with more, smaller files than -iceberg-java's would. Disabling the fanout writer (`write.spark.fanout.enabled=false`) or a larger -`spark.memory.offHeap.size` avoids the extra files. A write that keeps one file open, unpartitioned -or clustered, has no partition to close, so when that file outgrows the pool the task fails with -`Additional allocation failed for IcebergWriteExec`. A smaller `write.parquet.row-group-size-bytes` -or a larger `spark.memory.offHeap.size` lets such a write fit. With -`spark.comet.write.iceberg.enabled=false`, iceberg-java writes both, as before. See -[Failure handling](iceberg-writes.md#failure-handling). +Comet's Iceberg write path is now switched by one setting, `spark.comet.write.iceberg.enabled`, +which stays off by default. With it on, Comet plans an Iceberg `INSERT INTO`, `INSERT OVERWRITE`, +and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite` under +`IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of each +eligible write natively with iceberg-rust. A write that is not eligible uses iceberg-java's writer, +and iceberg-java commits every write. Before turning it on, see [Iceberg Writes](iceberg-writes.md) +for how natively written files differ from iceberg-java's, the footer reads the native path makes on +S3 or GCS, and how the native writer draws on Comet's off-heap memory pool. + +Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the testing category, and it +took effect only together with `spark.comet.write.iceberg.splitOperator.enabled`. Comet now ignores +the old name, so a deployment that turned native writes on that way must set +`spark.comet.write.iceberg.enabled=true` to keep them. +`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting. Comet 1.1.0 documented +it as the switch for the split plan, and a deployment that still sets it to `true` keeps the split +plan, with iceberg-java writing the data files, and on Spark 3.5+ it now also plans merge-on-read +writes as the split plan. Unset it to plan Spark's own operator. ## Upgrading to Comet 1.1.0 diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index d63548ed805..9a0a430fc86 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -132,12 +132,12 @@ 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. | -| Iceberg writes: `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Planned as `IcebergWrite` and `IcebergCommit` by default (`spark.comet.write.iceberg.enabled`), and eligible data files are written natively (`CometIcebergWrite`). Other writes use Iceberg's Java writer. Covers `INSERT INTO`, `INSERT OVERWRITE`, and copy-on-write `DELETE` / `UPDATE` / `MERGE`. Merge-on-read writes (`WriteDeltaExec`) fall back. See [Iceberg Writes](iceberg-writes.md). | -| `MergeRowsExec` | ⚠️ | Spark 3.5+. Experimental, disabled by default (`spark.comet.exec.mergeRows.enabled`). On Spark 4.1+, stock V2 writers retain Spark MergeRowsExec; Comet's split Iceberg path can run it natively while preserving MergeSummary. See [MERGE INTO](compatibility/operators.md#merge-into-mergerowsexec). | +| 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` | ⚠️ | Disabled by default (opt-in). `spark.comet.write.iceberg.enabled=true` plans the write as `IcebergWrite` and `IcebergCommit` and writes eligible data files natively (`CometIcebergWrite`). Other writes use Iceberg's Java writer. Covers `INSERT INTO`, `INSERT OVERWRITE`, and copy-on-write `DELETE` / `UPDATE` / `MERGE`. Merge-on-read writes (`WriteDeltaExec`) fall back. See [Iceberg Writes](iceberg-writes.md). | +| `MergeRowsExec` | ⚠️ | Spark 3.5+. Experimental, disabled by default (`spark.comet.exec.mergeRows.enabled`). On Spark 4.1+, stock V2 writers retain Spark MergeRowsExec; Comet's split Iceberg path can run it natively while preserving MergeSummary. See [MERGE INTO](compatibility/operators.md#merge-into-mergerowsexec). | ## Python and UDF diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index da376f8cbc9..f844f7e8f80 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -135,15 +135,14 @@ object CometConf extends ShimCometConf { val COMET_ICEBERG_NATIVE_WRITE_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.write.iceberg.enabled") .category(CATEGORY_EXEC) - .doc( - "Whether Comet plans Iceberg writes and writes their data files natively. Comet " + - "replaces Spark's combined V2 write operator with a file writer (inside AQE) under a " + - "committer (outside AQE), and writes the data files of each eligible write with its " + - "native (iceberg-rust) writer. Other writes use Iceberg's own writer, and Iceberg " + - "commits every write. Merge-on-read writes keep Spark's operator. Set this to false " + - "to plan Spark's own V2 write operator.") + .doc("Whether Comet plans Iceberg writes and writes their data files natively. When " + + "enabled, Comet replaces Spark's combined V2 write operator with a file writer " + + "(inside AQE) under a committer (outside AQE), and writes the data files of each " + + "eligible write with its native (iceberg-rust) writer. Other writes use Iceberg's own " + + "writer, and Iceberg commits every write. Merge-on-read writes keep Spark's operator. " + + "When disabled, Spark plans its own V2 write operator.") .booleanConf - .createWithDefault(true) + .createWithDefault(false) val COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT: ConfigEntry[Int] = conf("spark.comet.scan.icebergNative.dataFileConcurrencyLimit") diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 5910527a774..c5668976c4c 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -149,9 +149,9 @@ class CometIcebergWriteActionSuite } // Comet 1.1.0 documented the split flag as a user setting, so a deployment can carry it into 1.2.0. - // With it on, `spark.comet.write.iceberg.enabled=false` only turns the native writer off: the split - // plan stays, with iceberg-java writing the files. The upgrade guide tells such deployments to - // unset the split flag to get Spark's own operator back. + // With it on and `spark.comet.write.iceberg.enabled` off, the split plan stays, with iceberg-java + // writing the files. The upgrade guide tells such deployments to unset the split flag to get + // Spark's own operator back. test( "the split flag keeps the split plan with iceberg-java's writer when the write flag is off") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -178,9 +178,10 @@ class CometIcebergWriteActionSuite } // The suite pins both write flags, so this test unsets them to see what an application that - // sets neither gets. A Parquet scan is a native input, so the write is eligible. + // sets neither gets: Spark's own write operator, even for a write the native writer could take. + // A Parquet scan is a native input, so the write is eligible. // https://github.com/apache/datafusion-comet/issues/5644 - test("an eligible Iceberg write runs natively under the split plan when no flag is set") { + test("an eligible Iceberg write keeps Spark's own write plan when no flag is set") { assume(icebergAvailable, "Iceberg not available in classpath") withIcebergCatalog { _ => withTempPath { dir => @@ -201,6 +202,40 @@ class CometIcebergWriteActionSuite } } assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got ${snapshot.snapshotDelta}") + val (commits, writes) = collectIcebergWriteOps(snapshot.plans) + assert( + commits.isEmpty && writes.isEmpty, + "expected Spark's own write plan with no flag set. Plans:\n" + + snapshot.plans.mkString("\n--\n")) + assertRows("write_defaults", expectedIds = 0 until 10) + } + } + } + + // An application turns Comet's Iceberg writes on with spark.comet.write.iceberg.enabled alone, + // so this test leaves the testing split flag unset. A Parquet scan is a native input, so the + // write is eligible. + test("an eligible Iceberg write runs natively when only the write flag is set") { + assume(icebergAvailable, "Iceberg not available in classpath") + withIcebergCatalog { _ => + withTempPath { dir => + spark + .range(10) + .selectExpr("CAST(id AS INT) AS id", "'eu' AS region", "CAST(id AS DOUBLE) AS amount") + .write + .parquet(dir.getCanonicalPath) + createTable("write_flag_only", partitionSpec = "") + val snapshot = captureWrite("write_flag_only") { + withSessionConf( + CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None, + CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> Some("true")) { + spark.read + .parquet(dir.getCanonicalPath) + .writeTo(s"$catalog.$ns.write_flag_only") + .append() + } + } + assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got ${snapshot.snapshotDelta}") val (commits, _) = collectIcebergWriteOps(snapshot.plans) val nativeWrites = snapshot.plans.flatMap { plan => collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e } @@ -209,15 +244,16 @@ class CometIcebergWriteActionSuite commits.nonEmpty && nativeWrites.nonEmpty, "expected an IcebergCommitExec over a CometIcebergWriteExec. Plans:\n" + snapshot.plans.mkString("\n--\n")) - assertRows("write_defaults", expectedIds = 0 until 10) + assertRows("write_flag_only", expectedIds = 0 until 10) } } } - // Iceberg's DeltaWriter writes a merge-on-read write's rows under either plan, so with no flag - // set the write keeps Spark's own operator. Only the testing split flag plans it as Comet's. + // Iceberg's DeltaWriter writes a merge-on-read write's rows under either plan, so with only the + // write flag set the write keeps Spark's own operator. Only the testing split flag plans it as + // Comet's. // https://github.com/apache/datafusion-comet/issues/6240 - test("a merge-on-read write keeps Spark's own write plan when no flag is set") { + test("a merge-on-read write keeps Spark's own write plan when only the write flag is set") { assume(icebergAvailable, "Iceberg not available in classpath") assume(isSpark35Plus, "WriteDelta interception starts with Spark 3.5") withIcebergCatalog { _ => @@ -231,7 +267,7 @@ class CometIcebergWriteActionSuite val snapshot = captureWrite("mor_defaults") { withSessionConf( CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None, - CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) { + CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> Some("true")) { spark.sql(s"DELETE FROM $catalog.$ns.mor_defaults WHERE id = 1") } } @@ -250,13 +286,13 @@ class CometIcebergWriteActionSuite } } - // With no flag set, an eligible write becomes a CometIcebergWriteExec, so reverting a + // With only the write flag set, an eligible write becomes a CometIcebergWriteExec, so reverting a // transition-heavy stage has to put a JVM writer back rather than drop the write. The same write // runs first under the default threshold, which leaves its stage alone, so that the reverted run // is known to start from a native write rather than from one that was never eligible. // https://github.com/apache/datafusion-comet/issues/5719 for (adaptive <- Seq(false, true)) { - test(s"transition-heavy fallback keeps the write when no flag is set with AQE=$adaptive") { + test(s"transition-heavy fallback keeps the write under the write flag alone, AQE=$adaptive") { assume(icebergAvailable, "Iceberg not available in classpath") withIcebergCatalog { _ => withTempPath { dir => @@ -270,8 +306,8 @@ class CometIcebergWriteActionSuite .parquet(dir.getCanonicalPath) val table = s"transition_defaults_${if (adaptive) "aqe" else "no_aqe"}" createTable(table, partitionSpec = "") - // Appends the source with no write flag set, checks that it committed once, and returns - // the plans it ran. + // Appends the source with only the write flag set, checks that it committed once, and + // returns the plans it ran. def append(maxTransitions: Int): Seq[SparkPlan] = { var plans = Seq.empty[SparkPlan] // withSQLConf returns Unit before Spark 4.0, so the plans are kept from inside it. @@ -282,7 +318,7 @@ class CometIcebergWriteActionSuite val snapshot = captureWrite(table) { withSessionConf( CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None, - CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) { + CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> Some("true")) { spark.read .parquet(dir.getCanonicalPath) .writeTo(s"$catalog.$ns.$table") @@ -4356,8 +4392,8 @@ class CometIcebergWriteActionSuite /** * Sets each key to its value, or unsets it for `None`, for the duration of `action`, then puts * back the value each key had before. It writes the session conf directly; see - * [[withNativeEnabled]] for why. A key is restored rather than unset because unsetting the - * native write flag would turn it on: it defaults to true while this suite pins it off. + * [[withNativeEnabled]] for why. A key is restored rather than unset because this suite pins + * both write flags, and unsetting one would leave its default in place of the suite's value. */ private def withSessionConf[T](settings: (String, Option[String])*)(action: => T): T = { val conf = spark.sessionState.conf