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
@@ -153,16 +166,28 @@ Parquet Rust encoder that iceberg-rust drives).
 
 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.
+
 ## 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 +231,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 +305,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

@@ -237,6 +237,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())
+}

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

@@ -39,9 +39,8 @@ use datafusion::physical_plan::{
 };
 use futures::{Stream, StreamExt, TryStreamExt};
 use iceberg::arrow::ScanMetrics;
-use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
-use iceberg_storage_opendal::OpenDalStorageFactory;
 
+use crate::execution::operators::iceberg_common::load_file_io;
 use crate::execution::operators::ExecutionError;
 use crate::parquet::parquet_support::SparkParquetOptions;
 use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
@@ -154,7 +153,7 @@ impl IcebergScanExec {
         context: Arc<TaskContext>,
     ) -> DFResult<SendableRecordBatchStream> {
         let output_schema = Arc::clone(&self.output_schema);
-        let file_io = Self::load_file_io(&self.catalog_properties, &self.metadata_location)?;
+        let file_io = load_file_io(&self.catalog_properties, &self.metadata_location)?;
         let batch_size = context.session_config().batch_size();
 
         let metrics = IcebergScanMetrics::new(&self.metrics);
@@ -198,39 +197,6 @@ impl IcebergScanExec {
 
         Ok(Box::pin(wrapped_stream))
     }
-
-    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)),
-            "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}"
-            ))),
-        }
-    }
-
-    fn load_file_io(
-        catalog_properties: &HashMap<String, String>,
-        metadata_location: &str,
-    ) -> Result<FileIO, DataFusionError> {
-        let factory = Self::storage_factory_for(metadata_location)?;
-        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())
-    }
 }
 
 /// Metrics for IcebergScanExec