Flink: Fix same-checkpoint upserts after schema evolution

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

@@ -35,12 +35,10 @@
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.deletes.SortingPositionOnlyDeleteWriter;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
-import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 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;
-import org.apache.iceberg.util.StructLikeMap;
 import org.apache.iceberg.util.StructLikeUtil;
 import org.apache.iceberg.util.StructProjection;
 import org.apache.iceberg.util.Tasks;
@@ -183,7 +181,9 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {
     private RollingFileWriter dataWriter;
     private RollingEqDeleteWriter eqDeleteWriter;
     private PartitioningWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
-    private Map<StructLike, PathOffset> insertedRowMap;
+    private Map<StructLike, PositionDeleteTracker.PathOffset> insertedRowMap;
+    private final PositionDeleteTracker insertedRowTracker;
+    private final boolean closeInsertedRowTracker;
     private boolean closePosDeleteWriter;
 
     protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
@@ -204,6 +204,16 @@ protected BaseEqualityDeltaWriter(
         Schema deleteSchema,
         DeleteGranularity deleteGranularity,
         PartitioningDVWriter<T> posDeleteWriter) {
+      this(partition, schema, deleteSchema, deleteGranularity, posDeleteWriter, null);
+    }
+
+    protected BaseEqualityDeltaWriter(
+        StructLike partition,
+        Schema schema,
+        Schema deleteSchema,
+        DeleteGranularity deleteGranularity,
+        PartitioningDVWriter<T> posDeleteWriter,
+        PositionDeleteTracker insertedRowTracker) {
       Preconditions.checkNotNull(schema, "Iceberg table schema cannot be null.");
       Preconditions.checkNotNull(deleteSchema, "Equality-delete schema cannot be null.");
       this.structProjection = StructProjection.create(schema, deleteSchema);
@@ -215,7 +225,12 @@ protected BaseEqualityDeltaWriter(
           posDeleteWriter != null
               ? posDeleteWriter
               : createPosDeleteWriter(partition, deleteGranularity);
-      this.insertedRowMap = StructLikeMap.create(deleteSchema.asStruct());
+      this.insertedRowTracker =
+          insertedRowTracker != null
+              ? insertedRowTracker
+              : new PositionDeleteTracker(deleteSchema.asStruct());
+      this.closeInsertedRowTracker = insertedRowTracker == null;
+      this.insertedRowMap = this.insertedRowTracker.insertedRows();
       this.partitionKey = partition;
     }
 
@@ -226,13 +241,14 @@ protected BaseEqualityDeltaWriter(
     protected abstract StructLike asStructLikeKey(T key);
 
     public void write(T row) throws IOException {
-      PathOffset pathOffset = PathOffset.of(dataWriter.currentPath(), dataWriter.currentRows());
+      PositionDeleteTracker.PathOffset pathOffset =
+          PositionDeleteTracker.PathOffset.of(dataWriter.currentPath(), dataWriter.currentRows());
 
       // Create a copied key from this row.
       StructLike copiedKey = StructLikeUtil.copy(structProjection.wrap(asStructLike(row)));
 
       // Adding a pos-delete to replace the old path-offset.
-      PathOffset previous = insertedRowMap.put(copiedKey, pathOffset);
+      PositionDeleteTracker.PathOffset previous = insertedRowMap.put(copiedKey, pathOffset);
       if (previous != null) {
         // TODO attach the previous row if has a positional-delete row schema in appender factory.
         writePosDelete(previous);
@@ -260,8 +276,8 @@ private EncryptedOutputFile newOutputFile(StructLike partition) {
       }
     }
 
-    private void writePosDelete(PathOffset pathOffset) {
-      positionDelete.set(pathOffset.path, pathOffset.rowOffset);
+    private void writePosDelete(PositionDeleteTracker.PathOffset pathOffset) {
+      positionDelete.set(pathOffset.path(), pathOffset.rowOffset());
       posDeleteWriter.write(positionDelete, spec, partitionKey);
     }
 
@@ -271,7 +287,7 @@ private void writePosDelete(PathOffset pathOffset) {
      * @param key has the same columns with the equality fields.
      */
     private boolean internalPosDelete(StructLike key) {
-      PathOffset previous = insertedRowMap.remove(key);
+      PositionDeleteTracker.PathOffset previous = insertedRowMap.remove(key);
 
       if (previous != null) {
         // TODO attach the previous row if has a positional-delete row schema in appender factory.
@@ -327,8 +343,8 @@ public void close() throws IOException {
           }
         }
 
-        if (insertedRowMap != null) {
-          insertedRowMap.clear();
+        if (closeInsertedRowTracker) {
+          insertedRowTracker.clear();
           insertedRowMap = null;
         }
 
@@ -351,28 +367,6 @@ public void close() throws IOException {
     }
   }
 
-  private static class PathOffset {
-    private final CharSequence path;
-    private final long rowOffset;
-
-    private PathOffset(CharSequence path, long rowOffset) {
-      this.path = path;
-      this.rowOffset = rowOffset;
-    }
-
-    private static PathOffset of(CharSequence path, long rowOffset) {
-      return new PathOffset(path, rowOffset);
-    }
-
-    @Override
-    public String toString() {
-      return MoreObjects.toStringHelper(this)
-          .add("path", path)
-          .add("row_offset", rowOffset)
-          .toString();
-    }
-  }
-
   private abstract class BaseRollingWriter<W extends Closeable> implements Closeable {
     private static final int ROWS_DIVISOR = 1000;
     private final StructLike partitionKey;

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.io;
+
+import java.util.Map;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeMap;
+
+/** Tracks data-file positions for equality keys written by related task writers. */
+public final class PositionDeleteTracker {
+  private final Map<StructLike, PathOffset> insertedRowMap;
+
+  /** Creates a tracker for the equality-key struct type. */
+  public PositionDeleteTracker(Types.StructType keyType) {
+    this.insertedRowMap =
+        StructLikeMap.create(Preconditions.checkNotNull(keyType, "Key type cannot be null"));
+  }
+
+  Map<StructLike, PathOffset> insertedRows() {
+    return insertedRowMap;
+  }
+
+  /** Clears all tracked rows. */
+  public void clear() {
+    insertedRowMap.clear();
+  }
+
+  static final class PathOffset {
+    private final CharSequence path;
+    private final long rowOffset;
+
+    private PathOffset(CharSequence path, long rowOffset) {
+      this.path = path;
+      this.rowOffset = rowOffset;
+    }
+
+    CharSequence path() {
+      return path;
+    }
+
+    long rowOffset() {
+      return rowOffset;
+    }
+
+    static PathOffset of(CharSequence path, long rowOffset) {
+      return new PathOffset(path, rowOffset);
+    }
+  }
+}

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java

@@ -36,6 +36,7 @@
 import org.apache.iceberg.io.FileWriterFactory;
 import org.apache.iceberg.io.OutputFileFactory;
 import org.apache.iceberg.io.PartitioningDVWriter;
+import org.apache.iceberg.io.PositionDeleteTracker;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.TypeUtil;
 
@@ -47,6 +48,7 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
   private final RowDataWrapper keyWrapper;
   private final RowDataProjection keyProjection;
   private final boolean upsert;
+  private final PositionDeleteTracker positionDeleteTracker;
 
   BaseDeltaTaskWriter(
       PartitionSpec spec,
@@ -59,7 +61,8 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
       RowType flinkSchema,
       Set<Integer> equalityFieldIds,
       boolean upsert,
-      boolean useDv) {
+      boolean useDv,
+      PositionDeleteTracker positionDeleteTracker) {
     super(spec, format, fileWriterFactory, fileFactory, io, targetFileSize, useDv);
     this.schema = schema;
     this.deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
@@ -69,6 +72,7 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
     this.keyProjection =
         RowDataProjection.create(flinkSchema, schema.asStruct(), deleteSchema.asStruct());
     this.upsert = upsert;
+    this.positionDeleteTracker = positionDeleteTracker;
   }
 
   abstract RowDataDeltaWriter route(RowData row);
@@ -112,7 +116,13 @@ public void write(RowData row) throws IOException {
 
   protected class RowDataDeltaWriter extends BaseEqualityDeltaWriter {
     RowDataDeltaWriter(PartitionKey partition, PartitioningDVWriter<RowData> dvFileWriter) {
-      super(partition, schema, deleteSchema, DeleteGranularity.FILE, dvFileWriter);
+      super(
+          partition,
+          schema,
+          deleteSchema,
+          DeleteGranularity.FILE,
+          dvFileWriter,
+          positionDeleteTracker);
     }
 
     @Override

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/PartitionedDeltaWriter.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.FileWriterFactory;
 import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.PositionDeleteTracker;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.Tasks;
 
@@ -51,7 +52,8 @@ class PartitionedDeltaWriter extends BaseDeltaTaskWriter {
       RowType flinkSchema,
       Set<Integer> equalityFieldIds,
       boolean upsert,
-      boolean useDv) {
+      boolean useDv,
+      PositionDeleteTracker positionDeleteTracker) {
     super(
         spec,
         format,
@@ -63,7 +65,8 @@ class PartitionedDeltaWriter extends BaseDeltaTaskWriter {
         flinkSchema,
         equalityFieldIds,
         upsert,
-        useDv);
+        useDv,
+        positionDeleteTracker);
     this.partitionKey = new PartitionKey(spec, schema);
   }
 

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

@@ -35,6 +35,7 @@
 import org.apache.iceberg.io.FileWriterFactory;
 import org.apache.iceberg.io.OutputFileFactory;
 import org.apache.iceberg.io.PartitionedFanoutWriter;
+import org.apache.iceberg.io.PositionDeleteTracker;
 import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.UnpartitionedWriter;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -184,6 +185,10 @@ public void initialize(int taskId, int attemptId) {
 
   @Override
   public TaskWriter<RowData> create() {
+    return create(null);
+  }
+
+  public TaskWriter<RowData> create(PositionDeleteTracker positionDeleteTracker) {
     Preconditions.checkNotNull(
         outputFileFactory,
         "The outputFileFactory shouldn't be null if we have invoked the initialize().");
@@ -225,7 +230,8 @@ public TaskWriter<RowData> create() {
             flinkSchema,
             equalityFieldIds,
             upsert,
-            useDv);
+            useDv,
+            positionDeleteTracker);
       } else {
         return new PartitionedDeltaWriter(
             spec,
@@ -238,7 +244,8 @@ public TaskWriter<RowData> create() {
             flinkSchema,
             equalityFieldIds,
             upsert,
-            useDv);
+            useDv,
+            positionDeleteTracker);
       }
     }
   }

flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/UnpartitionedDeltaWriter.java

@@ -28,6 +28,7 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.FileWriterFactory;
 import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.PositionDeleteTracker;
 
 class UnpartitionedDeltaWriter extends BaseDeltaTaskWriter {
   private final RowDataDeltaWriter writer;
@@ -43,7 +44,8 @@ class UnpartitionedDeltaWriter extends BaseDeltaTaskWriter {
       RowType flinkSchema,
       Set<Integer> equalityFieldIds,
       boolean upsert,
-      boolean useDv) {
+      boolean useDv,
+      PositionDeleteTracker positionDeleteTracker) {
     super(
         spec,
         format,
@@ -55,7 +57,8 @@ class UnpartitionedDeltaWriter extends BaseDeltaTaskWriter {
         flinkSchema,
         equalityFieldIds,
         upsert,
-        useDv);
+        useDv,
+        positionDeleteTracker);
     this.writer = new RowDataDeltaWriter(null, dvFileWriter());
   }