feat: support native Iceberg UUID writes

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` 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
@@ -461,6 +466,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

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

@@ -494,7 +494,7 @@ A write is eligible only when ALL of the following hold:
 | 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.<data-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`                                                                                                                                                                                                                                                                                                             |
+| column types                                                                                                                                | any except 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.
@@ -587,6 +587,14 @@ or GCS each read is a GET request, so the task waits two request round trips per
 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.
 
+UUID columns, including struct fields, array elements and map keys/values, are normalized
+from Spark strings using Java's `UUID.fromString` rules. This accepts uppercase, shortened
+groups and the other non-canonical forms accepted by Java; an invalid visible value fails the
+task and triggers ordinary write cleanup. `fixed(16)` columns retain binary semantics.
+UUID columns used as partition or sort sources currently fall back, because Spark distributes
+and orders the input strings before UUID normalization. UUID data columns remain eligible in
+unpartitioned, clustered and fanout writes partitioned or sorted by other columns.
+
 ### Failure handling
 
 Eligibility is decided entirely at plan time. That includes the reflection surface: every
@@ -656,16 +664,23 @@ unchanged.
 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
+different blast radius: differences confined to the physical bytes of a data file, 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)
+#### Physical file layout
 
-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:
+Iceberg readers use the table schema and recover the same values from these layouts.
+Tools that inspect Parquet files without the Iceberg schema can observe the UUID annotation
+difference below; the remaining differences are cosmetic:
 
+- UUID leaves use `FIXED_LEN_BYTE_ARRAY(16)` with Iceberg field ids and big-endian UUID bytes,
+  but the pinned iceberg-rust writer omits the Parquet `UUID` logical annotation that
+  iceberg-java emits. This applies to nested UUIDs too. Iceberg readers interpret these fields
+  using the table schema; a standalone Parquet reader relying on the logical annotation sees
+  fixed-length binary instead of UUID. Native writes therefore do not provide full UUID footer
+  parity ([#6959](https://github.com/apache/datafusion-comet/issues/6959)).
 - 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`).