jordepic opened a new pull request, #5361: URL: https://github.com/apache/datafusion-comet/pull/5361
## Which issue does this PR close? (Third of three) Closes #4322 and closes #5308. Part one was #4658 (split-operator plan, merged); part two was #5298 (detection + fall-back reasons, merged). ## Rationale for this change With detection in place (#5298), this PR wires in the native writer itself: when `spark.comet.iceberg.write.enabled=true` and a write passes the eligibility gate, the `IcebergWrite` operator's per-task Parquet write is delegated to iceberg-rust through Comet's native execution pipeline. Commit semantics are untouched — the JVM still assembles `SparkWrite$TaskCommit` messages and `IcebergCommit` performs the normal `BatchWrite.commit`, so snapshot assignment, validation, and retries are iceberg-java's. The key design point (a divergence from the original #4487, as discussed on #5298) is that manifest metadata is Java's decision by construction: iceberg-rust returns each task's `DataFile`s packed as an in-memory Iceberg V2 data manifest, the JVM decodes them with `ManifestFiles.read`, and then re-derives every file's manifest metrics from the written Parquet footer through the version-matched `ParquetUtil.footerMetrics` + `MetricsConfig.forTable` (float/double NaN counts and bounds are carried over from the rust writer, mirroring Java's writer-tracked `FieldMetrics`). This makes metrics modes, `truncate(N)` bound adjustment, per-column overrides, sorted-column promotion, and the inferred-column cap all run inside the linked Iceberg's own code — which is why the `counts`/`none` metrics-mode gates from #5298 are now lifted (every `write.metadata.metrics.*` value is supported). The native writer always emits full, untruncated footer statistics, matching parquet-mr. Remaining unconditional differences behind the toggle (compressed bytes, footer key-value metadata, file rolling cadence, float zero-sign bound normalization, etc.) are documented in `iceberg-writes.md` under "Accepted divergences". This also addresses @unikdahal's review comment on #4487: `IcebergReflection.getMetadataLocation` now uses `Option(...)` instead of `Some(...)` and the write path defaults a missing metadata location to `""`, so CTAS/RTAS (where no metadata file exists yet) run natively instead of falling back with an opaque NPE message. ## What changes are included in this PR? 1) `iceberg_write.rs`: a native operator that drives iceberg-rust's writer stack (`ParquetWriterBuilder` → `RollingFileWriterBuilder` → `DataFileWriterBuilder` → unpartitioned/fanout/clustered partitioning writers) and emits the task's manifest bytes as a single Binary row. Notable: the partition splitter returns parts in HashMap order while `ClusteredWriter` rejects revisits of closed partitions, so the operator restores each batch's first-occurrence partition order before writing. 2) `CometIcebergWriteExec`: stamps per-task IDs onto the plan, drains the native output, decodes the manifest, stamps `sort_order_id`, rebuilds each `DataFile`'s metrics from the footer (see above), and packages the same commit message the JVM writer would have produced. 3) `CometIcebergNativeWrite.convert`/`createExec` now actually build the native plan (the serde's gate is unchanged apart from lifting the metrics-mode rules and adding one new rule: a non-integer `write.parquet.compression-level` falls back, because iceberg-java throws for it at write time while the native translation would have silently substituted a default). 4) Driver-side proto translation (`IcebergWriteProtoTranslation`) for the vetted parquet settings, substituting parquet-mr's zstd default level 3 for byte-size parity. 5) Docs: `iceberg-writes.md` gains "How the native write works" and updated eligibility/divergence tables. ## How are these changes tested? - `CometIcebergWriteActionSuite` gains a "native acceleration" section (~20 new tests): appends (unpartitioned/partitioned/fanout with unsorted input), overwrite static/dynamic, CoW DELETE/UPDATE natively and a pin that CoW MERGE intentionally falls back (`MergeRowsExec` is not Comet-native), CTAS/RTAS on Spark 3.5+ (pinning the empty-metadata-location path), complex types with nested field IDs, multi-file rolling via `write.target-file-size-bytes`, empty appends (one commit, zero files), and `sort_order_id` parity with the JVM writer (0 through Iceberg 1.10; the resolved order id on 1.11+). - Manifest-metadata parity tests write identical data through the native and JVM paths and compare the committed manifest metrics: NaN float/double counts and bounds (incl. the all-NaN bounds-drop), string bound truncation under the default `truncate(16)`, metrics mode `none` with a per-column `full` override, and mode `counts` with `WRITE ORDERED BY` (pinning the sorted-column promotion inside `MetricsConfig.forTable`). - Rewrite actions (`CometIcebergRewriteActionSuite`) now assert compaction data files were physically written by iceberg-rust via the parquet footer `created_by` signature. - Rust unit + integration tests for the writer translation, manifest round-trip, partitioning writers (including a regression test for a partition spanning a batch boundary under the clustered writer), and file naming. - All suites pass locally on all four profiles: spark-3.4 (Iceberg 1.5.2, Scala 2.12), spark-3.5 (1.8.1), spark-4.0 (1.10.0), spark-4.1 (1.11.0) — the version spread matters because the metrics rebuild reflects version-specific Iceberg internals (e.g. Iceberg 1.10+ has two 6-arg `FieldMetrics` constructors, and 1.11 changed what `sort_order_id` the Spark writer stamps). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
