diff --git a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md index f5c832ec898..1666c0ad40d 100644 --- a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md +++ b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md @@ -50,10 +50,11 @@ not, inherits it. There is no query that fails to tell you. | 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. 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. +user turns it on, a change to the split plan affects every append, overwrite and copy-on-write +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 @@ -205,7 +206,7 @@ Read the ownership table in the contributor guide before reviewing any change ne | Suite | Covers | | ----------------------------------- | ----------------------------------------------------------------------------------------------- | | `CometIcebergWriteActionSuite` | Both layers end to end, parity with iceberg-java, DML, failure cleanup, AQE re-plans | -| `CometIcebergWriteDetectionSuite` | One case per eligibility rule | +| `CometIcebergWriteDetectionSuite` | Most eligibility rules, accepted and declined | | `CometIcebergSystemFunctionSuite` | Native partition transforms keeping partitioned writes native | | `CometIcebergRewriteActionSuite` | `rewrite_data_files` through the split plan and the native writer | | `IcebergWriteProtoTranslationSuite` | Property to `IcebergParquetWriteSettings` translation | @@ -231,7 +232,8 @@ Ask specifically: default Spark profile. Reflection, metrics conventions and path spelling differ across Iceberg versions, so ask for the `run-iceberg-tests` label or a local run where they matter. - [ ] **Is a green Iceberg Spark test run being cited as evidence?** It does not show which writer - handled a test ([#6148](https://github.com/apache/datafusion-comet/issues/6148)). + handled a test. The job summary counts writes per writer without naming tests (see "Which + writer ran each Iceberg write" in `contributor-guide/iceberg-spark-tests.md`). ## 8. Do the PR Changes Make the Docs Stale? diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md b/docs/source/contributor-guide/iceberg-spark-tests.md index dcee0903436..e0cdccd7ee1 100644 --- a/docs/source/contributor-guide/iceberg-spark-tests.md +++ b/docs/source/contributor-guide/iceberg-spark-tests.md @@ -65,10 +65,11 @@ are still the reference, and are what you want when updating a diff. ## 1. Install Comet -Run `make release` in Comet to install the Comet JAR into the local Maven repository, specifying the Spark version. +Run `make release` in Comet to install the Comet JAR into the local Maven repository, specifying the Spark +and Scala versions the Iceberg tests will run against. The example below matches step 3. ```shell -PROFILES="-Pspark-4.1" make release +PROFILES="-Pspark-3.5 -Pscala-2.13" make release ``` ## 2. Clone Iceberg and Apply Diff @@ -121,12 +122,12 @@ diff must be generated against its own tag. ## Running Tests in CI -The `iceberg_spark_test_.yml` workflows apply these diffs and run the three Gradle targets above -against each Iceberg version, all with Java 17. Iceberg 1.8.1 runs against Spark 3.4.3; Iceberg 1.9.1 and 1.10.0 -run against Spark 3.5.9; Iceberg 1.11.0 runs against Spark 4.1.3. Iceberg 1.11.0 runs in the -merge queue; 1.8.1, 1.9.1 and 1.10.0 run once a night against `main`. All four run earlier on a -pull request labeled `run-iceberg-tests`; none runs on an unlabeled pull request. All caller -workflows delegate to `iceberg_spark_test_reusable.yml`, which holds the build and test job logic. See +The `iceberg_1_8`, `iceberg_1_9`, `iceberg_1_10` and `iceberg_1_11` jobs in `ci.yml` apply these diffs and +run the three Gradle targets above against each Iceberg version, all with Java 17. Iceberg 1.8.1 runs against +Spark 3.4.3; Iceberg 1.9.1 and 1.10.0 run against Spark 3.5.9; Iceberg 1.11.0 runs against Spark 4.1.3. +Iceberg 1.11.0 runs in the merge queue; 1.8.1, 1.9.1 and 1.10.0 run once a night against `main`. All four +run earlier on a pull request labeled `run-iceberg-tests`; none runs on an unlabeled pull request. Each job +calls `iceberg_spark_test_reusable.yml`, which holds the build and test job logic. See [.github/workflows/README.md](https://github.com/apache/datafusion-comet/blob/main/.github/workflows/README.md) for how the pull-request, merge-queue and nightly tiers differ. diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index 05653338c9c..e9e8e8cddc8 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -38,7 +38,8 @@ Two layers, the second built on the first, both switched by `spark.comet.write.i | The native writer | Who writes the data files. An eligible `IcebergWrite` becomes `CometIcebergWrite`, which writes Parquet with iceberg-rust. | 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. +split plan reaches every append, overwrite and copy-on-write 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 @@ -170,7 +171,8 @@ The gate has these properties, and a new rule has to keep them: `scheme_of` split a location's scheme differently). Prefer sharing one implementation, or pin the pair with a test that feeds both the same inputs. -Every rule is pinned by `CometIcebergWriteDetectionSuite`. Add a case there for any new rule, for +Most rules are pinned by `CometIcebergWriteDetectionSuite`, and the row lineage rule by +`CometIcebergWriteActionSuite`. Add a case to the detection suite for any new rule, for both the accepted and the declined side. Update the eligibility table in the user guide in the same change. Whether each existing restriction is permanent is tracked in [#5643](https://github.com/apache/datafusion-comet/issues/5643). @@ -238,9 +240,9 @@ Points where Comet adapts iceberg-rust to match iceberg-java: - **Location generation.** `CometLocationGenerator` (`iceberg_partition_path.rs`) replaces iceberg-rust's `DefaultLocationGenerator` and renders partition directories the way iceberg-java's `PartitionSpec.partitionToPath` does. iceberg-rust's own rendering differs for several types, and - panics for pre-1970 `timestamptz` and for dates past `chrono`'s calendar. Float and double values - use Comet's Java `Double.toString` renderer (`java_float_string`), shared with - `cast(float as string)`. The partition type is resolved once when the generator is built, because + panics for dates past `chrono`'s calendar. Float and double values use Comet's Java + `Double.toString` renderer (`java_float_string`), shared with `cast(float as string)`. The + partition type is resolved once when the generator is built, because `LocationGenerator::generate_location` cannot return an error. - **Partition values.** `PartitionValueCalculator` (`iceberg_partition_value.rs`) replaces iceberg-rust's calculator of the same name, and `PartitionSplitter` replaces its @@ -264,9 +266,10 @@ Points where Comet adapts iceberg-rust to match iceberg-java: column's encoding when the file opens ([#6114](https://github.com/apache/datafusion-comet/issues/6114)). So each partition's `PartitionFeed` holds back the partition's first page of rows (`write.parquet.page-row-limit`, - capped by `write.parquet.row-group-size-bytes`, which the fanout feeds share) before any of them - reaches the writer. `DictionaryChooser` (`iceberg_dictionary.rs`) replays parquet-mr's accounting - over those rows and turns dictionary encoding off for the columns parquet-mr would write plain. + at least 100, capped by `write.parquet.row-group-size-bytes`, which the fanout feeds share) before + any of them reaches the writer, unless no column has a dictionary choice to make. + `DictionaryChooser` (`iceberg_dictionary.rs`) replays parquet-mr's accounting over those rows and + turns dictionary encoding off for the columns parquet-mr would write plain. It reads only the columns there is a choice for, one at a time, and walks each lazily only as far as its first page, so beyond the held rows it keeps no more than one column's dictionary. The accounting follows `FallbackValuesWriter`, `DictionaryValuesWriter` and @@ -347,7 +350,8 @@ backend ignores them. A fallback reason must contain property names only, never Each task emits exactly one batch with one row and two `BINARY` columns (`build_output_schema`): 1. The task's `DataFile`s encoded as an Iceberg v2 data manifest, written by iceberg-rust's - `ManifestWriter` into an in-process `memory:` `FileIO`. Empty when the task wrote no files. + `ManifestWriter` into an in-process `memory:` `FileIO`. A task that wrote no files sends a + manifest with no entries. 2. Every location the task's writers were handed (`encode_locations`: a big-endian count, then a length-prefixed UTF-8 string per location). @@ -368,7 +372,8 @@ Each task emits exactly one batch with one row and two `BINARY` columns (`build_ Rebuilding metrics on the JVM is the main reason manifest parity holds: metrics modes, truncation, the inferred-column cap and list/map bounds suppression are decided by iceberg-java's code. When touching the native writer, do not try to make iceberg-rust's own metrics authoritative; the JVM -discards them. The price is one footer read per written file. +discards them. The price is reading each written file's footer back from storage, two small reads +per file. ## Failure Handling and Cleanup Ownership @@ -438,7 +443,7 @@ the writer: run the write suites and the Iceberg Spark tests. The pin policy is | Suite | What it covers | | ---------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `CometIcebergWriteActionSuite` | End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, writes outgrowing the memory pool, AQE re-planning. | -| `CometIcebergWriteDetectionSuite` | One case per eligibility rule, accepted and declined. | +| `CometIcebergWriteDetectionSuite` | Most eligibility rules, each accepted and declined. | | `CometIcebergSystemFunctionSuite` | Native `bucket`, `truncate`, `years`/`months`/`days`/`hours`, which keep a partitioned write's hash distribution and sort native end to end. | | `CometIcebergRewriteActionSuite` | Iceberg's `rewrite_data_files` with the split plan and the native writer. | | `IcebergWriteProtoTranslationSuite` | Translation of properties into `IcebergParquetWriteSettings` and the writer mode. | @@ -448,9 +453,11 @@ the writer: run the write suites and the Iceberg Spark tests. The pin policy is | `CometIcebergEndToEndBenchmark` | An Iceberg read, hash repartition and Iceberg write against stock Spark, over primitive, struct, array, map and nested columns. It checks each arm's plan and the rows it wrote before timing it. | The Comet suites run against the Iceberg version each Spark profile pins in `spark/pom.xml`: 1.5.2 -for Spark 3.4, 1.8.1 for 3.5, 1.10.0 for 4.0 and 4.2, and 1.11.0 for 4.1. Only the default profile -runs on every pull request. Behaviour that differs between Iceberg versions, such as reflection -targets, metrics conventions and partition path spelling, needs a test that runs on each. +for Spark 3.4, 1.8.1 for 3.5, 1.10.0 for 4.0, and 1.11.0 for 4.1. No Iceberg runtime is published +for Spark 4.2, so the 4.2 profile builds against the 4.0 runtime and the Iceberg suites skip there +(`icebergAvailable` is false on Spark 4.2). Only the default profile runs on every pull request. +Behaviour that differs between Iceberg versions, such as reflection targets, metrics conventions +and partition path spelling, needs a test that runs on each. When writing a native-write test: @@ -471,9 +478,10 @@ When writing a native-write test: The upstream Iceberg Spark tests also run with the native writer and `localTableScan` enabled (see [Running Iceberg Spark Tests](iceberg-spark-tests.md)). They are a broad regression net, but they do -not assert which writer ran, and Comet's fallback reasons do not appear in their CI logs, so a green -run is not evidence that the native writer handled a given test -([#6148](https://github.com/apache/datafusion-comet/issues/6148)). +not assert which writer ran, so a green run is not evidence that the native writer handled a given +test. Each job's summary page counts the writes each writer ran and lists the most common reasons +Comet declined the rest (see +[Which writer ran each Iceberg write](iceberg-spark-tests.md#which-writer-ran-each-iceberg-write)). ## Pitfalls diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index 5d50d34cc18..bba4ea8ec14 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -116,6 +116,7 @@ supports: - `INSERT OVERWRITE`, static and dynamic (`OverwriteByExpression`, `OverwritePartitionsDynamic`) - Copy-on-write `DELETE` / `UPDATE` / `MERGE` (`ReplaceData`) +- Iceberg's `rewrite_data_files` procedure, which writes each rewritten file group as an append For an unpartitioned copy-on-write `MERGE`, the native Iceberg writer is reachable on Spark 3.5+ when `spark.comet.exec.mergeRows.enabled=true`. With the flag disabled, the JVM @@ -154,10 +155,12 @@ The rewrite is skipped — and the write runs through Spark's stock combined ope `spark.comet.write.iceberg.splitOperator.enabled` is on with Spark 3.5+; - the statement is CTAS / RTAS on Spark 3.4, where the staged exec writes inline; on Spark 3.5+ those statements re-plan their inner append, which is intercepted normally; +- the write is a streaming write (`writeStream` into an Iceberg table); - the write requires Spark's commit coordinator, which Comet's per-task commit protocol does not use; - Comet cannot reflect the Iceberg internals needed to build the two-operator plan (for - example an unrecognised write class or a `ReplaceData` projection it cannot map). + example an unrecognised write class, or an Iceberg 1.5 `ReplaceIcebergData` plan whose fields + it cannot read). In every fallback case the write is planned as if Comet were absent; there is no correctness trade-off, only no plan change. @@ -173,10 +176,10 @@ depends on is one the native path reproduces exactly, and additionally only when feeding the write is fully Comet-native. For a partitioned table that plan includes the hash distribution and local sort Iceberg requests on its partition transforms; those stay native because the transforms themselves have native implementations (see -[Iceberg system functions](iceberg.md)). Ineligible writes run through iceberg-java unchanged, -with the reason reported as a fall-back reason in Comet's extended EXPLAIN output. A write that -runs natively shows `CometIcebergWrite` under `IcebergCommit` in the physical plan; an ineligible -write keeps `IcebergWrite`. +[Iceberg system functions](iceberg.md#iceberg-system-functions)). Ineligible writes run through +iceberg-java unchanged, with the reason reported as a fall-back reason in Comet's extended EXPLAIN +output. A write that runs natively shows `CometIcebergWrite` under `IcebergCommit` in the physical +plan; an ineligible write keeps `IcebergWrite`. The native writer reads its input as Arrow batches from a Comet operator, so the write's input must itself run in Comet. A write whose input is a local relation, such as `INSERT ... VALUES` or @@ -295,7 +298,7 @@ writer produce different data files: distribution and ordering settings shape th identically on both paths, WAP / branch / snapshot properties act on the JVM committer, `write.avro.*` / `write.orc.*` apply only to formats already excluded, and merge-on-read settings route the write through `WriteDelta`, whose task writer remains iceberg-java rather -than the native writer. Every native eligibility rule is pinned by +than the native writer. Most native eligibility rules are pinned by `CometIcebergWriteDetectionSuite`. Manifest `DataFile` metrics are assembled on the JVM before commit: each written file's @@ -338,7 +341,7 @@ partitions needs memory in proportion to them. When the pool cannot grant it, th and closes the partitions holding the most memory until what is left fits, and a closed partition's next rows open a new file. The write then finishes with more, smaller files than iceberg-java's would. When every partition keeps getting rows, a task given an eighth of the memory -its partitions need writes about four files per partition instead of one, however many partitions +its partitions need writes four to five files per partition instead of one, however many partitions it writes to. The `files closed early to free memory` metric of its `CometIcebergWrite` operator counts the files it closed early. Disabling the fanout writer (`write.spark.fanout.enabled=false`) avoids the extra files: Spark then sorts each task's rows by partition, and the task keeps one file @@ -370,8 +373,9 @@ normal `remove_orphan_files` maintenance. When one task fails, the tasks that had already completed leave committed-nothing data files too. The committer collects each task's commit message as that task finishes, so on a job failure it aborts with the completed messages and deletes their data files through the table -`FileIO`. (Iceberg's own `SparkWrite.abort` skips cleanup unless a commit failed with a -cleanable error, so on the stock path those files are left for `remove_orphan_files`.) A +`FileIO`. (From Iceberg 1.6 on, Iceberg's own `SparkWrite.abort` skips cleanup unless a commit +failed with a cleanable error, so on the stock path those files are left for +`remove_orphan_files`. Iceberg 1.5, which the Spark 3.4 profile pins, deletes them.) A failure during the driver-side commit itself behaves exactly as on the stock path: the commit messages carry genuine `SparkWrite$TaskCommit` objects, so Iceberg's own `SparkWrite.abort` cleanup (which deletes the files listed in the commit messages for cleanable failures) applies @@ -381,11 +385,11 @@ unchanged. 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. Enabling -`spark.comet.write.iceberg.enabled` accepts them. They fall into three classes with very +`spark.comet.write.iceberg.enabled` accepts them. They fall into two 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 -below), and one operational path-layout caveat. +no reader decision is based on them), and differences visible in manifest metadata (these +outlive the write and feed later readers' pruning decisions, so each one is analyzed +individually below). ### Physical file layout only (cosmetic) @@ -395,13 +399,12 @@ a data file but not what any reader computes from it: - Footer key-value metadata differs: native files carry an `ARROW:schema` entry and no `iceberg.schema` entry; iceberg-java files are the opposite. - The Parquet root schema element is named `arrow_schema` (iceberg-java: `table`). -- `created_by` identifies parquet-rs, not parquet-mr. -- No page CRC checksums and no page-header statistics (parquet-mr writes both by default). - Absent page-header statistics can only make a reader scan more pages, never skip pages it - should have read; page pruning uses the column index, which the native writer does produce. +- `created_by` reads `Apache Iceberg (Comet)`, where iceberg-java's files carry + parquet-mr's version string. +- No page CRC checksums: iceberg-java writes one in every page header, and parquet-rs writes none. + Neither writer puts statistics in page headers; page pruning uses the column index, which both + writers produce. - Dictionary-encoded pages are labeled `RLE_DICTIONARY` (parquet-mr v1 files: `PLAIN_DICTIONARY`). -- Fixed-length binary columns (`uuid`, `fixed`, decimals with precision > 18) are not - dictionary-encoded (parquet-mr dictionary-encodes them). - The native writer uses parquet-mr's size accounting to choose dictionary encoding: a column whose sampled rows show the dictionary saving no space is written plain, with no dictionary page, instead of carrying a dictionary page that every selective read of it would have to fetch @@ -419,10 +422,12 @@ a data file but not what any reader computes from it: space. A column that keeps its dictionary and later fills it falls back to plain on both writers, but parquet-rs's dictionary page then holds every entry, where parquet-mr's holds only the entries earlier pages used. -- Row-group boundaries: parquet-mr flushes by byte size at a record-count check cadence, - parquet-rs buffers by row count. File naming follows the same cadence-style difference - (iceberg-java names files `---`; iceberg-rust uses a - process-local counter). +- Row-group boundaries: both writers close a row group by size, but parquet-mr checks its buffered + size only at record-count checkpoints, while parquet-rs checks its own estimate as it writes each + batch, so the boundaries land on different rows. +- File names: both writers name files `---`, counting + per task. The native writer zero-pads the task attempt id and counts from `00000`; iceberg-java + leaves the id unpadded and counts from `00001`. - Partition directory names match iceberg-java 1.8+'s `PartitionSpec.partitionToPath` for every partition type the native writer accepts. Identity partitions on `float` and `double` columns fall back (see [Native Parquet write eligibility](#native-parquet-write-eligibility)), so @@ -460,8 +465,9 @@ a data file but not what any reader computes from it: and on which partitions it closes early to free memory. Clustered and unpartitioned writes append in creation order on both paths and are unaffected. - Compressed page bytes are implementation-defined: the codec and any explicit level are - translated, but parquet-rs and parquet-mr embed different encoder implementations and - defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even + translated, and a `zstd` table with no level is written at parquet-mr's default level of 3, but + parquet-rs and parquet-mr embed different encoder implementations (LZ4 framing, for one), and + their pages and row groups end at different rows, so byte-identical output is not achievable even for a default `zstd` table. The decompressed data is identical. For the same reason, codec-level side channels (`zlib.compress.level`, `compression.brotli.quality`, `io.compression.codec.zstd.level` — the last is present in every Hadoop configuration by @@ -492,5 +498,5 @@ iceberg-java's _writer-tracked_ state would have recorded, and both are analyzed the same amount, so the derived null ratios and `IS NULL` / `IS NOT NULL` pruning decisions are unaffected. -All content not listed above — the logical data, encodings for non-FLBA columns, statistics -values, and manifest metadata — must match iceberg-java exactly, or the write falls back. +All content not listed above — the logical data, statistics values, and manifest metadata — +must match iceberg-java exactly, or the write falls back. diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 81fbac586c6..24713a32b7d 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -198,10 +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 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. +Writes are not covered by this list. By default Spark plans Iceberg writes with its own V2 write +operator, and iceberg-java writes the data files. 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 c0aadb4cb96..c0fae0e96b0 100644 --- a/docs/source/user-guide/latest/migration-guide.md +++ b/docs/source/user-guide/latest/migration-guide.md @@ -101,10 +101,11 @@ Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the testi 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. +`spark.comet.write.iceberg.splitOperator.enabled` is a testing setting, as it was in 1.1.0, but +Comet 1.1.0's Iceberg Writes guide documented it as the switch for the split plan. 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 ccaa8fbf2c6..aa4fd589679 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -63,7 +63,7 @@ omitted from the tables below and may be reconsidered based on demand: | ----------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `FileSourceScanExec` | ✅ | Parquet only. Some types and configurations fall back. See [Parquet Scan Compatibility](compatibility/scans.md). | | `BatchScanExec` | ✅ | Apache Iceberg Parquet scans run natively. Native CSV scans are experimental and disabled by default. DataSource V2 Parquet scans are not accelerated. See [Parquet Scan Compatibility](compatibility/scans.md) and the [Iceberg Guide](iceberg.md). | -| `LocalTableScanExec` | ⚠️ | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). | +| `LocalTableScanExec` | ⚠️ | Disabled by default. Converting it rarely speeds up the scan itself, but a native Iceberg write needs it to write the rows of `INSERT ... VALUES` or a local DataFrame natively; see [Iceberg Writes](iceberg-writes.md#native-parquet-write-eligibility). Enable it with `spark.comet.exec.localTableScan.enabled=true` ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). | | `EmptyRelationExec` | ✅ | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support, including native Parquet writes. | | `RangeExec` | ⚠️ | Disabled by default. Set `spark.comet.exec.range.enabled=true` to generate the rows of `spark.range` and SQL `range()` in native code, so the operators above them run natively. It can be slower than Spark when those operators are only cheap expressions, such as a filter, which Spark compiles together with the range into one loop. Ranges whose arithmetic overflows the `Long` range fall back to Spark. | | `InMemoryTableScanExec` | ⚠️ | Experimental, disabled by default. Set `spark.comet.exec.inMemoryCache.enabled=true` before the application starts so Comet installs its Arrow cache serializer. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](in-memory-cache.md). | diff --git a/docs/source/user-guide/latest/tuning/memory.md b/docs/source/user-guide/latest/tuning/memory.md index 2632369df24..51fb9754b83 100644 --- a/docs/source/user-guide/latest/tuning/memory.md +++ b/docs/source/user-guide/latest/tuning/memory.md @@ -39,12 +39,13 @@ off-heap consumers draw down one number. A larger pool lets the operators that can spill, such as sorts, aggregations and the shuffle writer, keep more in memory before they spill to disk. An operator that cannot spill fails the task when it cannot reserve the memory it -needs. +needs. A native Iceberg fanout write is the exception: it closes some of its open files early instead (see +[Iceberg Writes](../iceberg-writes.md#failure-handling)). Comet's memory pool only tracks memory that an operator explicitly reserves, which in practice means the batches -an operator deliberately accumulates: the sort buffer, the build side of a hash join, hash aggregation state, and the -shuffle writer's buffered partitions. Memory that is not reserved is invisible to the pool no matter how much of it -there is. That includes: +an operator deliberately accumulates: the sort buffer, the build side of a hash join, hash aggregation state, the +shuffle writer's buffered partitions, and the files a native writer holds open. Memory that is not reserved is +invisible to the pool no matter how much of it there is. That includes: - per-batch working memory in expression kernels and Arrow array builders, - decompression buffers and Parquet reader structures, diff --git a/docs/source/user-guide/latest/understanding-comet-plans.md b/docs/source/user-guide/latest/understanding-comet-plans.md index f77cd1b5165..ccaa8d27740 100644 --- a/docs/source/user-guide/latest/understanding-comet-plans.md +++ b/docs/source/user-guide/latest/understanding-comet-plans.md @@ -306,7 +306,9 @@ by role. Names match what is shown in the plan output. ### 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. +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 | | ------------------------------ | --------------------------------------------------------------------------------- | @@ -329,7 +331,7 @@ consecutively in a plan, they execute as a single fused block. | `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) | +| `CometIcebergWrite` | `IcebergWrite` (native Iceberg data-file writes, opt-in) | ### JVM-Side Operators