Parquet: Fix variant STRING metrics bounds to use UTF-8 byte order

parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java

@@ -184,10 +184,13 @@ static <T> T convertValue(PhysicalType primitive, int scale, Object value) {
    */
   @SuppressWarnings("unchecked")
   static <T> Comparator<T> comparator(PhysicalType primitive) {
-    if (primitive == PhysicalType.BINARY) {
-      return (Comparator<T>) Comparators.unsignedBytes();
-    } else {
-      return (Comparator<T>) Comparator.naturalOrder();
+    switch (primitive) {
+      case BINARY:
+        return (Comparator<T>) Comparators.unsignedBytes();
+      case STRING:
+        return (Comparator<T>) Comparators.charSequences();
+      default:
+        return (Comparator<T>) Comparator.naturalOrder();
     }
   }
 

parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.parquet;
 
+import static org.apache.iceberg.TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
@@ -273,6 +274,33 @@ public void testShreddedBinaryUpperBoundOverflow() throws IOException {
         .isNull();
   }
 
+  @Test
+  public void testShreddedStringBoundsAcrossRowGroups() throws IOException {
+    // shredded string bounds must use UTF-8 order (Comparators.charSequences), like the scan path;
+    // String.compareTo (UTF-16) sorts a supplementary char below U+E000 and would invert the bounds
+    String belowSurrogate = new String(Character.toChars(0xE000));
+    String supplementary = new String(Character.toChars(0x10000));
+
+    Variant[] rows = new Variant[300];
+    for (int i = 0; i < rows.length; i += 1) {
+      rows[i] = Variant.of(EMPTY, Variants.of(i < 150 ? belowSurrogate : supplementary));
+    }
+
+    // a tiny row-group size forces multiple row groups so cross-chunk bound aggregation runs
+    Metrics metrics =
+        writeParquetWithRowGroupSize(
+            (id, name) -> ParquetVariantUtil.toParquetSchema(Variants.of(belowSurrogate)),
+            "1",
+            rows);
+
+    assertThat(metrics.lowerBounds().get(2))
+        .extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
+        .isEqualTo(Variants.of(belowSurrogate));
+    assertThat(metrics.upperBounds().get(2))
+        .extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
+        .isEqualTo(Variants.of(supplementary));
+  }
+
   @Test
   public void testVariantFloatNaN() throws IOException {
     // NaN values are not counted because there is no ID for FieldMetrics
@@ -578,16 +606,27 @@ public void testShreddedValueColumnWithEmptyStats() throws IOException {
 
   private Metrics writeParquet(VariantShreddingFunction shredding, Variant... variants)
       throws IOException {
-    OutputFile out = new InMemoryOutputFile();
-    GenericRecord record = GenericRecord.create(SCHEMA);
+    return writeRecords(writeBuilder(shredding), variants);
+  }
+
+  private Metrics writeParquetWithRowGroupSize(
+      VariantShreddingFunction shredding, String rowGroupSizeBytes, Variant... variants)
+      throws IOException {
+    return writeRecords(
+        writeBuilder(shredding).set(PARQUET_ROW_GROUP_SIZE_BYTES, rowGroupSizeBytes), variants);
+  }
 
-    FileAppender<Record> writer =
-        Parquet.write(out)
-            .schema(SCHEMA)
-            .variantShreddingFunc(shredding)
-            .createWriterFunc(fileSchema -> InternalWriter.create(SCHEMA.asStruct(), fileSchema))
-            .build();
+  private Parquet.WriteBuilder writeBuilder(VariantShreddingFunction shredding) {
+    return Parquet.write(new InMemoryOutputFile())
+        .schema(SCHEMA)
+        .variantShreddingFunc(shredding)
+        .createWriterFunc(fileSchema -> InternalWriter.create(SCHEMA.asStruct(), fileSchema));
+  }
 
+  private Metrics writeRecords(Parquet.WriteBuilder builder, Variant... variants)
+      throws IOException {
+    GenericRecord record = GenericRecord.create(SCHEMA);
+    FileAppender<Record> writer = builder.build();
     try (writer) {
       for (int id = 0; id < variants.length; id += 1) {
         record.setField("id", (long) id);