Flink: Add dynamic writer and committer

This adds the dynamic version of the writer and committer for the Flink Dynamic
Iceberg Sink. Conceptually, they work similar to the IcebergSink, but they
support writing to multiple tables. Write results from each table are aggregated
from the DynamicWriter in the DynamicWriteResultAggregator, from where they are
sent to the DynamicCommitter.

Broken out of apache#12424, depends on apache#13032.

Author: mxm

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

@@ -19,13 +19,14 @@
 package org.apache.iceberg.flink.sink;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.NavigableMap;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.util.ScanTaskUtil;
 
-class CommitSummary {
+public class CommitSummary {
 
   private final AtomicLong dataFilesCount = new AtomicLong();
   private final AtomicLong dataFilesRecordCount = new AtomicLong();
@@ -34,30 +35,35 @@ class CommitSummary {
   private final AtomicLong deleteFilesRecordCount = new AtomicLong();
   private final AtomicLong deleteFilesByteCount = new AtomicLong();
 
-  CommitSummary(NavigableMap<Long, WriteResult> pendingResults) {
-    pendingResults
-        .values()
+  public CommitSummary() {}
+
+  public CommitSummary(NavigableMap<Long, WriteResult> pendingResults) {
+    pendingResults.values().forEach(this::addWriteResult);
+  }
+
+  public void addAll(NavigableMap<Long, List<WriteResult>> pendingResults) {
+    pendingResults.values().forEach(writeResults -> writeResults.forEach(this::addWriteResult));
+  }
+
+  private void addWriteResult(WriteResult writeResult) {
+    dataFilesCount.addAndGet(writeResult.dataFiles().length);
+    Arrays.stream(writeResult.dataFiles())
+        .forEach(
+            dataFile -> {
+              dataFilesRecordCount.addAndGet(dataFile.recordCount());
+              dataFilesByteCount.addAndGet(dataFile.fileSizeInBytes());
+            });
+    deleteFilesCount.addAndGet(writeResult.deleteFiles().length);
+    Arrays.stream(writeResult.deleteFiles())
         .forEach(
-            writeResult -> {
-              dataFilesCount.addAndGet(writeResult.dataFiles().length);
-              Arrays.stream(writeResult.dataFiles())
-                  .forEach(
-                      dataFile -> {
-                        dataFilesRecordCount.addAndGet(dataFile.recordCount());
-                        dataFilesByteCount.addAndGet(dataFile.fileSizeInBytes());
-                      });
-              deleteFilesCount.addAndGet(writeResult.deleteFiles().length);
-              Arrays.stream(writeResult.deleteFiles())
-                  .forEach(
-                      deleteFile -> {
-                        deleteFilesRecordCount.addAndGet(deleteFile.recordCount());
-                        long deleteBytes = ScanTaskUtil.contentSizeInBytes(deleteFile);
-                        deleteFilesByteCount.addAndGet(deleteBytes);
-                      });
+            deleteFile -> {
+              deleteFilesRecordCount.addAndGet(deleteFile.recordCount());
+              long deleteBytes = ScanTaskUtil.contentSizeInBytes(deleteFile);
+              deleteFilesByteCount.addAndGet(deleteBytes);
             });
   }
 
-  long dataFilesCount() {
+  public long dataFilesCount() {
     return dataFilesCount.get();
   }
 
@@ -69,7 +75,7 @@ long dataFilesByteCount() {
     return dataFilesByteCount.get();
   }
 
-  long deleteFilesCount() {
+  public long deleteFilesCount() {
     return deleteFilesCount.get();
   }
 

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

@@ -23,7 +23,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
-class DeltaManifests {
+public class DeltaManifests {
 
   private static final CharSequence[] EMPTY_REF_DATA_FILES = new CharSequence[0];
 
@@ -56,7 +56,7 @@ CharSequence[] referencedDataFiles() {
     return referencedDataFiles;
   }
 
-  List<ManifestFile> manifests() {
+  public List<ManifestFile> manifests() {
     List<ManifestFile> manifests = Lists.newArrayListWithCapacity(2);
     if (dataManifest != null) {
       manifests.add(dataManifest);

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

@@ -28,12 +28,12 @@
 import org.apache.iceberg.ManifestFiles;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
-class DeltaManifestsSerializer implements SimpleVersionedSerializer<DeltaManifests> {
+public class DeltaManifestsSerializer implements SimpleVersionedSerializer<DeltaManifests> {
   private static final int VERSION_1 = 1;
   private static final int VERSION_2 = 2;
   private static final byte[] EMPTY_BINARY = new byte[0];
 
-  static final DeltaManifestsSerializer INSTANCE = new DeltaManifestsSerializer();
+  public static final DeltaManifestsSerializer INSTANCE = new DeltaManifestsSerializer();
 
   @Override
   public int getVersion() {

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

@@ -38,7 +38,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class FlinkManifestUtil {
+public class FlinkManifestUtil {
 
   private static final Logger LOG = LoggerFactory.getLogger(FlinkManifestUtil.class);
   private static final int FORMAT_V2 = 2;
@@ -66,7 +66,7 @@ static List<DataFile> readDataFiles(
     }
   }
 
-  static ManifestOutputFileFactory createOutputFileFactory(
+  public static ManifestOutputFileFactory createOutputFileFactory(
       Supplier<Table> tableSupplier,
       Map<String, String> tableProps,
       String flinkJobId,
@@ -83,7 +83,7 @@ static ManifestOutputFileFactory createOutputFileFactory(
    * @param result all those DataFiles/DeleteFiles in this WriteResult should be written with same
    *     partition spec
    */
-  static DeltaManifests writeCompletedFiles(
+  public static DeltaManifests writeCompletedFiles(
       WriteResult result, Supplier<OutputFile> outputFileSupplier, PartitionSpec spec)
       throws IOException {
 
@@ -114,7 +114,7 @@ static DeltaManifests writeCompletedFiles(
     return new DeltaManifests(dataManifest, deleteManifest, result.referencedDataFiles());
   }
 
-  static WriteResult readCompletedFiles(
+  public static WriteResult readCompletedFiles(
       DeltaManifests deltaManifests, FileIO io, Map<Integer, PartitionSpec> specsById)
       throws IOException {
     WriteResult.Builder builder = WriteResult.builder();
@@ -135,7 +135,7 @@ static WriteResult readCompletedFiles(
     return builder.addReferencedDataFiles(deltaManifests.referencedDataFiles()).build();
   }
 
-  static void deleteCommittedManifests(
+  public static void deleteCommittedManifests(
       Table table, List<ManifestFile> manifests, String newFlinkJobId, long checkpointId) {
     for (ManifestFile manifest : manifests) {
       try {

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

@@ -31,7 +31,7 @@
  * <p>{@link IcebergCommittableSerializer} is used for serializing the objects between the Writer
  * and the Aggregator operator and between the Aggregator and the Committer as well.
  */
-class IcebergCommittable implements Serializable {
+public class IcebergCommittable implements Serializable {
   private final byte[] manifest;
   private final String jobId;
   private final String operatorId;

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

@@ -30,7 +30,7 @@
  *
  * <p>In both cases only the respective part is serialized.
  */
-class IcebergCommittableSerializer implements SimpleVersionedSerializer<IcebergCommittable> {
+public class IcebergCommittableSerializer implements SimpleVersionedSerializer<IcebergCommittable> {
   private static final int VERSION = 1;
 
   @Override

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

@@ -24,7 +24,7 @@
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.iceberg.flink.util.ElapsedTimeGauge;
 
-class IcebergFilesCommitterMetrics {
+public class IcebergFilesCommitterMetrics {
   private final AtomicLong lastCheckpointDurationMs = new AtomicLong();
   private final AtomicLong lastCommitDurationMs = new AtomicLong();
   private final ElapsedTimeGauge elapsedSecondsSinceLastSuccessfulCommit;
@@ -35,7 +35,7 @@ class IcebergFilesCommitterMetrics {
   private final Counter committedDeleteFilesRecordCount;
   private final Counter committedDeleteFilesByteCount;
 
-  IcebergFilesCommitterMetrics(MetricGroup metrics, String fullTableName) {
+  public IcebergFilesCommitterMetrics(MetricGroup metrics, String fullTableName) {
     MetricGroup committerMetrics =
         metrics.addGroup("IcebergFilesCommitter").addGroup("table", fullTableName);
     committerMetrics.gauge("lastCheckpointDurationMs", lastCheckpointDurationMs::get);
@@ -52,16 +52,16 @@ class IcebergFilesCommitterMetrics {
     this.committedDeleteFilesByteCount = committerMetrics.counter("committedDeleteFilesByteCount");
   }
 
-  void checkpointDuration(long checkpointDurationMs) {
+  public void checkpointDuration(long checkpointDurationMs) {
     lastCheckpointDurationMs.set(checkpointDurationMs);
   }
 
-  void commitDuration(long commitDurationMs) {
+  public void commitDuration(long commitDurationMs) {
     lastCommitDurationMs.set(commitDurationMs);
   }
 
   /** This is called upon a successful commit. */
-  void updateCommitSummary(CommitSummary stats) {
+  public void updateCommitSummary(CommitSummary stats) {
     elapsedSecondsSinceLastSuccessfulCommit.refreshLastRecordedTime();
     committedDataFilesCount.inc(stats.dataFilesCount());
     committedDataFilesRecordCount.inc(stats.dataFilesRecordCount());

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

@@ -28,7 +28,7 @@
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.util.ScanTaskUtil;
 
-class IcebergStreamWriterMetrics {
+public class IcebergStreamWriterMetrics {
   // 1,024 reservoir size should cost about 8KB, which is quite small.
   // It should also produce good accuracy for histogram distribution (like percentiles).
   private static final int HISTOGRAM_RESERVOIR_SIZE = 1024;
@@ -40,7 +40,7 @@ class IcebergStreamWriterMetrics {
   private final Histogram dataFilesSizeHistogram;
   private final Histogram deleteFilesSizeHistogram;
 
-  IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) {
+  public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) {
     MetricGroup writerMetrics =
         metrics.addGroup("IcebergStreamWriter").addGroup("table", fullTableName);
     this.flushedDataFiles = writerMetrics.counter("flushedDataFiles");
@@ -63,7 +63,7 @@ class IcebergStreamWriterMetrics {
             new DropwizardHistogramWrapper(dropwizardDeleteFilesSizeHistogram));
   }
 
-  void updateFlushResult(WriteResult result) {
+  public void updateFlushResult(WriteResult result) {
     flushedDataFiles.inc(result.dataFiles().length);
     flushedDeleteFiles.inc(result.deleteFiles().length);
     flushedReferencedDataFiles.inc(result.referencedDataFiles().length);
@@ -84,7 +84,7 @@ void updateFlushResult(WriteResult result) {
             });
   }
 
-  void flushDuration(long flushDurationMs) {
+  public void flushDuration(long flushDurationMs) {
     lastFlushDurationMs.set(flushDurationMs);
   }
 }

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

@@ -30,7 +30,7 @@
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Strings;
 
-class ManifestOutputFileFactory {
+public class ManifestOutputFileFactory {
   // Users could define their own flink manifests directory by setting this value in table
   // properties.
   @VisibleForTesting static final String FLINK_MANIFEST_LOCATION = "flink.manifests.location";
@@ -70,7 +70,7 @@ private String generatePath(long checkpointId) {
             fileCount.incrementAndGet()));
   }
 
-  OutputFile create(long checkpointId) {
+  public OutputFile create(long checkpointId) {
     String flinkManifestDir = props.get(FLINK_MANIFEST_LOCATION);
     TableOperations ops = ((HasTableOperations) tableSupplier.get()).operations();
 

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

@@ -80,6 +80,28 @@ public RowDataTaskWriterFactory(
       Map<String, String> writeProperties,
       List<Integer> equalityFieldIds,
       boolean upsert) {
+    this(
+        tableSupplier,
+        flinkSchema,
+        targetFileSizeBytes,
+        format,
+        writeProperties,
+        equalityFieldIds,
+        upsert,
+        tableSupplier.get().schema(),
+        tableSupplier.get().spec());
+  }
+
+  public RowDataTaskWriterFactory(
+      SerializableSupplier<Table> tableSupplier,
+      RowType flinkSchema,
+      long targetFileSizeBytes,
+      FileFormat format,
+      Map<String, String> writeProperties,
+      List<Integer> equalityFieldIds,
+      boolean upsert,
+      Schema schema,
+      PartitionSpec spec) {
     this.tableSupplier = tableSupplier;
 
     Table table;
@@ -90,9 +112,9 @@ public RowDataTaskWriterFactory(
       table = tableSupplier.get();
     }
 
-    this.schema = table.schema();
+    this.schema = schema;
     this.flinkSchema = flinkSchema;
-    this.spec = table.spec();
+    this.spec = spec;
     this.targetFileSizeBytes = targetFileSizeBytes;
     this.format = format;
     this.equalityFieldIds = equalityFieldIds;
@@ -148,6 +170,7 @@ public void initialize(int taskId, int attemptId) {
         OutputFileFactory.builderFor(table, taskId, attemptId)
             .format(format)
             .ioSupplier(() -> tableSupplier.get().io())
+            .defaultSpec(spec)
             .build();
   }
 

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

@@ -26,7 +26,7 @@
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.iceberg.io.WriteResult;
 
-class WriteResultSerializer implements SimpleVersionedSerializer<WriteResult> {
+public class WriteResultSerializer implements SimpleVersionedSerializer<WriteResult> {
   private static final int VERSION = 1;
 
   @Override