Flink: Refactor WriteResult aggregation in DynamicIcebergSink

core/src/main/java/org/apache/iceberg/io/WriteResult.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.util.CharSequenceSet;
@@ -134,4 +135,14 @@ public WriteResult build() {
       return new WriteResult(dataFiles, deleteFiles, referencedDataFiles, rewrittenDeleteFiles);
     }
   }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("dataFiles", dataFiles)
+        .add("deleteFiles", deleteFiles)
+        .add("referencedDataFiles", referencedDataFiles)
+        .add("rewrittenDeleteFiles", rewrittenDeleteFiles)
+        .toString();
+  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/CommitSummary.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.flink.sink;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.NavigableMap;
 import java.util.concurrent.atomic.AtomicLong;
@@ -44,7 +45,11 @@ public CommitSummary(NavigableMap<Long, WriteResult> pendingResults) {
   }
 
   public void addAll(NavigableMap<Long, List<WriteResult>> pendingResults) {
-    pendingResults.values().forEach(writeResults -> writeResults.forEach(this::addWriteResult));
+    pendingResults.values().forEach(this::addAll);
+  }
+
+  public void addAll(Collection<WriteResult> pendingResults) {
+    pendingResults.forEach(this::addWriteResult);
   }
 
   private void addWriteResult(WriteResult writeResult) {

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java

@@ -26,35 +26,35 @@
 
 /**
  * The aggregated results of a single checkpoint which should be committed. Containing the
- * serialized {@link DeltaManifests} file - which contains the commit data, and the jobId,
+ * serialized {@link DeltaManifests} files - which contains the commit data, and the jobId,
  * operatorId, checkpointId triplet to identify the specific commit.
  *
  * <p>{@link DynamicCommittableSerializer} is used to serialize {@link DynamicCommittable} between
  * the {@link DynamicWriter} and the {@link DynamicWriteResultAggregator}.
  */
 class DynamicCommittable implements Serializable {
 
-  private final WriteTarget key;
-  private final byte[] manifest;
+  private final TableKey key;
+  private final byte[][] manifests;
   private final String jobId;
   private final String operatorId;
   private final long checkpointId;
 
   DynamicCommittable(
-      WriteTarget key, byte[] manifest, String jobId, String operatorId, long checkpointId) {
+      TableKey key, byte[][] manifests, String jobId, String operatorId, long checkpointId) {
     this.key = key;
-    this.manifest = manifest;
+    this.manifests = manifests;
     this.jobId = jobId;
     this.operatorId = operatorId;
     this.checkpointId = checkpointId;
   }
 
-  WriteTarget key() {
+  TableKey key() {
     return key;
   }
 
-  byte[] manifest() {
-    return manifest;
+  byte[][] manifests() {
+    return manifests;
   }
 
   String jobId() {
@@ -78,14 +78,14 @@ public boolean equals(Object o) {
     DynamicCommittable that = (DynamicCommittable) o;
     return checkpointId == that.checkpointId
         && Objects.equals(key, that.key)
-        && Objects.deepEquals(manifest, that.manifest)
+        && Arrays.deepEquals(manifests, that.manifests)
         && Objects.equals(jobId, that.jobId)
         && Objects.equals(operatorId, that.operatorId);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(key, Arrays.hashCode(manifest), jobId, operatorId, checkpointId);
+    return Objects.hash(key, Arrays.deepHashCode(manifests), jobId, operatorId, checkpointId);
   }
 
   @Override
@@ -97,8 +97,4 @@ public String toString() {
         .add("operatorId", operatorId)
         .toString();
   }
-
-  public WriteTarget writeTarget() {
-    return key;
-  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java

@@ -31,11 +31,12 @@
  */
 class DynamicCommittableSerializer implements SimpleVersionedSerializer<DynamicCommittable> {
 
-  private static final int VERSION = 1;
+  private static final int VERSION_1 = 1;
+  private static final int VERSION_2 = 2;
 
   @Override
   public int getVersion() {
-    return VERSION;
+    return VERSION_2;
   }
 
   @Override
@@ -46,26 +47,61 @@ public byte[] serialize(DynamicCommittable committable) throws IOException {
     view.writeUTF(committable.jobId());
     view.writeUTF(committable.operatorId());
     view.writeLong(committable.checkpointId());
-    view.writeInt(committable.manifest().length);
-    view.write(committable.manifest());
+
+    int numManifests = committable.manifests().length;
+    view.writeInt(numManifests);
+    for (int i = 0; i < numManifests; i++) {
+      byte[] manifest = committable.manifests()[i];
+      view.writeInt(manifest.length);
+      view.write(manifest);
+    }
+
     return out.toByteArray();
   }
 
   @Override
   public DynamicCommittable deserialize(int version, byte[] serialized) throws IOException {
-    if (version == 1) {
-      DataInputDeserializer view = new DataInputDeserializer(serialized);
-      WriteTarget key = WriteTarget.deserializeFrom(view);
-      String jobId = view.readUTF();
-      String operatorId = view.readUTF();
-      long checkpointId = view.readLong();
-      int manifestLen = view.readInt();
-      byte[] manifestBuf;
-      manifestBuf = new byte[manifestLen];
-      view.read(manifestBuf);
-      return new DynamicCommittable(key, manifestBuf, jobId, operatorId, checkpointId);
+    if (version == VERSION_1) {
+      return deserializeV1(serialized);
+    } else if (version == VERSION_2) {
+      return deserializeV2(serialized);
     }
 
     throw new IOException("Unrecognized version or corrupt state: " + version);
   }
+
+  private DynamicCommittable deserializeV1(byte[] serialized) throws IOException {
+    DataInputDeserializer view = new DataInputDeserializer(serialized);
+    WriteTarget key = WriteTarget.deserializeFrom(view);
+    String jobId = view.readUTF();
+    String operatorId = view.readUTF();
+    long checkpointId = view.readLong();
+    int manifestLen = view.readInt();
+    byte[] manifestBuf;
+    manifestBuf = new byte[manifestLen];
+    view.read(manifestBuf);
+    return new DynamicCommittable(
+        new TableKey(key.tableName(), key.branch()),
+        new byte[][] {manifestBuf},
+        jobId,
+        operatorId,
+        checkpointId);
+  }
+
+  private DynamicCommittable deserializeV2(byte[] serialized) throws IOException {
+    DataInputDeserializer view = new DataInputDeserializer(serialized);
+    TableKey key = TableKey.deserializeFrom(view);
+    String jobId = view.readUTF();
+    String operatorId = view.readUTF();
+    long checkpointId = view.readLong();
+
+    byte[][] manifestsBuf = new byte[view.readInt()][];
+    for (int i = 0; i < manifestsBuf.length; i++) {
+      byte[] manifest = new byte[view.readInt()];
+      view.read(manifest);
+      manifestsBuf[i] = manifest;
+    }
+
+    return new DynamicCommittable(key, manifestsBuf, jobId, operatorId, checkpointId);
+  }
 }