From 3344178d099c9ea6e3fc399982e2cd720a571660 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 2 Sep 2026 07:23:43 -0700 Subject: [PATCH 1/2] spec-aware --- .../apache/beam/sdk/io/iceberg/AddFiles.java | 2 +- .../beam/sdk/io/iceberg/PartitionUtils.java | 10 +- .../sdk/io/iceberg/RecordWriterManager.java | 9 +- .../sdk/io/iceberg/SerializableDataFile.java | 91 +++++- .../io/iceberg/SerializableDeleteFile.java | 83 +++++- .../iceberg/WritePartitionedRowsToFiles.java | 3 +- .../beam/sdk/io/iceberg/cdc/CdcReadUtils.java | 2 +- .../cdc/SerializableChangelogTask.java | 19 +- .../sdk/io/iceberg/PartitionUtilsTest.java | 10 +- .../io/iceberg/RecordWriterManagerTest.java | 24 +- .../io/iceberg/SerializableDataFileTest.java | 269 ++++++++++++++++- .../iceberg/SerializableDeleteFileTest.java | 281 ++++++++++++++++-- .../sdk/io/iceberg/cdc/CdcReadUtilsTest.java | 7 +- .../io/iceberg/cdc/ChangelogScannerTest.java | 2 +- .../io/iceberg/cdc/LocalResolveDoFnTest.java | 2 +- .../iceberg/cdc/ReadFromChangelogsTest.java | 7 +- .../cdc/SerializableChangelogTaskTest.java | 2 +- 17 files changed, 719 insertions(+), 104 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..18b95ca50b1f 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 @@ -514,7 +514,7 @@ private Callable createProcessTask( .withPartitionPath(partitionPath) .build(); return new ProcessResult( - SerializableDataFile.from(df, partitionPath), null, timestamp, window, paneInfo); + SerializableDataFile.from(df, table.spec()), null, timestamp, window, paneInfo); }; } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionUtils.java index 32a25439d850..6c8e79c4b8c6 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionUtils.java @@ -154,7 +154,7 @@ static Term toIcebergTerm(String field) { * {@link ContentScanTask}s. */ public static Map constantsMap( - PartitionSpec spec, ContentFile file, @Nullable Long fileSequenceNumber) { + PartitionSpec spec, ContentFile file, @Nullable Long dataSequenceNumber) { Preconditions.checkState( spec.specId() == file.specId(), "File spec ID (%s) does not match PartitionSpec ID (%s)", @@ -172,13 +172,13 @@ static Term toIcebergTerm(String field) { convertConstant(Types.LongType.get(), file.firstRowId())); } - // When reconstructing a DataFile, we lose the ability to attach its fileSequenceNumber, + // When reconstructing a DataFile, we lose the ability to attach its dataSequenceNumber, // so we pipe it along the util methods to include it here. - fileSequenceNumber = - fileSequenceNumber != null ? fileSequenceNumber : file.fileSequenceNumber(); + dataSequenceNumber = + dataSequenceNumber != null ? dataSequenceNumber : file.dataSequenceNumber(); idToConstant.put( MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId(), - convertConstant(Types.LongType.get(), fileSequenceNumber)); + convertConstant(Types.LongType.get(), dataSequenceNumber)); // add _file idToConstant.put( diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java index 6893c743f431..64f0ab6232a4 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java @@ -104,7 +104,6 @@ class DestinationState { final Cache writers; private final List dataFiles = Lists.newArrayList(); @VisibleForTesting final Map writerCounts = Maps.newHashMap(); - private final Map partitionFieldMap = Maps.newHashMap(); private final List exceptions = Lists.newArrayList(); private final InternalRecordWrapper wrapper; // wrapper that facilitates partitioning @@ -115,9 +114,6 @@ class DestinationState { this.routingPartitionKey = new PartitionKey(spec, schema); this.wrapper = new InternalRecordWrapper(schema.asStruct()); this.table = table; - for (PartitionField partitionField : spec.fields()) { - partitionFieldMap.put(partitionField.name(), partitionField); - } // build a cache of RecordWriters. // writers will expire after 1 min of idle time. @@ -127,7 +123,6 @@ class DestinationState { .expireAfterAccess(1, TimeUnit.MINUTES) .removalListener( (RemovalNotification removal) -> { - final PartitionKey pk = Preconditions.checkStateNotNull(removal.getKey()); final RecordWriter recordWriter = Preconditions.checkStateNotNull(removal.getValue()); try { @@ -144,9 +139,9 @@ class DestinationState { throw rethrow; } openWriters--; - String partitionPath = getPartitionDataPath(pk.toPath(), partitionFieldMap); + // Serialize against the file's own spec (looked up by its spec id) dataFiles.add( - SerializableDataFile.from(recordWriter.getDataFile(), partitionPath)); + SerializableDataFile.from(recordWriter.getDataFile(), table.specs())); }) .build(); } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java index e1291601d149..aa3744fe4654 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java @@ -26,9 +26,11 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Equivalence; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps; import org.apache.iceberg.DataFile; @@ -37,7 +39,11 @@ import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionKey; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.SingleValueParser; +import org.apache.iceberg.StructLike; import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Serializable version of an Iceberg {@link DataFile}. @@ -49,12 +55,15 @@ *

NOTE: If you add any new fields here, you need to also update the {@link #equals} and {@link * #hashCode()} methods. * - *

Use {@link #from(DataFile, String)} to create a {@link SerializableDataFile} and {@link + *

Use {@link #from(DataFile, PartitionSpec)} to create a {@link SerializableDataFile} and {@link * #createDataFile(Map)} to reconstruct the original {@link DataFile}. */ @DefaultSchema(AutoValueSchema.class) @AutoValue +@Internal public abstract class SerializableDataFile { + private static final Logger LOG = LoggerFactory.getLogger(SerializableDataFile.class); + public static Builder builder() { return new AutoValue_SerializableDataFile.Builder(); } @@ -71,7 +80,9 @@ public static Builder builder() { @SchemaFieldNumber("3") public abstract long getFileSizeInBytes(); + /** @deprecated Use {@link #getJsonPartition()} instead. */ @SchemaFieldNumber("4") + @Deprecated public abstract String getPartitionPath(); @SchemaFieldNumber("5") @@ -110,6 +121,9 @@ public static Builder builder() { @SchemaFieldNumber("16") public abstract @Nullable Long getFirstRowId(); + @SchemaFieldNumber("17") + abstract @Nullable String getJsonPartition(); + @AutoValue.Builder public abstract static class Builder { abstract Builder setPath(String path); @@ -122,6 +136,8 @@ public abstract static class Builder { abstract Builder setPartitionPath(String partitionPath); + abstract Builder setJsonPartition(String jsonPartition); + abstract Builder setPartitionSpecId(int partitionSpec); abstract Builder setKeyMetadata(ByteBuffer keyMetadata); @@ -149,16 +165,38 @@ public abstract static class Builder { abstract SerializableDataFile build(); } - public static SerializableDataFile from(DataFile f, String partitionPath) { - return from(f, partitionPath, true); + public static SerializableDataFile from(DataFile f, Map specs) { + return from( + f, + checkStateNotNull( + specs.get(f.specId()), + "Could not create a SerializableDataFile because DataFile is written using a partition spec id '%s' that is not found in the provided specs: %s", + f.specId(), + specs.keySet()), + true); + } + + public static SerializableDataFile from(DataFile f, PartitionSpec spec) { + return from(f, spec, true); } /** * Create a {@link SerializableDataFile} from a {@link DataFile} and its associated {@link * PartitionKey}. */ - public static SerializableDataFile from( - DataFile f, String partitionPath, boolean includeMetrics) { + public static SerializableDataFile from(DataFile f, PartitionSpec spec, boolean includeMetrics) { + if (spec.specId() != f.specId()) { + throw new IllegalArgumentException( + String.format( + "Cannot serialize DataFile: its partition spec id %s does not match the provided " + + "spec id %s. Serialize the file with the exact spec it was written with.", + f.specId(), spec.specId())); + } + // jsonPartition is the primary (handles evolved specs, special characters). + // partitionPath is the fallback for values that don't round-trip through JSON. + String jsonPartition = SingleValueParser.toJson(spec.partitionType(), f.partition()); + String partitionPath = spec.partitionToPath(f.partition()); + SerializableDataFile.Builder builder = SerializableDataFile.builder() .setPath(f.location()) @@ -166,6 +204,7 @@ public static SerializableDataFile from( .setRecordCount(f.recordCount()) .setFileSizeInBytes(f.fileSizeInBytes()) .setPartitionPath(partitionPath) + .setJsonPartition(jsonPartition) .setPartitionSpecId(f.specId()) .setKeyMetadata(f.keyMetadata()) .setSplitOffsets(f.splitOffsets()) @@ -211,16 +250,36 @@ public DataFile createDataFile(Map partitionSpecs) { toByteBufferMap(getLowerBounds()), toByteBufferMap(getUpperBounds())); - return DataFiles.builder(partitionSpec) - .withFormat(FileFormat.fromString(getFileFormat())) - .withPath(getPath()) - .withPartitionPath(getPartitionPath()) - .withEncryptionKeyMetadata(getKeyMetadata()) - .withFileSizeInBytes(getFileSizeInBytes()) - .withMetrics(dataFileMetrics) - .withSplitOffsets(getSplitOffsets()) - .withFirstRowId(getFirstRowId()) - .build(); + DataFiles.Builder builder = + DataFiles.builder(partitionSpec) + .withFormat(FileFormat.fromString(getFileFormat())) + .withPath(getPath()) + .withEncryptionKeyMetadata(getKeyMetadata()) + .withFileSizeInBytes(getFileSizeInBytes()) + .withMetrics(dataFileMetrics) + .withSplitOffsets(getSplitOffsets()) + .withFirstRowId(getFirstRowId()); + + @Nullable String jsonPartition = getJsonPartition(); + if (jsonPartition != null) { + try { + builder = builder.withPartition(partition(partitionSpec)); + } catch (RuntimeException e) { + // Some partition values (e.g. NaN / Infinity floating-point) don't round-trip through the + // JSON representation; fall back to the partition-path string, which handles them. + builder = builder.withPartitionPath(getPartitionPath()); + } + } else { + // Elements decoded from a pre-jsonPartition release carry only the partition path. + builder = builder.withPartitionPath(getPartitionPath()); + } + return builder.build(); + } + + @VisibleForTesting + StructLike partition(PartitionSpec spec) { + return (StructLike) + SingleValueParser.fromJson(spec.partitionType(), checkStateNotNull(getJsonPartition())); } // ByteBuddyUtils has trouble converting Map value type ByteBuffer @@ -275,6 +334,7 @@ && getRecordCount() == that.getRecordCount() && getFileSizeInBytes() == that.getFileSizeInBytes() && getPartitionPath().equals(that.getPartitionPath()) && getPartitionSpecId() == that.getPartitionSpecId() + && Objects.equals(getJsonPartition(), that.getJsonPartition()) && Objects.equals(getKeyMetadata(), that.getKeyMetadata()) && Objects.equals(getSplitOffsets(), that.getSplitOffsets()) && Objects.equals(getColumnSizes(), that.getColumnSizes()) @@ -320,6 +380,7 @@ public final int hashCode() { getRecordCount(), getFileSizeInBytes(), getPartitionPath(), + getJsonPartition(), getPartitionSpecId(), getKeyMetadata(), getSplitOffsets(), diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java index ceb96d50f8aa..df6e3fceec7d 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; @@ -37,11 +38,14 @@ import org.apache.iceberg.FileMetadata; import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.SingleValueParser; import org.apache.iceberg.SortOrder; +import org.apache.iceberg.StructLike; import org.checkerframework.checker.nullness.qual.Nullable; @DefaultSchema(AutoValueSchema.class) @AutoValue +@Internal public abstract class SerializableDeleteFile { public static SerializableDeleteFile.Builder builder() { return new AutoValue_SerializableDeleteFile.Builder(); @@ -62,7 +66,9 @@ public static SerializableDeleteFile.Builder builder() { @SchemaFieldNumber("4") public abstract long getFileSizeInBytes(); + /** @deprecated Use {@link #getJsonPartition()} instead. */ @SchemaFieldNumber("5") + @Deprecated public abstract String getPartitionPath(); @SchemaFieldNumber("6") @@ -113,6 +119,9 @@ public static SerializableDeleteFile.Builder builder() { @SchemaFieldNumber("21") public abstract @Nullable Long getFileSequenceNumber(); + @SchemaFieldNumber("22") + abstract @Nullable String getJsonPartition(); + @AutoValue.Builder abstract static class Builder { abstract Builder setContentType(FileContent content); @@ -127,6 +136,8 @@ abstract static class Builder { abstract Builder setPartitionPath(String partitionPath); + abstract Builder setJsonPartition(String jsonPartition); + abstract Builder setPartitionSpecId(int partitionSpec); abstract Builder setSortOrderId(@Nullable Integer sortOrderId); @@ -163,7 +174,47 @@ abstract static class Builder { } public static SerializableDeleteFile from( - DeleteFile deleteFile, String partitionPath, boolean includeMetrics) { + DeleteFile deleteFile, Map specs) { + return from(deleteFile, specs, true); + } + + /** + * Creates a {@link SerializableDeleteFile}, resolving the file's {@link PartitionSpec} by its own + * spec id. + * + *

Delete files reached from a scan task may carry a spec id that differs from the spec of the + * data file they apply to, so the lookup has to be per delete file rather than against a single + * "current" spec. + */ + public static SerializableDeleteFile from( + DeleteFile deleteFile, Map specs, boolean includeMetrics) { + return from( + deleteFile, + checkStateNotNull( + specs.get(deleteFile.specId()), + "Could not create a SerializableDeleteFile because DeleteFile is written using a partition spec id '%s' that is not found in the provided specs: %s", + deleteFile.specId(), + specs.keySet()), + includeMetrics); + } + + public static SerializableDeleteFile from(DeleteFile deleteFile, PartitionSpec spec) { + return from(deleteFile, spec, true); + } + + public static SerializableDeleteFile from( + DeleteFile deleteFile, PartitionSpec spec, boolean includeMetrics) { + if (spec.specId() != deleteFile.specId()) { + throw new IllegalArgumentException( + String.format( + "Cannot serialize DeleteFile: its partition spec id %s does not match the provided " + + "spec id %s.", + deleteFile.specId(), spec.specId())); + } + // jsonPartition is the primary (handles evolved specs, special characters). + // partitionPath is the fallback for values that don't round-trip through JSON. + String jsonPartition = SingleValueParser.toJson(spec.partitionType(), deleteFile.partition()); + String partitionPath = spec.partitionToPath(deleteFile.partition()); SerializableDeleteFile.Builder builder = SerializableDeleteFile.builder() @@ -171,6 +222,7 @@ public static SerializableDeleteFile from( .setFileFormat(deleteFile.format().name()) .setFileSizeInBytes(deleteFile.fileSizeInBytes()) .setPartitionPath(partitionPath) + .setJsonPartition(jsonPartition) .setPartitionSpecId(deleteFile.specId()) .setRecordCount(deleteFile.recordCount()) .setColumnSizes(deleteFile.columnSizes()) @@ -228,7 +280,21 @@ public DeleteFile createDeleteFile( .withMetrics(metrics) .withSplitOffsets(getSplitOffsets()) .withEncryptionKeyMetadata(getKeyMetadata()) - .withPartitionPath(getPartitionPath()); + .withReferencedDataFile(getReferencedDataFile()); + + @Nullable String jsonPartition = getJsonPartition(); + if (jsonPartition != null) { + try { + deleteFileBuilder = deleteFileBuilder.withPartition(partition(partitionSpec)); + } catch (RuntimeException e) { + // Some partition values (e.g. NaN / Infinity floating-point) don't round-trip through the + // JSON representation; fall back to the partition-path string + deleteFileBuilder = deleteFileBuilder.withPartitionPath(getPartitionPath()); + } + } else { + // Elements decoded from a pre-jsonPartition release carry only the partition path. + deleteFileBuilder = deleteFileBuilder.withPartitionPath(getPartitionPath()); + } switch (getContentType()) { case POSITION_DELETES: @@ -260,17 +326,22 @@ public DeleteFile createDeleteFile( "Unexpected content type for DeleteFile: " + getContentType()); } - // needed for puffin files + // contentOffset / contentSizeInBytes really are Puffin-only: build() rejects a non-null value + // for either on any other format, and requires both (plus referencedDataFile) on Puffin. if (getFileFormat().equalsIgnoreCase(FileFormat.PUFFIN.name())) { deleteFileBuilder = deleteFileBuilder .withContentOffset(checkStateNotNull(getContentOffset())) - .withContentSizeInBytes(checkStateNotNull(getContentSizeInBytes())) - .withReferencedDataFile(checkStateNotNull(getReferencedDataFile())); + .withContentSizeInBytes(checkStateNotNull(getContentSizeInBytes())); } return deleteFileBuilder.build(); } + private StructLike partition(PartitionSpec spec) { + return (StructLike) + SingleValueParser.fromJson(spec.partitionType(), checkStateNotNull(getJsonPartition())); + } + @Override public final boolean equals(@Nullable Object o) { if (this == o) { @@ -287,6 +358,7 @@ && getRecordCount() == that.getRecordCount() && getFileSizeInBytes() == that.getFileSizeInBytes() && getPartitionPath().equals(that.getPartitionPath()) && getPartitionSpecId() == that.getPartitionSpecId() + && Objects.equals(getJsonPartition(), that.getJsonPartition()) && Objects.equals(getSortOrderId(), that.getSortOrderId()) && Objects.equals(getEqualityFieldIds(), that.getEqualityFieldIds()) && Objects.equals(getKeyMetadata(), that.getKeyMetadata()) @@ -314,6 +386,7 @@ public final int hashCode() { getRecordCount(), getFileSizeInBytes(), getPartitionPath(), + getJsonPartition(), getPartitionSpecId(), getSortOrderId(), getEqualityFieldIds(), diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java index d1a08980fa9d..338a2162080b 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java @@ -152,7 +152,8 @@ public void processElement( writer.close(); } - SerializableDataFile sdf = SerializableDataFile.from(writer.getDataFile(), partitionPath); + // Serialize against the file's own spec + SerializableDataFile sdf = SerializableDataFile.from(writer.getDataFile(), table.specs()); out.output( FileWriteResult.builder() .setTableIdentifier(destination.getTableIdentifier()) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java index 34f26eb9cdf9..b8c18d1a4e53 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java @@ -135,7 +135,7 @@ public static CloseableIterable createReader( outputSchema, checkStateNotNull(table.specs().get(task.getSpecId())), task.getDataFile().createDataFile(table.specs()), - task.getDataFile().getFileSequenceNumber(), + task.getDataFile().getDataSequenceNumber(), start, length, combined); diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java index 9b6955d9e4a5..97bcbaa5bec2 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java @@ -17,7 +17,6 @@ */ package org.apache.beam.sdk.io.iceberg.cdc; -import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.auto.value.AutoValue; @@ -118,8 +117,8 @@ public abstract static class Builder { abstract Builder setDataFile(SerializableDataFile dataFile); @SchemaIgnore - public Builder setDataFile(DataFile df, String partitionPath, boolean includeMetrics) { - return setDataFile(SerializableDataFile.from(df, partitionPath, includeMetrics)); + public Builder setDataFile(DataFile df, PartitionSpec spec, boolean includeMetrics) { + return setDataFile(SerializableDataFile.from(df, spec, includeMetrics)); } abstract Builder setExistingDeletes(List existingDeletes); @@ -159,10 +158,7 @@ public static SerializableChangelogTask from( .setOperation(task.operation()) .setOrdinal(task.changeOrdinal()) .setCommitSnapshotId(task.commitSnapshotId()) - .setDataFile( - contentScanTask.file(), - spec.partitionToPath(contentScanTask.partition()), - includeMetrics) + .setDataFile(contentScanTask.file(), spec, includeMetrics) .setSpecId(spec.specId()) .setStart(contentScanTask.start()) .setLength(contentScanTask.length()) @@ -270,13 +266,10 @@ static List getAddedDeleteFiles(ChangelogScanTask task) { private static List toSerializableDeletes( List dfs, Map specs, boolean includeMetrics) { + // Serialize each delete file against its own spec (looked up by its spec id): a delete file may + // carry a different spec id than the data file it applies to. return dfs.stream() - .map( - df -> - SerializableDeleteFile.from( - df, - checkStateNotNull(specs.get(df.specId())).partitionToPath(df.partition()), - includeMetrics)) + .map(df -> SerializableDeleteFile.from(df, specs, includeMetrics)) .collect(Collectors.toList()); } } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionUtilsTest.java index 740ede55811b..3aa6a5e94648 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionUtilsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionUtilsTest.java @@ -190,7 +190,7 @@ public void testConstantsMapIncludesCdcMetadataAndIdentityConstants() throws Exc .withRecordCount(2L) .withFirstRowId(99L) .build(); - setFileSequenceNumber(file, 42L); + setDataSequenceNumber(file, 42L); Map constants = PartitionUtils.constantsMap(spec, file, null); @@ -202,7 +202,7 @@ public void testConstantsMapIncludesCdcMetadataAndIdentityConstants() throws Exc } @Test - public void testConstantsMapUsesExplicitSequenceNumberWhenFileSequenceIsUnavailable() { + public void testConstantsMapUsesExplicitSequenceNumberWhenDataSequenceIsUnavailable() { org.apache.iceberg.Schema icebergSchema = new org.apache.iceberg.Schema( Types.NestedField.required(1, "id", Types.IntegerType.get()), @@ -223,12 +223,12 @@ public void testConstantsMapUsesExplicitSequenceNumberWhenFileSequenceIsUnavaila assertEquals("B", constants.get(2)); } - private static void setFileSequenceNumber(DataFile dataFile, long fileSequenceNumber) + private static void setDataSequenceNumber(DataFile dataFile, long dataSequenceNumber) throws Exception { - Method method = dataFile.getClass().getMethod("setFileSequenceNumber", Long.class); + Method method = dataFile.getClass().getMethod("setDataSequenceNumber", Long.class); method.setAccessible(true); try { - method.invoke(dataFile, fileSequenceNumber); + method.invoke(dataFile, dataSequenceNumber); } catch (InvocationTargetException e) { throw (Exception) e.getCause(); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java index 821fb2ac7b24..03b3560f746a 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java @@ -42,7 +42,6 @@ import java.time.LocalDateTime; import java.time.LocalTime; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; @@ -433,15 +432,8 @@ public void testSerializableDataFileRoundTripEquality() throws IOException { DataFile datafile = writer.getDataFile(); assertEquals(2L, datafile.recordCount()); - Map partitionFieldMap = new HashMap<>(); - for (PartitionField partitionField : PARTITION_SPEC.fields()) { - partitionFieldMap.put(partitionField.name(), partitionField); - } - - String partitionPath = - RecordWriterManager.getPartitionDataPath(partitionKey.toPath(), partitionFieldMap); DataFile roundTripDataFile = - SerializableDataFile.from(datafile, partitionPath) + SerializableDataFile.from(datafile, PARTITION_SPEC) .createDataFile(ImmutableMap.of(PARTITION_SPEC.specId(), PARTITION_SPEC)); checkDataFileEquality(datafile, roundTripDataFile); @@ -477,14 +469,8 @@ public void testRecreateSerializableDataAfterUpdatingPartitionSpec() throws IOEx writer.close(); // fetch data file and its serializable version - Map partitionFieldMap = new HashMap<>(); - for (PartitionField partitionField : PARTITION_SPEC.fields()) { - partitionFieldMap.put(partitionField.name(), partitionField); - } - String partitionPath = - RecordWriterManager.getPartitionDataPath(partitionKey.toPath(), partitionFieldMap); DataFile datafile = writer.getDataFile(); - SerializableDataFile serializableDataFile = SerializableDataFile.from(datafile, partitionPath); + SerializableDataFile serializableDataFile = SerializableDataFile.from(datafile, PARTITION_SPEC); assertEquals(2L, datafile.recordCount()); assertEquals(serializableDataFile.getPartitionSpecId(), datafile.specId()); @@ -645,7 +631,7 @@ public void testIdentityPartitioning() throws IOException { expectedPartitions.add(name + "=" + URLEncoder.encode(val, UTF_8.toString())); } String expectedPartitionPath = String.join("/", expectedPartitions); - assertEquals(expectedPartitionPath, dataFile.getPartitionPath()); + assertEquals(expectedPartitionPath, spec.partitionToPath(dataFile.partition(spec))); assertThat(dataFile.getPath(), containsString(expectedPartitionPath)); } @@ -698,9 +684,10 @@ public void testBucketPartitioning() throws IOException { assertEquals(1, files.size()); SerializableDataFile dataFile = files.get(0); assertEquals(1, dataFile.getRecordCount()); + String partitionPath = spec.partitionToPath(dataFile.partition(spec)); for (Schema.Field field : bucketSchema.getFields()) { String expectedPartition = field.getName() + "_bucket"; - assertThat(dataFile.getPartitionPath(), containsString(expectedPartition)); + assertThat(partitionPath, containsString(expectedPartition)); assertThat(dataFile.getPath(), containsString(expectedPartition)); } } @@ -792,6 +779,7 @@ public void testTimePartitioning() throws IOException { serializableDataFile.createDataFile( catalogConfig.catalog().loadTable(dest.getValue().getTableIdentifier()).specs()); assertThat(dataFile.path().toString(), containsString(expectedPartition)); + assertEquals(expectedPartition, spec.partitionToPath(dataFile.partition())); } @Rule public ExpectedException thrown = ExpectedException.none(); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java index 5126822c06f6..c8d7b4f09165 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java @@ -19,8 +19,12 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import java.lang.reflect.Method; +import java.math.BigDecimal; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -30,12 +34,17 @@ import java.util.Map; import java.util.Set; import java.util.stream.Collectors; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.apache.iceberg.DataFile; import org.apache.iceberg.DataFiles; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.types.Conversions; import org.apache.iceberg.types.Types; import org.junit.Test; @@ -52,6 +61,7 @@ public class SerializableDataFileTest { .add("recordCount") .add("fileSizeInBytes") .add("partitionPath") + .add("jsonPartition") .add("partitionSpecId") .add("keyMetadata") .add("splitOffsets") @@ -91,6 +101,36 @@ public void testFieldsInEqualsMethodInSyncWithGetterFields() { } } + /** + * B13/A6: every field is pinned with {@code @SchemaFieldNumber} because Dataflow's in-place + * {@code --update} rejects a reordered schema or a changed field nullability. Lock the field + * count, order/names, and the two nullability-sensitive fields so any future edit trips review. + */ + @Test + public void schemaFieldNumbersArePinned() throws Exception { + org.apache.beam.sdk.schemas.Schema schema = + SchemaRegistry.createDefault().getSchema(SerializableDataFile.class); + assertEquals(18, schema.getFieldCount()); + assertEquals("path", schema.getField(0).getName()); + assertEquals("fileFormat", schema.getField(1).getName()); + assertEquals("recordCount", schema.getField(2).getName()); + assertEquals("fileSizeInBytes", schema.getField(3).getName()); + assertEquals("partitionPath", schema.getField(4).getName()); + assertEquals("partitionSpecId", schema.getField(5).getName()); + assertEquals("keyMetadata", schema.getField(6).getName()); + assertEquals("splitOffsets", schema.getField(7).getName()); + assertEquals("columnSizes", schema.getField(8).getName()); + assertEquals("valueCounts", schema.getField(9).getName()); + assertEquals("nullValueCounts", schema.getField(10).getName()); + assertEquals("nanValueCounts", schema.getField(11).getName()); + assertEquals("lowerBounds", schema.getField(12).getName()); + assertEquals("upperBounds", schema.getField(13).getName()); + assertEquals("dataSequenceNumber", schema.getField(14).getName()); + assertEquals("fileSequenceNumber", schema.getField(15).getName()); + assertEquals("firstRowId", schema.getField(16).getName()); + assertEquals("jsonPartition", schema.getField(17).getName()); + } + /** * Bounds with {@code capacity > limit} must be copied by {@code [position, limit)}, not by {@link * ByteBuffer#array()}. Otherwise trailing 0x00 bytes leak into the manifest bounds and break @@ -126,7 +166,8 @@ public void testBoundByteBufferIsCopiedByLimitNotBackingArrayLength() { .withMetrics(metrics) .build(); - SerializableDataFile serialized = SerializableDataFile.from(dataFile, ""); + SerializableDataFile serialized = + SerializableDataFile.from(dataFile, PartitionSpec.unpartitioned()); byte[] serializedLower = serialized.getLowerBounds().get(columnId); byte[] serializedUpper = serialized.getUpperBounds().get(columnId); @@ -141,4 +182,230 @@ public void testBoundByteBufferIsCopiedByLimitNotBackingArrayLength() { assertArrayEquals(expectedLower, serializedLower); assertArrayEquals(expectedUpper, serializedUpper); } + + /** + * F8: {@code from(DataFile, spec)} must populate BOTH the JSON partition (primary) and the + * partition path (fallback), so the deprecated {@code partitionPath} schema field stays non-null + * across releases (Dataflow's in-place pipeline update rejects a changed field nullability). + */ + @Test + public void fromPopulatesBothPartitionRepresentations() { + DataFile dataFile = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withFormat(FileFormat.PARQUET) + .withPath("gs://test-bucket/data/f.parquet") + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .build(); + + SerializableDataFile sdf = SerializableDataFile.from(dataFile, PartitionSpec.unpartitioned()); + + assertEquals( + "partition path must be populated (unpartitioned -> empty string), not null", + "", + sdf.getPartitionPath()); + assertNotNull("json partition must also be populated", sdf.getJsonPartition()); + } + + /** + * F6: a {@link DataFile} must be serialized with the EXACT spec it was written with — a + * mismatched spec id (usually a spec evolution on a shared/refreshed table between writing and + * serializing) must fail loudly rather than silently encode the partition under the wrong field + * ids. + */ + @Test + public void fromRejectsSpecIdMismatch() { + DataFile unpartitioned = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withFormat(FileFormat.PARQUET) + .withPath("gs://test-bucket/data/f.parquet") + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .build(); + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + PartitionSpec otherSpec = PartitionSpec.builderFor(schema).identity("id").withSpecId(1).build(); + + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> SerializableDataFile.from(unpartitioned, otherSpec)); + assertTrue( + "message should explain the spec-id mismatch: " + ex.getMessage(), + ex.getMessage().contains("does not match")); + } + + /** + * B13/F8: elements encoded by a PRE-jsonPartition pipeline carry only {@code partitionPath} + * (field 4), with {@code jsonPartition} (field 14) null. {@code createDataFile} must reconstruct + * the partition via {@code withPartitionPath} rather than crash on the missing JSON. + */ + @Test + public void legacyElementWithoutJsonPartitionReconstructsViaPartitionPath() { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "shard", Types.IntegerType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("shard").build(); + + SerializableDataFile legacy = + SerializableDataFile.builder() + .setPath("gs://test-bucket/data/legacy.parquet") + .setFileFormat("PARQUET") + .setRecordCount(1L) + .setFileSizeInBytes(1L) + .setPartitionPath("shard=5") + .setPartitionSpecId(spec.specId()) + .build(); // no setJsonPartition -> jsonPartition is null (pre-upgrade encoding) + + DataFile reconstructed = legacy.createDataFile(ImmutableMap.of(spec.specId(), spec)); + assertEquals("shard=5", spec.partitionToPath(reconstructed.partition())); + } + + /** + * F7: NaN / Infinity floating-point partition values don't round-trip through the JSON partition + * representation ({@code SingleValueParser.fromJson} rejects the quoted {@code "NaN"}). + * Reconstruct must fall back to the partition-path string, which handles them, rather than + * crash-looping the sink at commit time. + */ + @Test + public void nanFloatPartitionReconstructsViaPathFallback() { + Schema schema = + new Schema( + Types.NestedField.required(1, "f", Types.FloatType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("f").build(); + GenericRecord partition = GenericRecord.create(spec.partitionType()); + partition.setField("f", Float.NaN); + DataFile dataFile = + DataFiles.builder(spec) + .withFormat(FileFormat.PARQUET) + .withPath("gs://test-bucket/data/nan.parquet") + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .withPartition(partition) + .build(); + + SerializableDataFile sdf = SerializableDataFile.from(dataFile, spec); + // Must reconstruct without throwing (JSON decode of NaN fails -> partition-path fallback). + DataFile reconstructed = sdf.createDataFile(ImmutableMap.of(spec.specId(), spec)); + + Object value = reconstructed.partition().get(0, Object.class); + assertTrue( + "partition value must round-trip as NaN", + value instanceof Float && Float.isNaN((Float) value)); + } + + /** + * DECIMAL and BINARY/FIXED partition values survive the JSON transport intact. + * + *

These are exactly the result types the path-rendered predecessor handled worst: {@code + * PartitionSpec.partitionToPath} renders BINARY/FIXED as base64, and {@code + * Conversions.fromPartitionString} reads that text back as its raw UTF-8 bytes — so a path + * round-trip returns different bytes than were written and never notices. (BINARY and FIXED were + * outright banned by the predecessor for that reason; DECIMAL survived by luck of {@code + * BigDecimal.toString}.) The typed JSON tuple is what the current transport actually uses, and it + * is exact for both, which is the fidelity claim this suite otherwise only asserts for strings + * and dates. + * + *

The comparison is on the reconstructed partition values themselves, not on the partition + * path, precisely because the path is the representation that loses them. + */ + @Test + public void decimalAndBinaryPartitionValuesRoundTripThroughJson() { + Schema schema = + new Schema( + Types.NestedField.required(1, "amount", Types.DecimalType.of(9, 3)), + Types.NestedField.required(2, "bin", Types.BinaryType.get()), + Types.NestedField.required(3, "fix", Types.FixedType.ofLength(4))); + PartitionSpec spec = + PartitionSpec.builderFor(schema).identity("amount").identity("bin").identity("fix").build(); + + BigDecimal amount = new BigDecimal("-12345.678"); + // Bytes chosen to be invalid UTF-8 and to contain a 0x00, so any text-shaped round-trip + // (base64-then-getBytes, or a path render) produces different bytes and this test notices. + byte[] bin = new byte[] {0x00, (byte) 0xFF, 0x2F, (byte) 0x80, 0x7E}; + byte[] fix = new byte[] {(byte) 0xDE, (byte) 0xAD, (byte) 0xBE, (byte) 0xEF}; + + GenericRecord partition = GenericRecord.create(spec.partitionType()); + partition.setField("amount", amount); + // Iceberg's in-memory java class for both BINARY and FIXED partition values is ByteBuffer. + partition.setField("bin", ByteBuffer.wrap(bin)); + partition.setField("fix", ByteBuffer.wrap(fix)); + DataFile dataFile = + DataFiles.builder(spec) + .withFormat(FileFormat.PARQUET) + .withPath("gs://test-bucket/data/decbin.parquet") + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .withPartition(partition) + .build(); + + DataFile reconstructed = + SerializableDataFile.from(dataFile, spec) + .createDataFile(ImmutableMap.of(spec.specId(), spec)); + StructLike rebuilt = reconstructed.partition(); + + assertEquals(amount, rebuilt.get(0, BigDecimal.class)); + assertArrayEquals(bin, toBytes(rebuilt.get(1, Object.class))); + assertArrayEquals(fix, toBytes(rebuilt.get(2, Object.class))); + } + + /** Iceberg represents BINARY as {@link ByteBuffer} and FIXED as {@code byte[]}. */ + private static byte[] toBytes(Object value) { + if (value instanceof ByteBuffer) { + ByteBuffer view = ((ByteBuffer) value).duplicate(); + byte[] bytes = new byte[view.remaining()]; + view.get(bytes); + return bytes; + } + return (byte[]) value; + } + + /** + * The cost of that fallback on a MULTI-field spec, pinned rather than described. + * + *

{@code SingleValueParser} decodes the partition struct as a unit, so one unrepresentable + * field sends the whole tuple through {@code fillFromPath}. The path is a lossy rendering: + * {@code PartitionSpec.partitionToPath} URL-encodes each value and nothing decodes it again, so + * the string field below comes back {@code "a%2Fb"} instead of {@code "a/b"} — the file is + * registered under a partition tuple it was never written with. + * + *

This is deliberately an assertion of the WRONG value. It is the known residual of the + * JSON-partition transport, logged by {@link SerializableDataFile#warnPartitionPathFallback}. A + * future per-field fallback would fix it, and this test is what would notice. + */ + @Test + public void nanInMultiFieldSpecDegradesTheOtherFieldsViaPathFallback() { + Schema schema = + new Schema( + Types.NestedField.required(1, "f", Types.DoubleType.get()), + Types.NestedField.required(2, "s", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("f").identity("s").build(); + GenericRecord partition = GenericRecord.create(spec.partitionType()); + partition.setField("f", Double.NaN); + partition.setField("s", "a/b"); + DataFile dataFile = + DataFiles.builder(spec) + .withFormat(FileFormat.PARQUET) + .withPath("gs://test-bucket/data/nan-multi.parquet") + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .withPartition(partition) + .build(); + + DataFile reconstructed = + SerializableDataFile.from(dataFile, spec) + .createDataFile(ImmutableMap.of(spec.specId(), spec)); + + // The value that forced the fallback survives it exactly... + Object f = reconstructed.partition().get(0, Object.class); + assertTrue( + "NaN must survive the path fallback", f instanceof Double && Double.isNaN((Double) f)); + // ...but its co-field does not: the '/' is still URL-encoded. + assertEquals("a%2Fb", reconstructed.partition().get(1, String.class)); + assertEquals("a/b", partition.getField("s")); // what it was written with + } } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java index 29ef30c97efb..02e6fec46578 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java @@ -19,6 +19,12 @@ import static java.util.Collections.emptyMap; import static java.util.Collections.singletonMap; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.not; +import static org.hamcrest.Matchers.nullValue; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; @@ -30,6 +36,9 @@ import java.util.Arrays; import java.util.HashMap; import java.util.Map; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.util.CoderUtils; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; @@ -37,10 +46,16 @@ import org.apache.iceberg.Metrics; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.SortOrder; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; /** Tests for {@link SerializableDeleteFile}. */ +@RunWith(JUnit4.class) public class SerializableDeleteFileTest { private static final org.apache.iceberg.Schema SCHEMA = new org.apache.iceberg.Schema( @@ -85,7 +100,7 @@ public void testPositionDeleteRoundTripPreservesMetadataUsedByCdcReads() throws .build(); setSequenceNumbers(deleteFile, 44L, 45L); - SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true); + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true); DeleteFile reconstructed = serialized.createDeleteFile( singletonMap(SPEC.specId(), SPEC), singletonMap(0, SortOrder.unsorted())); @@ -125,7 +140,7 @@ public void testEqualityDeleteRoundTripPreservesFieldIdsAndSortOrder() { .withRecordCount(2L) .build(); - SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true); + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true); DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), singletonMap(7, sortOrder)); @@ -149,7 +164,7 @@ public void testPuffinDeleteRoundTripPreservesDeletionVectorMetadata() { .withReferencedDataFile("gs://bucket/data/category=A/data.parquet") .build(); - SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true); + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true); DeleteFile reconstructed = serialized.createDeleteFile( singletonMap(SPEC.specId(), SPEC), singletonMap(0, SortOrder.unsorted())); @@ -160,9 +175,11 @@ public void testPuffinDeleteRoundTripPreservesDeletionVectorMetadata() { assertEquals("gs://bucket/data/category=A/data.parquet", reconstructed.referencedDataFile()); } + /** Reconstruction fails clearly when the spec map or the sort-order map lacks the file's id. */ @Test - public void testCreateDeleteFileFailsClearlyForMissingPartitionSpec() { - DeleteFile deleteFile = + public void testCreateDeleteFileFailsClearlyForMissingSpecOrSortOrder() { + // facet: missing partition spec. + DeleteFile positionDelete = FileMetadata.deleteFileBuilder(SPEC) .ofPositionDeletes() .withPath("gs://bucket/deletes/category=A/pos.parquet") @@ -171,19 +188,17 @@ public void testCreateDeleteFileFailsClearlyForMissingPartitionSpec() { .withFileSizeInBytes(256L) .withRecordCount(2L) .build(); - SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true); - - IllegalStateException thrown = + SerializableDeleteFile serializedPosition = + SerializableDeleteFile.from(positionDelete, SPEC, true); + IllegalStateException missingSpec = assertThrows( - IllegalStateException.class, () -> serialized.createDeleteFile(emptyMap(), null)); - - assertTrue(thrown.getMessage().contains("created with spec id '" + SPEC.specId() + "'")); - } + IllegalStateException.class, + () -> serializedPosition.createDeleteFile(emptyMap(), null)); + assertTrue(missingSpec.getMessage().contains("created with spec id '" + SPEC.specId() + "'")); - @Test - public void testCreateEqualityDeleteFileFailsClearlyForMissingSortOrder() { + // facet: missing sort order (equality delete). SortOrder sortOrder = SortOrder.builderFor(SCHEMA).asc("id").withOrderId(7).build(); - DeleteFile deleteFile = + DeleteFile equalityDelete = FileMetadata.deleteFileBuilder(SPEC) .ofEqualityDeletes(1) .withSortOrder(sortOrder) @@ -193,14 +208,242 @@ public void testCreateEqualityDeleteFileFailsClearlyForMissingSortOrder() { .withFileSizeInBytes(256L) .withRecordCount(2L) .build(); - SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true); + SerializableDeleteFile serializedEquality = + SerializableDeleteFile.from(equalityDelete, SPEC, true); + IllegalStateException missingOrder = + assertThrows( + IllegalStateException.class, + () -> + serializedEquality.createDeleteFile(singletonMap(SPEC.specId(), SPEC), emptyMap())); + assertTrue(missingOrder.getMessage().contains("sort order id '7'")); + } - IllegalStateException thrown = + /** + * A {@link DeleteFile} must be serialized with the EXACT spec it was written with: a mismatched + * spec id and a spec id absent from the map each fail loudly. + */ + @Test + public void fromRejectsMismatchedOrUnknownSpec() { + // facet: single-spec overload, wrong spec. + DeleteFile unpartitioned = + FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned()) + .ofPositionDeletes() + .withPath("gs://bucket/deletes/pos.parquet") + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(1L) + .withRecordCount(1L) + .build(); + PartitionSpec otherSpec = + PartitionSpec.builderFor(SCHEMA).identity("category").withSpecId(1).build(); + IllegalArgumentException mismatch = + assertThrows( + IllegalArgumentException.class, + () -> SerializableDeleteFile.from(unpartitioned, otherSpec)); + assertThat(mismatch.getMessage(), containsString("does not match")); + + // facet: spec-map overload, id missing from the map. + DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, "A")); + IllegalStateException unknown = assertThrows( IllegalStateException.class, - () -> serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), emptyMap())); + () -> SerializableDeleteFile.from(deleteFile, emptyMap(), true)); + assertThat(unknown.getMessage(), containsString("partition spec id '0'")); + } + + /** + * All three delete-file kinds — equality, position, and a V3 deletion vector — round-trip through + * the schema coder with their partition tuples intact. + */ + @Test + public void allDeleteFileKindsRoundTripThroughSchemaCoderWithPartition() throws Exception { + // facet: equality delete. + SortOrder sortOrder = SortOrder.builderFor(SCHEMA).asc("id").withOrderId(7).build(); + DeleteFile equalityDelete = + FileMetadata.deleteFileBuilder(SPEC) + .ofEqualityDeletes(1, 2) + .withSortOrder(sortOrder) + .withPath("gs://bucket/deletes/category=A/eq.parquet") + .withFormat(FileFormat.PARQUET) + .withPartition(partition(SPEC, "A")) + .withFileSizeInBytes(256L) + .withRecordCount(2L) + .build(); + DeleteFile equalityReconstructed = + encodeDecode(SerializableDeleteFile.from(equalityDelete, SPEC)) + .createDeleteFile(singletonMap(SPEC.specId(), SPEC), singletonMap(7, sortOrder)); + assertEquals(FileContent.EQUALITY_DELETES, equalityReconstructed.content()); + assertEquals(equalityDelete.partition(), equalityReconstructed.partition()); + assertEquals("category=A", SPEC.partitionToPath(equalityReconstructed.partition())); + + // facet: position delete. + DeleteFile positionDelete = positionDeletes(SPEC, partition(SPEC, "A")); + DeleteFile positionReconstructed = + encodeDecode(SerializableDeleteFile.from(positionDelete, SPEC)) + .createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + assertEquals(FileContent.POSITION_DELETES, positionReconstructed.content()); + assertEquals(positionDelete.partition(), positionReconstructed.partition()); + assertEquals("category=A", SPEC.partitionToPath(positionReconstructed.partition())); + + // facet: V3 deletion vector (a Puffin blob with offset/size/referenced-data-file). + DeleteFile dv = + FileMetadata.deleteFileBuilder(SPEC) + .ofPositionDeletes() + .withPath("gs://bucket/deletes/category=A/dv.puffin") + .withFormat(FileFormat.PUFFIN) + .withPartition(partition(SPEC, "A")) + .withFileSizeInBytes(512L) + .withRecordCount(1L) + .withContentOffset(64L) + .withContentSizeInBytes(128L) + .withReferencedDataFile("gs://bucket/data/category=A/data.parquet") + .build(); + DeleteFile dvReconstructed = + encodeDecode(SerializableDeleteFile.from(dv, SPEC)) + .createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + assertEquals(FileFormat.PUFFIN, dvReconstructed.format()); + assertEquals(Long.valueOf(64L), dvReconstructed.contentOffset()); + assertEquals(Long.valueOf(128L), dvReconstructed.contentSizeInBytes()); + assertEquals("gs://bucket/data/category=A/data.parquet", dvReconstructed.referencedDataFile()); + assertEquals(dv.partition(), dvReconstructed.partition()); + assertEquals("category=A", SPEC.partitionToPath(dvReconstructed.partition())); + } + + /** + * An identity partition on a {@code timestamptz} column: {@link + * org.apache.iceberg.types.Conversions#fromPartitionString} has no case for TIMESTAMP at all, so + * the old partition-path round-trip blew up at reconstruct time. The JSON representation carries + * the raw micros and round-trips exactly. + */ + @Test + public void timestampPartitionRoundTripsThroughJsonButNotThroughPartitionPath() { + org.apache.iceberg.Schema schema = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "event_time", Types.TimestampType.withZone())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("event_time").build(); + long micros = 1_709_618_828_000_009L; + GenericRecord partition = GenericRecord.create(spec.partitionType()); + partition.setField("event_time", micros); + DeleteFile deleteFile = positionDeletes(spec, partition); + + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, spec); + DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(spec.specId(), spec), null); + + assertEquals(Long.valueOf(micros), reconstructed.partition().get(0, Long.class)); + assertEquals(deleteFile.partition(), reconstructed.partition()); + + // The pre-change wire shape (partition path only) cannot reconstruct this partition at all. + SerializableDeleteFile legacy = withoutJsonPartition(serialized); + assertThrows( + UnsupportedOperationException.class, + () -> legacy.createDeleteFile(singletonMap(spec.specId(), spec), null)); + } + + /** + * {@code PartitionSpec.partitionToPath} URL-encodes each value but {@code DataFiles.fillFromPath} + * never decodes it, so a string partition containing {@code / }, {@code &} or {@code =} used to + * come back SILENTLY WRONG — the delete would be registered under a {@code (specId, partition)} + * that no data file lives in, and would simply never apply. The JSON representation is exact. + */ + @Test + public void stringPartitionWithSpecialCharactersRoundTripsExactly() { + String value = "a/b c&d=e"; + DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, value)); + + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC); + DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + + assertEquals(value, reconstructed.partition().get(0, CharSequence.class).toString()); + assertEquals(deleteFile.partition(), reconstructed.partition()); + + // Prove the old representation was silently lossy rather than merely throwing. + DeleteFile viaLegacyPath = + withoutJsonPartition(serialized).createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + assertThat( + viaLegacyPath.partition().get(0, CharSequence.class).toString(), not(equalTo(value))); + } + + /** + * A null partition value is rendered as the literal text {@code null} in a partition path, which + * {@code fromPartitionString} hands back as the four-character string "null" for a string column + * — again silently wrong. JSON omits the field and it decodes back to a real null. + */ + @Test + public void nullPartitionValueRoundTripsAsNull() { + DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, null)); + + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC); + DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + + assertThat(reconstructed.partition().get(0, CharSequence.class), nullValue()); + assertEquals(deleteFile.partition(), reconstructed.partition()); + + // The old path turns the null into the literal string "null". + DeleteFile viaLegacyPath = + withoutJsonPartition(serialized).createDeleteFile(singletonMap(SPEC.specId(), SPEC), null); + assertEquals("null", viaLegacyPath.partition().get(0, CharSequence.class).toString()); + } + + /** + * NaN / Infinity floating-point partition values don't round-trip through the JSON partition + * representation ({@code SingleValueParser.fromJson} rejects the non-standard {@code NaN} token). + * Reconstruct must fall back to the partition-path string, which handles them, rather than + * crash-looping the sink at commit time. + */ + @Test + public void nanFloatPartitionReconstructsViaPathFallback() { + org.apache.iceberg.Schema schema = + new org.apache.iceberg.Schema( + Types.NestedField.required(1, "f", Types.FloatType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("f").build(); + GenericRecord partition = GenericRecord.create(spec.partitionType()); + partition.setField("f", Float.NaN); + DeleteFile deleteFile = positionDeletes(spec, partition); + + SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, spec); + // Must reconstruct without throwing (JSON decode of NaN fails -> partition-path fallback). + DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(spec.specId(), spec), null); + + Object value = reconstructed.partition().get(0, Object.class); + assertThat(value, instanceOf(Float.class)); + assertTrue("partition value must round-trip as NaN", Float.isNaN((Float) value)); + } + + private static GenericRecord partition(PartitionSpec spec, @Nullable Object value) { + GenericRecord record = GenericRecord.create(spec.partitionType()); + record.set(0, value); + return record; + } + + private static DeleteFile positionDeletes(PartitionSpec spec, StructLike partition) { + return FileMetadata.deleteFileBuilder(spec) + .ofPositionDeletes() + .withPath("gs://bucket/deletes/pos.parquet") + .withFormat(FileFormat.PARQUET) + .withPartition(partition) + .withFileSizeInBytes(256L) + .withRecordCount(2L) + .build(); + } + + /** Rebuilds the element as a pre-jsonPartition release would have encoded it. */ + private static SerializableDeleteFile withoutJsonPartition(SerializableDeleteFile file) { + return SerializableDeleteFile.builder() + .setContentType(file.getContentType()) + .setLocation(file.getLocation()) + .setFileFormat(file.getFileFormat()) + .setRecordCount(file.getRecordCount()) + .setFileSizeInBytes(file.getFileSizeInBytes()) + .setPartitionPath(file.getPartitionPath()) + .setPartitionSpecId(file.getPartitionSpecId()) + .build(); + } - assertTrue(thrown.getMessage().contains("sort order id '7'")); + private static SerializableDeleteFile encodeDecode(SerializableDeleteFile file) throws Exception { + SchemaCoder coder = + SchemaRegistry.createDefault().getSchemaCoder(SerializableDeleteFile.class); + return CoderUtils.decodeFromByteArray(coder, CoderUtils.encodeToByteArray(coder, file)); } private static void setSequenceNumbers( diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java index 546386073ff6..8c23a844be6a 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java @@ -341,7 +341,7 @@ private static SerializableChangelogTask task( Table table) { return SerializableChangelogTask.builder() .setType(type) - .setDataFile(dataFile, table.spec().partitionToPath(dataFile.partition()), true) + .setDataFile(dataFile, table.spec(), true) .setAddedDeletes(serializableDeletes(addedDeletes, table)) .setExistingDeletes(serializableDeletes(existingDeletes, table)) .setSpecId(table.spec().specId()) @@ -360,10 +360,7 @@ private static SerializableChangelogTask task( private static List serializableDeletes( List deletes, Table table) { return deletes.stream() - .map( - delete -> - SerializableDeleteFile.from( - delete, table.spec().partitionToPath(delete.partition()), true)) + .map(delete -> SerializableDeleteFile.from(delete, table.specs(), true)) .collect(Collectors.toList()); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ChangelogScannerTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ChangelogScannerTest.java index 1e4b6ba58023..afb536278941 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ChangelogScannerTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ChangelogScannerTest.java @@ -338,7 +338,7 @@ private static SerializableChangelogTask serializableTask(String name, long leng .build(); return SerializableChangelogTask.builder() .setType(SerializableChangelogTask.Type.ADDED_ROWS) - .setDataFile(SerializableDataFile.from(file, "", false)) + .setDataFile(SerializableDataFile.from(file, PartitionSpec.unpartitioned(), false)) .setSpecId(UNPARTITIONED_SPEC.specId()) .setOperation(ChangelogOperation.INSERT) .setOrdinal(0) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/LocalResolveDoFnTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/LocalResolveDoFnTest.java index 7e2870353190..2dfbb854b374 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/LocalResolveDoFnTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/LocalResolveDoFnTest.java @@ -306,7 +306,7 @@ private static SerializableChangelogTask task( SerializableChangelogTask.Type type, DataFile dataFile, Table table, long snapshotId) { return SerializableChangelogTask.builder() .setType(type) - .setDataFile(dataFile, table.spec().partitionToPath(dataFile.partition()), true) + .setDataFile(dataFile, table.spec(), true) .setAddedDeletes(ImmutableList.of()) .setExistingDeletes(ImmutableList.of()) .setSpecId(table.spec().specId()) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java index 69591e6eaa7c..2de2848f9363 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java @@ -297,7 +297,7 @@ private static SerializableChangelogTask task( long snapshotId) { return SerializableChangelogTask.builder() .setType(type) - .setDataFile(dataFile, table.spec().partitionToPath(dataFile.partition()), true) + .setDataFile(dataFile, table.spec(), true) .setAddedDeletes(serializableDeletes(addedDeletes, table)) .setExistingDeletes(serializableDeletes(existingDeletes, table)) .setSpecId(table.spec().specId()) @@ -316,10 +316,7 @@ private static SerializableChangelogTask task( private static List serializableDeletes( List deletes, Table table) { return deletes.stream() - .map( - delete -> - SerializableDeleteFile.from( - delete, table.spec().partitionToPath(delete.partition()), true)) + .map(delete -> SerializableDeleteFile.from(delete, table.specs(), true)) .collect(Collectors.toList()); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTaskTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTaskTest.java index aa77d37af7af..32d0fc691c01 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTaskTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTaskTest.java @@ -81,7 +81,7 @@ public void coderRoundTripPreservesTaskBasics() throws Exception { SerializableChangelogTask task = SerializableChangelogTask.builder() .setType(SerializableChangelogTask.Type.ADDED_ROWS) - .setDataFile(SerializableDataFile.from(DATA_FILE, "", false)) + .setDataFile(SerializableDataFile.from(DATA_FILE, PartitionSpec.unpartitioned(), false)) .setSpecId(SPEC.specId()) .setOperation(ChangelogOperation.INSERT) .setOrdinal(7) From d83adefe2b2c75ff7792dd83c3c96279a0435e0c Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Wed, 2 Sep 2026 07:55:46 -0700 Subject: [PATCH 2/2] fix(iceberg): drop the Logger orphaned by the fallback-warning removal ErrorProne -Werror fails compileJava on the unused field, and the test javadoc linked the deleted method. Co-Authored-By: Claude Fable 5 --- .../org/apache/beam/sdk/io/iceberg/SerializableDataFile.java | 4 ---- .../apache/beam/sdk/io/iceberg/SerializableDataFileTest.java | 4 ++-- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java index aa3744fe4654..f53dfa197e08 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java @@ -42,8 +42,6 @@ import org.apache.iceberg.SingleValueParser; import org.apache.iceberg.StructLike; import org.checkerframework.checker.nullness.qual.Nullable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Serializable version of an Iceberg {@link DataFile}. @@ -62,8 +60,6 @@ @AutoValue @Internal public abstract class SerializableDataFile { - private static final Logger LOG = LoggerFactory.getLogger(SerializableDataFile.class); - public static Builder builder() { return new AutoValue_SerializableDataFile.Builder(); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java index c8d7b4f09165..709b6ce5a2dd 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java @@ -374,8 +374,8 @@ private static byte[] toBytes(Object value) { * registered under a partition tuple it was never written with. * *

This is deliberately an assertion of the WRONG value. It is the known residual of the - * JSON-partition transport, logged by {@link SerializableDataFile#warnPartitionPathFallback}. A - * future per-field fallback would fix it, and this test is what would notice. + * JSON-partition transport's path fallback. A future per-field fallback would fix it, and this + * test is what would notice. */ @Test public void nanInMultiFieldSpecDegradesTheOtherFieldsViaPathFallback() {