diff --git a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md index 5b3f317cced..45233220329 100644 --- a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md +++ b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md @@ -33,7 +33,7 @@ bar, and the output format. This skill only covers Iceberg writes. | Doc | What you need from it | | ------------------------------------------------------ | --------------------------------------------------------------------------------------- | | `docs/source/contributor-guide/iceberg-writes.md` | The two layers, the gate's properties, the wire format both ways, cleanup ownership | -| `docs/source/user-guide/latest/iceberg-writes.md` | The eligibility table and the accepted divergences from iceberg-java, as users see them | +| `docs/source/user-guide/latest/iceberg.md` (Writes) | The eligibility table and the accepted divergences from iceberg-java, as users see them | | `docs/source/contributor-guide/iceberg-spark-tests.md` | What the Iceberg Spark test diffs enable, and what a green run does and does not prove | The organizing rule is that **the native writer produces what iceberg-java would have produced, or @@ -123,7 +123,7 @@ only the latest. these transforms back to iceberg-rust, or that makes the kernels read the column's timezone label, breaks that agreement. iceberg-rust's `years` and `months` follow the label ([apache/iceberg-rust#3142](https://github.com/apache/iceberg-rust/issues/3142)). Ask for a - test in a non-UTC session for a change here. "Partition values" in `iceberg-writes.md` lists + test in a non-UTC session for a change here. "Partition values" in the Writes section of the user-guide `iceberg.md` lists where iceberg-rust differs, and `timezones.md` covers how Comet labels timestamps. - [ ] **Partition paths use the Java renderers.** Directory names come from `CometLocationGenerator` / `partition_to_path` in `iceberg_partition_path.rs`, which follow @@ -241,9 +241,10 @@ Ask specifically: properties, the proto description, the native adaptations list, the payload columns and the JVM steps after them, the cleanup ownership table, the test table, and the Iceberg version per Spark profile -- **`user-guide/latest/iceberg-writes.md`**: the eligibility table, the "falls back" lists, the +- **`user-guide/latest/iceberg.md`, Writes section**: the eligibility table, the "falls back" lists, the failure-handling section, and the accepted-divergences list. A PR that lifts or adds a rule, or changes a divergence, must update these. A divergence the code has since fixed must be removed. -- **`user-guide/latest/iceberg.md`**: its unsupported-features list mentions writes +- **`user-guide/latest/iceberg.md`, Native Reader section**: its unsupported-features list mentions writes. The + Known Issues section lists the open write issues - **`contributor-guide/iceberg-spark-tests.md`**: the flags the diffs set - `CometConf.scala` doc strings for the two flags, when defaults or scope change diff --git a/docs/source/_templates/redirect.html b/docs/source/_templates/redirect.html new file mode 100644 index 00000000000..7ebc375872a --- /dev/null +++ b/docs/source/_templates/redirect.html @@ -0,0 +1,20 @@ + + + diff --git a/docs/source/about/faq.md b/docs/source/about/faq.md index 8e411caa693..29c39910ed1 100644 --- a/docs/source/about/faq.md +++ b/docs/source/about/faq.md @@ -142,7 +142,7 @@ Common causes of a small speedup are: Not yet. Comet accelerates [Apache Iceberg](../user-guide/latest/iceberg.md) tables, with a native 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, +[native writes](../user-guide/latest/iceberg.md#writes). 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 4065d19ef83..c1942ac4755 100644 --- a/docs/source/about/gluten_comparison.md +++ b/docs/source/about/gluten_comparison.md @@ -124,7 +124,7 @@ native integrations for Delta Lake, Hudi, or Paimon. See the [Comet Iceberg guid features and known limitations. [Comet Iceberg guide]: /user-guide/latest/iceberg.md -[Comet Iceberg writes]: /user-guide/latest/iceberg-writes.md +[Comet Iceberg writes]: /user-guide/latest/iceberg.md#writes Gluten ships dedicated modules for Iceberg, Delta Lake (2.3 through 4.1, depending on the Spark version), Hudi, and Paimon. Its [Iceberg documentation] says that reads of unpartitioned tables are offloaded, including tables with diff --git a/docs/source/conf.py b/docs/source/conf.py index 05c3e36accf..ffa1a24e808 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -186,6 +186,10 @@ # enable nice rendering of checkboxes for the task lists myst_enable_extensions = ["colon_fence", "deflist", "tasklist"] +# Like the extension's default template, but when the old URL carries an anchor, the anchor +# replaces any anchor in the target, so deep links into a moved page still land on their section. +redirect_html_template_file = "_templates/redirect.html" + redirects = { "overview.html": "index.html", "gluten_comparison.html": "about/gluten_comparison.html", @@ -205,6 +209,11 @@ "user-guide/tuning.html": "latest/tuning.html", # The compatibility guide became a directory of pages in #4055. "user-guide/latest/compatibility.html": "compatibility/index.html", + # Parquet and Iceberg documentation was consolidated into one page each. + "user-guide/latest/compatibility/scans.html": "../parquet.html#reading", + "user-guide/latest/compatibility/parquet-iceberg-known-issues.html": "../parquet.html#known-issues", + "user-guide/latest/tuning/scans.html": "../parquet.html#tuning", + "user-guide/latest/iceberg-writes.html": "iceberg.html", } diff --git a/docs/source/contributor-guide/data-source-audits/iceberg.md b/docs/source/contributor-guide/data-source-audits/iceberg.md index 41f89bd1e0c..12c751b3c1f 100644 --- a/docs/source/contributor-guide/data-source-audits/iceberg.md +++ b/docs/source/contributor-guide/data-source-audits/iceberg.md @@ -21,7 +21,7 @@ > Audit notes for the native Iceberg scan and the native Iceberg writer. Absence of an entry means > the area has not been audited yet. Open issues are listed for users in -> [Known Issues in Parquet and Iceberg Reads and Writes]. +> [Known Issues in the Iceberg guide]. ## Native Iceberg scan @@ -86,7 +86,7 @@ - **Not covered:** merge-on-read writes (`WriteDelta`), format v3 types, encryption, object stores, and write throughput, which needs a release build. -[Known Issues in Parquet and Iceberg Reads and Writes]: ../../user-guide/latest/compatibility/parquet-iceberg-known-issues.md +[Known Issues in the Iceberg guide]: ../../user-guide/latest/iceberg.md#known-issues [apache/iceberg#17320]: https://github.com/apache/iceberg/pull/17320 [apache/iceberg#17560]: https://github.com/apache/iceberg/pull/17560 [apache/iceberg#17561]: https://github.com/apache/iceberg/issues/17561 diff --git a/docs/source/contributor-guide/data-source-audits/index.md b/docs/source/contributor-guide/data-source-audits/index.md index bc0df6d06a9..3a886df9684 100644 --- a/docs/source/contributor-guide/data-source-audits/index.md +++ b/docs/source/contributor-guide/data-source-audits/index.md @@ -31,8 +31,9 @@ audits of expressions. Each page holds dated entries per component. An entry rec - what the audit did not cover A later audit of the same area adds a new dated entry instead of editing an old one, so the history -stays visible. The open issues that users can hit are listed in the user guide, in -[Known Issues in Parquet and Iceberg Reads and Writes](../../user-guide/latest/compatibility/parquet-iceberg-known-issues.md). +stays visible. The open issues that users can hit are listed in the Known Issues +sections of the [Parquet](../../user-guide/latest/parquet.md#known-issues) and +[Iceberg](../../user-guide/latest/iceberg.md#known-issues) guides. ```{toctree} :maxdepth: 1 diff --git a/docs/source/contributor-guide/data-source-audits/parquet.md b/docs/source/contributor-guide/data-source-audits/parquet.md index ad7fad0efd4..e574ef420b7 100644 --- a/docs/source/contributor-guide/data-source-audits/parquet.md +++ b/docs/source/contributor-guide/data-source-audits/parquet.md @@ -21,7 +21,7 @@ > Audit notes for the native Parquet scan and the native Parquet writer. Absence of an entry means > the area has not been audited yet. Open issues are listed for users in -> [Known Issues in Parquet and Iceberg Reads and Writes]. +> [Known Issues in the Parquet guide]. ## Native Parquet scan @@ -87,7 +87,7 @@ not support ([#3193], [#3194]), object-store and HDFS outputs, and write throughput, which needs a release build. -[Known Issues in Parquet and Iceberg Reads and Writes]: ../../user-guide/latest/compatibility/parquet-iceberg-known-issues.md +[Known Issues in the Parquet guide]: ../../user-guide/latest/parquet.md#known-issues [apache/datafusion#26393]: https://github.com/apache/datafusion/issues/26393 [#3193]: https://github.com/apache/datafusion-comet/issues/3193 [#3194]: https://github.com/apache/datafusion-comet/issues/3194 diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index 25f522a9643..5504bab8e76 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -25,7 +25,7 @@ the JVM/native boundary in each direction, who owns cleanup when a task fails, a change to any of it. For the user-facing view (configuration, the eligibility table, and the accepted differences from -iceberg-java's output), see [Iceberg Writes](../user-guide/latest/iceberg-writes.md). This page +iceberg-java's output), see [Iceberg Writes](../user-guide/latest/iceberg.md#writes). This page does not repeat those lists; it explains the code that implements them. ## Overview diff --git a/docs/source/contributor-guide/timezones.md b/docs/source/contributor-guide/timezones.md index ee66adcc667..951b8b93ee5 100644 --- a/docs/source/contributor-guide/timezones.md +++ b/docs/source/contributor-guide/timezones.md @@ -24,7 +24,7 @@ aimed at contributors working on datetime expressions, casts, scans, or any nati produces or consumes timestamp columns. For user-facing differences from Spark, see the datetime and cast sections of the [Expression Compatibility](../user-guide/latest/compatibility/expressions/index.md) pages, and the -[scan compatibility](../user-guide/latest/compatibility/scans.md) notes. Known bugs are tracked in +[Parquet](../user-guide/latest/parquet.md) notes. Known bugs are tracked in [#6335](https://github.com/apache/datafusion-comet/issues/6335). The short version: Comet never converts timestamp values at the JVM/native boundary. The session diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index bc364e14a48..be55a72085e 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -23,9 +23,9 @@ Comet aims to provide consistent results with the version of Apache Spark that i This guide documents areas where Comet's behavior is known to differ from Spark. Topics are grouped by subsystem: -- **Parquet**: limitations when reading Parquet files. -- **Parquet and Iceberg known issues**: open issues in which Parquet and Iceberg reads and writes - give different results from Spark, or fail where Spark succeeds, without falling back. +- **Parquet and Iceberg**: limitations, differences from Spark, and known issues for reading and + writing Parquet files and Iceberg tables. These live with the data source, on the + [Parquet](../parquet.md) and [Iceberg](../iceberg.md) pages. - **Floating-point comparison**: NaN and signed-zero handling in comparisons. - **Regular expressions**: differences between the Rust regexp crate and Java's regex engine. - **Operators**: operator-level compatibility notes, including window functions and round-robin partitioning. @@ -122,8 +122,8 @@ divergence: The following native paths silently return values that differ from Spark for edge-case inputs. Most also have entries in the per-category expression pages linked above; they are collected here so users hunting an unexpected value have a single place to check. Parquet and Iceberg reads and -writes have their own list, in -[Known Issues in Parquet and Iceberg Reads and Writes](parquet-iceberg-known-issues.md#results-that-differ-from-spark): +writes have their own lists, under Known Issues on the +[Parquet](../parquet.md#known-issues) and [Iceberg](../iceberg.md#known-issues) pages: - Native `RANGE` window frames with an explicit `PRECEDING` / `FOLLOWING` offset diverge from Spark when the boundary arithmetic overflows for `DATE` or `DECIMAL` `ORDER BY` columns diff --git a/docs/source/user-guide/latest/compatibility/parquet-iceberg-known-issues.md b/docs/source/user-guide/latest/compatibility/parquet-iceberg-known-issues.md deleted file mode 100644 index 77a0a2501d9..00000000000 --- a/docs/source/user-guide/latest/compatibility/parquet-iceberg-known-issues.md +++ /dev/null @@ -1,266 +0,0 @@ - - -# Known Issues in Parquet and Iceberg Reads and Writes - -This page lists open issues in Comet's Parquet and Iceberg reads and writes where Comet does not -fall back to Spark. Each entry links to its tracking issue. The entries are in two sections: - -- [Results that differ from Spark](#results-that-differ-from-spark): wrong values, a NULL where - Spark returns a value or the reverse, rows returned for a read that Spark rejects, and written - files that Spark or other engines read differently from the files Spark writes. -- [Queries that fail where Spark succeeds](#queries-that-fail-where-spark-succeeds). - -The cases where Comet falls back to Spark are documented in [Parquet Compatibility](scans.md), in -the [current limitations](../iceberg.md#current-limitations) of the Iceberg guide and in -[Iceberg Writes](../iceberg-writes.md). - -Which entries apply depends on which native paths are enabled: - -| Native path | Setting | Default | -| -------------- | ---------------------------------------- | ---------------------- | -| Parquet scan | `spark.comet.scan.enabled` | `true` | -| Iceberg scan | `spark.comet.scan.icebergNative.enabled` | `true` | -| Iceberg writer | `spark.comet.write.iceberg.enabled` | `false` | -| Parquet writer | `spark.comet.write.parquet.enabled` | `false` (testing only) | - -`spark.comet.scan.enabled` is a testing setting that Comet's own test suites use. It is not meant -for production, so do not turn it off to avoid an entry on this page. - -## Results that differ from Spark - -### Parquet reads - -- **Dates and timestamps written with the legacy calendar.** Spark before 3.0, and later versions - with `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`, write these values in the hybrid - Julian/Gregorian calendar. The native Parquet scan reads them as proleptic Gregorian, so values - before October 15, 1582 are wrong in projections and predicates. It ignores - `spark.sql.parquet.datetimeRebaseModeInRead` and the file's `org.apache.spark.legacyDateTime` - metadata ([#5010](https://github.com/apache/datafusion-comet/issues/5010)). The - [Delta scan](../delta.md) rebases. -- **INT96 timestamps inside legacy-format lists.** INT96 timestamps outside the years 1677 to 2262 - read back wrong inside lists written in a legacy layout, as Hive and - `spark.sql.parquet.writeLegacyFormat=true` write them - ([#7029](https://github.com/apache/datafusion-comet/issues/7029)). -- **A struct or array with none of its requested fields in the file.** Spark returns NULL for the - column. Comet returns a non-null struct or array that holds NULLs. This applies to Spark 3.4, 3.5 - and 4.0, and to Spark 4.1 and later with - `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true` - ([#7030](https://github.com/apache/datafusion-comet/issues/7030)). -- **Nested BINARY values that are not valid UTF-8, read as STRING.** This applies to a BINARY - column without the UTF8 annotation that is read as `string`, through the read schema or - `spark.sql.parquet.binaryAsString=true`. Spark returns the bytes. Comet returns NULL for each - such nested value. Top-level values match Spark in a release build, because the invalid bytes - reach an Arrow string array unchecked. A debug build panics on them - ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of - [#4764](https://github.com/apache/datafusion-comet/issues/4764)). -- **Spark 3.4: a struct matched by field id under a different name.** With - `spark.sql.parquet.fieldId.read.enabled=true`, a struct column that the read schema matches to - the file by field id but under another name reads as NULL when Spark prunes its nested fields - ([#7031](https://github.com/apache/datafusion-comet/issues/7031)). -- **Field-id reads where one file column matches a field by id and a sibling by name.** With - `spark.sql.parquet.fieldId.read.enabled=true`, Comet returns NULL for the sibling at the top - level. Inside a struct, it returns the values of the field matched by id. Spark's own readers - disagree here ([#6994](https://github.com/apache/datafusion-comet/issues/6994)). -- **`TIMESTAMP_MILLIS` values that overflow when converted to microseconds.** In nested fields, and - in every column of a scan that has data filters, Comet returns NULL where Spark fails with - `ArithmeticException: long overflow` - ([#5553](https://github.com/apache/datafusion-comet/issues/5553), - [#5739](https://github.com/apache/datafusion-comet/issues/5739)). -- **Reads that Spark rejects.** Comet returns rows where Spark fails the read for - `TIMESTAMP(NANOS)` columns read as `timestamp` or `timestamp_ntz` - ([#7034](https://github.com/apache/datafusion-comet/issues/7034)), `INT` columns read as - `INTERVAL DAY TO SECOND` ([#5188](https://github.com/apache/datafusion-comet/issues/5188)), and - Variant-annotated fields read as a plain struct - ([#5741](https://github.com/apache/datafusion-comet/issues/5741)). - -### Iceberg reads - -- **Migrated tables that use a name mapping.** The `migrate`, `snapshot` and `add_files` - procedures set `schema.name-mapping.default` for files without Iceberg field ids. When a query - also reads a top-level primitive column, the native scan returns NULL for nested fields (struct - fields, list elements and map values). Filters on those fields then select the wrong rows - ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The fix needs an iceberg-rust - upgrade ([#6934](https://github.com/apache/datafusion-comet/issues/6934)). -- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such - a file as NULL, and the native scan returns the values stored in the file - ([#6794](https://github.com/apache/datafusion-comet/issues/6794)). -- **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as - NULL, and two columns that swapped names return each other's values - ([#6822](https://github.com/apache/datafusion-comet/issues/6822)). -- **Reflection failures while the scan is planned.** If Comet fails to read the table's schema or - metadata through reflection, it continues with an empty or partial result instead of falling - back, which can resolve a column to a different field id - ([#5257](https://github.com/apache/datafusion-comet/issues/5257)). - -### Parquet writes - -The native Parquet writer is experimental and disabled by default. With -`spark.comet.write.parquet.enabled=true`: - -- **The datetime rebase modes are ignored.** On Spark 3.4 and 3.5, whose default rebase mode is - `EXCEPTION`, the native writer writes dates and timestamps before October 15, 1582 that Spark - refuses to write, and Spark then fails to read them back under its default read mode. With - `spark.sql.parquet.datetimeRebaseModeInRead=LEGACY`, Spark reads them shifted by several days, on - every version. The files lack the `org.apache.spark.version` footer key that tells Spark they need - no rebasing ([#7047](https://github.com/apache/datafusion-comet/issues/7047), - [#3427](https://github.com/apache/datafusion-comet/issues/3427)). -- **`spark.sql.parquet.outputTimestampType` and `spark.sql.parquet.writeLegacyFormat` are - ignored.** Timestamps are always written as `INT64` microseconds, never as Spark's default - `INT96`, so with `TIMESTAMP_MILLIS` they keep the sub-millisecond digits that Spark drops. Spark's - schema footer key is not written either, so a path-based read loses column metadata such as - comments, and with `spark.sql.parquet.inferTimestampNTZ.enabled=false` it reads `TIMESTAMP_NTZ` - columns as `TIMESTAMP` ([#7048](https://github.com/apache/datafusion-comet/issues/7048), - [#3425](https://github.com/apache/datafusion-comet/issues/3425)). -- **Spark 3.4 and 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs - after a write: file index refresh, cache refresh, partition refresh and statistics update. Later - queries can return the table's previous contents until `REFRESH TABLE` - ([#6315](https://github.com/apache/datafusion-comet/issues/6315)). -- **Spark's Parquet writer properties are ignored.** `parquet.block.size`, page size, dictionary, - writer version and statistics settings never reach the native writer - ([#5304](https://github.com/apache/datafusion-comet/issues/5304)). -- **An input with no partitions writes no file.** Spark writes one file that holds the schema - ([#5303](https://github.com/apache/datafusion-comet/issues/5303)). -- **Spark SQL tests that still fail with the native writer**, not yet classified: - - CTAS schema ([#3418](https://github.com/apache/datafusion-comet/issues/3418)) - - CTAS with `CHAR_AS_VARCHAR` ([#3419](https://github.com/apache/datafusion-comet/issues/3419)) - - CTAS with a union ([#3429](https://github.com/apache/datafusion-comet/issues/3429)) - - `INSERT INTO` with default columns ([#3421](https://github.com/apache/datafusion-comet/issues/3421)) - - `INSERT INTO` after `ALTER TABLE ADD COLUMNS` ([#3422](https://github.com/apache/datafusion-comet/issues/3422)) - - `INSERT INTO` from non-projection queries ([#3423](https://github.com/apache/datafusion-comet/issues/3423)) - - the `INSERT OVERWRITE` output schema ([#3424](https://github.com/apache/datafusion-comet/issues/3424)) - - multiple inserts ([#3430](https://github.com/apache/datafusion-comet/issues/3430)) - - `INSERT OVERWRITE LOCAL DIRECTORY` ([#3420](https://github.com/apache/datafusion-comet/issues/3420)) - - empty files ([#3417](https://github.com/apache/datafusion-comet/issues/3417)) - - `onTaskCommit` ([#3428](https://github.com/apache/datafusion-comet/issues/3428)) - -### Iceberg writes - -The native Iceberg writer is disabled by default. With `spark.comet.write.iceberg.enabled=true`: - -- **Brotli tables.** The native writer commits brotli-compressed files. JVM Parquet readers can - read them only if Hadoop's `BrotliCodec` is on the classpath. iceberg-java fails the write - instead ([#6960](https://github.com/apache/datafusion-comet/issues/6960)). -- **`write.parquet.compression-codec=none`.** The native writer writes the table uncompressed, and - iceberg-java rejects the value. The native writer also trims or lowercases a few other settings - that iceberg-java reads as written - ([#7008](https://github.com/apache/datafusion-comet/issues/7008)). - -The physical differences that every native data file has, and why they are safe, are listed in -[Accepted divergences](../iceberg-writes.md#accepted-divergences). - -## Queries that fail where Spark succeeds - -In these cases a Comet read or write fails with an error where Spark, or iceberg-java, succeeds. -Comet does not fall back to Spark for them. - -### Parquet reads that fail - -- **STRING columns that hold invalid UTF-8.** The read fails with - `Parquet error: encountered non UTF-8 data`, where Spark returns the bytes - ([#4121](https://github.com/apache/datafusion-comet/issues/4121), part of - [#4764](https://github.com/apache/datafusion-comet/issues/4764)). -- **Duplicate field names.** Selecting a field by name when several physical siblings match it, - including inside structs, arrays and maps, fails with a duplicate-field error - ([#5884](https://github.com/apache/datafusion-comet/issues/5884), - [#5964](https://github.com/apache/datafusion-comet/issues/5964)). - [Parquet Compatibility](scans.md) describes the reads this affects. -- **A duplicate field id in a column the query does not read.** With a valid read schema and - `spark.sql.parquet.fieldId.read.enabled=true`, a file in which two root columns carry the same - field id fails the scan whenever the table schema has a field with that id, even when the query - does not read that field ([#7006](https://github.com/apache/datafusion-comet/issues/7006)). -- **A nested encoding that arrow-rs rejects.** A file fails to read when any column in its schema - uses such an encoding, even a column the query does not read - ([#7035](https://github.com/apache/datafusion-comet/issues/7035)). -- **A `DATE` column read as `INTERVAL YEAR TO MONTH`.** Spark reads it, and Comet fails the read - ([#5188](https://github.com/apache/datafusion-comet/issues/5188)). - -### Iceberg reads that fail - -- **A data file that omits an optional nested field.** The native scan fails on a data file that - lacks an optional struct field, which a write with `write.spark.accept-any-schema=true` can - produce without any schema change - ([#7033](https://github.com/apache/datafusion-comet/issues/7033)). -- **`VERSION AS OF` reads of a column whose type was promoted after the snapshot** - ([#6993](https://github.com/apache/datafusion-comet/issues/6993)). -- **An equality delete keyed on a nested field.** Every query the native scan runs on the table - fails with `field not found`, even one that reads the key - ([#6782](https://github.com/apache/datafusion-comet/issues/6782)). -- **Column names that only newer Unicode case data folds together**, such as a pair of Vithkuqi - letters, which Unicode 14 added. Queries on such a table fail with a duplicate-field error - ([#5540](https://github.com/apache/datafusion-comet/issues/5540)). - -### Object store and credential settings - -The native Parquet and Iceberg scans resolve storage settings with their own clients rather than -with Hadoop's. Where the two disagree, a native read can use a different identity or route than -Spark, and fail where Spark succeeds: - -- **S3 settings that Hadoop's S3A client resolves differently**, such as JCEKS-backed secrets and - the precedence of per-bucket settings. The native Parquet scan forwards them without checking - for a difference ([#5662](https://github.com/apache/datafusion-comet/issues/5662)). -- **AWS profile credential providers.** The native scans ignore `fs.s3a.auth.profile.name` and - `fs.s3a.auth.profile.file` ([#4245](https://github.com/apache/datafusion-comet/issues/4245)), - and resolve some profiles differently from Hadoop 3.4 and later, for example profiles that use - `credential_process` or chains of assumed roles - ([#6575](https://github.com/apache/datafusion-comet/issues/6575)). -- **Azure credentials.** Container-scoped OAuth settings, Fabric hosts and a few other - configurations resolve differently from Hadoop's ABFS driver - ([#6605](https://github.com/apache/datafusion-comet/issues/6605)). -- **Iceberg tables whose metadata location has no host.** The native Iceberg scan skips the - credential provider configured with `s3.comet.credential.provider.class` - ([#6536](https://github.com/apache/datafusion-comet/issues/6536)). - -### Parquet writes that fail - -With `spark.comet.write.parquet.enabled=true`: - -- **Row groups that outgrow the memory pool.** The native writer cuts row groups by row count - only, where Spark uses `parquet.block.size`, and it cannot spill. With wide or nested rows, the - open row group can outgrow the off-heap pool and fail the task. The audit did not reproduce a - failure, because a 239 MB row group fit in a 256 MB pool - ([#5304](https://github.com/apache/datafusion-comet/issues/5304)). - -### Iceberg writes that fail - -With `spark.comet.write.iceberg.enabled=true`: - -- **A write that keeps one file open and outgrows the off-heap pool.** An unpartitioned or - clustered write fails its task when its open file outgrows the pool, where iceberg-java writes - it on the JVM heap. [Failure handling](../iceberg-writes.md#failure-handling) describes the - error and the settings that let such a write fit. - -## Differences that are not Comet bugs - -In these cases Comet's result differs from Spark's, but the bug is in Spark or iceberg-java: - -- **iceberg-java 1.9 and later: `IS NULL` on a float or double field under a NULL struct.** - iceberg-java leaves the rows where the struct is NULL out of the field's null count, so a filter - such as `WHERE s.x IS NULL` can skip data files that hold matching rows, with or without Comet - ([apache/iceberg#17561](https://github.com/apache/iceberg/issues/17561)). Files written by Comet's - native Iceberg writer are not affected. -- **iceberg-java before 1.12: a NULL struct whose first field is an identity partition source.** - iceberg-java reads the struct as a struct of NULLs, and Comet returns NULL, as iceberg-java 1.12 - does. [apache/iceberg#17320](https://github.com/apache/iceberg/pull/17320) is titled for nested - initial defaults, but it fixes this case too: it keeps one file column in the read, so the - struct's definition level survives when every projected field is a constant, and an identity - partition value is such a constant. -- **Field-id reads with a read schema that swaps two nested field names.** Spark's vectorized - reader pairs the file's columns with the read fields by the file's names, so it fails with a type - mismatch or returns no rows. Comet resolves the fields by id. diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md deleted file mode 100644 index ec6142713f2..00000000000 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ /dev/null @@ -1,149 +0,0 @@ - - -# Parquet Compatibility - -Comet's Parquet scan offloads decoding to native code and produces Arrow batches for the rest of -the plan. Comet falls back to Spark when the scan cannot be converted (for example, due to one of -the unsupported features listed below). - -## Parquet Scan Limitations - -The following features are not supported and cause Comet to fall back to Spark: - -- Decimals encoded in binary format. -- `ShortType` columns, by default. When reading Parquet files written by systems other than Spark that contain - columns with the logical type `UINT_8` (unsigned 8-bit integers), Comet may produce different results than Spark. - Spark maps `UINT_8` to `ShortType`, but Comet's Arrow-based readers respect the unsigned type and read the data as - unsigned rather than signed. Since Comet cannot distinguish `ShortType` columns that came from `UINT_8` versus - signed `INT16`, by default Comet falls back to Spark when scanning Parquet files containing `ShortType` columns. - This behavior can be disabled by setting `spark.comet.scan.unsignedSmallIntSafetyCheck=false`. Note that `ByteType` - columns are always safe because they can only come from signed `INT8`, where truncation preserves the signed value. -- Default values that are nested types (e.g., maps, arrays, structs). Literal default values are supported. -- Spark's Datasource V2 API. When `spark.sql.sources.useV1SourceList` does not include `parquet`, Spark uses the - V2 API for Parquet scans. Comet's Parquet scan only supports the V1 API. -- `_metadata.row_index`, `_metadata.file_block_start` and `_metadata.file_block_length`. When Spark splits a file, - the native scan and Spark can read a row group in different splits, so the block values would differ. Other - `_metadata` columns (`file_path`, `file_name`, `file_size`, `file_modification_time`) are supported. -- No support for `input_file_name()`, `input_file_block_start()`, or `input_file_block_length()` SQL functions. - Comet's Parquet scan does not use Spark's `FileScanRDD`, so these functions cannot populate their values. -- No support for `ignoreMissingFiles` or `ignoreCorruptFiles` being set to `true` -- `spark.sql.parquet.enableVectorizedReader=false`. Disabling the vectorized reader opts into - Spark's parquet-mr semantics (silent overflow, null-on-narrowing), which Comet's native reader - does not replicate. By default Comet falls back to Spark in this case. Set - `spark.comet.scan.allowDisabledParquetVectorizedReader=true` to opt in to running the - Comet Parquet scan regardless. -- A read schema that repeats a Parquet field id, at the top level or within a struct, when - `spark.sql.parquet.fieldId.read.enabled=true`. -- A read schema with sibling struct fields whose names collide case-insensitively, when - `spark.sql.caseSensitive=false`. Spark's analyzer normally rejects such a schema before the scan - is planned. - -Some Parquet reads return results that differ from Spark without falling back to Spark, such as -reads of dates and timestamps written with the legacy hybrid Julian/Gregorian calendar -([#5010](https://github.com/apache/datafusion-comet/issues/5010)). -[Known Issues](parquet-iceberg-known-issues.md#parquet-reads) lists them, with the -issues that track them. - -The following limitations raise an error at scan time rather than falling back to Spark. More -reads that fail where Spark succeeds are listed in -[Known Issues](parquet-iceberg-known-issues.md#parquet-reads-that-fail): - -- Selecting a field by name when multiple physical siblings match, including inside structs, - arrays, and maps. Comet raises a duplicate-field error instead of resolving the collision. - Checks cover referenced columns, including predicates; unselected roots do not prevent - reading a unique field by name or field ID. Exact-name projections of unique children in - structs and arrays of structs remain supported. Casts that cannot use this pruning reject - byte-identical duplicate siblings anywhere in the decoded physical subtree, including maps. - Field-ID resolution retains precedence, but selecting a byte-identically duplicated physical - root name still raises a duplicate-field error, even when the requested field is renamed. - Names in separate groups do not collide. Spark may read a duplicate-bearing file with an - explicit schema in case-sensitive mode, but its choice of sibling depends on the field shape - and can produce unexpected values. Spark rejects schema inference from a single file with - duplicate names; inference across files can depend on merge order. - Resolution is tracked in [#5884](https://github.com/apache/datafusion-comet/issues/5884), - with mixed-type behavior in [#5964](https://github.com/apache/datafusion-comet/issues/5964). -- Invalid UTF-8 bytes in `STRING` columns. Spark permits arbitrary byte sequences in a `STRING` - column (for example from `CAST(X'C1' AS STRING)`), but Comet's native execution path is built on - Arrow, whose string type is strictly UTF-8. Reading a Parquet file whose `STRING` column contains - non-UTF-8 bytes fails with `Parquet error: encountered non UTF-8 data`. Disable Comet for the - query, or cast the column to `BINARY` before persisting, if you need to preserve non-UTF-8 bytes. - By contrast, Comet decodes non-UTF-8 bytes at the JVM-to-native Arrow FFI boundaries using - JVM-compatible replacement semantics. This covers native query input from the JVM-exported Arrow - stream (including JVM scans, shuffle reads, and `mapInArrow`), columnar-to-row conversion, and JVM - UDF results. It does not change the native Parquet reader: rejecting invalid UTF-8 there remains - Gap A of [#4764](https://github.com/apache/datafusion-comet/issues/4764). See - [Strings with non-UTF-8 bytes](index.md#strings-with-non-utf-8-bytes), - [#4121](https://github.com/apache/datafusion-comet/issues/4121), and the tracking issue above. -- Reading `TimestampLTZ` as `TimestampNTZ` on Spark 3.x. Spark raises an error per - [SPARK-36182](https://issues.apache.org/jira/browse/SPARK-36182) because LTZ encodes UTC-adjusted - instants that cannot be safely reinterpreted as timezone-free values, and Comet matches this by - rejecting the read. This applies to all LTZ physical encodings (INT96, TIMESTAMP_MICROS, - TIMESTAMP_MILLIS). On Spark 4.0+, this read is permitted - ([SPARK-47447](https://issues.apache.org/jira/browse/SPARK-47447)) and Comet matches Spark's - behavior. - -### Schema Mismatch Handling - -The issues in this subsection apply only when the requested read schema differs from the schema written -to the Parquet file. They do **not** affect a plain `spark.read.parquet(path)` that infers the schema -from file metadata, because in that case the requested schema and file schema match by construction. -Schema mismatch happens in two real-world scenarios: - -1. The user provides an explicit read schema: `spark.read.schema().parquet(path)` (or the - equivalent DataFrame API). -2. **Schema evolution / partitioned reads** where files in a single dataset were written at different - times with different types, or a table-format catalog (Iceberg, Delta) records a logical schema - that has evolved past one or more underlying Parquet files. Spark coerces the file types to the - table types at read time. - -Spark's vectorized Parquet reader fully validates these conversions in `ParquetVectorUpdaterFactory.getUpdater` -and throws `SchemaColumnConvertNotSupportedException` for unsupported pairs. Comet's Parquet scan -mirrors that validation in its schema adapter; the entries below are the remaining gaps. - -Note that the exact set of accepted conversions has changed between Spark versions -(for example, Spark 3.x's `schemaEvolution.enabled` flag gates `INT32 → INT64`, `FLOAT → DOUBLE`, -and `INT32 → DOUBLE` widening that Spark 4.0+ accepts unconditionally; `TimestampLTZ → TimestampNTZ` -is rejected by Spark 3.x but accepted by Spark 4.0+). Comet aims to follow the per-version Spark -behavior. - -- **Conversion errors precede pushed row filters**. Empty files and files whose row groups or - pages are all pruned read without decode-time conversion errors. When row-filter pushdown is - enabled, Comet checks rejected conversions before reading surviving data pages, so a row filter - cannot suppress the error by discarding every row. Legacy LIST shape mismatches that Spark cannot - clip are rejected when the file is opened, including for empty and fully pruned files. -- **List conversion error paths assume Spark's standard encoding**. Comet inserts `list` - before the element name when reporting a rejected array element conversion. Arrow's schema - omits the repeated group name, so paths for legacy LIST encodings or custom group names may - differ from Spark's Parquet column path. -- **`ParquetSchemaConvert` errors do not include the file path**. The mismatch itself is detected and - rejected correctly, but the resulting Spark error message reads - `Encountered error while reading file . Data type mismatches…` (note the empty path). Behavior is - consistent across Spark versions. See - [#4316](https://github.com/apache/datafusion-comet/issues/4316). -- **Spark 3.x: extra `SparkException` layer in the cause chain**. The native error is translated to a - `SparkException` whose cause is `SchemaColumnConvertNotSupportedException` (matching what Spark would - throw); on Spark 3.x the executor / task error handling re-wraps this once more on the way back to - the driver, producing a two-level chain (`SparkException → SparkException → -SchemaColumnConvertNotSupportedException`) instead of the one-level chain Spark's own vectorized - reader produces. Code that catches `SparkException` and inspects only the immediate cause via - `e.getCause.isInstanceOf[SchemaColumnConvertNotSupportedException]` will see the inner - `SparkException` instead. Walk the cause chain to recover the - `SchemaColumnConvertNotSupportedException`. Spark 4.0+ produces a single-level chain, matching - vanilla Spark. See [#4354](https://github.com/apache/datafusion-comet/issues/4354). diff --git a/docs/source/user-guide/latest/compatibility/spark-versions.md b/docs/source/user-guide/latest/compatibility/spark-versions.md index f178b679f94..3ae6fb1b336 100644 --- a/docs/source/user-guide/latest/compatibility/spark-versions.md +++ b/docs/source/user-guide/latest/compatibility/spark-versions.md @@ -43,7 +43,7 @@ release than on the other supported versions. We recommend moving to Spark 3.5 o - **Extra `SparkException` layer in Parquet schema mismatch errors**: when a Parquet read is rejected because a file's type cannot be converted to the requested type, the error's cause chain has one more `SparkException` layer than Spark's own reader produces. - See [Parquet Compatibility](scans.md#schema-mismatch-handling) for details. + See [Parquet](../parquet.md#schema-mismatch-handling) for details. ## Spark 3.5 @@ -54,7 +54,7 @@ Spark 3.5.9 is supported with Java 17 and Scala 2.12/2.13. - **Extra `SparkException` layer in Parquet schema mismatch errors**: when a Parquet read is rejected because a file's type cannot be converted to the requested type, the error's cause chain has one more `SparkException` layer than Spark's own reader produces. - See [Parquet Compatibility](scans.md#schema-mismatch-handling) for details. + See [Parquet](../parquet.md#schema-mismatch-handling) for details. ## Spark 4.0 diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 9d102e71d5e..f687d3ab3b3 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -23,17 +23,14 @@ ### Parquet -Parquet scans run in Rust via DataFusion if all data types in the schema are supported. When the scan -falls back to Spark, enabling `spark.comet.convert.parquet.enabled` will immediately convert the data into -Arrow format, allowing the Comet pipeline to take over after that, but the process may not be efficient. +Parquet scans run in Rust via DataFusion if all data types in the schema are supported. See the +[Parquet](parquet.md) page for what falls back to Spark, differences from Spark, tuning, and known +issues. ### Apache Iceberg 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 +writes natively, an opt-in feature. See the [Iceberg](iceberg.md) page for more information. ### CSV @@ -109,7 +106,7 @@ them, and the conversion and the Comet operators above it would read ahead of th ### Apache Iceberg -See the dedicated [Comet and Iceberg Guide](iceberg.md). +See the dedicated [Iceberg](iceberg.md) page. ## Supported Storages @@ -154,7 +151,7 @@ the HDFS read itself, typically while an unrelated query is running. ``` Native Iceberg scans do not support HDFS-backed tables; those scans fall back to Spark. See the -[Comet and Iceberg Guide](iceberg.md). +[Iceberg](iceberg.md) page. ### Building Comet with HDFS support @@ -383,7 +380,7 @@ Once translated, these values are applied at the same `fs.s3a.bucket.{bucket-nam described in [Additional S3 Configuration Options](#additional-s3-configuration-options), so the credential providers and options documented above also apply to alias-scheme URLs. The same translation feeds the native Iceberg scan; see -[Object store configuration (S3)](iceberg.md#object-store-configuration-s3) in the Iceberg guide. +[Object store configuration (S3)](iceberg.md#object-store-configuration-s3) on the Iceberg page. A native Parquet or CSV scan whose alias-scheme paths span more than one bucket resolves the alias settings for each bucket. Alias schemes apply to native scans only: a native Iceberg write to an @@ -419,7 +416,7 @@ $SPARK_HOME/bin/spark-shell \ ... ``` -### Limitations +### S3 Limitations Comet's S3 support has the following limitations: @@ -427,6 +424,10 @@ Comet's S3 support has the following limitations: 2. **Custom credential providers**: Custom credential provider classes named in `fs.s3a.aws.credentials.provider` are not supported; only the standard providers listed in the table above are. To route credential requests through your own Java code, implement Comet's `CometS3CredentialProvider` SPI; see [S3 Credential Providers](s3-credential-providers.md). Broader Hadoop S3A integration is tracked in [#1829](https://github.com/apache/datafusion-comet/issues/1829). +3. **Settings that Hadoop's S3A client resolves differently**: The native scans resolve storage settings with their own client rather than with Hadoop's. Where the two disagree, a native read can use a different identity or route than Spark and fail where Spark succeeds. Examples are JCEKS-backed secrets and the precedence of per-bucket settings, which the native Parquet scan forwards without checking for a difference ([#5662](https://github.com/apache/datafusion-comet/issues/5662)). + +4. **AWS profile credential providers**: The native scans ignore `fs.s3a.auth.profile.name` and `fs.s3a.auth.profile.file` ([#4245](https://github.com/apache/datafusion-comet/issues/4245)), and resolve some profiles differently from Hadoop 3.4 and later, for example profiles that use `credential_process` or chains of assumed roles ([#6575](https://github.com/apache/datafusion-comet/issues/6575)). + ## Azure Comet's Parquet scan reads Azure Data Lake Storage Gen2 (ADLS Gen2) through the @@ -542,7 +543,7 @@ Comet 1.1 started from the environment and overlaid the Hadoop keys it found by - A lower-case `oauth` auth type, or OAuth keys without `fs.azure.account.auth.type`, fail the way they fail for Spark. In the second case Hadoop defaults to `SharedKey` and the key provider fails. On hadoop-azure 3.4.1 and later the driver log names the missing key. 3.3.4 reports only `Failure to initialize configuration`. - A configured mechanism the native scan cannot build fails with an error instead of silently reading the environment. -### Limitations +### Azure Limitations 1. **Authentication keys only**: Only the keys in the table above reach the native store. Other `fs.azure.*` settings (read-ahead, retries, HTTP tuning) do not apply to the native scan. @@ -551,3 +552,5 @@ Comet 1.1 started from the environment and overlaid the Hadoop keys it found by 3. **URL shape**: URLs must include the account in the host, i.e. `abfss://@.dfs.core.windows.net/`. Bare `abfs:///` (fsspec-style, no account in the URL) is not supported because Comet cannot resolve the storage account name. 4. **Endpoint hosts**: `object_store` builds an Azure store only for hosts under `dfs.core.windows.net`, `blob.core.windows.net`, `dfs.fabric.microsoft.com` and `blob.fabric.microsoft.com`. Sovereign-cloud endpoints such as `dfs.core.chinacloudapi.cn` are not supported, whatever keys are configured for them. + +5. **Credentials that resolve differently from Hadoop's ABFS driver**: Container-scoped OAuth settings, Fabric hosts and a few other configurations resolve differently from Hadoop's ABFS driver, so a native read can fail where Spark succeeds ([#6605](https://github.com/apache/datafusion-comet/issues/6605)). diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md index 6ffedb1f1ee..061d7ae46bf 100644 --- a/docs/source/user-guide/latest/datatypes.md +++ b/docs/source/user-guide/latest/datatypes.md @@ -131,6 +131,6 @@ Shredded reconstruction can be slower than Spark's reader; see the ## See also - [Comet Compatibility Guide](compatibility/index.md) - known incompatibilities and edge cases. -- [Parquet Scan Compatibility](compatibility/scans.md) - per-type behavior at scan time. +- [Parquet](parquet.md#reading) - per-type behavior at scan time. - [Supported Spark Operators](operators.md) - the equivalent reference for operators. - [Supported Spark Expressions](expressions.md) - the equivalent reference for expressions. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index dfe819139de..ddf3df28923 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -71,7 +71,7 @@ expressions. The following function families are **not currently planned** for n - **UTF-8 validation** (`is_valid_utf8`, `make_valid_utf8`, `validate_utf8`, `try_validate_utf8`): niche Spark 4.x string-validation helpers. - **Miscellaneous niche** (`histogram_numeric`, `version`, `sentences`, `quote`): low-value or specialized functions with little benefit from native acceleration. -The file-metadata functions `input_file_name`, `input_file_block_start`, and `input_file_block_length` depend on scan-internal per-row file information rather than the expression layer; their support status is covered in the [scan compatibility guide](compatibility/scans.md). +The file-metadata functions `input_file_name`, `input_file_block_start`, and `input_file_block_length` depend on scan-internal per-row file information rather than the expression layer; their support status is covered in the [Parquet page](parquet.md#reading). Note that `median` and `mode` are supported: they are mainstream exact aggregates. `mode` runs natively only with `spark.comet.expression.Mode.allowIncompatible=true`. `approx_count_distinct` is supported because Comet ports Spark's `HyperLogLogPlusPlus` exactly, so its result is bit-identical to Spark. diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md deleted file mode 100644 index 399d98fbaea..00000000000 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ /dev/null @@ -1,498 +0,0 @@ - - -# Iceberg Writes: Comet's Split-Operator Plan and Native Writer - -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). - -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 - -Spark writes an Iceberg table through a single physical operator that combines data-file -writing with metadata writing, committing, and catalog validation. Spark's Adaptive Query -Execution (AQE) already re-plans the sub-query feeding that operator — the scans, projects, -sorts, and exchanges producing the rows — but the operator itself sits outside AQE, so the -data-file writing cannot be re-planned in response to how its input ran. And because data-file -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`, 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 - feeding it run inside AQE. -2. **`IcebergCommit`** — collects the commit messages on the driver and performs the normal - Iceberg commit (including commit-time validation), outside AQE, exactly once. - -The split moves data-file writing inside AQE and separates it from the commit, which gives Comet -a step to replace: when the write passes the eligibility check below, the `IcebergWrite` -operator's per-task Parquet write is delegated to -[iceberg-rust](https://github.com/apache/iceberg-rust) via Comet's native execution pipeline -([#5361](https://github.com/apache/datafusion-comet/pull/5361)). A write that does not pass keeps -`IcebergWrite`, and iceberg-java writes its data files; only the plan shape changes. - -## How the native write works - -The JVM-side planner marshals everything iceberg-rust needs — the write schema and partition -spec as JSON, the data location, the resolved parquet writer settings, the writer mode -(unpartitioned / fanout / clustered, mirroring `SparkWrite`'s own choice), object-store -configuration (the table's `FileIO` properties, e.g. REST-vended credentials, merged over -`fs.s3a.*` settings translated from the effective Hadoop configuration carried by the table's -`FileIO`, including catalog-specific `hadoop.*` overrides, since `HadoopFileIO` carries its S3 -configuration in the Hadoop Configuration rather than in `FileIO` properties), and per-task IDs — -into the serialized native plan. On each task, iceberg-rust writes the Parquet files and -returns its `DataFile` metadata packed as a single in-memory Iceberg V2 data manifest; the JVM -decodes those bytes with Iceberg's own `ManifestFiles.read`, re-derives each file's manifest -metrics from the written Parquet footer with Iceberg's `MetricsConfig` logic (so metrics modes, -truncation, and bounds decisions are iceberg-java's by construction), and wraps the result in -the same `TaskCommit` message the JVM writer would have produced. Everything iceberg-java does -post-write — snapshot assignment, manifest-list aggregation, commit validation and retries — -is untouched: `IcebergCommit` performs the normal `BatchWrite.commit`. - -For `ResolvingFileIO`, Hadoop settings are taken from the delegate opening the data location. -An S3 location handled by `S3FileIO` uses its initialized FileIO properties, so Hadoop options -on the wrapper do not change the native endpoint or encryption settings. If that delegate cannot -be resolved, the write falls back to iceberg-java. - -Before a task opens a partition's first file, it holds the partition's initial rows in memory -to choose which columns to dictionary-encode using parquet-mr's size accounting (see the -accepted divergences below). It normally waits for at least `write.parquet.page-row-limit` -rows (at least 100 when the configured limit is lower), or until the partition ends. The -buffering threshold is `write.parquet.row-group-size-bytes`, shared across all partitions in -a fanout write. Reaching it makes each partition still holding rows choose from the rows it -already has, even if they do not fill its first page. - -## Configuration - -Standard Comet + Iceberg setup (see [`iceberg.md`](iceberg.md)) plus the write setting, which is -off by default: - -``` -# Standard Comet / Iceberg wiring -spark.plugins=org.apache.spark.CometPlugin -spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions -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 -``` - -## Supported operations - -The split-operator plan supports the following operations on every Spark version Comet -supports: - -- `INSERT INTO` / DataFrame `append` (`AppendData`) -- `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 -`MergeRowsExec` breaks the fully-native child chain required by `CometIcebergWriteExec`. -On Spark 4.1+, the native MergeRows path also preserves the semantic counters required by the -summary-aware writer commit contract. - -For copy-on-write row-level DML, the mechanism differs by Spark version: on Spark 4.0+ the -analyzer emits operation-coded rows that Comet's writer dispatches through `ReplaceData`'s -projections, while on Spark 3.4/3.5 the rewritten rows are written as a plain row stream. The -supported set of operations is the same either way. - -Merge-on-read uses Spark's `WriteDelta`, and Iceberg's JVM `DeltaWriter` writes its rows under -either plan, so it keeps Spark's stock write plan. On Spark 3.5+ the testing setting -`spark.comet.write.iceberg.splitOperator.enabled` plans it as the split plan, with the same driver -commit and reporting path, but `CometIcebergWriteExec` is never used for position-delta rows. -Spark 3.4 always leaves `WriteDelta` on Spark's stock write plan. - -On Spark 4.1+ the split plan matches two further stock-Spark behaviours: MERGE metrics are -forwarded to the writer's commit (Iceberg 1.11+ records them in the snapshot summary), and -cached catalog tables are recached by name after a write so cache entries survive schema -changes. - -## When Comet falls back to Spark's write operator - -The rewrite is skipped — and the write runs through Spark's stock combined operator — when: - -- `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 - (`spark.comet.exec.enabled=false`); -- Comet is in plan-only mode (`spark.comet.explain.planOnly.enabled=true`); -- the write is neither an Iceberg `SparkWrite` nor a supported Iceberg position-delta write; -- the table uses merge-on-read (`WriteDelta`), unless the testing setting - `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 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. - -## Native Parquet write eligibility - -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, -statistics, and manifest metadata — so a write is only eligible when every table property it -depends on is one the native path reproduces exactly, and additionally only when the plan -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#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 -`df.writeTo(...).append()` on a DataFrame built from local data, is fed by Spark's -`LocalTableScanExec`, which Comet only converts when `spark.comet.exec.localTableScan.enabled=true` -(off by default). Without that setting such writes keep `IcebergWrite` and run through -iceberg-java. - -**Most Iceberg write settings are not supported.** Detection is an allowlist: a write is -eligible only when its entire effective configuration matches the table below, and anything -else — any other write-affecting property, any key added by a future Iceberg version, any -value outside the supported set, any reflection failure while inspecting the write — falls -back to iceberg-java with a reason reported in extended EXPLAIN. Checks run on the effective -configuration: table properties overlaid with `SparkWrite.writeProperties`, which is where -iceberg-java resolves per-write options and `spark.sql.iceberg.*` session overrides. - -A write is eligible only when ALL of the following hold: - -| Setting | Supported values | -| ------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| resolved write format (`write-format` option overlaid on `write.format.default`) | `parquet` | -| `format-version` | `1`, `2` or `3` (on `3`, writes that carry row lineage fall back; see below) | -| `write.parquet.compression-codec` / `compression-level` / `row-group-size-bytes` / `page-size-bytes` / `page-row-limit` / `dict-size-bytes` | the sizes/limit must be positive Java ints (`Integer.parseInt` semantics — no trimming, no values past `Int.MaxValue` — matching iceberg-java, whose writer fails on anything else); `compression-level` must be a Java int within the native writer's per-codec range (zstd 1–22, gzip 0–9, brotli 0–11; ignored by both writers for snappy/lz4/none) — iceberg-java never validates the level, so an out-of-range value falls back rather than becoming a native task failure | -| `write.parquet.row-group-check-min-record-count` | unset or `100` (the default) | -| `write.parquet.row-group-check-max-record-count` | unset or `10000` (the default) | -| `write.parquet.page-version` | unset or `v1` | -| `write.parquet.shred-variants` | unset or `false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write) | -| `write.parquet.variant-inference-buffer-size` | any value (only meaningful when shredding, which is gated) | -| `write.parquet.bloom-filter-enabled.column.` | unset or `false` | -| `write.metadata.metrics.*` | any value (manifest metrics are re-derived on the JVM with Iceberg's own logic) | -| `write.spark.fanout.enabled` | any value (the native writer implements both clustered and fanout modes) | -| `write.target-file-size-bytes` | any value (the two writers can choose different roll points; see accepted divergences) | -| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs`, matched case-sensitively (`S3://` falls back). `s3`, `s3a` and `gs` need a bucket in the authority (`s3://bucket/...`), so a hostless form such as `s3:/bucket/key` falls back. `gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below | -| resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` | -| Hadoop S3A settings for an `s3` / `s3a` data location | only `fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`, `endpoint.region`, and `path.style.access`, including their `fs.s3a.bucket..*` forms; any other effective `fs.s3a.*` setting falls back, apart from the ignored settings listed below | -| Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location | the S3 endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`, `kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain settings parsed by the pinned iceberg-rust version, plus Comet's credential-provider class and built-in web-identity properties. When a custom provider is configured, its vendor-owned `s3.*` / `client.*` properties are also forwarded; unsupported Iceberg-defined S3 settings still fall back | -| partition spec | any, except a `void` field whose source column was dropped beside a live field (see below) | -| column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) and the v3 types `variant`, `unknown`, `timestamp_ns`, `geometry` and `geography` | - -Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` — -everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g. -`bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions), -any `parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*` -key in the session Hadoop configuration other than the reader-only -`parquet.hadoop.vectored.io.enabled` (with `HadoopFileIO`-backed output those reach -iceberg-java's writer but not the native one). Also gated explicitly: any `encryption.*` key, -`write.object-storage.enabled=true`, `write.location-provider.impl`, and `io-impl`. - -Three checks look past properties at the table's instantiated state, because they can be -configured at the catalog level (or by a custom `TableOperations`) without any table or write -property changing: `table.locationProvider()` must be Iceberg's `DefaultLocationProvider`; -`table.io()` must be a recognized `FileIO` (the same allowlist the native scan uses, minus the -`EncryptingFileIO` family — the native writer produces plaintext files, so an encrypting `FileIO` -is rejected on the write side); and `table.encryption()` must be Iceberg's -`PlaintextEncryptionManager`. Anything else falls back. - -A `gs` data location additionally requires that the `FileIO` actually opening it is a -`GCSFileIO` (for a `ResolvingFileIO`, the delegate it instantiates for that location, -which is a `HadoopFileIO` when the GCS `FileIO` cannot be loaded or initialized). A -`HadoopFileIO` takes its GCS credentials, endpoint and project from `fs.gs.*` in the Hadoop -Configuration, and only `fs.s3a.*` is translated into the native `FileIO`, so the native writer -could resolve a different storage identity or endpoint than the JVM writer would. That -combination falls back; a `GCSFileIO` carries its `gcs.*` settings in `FileIO.properties()`, -which are forwarded. - -For an `s3` or `s3a` data location, the gate also inspects both the table FileIO's effective Hadoop -configuration and `table.io().properties()`. These are separate allowlists because Hadoop S3A -keys are translated before they reach iceberg-rust, while Iceberg `FileIO` keys are forwarded -directly. When the FileIO exposes a Hadoop configuration, its initialized values govern both the -gate and native translation, including after session or catalog options change. FileIO -implementations without a Hadoop configuration, such as `S3FileIO`, use only their initialized -properties; session and catalog Hadoop options are neither checked nor forwarded. Hadoop's built-in -`core-default.xml` values are not treated as explicit settings, but -programmatic settings and values from site or custom `*-default.xml` resources are. Spark's -session-wide S3A vectored-read and `downgrade.syncable.exceptions` compatibility settings are also -ignored, as are the S3A committer (`committer.name`, `committer.magic.enabled`, -`committer.threads`), `bulk.delete.page.size` and `experimental.input.fadvise` settings, because -none of them can alter an Iceberg data-file write request. Spark 4.1 sets `committer.name` and -`committer.magic.enabled` itself when `spark-hadoop-cloud` is on the classpath. Ignored settings are -matched by their plain `fs.s3a.` keys, so a per-bucket `fs.s3a.bucket..` -form still falls back. Unknown explicit -`fs.s3a.*`, `s3.*`, or `client.*` settings therefore fall back at planning time instead of being -silently ignored by the native storage backend. The exception is a vendor-owned `s3.*` / -`client.*` property when -`s3.comet.credential.provider.class` is configured: the provider receives the unfiltered FileIO -bag and can consume that property. Iceberg-defined settings that the native storage path cannot -honour still fall back even with a provider. A per-bucket Hadoop setting counts only for the exact -data-bucket name, so configuration for a longer dotted bucket does not by itself disable the -native write. If Iceberg's AWS property classes cannot be loaded, vendor `s3.*` / `client.*` keys -fall back too and planning still completes. The fall-back reason reports only sorted property -names, never their values, so credentials and tokens do not enter EXPLAIN or plan logs. - -On a format-version 3 table with Iceberg 1.10 or newer, a write that rewrites existing rows falls -back. Copy-on-write `DELETE`, `UPDATE` and `MERGE` and `rewrite_data_files` write the row lineage -columns `_row_id` and `_last_updated_sequence_number` into the new data files, so that rewritten -rows keep their ids, and the native writer does not write those columns. Appends, overwrites, -`CREATE TABLE ... AS SELECT` and `REPLACE TABLE ... AS SELECT` write the data columns only, and -Iceberg assigns the new rows their ids when it commits, so those run natively. Earlier Iceberg -versions write no lineage columns, so the rule never applies to them. - -A format-version-1 table keeps a dropped partition field as a `void` field, and its source column -can be dropped afterwards. A spec that mixes such a field with a live one falls back: iceberg-java -cannot write through it either, so the write fails with iceberg-java's own error rather than in -the native writer ([#6141](https://github.com/apache/datafusion-comet/issues/6141)). A spec whose -fields are all `void` writes unpartitioned and stays eligible. - -Other `write.*` properties are intentionally not gated because they cannot make the native -writer produce different data files: distribution and ordering settings shape the Spark plan -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. Most native eligibility rules are pinned by -`CometIcebergWriteDetectionSuite`. - -Manifest `DataFile` metrics are assembled on the JVM before commit: each written file's -metrics are re-derived from its parquet footer through the version-matched -`ParquetUtil.footerMetrics` and `MetricsConfig.forTable`, with float/double NaN counts and -bounds carried over from the native writer's tracked state. iceberg-java's metadata decisions -— metrics modes, the inferred-column cap -(`write.metadata.metrics.max-inferred-column-defaults`), bound truncation, and list/map bounds -suppression — are therefore applied by iceberg-java's own code regardless of what the native -writer reports. - -iceberg-java's writer takes these metrics from the footer it still holds in memory, but the native -path reads each footer back from storage: two small reads per written file, one for the footer's -length and one for the footer itself, made one file after another before the task finishes. On S3 -or GCS each read is a GET request, so the task waits two request round trips per file. That is -small next to uploading a large file, but a write that produces many small files, such as a fanout -write over many partitions, pays it for every one of them. - -## Failure handling - -Eligibility is decided entirely at plan time. That includes the reflection surface: every -iceberg-java class, method, and constructor the executor-side commit-message assembly uses is -eagerly resolved by the eligibility gate on the driver, so an Iceberg release that moves any -of them declines the native path with a fall-back reason instead of failing tasks mid-write. -Once planned, the physical plan is fixed — there is no per-task re-decision or runtime switch -back to the JVM writer. - -When a native write fails partway through a task (an object-store error, a data-dependent cast -failure), the error propagates as an ordinary Spark task failure and Spark's task retry -re-executes it — through the native writer again. Retries cannot collide: each attempt's task -attempt id is embedded in its data file names. - -The native writer's buffers are charged to Comet's memory pool, the off-heap budget Comet's other -native operators draw on, 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. Each open file holds the row group it is -writing in memory, up to `write.parquet.row-group-size-bytes`, and on S3 or GCS also the last row -group it flushed, which is uploaded once the next one is complete or the file closes. Each -partition also holds the rows that have not reached its file yet. So a task writing to many -partitions needs memory in proportion to them. When the pool cannot grant it, the task writes out -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 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 -open at a time. A larger `spark.memory.offHeap.size` also helps, and Iceberg's `rewrite_data_files` -procedure compacts small files once they are written. - -A write that keeps one file open, unpartitioned or clustered, has no partition to close. When that -file outgrows the pool, the task fails with a `CometNativeException` reading -`Additional allocation failed for IcebergWriteExec` instead of exceeding the executor's memory, and -Spark retries it like any other task failure. A smaller `write.parquet.row-group-size-bytes` or a -larger `spark.memory.offHeap.size` lets such a write fit. - -Partial results are never committed. The commit set is exactly the commit messages returned by -successful tasks — a failed task contributes none — and if the job fails, the driver-side -commit operator aborts without committing anything. A failed task attempt also deletes the -data files it created, as iceberg-java's writer abort does. The native writer records every -location it hands to a file writer, and cleanup has no ownership gap. Native keeps its cleanup -guard armed after yielding the output batch. The JVM reads those locations into a task failure -listener first, then polls the native output to EOF; that EOF is the acknowledgement that lets -native disarm. A failure in this narrow handoff window can therefore trigger best-effort deletion -on both sides, which is harmless. The native guard still covers a failed write, a task torn down -before the write completed — for example because the operator feeding it threw — and a failure -encoding the manifest or building the output batch. The handoff does not depend on decoding the -manifest: the locations are read before the manifest is decoded, so a failure in that decode still -cleans up. Cleanup never masks the original failure; anything it misses is invisible to every -reader, since readers resolve files through committed manifests only, and is reclaimed by Iceberg's -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`. (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 -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. Enabling -`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), 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) - -No Iceberg reader bases a planning or correctness decision on these; they change the bytes of -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` 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`). -- 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 - ([#6114](https://github.com/apache/datafusion-comet/issues/6114)). The native writer decides - once per partition, normally from that partition's first page of rows in the task, and keeps the - decision for every file and row group it writes for the partition; parquet-mr decides again - for every row group. If buffered rows reach `write.parquet.row-group-size-bytes`, the choice - uses the rows collected so far. In a fanout write, all partitions share this threshold, so - each partition still buffering can make its choice before it has a full first page. Later - rows could have changed parquet-mr's decision, so this difference is not limited to columns - close to the size cut-off. Where the page size rather than `write.parquet.page-row-limit` ends a - column's first page, the native page ends at the first row past parquet-mr's size threshold, - while parquet-mr only ends it at its next periodic size check, so a column close to the - cut-off can be decided the other way. Close to the cut-off both encodings take about the same - 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: 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. On Iceberg 1.5.x, - which the Spark 3.4 profile pins, iceberg-java itself spelled `timestamp` and `timestamptz` - directories with `LocalDateTime.toString()` / `OffsetDateTime.toString()` - (`ts=1969-12-31T23:59:58.500Z`) and left the partition field name unescaped; Comet uses the - 1.8+ spelling on every profile. Distinct partition values still get distinct directories in all - cases, and no reader parses these names: files are resolved through committed manifests. -- File rolling lands on the same row grid as iceberg-java but not necessarily on the same row. - Both writers re-check the current file's size against `write.target-file-size-bytes` once - every 1000 rows of that file (iceberg-java's `RollingFileWriter.ROWS_DIVISOR`; Comet hands the - iceberg-rust writer rows in 1000-row units, per partition file, to get the same grid), so each - writer rolls only on a 1000-row boundary of its own file. - The shared grid is all that is shared. What each writer compares against the target differs — - flushed bytes plus parquet-rs's estimate of the open row group, versus parquet-mr's file - position plus its buffered size — and the two use different threshold comparisons. These are - independent size estimates, so nothing bounds how far apart the two writers' roll points are: - they may cross the target several grid steps apart, and the resulting files can differ in row - count by an arbitrary number of 1000-row blocks. Do not rely on file-layout parity between the - two writers; rely only on each file rolling on its own 1000-row boundary. -- A fanout write that the memory pool cannot hold closes partitions before the task ends (see - [Failure handling](#failure-handling)), where iceberg-java's fanout writer keeps every file open - until then. So a partition can have more, smaller files than iceberg-java writes, and a file - closed early ends off the 1000-row grid, as the last file of a task does. A partition closed - before its rows filled a first page makes its dictionary choice from the rows it has. -- A fanout write lists a task's data files in file-path order, where iceberg-java lists them in - its own `StructLikeMap` iteration order. Both are stable across runs, and neither is a - documented ordering, but the manifest entry order becomes the scan-task order and so the row - order of an unordered `SELECT *`. On a format-version 3 table it also decides which row ids the - commit gives each file's rows, so the same rows can get different `_row_id` values from the two - writers. The ids are unique either way, and across tasks iceberg-java's own assignment already - depends on the order in which the tasks finish. Only the sorted order is reproducible on the - native path: the order the native writer closes its files in depends on the order rows arrive in - 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, 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 - default) are not gated: they can only shift compressed bytes, which are already accepted as - divergent. - -### Manifest metadata visible to later readers - -Manifest metrics drive partition- and file-level pruning for every future reader of the table, -so a divergence here would outlive the write. This class is deliberately kept almost empty: -`DataFile` metrics are not taken from the native writer's manifest but re-derived on the JVM -from each written file's parquet footer through iceberg-java's own `ParquetUtil.footerMetrics` -and `MetricsConfig.forTable` (see above). Metrics modes, lower/upper bound truncation, the -null-count conventions, and list/map bounds suppression are therefore iceberg-java's code -making iceberg-java's decisions. The parity tests write the same rows through both writers and -compare the committed `readable_metrics` (value, null and NaN counts, and lower and upper bounds) -for the column types they cover. Two footer-derived values can still differ from what -iceberg-java's _writer-tracked_ state would have recorded, and both are analyzed safe: - -- Float/double bounds involving zero may differ in sign: parquet-rs normalises footer - statistics to min `-0.0` / max `+0.0` (the parquet-format recommendation), while - iceberg-java's writer-tracked bounds preserve the exact sign it saw. The native path's - manifest bounds inherit the normalised values — a strictly conservative widening that cannot - change pruning decisions. -- On Iceberg 1.9+, manifest `value_counts` / `null_value_counts` for float/double columns - nested under a nullable struct count rows whose parent struct is null (they come from the - parquet footer), while iceberg-java's writer-tracked counts do not. The native counts are the - correct ones. iceberg-java's counts leave out rows whose struct is null. When every non-null - struct in a file holds a value, iceberg-java records a null count of 0, and `IS NULL` prunes a - file that contains matching rows - ([apache/iceberg#17561](https://github.com/apache/iceberg/issues/17561)). The native counts - never prune such a file. `IS NOT NULL` pruning is the same for both. - -All content not listed above — the logical data, statistics values, and manifest metadata — -must match iceberg-java exactly, or the write falls back. The open bugs in which a native write -still differs are listed in -[Known Issues](compatibility/parquet-iceberg-known-issues.md#iceberg-writes). diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 0df6319f60b..b24bc258539 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -17,7 +17,17 @@ under the License. --> -# Accelerating Apache Iceberg Parquet Scans using Comet +# Apache Iceberg + +Comet accelerates Iceberg tables whose data files are Parquet, in two ways: + +- **[Reading](#native-reader)**: the native Iceberg reader scans the table's data and delete + files in Rust. It is enabled by default. +- **[Writing](#writes)**: with `spark.comet.write.iceberg.enabled=true`, Comet writes the data + files of eligible writes natively. It is disabled by default. + +This page covers both, followed by the [known issues](#known-issues). Behavior that applies to +Parquet files in general is documented on the [Parquet](parquet.md) page. ## Native Reader @@ -56,7 +66,8 @@ Catalog configuration is standard Iceberg-on-Spark and independent of Comet. The Comet’s native Iceberg reader supports fetching multiple files in parallel to hide I/O latency with the config `spark.comet.scan.icebergNative.dataFileConcurrencyLimit`. This value defaults to 1 to maintain test behavior on Iceberg Java tests without `ORDER BY` clauses, but we suggest increasing it to -values between 2 and 8 based on your workload. +values between 2 and 8 based on your workload, at the cost of extra memory. This helps most for +tables with many small files or high-latency storage. ### Supported features @@ -202,13 +213,12 @@ The following scenarios will fall back to the JVM Iceberg reader: column the same name) The cases in which the native reader returns different results from Spark, or fails where Spark -succeeds, without falling back are listed in -[Known Issues](compatibility/parquet-iceberg-known-issues.md). +succeeds, without falling back are listed under [Known Issues](#known-issues). 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. +see [Writes](#writes) for the plan, the native writer, and when each applies. ### Iceberg UDFs @@ -261,7 +271,7 @@ Every other `truncate` input type, and `bucket` on decimals, runs natively. This matters most for writes. A partitioned table with the default `write.distribution-mode` (`hash`) is planned with a shuffle and a local sort keyed on the partition transforms, and with -these functions native the whole sub-plan feeding the [native Iceberg writer](iceberg-writes.md) +these functions native the whole sub-plan feeding the [native Iceberg writer](#writes) stays in Comet. A `numBuckets` or `width` argument that is not a positive integer literal makes the expression fall back to Spark. @@ -270,3 +280,574 @@ the expression fall back to Spark. The native Iceberg reader populates Spark's task-level `inputMetrics.bytesRead` (visible in the Spark UI Stages tab) using the `bytes_read` counter from iceberg-rust's `ScanMetrics`. This counter includes bytes read from both data files and delete files. The scan's SQL metrics, including Iceberg's planning counters, are listed in the [Metrics Guide](metrics.md#cometicebergnativescan). Iceberg Java does not explicitly report `bytesRead` to Spark's task input metrics. On the iceberg Java path, any `bytesRead` value comes from Hadoop's filesystem-level I/O counters, not from Iceberg itself. Because Comet's native reader and the Hadoop filesystem use different counting mechanisms, the exact byte counts will differ between the two paths. + +## Writes + +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). + +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 + +Spark writes an Iceberg table through a single physical operator that combines data-file +writing with metadata writing, committing, and catalog validation. Spark's Adaptive Query +Execution (AQE) already re-plans the sub-query feeding that operator — the scans, projects, +sorts, and exchanges producing the rows — but the operator itself sits outside AQE, so the +data-file writing cannot be re-planned in response to how its input ran. And because data-file +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`, 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 + feeding it run inside AQE. +2. **`IcebergCommit`** — collects the commit messages on the driver and performs the normal + Iceberg commit (including commit-time validation), outside AQE, exactly once. + +The split moves data-file writing inside AQE and separates it from the commit, which gives Comet +a step to replace: when the write passes the eligibility check below, the `IcebergWrite` +operator's per-task Parquet write is delegated to +[iceberg-rust](https://github.com/apache/iceberg-rust) via Comet's native execution pipeline +([#5361](https://github.com/apache/datafusion-comet/pull/5361)). A write that does not pass keeps +`IcebergWrite`, and iceberg-java writes its data files; only the plan shape changes. + +### How the native write works + +The JVM-side planner marshals everything iceberg-rust needs — the write schema and partition +spec as JSON, the data location, the resolved parquet writer settings, the writer mode +(unpartitioned / fanout / clustered, mirroring `SparkWrite`'s own choice), object-store +configuration (the table's `FileIO` properties, e.g. REST-vended credentials, merged over +`fs.s3a.*` settings translated from the effective Hadoop configuration carried by the table's +`FileIO`, including catalog-specific `hadoop.*` overrides, since `HadoopFileIO` carries its S3 +configuration in the Hadoop Configuration rather than in `FileIO` properties), and per-task IDs — +into the serialized native plan. On each task, iceberg-rust writes the Parquet files and +returns its `DataFile` metadata packed as a single in-memory Iceberg V2 data manifest; the JVM +decodes those bytes with Iceberg's own `ManifestFiles.read`, re-derives each file's manifest +metrics from the written Parquet footer with Iceberg's `MetricsConfig` logic (so metrics modes, +truncation, and bounds decisions are iceberg-java's by construction), and wraps the result in +the same `TaskCommit` message the JVM writer would have produced. Everything iceberg-java does +post-write — snapshot assignment, manifest-list aggregation, commit validation and retries — +is untouched: `IcebergCommit` performs the normal `BatchWrite.commit`. + +For `ResolvingFileIO`, Hadoop settings are taken from the delegate opening the data location. +An S3 location handled by `S3FileIO` uses its initialized FileIO properties, so Hadoop options +on the wrapper do not change the native endpoint or encryption settings. If that delegate cannot +be resolved, the write falls back to iceberg-java. + +Before a task opens a partition's first file, it holds the partition's initial rows in memory +to choose which columns to dictionary-encode using parquet-mr's size accounting (see the +accepted divergences below). It normally waits for at least `write.parquet.page-row-limit` +rows (at least 100 when the configured limit is lower), or until the partition ends. The +buffering threshold is `write.parquet.row-group-size-bytes`, shared across all partitions in +a fanout write. Reaching it makes each partition still holding rows choose from the rows it +already has, even if they do not fill its first page. + +### Configuration + +Standard Comet + Iceberg setup (see [Native Reader](#native-reader)) plus the write setting, which is +off by default: + +``` +# Standard Comet / Iceberg wiring +spark.plugins=org.apache.spark.CometPlugin +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions +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 +``` + +### Supported operations + +The split-operator plan supports the following operations on every Spark version Comet +supports: + +- `INSERT INTO` / DataFrame `append` (`AppendData`) +- `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 +`MergeRowsExec` breaks the fully-native child chain required by `CometIcebergWriteExec`. +On Spark 4.1+, the native MergeRows path also preserves the semantic counters required by the +summary-aware writer commit contract. + +For copy-on-write row-level DML, the mechanism differs by Spark version: on Spark 4.0+ the +analyzer emits operation-coded rows that Comet's writer dispatches through `ReplaceData`'s +projections, while on Spark 3.4/3.5 the rewritten rows are written as a plain row stream. The +supported set of operations is the same either way. + +Merge-on-read uses Spark's `WriteDelta`, and Iceberg's JVM `DeltaWriter` writes its rows under +either plan, so it keeps Spark's stock write plan. On Spark 3.5+ the testing setting +`spark.comet.write.iceberg.splitOperator.enabled` plans it as the split plan, with the same driver +commit and reporting path, but `CometIcebergWriteExec` is never used for position-delta rows. +Spark 3.4 always leaves `WriteDelta` on Spark's stock write plan. + +On Spark 4.1+ the split plan matches two further stock-Spark behaviours: MERGE metrics are +forwarded to the writer's commit (Iceberg 1.11+ records them in the snapshot summary), and +cached catalog tables are recached by name after a write so cache entries survive schema +changes. + +### When Comet falls back to Spark's write operator + +The rewrite is skipped — and the write runs through Spark's stock combined operator — when: + +- `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 + (`spark.comet.exec.enabled=false`); +- Comet is in plan-only mode (`spark.comet.explain.planOnly.enabled=true`); +- the write is neither an Iceberg `SparkWrite` nor a supported Iceberg position-delta write; +- the table uses merge-on-read (`WriteDelta`), unless the testing setting + `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 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. + +### Native Parquet write eligibility + +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, +statistics, and manifest metadata — so a write is only eligible when every table property it +depends on is one the native path reproduces exactly, and additionally only when the plan +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-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 +`df.writeTo(...).append()` on a DataFrame built from local data, is fed by Spark's +`LocalTableScanExec`, which Comet only converts when `spark.comet.exec.localTableScan.enabled=true` +(off by default). Without that setting such writes keep `IcebergWrite` and run through +iceberg-java. + +**Most Iceberg write settings are not supported.** Detection is an allowlist: a write is +eligible only when its entire effective configuration matches the table below, and anything +else — any other write-affecting property, any key added by a future Iceberg version, any +value outside the supported set, any reflection failure while inspecting the write — falls +back to iceberg-java with a reason reported in extended EXPLAIN. Checks run on the effective +configuration: table properties overlaid with `SparkWrite.writeProperties`, which is where +iceberg-java resolves per-write options and `spark.sql.iceberg.*` session overrides. + +A write is eligible only when ALL of the following hold: + +| Setting | Supported values | +| ------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| resolved write format (`write-format` option overlaid on `write.format.default`) | `parquet` | +| `format-version` | `1`, `2` or `3` (on `3`, writes that carry row lineage fall back; see below) | +| `write.parquet.compression-codec` / `compression-level` / `row-group-size-bytes` / `page-size-bytes` / `page-row-limit` / `dict-size-bytes` | the sizes/limit must be positive Java ints (`Integer.parseInt` semantics — no trimming, no values past `Int.MaxValue` — matching iceberg-java, whose writer fails on anything else); `compression-level` must be a Java int within the native writer's per-codec range (zstd 1–22, gzip 0–9, brotli 0–11; ignored by both writers for snappy/lz4/none) — iceberg-java never validates the level, so an out-of-range value falls back rather than becoming a native task failure | +| `write.parquet.row-group-check-min-record-count` | unset or `100` (the default) | +| `write.parquet.row-group-check-max-record-count` | unset or `10000` (the default) | +| `write.parquet.page-version` | unset or `v1` | +| `write.parquet.shred-variants` | unset or `false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write) | +| `write.parquet.variant-inference-buffer-size` | any value (only meaningful when shredding, which is gated) | +| `write.parquet.bloom-filter-enabled.column.` | unset or `false` | +| `write.metadata.metrics.*` | any value (manifest metrics are re-derived on the JVM with Iceberg's own logic) | +| `write.spark.fanout.enabled` | any value (the native writer implements both clustered and fanout modes) | +| `write.target-file-size-bytes` | any value (the two writers can choose different roll points; see accepted divergences) | +| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs`, matched case-sensitively (`S3://` falls back). `s3`, `s3a` and `gs` need a bucket in the authority (`s3://bucket/...`), so a hostless form such as `s3:/bucket/key` falls back. `gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below | +| resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` | +| Hadoop S3A settings for an `s3` / `s3a` data location | only `fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`, `endpoint.region`, and `path.style.access`, including their `fs.s3a.bucket..*` forms; any other effective `fs.s3a.*` setting falls back, apart from the ignored settings listed below | +| Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location | the S3 endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`, `kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain settings parsed by the pinned iceberg-rust version, plus Comet's credential-provider class and built-in web-identity properties. When a custom provider is configured, its vendor-owned `s3.*` / `client.*` properties are also forwarded; unsupported Iceberg-defined S3 settings still fall back | +| partition spec | any, except a `void` field whose source column was dropped beside a live field (see below) | +| column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) and the v3 types `variant`, `unknown`, `timestamp_ns`, `geometry` and `geography` | + +Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` — +everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g. +`bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions), +any `parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*` +key in the session Hadoop configuration other than the reader-only +`parquet.hadoop.vectored.io.enabled` (with `HadoopFileIO`-backed output those reach +iceberg-java's writer but not the native one). Also gated explicitly: any `encryption.*` key, +`write.object-storage.enabled=true`, `write.location-provider.impl`, and `io-impl`. + +Three checks look past properties at the table's instantiated state, because they can be +configured at the catalog level (or by a custom `TableOperations`) without any table or write +property changing: `table.locationProvider()` must be Iceberg's `DefaultLocationProvider`; +`table.io()` must be a recognized `FileIO` (the same allowlist the native scan uses, minus the +`EncryptingFileIO` family — the native writer produces plaintext files, so an encrypting `FileIO` +is rejected on the write side); and `table.encryption()` must be Iceberg's +`PlaintextEncryptionManager`. Anything else falls back. + +A `gs` data location additionally requires that the `FileIO` actually opening it is a +`GCSFileIO` (for a `ResolvingFileIO`, the delegate it instantiates for that location, +which is a `HadoopFileIO` when the GCS `FileIO` cannot be loaded or initialized). A +`HadoopFileIO` takes its GCS credentials, endpoint and project from `fs.gs.*` in the Hadoop +Configuration, and only `fs.s3a.*` is translated into the native `FileIO`, so the native writer +could resolve a different storage identity or endpoint than the JVM writer would. That +combination falls back; a `GCSFileIO` carries its `gcs.*` settings in `FileIO.properties()`, +which are forwarded. + +For an `s3` or `s3a` data location, the gate also inspects both the table FileIO's effective Hadoop +configuration and `table.io().properties()`. These are separate allowlists because Hadoop S3A +keys are translated before they reach iceberg-rust, while Iceberg `FileIO` keys are forwarded +directly. When the FileIO exposes a Hadoop configuration, its initialized values govern both the +gate and native translation, including after session or catalog options change. FileIO +implementations without a Hadoop configuration, such as `S3FileIO`, use only their initialized +properties; session and catalog Hadoop options are neither checked nor forwarded. Hadoop's built-in +`core-default.xml` values are not treated as explicit settings, but +programmatic settings and values from site or custom `*-default.xml` resources are. Spark's +session-wide S3A vectored-read and `downgrade.syncable.exceptions` compatibility settings are also +ignored, as are the S3A committer (`committer.name`, `committer.magic.enabled`, +`committer.threads`), `bulk.delete.page.size` and `experimental.input.fadvise` settings, because +none of them can alter an Iceberg data-file write request. Spark 4.1 sets `committer.name` and +`committer.magic.enabled` itself when `spark-hadoop-cloud` is on the classpath. Ignored settings are +matched by their plain `fs.s3a.` keys, so a per-bucket `fs.s3a.bucket..` +form still falls back. Unknown explicit +`fs.s3a.*`, `s3.*`, or `client.*` settings therefore fall back at planning time instead of being +silently ignored by the native storage backend. The exception is a vendor-owned `s3.*` / +`client.*` property when +`s3.comet.credential.provider.class` is configured: the provider receives the unfiltered FileIO +bag and can consume that property. Iceberg-defined settings that the native storage path cannot +honour still fall back even with a provider. A per-bucket Hadoop setting counts only for the exact +data-bucket name, so configuration for a longer dotted bucket does not by itself disable the +native write. If Iceberg's AWS property classes cannot be loaded, vendor `s3.*` / `client.*` keys +fall back too and planning still completes. The fall-back reason reports only sorted property +names, never their values, so credentials and tokens do not enter EXPLAIN or plan logs. + +On a format-version 3 table with Iceberg 1.10 or newer, a write that rewrites existing rows falls +back. Copy-on-write `DELETE`, `UPDATE` and `MERGE` and `rewrite_data_files` write the row lineage +columns `_row_id` and `_last_updated_sequence_number` into the new data files, so that rewritten +rows keep their ids, and the native writer does not write those columns. Appends, overwrites, +`CREATE TABLE ... AS SELECT` and `REPLACE TABLE ... AS SELECT` write the data columns only, and +Iceberg assigns the new rows their ids when it commits, so those run natively. Earlier Iceberg +versions write no lineage columns, so the rule never applies to them. + +A format-version-1 table keeps a dropped partition field as a `void` field, and its source column +can be dropped afterwards. A spec that mixes such a field with a live one falls back: iceberg-java +cannot write through it either, so the write fails with iceberg-java's own error rather than in +the native writer ([#6141](https://github.com/apache/datafusion-comet/issues/6141)). A spec whose +fields are all `void` writes unpartitioned and stays eligible. + +Other `write.*` properties are intentionally not gated because they cannot make the native +writer produce different data files: distribution and ordering settings shape the Spark plan +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. Most native eligibility rules are pinned by +`CometIcebergWriteDetectionSuite`. + +Manifest `DataFile` metrics are assembled on the JVM before commit: each written file's +metrics are re-derived from its parquet footer through the version-matched +`ParquetUtil.footerMetrics` and `MetricsConfig.forTable`, with float/double NaN counts and +bounds carried over from the native writer's tracked state. iceberg-java's metadata decisions +— metrics modes, the inferred-column cap +(`write.metadata.metrics.max-inferred-column-defaults`), bound truncation, and list/map bounds +suppression — are therefore applied by iceberg-java's own code regardless of what the native +writer reports. + +iceberg-java's writer takes these metrics from the footer it still holds in memory, but the native +path reads each footer back from storage: two small reads per written file, one for the footer's +length and one for the footer itself, made one file after another before the task finishes. On S3 +or GCS each read is a GET request, so the task waits two request round trips per file. That is +small next to uploading a large file, but a write that produces many small files, such as a fanout +write over many partitions, pays it for every one of them. + +### Failure handling + +Eligibility is decided entirely at plan time. That includes the reflection surface: every +iceberg-java class, method, and constructor the executor-side commit-message assembly uses is +eagerly resolved by the eligibility gate on the driver, so an Iceberg release that moves any +of them declines the native path with a fall-back reason instead of failing tasks mid-write. +Once planned, the physical plan is fixed — there is no per-task re-decision or runtime switch +back to the JVM writer. + +When a native write fails partway through a task (an object-store error, a data-dependent cast +failure), the error propagates as an ordinary Spark task failure and Spark's task retry +re-executes it — through the native writer again. Retries cannot collide: each attempt's task +attempt id is embedded in its data file names. + +The native writer's buffers are charged to Comet's memory pool, the off-heap budget Comet's other +native operators draw on, 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. Each open file holds the row group it is +writing in memory, up to `write.parquet.row-group-size-bytes`, and on S3 or GCS also the last row +group it flushed, which is uploaded once the next one is complete or the file closes. Each +partition also holds the rows that have not reached its file yet. So a task writing to many +partitions needs memory in proportion to them. When the pool cannot grant it, the task writes out +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 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 +open at a time. A larger `spark.memory.offHeap.size` also helps, and Iceberg's `rewrite_data_files` +procedure compacts small files once they are written. + +A write that keeps one file open, unpartitioned or clustered, has no partition to close. When that +file outgrows the pool, the task fails with a `CometNativeException` reading +`Additional allocation failed for IcebergWriteExec` instead of exceeding the executor's memory, and +Spark retries it like any other task failure. A smaller `write.parquet.row-group-size-bytes` or a +larger `spark.memory.offHeap.size` lets such a write fit. + +Partial results are never committed. The commit set is exactly the commit messages returned by +successful tasks — a failed task contributes none — and if the job fails, the driver-side +commit operator aborts without committing anything. A failed task attempt also deletes the +data files it created, as iceberg-java's writer abort does. The native writer records every +location it hands to a file writer, and cleanup has no ownership gap. Native keeps its cleanup +guard armed after yielding the output batch. The JVM reads those locations into a task failure +listener first, then polls the native output to EOF; that EOF is the acknowledgement that lets +native disarm. A failure in this narrow handoff window can therefore trigger best-effort deletion +on both sides, which is harmless. The native guard still covers a failed write, a task torn down +before the write completed — for example because the operator feeding it threw — and a failure +encoding the manifest or building the output batch. The handoff does not depend on decoding the +manifest: the locations are read before the manifest is decoded, so a failure in that decode still +cleans up. Cleanup never masks the original failure; anything it misses is invisible to every +reader, since readers resolve files through committed manifests only, and is reclaimed by Iceberg's +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`. (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 +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. Enabling +`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), 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) + +No Iceberg reader bases a planning or correctness decision on these; they change the bytes of +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` 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`). +- 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 + ([#6114](https://github.com/apache/datafusion-comet/issues/6114)). The native writer decides + once per partition, normally from that partition's first page of rows in the task, and keeps the + decision for every file and row group it writes for the partition; parquet-mr decides again + for every row group. If buffered rows reach `write.parquet.row-group-size-bytes`, the choice + uses the rows collected so far. In a fanout write, all partitions share this threshold, so + each partition still buffering can make its choice before it has a full first page. Later + rows could have changed parquet-mr's decision, so this difference is not limited to columns + close to the size cut-off. Where the page size rather than `write.parquet.page-row-limit` ends a + column's first page, the native page ends at the first row past parquet-mr's size threshold, + while parquet-mr only ends it at its next periodic size check, so a column close to the + cut-off can be decided the other way. Close to the cut-off both encodings take about the same + 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: 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. On Iceberg 1.5.x, + which the Spark 3.4 profile pins, iceberg-java itself spelled `timestamp` and `timestamptz` + directories with `LocalDateTime.toString()` / `OffsetDateTime.toString()` + (`ts=1969-12-31T23:59:58.500Z`) and left the partition field name unescaped; Comet uses the + 1.8+ spelling on every profile. Distinct partition values still get distinct directories in all + cases, and no reader parses these names: files are resolved through committed manifests. +- File rolling lands on the same row grid as iceberg-java but not necessarily on the same row. + Both writers re-check the current file's size against `write.target-file-size-bytes` once + every 1000 rows of that file (iceberg-java's `RollingFileWriter.ROWS_DIVISOR`; Comet hands the + iceberg-rust writer rows in 1000-row units, per partition file, to get the same grid), so each + writer rolls only on a 1000-row boundary of its own file. + The shared grid is all that is shared. What each writer compares against the target differs — + flushed bytes plus parquet-rs's estimate of the open row group, versus parquet-mr's file + position plus its buffered size — and the two use different threshold comparisons. These are + independent size estimates, so nothing bounds how far apart the two writers' roll points are: + they may cross the target several grid steps apart, and the resulting files can differ in row + count by an arbitrary number of 1000-row blocks. Do not rely on file-layout parity between the + two writers; rely only on each file rolling on its own 1000-row boundary. +- A fanout write that the memory pool cannot hold closes partitions before the task ends (see + [Failure handling](#failure-handling)), where iceberg-java's fanout writer keeps every file open + until then. So a partition can have more, smaller files than iceberg-java writes, and a file + closed early ends off the 1000-row grid, as the last file of a task does. A partition closed + before its rows filled a first page makes its dictionary choice from the rows it has. +- A fanout write lists a task's data files in file-path order, where iceberg-java lists them in + its own `StructLikeMap` iteration order. Both are stable across runs, and neither is a + documented ordering, but the manifest entry order becomes the scan-task order and so the row + order of an unordered `SELECT *`. On a format-version 3 table it also decides which row ids the + commit gives each file's rows, so the same rows can get different `_row_id` values from the two + writers. The ids are unique either way, and across tasks iceberg-java's own assignment already + depends on the order in which the tasks finish. Only the sorted order is reproducible on the + native path: the order the native writer closes its files in depends on the order rows arrive in + 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, 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 + default) are not gated: they can only shift compressed bytes, which are already accepted as + divergent. + +#### Manifest metadata visible to later readers + +Manifest metrics drive partition- and file-level pruning for every future reader of the table, +so a divergence here would outlive the write. This class is deliberately kept almost empty: +`DataFile` metrics are not taken from the native writer's manifest but re-derived on the JVM +from each written file's parquet footer through iceberg-java's own `ParquetUtil.footerMetrics` +and `MetricsConfig.forTable` (see above). Metrics modes, lower/upper bound truncation, the +null-count conventions, and list/map bounds suppression are therefore iceberg-java's code +making iceberg-java's decisions. The parity tests write the same rows through both writers and +compare the committed `readable_metrics` (value, null and NaN counts, and lower and upper bounds) +for the column types they cover. Two footer-derived values can still differ from what +iceberg-java's _writer-tracked_ state would have recorded, and both are analyzed safe: + +- Float/double bounds involving zero may differ in sign: parquet-rs normalises footer + statistics to min `-0.0` / max `+0.0` (the parquet-format recommendation), while + iceberg-java's writer-tracked bounds preserve the exact sign it saw. The native path's + manifest bounds inherit the normalised values — a strictly conservative widening that cannot + change pruning decisions. +- On Iceberg 1.9+, manifest `value_counts` / `null_value_counts` for float/double columns + nested under a nullable struct count rows whose parent struct is null (they come from the + parquet footer), while iceberg-java's writer-tracked counts do not. The native counts are the + correct ones. iceberg-java's counts leave out rows whose struct is null. When every non-null + struct in a file holds a value, iceberg-java records a null count of 0, and `IS NULL` prunes a + file that contains matching rows + ([apache/iceberg#17561](https://github.com/apache/iceberg/issues/17561)). The native counts + never prune such a file. `IS NOT NULL` pruning is the same for both. + +All content not listed above — the logical data, statistics values, and manifest metadata — +must match iceberg-java exactly, or the write falls back. The open bugs in which a native write +still differs are listed in +[Known Issues](#writes-that-differ-from-spark). + +## Known Issues + +This section lists open issues in which Comet's Iceberg reads and writes give a different result +from Spark or iceberg-java, or fail where they succeed, without falling back. Each entry links to +its tracking issue. The cases where Comet falls back to Spark are listed under +[Current limitations](#current-limitations) and [When Comet falls back to Spark's write +operator](#when-comet-falls-back-to-sparks-write-operator). Storage-specific issues are listed under +the [S3](datasources.md#s3-limitations) and [Azure](datasources.md#azure-limitations) limitations. + +Which entries apply depends on which native paths are enabled: + +| Native path | Setting | Default | +| -------------- | ---------------------------------------- | ------- | +| Iceberg scan | `spark.comet.scan.icebergNative.enabled` | `true` | +| Iceberg writer | `spark.comet.write.iceberg.enabled` | `false` | + +### Reads that differ from Spark + +- **Migrated tables that use a name mapping.** The `migrate`, `snapshot` and `add_files` + procedures set `schema.name-mapping.default` for files without Iceberg field ids. When a query + also reads a top-level primitive column, the native scan returns NULL for nested fields (struct + fields, list elements and map values). Filters on those fields then select the wrong rows + ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The fix needs an iceberg-rust + upgrade ([#6934](https://github.com/apache/datafusion-comet/issues/6934)). +- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such + a file as NULL, and the native scan returns the values stored in the file + ([#6794](https://github.com/apache/datafusion-comet/issues/6794)). +- **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as + NULL, and two columns that swapped names return each other's values + ([#6822](https://github.com/apache/datafusion-comet/issues/6822)). +- **Reflection failures while the scan is planned.** If Comet fails to read the table's schema or + metadata through reflection, it continues with an empty or partial result instead of falling + back, which can resolve a column to a different field id + ([#5257](https://github.com/apache/datafusion-comet/issues/5257)). + +### Writes that differ from Spark + +The native Iceberg writer is disabled by default. With `spark.comet.write.iceberg.enabled=true`: + +- **Brotli tables.** The native writer commits brotli-compressed files. JVM Parquet readers can + read them only if Hadoop's `BrotliCodec` is on the classpath. iceberg-java fails the write + instead ([#6960](https://github.com/apache/datafusion-comet/issues/6960)). +- **`write.parquet.compression-codec=none`.** The native writer writes the table uncompressed, and + iceberg-java rejects the value. The native writer also trims or lowercases a few other settings + that iceberg-java reads as written + ([#7008](https://github.com/apache/datafusion-comet/issues/7008)). + +The physical differences that every native data file has, and why they are safe, are listed in +[Accepted divergences](#accepted-divergences). + +### Reads that fail where Spark succeeds + +- **A data file that omits an optional nested field.** The native scan fails on a data file that + lacks an optional struct field, which a write with `write.spark.accept-any-schema=true` can + produce without any schema change + ([#7033](https://github.com/apache/datafusion-comet/issues/7033)). +- **`VERSION AS OF` reads of a column whose type was promoted after the snapshot** + ([#6993](https://github.com/apache/datafusion-comet/issues/6993)). +- **An equality delete keyed on a nested field.** Every query the native scan runs on the table + fails with `field not found`, even one that reads the key + ([#6782](https://github.com/apache/datafusion-comet/issues/6782)). +- **Column names that only newer Unicode case data folds together**, such as a pair of Vithkuqi + letters, which Unicode 14 added. Queries on such a table fail with a duplicate-field error + ([#5540](https://github.com/apache/datafusion-comet/issues/5540)). +- **Iceberg tables whose metadata location has no host.** The native Iceberg scan skips the + credential provider configured with `s3.comet.credential.provider.class` + ([#6536](https://github.com/apache/datafusion-comet/issues/6536)). + +### Writes that fail where Spark succeeds + +With `spark.comet.write.iceberg.enabled=true`: + +- **A write that keeps one file open and outgrows the off-heap pool.** An unpartitioned or + clustered write fails its task when its open file outgrows the pool, where iceberg-java writes + it on the JVM heap. [Failure handling](#failure-handling) describes the + error and the settings that let such a write fit. + +### Differences that are not Comet bugs + +- **iceberg-java 1.9 and later: `IS NULL` on a float or double field under a NULL struct.** + iceberg-java leaves the rows where the struct is NULL out of the field's null count, so a filter + such as `WHERE s.x IS NULL` can skip data files that hold matching rows, with or without Comet + ([apache/iceberg#17561](https://github.com/apache/iceberg/issues/17561)). Files written by Comet's + native Iceberg writer are not affected. +- **iceberg-java before 1.12: a NULL struct whose first field is an identity partition source.** + iceberg-java reads the struct as a struct of NULLs, and Comet returns NULL, as iceberg-java 1.12 + does. [apache/iceberg#17320](https://github.com/apache/iceberg/pull/17320) is titled for nested + initial defaults, but it fixes this case too: it keeps one file column in the read, so the + struct's definition level survives when every projected field is a constant, and an identity + partition value is such a constant. diff --git a/docs/source/user-guide/latest/index.rst b/docs/source/user-guide/latest/index.rst index b4cfe077ca8..1217d3ea798 100644 --- a/docs/source/user-guide/latest/index.rst +++ b/docs/source/user-guide/latest/index.rst @@ -28,7 +28,7 @@ types, operators, and expressions that Comet supports, along with a compatibilit known differences from Apache Spark. Operational topics include reading and understanding Comet query plans, tuning, available metrics, -and integration guides for Apache Iceberg, Apache Celeborn, and Kubernetes. Select a topic from +and integration guides for Apache Celeborn and Kubernetes. The data source pages cover Parquet, Apache Iceberg, and Delta Lake. Select a topic from the navigation menu to read more. .. _toc.user-guide-links-$COMET_VERSION: @@ -43,10 +43,20 @@ the navigation menu to read more. .. toctree:: :maxdepth: 1 - :caption: What Comet Supports + :caption: Data Sources :hidden: Supported Data Sources + Parquet + Apache Iceberg + Delta Lake (experimental) + S3 Credential Providers + +.. toctree:: + :maxdepth: 1 + :caption: What Comet Supports + :hidden: + Supported Data Types Supported Operators Supported Expressions @@ -59,8 +69,6 @@ the navigation menu to read more. :hidden: Overview - compatibility/scans - compatibility/parquet-iceberg-known-issues compatibility/floating-point compatibility/regex compatibility/operators @@ -87,7 +95,6 @@ the navigation menu to read more. Memory Shuffle Celeborn Shuffle - Scans Operators Row/Columnar Transitions @@ -96,11 +103,7 @@ the navigation menu to read more. :caption: Integrations :hidden: - Delta Lake (experimental) - Iceberg Guide - Iceberg Writes Celeborn Guide - S3 Credential Providers Kubernetes Guide .. toctree:: diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md index c0fae0e96b0..d48cc116834 100644 --- a/docs/source/user-guide/latest/migration-guide.md +++ b/docs/source/user-guide/latest/migration-guide.md @@ -93,7 +93,7 @@ which stays off by default. With it on, Comet plans an Iceberg `INSERT INTO`, `I 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) +and iceberg-java commits every write. Before turning it on, see [Iceberg Writes](iceberg.md#writes) 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. diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index 566ec058f15..7ac21920b56 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -61,9 +61,9 @@ omitted from the tables below and may be reconsidered based on demand: | Operator | Status | Notes | | ----------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `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. 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)). | +| `FileSourceScanExec` | ✅ | Parquet only. Some types and configurations fall back. See [Parquet](parquet.md#reading). | +| `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](parquet.md#reading) and the [Iceberg Guide](iceberg.md). | +| `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.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). | @@ -137,7 +137,7 @@ combined with `OR`, which Spark plans as an existence `BroadcastNestedLoopJoinEx | ------------------------------------------------------------------------------------------------------------------ | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `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). | +| 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.md#writes). | | `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/docs/source/user-guide/latest/parquet.md b/docs/source/user-guide/latest/parquet.md new file mode 100644 index 00000000000..57945d06518 --- /dev/null +++ b/docs/source/user-guide/latest/parquet.md @@ -0,0 +1,351 @@ + + +# Parquet + +Comet's Parquet scan offloads decoding to native code (Rust, through DataFusion) and produces Arrow +batches for the rest of the plan. Comet falls back to Spark when the scan cannot be converted, for +example because the schema holds an unsupported type or the query uses one of the features listed +below. When the scan falls back to Spark, enabling `spark.comet.convert.parquet.enabled` converts +the data into Arrow format immediately, so the Comet pipeline can take over after the scan, although +the conversion may not be efficient. + +This page covers what the scan supports, where its results can differ from Spark, how to tune it, +and the known issues. Parquet files managed by Apache Iceberg are read by the native Iceberg +reader, described on the [Iceberg](iceberg.md) page. Reading from object stores and HDFS is +described in [Supported Data Sources](datasources.md#supported-storages). + +## Reading + +### Unsupported features: fall back to Spark + +The following features are not supported and cause Comet to fall back to Spark: + +- Decimals encoded in binary format. +- `ShortType` columns, by default. When reading Parquet files written by systems other than Spark that contain + columns with the logical type `UINT_8` (unsigned 8-bit integers), Comet may produce different results than Spark. + Spark maps `UINT_8` to `ShortType`, but Comet's Arrow-based readers respect the unsigned type and read the data as + unsigned rather than signed. Since Comet cannot distinguish `ShortType` columns that came from `UINT_8` versus + signed `INT16`, by default Comet falls back to Spark when scanning Parquet files containing `ShortType` columns. + This behavior can be disabled by setting `spark.comet.scan.unsignedSmallIntSafetyCheck=false`. Note that `ByteType` + columns are always safe because they can only come from signed `INT8`, where truncation preserves the signed value. +- Default values that are nested types (e.g., maps, arrays, structs). Literal default values are supported. +- Spark's Datasource V2 API. When `spark.sql.sources.useV1SourceList` does not include `parquet`, Spark uses the + V2 API for Parquet scans. Comet's Parquet scan only supports the V1 API. +- `_metadata.row_index`, `_metadata.file_block_start` and `_metadata.file_block_length`. When Spark splits a file, + the native scan and Spark can read a row group in different splits, so the block values would differ. Other + `_metadata` columns (`file_path`, `file_name`, `file_size`, `file_modification_time`) are supported. +- No support for `input_file_name()`, `input_file_block_start()`, or `input_file_block_length()` SQL functions. + Comet's Parquet scan does not use Spark's `FileScanRDD`, so these functions cannot populate their values. +- No support for `ignoreMissingFiles` or `ignoreCorruptFiles` being set to `true` +- `spark.sql.parquet.enableVectorizedReader=false`. Disabling the vectorized reader opts into + Spark's parquet-mr semantics (silent overflow, null-on-narrowing), which Comet's native reader + does not replicate. By default Comet falls back to Spark in this case. Set + `spark.comet.scan.allowDisabledParquetVectorizedReader=true` to opt in to running the + Comet Parquet scan regardless. +- A read schema that repeats a Parquet field id, at the top level or within a struct, when + `spark.sql.parquet.fieldId.read.enabled=true`. +- A read schema with sibling struct fields whose names collide case-insensitively, when + `spark.sql.caseSensitive=false`. Spark's analyzer normally rejects such a schema before the scan + is planned. + +Some Parquet reads return results that differ from Spark without falling back to Spark, such as +reads of dates and timestamps written with the legacy hybrid Julian/Gregorian calendar +([#5010](https://github.com/apache/datafusion-comet/issues/5010)). +[Reads that differ from Spark](#reads-that-differ-from-spark) lists them, with the +issues that track them. + +### Unsupported features: error at scan time + +The following limitations raise an error at scan time rather than falling back to Spark. More +reads that fail where Spark succeeds are listed in +[Reads that fail where Spark succeeds](#reads-that-fail-where-spark-succeeds): + +- Selecting a field by name when multiple physical siblings match, including inside structs, + arrays, and maps. Comet raises a duplicate-field error instead of resolving the collision. + Checks cover referenced columns, including predicates; unselected roots do not prevent + reading a unique field by name or field ID. Exact-name projections of unique children in + structs and arrays of structs remain supported. Casts that cannot use this pruning reject + byte-identical duplicate siblings anywhere in the decoded physical subtree, including maps. + Field-ID resolution retains precedence, but selecting a byte-identically duplicated physical + root name still raises a duplicate-field error, even when the requested field is renamed. + Names in separate groups do not collide. Spark may read a duplicate-bearing file with an + explicit schema in case-sensitive mode, but its choice of sibling depends on the field shape + and can produce unexpected values. Spark rejects schema inference from a single file with + duplicate names; inference across files can depend on merge order. + Resolution is tracked in [#5884](https://github.com/apache/datafusion-comet/issues/5884), + with mixed-type behavior in [#5964](https://github.com/apache/datafusion-comet/issues/5964). +- Invalid UTF-8 bytes in `STRING` columns. Spark permits arbitrary byte sequences in a `STRING` + column (for example from `CAST(X'C1' AS STRING)`), but Comet's native execution path is built on + Arrow, whose string type is strictly UTF-8. Reading a Parquet file whose `STRING` column contains + non-UTF-8 bytes fails with `Parquet error: encountered non UTF-8 data`. Disable Comet for the + query, or cast the column to `BINARY` before persisting, if you need to preserve non-UTF-8 bytes. + By contrast, Comet decodes non-UTF-8 bytes at the JVM-to-native Arrow FFI boundaries using + JVM-compatible replacement semantics. This covers native query input from the JVM-exported Arrow + stream (including JVM scans, shuffle reads, and `mapInArrow`), columnar-to-row conversion, and JVM + UDF results. It does not change the native Parquet reader: rejecting invalid UTF-8 there remains + Gap A of [#4764](https://github.com/apache/datafusion-comet/issues/4764). See + [Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes), + [#4121](https://github.com/apache/datafusion-comet/issues/4121), and the tracking issue above. +- Reading `TimestampLTZ` as `TimestampNTZ` on Spark 3.x. Spark raises an error per + [SPARK-36182](https://issues.apache.org/jira/browse/SPARK-36182) because LTZ encodes UTC-adjusted + instants that cannot be safely reinterpreted as timezone-free values, and Comet matches this by + rejecting the read. This applies to all LTZ physical encodings (INT96, TIMESTAMP_MICROS, + TIMESTAMP_MILLIS). On Spark 4.0+, this read is permitted + ([SPARK-47447](https://issues.apache.org/jira/browse/SPARK-47447)) and Comet matches Spark's + behavior. + +### Schema Mismatch Handling + +The issues in this subsection apply only when the requested read schema differs from the schema written +to the Parquet file. They do **not** affect a plain `spark.read.parquet(path)` that infers the schema +from file metadata, because in that case the requested schema and file schema match by construction. +Schema mismatch happens in two real-world scenarios: + +1. The user provides an explicit read schema: `spark.read.schema().parquet(path)` (or the + equivalent DataFrame API). +2. **Schema evolution / partitioned reads** where files in a single dataset were written at different + times with different types, or a table-format catalog (Iceberg, Delta) records a logical schema + that has evolved past one or more underlying Parquet files. Spark coerces the file types to the + table types at read time. + +Spark's vectorized Parquet reader fully validates these conversions in `ParquetVectorUpdaterFactory.getUpdater` +and throws `SchemaColumnConvertNotSupportedException` for unsupported pairs. Comet's Parquet scan +mirrors that validation in its schema adapter; the entries below are the remaining gaps. + +Note that the exact set of accepted conversions has changed between Spark versions +(for example, Spark 3.x's `schemaEvolution.enabled` flag gates `INT32 → INT64`, `FLOAT → DOUBLE`, +and `INT32 → DOUBLE` widening that Spark 4.0+ accepts unconditionally; `TimestampLTZ → TimestampNTZ` +is rejected by Spark 3.x but accepted by Spark 4.0+). Comet aims to follow the per-version Spark +behavior. + +- **Conversion errors precede pushed row filters**. Empty files and files whose row groups or + pages are all pruned read without decode-time conversion errors. When row-filter pushdown is + enabled, Comet checks rejected conversions before reading surviving data pages, so a row filter + cannot suppress the error by discarding every row. Legacy LIST shape mismatches that Spark cannot + clip are rejected when the file is opened, including for empty and fully pruned files. +- **List conversion error paths assume Spark's standard encoding**. Comet inserts `list` + before the element name when reporting a rejected array element conversion. Arrow's schema + omits the repeated group name, so paths for legacy LIST encodings or custom group names may + differ from Spark's Parquet column path. +- **`ParquetSchemaConvert` errors do not include the file path**. The mismatch itself is detected and + rejected correctly, but the resulting Spark error message reads + `Encountered error while reading file . Data type mismatches…` (note the empty path). Behavior is + consistent across Spark versions. See + [#4316](https://github.com/apache/datafusion-comet/issues/4316). +- **Spark 3.x: extra `SparkException` layer in the cause chain**. The native error is translated to a + `SparkException` whose cause is `SchemaColumnConvertNotSupportedException` (matching what Spark would + throw); on Spark 3.x the executor / task error handling re-wraps this once more on the way back to + the driver, producing a two-level chain (`SparkException → SparkException → +SchemaColumnConvertNotSupportedException`) instead of the one-level chain Spark's own vectorized + reader produces. Code that catches `SparkException` and inspects only the immediate cause via + `e.getCause.isInstanceOf[SchemaColumnConvertNotSupportedException]` will see the inner + `SparkException` instead. Walk the cause chain to recover the + `SchemaColumnConvertNotSupportedException`. Spark 4.0+ produces a single-level chain, matching + vanilla Spark. See [#4354](https://github.com/apache/datafusion-comet/issues/4354). + +## Writing + +Comet can also write Parquet files natively. The native Parquet writer is for testing only and +should not be used in production. It is disabled by default and enabled with +`spark.comet.write.parquet.enabled=true`. It handles only non-partitioned, non-bucketed writes; +partitioned and bucketed writes fall back to Spark, as do writes with +`spark.sql.files.maxRecordsPerFile` set on Spark 4.0 and later. Parts of Spark's Parquet write +behavior are not reproduced, as listed under [Known Issues](#known-issues). Iceberg writes use a +different native writer, described on the [Iceberg](iceberg.md#writes) page. + +## Tuning + +### Filter Pushdown / Late Materialization + +Setting `spark.comet.parquet.rowFilterPushdown.enabled=true` pushes filter evaluation into the Parquet +decode step and lazily materializes projected columns for surviving rows. This can significantly reduce +CPU and memory when the filter is highly selective on a small subset of columns. It is disabled by default +because it can hurt when the filter is not selective or when most columns must be read anyway. Row-group, +page-index, and bloom-filter pruning happen regardless of this flag whenever Spark's +`spark.sql.parquet.filterPushdown` is on, except for filters that compare `FLOAT` or `DOUBLE` values, +which do not prune while this flag is on (see [Floating-point Number Comparison](compatibility/floating-point.md)). + +### Row Group Assignment to Splits + +Spark and DataFusion's native Parquet scans use different rules to decide which row groups belong to a +given scan range (split). Spark assigns a row group to a split if the row group's start offset falls +within `[split.start, split.start + split.length)`, guaranteeing that every task Spark plans reads at +least one row group when the file layout permits. DataFusion's `prune_by_range` also checks whether a +row group's start offset falls within the split's byte range, but because row group sizes are not aligned +with Spark's split boundaries, the two systems can disagree on which split "owns" a given row group. + +When a file contains row groups whose sizes are close to `spark.sql.files.maxPartitionBytes`, this +mismatch can leave some Comet scan tasks with no row groups to read. Those tasks still load Parquet +metadata but return zero rows, while neighboring tasks end up reading more row groups than Spark +intended. The overall effect is that Comet uses only a fraction of the parallelism that Spark planned +for the scan stage, and end-to-end scan latency increases even though the total amount of data read +is unchanged. + +Symptoms to look for: + +- A subset of scan tasks completes almost immediately and reports 0 input rows, while the remaining + tasks read noticeably more rows than the equivalent Spark tasks would. +- The Comet scan stage has the same number of planned tasks as Spark but a much lower count of tasks + that actually do work. + +Workaround: lower `spark.sql.files.maxPartitionBytes` so that each split is smaller than a single row +group. For example, if the file's row groups are around 120 MB and `spark.sql.files.maxPartitionBytes` +is left at the 128 MB default, most splits will contain at most one row group boundary and the +mismatch is amplified; setting `spark.sql.files.maxPartitionBytes` below 120 MB (for example, 64 MB) +distributes row groups across more splits and reduces the number of idle tasks. Smaller values produce +more splits overall, so some idle tasks may remain — tune the value against your file layout. + +See [#3817](https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4193279630) for a +worked example and further discussion. + +## Known Issues + +This section lists open issues in which Comet's Parquet reads and writes give a different result +from Spark, or fail where Spark succeeds, without falling back. Each entry links to its tracking +issue. The cases where Comet falls back to Spark are listed under [Reading](#reading). Iceberg +tables have their own list under the Iceberg page's [Known Issues](iceberg.md#known-issues). +Storage-specific issues are listed under the [S3](datasources.md#s3-limitations) and +[Azure](datasources.md#azure-limitations) limitations. + +Which entries apply depends on which native paths are enabled: + +| Native path | Setting | Default | +| -------------- | ----------------------------------- | ---------------------- | +| Parquet scan | `spark.comet.scan.enabled` | `true` | +| Parquet writer | `spark.comet.write.parquet.enabled` | `false` (testing only) | + +`spark.comet.scan.enabled` is a testing setting that Comet's own test suites use. It is not meant +for production, so do not turn it off to avoid an entry on this page. + +### Reads that differ from Spark + +- **Dates and timestamps written with the legacy calendar.** Spark before 3.0, and later versions + with `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`, write these values in the hybrid + Julian/Gregorian calendar. The native Parquet scan reads them as proleptic Gregorian, so values + before October 15, 1582 are wrong in projections and predicates. It ignores + `spark.sql.parquet.datetimeRebaseModeInRead` and the file's `org.apache.spark.legacyDateTime` + metadata ([#5010](https://github.com/apache/datafusion-comet/issues/5010)). The + [Delta scan](delta.md) rebases. +- **INT96 timestamps inside legacy-format lists.** INT96 timestamps outside the years 1677 to 2262 + read back wrong inside lists written in a legacy layout, as Hive and + `spark.sql.parquet.writeLegacyFormat=true` write them + ([#7029](https://github.com/apache/datafusion-comet/issues/7029)). +- **A struct or array with none of its requested fields in the file.** Spark returns NULL for the + column. Comet returns a non-null struct or array that holds NULLs. This applies to Spark 3.4, 3.5 + and 4.0, and to Spark 4.1 and later with + `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true` + ([#7030](https://github.com/apache/datafusion-comet/issues/7030)). +- **Nested BINARY values that are not valid UTF-8, read as STRING.** This applies to a BINARY + column without the UTF8 annotation that is read as `string`, through the read schema or + `spark.sql.parquet.binaryAsString=true`. Spark returns the bytes. Comet returns NULL for each + such nested value. Top-level values match Spark in a release build, because the invalid bytes + reach an Arrow string array unchecked. A debug build panics on them + ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of + [#4764](https://github.com/apache/datafusion-comet/issues/4764)). +- **Spark 3.4: a struct matched by field id under a different name.** With + `spark.sql.parquet.fieldId.read.enabled=true`, a struct column that the read schema matches to + the file by field id but under another name reads as NULL when Spark prunes its nested fields + ([#7031](https://github.com/apache/datafusion-comet/issues/7031)). +- **Field-id reads where one file column matches a field by id and a sibling by name.** With + `spark.sql.parquet.fieldId.read.enabled=true`, Comet returns NULL for the sibling at the top + level. Inside a struct, it returns the values of the field matched by id. Spark's own readers + disagree here ([#6994](https://github.com/apache/datafusion-comet/issues/6994)). +- **`TIMESTAMP_MILLIS` values that overflow when converted to microseconds.** In nested fields, and + in every column of a scan that has data filters, Comet returns NULL where Spark fails with + `ArithmeticException: long overflow` + ([#5553](https://github.com/apache/datafusion-comet/issues/5553), + [#5739](https://github.com/apache/datafusion-comet/issues/5739)). +- **Reads that Spark rejects.** Comet returns rows where Spark fails the read for + `TIMESTAMP(NANOS)` columns read as `timestamp` or `timestamp_ntz` + ([#7034](https://github.com/apache/datafusion-comet/issues/7034)), `INT` columns read as + `INTERVAL DAY TO SECOND` ([#5188](https://github.com/apache/datafusion-comet/issues/5188)), and + Variant-annotated fields read as a plain struct + ([#5741](https://github.com/apache/datafusion-comet/issues/5741)). + +### Writes that differ from Spark + +With `spark.comet.write.parquet.enabled=true`: + +- **The datetime rebase modes are ignored.** On Spark 3.4 and 3.5, whose default rebase mode is + `EXCEPTION`, the native writer writes dates and timestamps before October 15, 1582 that Spark + refuses to write, and Spark then fails to read them back under its default read mode. With + `spark.sql.parquet.datetimeRebaseModeInRead=LEGACY`, Spark reads them shifted by several days, on + every version. The files lack the `org.apache.spark.version` footer key that tells Spark they need + no rebasing ([#7047](https://github.com/apache/datafusion-comet/issues/7047), + [#3427](https://github.com/apache/datafusion-comet/issues/3427)). +- **`spark.sql.parquet.outputTimestampType` and `spark.sql.parquet.writeLegacyFormat` are + ignored.** Timestamps are always written as `INT64` microseconds, never as Spark's default + `INT96`, so with `TIMESTAMP_MILLIS` they keep the sub-millisecond digits that Spark drops. Spark's + schema footer key is not written either, so a path-based read loses column metadata such as + comments, and with `spark.sql.parquet.inferTimestampNTZ.enabled=false` it reads `TIMESTAMP_NTZ` + columns as `TIMESTAMP` ([#7048](https://github.com/apache/datafusion-comet/issues/7048), + [#3425](https://github.com/apache/datafusion-comet/issues/3425)). +- **Spark 3.4 and 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs + after a write: file index refresh, cache refresh, partition refresh and statistics update. Later + queries can return the table's previous contents until `REFRESH TABLE` + ([#6315](https://github.com/apache/datafusion-comet/issues/6315)). +- **Spark's Parquet writer properties are ignored.** `parquet.block.size`, page size, dictionary, + writer version and statistics settings never reach the native writer + ([#5304](https://github.com/apache/datafusion-comet/issues/5304)). +- **An input with no partitions writes no file.** Spark writes one file that holds the schema + ([#5303](https://github.com/apache/datafusion-comet/issues/5303)). +- **Spark SQL tests that still fail with the native writer**, not yet classified: + - CTAS schema ([#3418](https://github.com/apache/datafusion-comet/issues/3418)) + - CTAS with `CHAR_AS_VARCHAR` ([#3419](https://github.com/apache/datafusion-comet/issues/3419)) + - CTAS with a union ([#3429](https://github.com/apache/datafusion-comet/issues/3429)) + - `INSERT INTO` with default columns ([#3421](https://github.com/apache/datafusion-comet/issues/3421)) + - `INSERT INTO` after `ALTER TABLE ADD COLUMNS` ([#3422](https://github.com/apache/datafusion-comet/issues/3422)) + - `INSERT INTO` from non-projection queries ([#3423](https://github.com/apache/datafusion-comet/issues/3423)) + - the `INSERT OVERWRITE` output schema ([#3424](https://github.com/apache/datafusion-comet/issues/3424)) + - multiple inserts ([#3430](https://github.com/apache/datafusion-comet/issues/3430)) + - `INSERT OVERWRITE LOCAL DIRECTORY` ([#3420](https://github.com/apache/datafusion-comet/issues/3420)) + - empty files ([#3417](https://github.com/apache/datafusion-comet/issues/3417)) + - `onTaskCommit` ([#3428](https://github.com/apache/datafusion-comet/issues/3428)) + +### Reads that fail where Spark succeeds + +- **STRING columns that hold invalid UTF-8**, and **duplicate field names**. See + [Unsupported features: error at scan time](#unsupported-features-error-at-scan-time). +- **A duplicate field id in a column the query does not read.** With a valid read schema and + `spark.sql.parquet.fieldId.read.enabled=true`, a file in which two root columns carry the same + field id fails the scan whenever the table schema has a field with that id, even when the query + does not read that field ([#7006](https://github.com/apache/datafusion-comet/issues/7006)). +- **A nested encoding that arrow-rs rejects.** A file fails to read when any column in its schema + uses such an encoding, even a column the query does not read + ([#7035](https://github.com/apache/datafusion-comet/issues/7035)). +- **A `DATE` column read as `INTERVAL YEAR TO MONTH`.** Spark reads it, and Comet fails the read + ([#5188](https://github.com/apache/datafusion-comet/issues/5188)). + +### Writes that fail where Spark succeeds + +With `spark.comet.write.parquet.enabled=true`: + +- **Row groups that outgrow the memory pool.** The native writer cuts row groups by row count + only, where Spark uses `parquet.block.size`, and it cannot spill. With wide or nested rows, the + open row group can outgrow the off-heap pool and fail the task. The audit did not reproduce a + failure, because a 239 MB row group fit in a 256 MB pool + ([#5304](https://github.com/apache/datafusion-comet/issues/5304)). + +### Differences that are not Comet bugs + +- **Field-id reads with a read schema that swaps two nested field names.** With + `spark.sql.parquet.fieldId.read.enabled=true`, Spark's vectorized reader pairs the file's columns + with the read fields by the file's names, so it fails with a type mismatch or returns no rows. + Comet resolves the fields by id. diff --git a/docs/source/user-guide/latest/s3-credential-providers.md b/docs/source/user-guide/latest/s3-credential-providers.md index 4a35f477ed2..c32dcd75d04 100644 --- a/docs/source/user-guide/latest/s3-credential-providers.md +++ b/docs/source/user-guide/latest/s3-credential-providers.md @@ -388,7 +388,7 @@ public final class IcebergRESTVendedS3Provider implements CometS3CredentialProvi | Value | Used for | | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `READ` | All native scan paths (raw Parquet, Iceberg). | -| `WRITE` | The native Iceberg writer (see [Iceberg Writes](iceberg-writes.md)). If the configured provider fails to initialize, the write fails rather than falling back to the default credential chain. | +| `WRITE` | The native Iceberg writer (see [Iceberg Writes](iceberg.md#writes)). If the configured provider fails to initialize, the write fails rather than falling back to the default credential chain. | A `WRITE` credential is not implicitly read-capable. Vendors that need read-during-write workflows include the required read permissions in the IAM policy attached to their `WRITE` credentials. diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index ede8eecb983..99bb05d0d20 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/docs/source/user-guide/latest/tuning.md @@ -29,8 +29,10 @@ guide is split into the following pages: implementations, and shuffle compression. - [Remote Shuffle with Celeborn](tuning/celeborn.md): bounding frame sizes and memory admission for native remote shuffle. -- [Scan Tuning](tuning/scans.md): Parquet filter pushdown, Parquet split sizing, and Iceberg data - file concurrency. +- [Parquet Scan Tuning](parquet.md#tuning): row filter pushdown and late materialization, and + sizing splits so that row groups spread evenly across scan tasks. +- [Iceberg Scan Tuning](iceberg.md#tuning): reading data files in parallel with + `spark.comet.scan.icebergNative.dataFileConcurrencyLimit`. - [Operator Tuning](tuning/operators.md): joins, adaptive partial aggregation, and sorting on floating-point values. - [Reducing Row/Columnar Conversion Overhead](tuning/transitions.md): stages in which many diff --git a/docs/source/user-guide/latest/tuning/memory.md b/docs/source/user-guide/latest/tuning/memory.md index 51fb9754b83..828dcc95272 100644 --- a/docs/source/user-guide/latest/tuning/memory.md +++ b/docs/source/user-guide/latest/tuning/memory.md @@ -40,7 +40,7 @@ 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. 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)). +[Iceberg Writes](../iceberg.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, the diff --git a/docs/source/user-guide/latest/tuning/scans.md b/docs/source/user-guide/latest/tuning/scans.md deleted file mode 100644 index 40bc82634ec..00000000000 --- a/docs/source/user-guide/latest/tuning/scans.md +++ /dev/null @@ -1,72 +0,0 @@ - - -# Scan Tuning - -## Parquet Reader Tuning - -### Filter Pushdown / Late Materialization - -Setting `spark.comet.parquet.rowFilterPushdown.enabled=true` pushes filter evaluation into the Parquet -decode step and lazily materializes projected columns for surviving rows. This can significantly reduce -CPU and memory when the filter is highly selective on a small subset of columns. It is disabled by default -because it can hurt when the filter is not selective or when most columns must be read anyway. Row-group, -page-index, and bloom-filter pruning happen regardless of this flag whenever Spark's -`spark.sql.parquet.filterPushdown` is on, except for filters that compare `FLOAT` or `DOUBLE` values, -which do not prune while this flag is on (see [Floating-point Number Comparison](../compatibility/floating-point.md)). - -### Parquet Native Scans - -Spark and DataFusion's native Parquet scans use different rules to decide which row groups belong to a -given scan range (split). Spark assigns a row group to a split if the row group's start offset falls -within `[split.start, split.start + split.length)`, guaranteeing that every task Spark plans reads at -least one row group when the file layout permits. DataFusion's `prune_by_range` also checks whether a -row group's start offset falls within the split's byte range, but because row group sizes are not aligned -with Spark's split boundaries, the two systems can disagree on which split "owns" a given row group. - -When a file contains row groups whose sizes are close to `spark.sql.files.maxPartitionBytes`, this -mismatch can leave some Comet scan tasks with no row groups to read. Those tasks still load Parquet -metadata but return zero rows, while neighboring tasks end up reading more row groups than Spark -intended. The overall effect is that Comet uses only a fraction of the parallelism that Spark planned -for the scan stage, and end-to-end scan latency increases even though the total amount of data read -is unchanged. - -Symptoms to look for: - -- A subset of scan tasks completes almost immediately and reports 0 input rows, while the remaining - tasks read noticeably more rows than the equivalent Spark tasks would. -- The Comet scan stage has the same number of planned tasks as Spark but a much lower count of tasks - that actually do work. - -Workaround: lower `spark.sql.files.maxPartitionBytes` so that each split is smaller than a single row -group. For example, if the file's row groups are around 120 MB and `spark.sql.files.maxPartitionBytes` -is left at the 128 MB default, most splits will contain at most one row group boundary and the -mismatch is amplified; setting `spark.sql.files.maxPartitionBytes` below 120 MB (for example, 64 MB) -distributes row groups across more splits and reduces the number of idle tasks. Smaller values produce -more splits overall, so some idle tasks may remain — tune the value against your file layout. - -See [#3817](https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4193279630) for a -worked example and further discussion. - -## Iceberg Scan Tuning - -Comet's native Iceberg scan (`spark.comet.scan.icebergNative.enabled`, enabled by default) reads each -task's data files one at a time by default. For tables with many small files or high-latency storage, -increase `spark.comet.scan.icebergNative.dataFileConcurrencyLimit` (default `1`; values of 2–8 are -suggested) to overlap I/O across files at the cost of extra memory. diff --git a/docs/source/user-guide/latest/understanding-comet-plans.md b/docs/source/user-guide/latest/understanding-comet-plans.md index ccaa8d27740..7789e8eacc2 100644 --- a/docs/source/user-guide/latest/understanding-comet-plans.md +++ b/docs/source/user-guide/latest/understanding-comet-plans.md @@ -345,7 +345,7 @@ These keep their data on the JVM but participate in the Comet pipeline. | `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-writes.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