feat: support native Iceberg UUID writes

.ai/skills/review-comet-iceberg-write-pr/SKILL.md

@@ -132,7 +132,7 @@ only the latest.
 - [ ] **Nothing nondeterministic reaches the output.** Map iteration order leaks into file names,
       manifest order and the row order of unordered reads. The fanout arm sorts its `DataFile`s by
       path for this reason ([#5776](https://github.com/apache/datafusion-comet/issues/5776)).
-- [ ] **Casts stay strict.** `decorate_batch_with_field_ids` casts with `safe: false`, so a type
+- [ ] **Casts stay strict.** `UuidAlignment::convert` casts with `safe: false`, so a type
       mismatch fails the task. A safe cast would write NULLs.
 
 ## 4. Failure Handling and Cleanup

dev/GenerateIcebergUuidOracle.java

@@ -0,0 +1,137 @@
+/*
+ * 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.
+ */
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+/** Run with JDK 17 and 21; the generated fixtures must match byte for byte. */
+class GenerateIcebergUuidOracle {
+  private static final String LICENSE =
+      """
+      # 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.
+      """;
+
+  public static void main(String[] args) throws Exception {
+    System.err.println(System.getProperty("java.vendor") + " " + Runtime.version());
+    var lines = new ArrayList<>(LICENSE.lines().toList());
+    lines.add(
+        "# Generated by dev/GenerateIcebergUuidOracle.java; UTF-8 hex input, UUID hex or error");
+    var inputs =
+        new ArrayList<>(
+            List.of(
+                "",
+                "1-1-1-1-1",
+                "+1-1-1-1-1",
+                "100000000-1-1-1-1",
+                "1-10000-1-1-1",
+                "１-１-１-１-１",
+                "١-١-١-١-١",
+                "Ａ-ｂ-Ｃ-ｄ-Ｅ",
+                "𝟙-1-1-1-1",
+                "-1-1-1-1-1",
+                "1--1-1-1",
+                "1-1-1-1-8000000000000000",
+                "f47ac10b58cc4372a5670e02b2c3d479",
+                "{f47ac10b-58cc-4372-a567-0e02b2c3d479}",
+                "f47ac10b-58cc-4372-a567-0e02b2c3d479",
+                "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
+                "00000000-0000-0000-0000-000000000000",
+                " 1-1-1-1-1",
+                "1-1-1-1-1\n"));
+    for (int group = 0; group < 5; group++) {
+      for (String value :
+          List.of(
+              "",
+              "+",
+              "++1",
+              "g",
+              "７",
+              "+f",
+              "00000000000000001",
+              "7fffffffffffffff",
+              "8000000000000000",
+              "ffffffffffffffff",
+              "0000000000000000000001")) {
+        var groups = new String[] {"1", "1", "1", "1", "1"};
+        groups[group] = value;
+        inputs.add(String.join("-", groups));
+      }
+    }
+    // Check Character.digit(char, 16) over the entire BMP, including rejected numeric characters.
+    var digits = new ArrayList<>(LICENSE.lines().toList());
+    for (int c = 0; c <= 0xffff; c++) {
+      int digit = Character.digit((char) c, 16);
+      if (digit >= 0) {
+        digits.add(String.format("%04x\t%d", c, digit));
+        inputs.add((char) c + "-1-1-1-1");
+      }
+    }
+    var random = new Random(6959);
+    String alphabet = "0123456789abcdefABCDEF+- g\t１١Ａ²Ⅳ𝟙";
+    // Include valid groups of irregular widths, then mutate separators, signs and characters.
+    for (int i = 0; i < 1024; i++) {
+      var groups = new ArrayList<String>();
+      for (int g = 0; g < 5; g++) {
+        String hex = Long.toHexString(random.nextLong() & Long.MAX_VALUE);
+        groups.add(hex.substring(0, 1 + random.nextInt(hex.length())));
+      }
+      String value = String.join("-", groups);
+      inputs.add(value);
+      int at = random.nextInt(value.length());
+      // Use code points to avoid creating unpaired surrogates that UTF-8 cannot represent.
+      int[] chars = alphabet.codePoints().toArray();
+      String replacement = new String(Character.toChars(chars[random.nextInt(chars.length)]));
+      inputs.add(value.substring(0, at) + replacement + value.substring(at + 1));
+    }
+    for (String input : inputs) {
+      String result;
+      try {
+        result = UUID.fromString(input).toString().replace("-", "");
+      } catch (IllegalArgumentException e) {
+        result = "error";
+      }
+      lines.add(HexFormat.of().formatHex(input.getBytes(StandardCharsets.UTF_8)) + "\t" + result);
+    }
+    Path out = Path.of(args[0]);
+    Files.createDirectories(out);
+    Files.write(out.resolve("uuid-java.tsv"), lines, StandardCharsets.UTF_8);
+    Files.write(out.resolve("uuid-java-digits.tsv"), digits, StandardCharsets.UTF_8);
+  }
+}

docs/source/contributor-guide/iceberg-writes.md

@@ -274,9 +274,14 @@ Points where Comet adapts iceberg-rust to match iceberg-java:
   accounting follows `FallbackValuesWriter`, `DictionaryValuesWriter` and
   `RunLengthBitPackingHybridEncoder`, and its tests pin answers recorded from parquet-mr itself. The
   held rows then go through `RowPacer`, so the roll grid does not move.
-- **Field ids and casting.** `decorate_batch_with_field_ids` casts each batch to the
+- **Field ids and casting.** `UuidAlignment::convert` (`iceberg_uuid.rs`) casts each batch to the
   field-id-annotated Arrow schema derived from the Iceberg schema, with `safe: false`, so a type
-  mismatch fails the task instead of writing NULLs.
+  mismatch fails the task instead of writing NULLs. It identifies UUID leaves by Iceberg field
+  id and parses their strings with Java-compatible UUID semantics into 16 big-endian bytes.
+  Conversion descriptors are built once per task; UUID-free subtrees keep the ordinary cast path.
+  Struct validity and list/map offset windows mask invisible values before parsing, including
+  required children under null parents. Invalid visible UUIDs fail through the existing abort
+  guard. `fixed(16)` is not treated as UUID.
 - **Deterministic output order.** `FanoutPartitions`, Comet's version of iceberg-rust's
   `FanoutWriter`, closes its files in an order that depends on how the rows arrived and on which
   partitions it closed early, so the fanout path sorts its `DataFile`s by path before returning them
@@ -442,13 +447,14 @@ the writer: run the write suites and the Iceberg Spark tests. The pin policy is
 
 | Suite                                                                | What it covers                                                                                                                                                                                                                 |
 | -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| `CometIcebergWriteActionSuite`                                       | End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, writes outgrowing the memory pool, AQE re-planning. |
+| `CometIcebergWriteActionSuite`                                       | End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, writes outgrowing the memory pool, AQE re-planning. UUID tests compare rows, nested fields, metrics and footers in all writer modes with AQE on/off, and check cleanup after an invalid UUID in a later batch. |
 | [`CometIcebergSchedulerFailureSuite`](iceberg-scheduler-failures.md) | Manual multi-host speculation and executor-loss recovery, with scheduler evidence and a manifest-to-storage audit.                                                                                                             |
-| `CometIcebergWriteDetectionSuite`                                    | Most eligibility rules, each accepted and declined.                                                                                                                                                                            |
+| `CometIcebergWriteDetectionSuite`                                    | Most eligibility rules, each accepted and declined, including UUID data column support and fallback for UUID partition or sort sources.                                                                                                                                                                            |
 | `CometIcebergSystemFunctionSuite`                                    | Native `bucket`, `truncate`, `years`/`months`/`days`/`hours`, which keep a partitioned write's hash distribution and sort native end to end.                                                                                   |
 | `CometIcebergRewriteActionSuite`                                     | Iceberg's `rewrite_data_files` with the split plan and the native writer.                                                                                                                                                      |
 | `IcebergWriteProtoTranslationSuite`                                  | Translation of properties into `IcebergParquetWriteSettings` and the writer mode.                                                                                                                                              |
 | Rust tests in `iceberg_write.rs` and in `iceberg_partition_*.rs`     | File rolling on the 1000-row grid, fanout order, clustered input checks, cleanup guard, manifest round trip, memory reservation, partition path rendering, partition values past `chrono`'s calendar.                          |
+| Rust tests in `iceberg_uuid.rs`                                     | Java-compatible UUID parsing against oracle fixtures, Unicode digits, field-id-based UUID detection, strict casts for non-UUID fields, and visibility masking for null parents and sliced lists/maps. |
 | Rust tests in `iceberg_dictionary.rs`                                | The per-column dictionary choice against answers recorded from parquet-mr, including columns either side of its cut-off.                                                                                                       |
 | `iceberg_dictionary_chooser` benchmark in `native/core/benches`      | The time one partition's dictionary choice takes, over primitive, struct, array, map and nested columns.                                                                                                                       |
 | `CometIcebergWriteBenchmark`                                         | Native versus iceberg-java for unpartitioned, clustered, fanout and copy-on-write delete writes. It checks each arm's plan before timing it.                                                                                   |
@@ -461,6 +467,27 @@ for Spark 4.2, so the 4.2 profile builds against the 4.0 runtime and the Iceberg
 Behaviour that differs between Iceberg versions, such as reflection targets, metrics conventions
 and partition path spelling, needs a test that runs on each.
 
+The native writer's UUID parser is pinned against fixtures generated with JDK 17 and 21;
+`dev/GenerateIcebergUuidOracle.java` records Java's acceptance and 16-byte output, plus
+`Character.digit` for the entire BMP. Regeneration instructions and the exact tested JDK versions
+are in `native/core/src/execution/operators/testdata/README.md`.
+
+Select the UUID writer benchmark separately with:
+
+```sh
+PROFILES=-Pspark-4.1 SPARK_GENERATE_BENCHMARK_FILES=1 \
+  make benchmark-org.apache.spark.sql.benchmark.CometIcebergWriteBenchmark uuid
+```
+
+It uses real Iceberg UUID columns, four million rows, eight source Parquet files, and three
+corpora: canonical UUIDs, canonical UUIDs with one null in eight, and Java-compatible shortened
+groups. Source generation and table creation are untimed. It rejects native fallback, checks
+normalized checksums, and reports five or more measured iterations with median, range, rows/s,
+writer factor, file count and output bytes. The default benchmark selection remains the non-UUID
+control (`control` selects only its unpartitioned insert). Comparing that control before and
+after a change requires matching builds and settings;
+the UUID writer factor alone is not evidence of a regression-free non-UUID path.
+
 When writing a native-write test:
 
 - **Assert that the native writer ran.** Every fallback is silent to the query, so a passing