From 9449b8ea65c99be20911be19d4322bba809eb788 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Tue, 28 Jul 2026 19:09:25 +0530 Subject: [PATCH] Parquet: Fix variant STRING metrics bounds to use UTF-8 byte order --- .../iceberg/parquet/ParquetVariantUtil.java | 11 ++-- .../iceberg/parquet/TestVariantMetrics.java | 55 ++++++++++++++++--- 2 files changed, 54 insertions(+), 12 deletions(-) diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java index a9b088aaf294..9d5dbbba0af7 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java @@ -184,10 +184,13 @@ static T convertValue(PhysicalType primitive, int scale, Object value) { */ @SuppressWarnings("unchecked") static Comparator comparator(PhysicalType primitive) { - if (primitive == PhysicalType.BINARY) { - return (Comparator) Comparators.unsignedBytes(); - } else { - return (Comparator) Comparator.naturalOrder(); + switch (primitive) { + case BINARY: + return (Comparator) Comparators.unsignedBytes(); + case STRING: + return (Comparator) Comparators.charSequences(); + default: + return (Comparator) Comparator.naturalOrder(); } } diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java index 8381052f58d3..3453b5c928f3 100644 --- a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java +++ b/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 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 writer = builder.build(); try (writer) { for (int id = 0; id < variants.length; id += 1) { record.setField("id", (long) id);