mxm commented on code in PR #18675:
URL: https://github.com/apache/iceberg/pull/18675#discussion_r4400983541


##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java:
##########
@@ -81,83 +90,157 @@ class DynamicWriter implements 
CommittingSinkWriter<DynamicRecordInternal, Dynam
     this.attemptId = attemptId;
     this.taskWriterFactories = new LRUCache<>(cacheMaximumSize);
     this.writers = Maps.newHashMap();
+    this.insertedRowTrackers = Maps.newHashMap();
 
     LOG.debug("DynamicIcebergSinkWriter created for subtask {} attemptId {}", 
subTaskId, attemptId);
   }
 
   @Override
   public void write(DynamicRecordInternal element, Context context)
       throws IOException, InterruptedException {
-    writers
-        .computeIfAbsent(
-            new WriteTarget(
-                element.tableName(),
-                element.branch(),
-                element.schema().schemaId(),
-                element.spec().specId(),
-                element.upsertMode(),
-                element.equalityFields()),
-            writerKey -> {
-              RowDataTaskWriterFactory taskWriterFactory =
-                  taskWriterFactories.computeIfAbsent(
-                      writerKey,
-                      factoryKey -> {
-                        Table table =
-                            
catalog.loadTable(TableIdentifier.parse(factoryKey.tableName()));
-
-                        Set<Integer> equalityFieldIds =
-                            getEqualityFields(table, element.equalityFields());
-                        if (element.upsertMode()) {
-                          Preconditions.checkState(
-                              !equalityFieldIds.isEmpty(),
-                              "Equality field columns shouldn't be empty when 
configuring to use UPSERT data.");
-
-                          if (!table.spec().isUnpartitioned()) {
-                            for (PartitionField partitionField : 
table.spec().fields()) {
-                              Preconditions.checkState(
-                                  
equalityFieldIds.contains(partitionField.sourceId()),
-                                  "In UPSERT mode, partition field '%s' should 
be included in equality fields: '%s'",
-                                  partitionField,
-                                  equalityFieldIds);
-                            }
-                          }
-                        }
-
-                        FlinkWriteConf flinkWriteConf =
-                            new FlinkWriteConf(table, commonWriteProperties, 
flinkConfig);
-                        Map<String, String> tableWriteProperties =
-                            SinkUtil.writeProperties(
-                                flinkWriteConf.dataFileFormat(), 
flinkWriteConf, table);
-
-                        LOG.debug("Creating new writer factory for table 
'{}'", table.name());
-                        return new RowDataTaskWriterFactory(
-                            () -> table,
-                            FlinkSchemaUtil.convert(element.schema()),
-                            flinkWriteConf.targetDataFileSize(),
-                            flinkWriteConf.dataFileFormat(),
-                            tableWriteProperties,
-                            Lists.newArrayList(equalityFieldIds),
-                            element.upsertMode(),
-                            element.schema(),
-                            element.spec());
-                      });
-
-              taskWriterFactory.initialize(subTaskId, attemptId);
-              return taskWriterFactory.create();
-            })
-        .write(element.rowData());
+    WriteTarget writeTarget =
+        new WriteTarget(
+            element.tableName(),
+            element.branch(),
+            element.schema().schemaId(),
+            element.spec().specId(),
+            element.upsertMode(),
+            element.equalityFields());
+
+    TaskWriter<RowData> writer = writers.get(writeTarget);
+    if (writer == null) {
+      warnIfSplitsTable(writeTarget);

Review Comment:
   Do we need `warnIfSplitsTable`? It logs a WARN whenever a table gets a 
second writer within a checkpoint, including append-only tables, where the 
message isn't helpful. If not, we could keep the existing `computeIfAbsent`.



##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java:
##########
@@ -326,6 +342,446 @@ void testUniqueFileSuffixOnFactoryRecreation() throws 
Exception {
     assertThat(firstFile.getName()).isNotEqualTo(secondFile.getName());
   }
 
+  @Test
+  void testUpsertOfSameKeyAcrossSchemaEvolutionWithinCheckpoint() throws 
Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"before")), null);
+
+    // A column added mid-checkpoint gives the table a new schema ID, so the 
next record for the
+    // same table opens a second writer. Its re-write of key 1 must retire the 
first writer's row.
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(upsertRecord(table, schemaAfter, rowData(1, "after", 
"x")), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).hasSize(2);
+    assertThat(deleteFiles(results))
+        .filteredOn(file -> file.content() == FileContent.POSITION_DELETES)
+        .hasSize(1);
+    assertThat(deleteFiles(results))
+        .as("Only the first writer's insert of a key it had not seen emits an 
equality delete")
+        .filteredOn(file -> file.content() == FileContent.EQUALITY_DELETES)
+        .hasSize(1);
+
+    DynamicWriteResult firstWriter = resultWithDelete(results, 
FileContent.EQUALITY_DELETES);
+    DynamicWriteResult secondWriter = resultWithDelete(results, 
FileContent.POSITION_DELETES);
+    assertThat(secondWriter.writeResult().referencedDataFiles())
+        .as("Second writer's position delete must reference the first writer's 
data file")
+        .containsExactly(firstWriter.writeResult().dataFiles()[0].location());
+
+    commit(table, results);
+    assertTableRows(table, SimpleDataUtil.createRecord(1, "after", "x"));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testUpsertAlternatingBetweenSchemaVersionsWithinCheckpoint() throws 
Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"100")), null);
+
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(upsertRecord(table, schemaAfter, rowData(1, "200", 
"x")), null);
+
+    // An input still shaped like the old schema resolves to the old table 
schema, so traffic can
+    // return to the first writer after the second one exists. The newest 
write must win.
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"300")), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).hasSize(2);
+    assertThat(deleteFiles(results))
+        .as("Each writer retires the other writer's row with a position 
delete")
+        .filteredOn(file -> file.content() == FileContent.POSITION_DELETES)
+        .hasSize(2);
+
+    commit(table, results);
+    assertTableRows(table, SimpleDataUtil.createRecord(1, "300", null));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testUpsertOfDistinctKeysAcrossSchemaEvolutionWithinCheckpoint() throws 
Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"one")), null);
+
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(upsertRecord(table, schemaAfter, rowData(2, "two", 
"x")), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).hasSize(2);
+    assertThat(deleteFiles(results))
+        .as("Distinct keys never hit the shared tracker")
+        .noneMatch(file -> file.content() == FileContent.POSITION_DELETES);
+
+    commit(table, results);
+    assertTableRows(
+        table,
+        SimpleDataUtil.createRecord(1, "one", null),
+        SimpleDataUtil.createRecord(2, "two", "x"));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testUpsertOfSameKeyAcrossSchemaEvolutionWithinCheckpointPartitioned() 
throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    PartitionSpec spec = 
PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("id").build();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA, spec);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"before")), null);
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(2, 
"other")), null);
+
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(upsertRecord(table, schemaAfter, rowData(1, "after", 
"x")), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(deleteFiles(results))
+        .filteredOn(file -> file.content() == FileContent.POSITION_DELETES)
+        .hasSize(1);
+
+    commit(table, results);
+    assertTableRows(
+        table,
+        SimpleDataUtil.createRecord(1, "after", "x"),
+        SimpleDataUtil.createRecord(2, "other", null));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testUpsertAcrossEqualityFieldTypePromotionFallsBackToPrivateTracker() 
throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"before")), null);
+
+    // Promoting the equality field changes the tracker key type, so the 
writers cannot share a
+    // tracker. The second writer must still work, with a private tracker; the 
re-write is then
+    // retired by an equality delete, which cannot apply within this commit 
(documented limitation).
+    table.updateSchema().updateColumn("id", Types.LongType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaAfter, GenericRowData.of(1L, 
StringData.fromString("after"))),
+        null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).hasSize(2);
+    assertThat(deleteFiles(results))
+        .as("Each writer emits its own insert-time equality delete and no 
position delete")
+        .hasSize(2)
+        .allMatch(file -> file.content() == FileContent.EQUALITY_DELETES);
+
+    commit(table, results);
+    Record before = GenericRecord.create(table.schema());
+    before.setField("id", 1L);
+    before.setField("data", "before");
+    Record after = GenericRecord.create(table.schema());
+    after.setField("id", 1L);
+    after.setField("data", "after");
+    assertTableRows(table, before, after);
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testUpsertOfSameKeyAcrossColumnDocChangeWithinCheckpoint() throws 
Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"before")), null);
+
+    // The sink evolves column docs on its own; a doc change on the key column 
must not stop the
+    // writers from sharing a tracker, since key values compare by type, not 
by name or doc.
+    table.updateSchema().updateColumnDoc("id", "primary key").commit();
+    Schema schemaAfter = table.schema();
+    assertThat(schemaAfter.schemaId()).isNotEqualTo(schemaBefore.schemaId());
+    dynamicWriter.write(
+        upsertRecord(table, schemaAfter, SimpleDataUtil.createRowData(1, 
"after")), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(deleteFiles(results))
+        .filteredOn(file -> file.content() == FileContent.POSITION_DELETES)
+        .hasSize(1);
+
+    commit(table, results);
+    assertTableRows(table, SimpleDataUtil.createRecord(1, "after"));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testDeleteThroughSecondWriterWithinCheckpoint() throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        upsertRecord(table, schemaBefore, SimpleDataUtil.createRowData(1, 
"before")), null);
+
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    RowData delete =
+        GenericRowData.ofKind(
+            RowKind.DELETE, 1, StringData.fromString("before"), 
StringData.fromString("x"));
+    dynamicWriter.write(upsertRecord(table, schemaAfter, delete), null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).as("The delete writer produces no data 
file").hasSize(1);
+    assertThat(deleteFiles(results))
+        .filteredOn(file -> file.content() == FileContent.POSITION_DELETES)
+        .hasSize(1);
+
+    commit(table, results);
+    assertTableRows(table);
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testCdcUpdateAcrossSchemaEvolutionWithinCheckpoint() throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    Schema schemaBefore = table.schema();
+    dynamicWriter.write(
+        cdcRecord(table, schemaBefore, SimpleDataUtil.createInsert(1, 
"before")), null);
+
+    // Without upsert mode an update arrives as UPDATE_BEFORE + UPDATE_AFTER. 
The UPDATE_BEFORE
+    // through the second writer must retire the first writer's row with a 
position delete.
+    table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    Schema schemaAfter = table.schema();
+    dynamicWriter.write(
+        cdcRecord(
+            table,
+            schemaAfter,
+            GenericRowData.ofKind(RowKind.UPDATE_BEFORE, 1, 
StringData.fromString("before"), null)),
+        null);
+    dynamicWriter.write(
+        cdcRecord(
+            table,
+            schemaAfter,
+            GenericRowData.ofKind(
+                RowKind.UPDATE_AFTER,
+                1,
+                StringData.fromString("after"),
+                StringData.fromString("x"))),
+        null);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    assertThat(results).hasSize(2);
+    assertThat(dataFiles(results)).hasSize(2);
+    assertThat(deleteFiles(results))
+        .as("The only delete is the second writer's position delete; inserts 
emit no delete")
+        .hasSize(1)
+        .allMatch(file -> file.content() == FileContent.POSITION_DELETES);
+
+    DynamicWriteResult secondWriter = resultWithDelete(results, 
FileContent.POSITION_DELETES);
+    DataFile firstWriterDataFile =
+        results.stream()
+            .filter(result -> result != secondWriter)
+            .findFirst()
+            .orElseThrow()
+            .writeResult()
+            .dataFiles()[0];
+    assertThat(secondWriter.writeResult().referencedDataFiles())
+        .containsExactly(firstWriterDataFile.location());
+
+    commit(table, results);
+    assertTableRows(table, SimpleDataUtil.createRecord(1, "after", "x"));
+
+    dynamicWriter.close();
+  }
+
+  @Test
+  void testInsertedRowTrackersDoNotOutliveCheckpoint() throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    dynamicWriter.write(
+        upsertRecord(table, table.schema(), SimpleDataUtil.createRowData(1, 
"first")), null);
+    assertThat(insertedRowTrackers(dynamicWriter)).hasSize(1);
+
+    Collection<DynamicWriteResult> firstCheckpoint = 
dynamicWriter.prepareCommit();
+    assertThat(insertedRowTrackers(dynamicWriter)).isEmpty();
+    commit(table, firstCheckpoint);
+
+    // The key's row is now committed at a lower sequence number, so the next 
checkpoint retires it
+    // with an equality delete. A tracker surviving prepareCommit would 
instead emit a position
+    // delete against the previous checkpoint's data file.
+    dynamicWriter.write(
+        upsertRecord(table, table.schema(), SimpleDataUtil.createRowData(1, 
"second")), null);
+    Collection<DynamicWriteResult> secondCheckpoint = 
dynamicWriter.prepareCommit();
+    assertThat(deleteFiles(secondCheckpoint))
+        .hasSize(1)
+        .allMatch(file -> file.content() == FileContent.EQUALITY_DELETES);
+    
assertThat(secondCheckpoint.iterator().next().writeResult().referencedDataFiles()).isEmpty();
+
+    commit(table, secondCheckpoint);
+    assertTableRows(table, SimpleDataUtil.createRecord(1, "second"));
+
+    dynamicWriter.write(
+        upsertRecord(table, table.schema(), SimpleDataUtil.createRowData(2, 
"open")), null);
+    assertThat(insertedRowTrackers(dynamicWriter)).hasSize(1);
+    dynamicWriter.close();
+    assertThat(insertedRowTrackers(dynamicWriter)).isEmpty();
+  }
+
+  @Test
+  void testInsertedRowTrackersAreScopedPerTable() throws Exception {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    Table table1 = catalog.createTable(TABLE1, SimpleDataUtil.SCHEMA);
+    Table table2 = catalog.createTable(TABLE2, SimpleDataUtil.SCHEMA);
+    DynamicWriter dynamicWriter = createDynamicWriter(catalog);
+
+    dynamicWriter.write(
+        upsertRecord(table1, table1.schema(), SimpleDataUtil.createRowData(1, 
"table1")), null);
+    Schema table2Before = table2.schema();
+    dynamicWriter.write(
+        upsertRecord(table2, table2Before, SimpleDataUtil.createRowData(1, 
"before")), null);
+
+    // Only the second table evolves; its second writer shares a tracker with 
its first writer,
+    // never with the other table's writer for the same key.
+    table2.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+    dynamicWriter.write(upsertRecord(table2, table2.schema(), rowData(1, 
"after", "x")), null);
+    assertThat(insertedRowTrackers(dynamicWriter)).hasSize(2);
+
+    Collection<DynamicWriteResult> results = dynamicWriter.prepareCommit();
+    Collection<DynamicWriteResult> table1Results = resultsFor(results, TABLE1);
+    Collection<DynamicWriteResult> table2Results = resultsFor(results, TABLE2);
+    assertThat(table1Results).hasSize(1);
+    assertThat(table2Results).hasSize(2);
+    assertThat(deleteFiles(table1Results))
+        .noneMatch(file -> file.content() == FileContent.POSITION_DELETES);
+
+    DynamicWriteResult table2Second = resultWithDelete(table2Results, 
FileContent.POSITION_DELETES);
+    DynamicWriteResult table2First = resultWithDelete(table2Results, 
FileContent.EQUALITY_DELETES);
+    assertThat(table2Second.writeResult().referencedDataFiles())
+        .containsExactly(table2First.writeResult().dataFiles()[0].location());
+
+    commit(table1, table1Results);
+    commit(table2, table2Results);
+    assertTableRows(table1, SimpleDataUtil.createRecord(1, "table1"));
+    assertTableRows(table2, SimpleDataUtil.createRecord(1, "after", "x"));
+
+    dynamicWriter.close();
+  }
+
+  private static DynamicRecordInternal upsertRecord(Table table, Schema 
schema, RowData row) {
+    DynamicRecordInternal record = cdcRecord(table, schema, row);
+    record.setUpsertMode(true);
+    return record;
+  }
+
+  private static DynamicRecordInternal cdcRecord(Table table, Schema schema, 
RowData row) {
+    DynamicRecordInternal record = new DynamicRecordInternal();
+    record.setTableName(TableIdentifier.parse(table.name()).name());
+    record.setSchema(schema);
+    record.setSpec(table.spec());
+    record.setEqualityFieldIds(Sets.newHashSet(1));
+    record.setRowData(row);
+    return record;
+  }
+
+  private static Map<?, ?> insertedRowTrackers(DynamicWriter dynamicWriter) {
+    DynFields.BoundField<Map<?, ?>> trackersField =
+        DynFields.builder()
+            .hiddenImpl(DynamicWriter.class, "insertedRowTrackers")
+            .build(dynamicWriter);
+    return trackersField.get();
+  }
+
+  private static Collection<DynamicWriteResult> resultsFor(
+      Collection<DynamicWriteResult> results, TableIdentifier tableId) {
+    return results.stream()
+        .filter(result -> result.key().tableName().equals(tableId.name()))
+        .collect(Collectors.toList());
+  }
+
+  private static RowData rowData(int id, String data, String extra) {
+    return GenericRowData.of(id, StringData.fromString(data), 
StringData.fromString(extra));
+  }
+
+  private static List<DataFile> dataFiles(Collection<DynamicWriteResult> 
results) {
+    List<DataFile> files = Lists.newArrayList();
+    results.forEach(result -> 
files.addAll(Arrays.asList(result.writeResult().dataFiles())));
+    return files;
+  }
+
+  private static List<DeleteFile> deleteFiles(Collection<DynamicWriteResult> 
results) {
+    List<DeleteFile> files = Lists.newArrayList();
+    results.forEach(result -> 
files.addAll(Arrays.asList(result.writeResult().deleteFiles())));
+    return files;
+  }
+
+  private static DynamicWriteResult resultWithDelete(
+      Collection<DynamicWriteResult> results, FileContent content) {
+    return results.stream()
+        .filter(
+            result ->
+                Arrays.stream(result.writeResult().deleteFiles())
+                    .anyMatch(file -> file.content() == content))
+        .findFirst()
+        .orElseThrow();
+  }
+
+  // Compares by the table's struct type: IcebergGenerics appends the _pos 
metadata column to rows
+  // read from data files that have position deletes, which breaks 
element-wise Record equality.
+  private static void assertTableRows(Table table, Record... expected) throws 
Exception {
+    table.refresh();
+    assertThat(SimpleDataUtil.actualRowSet(table, "*"))
+        .isEqualTo(SimpleDataUtil.expectedRowSet(table, expected));
+  }
+
+  private static void commit(Table table, Collection<DynamicWriteResult> 
results) {

Review Comment:
   Could we also add a test to `TestDynamicIcebergSink` that updates a key 
after a schema change?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to