[core] ConflictDetection should check delta files for OVERWRITE

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java

@@ -190,10 +190,12 @@ static <T extends FileEntry> void mergeEntries(Iterable<T> entries, Map<Identifi
             Identifier identifier = entry.identifier();
             switch (entry.kind()) {
                 case ADD:
+                    T old = map.get(identifier);
                     checkState(
-                            !map.containsKey(identifier),
-                            "Trying to add file %s which is already added.",
-                            identifier);
+                            old == null,
+                            "Trying to add file %s which is already in the the map: %s",
+                            identifier,
+                            old);
                     map.put(identifier, entry);
                     break;
                 case DELETE:

paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java

@@ -137,6 +137,15 @@ public void checkNoConflictsOrFail(
 
         Function<Throwable, RuntimeException> conflictException =
                 conflictException(baseCommitUser, baseEntries, deltaEntries);
+
+        try {
+            // check the delta, it is important not to delete and add the same file. Since scan
+            // relies on map for deduplication, this may result in the loss of this file
+            FileEntry.mergeEntries(deltaEntries);
+        } catch (Throwable e) {
+            throw conflictException.apply(e);
+        }
+
         Collection<SimpleFileEntry> mergedEntries;
         try {
             // merge manifest entries and also check if the files we want to delete are still there

paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java

@@ -124,6 +124,29 @@
 /** Tests for {@link AppendOnlyFileStoreTable}. */
 public class AppendOnlySimpleTableTest extends SimpleTableTestBase {
 
+    @Test
+    public void testOverwriteSameFiles() throws Exception {
+        FileStoreTable table = createFileStoreTable();
+
+        // write files
+        List<CommitMessage> commitMessages;
+        try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) {
+            write.write(rowData(1, 10, 100L));
+            commitMessages = write.prepareCommit();
+        }
+
+        // first commit
+        try (BatchTableCommit commit = table.newBatchWriteBuilder().withOverwrite().newCommit()) {
+            commit.commit(commitMessages);
+        }
+
+        // second commit should throw exception
+        try (BatchTableCommit commit = table.newBatchWriteBuilder().withOverwrite().newCommit()) {
+            assertThatThrownBy(() -> commit.commit(commitMessages))
+                    .hasMessageContaining("File deletion conflicts detected! Give up committing");
+        }
+    }
+
     @Test
     public void testBucketedAppendTableWriteWithInit() throws Exception {
         innerTestBucketedAppendTableWriteInit(true);