Flink: add append capability to dynamic iceberg sink (#14526)

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

@@ -31,6 +31,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.connector.sink2.Committer;
 import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.ManifestFile;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.RowDelta;
@@ -302,30 +303,56 @@ private void commitDeltaTxn(
       CommitSummary summary,
       String newFlinkJobId,
       String operatorId) {
-    for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
-      long checkpointId = e.getKey();
-      List<WriteResult> writeResults = e.getValue();
-
-      RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
-      for (WriteResult result : writeResults) {
-        // Row delta validations are not needed for streaming changes that write equality deletes.
-        // Equality deletes are applied to data in all previous sequence numbers, so retries may
-        // push deletes further in the future, but do not affect correctness. Position deletes
-        // committed to the table in this path are used only to delete rows from data files that are
-        // being added in this commit. There is no way for data files added along with the delete
-        // files to be concurrently removed, so there is no need to validate the files referenced by
-        // the position delete files that are being committed.
-        Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
-        Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
+    if (summary.deleteFilesCount() == 0) {
+      // Use append snapshot operation where possible
+      AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
+      for (List<WriteResult> resultList : pendingResults.values()) {
+        for (WriteResult result : resultList) {
+          Preconditions.checkState(
+              result.referencedDataFiles().length == 0,
+              "Should have no referenced data files for append.");
+          Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
+        }
       }
 
-      // Every Flink checkpoint contains a set of independent changes which can be committed
-      // together. While it is technically feasible to combine append-only data across checkpoints,
-      // for the sake of simplicity, we do not implement this (premature) optimization. Multiple
-      // pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
-      // intervals or when concurrent checkpointing is enabled.
       commitOperation(
-          table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
+          table,
+          branch,
+          appendFiles,
+          summary,
+          "append",
+          newFlinkJobId,
+          operatorId,
+          pendingResults.lastKey());
+    } else {
+      for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
+        long checkpointId = e.getKey();
+        List<WriteResult> writeResults = e.getValue();
+
+        RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
+        for (WriteResult result : writeResults) {
+          // Row delta validations are not needed for streaming changes that write equality deletes.
+          // Equality deletes are applied to data in all previous sequence numbers, so retries may
+          // push deletes further in the future, but do not affect correctness. Position deletes
+          // committed to the table in this path are used only to delete rows from data files that
+          // are
+          // being added in this commit. There is no way for data files added along with the delete
+          // files to be concurrently removed, so there is no need to validate the files referenced
+          // by
+          // the position delete files that are being committed.
+          Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
+          Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
+        }
+
+        // Every Flink checkpoint contains a set of independent changes which can be committed
+        // together. While it is technically feasible to combine append-only data across
+        // checkpoints,
+        // for the sake of simplicity, we do not implement this (premature) optimization. Multiple
+        // pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
+        // intervals or when concurrent checkpointing is enabled.
+        commitOperation(
+            table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
+      }
     }
   }
 

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java

@@ -591,6 +591,71 @@ void testTableBranchAtomicCommitWithFailures() throws Exception {
                 .build());
   }
 
+  @Test
+  void testCommitDeltaTxnWithAppendFiles() throws Exception {
+    Table table = catalog.loadTable(TableIdentifier.of(TABLE1));
+    assertThat(table.snapshots()).isEmpty();
+
+    DynamicWriteResultAggregator aggregator =
+        new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize);
+    OneInputStreamOperatorTestHarness aggregatorHarness =
+        new OneInputStreamOperatorTestHarness(aggregator);
+    aggregatorHarness.open();
+
+    WriteTarget writeTarget1 =
+        new WriteTarget(TABLE1, "branch1", 42, 0, true, Sets.newHashSet(1, 2));
+    WriteTarget writeTarget2 = new WriteTarget(TABLE1, "branch1", 23, 0, true, Sets.newHashSet());
+
+    WriteResult writeResult1 = WriteResult.builder().addDataFiles(DATA_FILE).build();
+    WriteResult writeResult2 = WriteResult.builder().addDataFiles(DATA_FILE_2).build();
+
+    final String jobId = JobID.generate().toHexString();
+    final String operatorId = new OperatorID().toHexString();
+    final int checkpointId = 1;
+
+    byte[] deltaManifest1 =
+        aggregator.writeToManifest(
+            writeTarget1,
+            Sets.newHashSet(new DynamicWriteResult(writeTarget1, writeResult1)),
+            checkpointId);
+
+    CommitRequest<DynamicCommittable> commitRequest1 =
+        new MockCommitRequest<>(
+            new DynamicCommittable(writeTarget1, deltaManifest1, jobId, operatorId, checkpointId));
+
+    byte[] deltaManifest2 =
+        aggregator.writeToManifest(
+            writeTarget2,
+            Sets.newHashSet(new DynamicWriteResult(writeTarget2, writeResult2)),
+            checkpointId);
+
+    CommitRequest<DynamicCommittable> commitRequest2 =
+        new MockCommitRequest<>(
+            new DynamicCommittable(writeTarget2, deltaManifest2, jobId, operatorId, checkpointId));
+
+    boolean overwriteMode = false;
+    int workerPoolSize = 1;
+    String sinkId = "sinkId";
+    UnregisteredMetricsGroup metricGroup = new UnregisteredMetricsGroup();
+    DynamicCommitterMetrics committerMetrics = new DynamicCommitterMetrics(metricGroup);
+    DynamicCommitter dynamicCommitter =
+        new DynamicCommitter(
+            CATALOG_EXTENSION.catalog(),
+            Maps.newHashMap(),
+            overwriteMode,
+            workerPoolSize,
+            sinkId,
+            committerMetrics);
+
+    dynamicCommitter.commit(Sets.newHashSet(commitRequest1, commitRequest2));
+
+    table.refresh();
+    assertThat(table.snapshots()).hasSize(1);
+
+    Snapshot snapshot = Iterables.getFirst(table.snapshots(), null);
+    assertThat(snapshot.operation()).isEqualTo("append");
+  }
+
   @Test
   void testReplacePartitions() throws Exception {
     Table table1 = catalog.loadTable(TableIdentifier.of(TABLE1));