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..779bc3753c7d 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,11 @@ 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 +121,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 +138,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 +176,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 +224,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 +282,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 +328,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 +360,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 +388,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/cdc/SerializableChangelogTask.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java index 3410c0a9d7ee..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; @@ -267,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/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 75bfaa1775da..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 @@ -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/ReadFromChangelogsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java index 7a1e71d3d80b..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 @@ -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()); }