This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 369409ea492 [IcebergIO] Serialize using json partition (#39705)
369409ea492 is described below
commit 369409ea4920e62ef332444b6684544d5b66ea07
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Thu Aug 13 13:59:22 2026 -0700
[IcebergIO] Serialize using json partition (#39705)
* use json partitioning instead
* trigger ITs
---
.../IO_Iceberg_Integration_Tests.json | 2 +-
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 2 +-
.../apache/beam/sdk/io/iceberg/PartitionUtils.java | 10 +-
.../beam/sdk/io/iceberg/RecordWriterManager.java | 9 +-
.../beam/sdk/io/iceberg/SerializableDataFile.java | 88 ++++++++++--
.../io/iceberg/WritePartitionedRowsToFiles.java | 3 +-
.../beam/sdk/io/iceberg/cdc/CdcReadUtils.java | 2 +-
.../io/iceberg/cdc/SerializableChangelogTask.java | 9 +-
.../beam/sdk/io/iceberg/PartitionUtilsTest.java | 10 +-
.../sdk/io/iceberg/RecordWriterManagerTest.java | 24 +---
.../sdk/io/iceberg/SerializableDataFileTest.java | 156 ++++++++++++++++++++-
.../beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java | 2 +-
.../sdk/io/iceberg/cdc/ChangelogScannerTest.java | 2 +-
.../sdk/io/iceberg/cdc/LocalResolveDoFnTest.java | 2 +-
.../sdk/io/iceberg/cdc/ReadFromChangelogsTest.java | 2 +-
.../iceberg/cdc/SerializableChangelogTaskTest.java | 2 +-
16 files changed, 259 insertions(+), 66 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index b73af5e61a4..7ab7bcd9a9c 100644
--- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
+++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run.",
- "modification": 1
+ "modification": 2
}
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 f37935f89e8..18b95ca50b1 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
.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 32a25439d85..6c8e79c4b8c 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 @@ class PartitionUtils {
* {@link ContentScanTask}s.
*/
public static Map<Integer, ?> 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 @@ class PartitionUtils {
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 6893c743f43..64f0ab6232a 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 RecordWriterManager implements AutoCloseable {
final Cache<PartitionKey, RecordWriter> writers;
private final List<SerializableDataFile> dataFiles = Lists.newArrayList();
@VisibleForTesting final Map<PartitionKey, Integer> writerCounts =
Maps.newHashMap();
- private final Map<String, PartitionField> partitionFieldMap =
Maps.newHashMap();
private final List<Exception> exceptions = Lists.newArrayList();
private final InternalRecordWrapper wrapper; // wrapper that facilitates
partitioning
@@ -115,9 +114,6 @@ class RecordWriterManager implements AutoCloseable {
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 RecordWriterManager implements AutoCloseable {
.expireAfterAccess(1, TimeUnit.MINUTES)
.removalListener(
(RemovalNotification<PartitionKey, RecordWriter> removal) ->
{
- final PartitionKey pk =
Preconditions.checkStateNotNull(removal.getKey());
final RecordWriter recordWriter =
Preconditions.checkStateNotNull(removal.getValue());
try {
@@ -144,9 +139,9 @@ class RecordWriterManager implements AutoCloseable {
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 e1291601d14..01d50010606 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.HashMap;
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,6 +39,8 @@ import org.apache.iceberg.FileFormat;
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;
/**
@@ -49,11 +53,12 @@ import org.checkerframework.checker.nullness.qual.Nullable;
* <p>NOTE: If you add any new fields here, you need to also update the {@link
#equals} and {@link
* #hashCode()} methods.
*
- * <p>Use {@link #from(DataFile, String)} to create a {@link
SerializableDataFile} and {@link
+ * <p>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 {
public static Builder builder() {
return new AutoValue_SerializableDataFile.Builder();
@@ -71,7 +76,9 @@ public abstract class SerializableDataFile {
@SchemaFieldNumber("3")
public abstract long getFileSizeInBytes();
+ /** @deprecated Use {@link #getJsonPartition()} instead. */
@SchemaFieldNumber("4")
+ @Deprecated
public abstract String getPartitionPath();
@SchemaFieldNumber("5")
@@ -110,6 +117,9 @@ public abstract class SerializableDataFile {
@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 +132,8 @@ public abstract class SerializableDataFile {
abstract Builder setPartitionPath(String partitionPath);
+ abstract Builder setJsonPartition(String jsonPartition);
+
abstract Builder setPartitionSpecId(int partitionSpec);
abstract Builder setKeyMetadata(ByteBuffer keyMetadata);
@@ -149,16 +161,38 @@ public abstract class SerializableDataFile {
abstract SerializableDataFile build();
}
- public static SerializableDataFile from(DataFile f, String partitionPath) {
- return from(f, partitionPath, true);
+ public static SerializableDataFile from(DataFile f, Map<Integer,
PartitionSpec> 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 +200,7 @@ public abstract class SerializableDataFile {
.setRecordCount(f.recordCount())
.setFileSizeInBytes(f.fileSizeInBytes())
.setPartitionPath(partitionPath)
+ .setJsonPartition(jsonPartition)
.setPartitionSpecId(f.specId())
.setKeyMetadata(f.keyMetadata())
.setSplitOffsets(f.splitOffsets())
@@ -211,16 +246,36 @@ public abstract class SerializableDataFile {
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 +330,8 @@ public abstract class SerializableDataFile {
&& getFileSizeInBytes() == that.getFileSizeInBytes()
&& getPartitionPath().equals(that.getPartitionPath())
&& getPartitionSpecId() == that.getPartitionSpecId()
+ && Objects.equals(getPartitionPath(), that.getPartitionPath())
+ && Objects.equals(getJsonPartition(), that.getJsonPartition())
&& Objects.equals(getKeyMetadata(), that.getKeyMetadata())
&& Objects.equals(getSplitOffsets(), that.getSplitOffsets())
&& Objects.equals(getColumnSizes(), that.getColumnSizes())
@@ -320,6 +377,7 @@ public abstract class SerializableDataFile {
getRecordCount(),
getFileSizeInBytes(),
getPartitionPath(),
+ getJsonPartition(),
getPartitionSpecId(),
getKeyMetadata(),
getSplitOffsets(),
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 d1a08980fa9..338a2162080 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 @@ class WritePartitionedRowsToFiles
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 34f26eb9cdf..b8c18d1a4e5 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 final class CdcReadUtils {
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 9b6955d9e4a..3410c0a9d7e 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
@@ -118,8 +118,8 @@ public abstract class SerializableChangelogTask {
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<SerializableDeleteFile>
existingDeletes);
@@ -159,10 +159,7 @@ public abstract class SerializableChangelogTask {
.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())
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 740ede55811..3aa6a5e9464 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 class PartitionUtilsTest {
.withRecordCount(2L)
.withFirstRowId(99L)
.build();
- setFileSequenceNumber(file, 42L);
+ setDataSequenceNumber(file, 42L);
Map<Integer, ?> constants = PartitionUtils.constantsMap(spec, file, null);
@@ -202,7 +202,7 @@ public class PartitionUtilsTest {
}
@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 class PartitionUtilsTest {
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 821fb2ac7b2..03b3560f746 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.LocalDate;
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 class RecordWriterManagerTest {
DataFile datafile = writer.getDataFile();
assertEquals(2L, datafile.recordCount());
- Map<String, PartitionField> 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 class RecordWriterManagerTest {
writer.close();
// fetch data file and its serializable version
- Map<String, PartitionField> 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 class RecordWriterManagerTest {
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 class RecordWriterManagerTest {
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 class RecordWriterManagerTest {
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 5126822c06f..e02cfaff084 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,6 +19,9 @@ package org.apache.beam.sdk.io.iceberg;
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.nio.ByteBuffer;
@@ -30,12 +33,16 @@ import java.util.List;
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.data.GenericRecord;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.junit.Test;
@@ -52,6 +59,7 @@ public class SerializableDataFileTest {
.add("recordCount")
.add("fileSizeInBytes")
.add("partitionPath")
+ .add("jsonPartition")
.add("partitionSpecId")
.add("keyMetadata")
.add("splitOffsets")
@@ -91,6 +99,36 @@ public class SerializableDataFileTest {
}
}
+ /**
+ * 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 +164,8 @@ public class SerializableDataFileTest {
.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 +180,119 @@ public class SerializableDataFileTest {
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));
+ }
}
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 546386073ff..75bfaa1775d 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 @@ public class CdcReadUtilsTest {
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())
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 1e4b6ba5802..afb53627894 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 @@ public class ChangelogScannerTest {
.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 7e287035319..2dfbb854b37 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 @@ public class LocalResolveDoFnTest {
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 69591e6eaa7..7a1e71d3d80 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 @@ public class ReadFromChangelogsTest {
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())
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 aa77d37af7a..32d0fc691c0 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 class SerializableChangelogTaskTest {
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)
