[core] Add dv conflict detection during commit (#6303)

Author: Zouxxyy

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -298,7 +298,10 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                 options.commitMinRetryWait(),
                 options.commitMaxRetryWait(),
                 options.commitStrictModeLastSafeSnapshot().orElse(null),
-                options.rowTrackingEnabled());
+                options.rowTrackingEnabled(),
+                !schema.primaryKeys().isEmpty(),
+                options.deletionVectorsEnabled(),
+                newIndexFileHandler());
     }
 
     @Override

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

@@ -22,7 +22,6 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.Filter;
-import org.apache.paimon.utils.Preconditions;
 
 import javax.annotation.Nullable;
 
@@ -40,6 +39,7 @@
 
 import static org.apache.paimon.utils.ManifestReadThreadPool.randomlyExecuteSequentialReturn;
 import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+import static org.apache.paimon.utils.Preconditions.checkState;
 
 /** Entry representing a file. */
 public interface FileEntry {
@@ -77,7 +77,7 @@ class Identifier {
         public final int level;
         public final String fileName;
         public final List<String> extraFiles;
-        @Nullable private final byte[] embeddedIndex;
+        @Nullable public final byte[] embeddedIndex;
         @Nullable public final String externalPath;
 
         /* Cache the hash code for the string */
@@ -190,7 +190,7 @@ static <T extends FileEntry> void mergeEntries(Iterable<T> entries, Map<Identifi
             Identifier identifier = entry.identifier();
             switch (entry.kind()) {
                 case ADD:
-                    Preconditions.checkState(
+                    checkState(
                             !map.containsKey(identifier),
                             "Trying to add file %s which is already added.",
                             identifier);

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

@@ -81,6 +81,21 @@ public static SimpleFileEntry from(ManifestEntry entry) {
                 entry.externalPath());
     }
 
+    public SimpleFileEntry toDelete() {
+        return new SimpleFileEntry(
+                FileKind.DELETE,
+                partition,
+                bucket,
+                totalBuckets,
+                level,
+                fileName,
+                extraFiles,
+                embeddedIndex,
+                minKey,
+                maxKey,
+                externalPath);
+    }
+
     public static List<SimpleFileEntry> from(List<ManifestEntry> entries) {
         return entries.stream().map(SimpleFileEntry::from).collect(Collectors.toList());
     }
@@ -115,6 +130,11 @@ public String fileName() {
         return fileName;
     }
 
+    @Nullable
+    public byte[] embeddedIndex() {
+        return embeddedIndex;
+    }
+
     @Nullable
     @Override
     public String externalPath() {

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

@@ -0,0 +1,124 @@
+/*
+ * 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.paimon.manifest;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/** A {@link FileEntry} contains {@link SimpleFileEntry} and dv file name. */
+public class SimpleFileEntryWithDV extends SimpleFileEntry {
+
+    @Nullable private final String dvFileName;
+
+    public SimpleFileEntryWithDV(SimpleFileEntry entry, @Nullable String dvFileName) {
+        super(
+                entry.kind(),
+                entry.partition(),
+                entry.bucket(),
+                entry.totalBuckets(),
+                entry.level(),
+                entry.fileName(),
+                entry.extraFiles(),
+                entry.embeddedIndex(),
+                entry.minKey(),
+                entry.maxKey(),
+                entry.externalPath());
+        this.dvFileName = dvFileName;
+    }
+
+    public Identifier identifier() {
+        return new IdentifierWithDv(super.identifier(), dvFileName);
+    }
+
+    @Nullable
+    public String dvFileName() {
+        return dvFileName;
+    }
+
+    public SimpleFileEntry toDelete() {
+        return new SimpleFileEntryWithDV(super.toDelete(), dvFileName);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        SimpleFileEntryWithDV that = (SimpleFileEntryWithDV) o;
+        return Objects.equals(dvFileName, that.dvFileName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), dvFileName);
+    }
+
+    @Override
+    public String toString() {
+        return super.toString() + ", {dvFileName=" + dvFileName + '}';
+    }
+
+    /**
+     * The same {@link Identifier} indicates that the {@link ManifestEntry} refers to the same data
+     * file.
+     */
+    static class IdentifierWithDv extends Identifier {
+
+        private final String dvFileName;
+
+        public IdentifierWithDv(Identifier identifier, String dvFileName) {
+            super(
+                    identifier.partition,
+                    identifier.bucket,
+                    identifier.level,
+                    identifier.fileName,
+                    identifier.extraFiles,
+                    identifier.embeddedIndex,
+                    identifier.externalPath);
+            this.dvFileName = dvFileName;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            if (!super.equals(o)) {
+                return false;
+            }
+            IdentifierWithDv that = (IdentifierWithDv) o;
+            return Objects.equals(dvFileName, that.dvFileName);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(super.hashCode(), dvFileName);
+        }
+    }
+}