From bbda506be2d39bdf9349f289c10a850c30072c92 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 10:25:18 -0400 Subject: [PATCH] Add min==max check to ensure bucket transform correctness in AddFiles --- .../apache/beam/sdk/io/iceberg/AddFiles.java | 18 +++ .../beam/sdk/io/iceberg/AddFilesTest.java | 139 ++++++++++++++---- 2 files changed, 129 insertions(+), 28 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index f37935f89e87..1a4d5d47ddbd 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -650,6 +650,24 @@ static String getPartitionFromMetrics(Metrics metrics, InputFile inputFile, Tabl "Min and max transformed values were not equal, for column: " + field.name()); } + // Equal transformed bounds only cover the values in between for + // order-preserving transforms. For bucket, which hashes the value mod N + // , min and max can land in the same bucket while intermediate values + // land in others. The void transform maps every value to null and needs + // no check. + if (!transform.preservesOrder() + && !transform.isVoid() + && !Objects.deepEquals( + Conversions.fromByteBuffer(type, lowerBytes), + Conversions.fromByteBuffer(type, upperBytes))) { + throw new UnknownPartitionException( + "Transform " + + transform + + " does not preserve ordering, and min and max raw values were not equal," + + " for column: " + + field.name()); + } + pk.set(i, lowerTransformedValue); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 6a9ac5c261c7..5375035c70a5 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -26,6 +26,7 @@ import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.hasEntry; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -75,6 +76,7 @@ import org.apache.iceberg.mapping.MappingUtil; import org.apache.iceberg.mapping.NameMappingParser; import org.apache.iceberg.parquet.Parquet; +import org.apache.iceberg.transforms.Transforms; import org.apache.iceberg.types.Conversions; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.SerializableFunction; @@ -527,38 +529,32 @@ public void testGetPartitionFromMetrics() throws IOException, InterruptedExcepti .identity("age") .build(); + // The bucketed column ("id") is single-valued per file: bucket does not preserve + // ordering, so min/max stats can only prove a file's bucket when min == max. List testCases = Arrays.asList( PartitionTestCase.of( root + "data_1.parquet", - record(1, "aaaa", 10), Arrays.asList( - record(1, "aaaa123", 10), - record(10, "aaaa789", 10), - record(100, "aaaa456", 10)), + record(1, "aaaa123", 10), record(1, "aaaa789", 10), record(1, "aaaa456", 10)), Arrays.asList(1, CharBuffer.wrap("aaaa123"), 10), - Arrays.asList(100, CharBuffer.wrap("aaaa789"), 10), + Arrays.asList(1, CharBuffer.wrap("aaaa789"), 10), "id_bucket=0/name_trunc=aaaa/age=10"), PartitionTestCase.of( root + "data_2.parquet", - record(1, "bbbb", 30), Arrays.asList( - record(5, "bbbb789", 30), - record(55, "bbbb456", 30), - record(500, "bbbb123", 30)), + record(5, "bbbb789", 30), record(5, "bbbb456", 30), record(5, "bbbb123", 30)), Arrays.asList(5, CharBuffer.wrap("bbbb123"), 30), - Arrays.asList(500, CharBuffer.wrap("bbbb789"), 30), + Arrays.asList(5, CharBuffer.wrap("bbbb789"), 30), "id_bucket=1/name_trunc=bbbb/age=30")); - PartitionKey pk = new PartitionKey(partitionSpec, icebergSchema); MetricsConfig metricsConfig = MetricsConfig.fromProperties(tableProps); Table table = catalog.createTable(tableId, icebergSchema, partitionSpec); for (PartitionTestCase caze : testCases) { List records = caze.records; String fileName = caze.fileName; - pk.wrap(caze.partition); - DataWriter writer = createWriter(fileName, pk.copy()); + DataWriter writer = createWriter(fileName); for (Record record : records) { writer.write(record); @@ -598,32 +594,29 @@ public void testThrowPartitionMismatchError() throws IOException, InterruptedExc List testCases = Arrays.asList( + // Straddles two truncate(name, 4) partitions. PartitionTestCase.of( root + "data_1.parquet", - record(1, "aaaa", 10), Arrays.asList( - record(1, "aaaa123", 10), record(10, "abab", 10), record(100, "aaaa789", 10)), + record(1, "aaaa123", 10), record(1, "abab", 10), record(1, "aaaa789", 10)), Arrays.asList(1, CharBuffer.wrap("aaaa123"), 10), - Arrays.asList(100, CharBuffer.wrap("abab"), 10), + Arrays.asList(1, CharBuffer.wrap("abab"), 10), "error"), + // Straddles two identity(age) partitions. PartitionTestCase.of( root + "data_2.parquet", - record(1, "bbbb", 30), - Arrays.asList( - record(5, "bbbb", 30), record(55, "bbbb", 30), record(500, "bbbb", 50)), + Arrays.asList(record(5, "bbbb", 30), record(5, "bbbb", 30), record(5, "bbbb", 50)), Arrays.asList(5, CharBuffer.wrap("bbbb"), 30), - Arrays.asList(500, CharBuffer.wrap("bbbb"), 50), + Arrays.asList(5, CharBuffer.wrap("bbbb"), 50), "error")); - PartitionKey pk = new PartitionKey(partitionSpec, icebergSchema); MetricsConfig metricsConfig = MetricsConfig.fromProperties(tableProps); Table table = catalog.createTable(tableId, icebergSchema, partitionSpec); for (PartitionTestCase caze : testCases) { List records = caze.records; String fileName = caze.fileName; - pk.wrap(caze.partition); - DataWriter writer = createWriter(fileName, pk.copy()); + DataWriter writer = createWriter(fileName); for (Record record : records) { writer.write(record); @@ -654,9 +647,102 @@ public void testThrowPartitionMismatchError() throws IOException, InterruptedExc } } + /** + * Bucket is a hash, so equal transformed min/max bounds do not imply that the values in between + * fall in the same bucket. A file whose bounds collide into one bucket while an intermediate row + * hashes elsewhere must go to the DLQ; a silent wrong assignment would hide that row from + * partition-pruned queries. + */ + @Test + public void testBucketPartitionRejectsMultiValuedColumn() + throws IOException, InterruptedException { + PartitionSpec partitionSpec = PartitionSpec.builderFor(icebergSchema).bucket("id", 2).build(); + + // Deterministic collision: ids 1 and 4 hash to bucket 0, while 3 (between them) hashes to + // bucket 1. + SerializableFunction bucket = + Transforms.bucket(2).bind(Types.IntegerType.get()); + assertEquals(bucket.apply(1), bucket.apply(4)); + assertNotEquals(bucket.apply(1), bucket.apply(3)); + + String fileName = root + "bucket_straddle.parquet"; + DataWriter writer = createWriter(fileName); + writer.write(record(1, "aaaa", 10)); + writer.write(record(3, "aaaa", 10)); + writer.write(record(4, "aaaa", 10)); + writer.close(); + + Table table = catalog.createTable(tableId, icebergSchema, partitionSpec); + InputFile file = table.io().newInputFile(fileName); + Metrics metrics = + AddFiles.getFileMetrics( + file, + FileFormat.PARQUET, + MetricsConfig.fromProperties(tableProps), + MappingUtil.create(icebergSchema)); + + AddFiles.UnknownPartitionException e = + assertThrows( + AddFiles.UnknownPartitionException.class, + () -> getPartitionFromMetrics(metrics, file, table)); + assertThat(e.getMessage(), containsString("does not preserve ordering")); + assertThat(e.getMessage(), containsString("id_bucket")); + } + + @Test + public void testBucketPartitionSingleValuedColumnResolves() + throws IOException, InterruptedException { + PartitionSpec partitionSpec = PartitionSpec.builderFor(icebergSchema).bucket("id", 2).build(); + + String fileName = root + "bucket_single.parquet"; + DataWriter writer = createWriter(fileName); + writer.write(record(3, "aaaa", 10)); + writer.write(record(3, "bbbb", 20)); + writer.close(); + + Table table = catalog.createTable(tableId, icebergSchema, partitionSpec); + InputFile file = table.io().newInputFile(fileName); + Metrics metrics = + AddFiles.getFileMetrics( + file, + FileFormat.PARQUET, + MetricsConfig.fromProperties(tableProps), + MappingUtil.create(icebergSchema)); + + assertEquals("id_bucket=1", getPartitionFromMetrics(metrics, file, table)); + } + + /** + * The void transform does not preserve ordering either, but it maps every value to null, so any + * file is trivially single-partition and must not be rejected. + */ + @Test + public void testVoidTransformResolvesForMultiValuedColumn() + throws IOException, InterruptedException { + PartitionSpec partitionSpec = + PartitionSpec.builderFor(icebergSchema).alwaysNull("id").identity("age").build(); + + String fileName = root + "void_multi.parquet"; + DataWriter writer = createWriter(fileName); + writer.write(record(1, "aaaa", 10)); + writer.write(record(3, "aaaa", 10)); + writer.write(record(4, "aaaa", 10)); + writer.close(); + + Table table = catalog.createTable(tableId, icebergSchema, partitionSpec); + InputFile file = table.io().newInputFile(fileName); + Metrics metrics = + AddFiles.getFileMetrics( + file, + FileFormat.PARQUET, + MetricsConfig.fromProperties(tableProps), + MappingUtil.create(icebergSchema)); + + assertEquals("id_null=null/age=10", getPartitionFromMetrics(metrics, file, table)); + } + static class PartitionTestCase { String fileName; - StructLike partition; List records; List expectedLower; List expectedUpper; @@ -664,13 +750,11 @@ static class PartitionTestCase { PartitionTestCase( String fileName, - StructLike partition, List records, List expectedLower, List expectedUpper, String expectedPartition) { this.fileName = fileName; - this.partition = partition; this.records = records; this.expectedLower = expectedLower; this.expectedUpper = expectedUpper; @@ -679,13 +763,12 @@ static class PartitionTestCase { static PartitionTestCase of( String fileName, - StructLike partition, List records, List expectedLower, List expectedUpper, String expectedPartition) { return new PartitionTestCase( - fileName, partition, records, expectedLower, expectedUpper, expectedPartition); + fileName, records, expectedLower, expectedUpper, expectedPartition); } }