feat: implement native Iceberg V2 writer via iceberg-rust

Author: Jordan Epstein

.github/workflows/pr_build_linux.yml

@@ -343,6 +343,7 @@ jobs:
               org.apache.comet.CometIcebergWriteActionSuite
               org.apache.comet.CometIcebergWriteDetectionSuite
               org.apache.comet.iceberg.IcebergReflectionSuite
+              org.apache.comet.serde.operator.IcebergWriteProtoTranslationSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
               org.apache.comet.CometFuzzTestSuite
               org.apache.comet.CometFuzzIcebergSuite

.github/workflows/pr_build_macos.yml

@@ -183,6 +183,7 @@ jobs:
               org.apache.comet.CometIcebergWriteActionSuite
               org.apache.comet.CometIcebergWriteDetectionSuite
               org.apache.comet.iceberg.IcebergReflectionSuite
+              org.apache.comet.serde.operator.IcebergWriteProtoTranslationSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
               org.apache.comet.CometFuzzTestSuite
               org.apache.comet.CometFuzzIcebergSuite

docs/source/user-guide/latest/iceberg-writes.md

@@ -27,16 +27,14 @@ Comet accelerates Iceberg V2 copy-on-write table writes in two complementary lay
    `V2ExistingTableWriteExec` command into a pair of operators — a **committer** and a
    **writer** — so that AQE (and Comet's columnar rules) can see and re-optimise the
    data sub-query. The actual file-writing work still runs through Iceberg's stock JVM writer.
-2. **Native parquet write — detection only at this stage** (further toggle). Once the
-   split-operator plan is in place, a serde inspects each `IcebergWriteExec` to decide
-   whether the table's properties and write shape would allow delegating the per-task parquet
-   write to [iceberg-rust](https://github.com/apache/iceberg-rust). The detection / fall-back
-   table is wired up and tested; the actual native exec lands in a follow-up commit. With the
-   gate on today, writes still go through the JVM two-op path — only the diagnostic surface is
-   live.
+2. **Native parquet write** (further toggle). Once the split-operator plan is in place, the
+   writer can be swapped for a native variant that delegates the per-task parquet write to
+   [iceberg-rust](https://github.com/apache/iceberg-rust). The committer / table-metadata
+   update / cache-refresh stay on the JVM exactly as before.
 
-Both layers fall back to plain Iceberg-Java when they can't safely apply, and both layers are
-configured per-session.
+Both layers fall back to plain Iceberg-Java when they can't safely apply — and both layers are
+configured per-session, so you can opt into the split plan but skip the native write while it
+burns in.
 
 ## What changes about the Iceberg plan
 
@@ -89,7 +87,7 @@ spark.sql.catalog.<name>.warehouse=...
 
 # Comet write toggles (both off by default; can be turned on independently)
 spark.comet.write.iceberg.splitOperator.enabled=true              # layer 1: split-operator plan
-spark.comet.write.iceberg.nativeAcceleration=true                 # layer 2: native parquet write (detection only at this stage)
+spark.comet.write.iceberg.nativeAcceleration=true                 # layer 2: native parquet write
 ```
 
 `IcebergSparkSessionExtensions` is **mandatory on Spark 3.4** for `UPDATE` / `MERGE` on V2 tables —
@@ -119,6 +117,21 @@ realisable benefit.
 
 - [¹] Requires `IcebergSparkSessionExtensions` (see Configuration above).
 
+Native parquet write coverage is more selective:
+
+| Write                                        | Spark 3.4   | Spark 3.5 | Spark 4.0 |
+|----------------------------------------------|-------------|-----------|-----------|
+| `AppendData` (`INSERT INTO`)                 | native      | native    | native    |
+| `OverwriteByExpression` (static)             | native      | native    | native    |
+| `OverwritePartitionsDynamic`                 | native      | native    | native    |
+| `ReplaceData` (CoW `DELETE` / `UPDATE`)      | native      | native    | native    |
+| `ReplaceData` (CoW `MERGE`)                  | fallback[²] | fallback[²] | fallback[²] |
+
+- [²] MERGE flows through Spark's (or Iceberg 1.5.2's) `MergeRowsExec`, which per-row dispatches
+  the matched / not-matched / not-matched-by-source clauses. There's no native equivalent in
+  Comet *yet* — this is planned (see Known limitations). The scan / join / sort / shuffle parts
+  still run native; only the per-row merge dispatch and the file write stay JVM.
+
 ## Fallback triggers
 
 Before accepting a write into the native path, the serde checks for things iceberg-rust either
@@ -144,25 +157,61 @@ Parquet Rust encoder that iceberg-rust drives).
 | `write.parquet.bloom-filter-enabled.column.<col>` | any column set to `true` | **iceberg-rust**: bloom-filter sizing diverges from iceberg-java (same root cause as above — no cap clamping) |
 | `write.parquet.row-group-check-min-record-count` | set to a non-default value (default `100`) | **parquet-rs**: row-group close cadence is byte-driven and doesn't expose parquet-mr's per-row-count knobs, so iceberg-rust can't honour this property |
 | `write.parquet.row-group-check-max-record-count` | set to a non-default value (default `10000`) | **parquet-rs**: same as above |
-| `write.parquet.page-version` | set to a non-default value (default `v1`) | **parquet-rs**: the default writer does not implement DataPageV2 encoding; iceberg-java with `v2` produces a different on-disk format. |
-| `write.parquet.stats-enabled.column.<col>` | any column override set | **parquet-rs**: no per-column statistics-enabled override on `WriterProperties` matching parquet-mr's; iceberg-rust would silently ignore the request. |
+| `write.parquet.stats-enabled.column.<col>` | any column override set **and** the running Iceberg version defines `TableProperties.PARQUET_COLUMN_STATS_ENABLED_PREFIX` (1.10.0+) | **parquet-rs**: no per-column statistics-enabled override matching parquet-mr's. The property was only added in Iceberg 1.10.0; on 1.5.2 / 1.8.1 Iceberg-Java ignores it too, so there's nothing to diverge from and the write is **not** gated on those versions. |
 | `parquet.enable.dictionary` | set | **parquet-rs**: explicit dictionary override is not translated to the native writer settings; parquet-rs would fall back to its own default and diverge from parquet-mr. |
-| `write.metadata.metrics.max-inferred-column-defaults` | set to less than the number of fields in the schema (default `100`) | **iceberg-rust**: doesn't apply `MetricsModes.None` to columns past this limit; iceberg-java does, so the produced column statistics would diverge |
+| `write.metadata.metrics.max-inferred-column-defaults` | the number of projected field IDs exceeds this limit (default `100`) **and** no explicit `write.metadata.metrics.default` is set | **iceberg-rust**: doesn't apply `MetricsModes.None` to columns past this limit. Iceberg only applies that inferred-column truncation when no default mode is configured (a user-set default applies to all columns regardless of count), so the gate is skipped once a default is present. |
+
+> `write.parquet.page-version` is intentionally **not** a fallback trigger: no Iceberg version Comet
+> targets (1.5.2 / 1.8.1 / 1.10.0) maps a table property to the Parquet writer version — it is
+> hardwired to `PARQUET_1_0`. Setting it is a no-op in Iceberg-Java exactly as in the native path.
 | `io-impl` (catalog property) | set | **iceberg-rust**: native FileIO is selected by URI scheme (`OpenDalStorageFactory`), so an explicit Java `FileIO` class can't be honoured. |
 | Resolved data-location URI scheme | not in `{file, memory, s3, s3a, gs, oss}` | **iceberg-rust**: `iceberg_common::storage_factory_for` resolves only the listed schemes; `hdfs`, `abfs`/`abfss`, `wasb`/`wasbs`, and similar would crash at write time. |
 
 Every trigger has a pinning test in `CometIcebergWriteDetectionSuite`.
 
+## Known limitations and planned work
+
+- **CoW MERGE — native acceleration planned.** The split-operator plan already applies (scan,
+  join, sort, shuffle run native); only the per-row merge dispatch (`MergeRowsExec`) and the
+  file write itself stay JVM. The path forward is well-understood: add a `CometMergeRowsExec`
+  that mirrors Spark's instruction-driven dispatch model (`Keep` / `Discard` / `Split` over
+  matched / not-matched / not-matched-by-source clauses, plus a bitmap-backed cardinality
+  check). Tracked.
+
+- **`sort_order_id` is always stamped `0` (unsorted).** This matches Iceberg-Java for every
+  version Comet targets: the Spark `SparkWrite$WriterFactory` in Iceberg 1.5.2 / 1.8.1 / 1.10.0
+  builds `SparkFileWriterFactory` **without** `.dataSortOrder(...)`, so a stock batch append stamps
+  `sort_order_id = 0` on committed data files even for a table that has a non-default sort order.
+  The native writer does the same. (Iceberg 1.11+ adds
+  `SparkWriteConf.outputSortOrderId(writeRequirements)` and wires the resolved id into the writer
+  factory; Comet reflects that resolver when present, so behaviour stays correct if the pinned
+  runtime is bumped.)
+
+- **Nested (list/map) column bounds differ from Iceberg-Java.** iceberg-java's
+  `ParquetUtil.shouldStoreBounds` suppresses `lower_bounds` / `upper_bounds` for any field reached
+  through a repeated (list or map) field, so collection-element primitives get no manifest bounds.
+  iceberg-rust's `MinMaxColAggregator` computes bounds for every leaf primitive — including
+  list-element / map-key / map-value fields — so a native write to a schema containing list or map
+  columns produces **extra** `lower_bounds` / `upper_bounds` entries for those nested field IDs that
+  Iceberg-Java would omit. `column_sizes` / `value_counts` / `null_value_counts` match. The extra
+  bounds are not used by Iceberg's metrics evaluators (predicates can't address list/map elements),
+  so the functional risk is low, but the manifest is not byte-identical to the JVM path for nested
+  schemas. Strict parity would require stripping those nested-field bounds before manifest encoding
+  (or in iceberg-rust); not yet done.
+
 ## Tests
 
 - **`CometIcebergWriteActionSuite`** — end-to-end scenarios against a temporary Hadoop catalog
-  covering the copy-on-write V2 logical-write variants on the split-operator path, plus parity
-  vs Iceberg-Java's writer, the disabled-config fallback, and the commit-once invariant. Passes
-  on Spark 3.4 / 3.5 / 4.0 with `IcebergSparkSessionExtensions` loaded.
+  covering the copy-on-write V2 logical-write variants on both the split-operator-only and
+  full-native paths, plus parity vs Iceberg-Java, the disabled-config fallback, the
+  commit-once invariant, and the documented native-engagement-vs-fallback contract for MERGE.
+  Passes on Spark 3.4 / 3.5 / 4.0 with `IcebergSparkSessionExtensions` loaded.
 - **`CometIcebergWriteDetectionSuite`** — tests that build a planned-but-not-executed
   `IcebergWriteExec` per fall-back trigger and assert
   `CometIcebergNativeWrite.getSupportLevel` returns `Unsupported` with the expected reason
   fragment, plus a positive baseline (clean parquet V2 table → `Compatible`).
+- **`IcebergWriteProtoTranslationSuite`** — unit tests pinning the JVM-side property → proto
+  translation table-by-table.
 
 ## Abort behaviour
 
@@ -206,6 +255,66 @@ Two design choices keep this stable:
    serialisable-isolation validation to walk the right snapshot range. The strategy calls
    `toBatch` once and threads the result through both execs.
 
+## Other AQE / planning interactions
+
+- **AQE re-fire produces a fresh `IcebergWriteExec` over the existing `CometIcebergWriteExec`.**
+  Without protection this would stack two writers. `CometExecRule`'s `IcebergWriteExec` case
+  detects this (via `unwrapToCometIcebergWrite`) and returns the existing writer instead —
+  mirroring the `DataWritingCommandExec(WriteFilesExec(CometNativeWriteExec))` unwrap V1
+  parquet uses.
+- **`CometIcebergWriteExec` is a block boundary in the serialisation loop.** The post-convert
+  serialization loop in `CometExecRule` resets `firstNativeOp = true` after
+  `CometIcebergWriteExec` (alongside `CometNativeWriteExec`) so the upstream Comet chain's
+  outermost node gets `convertBlock()` called on it and populates its `serializedPlanOpt`.
+  Without this, an upstream `CometProject` reaches execution without a serialised plan and
+  `doExecuteColumnar` throws.
+- **`CometMetricNode.fromCometPlan` stops at non-Comet descendants.** AQE-stage nodes
+  (`AQEShuffleReadExec`) can have a null `session` field when constructed off the planning
+  thread; their lazy `metrics` val NPEs on `sparkContext`. They're not ours to update anyway,
+  so the metric tree walk just stops at the boundary.
+
+## Native acceleration internals
+
+The high-level shape: the JVM-side serde marshals everything iceberg-rust needs (table
+properties, write schema, partition spec, target file size, writer mode, per-task IDs, …) into
+a single proto, and on each task iceberg-rust returns one `BINARY` row containing an Iceberg V2
+data manifest (Avro container bytes). The JVM-side commit decodes those bytes back into
+`DataFile`s via `ManifestFiles.read` and feeds them to `BatchWrite.commit(messages)`. So
+everything iceberg-java would do post-write (snapshot id assignment, manifest list aggregation,
+commit retries, cache refresh) is untouched — only the per-task parquet emission moved.
+
+Notable details:
+
+- **Per-task IDs are stamped onto the proto.** `CometIcebergWriteExec.doExecuteColumnar` reads
+  `TaskContext.getPartitionId()` and `taskAttemptId()` and rebuilds the `IcebergWrite` proto's
+  `partition_id` / `task_attempt_id` per task before serialising. Without this every task mints
+  the same filename prefix (`00000-00000-<op_id>`) and OpenDAL's `CompleteWriter` trips on the
+  resulting concurrent-write byte-count drift. Mirrors `CometNativeWriteExec`'s parquet pattern.
+- **Spark 4.x `ReplaceData` carries extra columns** that the native writer can't pass through
+  as-is; `CometIcebergNativeWrite.maybeProjectReplaceDataColumns` splices a DataFusion
+  `Projection` between the FFI scan and `IcebergWrite` to strip them. The method docstring
+  has the full mechanics.
+
+## Iceberg 1.5.2 (Spark 3.4) reflection skew
+
+Iceberg 1.5.2 differs in a handful of small ways from 1.8+. All of them are papered over with
+reflection in `IcebergReflection` rather than per-version source shims:
+
+- **`useFanoutWriter` field renamed.** 1.5.2 calls it `partitionedFanoutEnabled`. The
+  `getUseFanoutWriterFromSparkWrite` helper tries the new name first and falls back to the old.
+- **`GenericManifestFile` constructor.** Iceberg 1.6 added a 3-arg form
+  `(InputFile, int, long)` that takes V3's `firstRowId`. 1.5.2 only has the 2-arg form
+  `(InputFile, int)`. `newDataManifestFile` tries the 3-arg form first and falls back.
+- **Null `snapshotId` rejected on read.** Iceberg 1.5.2's
+  `InheritableMetadataFactory.fromManifest` throws "Cannot read from ManifestFile with null
+  (unassigned) snapshot ID". 1.8+ relaxed that check. The helper reflectively writes a `0L`
+  placeholder onto `GenericManifestFile.snapshotId` after construction; it's overwritten by
+  `BatchWrite.commit` later and never reaches storage.
+- **Separate logical write node.** Iceberg 1.5.2 ships its own `ReplaceIcebergData` logical
+  node because Spark 3.4 lacks native row-level operation support. Field shape is identical to
+  Spark 3.5+'s stock `ReplaceData`, so `IcebergWriteStrategy` matches it by FQCN and the
+  extracted tuple reuses the existing dispatcher.
+
 ## Why a custom logical anchor (`IcebergWriteLogical`)
 
 Without `IcebergWriteLogical`, AQE's `reOptimize` would either:
@@ -220,10 +329,3 @@ Without `IcebergWriteLogical`, AQE's `reOptimize` would either:
 
 `IcebergWriteLogical` sits between the two and lets the strategy re-emit a fresh
 `IcebergWriteExec` on every iteration without touching the committer.
-
-## Iceberg 1.5.2 (Spark 3.4) logical-write skew
-
-Iceberg 1.5.2 (paired with Spark 3.4) ships its own `ReplaceIcebergData` logical node because
-Spark 3.4 lacks native row-level operation support. Field shape is identical to Spark 3.5+'s
-stock `ReplaceData`, so `IcebergWriteStrategy` matches it by FQCN and the extracted tuple feeds
-the same dispatcher.

native/core/src/execution/jni_api.rs

@@ -236,6 +236,7 @@ fn op_name(op: &OpStruct) -> &'static str {
         OpStruct::Window(_) => "Window",
         OpStruct::NativeScan(_) => "NativeScan",
         OpStruct::IcebergScan(_) => "IcebergScan",
+        OpStruct::IcebergWrite(_) => "IcebergWrite",
         OpStruct::ParquetWriter(_) => "ParquetWriter",
         OpStruct::Explode(_) => "Explode",
         OpStruct::CsvScan(_) => "CsvScan",

native/core/src/execution/operators/iceberg_common.rs

@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Helpers shared between the Iceberg scan and Iceberg write operators.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::common::DataFusionError;
+use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
+use iceberg_storage_opendal::OpenDalStorageFactory;
+
+/// Pick an OpenDAL storage backend from a URI's scheme. `file` (or no scheme) falls through to
+/// the local file system. `memory` is used by the write path to assemble manifest bytes that
+/// stay entirely in-process.
+pub(crate) fn storage_factory_for(path: &str) -> Result<Arc<dyn StorageFactory>, DataFusionError> {
+    let scheme = if path.contains("://") {
+        path.split("://").next().unwrap_or("file")
+    } else {
+        "file"
+    };
+    match scheme {
+        "file" => Ok(Arc::new(OpenDalStorageFactory::Fs)),
+        "memory" => Ok(Arc::new(OpenDalStorageFactory::Memory)),
+        "s3" | "s3a" => Ok(Arc::new(OpenDalStorageFactory::S3 {
+            customized_credential_load: None,
+        })),
+        "gs" => Ok(Arc::new(OpenDalStorageFactory::Gcs)),
+        "oss" => Ok(Arc::new(OpenDalStorageFactory::Oss)),
+        _ => Err(DataFusionError::Execution(format!(
+            "Unsupported storage scheme: {scheme}"
+        ))),
+    }
+}
+
+/// Build a `FileIO` whose storage scheme is inferred from `reference_path` and whose properties
+/// come from the catalog. The reference path is the metadata location for reads or the data
+/// location for writes — anything that carries the right URI scheme.
+pub(crate) fn load_file_io(
+    catalog_properties: &HashMap<String, String>,
+    reference_path: &str,
+) -> Result<FileIO, DataFusionError> {
+    let factory = storage_factory_for(reference_path)?;
+    let mut file_io_builder = FileIOBuilder::new(factory);
+    for (key, value) in catalog_properties {
+        file_io_builder = file_io_builder.with_prop(key, value);
+    }
+    Ok(file_io_builder.build())
+}