ahmedabu98 commented on code in PR #39720:
URL: https://github.com/apache/beam/pull/39720#discussion_r3760368733
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java:
##########
@@ -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<PartitionTestCase> 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<Record> records = caze.records;
String fileName = caze.fileName;
- pk.wrap(caze.partition);
- DataWriter<Record> writer = createWriter(fileName, pk.copy());
Review Comment:
This is meant to create a partitioned writer. Why remove it?
##########
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))) {
Review Comment:
This is too strict and will lead to false negatives. It's normal for a file
to contain different values that hash to the same bucket. This will incorrectly
turn them away.
Bucketing is a common transform so we need to be careful how we deal with it.
##########
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))) {
Review Comment:
We used to have a more thorough but costly validation In the initial
implementation.
If we encountered a bucket partitioned column, we would iterate through
every value of that column and apply the transform to check that they all map
to the same partition value. We ended up removing it though because we expected
it to be a big bottleneck
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
