This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 42b81c3d36 [core] Refactor conflict detection by table type (#9196)
42b81c3d36 is described below
commit 42b81c3d36530c3d258d409e8aabf9f57ba2d387
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Aug 13 10:06:21 2026 +0800
[core] Refactor conflict detection by table type (#9196)
---
.../java/org/apache/paimon/AbstractFileStore.java | 2 +-
.../paimon/operation/FileStoreCommitImpl.java | 121 +---
.../operation/commit/AppendConflictDetection.java | 71 +++
.../paimon/operation/commit/ConflictDetection.java | 591 ++++--------------
.../commit/DataEvolutionConflictDetection.java | 687 +++++++++++++++++++++
.../commit/PrimaryKeyConflictDetection.java | 163 +++++
.../operation/commit/RowIdConflictChecker.java | 6 -
.../paimon/operation/FileStoreCommitTest.java | 2 +-
.../operation/commit/ConflictDetectionTest.java | 607 +++++++++++++++++-
.../table/DataEvolutionDeletionVectorTest.java | 107 +++-
.../paimon/table/DataEvolutionTableTest.java | 35 ++
11 files changed, 1768 insertions(+), 624 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index 524dd4962c..1bf2a59e3f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -293,7 +293,7 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
}
ConflictDetection.Factory conflictDetectFactory =
scanner ->
- new ConflictDetection(
+ ConflictDetection.create(
tableName,
commitUser,
partitionType,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 7f9a02f602..eb6ef13cef 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -27,7 +27,6 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
@@ -54,9 +53,7 @@ import
org.apache.paimon.operation.commit.ManifestEntryChanges;
import org.apache.paimon.operation.commit.RetryCommitResult;
import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
import
org.apache.paimon.operation.commit.RetryCommitResult.ManifestMergeResult;
-import org.apache.paimon.operation.commit.RowIdColumnConflictChecker;
import org.apache.paimon.operation.commit.RowIdConflictChecker;
-import org.apache.paimon.operation.commit.RowIdRangeConflictChecker;
import
org.apache.paimon.operation.commit.RowTrackingCommitUtils.RowTrackingAssigned;
import org.apache.paimon.operation.commit.StrictModeChecker;
import org.apache.paimon.operation.commit.SuccessCommitResult;
@@ -84,7 +81,6 @@ import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.ListUtils;
import org.apache.paimon.utils.Pair;
-import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RetryWaiter;
import org.apache.paimon.utils.SnapshotManager;
@@ -99,7 +95,6 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -116,7 +111,6 @@ import static
org.apache.paimon.manifest.ManifestEntry.nullableRecordCount;
import static org.apache.paimon.manifest.ManifestEntry.recordCountAdd;
import static org.apache.paimon.manifest.ManifestEntry.recordCountDelete;
import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPartitions;
-import static
org.apache.paimon.operation.commit.RowIdConflictChecker.TriggerSource.MATERIALIZE_DV_COMPACTION;
import static
org.apache.paimon.operation.commit.RowTrackingCommitUtils.assignRowTracking;
import static
org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
import static
org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
@@ -1060,55 +1054,15 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
retryResult instanceof CommitFailRetryResult
? (CommitFailRetryResult) retryResult
: null;
- List<Range> changedRowRanges = changedRowRanges(deltaFiles,
indexFiles, commitKind);
- // An overwrite may replace the base manifest list without
recording the replacements
- // in its delta manifest, so the cached base cannot always be
refreshed incrementally.
- if (!changedRowRanges.isEmpty()) {
- baseDataFiles =
- scanner.readAllEntriesFromChangedRowRanges(
- latestSnapshot, changedPartitions,
changedRowRanges);
- Set<String> dataFilesToDelete =
- deltaFiles.stream()
- .filter(entry -> entry.kind() ==
FileKind.DELETE)
- .map(entry -> entry.file().fileName())
- .collect(Collectors.toSet());
- for (IndexManifestEntry indexFile : indexFiles) {
- if (indexFile.indexFile().dvRanges() != null) {
-
dataFilesToDelete.addAll(indexFile.indexFile().dvRanges().keySet());
- }
- }
- if (!dataFilesToDelete.isEmpty()) {
- baseDataFiles.addAll(
- scanner.readAllEntriesFromDataFiles(
- latestSnapshot, changedPartitions,
dataFilesToDelete));
- baseDataFiles =
- new ArrayList<>(
- baseDataFiles.stream()
- .collect(
- Collectors.toMap(
-
FileEntry::identifier,
- entry -> entry,
- (left, right) ->
left,
-
LinkedHashMap::new))
- .values());
- }
- } else if (commitFailRetry != null
- && commitFailRetry.latestSnapshot != null
- && commitFailRetry.baseDataFiles != null
- && !hasOverwriteSinceLastAttempt) {
- baseDataFiles = new ArrayList<>(commitFailRetry.baseDataFiles);
- List<SimpleFileEntry> incremental =
- scanner.readIncrementalChanges(
- commitFailRetry.latestSnapshot,
latestSnapshot, changedPartitions);
- if (!incremental.isEmpty()) {
- baseDataFiles.addAll(incremental);
- baseDataFiles = new
ArrayList<>(FileEntry.mergeEntries(baseDataFiles));
- }
- } else {
- baseDataFiles =
- scanner.readAllEntriesFromChangedPartitions(
- latestSnapshot, changedPartitions);
- }
+ baseDataFiles =
+ conflictDetection.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ commitKind,
+ commitFailRetry,
+ hasOverwriteSinceLastAttempt);
if (discardDuplicate) {
Set<FileEntry.Identifier> baseIdentifiers =
baseDataFiles.stream()
@@ -1119,36 +1073,9 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
.filter(entry ->
!baseIdentifiers.contains(entry.identifier()))
.collect(Collectors.toList());
}
- RowIdConflictChecker rowIdConflictChecker = null;
- if (conflictDetection.shouldCheckRowIdFromSnapshot(commitKind)) {
- List<DataFileMeta> rowIdConflictFiles;
- if (conflictDetection.rowIdConflictCheckTriggerSource()
- == MATERIALIZE_DV_COMPACTION) {
- // For materialize dv compaction jobs, we should check
each deleted file range
- // will not be erroneously restored by concurrent
merg-into updates.
- rowIdConflictFiles =
- deltaFiles.stream()
- .filter(entry -> entry.kind() ==
FileKind.DELETE)
- .map(ManifestEntry::file)
- .filter(file -> file.firstRowId() != null)
- .filter(
- file ->
-
!isBlobFile(file.fileName())
- &&
!isVectorStoreFile(file.fileName()))
- .collect(Collectors.toList());
-
- rowIdConflictChecker =
-
RowIdRangeConflictChecker.fromDataFiles(rowIdConflictFiles);
- } else {
- rowIdConflictFiles =
- deltaFiles.stream()
- .map(ManifestEntry::file)
- .collect(Collectors.toList());
- rowIdConflictChecker =
- RowIdColumnConflictChecker.fromDataFiles(
- schemaManager, rowIdConflictFiles);
- }
- }
+ RowIdConflictChecker rowIdConflictChecker =
+ conflictDetection.createRowIdConflictChecker(
+ schemaManager, deltaFiles, commitKind);
Optional<RuntimeException> exception =
conflictDetection.checkConflicts(
latestSnapshot,
@@ -1381,30 +1308,6 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return new SuccessCommitResult();
}
- private List<Range> changedRowRanges(
- List<ManifestEntry> deltaFiles,
- List<IndexManifestEntry> indexFiles,
- CommitKind commitKind) {
- if (!options.dataEvolutionEnabled() || commitKind !=
CommitKind.COMPACT) {
- return Collections.emptyList();
- }
-
- List<Range> ranges =
- deltaFiles.stream()
- .map(ManifestEntry::file)
- .filter(file -> file.firstRowId() != null)
- .map(DataFileMeta::nonNullRowIdRange)
- .collect(Collectors.toList());
- indexFiles.stream()
- .filter(entry -> entry.kind() == FileKind.ADD)
- .map(IndexManifestEntry::indexFile)
- .map(IndexFileMeta::globalIndexMeta)
- .filter(Objects::nonNull)
- .map(GlobalIndexMeta::rowRange)
- .forEach(ranges::add);
- return Range.sortAndMergeOverlap(ranges, true);
- }
-
@Nullable
private ManifestMergeReuse tryReuseManifestMergeResult(
@Nullable RetryCommitResult retryResult, List<ManifestFileMeta>
currentManifests) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/AppendConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/AppendConflictDetection.java
new file mode 100644
index 0000000000..f9b1d5df52
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/AppendConflictDetection.java
@@ -0,0 +1,71 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+/** Conflict detection for append-only tables. */
+public class AppendConflictDetection extends ConflictDetection {
+
+ public AppendConflictDetection(
+ String tableName,
+ String commitUser,
+ RowType partitionType,
+ FileStorePathFactory pathFactory,
+ BucketMode bucketMode,
+ boolean deletionVectorsEnabled,
+ IndexFileHandler indexFileHandler,
+ CommitScanner commitScanner) {
+ super(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ bucketMode,
+ deletionVectorsEnabled,
+ indexFileHandler,
+ commitScanner);
+ }
+
+ @Override
+ protected Optional<RuntimeException> checkTableSpecificConflicts(
+ Snapshot latestSnapshot,
+ List<SimpleFileEntry> baseEntries,
+ List<SimpleFileEntry> deltaEntries,
+ List<IndexManifestEntry> deltaIndexEntries,
+ Collection<SimpleFileEntry> mergedEntries,
+ @Nullable RowIdConflictChecker rowIdConflictChecker,
+ CommitKind commitKind,
+ String baseCommitUser) {
+ return Optional.empty();
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index ca7e57224e..b660d50ad4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -22,12 +22,9 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.errors.ErrorMessages;
import org.apache.paimon.index.DeletionVectorMeta;
-import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -35,13 +32,12 @@ import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.SimpleFileEntry;
import org.apache.paimon.manifest.SimpleFileEntryWithDV;
import org.apache.paimon.operation.PartitionExpire;
+import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Pair;
-import org.apache.paimon.utils.Range;
-import org.apache.paimon.utils.RangeHelper;
-import org.apache.paimon.utils.RowRangeIndex;
import org.apache.paimon.utils.SnapshotManager;
import org.slf4j.Logger;
@@ -65,15 +61,11 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
-import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
-import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPartitions;
-import static
org.apache.paimon.operation.commit.RowIdConflictChecker.TriggerSource.MATERIALIZE_DV_COMPACTION;
-import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static
org.apache.paimon.utils.InternalRowPartitionComputer.partToSimpleString;
import static org.apache.paimon.utils.Preconditions.checkState;
-/** Util class for detecting conflicts between base and delta files. */
-public class ConflictDetection {
+/** Base class for table-specific conflict detection between base and delta
files. */
+public abstract class ConflictDetection {
private static final Logger LOG =
LoggerFactory.getLogger(ConflictDetection.class);
private static final int FIXED_BUCKET_CHECK_CACHE_MAX_SIZE = 1000;
@@ -82,13 +74,9 @@ public class ConflictDetection {
private final String commitUser;
private final RowType partitionType;
private final FileStorePathFactory pathFactory;
- private final @Nullable Comparator<InternalRow> keyComparator;
private final BucketMode bucketMode;
private final boolean deletionVectorsEnabled;
- private final boolean dataEvolutionEnabled;
- private final boolean pkClusteringOverride;
private final IndexFileHandler indexFileHandler;
- private final SnapshotManager snapshotManager;
private final CommitScanner commitScanner;
private final Set<BinaryRow> checkedFixedBucketPartitions =
Collections.newSetFromMap(
@@ -101,70 +89,97 @@ public class ConflictDetection {
});
private @Nullable PartitionExpire partitionExpire;
- private @Nullable Long rowIdCheckFromSnapshot = null;
- private @Nullable RowIdConflictChecker.TriggerSource
rowIdConflictCheckTriggerSource = null;
- public ConflictDetection(
+ protected ConflictDetection(
String tableName,
String commitUser,
RowType partitionType,
FileStorePathFactory pathFactory,
- @Nullable Comparator<InternalRow> keyComparator,
BucketMode bucketMode,
boolean deletionVectorsEnabled,
- boolean dataEvolutionEnabled,
- boolean pkClusteringOverride,
IndexFileHandler indexFileHandler,
- SnapshotManager snapshotManager,
CommitScanner commitScanner) {
this.tableName = tableName;
this.commitUser = commitUser;
this.partitionType = partitionType;
this.pathFactory = pathFactory;
- this.keyComparator = keyComparator;
this.bucketMode = bucketMode;
this.deletionVectorsEnabled = deletionVectorsEnabled;
- this.dataEvolutionEnabled = dataEvolutionEnabled;
- this.pkClusteringOverride = pkClusteringOverride;
this.indexFileHandler = indexFileHandler;
- this.snapshotManager = snapshotManager;
this.commitScanner = commitScanner;
}
+ public static ConflictDetection create(
+ String tableName,
+ String commitUser,
+ RowType partitionType,
+ FileStorePathFactory pathFactory,
+ @Nullable Comparator<InternalRow> keyComparator,
+ BucketMode bucketMode,
+ boolean deletionVectorsEnabled,
+ boolean dataEvolutionEnabled,
+ boolean pkClusteringOverride,
+ IndexFileHandler indexFileHandler,
+ SnapshotManager snapshotManager,
+ CommitScanner commitScanner) {
+ if (dataEvolutionEnabled) {
+ return new DataEvolutionConflictDetection(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ bucketMode,
+ deletionVectorsEnabled,
+ indexFileHandler,
+ snapshotManager,
+ commitScanner);
+ }
+ if (keyComparator != null) {
+ return new PrimaryKeyConflictDetection(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ keyComparator,
+ bucketMode,
+ deletionVectorsEnabled,
+ pkClusteringOverride,
+ indexFileHandler,
+ commitScanner);
+ }
+ return new AppendConflictDetection(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ bucketMode,
+ deletionVectorsEnabled,
+ indexFileHandler,
+ commitScanner);
+ }
+
public void setRowIdCheckFromSnapshot(@Nullable Long
rowIdCheckFromSnapshot) {
- setRowIdCheckFromSnapshot(
- rowIdCheckFromSnapshot,
RowIdConflictChecker.TriggerSource.DATA_EVOLUTION_DML);
+ // Only Data Evolution tables support Row ID conflict detection.
}
public void setRowIdCheckFromSnapshotForMaterializeDvCompaction(
@Nullable Long rowIdCheckFromSnapshot) {
- setRowIdCheckFromSnapshot(rowIdCheckFromSnapshot,
MATERIALIZE_DV_COMPACTION);
- }
-
- private void setRowIdCheckFromSnapshot(
- @Nullable Long rowIdCheckFromSnapshot,
- RowIdConflictChecker.TriggerSource triggerSource) {
- this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
- this.rowIdConflictCheckTriggerSource =
- rowIdCheckFromSnapshot == null ? null : triggerSource;
+ // Only Data Evolution tables support Row ID conflict detection.
}
public boolean shouldCheckRowIdFromSnapshot(CommitKind commitKind) {
- return rowIdCheckFromSnapshot != null
- && (rowIdConflictCheckTriggerSource !=
MATERIALIZE_DV_COMPACTION
- || commitKind == CommitKind.COMPACT);
+ return false;
}
- public RowIdConflictChecker.TriggerSource
rowIdConflictCheckTriggerSource() {
- checkState(
- rowIdConflictCheckTriggerSource != null,
- "Row ID conflict check trigger source is not set.");
- return rowIdConflictCheckTriggerSource;
+ @Nullable
+ public RowIdConflictChecker createRowIdConflictChecker(
+ SchemaManager schemaManager, List<ManifestEntry> deltaFiles,
CommitKind commitKind) {
+ return null;
}
@Nullable
public Comparator<InternalRow> keyComparator() {
- return keyComparator;
+ return null;
}
public void withPartitionExpire(PartitionExpire partitionExpire) {
@@ -186,7 +201,7 @@ public class ConflictDetection {
return false;
}
- public Optional<RuntimeException> checkConflicts(
+ public final Optional<RuntimeException> checkConflicts(
Snapshot latestSnapshot,
List<SimpleFileEntry> baseEntries,
List<SimpleFileEntry> deltaEntries,
@@ -250,29 +265,66 @@ public class ConflictDetection {
if (exception.isPresent()) {
return exception;
}
- exception = checkKeyRange(baseEntries, deltaEntries, mergedEntries,
baseCommitUser);
- if (exception.isPresent()) {
- return exception;
- }
+ return checkTableSpecificConflicts(
+ latestSnapshot,
+ baseEntries,
+ deltaEntries,
+ deltaIndexEntries,
+ mergedEntries,
+ rowIdConflictChecker,
+ commitKind,
+ baseCommitUser);
+ }
- Long nextRowId = latestSnapshot.nextRowId();
- exception = checkRowIdExistence(baseEntries, deltaEntries, nextRowId,
commitKind);
- if (exception.isPresent()) {
- return exception;
- }
+ protected abstract Optional<RuntimeException> checkTableSpecificConflicts(
+ Snapshot latestSnapshot,
+ List<SimpleFileEntry> baseEntries,
+ List<SimpleFileEntry> deltaEntries,
+ List<IndexManifestEntry> deltaIndexEntries,
+ Collection<SimpleFileEntry> mergedEntries,
+ @Nullable RowIdConflictChecker rowIdConflictChecker,
+ CommitKind commitKind,
+ String baseCommitUser);
- exception = checkRowIdRangeConflicts(commitKind, mergedEntries);
- if (exception.isPresent()) {
- return exception;
- }
+ public List<SimpleFileEntry> scanBaseDataFiles(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ CommitKind commitKind,
+ @Nullable CommitFailRetryResult previousAttempt,
+ boolean hasOverwriteSincePreviousAttempt) {
+ return scanChangedPartitions(
+ latestSnapshot,
+ changedPartitions,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
- exception = checkGlobalIndexRowIdExistence(baseEntries,
deltaIndexEntries);
- if (exception.isPresent()) {
- return exception;
+ protected List<SimpleFileEntry> scanChangedPartitions(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ @Nullable CommitFailRetryResult previousAttempt,
+ boolean hasOverwriteSincePreviousAttempt) {
+ if (previousAttempt != null
+ && previousAttempt.latestSnapshot != null
+ && previousAttempt.baseDataFiles != null
+ && !hasOverwriteSincePreviousAttempt) {
+ List<SimpleFileEntry> baseDataFiles = new
ArrayList<>(previousAttempt.baseDataFiles);
+ List<SimpleFileEntry> incremental =
+ commitScanner.readIncrementalChanges(
+ previousAttempt.latestSnapshot, latestSnapshot,
changedPartitions);
+ if (!incremental.isEmpty()) {
+ baseDataFiles.addAll(incremental);
+ baseDataFiles = new
ArrayList<>(FileEntry.mergeEntries(baseDataFiles));
+ }
+ return baseDataFiles;
}
+ return
commitScanner.readAllEntriesFromChangedPartitions(latestSnapshot,
changedPartitions);
+ }
- return checkForRowIdFromSnapshot(
- latestSnapshot, deltaEntries, deltaIndexEntries,
rowIdConflictChecker);
+ protected CommitScanner commitScanner() {
+ return commitScanner;
}
public <T extends FileEntry> Map<BinaryRow, Integer>
collectUncheckedFixedBucketPartitions(
@@ -378,59 +430,6 @@ public class ConflictDetection {
partInfo, numBuckets, previousNumBuckets));
}
- private Optional<RuntimeException> checkKeyRange(
- List<SimpleFileEntry> baseEntries,
- List<SimpleFileEntry> deltaEntries,
- Collection<SimpleFileEntry> mergedEntries,
- String baseCommitUser) {
- // fast exit for file store without keys
- if (keyComparator == null) {
- return Optional.empty();
- }
-
- // Not clustering by pk, no need to check key range
- if (pkClusteringOverride) {
- return Optional.empty();
- }
-
- // group entries by partitions, buckets and levels
- Map<LevelIdentifier, List<SimpleFileEntry>> levels = new HashMap<>();
- for (SimpleFileEntry entry : mergedEntries) {
- int level = entry.level();
- if (level >= 1) {
- levels.computeIfAbsent(
- new LevelIdentifier(entry.partition(),
entry.bucket(), level),
- lv -> new ArrayList<>())
- .add(entry);
- }
- }
-
- // check for all LSM level >= 1, key ranges of files do not intersect
- for (List<SimpleFileEntry> entries : levels.values()) {
- entries.sort((a, b) -> keyComparator.compare(a.minKey(),
b.minKey()));
- for (int i = 0; i + 1 < entries.size(); i++) {
- SimpleFileEntry a = entries.get(i);
- SimpleFileEntry b = entries.get(i + 1);
- if (keyComparator.compare(a.maxKey(), b.minKey()) >= 0) {
- Pair<RuntimeException, RuntimeException> conflictException
=
- createConflictException(
- "LSM conflicts detected! Give up
committing. Conflict files are:\n"
- +
a.identifier().toString(pathFactory)
- + "\n"
- +
b.identifier().toString(pathFactory),
- baseCommitUser,
- baseEntries,
- deltaEntries,
- null);
-
- LOG.warn("", conflictException.getLeft());
- return Optional.of(conflictException.getRight());
- }
- }
- }
- return Optional.empty();
- }
-
private Function<Throwable, RuntimeException> conflictException(
String baseCommitUser,
List<SimpleFileEntry> baseEntries,
@@ -497,329 +496,6 @@ public class ConflictDetection {
return Optional.empty();
}
- private Optional<RuntimeException> checkRowIdRangeConflicts(
- CommitKind commitKind, Collection<SimpleFileEntry> mergedEntries) {
- if (!dataEvolutionEnabled) {
- return Optional.empty();
- }
- if (rowIdCheckFromSnapshot == null && commitKind !=
CommitKind.COMPACT) {
- return Optional.empty();
- }
-
- List<SimpleFileEntry> entries =
- mergedEntries.stream()
- .filter(file -> file.firstRowId() != null)
- .collect(Collectors.toList());
-
- RangeHelper<SimpleFileEntry> rangeHelper =
- new RangeHelper<>(SimpleFileEntry::nonNullRowIdRange);
- List<SimpleFileEntry> dataFiles =
- entries.stream()
- .filter(file -> !dedicatedStorageFile(file.fileName()))
- .collect(Collectors.toList());
-
- Optional<RuntimeException> exception =
- checkDataFileRowIdRangeConflicts(rangeHelper, dataFiles);
- if (exception.isPresent()) {
- return exception;
- }
-
- List<SimpleFileEntry> dedicatedFiles =
- entries.stream()
- .filter(file -> dedicatedStorageFile(file.fileName()))
- .collect(Collectors.toList());
- exception = checkDedicatedFileRowIdRangeConflicts(dataFiles,
dedicatedFiles);
- if (exception.isPresent()) {
- return exception;
- }
-
- return Optional.empty();
- }
-
- private Optional<RuntimeException> checkDataFileRowIdRangeConflicts(
- RangeHelper<SimpleFileEntry> rangeHelper, List<SimpleFileEntry>
dataFiles) {
- for (List<SimpleFileEntry> dataFileGroup :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
- if (!rangeHelper.areAllRangesSame(dataFileGroup)) {
- return Optional.of(
- new RuntimeException(
- "For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
- + "operations "
- + "have encountered conflicts, data
files: "
- + dataFileGroup));
- }
- }
-
- return Optional.empty();
- }
-
- private Optional<RuntimeException> checkDedicatedFileRowIdRangeConflicts(
- List<SimpleFileEntry> dataFiles, List<SimpleFileEntry>
dedicatedFiles) {
- if (dedicatedFiles.isEmpty()) {
- return Optional.empty();
- }
-
- RowRangeIndex dataFileRowRangeIndex = rowRangeIndex(dataFiles, false);
-
- for (SimpleFileEntry dedicatedFile : dedicatedFiles) {
- Range dedicatedRange = dedicatedFile.nonNullRowIdRange();
- if (dataFileRowRangeIndex.contains(dedicatedRange)) {
- continue;
- }
-
- List<Range> intersectingRanges =
-
dataFileRowRangeIndex.intersectedRanges(dedicatedRange.from, dedicatedRange.to);
- List<SimpleFileEntry> intersectingDataFiles =
- dataFiles.stream()
- .filter(
- dataFile ->
- dataFile.nonNullRowIdRange()
-
.hasIntersection(dedicatedRange))
- .collect(Collectors.toList());
- String conflictReason =
- intersectingRanges.size() > 1
- ? "spans multiple data file ranges"
- : "is not covered by one data file range";
- return Optional.of(
- new RuntimeException(
- String.format(
- "For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
- + "operations have encountered
conflicts, dedicated "
- + "file %s %s %s: %s",
- dedicatedFile,
- dedicatedRange,
- conflictReason,
- intersectingDataFiles)));
- }
-
- return Optional.empty();
- }
-
- private static RowRangeIndex rowRangeIndex(
- Collection<SimpleFileEntry> files, boolean mergeAdjacent) {
- return RowRangeIndex.create(
-
files.stream().map(SimpleFileEntry::nonNullRowIdRange).collect(Collectors.toList()),
- mergeAdjacent);
- }
-
- private Optional<RuntimeException> checkForRowIdFromSnapshot(
- Snapshot latestSnapshot,
- List<SimpleFileEntry> deltaEntries,
- List<IndexManifestEntry> deltaIndexEntries,
- @Nullable RowIdConflictChecker conflictChecker) {
- if (!dataEvolutionEnabled) {
- return Optional.empty();
- }
- if (rowIdCheckFromSnapshot == null) {
- return Optional.empty();
- }
- if (conflictChecker == null || conflictChecker.isEmpty()) {
- return Optional.empty();
- }
-
- List<BinaryRow> changedPartitions = changedPartitions(deltaEntries,
deltaIndexEntries);
-
- // check history row id ranges
- Long checkNextRowId =
snapshotManager.snapshot(rowIdCheckFromSnapshot).nextRowId();
- checkState(
- checkNextRowId != null,
- "Next row id cannot be null for snapshot %s.",
- rowIdCheckFromSnapshot);
- for (long i = rowIdCheckFromSnapshot + 1; i <= latestSnapshot.id();
i++) {
- Snapshot snapshot = snapshotManager.snapshot(i);
- if (snapshot.commitKind() == CommitKind.COMPACT) {
- continue;
- }
- List<ManifestEntry> changes =
- commitScanner.readIncrementalEntries(snapshot,
changedPartitions);
- for (ManifestEntry entry : changes) {
- if (!shouldCheckHistoricalRowIdEntry(entry.kind())) {
- continue;
- }
- DataFileMeta file = entry.file();
- if (file.firstRowId() != null
- && file.nonNullRowIdRange().from < checkNextRowId
- && conflictChecker.conflictsWith(file)) {
- LOG.debug(
- "Data evolution row id conflict detected for table
{}, commit user {}, "
- + "snapshot {}, file {}.",
- tableName,
- commitUser,
- snapshot.id(),
- file);
- return Optional.of(
- new RuntimeException(
-
ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE));
- }
- }
- }
-
- return Optional.empty();
- }
-
- boolean shouldCheckHistoricalRowIdEntry(FileKind kind) {
- return rowIdConflictCheckTriggerSource != MATERIALIZE_DV_COMPACTION ||
kind == FileKind.ADD;
- }
-
- private Optional<RuntimeException> checkGlobalIndexRowIdExistence(
- List<SimpleFileEntry> baseEntries, List<IndexManifestEntry>
deltaIndexEntries) {
- if (!dataEvolutionEnabled) {
- return Optional.empty();
- }
-
- List<IndexManifestEntry> indexesToCheck =
globalIndexFileAdditions(deltaIndexEntries);
- if (indexesToCheck.isEmpty()) {
- return Optional.empty();
- }
-
- Map<Pair<BinaryRow, Integer>, List<Range>> dataRanges = new
HashMap<>();
- for (SimpleFileEntry entry : baseEntries) {
- if (entry.kind() == FileKind.ADD && entry.firstRowId() != null) {
- dataRanges
- .computeIfAbsent(
- Pair.of(entry.partition(), entry.bucket()), k
-> new ArrayList<>())
- .add(entry.nonNullRowIdRange());
- }
- }
- Map<Pair<BinaryRow, Integer>, RowRangeIndex> rowRangeIndexes =
- dataRanges.entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- entry ->
RowRangeIndex.create(entry.getValue())));
-
- for (IndexManifestEntry indexEntry : indexesToCheck) {
- GlobalIndexMeta globalIndex =
indexEntry.indexFile().globalIndexMeta();
- checkState(globalIndex != null, "Global index meta must not be
null.");
- Range indexRange = globalIndex.rowRange();
- RowRangeIndex rowRangeIndex =
- rowRangeIndexes.get(Pair.of(indexEntry.partition(),
indexEntry.bucket()));
- if (rowRangeIndex == null || !rowRangeIndex.contains(indexRange)) {
- return Optional.of(
- new RuntimeException(
- String.format(
- "Global index row ID existence
conflict: index file '%s' "
- + "references row range %s,
but this range "
- + "is not fully covered by
current data "
- + "files. The referenced row
IDs may have been "
- + "reassigned or removed by a
concurrent commit.",
- indexEntry.indexFile().fileName(),
indexRange)));
- }
- }
- return Optional.empty();
- }
-
- private List<IndexManifestEntry> globalIndexFileAdditions(
- List<IndexManifestEntry> indexFileChanges) {
- List<IndexManifestEntry> result = new ArrayList<>();
- for (IndexManifestEntry entry : indexFileChanges) {
- if (entry.kind() == FileKind.ADD &&
entry.indexFile().globalIndexMeta() != null) {
- result.add(entry);
- }
- }
- return result;
- }
-
- Optional<RuntimeException> checkRowIdExistence(
- List<SimpleFileEntry> baseEntries,
- List<SimpleFileEntry> deltaEntries,
- @Nullable Long nextRowId,
- CommitKind commitKind) {
- if (!dataEvolutionEnabled) {
- return Optional.empty();
- }
-
- List<SimpleFileEntry> existingDataFiles =
- baseEntries.stream()
- .filter(
- base ->
- base.firstRowId() != null
- &&
!dedicatedStorageFile(base.fileName()))
- .collect(Collectors.toList());
-
- if (commitKind == CommitKind.COMPACT) {
- return checkCompactRowIdExistence(existingDataFiles, deltaEntries);
- }
- return checkNonCompactRowIdExistence(existingDataFiles, deltaEntries,
nextRowId);
- }
-
- /**
- * Checks conflicts between compaction and concurrent Row ID reassignment,
which may otherwise
- * cause reassigned Row IDs to fall back. For example, compaction produces
a file with Row IDs
- * [0, 9], then reassignment moves the current range to [10, 19].
Committing the stale
- * compaction would move the Row IDs back to [0, 9].
- */
- private Optional<RuntimeException> checkCompactRowIdExistence(
- List<SimpleFileEntry> existingDataFiles, List<SimpleFileEntry>
deltaEntries) {
- // A compaction output may cover multiple adjacent normal files, but
only within the same
- // partition and bucket.
- Map<Pair<BinaryRow, Integer>, List<SimpleFileEntry>> dataFilesByBucket
=
- existingDataFiles.stream()
- .collect(
- Collectors.groupingBy(
- file -> Pair.of(file.partition(),
file.bucket())));
- Map<Pair<BinaryRow, Integer>, RowRangeIndex> existingIndexes =
- dataFilesByBucket.entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- entry ->
rowRangeIndex(entry.getValue(), true)));
-
- for (SimpleFileEntry entry : deltaEntries) {
- if (entry.kind() != FileKind.ADD || entry.firstRowId() == null) {
- continue;
- }
-
- RowRangeIndex existingIndex =
- existingIndexes.get(Pair.of(entry.partition(),
entry.bucket()));
- if (existingIndex == null ||
!existingIndex.contains(entry.nonNullRowIdRange())) {
- return Optional.of(rowIdExistenceConflict(entry));
- }
- }
- return Optional.empty();
- }
-
- private Optional<RuntimeException> checkNonCompactRowIdExistence(
- List<SimpleFileEntry> existingDataFiles,
- List<SimpleFileEntry> deltaEntries,
- @Nullable Long nextRowId) {
- List<SimpleFileEntry> filesToCheck =
- deltaEntries.stream()
- .filter(
- e ->
- e.kind() == FileKind.ADD
- && e.firstRowId() != null
- && nextRowId != null
- && e.firstRowId() < nextRowId)
- .collect(Collectors.toList());
-
- if (filesToCheck.isEmpty()) {
- return Optional.empty();
- }
-
- RowRangeIndex existingIndex = rowRangeIndex(existingDataFiles, false);
-
- for (SimpleFileEntry entry : filesToCheck) {
- Range rowRange = entry.nonNullRowIdRange();
- boolean exists =
- dedicatedStorageFile(entry.fileName())
- ? existingIndex.contains(rowRange)
- : existingIndex.containsExactly(rowRange);
- if (!exists) {
- return Optional.of(rowIdExistenceConflict(entry));
- }
- }
- return Optional.empty();
- }
-
- private RowIdExistenceConflictException
rowIdExistenceConflict(SimpleFileEntry entry) {
- return new RowIdExistenceConflictException(
- entry.fileName(), entry.firstRowId(), entry.rowCount(),
entry.bucket());
- }
-
- private static boolean dedicatedStorageFile(String fileName) {
- return isBlobFile(fileName) || isVectorStoreFile(fileName);
- }
-
static List<SimpleFileEntry> buildBaseEntriesWithDV(
List<SimpleFileEntry> baseEntries, List<IndexManifestEntry>
baseIndexEntries) {
if (baseEntries.isEmpty()) {
@@ -937,7 +613,7 @@ public class ConflictDetection {
* simplified exception), The simplified exception is generated when the
entry length is larger
* than the max limit.
*/
- private Pair<RuntimeException, RuntimeException> createConflictException(
+ protected final Pair<RuntimeException, RuntimeException>
createConflictException(
String message,
String baseCommitUser,
List<SimpleFileEntry> baseEntries,
@@ -1020,35 +696,6 @@ public class ConflictDetection {
}
}
- private static class LevelIdentifier {
-
- private final BinaryRow partition;
- private final int bucket;
- private final int level;
-
- private LevelIdentifier(BinaryRow partition, int bucket, int level) {
- this.partition = partition;
- this.bucket = bucket;
- this.level = level;
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof LevelIdentifier)) {
- return false;
- }
- LevelIdentifier that = (LevelIdentifier) o;
- return Objects.equals(partition, that.partition)
- && bucket == that.bucket
- && level == that.level;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(partition, bucket, level);
- }
- }
-
/** Factory to create {@link ConflictDetection}. */
public interface Factory {
ConflictDetection create(CommitScanner scanner);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
new file mode 100644
index 0000000000..c04e3d4651
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java
@@ -0,0 +1,687 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.errors.ErrorMessages;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RangeHelper;
+import org.apache.paimon.utils.RowRangeIndex;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPartitions;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Conflict detection for Data Evolution tables. */
+public class DataEvolutionConflictDetection extends ConflictDetection {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionConflictDetection.class);
+
+ private final String tableName;
+ private final String commitUser;
+ private final SnapshotManager snapshotManager;
+
+ private @Nullable Long rowIdCheckFromSnapshot;
+ private @Nullable RowIdConflictCheckStrategy rowIdConflictCheckStrategy;
+
+ public DataEvolutionConflictDetection(
+ String tableName,
+ String commitUser,
+ RowType partitionType,
+ FileStorePathFactory pathFactory,
+ BucketMode bucketMode,
+ boolean deletionVectorsEnabled,
+ IndexFileHandler indexFileHandler,
+ SnapshotManager snapshotManager,
+ CommitScanner commitScanner) {
+ super(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ bucketMode,
+ deletionVectorsEnabled,
+ indexFileHandler,
+ commitScanner);
+ this.tableName = tableName;
+ this.commitUser = commitUser;
+ this.snapshotManager = snapshotManager;
+ }
+
+ @Override
+ public void setRowIdCheckFromSnapshot(@Nullable Long
rowIdCheckFromSnapshot) {
+ setRowIdCheckFromSnapshot(
+ rowIdCheckFromSnapshot,
DataEvolutionDmlRowIdConflictCheck.INSTANCE);
+ }
+
+ @Override
+ public void setRowIdCheckFromSnapshotForMaterializeDvCompaction(
+ @Nullable Long rowIdCheckFromSnapshot) {
+ setRowIdCheckFromSnapshot(rowIdCheckFromSnapshot,
MaterializeDvRowIdConflictCheck.INSTANCE);
+ }
+
+ private void setRowIdCheckFromSnapshot(
+ @Nullable Long rowIdCheckFromSnapshot,
+ RowIdConflictCheckStrategy conflictCheckStrategy) {
+ this.rowIdCheckFromSnapshot = rowIdCheckFromSnapshot;
+ this.rowIdConflictCheckStrategy =
+ rowIdCheckFromSnapshot == null ? null : conflictCheckStrategy;
+ }
+
+ @Override
+ public boolean shouldCheckRowIdFromSnapshot(CommitKind commitKind) {
+ return rowIdCheckFromSnapshot != null &&
rowIdConflictCheckStrategy().appliesTo(commitKind);
+ }
+
+ @Override
+ @Nullable
+ public RowIdConflictChecker createRowIdConflictChecker(
+ SchemaManager schemaManager, List<ManifestEntry> deltaFiles,
CommitKind commitKind) {
+ if (!shouldCheckRowIdFromSnapshot(commitKind)) {
+ return null;
+ }
+ return rowIdConflictCheckStrategy().createChecker(schemaManager,
deltaFiles);
+ }
+
+ private RowIdConflictCheckStrategy rowIdConflictCheckStrategy() {
+ checkState(
+ rowIdConflictCheckStrategy != null, "Row ID conflict check
strategy is not set.");
+ return rowIdConflictCheckStrategy;
+ }
+
+ @Override
+ public List<SimpleFileEntry> scanBaseDataFiles(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ CommitKind commitKind,
+ @Nullable CommitFailRetryResult previousAttempt,
+ boolean hasOverwriteSincePreviousAttempt) {
+ if (commitKind == CommitKind.COMPACT) {
+ return scanCompactBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
+ if (commitKind == CommitKind.OVERWRITE) {
+ return scanOverwriteBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
+ return super.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ commitKind,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
+
+ private List<SimpleFileEntry> scanCompactBaseDataFiles(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ @Nullable CommitFailRetryResult previousAttempt,
+ boolean hasOverwriteSincePreviousAttempt) {
+ List<Range> changedRowRanges = changedRowRanges(deltaFiles,
indexFiles);
+ if (changedRowRanges.isEmpty()) {
+ return super.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ CommitKind.COMPACT,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
+ return scanChangedRowRanges(
+ latestSnapshot,
+ changedPartitions,
+ changedRowRanges,
+ referencedDataFiles(deltaFiles, indexFiles));
+ }
+
+ private List<SimpleFileEntry> scanOverwriteBaseDataFiles(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> indexFiles,
+ @Nullable CommitFailRetryResult previousAttempt,
+ boolean hasOverwriteSincePreviousAttempt) {
+ List<Range> changedRowRanges = changedRowRanges(deltaFiles,
indexFiles);
+ Set<String> referencedDataFiles = referencedDataFiles(deltaFiles,
indexFiles);
+ if (!changedRowRanges.isEmpty()) {
+ return scanChangedRowRanges(
+ latestSnapshot, changedPartitions, changedRowRanges,
referencedDataFiles);
+ }
+ if (!referencedDataFiles.isEmpty()) {
+ return commitScanner()
+ .readAllEntriesFromDataFiles(
+ latestSnapshot, changedPartitions,
referencedDataFiles);
+ }
+ return super.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ deltaFiles,
+ indexFiles,
+ CommitKind.OVERWRITE,
+ previousAttempt,
+ hasOverwriteSincePreviousAttempt);
+ }
+
+ private List<SimpleFileEntry> scanChangedRowRanges(
+ Snapshot latestSnapshot,
+ List<BinaryRow> changedPartitions,
+ List<Range> changedRowRanges,
+ Set<String> referencedDataFiles) {
+ List<SimpleFileEntry> baseDataFiles =
+ new ArrayList<>(
+ commitScanner()
+ .readAllEntriesFromChangedRowRanges(
+ latestSnapshot, changedPartitions,
changedRowRanges));
+ referencedDataFiles.removeAll(
+
baseDataFiles.stream().map(SimpleFileEntry::fileName).collect(Collectors.toSet()));
+ if (referencedDataFiles.isEmpty()) {
+ return baseDataFiles;
+ }
+
+ baseDataFiles.addAll(
+ commitScanner()
+ .readAllEntriesFromDataFiles(
+ latestSnapshot, changedPartitions,
referencedDataFiles));
+ return new ArrayList<>(
+ baseDataFiles.stream()
+ .collect(
+ Collectors.toMap(
+ FileEntry::identifier,
+ entry -> entry,
+ (left, right) -> left,
+ LinkedHashMap::new))
+ .values());
+ }
+
+ private Set<String> referencedDataFiles(
+ List<ManifestEntry> deltaFiles, List<IndexManifestEntry>
indexFiles) {
+ Set<String> referencedDataFiles =
+ deltaFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.DELETE)
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toSet());
+ for (IndexManifestEntry indexFile : indexFiles) {
+ if (indexFile.indexFile().dvRanges() != null) {
+
referencedDataFiles.addAll(indexFile.indexFile().dvRanges().keySet());
+ }
+ }
+ return referencedDataFiles;
+ }
+
+ private List<Range> changedRowRanges(
+ List<ManifestEntry> deltaFiles, List<IndexManifestEntry>
indexFiles) {
+ List<Range> ranges =
+ deltaFiles.stream()
+ .map(ManifestEntry::file)
+ .filter(file -> file.firstRowId() != null)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList());
+ indexFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .map(IndexManifestEntry::indexFile)
+ .map(IndexFileMeta::globalIndexMeta)
+ .filter(Objects::nonNull)
+ .map(GlobalIndexMeta::rowRange)
+ .forEach(ranges::add);
+ return Range.sortAndMergeOverlap(ranges, true);
+ }
+
+ @Override
+ protected Optional<RuntimeException> checkTableSpecificConflicts(
+ Snapshot latestSnapshot,
+ List<SimpleFileEntry> baseEntries,
+ List<SimpleFileEntry> deltaEntries,
+ List<IndexManifestEntry> deltaIndexEntries,
+ Collection<SimpleFileEntry> mergedEntries,
+ @Nullable RowIdConflictChecker rowIdConflictChecker,
+ CommitKind commitKind,
+ String baseCommitUser) {
+ Optional<RuntimeException> exception =
+ checkRowIdExistence(
+ baseEntries, deltaEntries, latestSnapshot.nextRowId(),
commitKind);
+ if (exception.isPresent()) {
+ return exception;
+ }
+
+ exception = checkRowIdRangeConflicts(commitKind, mergedEntries);
+ if (exception.isPresent()) {
+ return exception;
+ }
+
+ exception = checkGlobalIndexRowIdExistence(baseEntries,
deltaIndexEntries);
+ if (exception.isPresent()) {
+ return exception;
+ }
+
+ return checkForRowIdFromSnapshot(
+ latestSnapshot, deltaEntries, deltaIndexEntries,
rowIdConflictChecker);
+ }
+
+ private Optional<RuntimeException> checkRowIdRangeConflicts(
+ CommitKind commitKind, Collection<SimpleFileEntry> mergedEntries) {
+ if (rowIdCheckFromSnapshot == null && commitKind !=
CommitKind.COMPACT) {
+ return Optional.empty();
+ }
+
+ List<SimpleFileEntry> entries =
+ mergedEntries.stream()
+ .filter(file -> file.firstRowId() != null)
+ .collect(Collectors.toList());
+
+ RangeHelper<SimpleFileEntry> rangeHelper =
+ new RangeHelper<>(SimpleFileEntry::nonNullRowIdRange);
+ List<SimpleFileEntry> dataFiles =
+ entries.stream()
+ .filter(file -> !dedicatedStorageFile(file.fileName()))
+ .collect(Collectors.toList());
+
+ Optional<RuntimeException> exception =
+ checkDataFileRowIdRangeConflicts(rangeHelper, dataFiles);
+ if (exception.isPresent()) {
+ return exception;
+ }
+
+ List<SimpleFileEntry> dedicatedFiles =
+ entries.stream()
+ .filter(file -> dedicatedStorageFile(file.fileName()))
+ .collect(Collectors.toList());
+ return checkDedicatedFileRowIdRangeConflicts(dataFiles,
dedicatedFiles);
+ }
+
+ private Optional<RuntimeException> checkDataFileRowIdRangeConflicts(
+ RangeHelper<SimpleFileEntry> rangeHelper, List<SimpleFileEntry>
dataFiles) {
+ for (List<SimpleFileEntry> dataFileGroup :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
+ if (!rangeHelper.areAllRangesSame(dataFileGroup)) {
+ return Optional.of(
+ new RuntimeException(
+ "For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
+ + "operations "
+ + "have encountered conflicts, data
files: "
+ + dataFileGroup));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private Optional<RuntimeException> checkDedicatedFileRowIdRangeConflicts(
+ List<SimpleFileEntry> dataFiles, List<SimpleFileEntry>
dedicatedFiles) {
+ if (dedicatedFiles.isEmpty()) {
+ return Optional.empty();
+ }
+
+ RowRangeIndex dataFileRowRangeIndex = rowRangeIndex(dataFiles, false);
+ for (SimpleFileEntry dedicatedFile : dedicatedFiles) {
+ Range dedicatedRange = dedicatedFile.nonNullRowIdRange();
+ if (dataFileRowRangeIndex.contains(dedicatedRange)) {
+ continue;
+ }
+
+ List<Range> intersectingRanges =
+
dataFileRowRangeIndex.intersectedRanges(dedicatedRange.from, dedicatedRange.to);
+ List<SimpleFileEntry> intersectingDataFiles =
+ dataFiles.stream()
+ .filter(
+ dataFile ->
+ dataFile.nonNullRowIdRange()
+
.hasIntersection(dedicatedRange))
+ .collect(Collectors.toList());
+ String conflictReason =
+ intersectingRanges.size() > 1
+ ? "spans multiple data file ranges"
+ : "is not covered by one data file range";
+ return Optional.of(
+ new RuntimeException(
+ String.format(
+ "For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
+ + "operations have encountered
conflicts, dedicated "
+ + "file %s %s %s: %s",
+ dedicatedFile,
+ dedicatedRange,
+ conflictReason,
+ intersectingDataFiles)));
+ }
+ return Optional.empty();
+ }
+
+ private Optional<RuntimeException> checkForRowIdFromSnapshot(
+ Snapshot latestSnapshot,
+ List<SimpleFileEntry> deltaEntries,
+ List<IndexManifestEntry> deltaIndexEntries,
+ @Nullable RowIdConflictChecker conflictChecker) {
+ if (rowIdCheckFromSnapshot == null
+ || conflictChecker == null
+ || conflictChecker.isEmpty()) {
+ return Optional.empty();
+ }
+
+ List<BinaryRow> changedPartitions = changedPartitions(deltaEntries,
deltaIndexEntries);
+ Long checkNextRowId =
snapshotManager.snapshot(rowIdCheckFromSnapshot).nextRowId();
+ checkState(
+ checkNextRowId != null,
+ "Next row id cannot be null for snapshot %s.",
+ rowIdCheckFromSnapshot);
+ for (long i = rowIdCheckFromSnapshot + 1; i <= latestSnapshot.id();
i++) {
+ Snapshot snapshot = snapshotManager.snapshot(i);
+ if (snapshot.commitKind() == CommitKind.COMPACT) {
+ continue;
+ }
+ List<ManifestEntry> changes =
+ commitScanner().readIncrementalEntries(snapshot,
changedPartitions);
+ for (ManifestEntry entry : changes) {
+ if (!shouldCheckHistoricalRowIdEntry(entry.kind())) {
+ continue;
+ }
+ DataFileMeta file = entry.file();
+ if (file.firstRowId() != null
+ && file.nonNullRowIdRange().from < checkNextRowId
+ && conflictChecker.conflictsWith(file)) {
+ LOG.debug(
+ "Data evolution row id conflict detected for table
{}, commit user {}, "
+ + "snapshot {}, file {}.",
+ tableName,
+ commitUser,
+ snapshot.id(),
+ file);
+ return Optional.of(
+ new RuntimeException(
+
ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE));
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ boolean shouldCheckHistoricalRowIdEntry(FileKind kind) {
+ return rowIdConflictCheckStrategy().shouldCheckHistoricalEntry(kind);
+ }
+
+ private interface RowIdConflictCheckStrategy {
+
+ boolean appliesTo(CommitKind commitKind);
+
+ RowIdConflictChecker createChecker(
+ SchemaManager schemaManager, List<ManifestEntry> deltaFiles);
+
+ boolean shouldCheckHistoricalEntry(FileKind kind);
+ }
+
+ private static class DataEvolutionDmlRowIdConflictCheck implements
RowIdConflictCheckStrategy {
+
+ private static final DataEvolutionDmlRowIdConflictCheck INSTANCE =
+ new DataEvolutionDmlRowIdConflictCheck();
+
+ @Override
+ public boolean appliesTo(CommitKind commitKind) {
+ return true;
+ }
+
+ @Override
+ public RowIdConflictChecker createChecker(
+ SchemaManager schemaManager, List<ManifestEntry> deltaFiles) {
+ return RowIdColumnConflictChecker.fromDataFiles(
+ schemaManager,
+
deltaFiles.stream().map(ManifestEntry::file).collect(Collectors.toList()));
+ }
+
+ @Override
+ public boolean shouldCheckHistoricalEntry(FileKind kind) {
+ return true;
+ }
+ }
+
+ private static class MaterializeDvRowIdConflictCheck implements
RowIdConflictCheckStrategy {
+
+ private static final MaterializeDvRowIdConflictCheck INSTANCE =
+ new MaterializeDvRowIdConflictCheck();
+
+ @Override
+ public boolean appliesTo(CommitKind commitKind) {
+ return commitKind == CommitKind.COMPACT;
+ }
+
+ @Override
+ public RowIdConflictChecker createChecker(
+ SchemaManager schemaManager, List<ManifestEntry> deltaFiles) {
+ // Materializing deletion vectors rewrites complete row ranges. A
concurrent ADD in a
+ // deleted normal-file range can otherwise restore logically
deleted rows.
+ List<DataFileMeta> deletedNormalFiles =
+ deltaFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.DELETE)
+ .map(ManifestEntry::file)
+ .filter(file -> file.firstRowId() != null)
+ .filter(
+ file ->
+ !isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ .collect(Collectors.toList());
+ return RowIdRangeConflictChecker.fromDataFiles(deletedNormalFiles);
+ }
+
+ @Override
+ public boolean shouldCheckHistoricalEntry(FileKind kind) {
+ return kind == FileKind.ADD;
+ }
+ }
+
+ private Optional<RuntimeException> checkGlobalIndexRowIdExistence(
+ List<SimpleFileEntry> baseEntries, List<IndexManifestEntry>
deltaIndexEntries) {
+ List<IndexManifestEntry> indexesToCheck =
globalIndexFileAdditions(deltaIndexEntries);
+ if (indexesToCheck.isEmpty()) {
+ return Optional.empty();
+ }
+
+ Map<Pair<BinaryRow, Integer>, List<Range>> dataRanges = new
HashMap<>();
+ for (SimpleFileEntry entry : baseEntries) {
+ if (entry.kind() == FileKind.ADD && entry.firstRowId() != null) {
+ dataRanges
+ .computeIfAbsent(
+ Pair.of(entry.partition(), entry.bucket()), k
-> new ArrayList<>())
+ .add(entry.nonNullRowIdRange());
+ }
+ }
+ Map<Pair<BinaryRow, Integer>, RowRangeIndex> rowRangeIndexes =
+ dataRanges.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
RowRangeIndex.create(entry.getValue())));
+
+ for (IndexManifestEntry indexEntry : indexesToCheck) {
+ GlobalIndexMeta globalIndex =
indexEntry.indexFile().globalIndexMeta();
+ checkState(globalIndex != null, "Global index meta must not be
null.");
+ Range indexRange = globalIndex.rowRange();
+ RowRangeIndex rowRangeIndex =
+ rowRangeIndexes.get(Pair.of(indexEntry.partition(),
indexEntry.bucket()));
+ if (rowRangeIndex == null || !rowRangeIndex.contains(indexRange)) {
+ return Optional.of(
+ new RuntimeException(
+ String.format(
+ "Global index row ID existence
conflict: index file '%s' "
+ + "references row range %s,
but this range "
+ + "is not fully covered by
current data "
+ + "files. The referenced row
IDs may have been "
+ + "reassigned or removed by a
concurrent commit.",
+ indexEntry.indexFile().fileName(),
indexRange)));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private List<IndexManifestEntry> globalIndexFileAdditions(
+ List<IndexManifestEntry> indexFileChanges) {
+ List<IndexManifestEntry> result = new ArrayList<>();
+ for (IndexManifestEntry entry : indexFileChanges) {
+ if (entry.kind() == FileKind.ADD &&
entry.indexFile().globalIndexMeta() != null) {
+ result.add(entry);
+ }
+ }
+ return result;
+ }
+
+ Optional<RuntimeException> checkRowIdExistence(
+ List<SimpleFileEntry> baseEntries,
+ List<SimpleFileEntry> deltaEntries,
+ @Nullable Long nextRowId,
+ CommitKind commitKind) {
+ List<SimpleFileEntry> existingDataFiles =
+ baseEntries.stream()
+ .filter(
+ base ->
+ base.firstRowId() != null
+ &&
!dedicatedStorageFile(base.fileName()))
+ .collect(Collectors.toList());
+
+ if (commitKind == CommitKind.COMPACT) {
+ return checkCompactRowIdExistence(existingDataFiles, deltaEntries);
+ }
+ return checkNonCompactRowIdExistence(existingDataFiles, deltaEntries,
nextRowId);
+ }
+
+ /**
+ * Checks conflicts between compaction and concurrent Row ID reassignment,
which may otherwise
+ * cause reassigned Row IDs to fall back. For example, compaction produces
a file with Row IDs
+ * [0, 9], then reassignment moves the current range to [10, 19].
Committing the stale
+ * compaction would move the Row IDs back to [0, 9].
+ */
+ private Optional<RuntimeException> checkCompactRowIdExistence(
+ List<SimpleFileEntry> existingDataFiles, List<SimpleFileEntry>
deltaEntries) {
+ Map<Pair<BinaryRow, Integer>, List<SimpleFileEntry>> dataFilesByBucket
=
+ existingDataFiles.stream()
+ .collect(
+ Collectors.groupingBy(
+ file -> Pair.of(file.partition(),
file.bucket())));
+ Map<Pair<BinaryRow, Integer>, RowRangeIndex> existingIndexes =
+ dataFilesByBucket.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
rowRangeIndex(entry.getValue(), true)));
+
+ for (SimpleFileEntry entry : deltaEntries) {
+ if (entry.kind() != FileKind.ADD || entry.firstRowId() == null) {
+ continue;
+ }
+ RowRangeIndex existingIndex =
+ existingIndexes.get(Pair.of(entry.partition(),
entry.bucket()));
+ if (existingIndex == null ||
!existingIndex.contains(entry.nonNullRowIdRange())) {
+ return Optional.of(rowIdExistenceConflict(entry));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private Optional<RuntimeException> checkNonCompactRowIdExistence(
+ List<SimpleFileEntry> existingDataFiles,
+ List<SimpleFileEntry> deltaEntries,
+ @Nullable Long nextRowId) {
+ List<SimpleFileEntry> filesToCheck =
+ deltaEntries.stream()
+ .filter(
+ entry ->
+ entry.kind() == FileKind.ADD
+ && entry.firstRowId() != null
+ && nextRowId != null
+ && entry.firstRowId() <
nextRowId)
+ .collect(Collectors.toList());
+ if (filesToCheck.isEmpty()) {
+ return Optional.empty();
+ }
+
+ RowRangeIndex existingIndex = rowRangeIndex(existingDataFiles, false);
+ for (SimpleFileEntry entry : filesToCheck) {
+ Range rowRange = entry.nonNullRowIdRange();
+ boolean exists =
+ dedicatedStorageFile(entry.fileName())
+ ? existingIndex.contains(rowRange)
+ : existingIndex.containsExactly(rowRange);
+ if (!exists) {
+ return Optional.of(rowIdExistenceConflict(entry));
+ }
+ }
+ return Optional.empty();
+ }
+
+ private RowIdExistenceConflictException
rowIdExistenceConflict(SimpleFileEntry entry) {
+ return new RowIdExistenceConflictException(
+ entry.fileName(), entry.firstRowId(), entry.rowCount(),
entry.bucket());
+ }
+
+ private static RowRangeIndex rowRangeIndex(
+ Collection<SimpleFileEntry> files, boolean mergeAdjacent) {
+ return RowRangeIndex.create(
+
files.stream().map(SimpleFileEntry::nonNullRowIdRange).collect(Collectors.toList()),
+ mergeAdjacent);
+ }
+
+ private static boolean dedicatedStorageFile(String fileName) {
+ return isBlobFile(fileName) || isVectorStoreFile(fileName);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/PrimaryKeyConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/PrimaryKeyConflictDetection.java
new file mode 100644
index 0000000000..2e8b4d34e3
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/PrimaryKeyConflictDetection.java
@@ -0,0 +1,163 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Conflict detection for primary-key tables. */
+public class PrimaryKeyConflictDetection extends ConflictDetection {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PrimaryKeyConflictDetection.class);
+
+ private final FileStorePathFactory pathFactory;
+ private final Comparator<InternalRow> keyComparator;
+ private final boolean pkClusteringOverride;
+
+ public PrimaryKeyConflictDetection(
+ String tableName,
+ String commitUser,
+ RowType partitionType,
+ FileStorePathFactory pathFactory,
+ Comparator<InternalRow> keyComparator,
+ BucketMode bucketMode,
+ boolean deletionVectorsEnabled,
+ boolean pkClusteringOverride,
+ IndexFileHandler indexFileHandler,
+ CommitScanner commitScanner) {
+ super(
+ tableName,
+ commitUser,
+ partitionType,
+ pathFactory,
+ bucketMode,
+ deletionVectorsEnabled,
+ indexFileHandler,
+ commitScanner);
+ this.pathFactory = pathFactory;
+ this.keyComparator = keyComparator;
+ this.pkClusteringOverride = pkClusteringOverride;
+ }
+
+ @Override
+ public Comparator<InternalRow> keyComparator() {
+ return keyComparator;
+ }
+
+ @Override
+ protected Optional<RuntimeException> checkTableSpecificConflicts(
+ Snapshot latestSnapshot,
+ List<SimpleFileEntry> baseEntries,
+ List<SimpleFileEntry> deltaEntries,
+ List<IndexManifestEntry> deltaIndexEntries,
+ Collection<SimpleFileEntry> mergedEntries,
+ @Nullable RowIdConflictChecker rowIdConflictChecker,
+ CommitKind commitKind,
+ String baseCommitUser) {
+ if (pkClusteringOverride) {
+ return Optional.empty();
+ }
+
+ Map<LevelIdentifier, List<SimpleFileEntry>> levels = new HashMap<>();
+ for (SimpleFileEntry entry : mergedEntries) {
+ int level = entry.level();
+ if (level >= 1) {
+ levels.computeIfAbsent(
+ new LevelIdentifier(entry.partition(),
entry.bucket(), level),
+ lv -> new ArrayList<>())
+ .add(entry);
+ }
+ }
+
+ for (List<SimpleFileEntry> entries : levels.values()) {
+ entries.sort((a, b) -> keyComparator.compare(a.minKey(),
b.minKey()));
+ for (int i = 0; i + 1 < entries.size(); i++) {
+ SimpleFileEntry a = entries.get(i);
+ SimpleFileEntry b = entries.get(i + 1);
+ if (keyComparator.compare(a.maxKey(), b.minKey()) >= 0) {
+ Pair<RuntimeException, RuntimeException> conflictException
=
+ createConflictException(
+ "LSM conflicts detected! Give up
committing. Conflict files are:\n"
+ +
a.identifier().toString(pathFactory)
+ + "\n"
+ +
b.identifier().toString(pathFactory),
+ baseCommitUser,
+ baseEntries,
+ deltaEntries,
+ null);
+ LOG.warn("", conflictException.getLeft());
+ return Optional.of(conflictException.getRight());
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ private static class LevelIdentifier {
+
+ private final BinaryRow partition;
+ private final int bucket;
+ private final int level;
+
+ private LevelIdentifier(BinaryRow partition, int bucket, int level) {
+ this.partition = partition;
+ this.bucket = bucket;
+ this.level = level;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof LevelIdentifier)) {
+ return false;
+ }
+ LevelIdentifier that = (LevelIdentifier) o;
+ return Objects.equals(partition, that.partition)
+ && bucket == that.bucket
+ && level == that.level;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partition, bucket, level);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdConflictChecker.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdConflictChecker.java
index e5e9808946..fb3614bfec 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdConflictChecker.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdConflictChecker.java
@@ -26,10 +26,4 @@ public interface RowIdConflictChecker {
boolean isEmpty();
boolean conflictsWith(DataFileMeta file);
-
- /** Defines the source operation which triggers this conflict check. */
- enum TriggerSource {
- DATA_EVOLUTION_DML,
- MATERIALIZE_DV_COMPACTION
- }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index f326be18dc..d427345fa7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1998,7 +1998,7 @@ public class FileStoreCommitTest {
Collections.emptyList(),
Collections.emptyList(),
scanner ->
- new ConflictDetection(
+ ConflictDetection.create(
tableName,
commitUser,
store.partitionType(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
index 71757cb49d..e200faaa0e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
@@ -23,13 +23,19 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.SimpleFileEntry;
import org.apache.paimon.manifest.SimpleFileEntryWithDV;
+import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
@@ -39,9 +45,11 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
@@ -50,9 +58,376 @@ import static org.apache.paimon.manifest.FileKind.DELETE;
import static
org.apache.paimon.operation.commit.ConflictDetection.buildBaseEntriesWithDV;
import static
org.apache.paimon.operation.commit.ConflictDetection.buildDeltaEntriesWithDV;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
class ConflictDetectionTest {
+ @Test
+ void testCreateConflictDetectionByTableType() {
+ ConflictDetection append = createConflictDetection(null, false, false);
+ assertThat(append).isInstanceOf(AppendConflictDetection.class);
+ assertThat(append.keyComparator()).isNull();
+ assertThat(createConflictDetection(null, false, true))
+ .isInstanceOf(PrimaryKeyConflictDetection.class);
+ assertThat(createConflictDetection(null, true, false))
+ .isInstanceOf(DataEvolutionConflictDetection.class);
+ // Data Evolution takes precedence even if a key comparator is
supplied.
+ assertThat(createConflictDetection(null, true, true))
+ .isInstanceOf(DataEvolutionConflictDetection.class);
+ }
+
+ @Test
+ void testPrimaryKeyClusteringOverrideSkipsLsmConflictCheck() {
+ ConflictDetection detection = createConflictDetection(null, false,
true, true, null);
+
+ assertThat(
+ detection.checkConflicts(
+ snapshot(1),
+
Collections.singletonList(createLevelFileEntry("base", 1)),
+
Collections.singletonList(createLevelFileEntry("delta", 1)),
+ Collections.emptyList(),
+ null,
+ Snapshot.CommitKind.COMPACT))
+ .isEmpty();
+ }
+
+ @Test
+ void testDataEvolutionCompactScansChangedRowRanges() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ BinaryRow partition = BinaryRow.singleColumn(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(partition);
+ Range changedRange = new Range(10, 19);
+
+ ManifestEntry delta = mock(ManifestEntry.class);
+ DataFileMeta dataFile = mock(DataFileMeta.class);
+ when(delta.kind()).thenReturn(ADD);
+ when(delta.file()).thenReturn(dataFile);
+ when(dataFile.firstRowId()).thenReturn(10L);
+ when(dataFile.nonNullRowIdRange()).thenReturn(changedRange);
+ when(scanner.readAllEntriesFromChangedRowRanges(
+ snapshot, changedPartitions,
Collections.singletonList(changedRange)))
+ .thenReturn(Collections.emptyList());
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.singletonList(delta),
+ Collections.emptyList(),
+ Snapshot.CommitKind.COMPACT,
+ null,
+ false))
+ .isEmpty();
+ verify(scanner)
+ .readAllEntriesFromChangedRowRanges(
+ snapshot, changedPartitions,
Collections.singletonList(changedRange));
+ }
+
+ @Test
+ void testDataEvolutionCompactSupplementsReferencedDataFiles() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ BinaryRow partition = BinaryRow.singleColumn(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(partition);
+ Range changedRange = new Range(10, 19);
+
+ ManifestEntry added = manifestEntry(ADD, "added", 10L, changedRange);
+ ManifestEntry deleted = manifestEntry(DELETE, "legacy", null, null);
+ SimpleFileEntry rangeBase =
+ createFileEntryWithRowId("range-base", ADD, partition, 0, 10L,
10L);
+ SimpleFileEntry legacyBase = createFileEntry("legacy", ADD);
+ SimpleFileEntry dvBase = createFileEntry("dv-base", ADD);
+ Set<String> referencedFiles = new HashSet<>(Arrays.asList("legacy",
"dv-base"));
+ when(scanner.readAllEntriesFromChangedRowRanges(
+ snapshot, changedPartitions,
Collections.singletonList(changedRange)))
+ .thenReturn(Collections.singletonList(rangeBase));
+ when(scanner.readAllEntriesFromDataFiles(snapshot, changedPartitions,
referencedFiles))
+ .thenReturn(Arrays.asList(legacyBase, dvBase));
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Arrays.asList(added, deleted),
+ Collections.singletonList(
+ createDvIndexEntry(
+ "dv", ADD,
Collections.singletonList("dv-base"))),
+ Snapshot.CommitKind.COMPACT,
+ null,
+ false))
+ .containsExactly(rangeBase, legacyBase, dvBase);
+ verify(scanner).readAllEntriesFromDataFiles(snapshot,
changedPartitions, referencedFiles);
+ }
+
+ @Test
+ void testDataEvolutionCompactWithoutRowRangesReusesRetryScan() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot previousSnapshot = snapshot(1);
+ Snapshot latestSnapshot = snapshot(2);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ SimpleFileEntry oldBase = createFileEntry("old", ADD);
+ SimpleFileEntry removedBase = createFileEntry("old", DELETE);
+ SimpleFileEntry newBase = createFileEntry("new", ADD);
+ List<SimpleFileEntry> cachedBase = Collections.singletonList(oldBase);
+ CommitFailRetryResult previousAttempt =
commitFailRetryResult(previousSnapshot, cachedBase);
+ when(scanner.readIncrementalChanges(previousSnapshot, latestSnapshot,
changedPartitions))
+ .thenReturn(Arrays.asList(removedBase, newBase));
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.COMPACT,
+ previousAttempt,
+ false))
+ .containsExactly(newBase);
+ assertThat(cachedBase).containsExactly(oldBase);
+ verify(scanner, never())
+ .readAllEntriesFromChangedPartitions(latestSnapshot,
changedPartitions);
+ }
+
+ @Test
+ void testRetryFallsBackWhenCachedScanCannotBeReused() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ ConflictDetection detection = createConflictDetection(scanner, false,
false);
+ Snapshot previousSnapshot = snapshot(1);
+ Snapshot latestSnapshot = snapshot(2);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ List<SimpleFileEntry> cachedBase =
+ Collections.singletonList(createFileEntry("cached", ADD));
+ List<SimpleFileEntry> expected =
Collections.singletonList(createFileEntry("latest", ADD));
+ when(scanner.readAllEntriesFromChangedPartitions(latestSnapshot,
changedPartitions))
+ .thenReturn(expected);
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.APPEND,
+ commitFailRetryResult(null, cachedBase),
+ false))
+ .isSameAs(expected);
+ assertThat(
+ detection.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.APPEND,
+ commitFailRetryResult(previousSnapshot, null),
+ false))
+ .isSameAs(expected);
+ assertThat(
+ detection.scanBaseDataFiles(
+ latestSnapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.APPEND,
+ commitFailRetryResult(previousSnapshot,
cachedBase),
+ true))
+ .isSameAs(expected);
+ verify(scanner, times(3))
+ .readAllEntriesFromChangedPartitions(latestSnapshot,
changedPartitions);
+ verify(scanner, never())
+ .readIncrementalChanges(previousSnapshot, latestSnapshot,
changedPartitions);
+ }
+
+ @Test
+ void testDataEvolutionOverwriteScansDeletedFileRowRange() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ BinaryRow partition = BinaryRow.singleColumn(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(partition);
+ Range changedRange = new Range(10, 19);
+
+ ManifestEntry deleted = manifestEntry(DELETE, "deleted", 10L,
changedRange);
+ SimpleFileEntry base = createFileEntryWithRowId("deleted", ADD,
partition, 0, 10L, 10L);
+ when(scanner.readAllEntriesFromChangedRowRanges(
+ snapshot, changedPartitions,
Collections.singletonList(changedRange)))
+ .thenReturn(Collections.singletonList(base));
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.singletonList(deleted),
+ Collections.emptyList(),
+ Snapshot.CommitKind.OVERWRITE,
+ null,
+ false))
+ .containsExactly(base);
+ verify(scanner, never())
+ .readAllEntriesFromDataFiles(
+ snapshot, changedPartitions,
Collections.singleton("deleted"));
+ verify(scanner, never()).readAllEntriesFromChangedPartitions(snapshot,
changedPartitions);
+ }
+
+ @Test
+ void testDataEvolutionOverwriteScansDeletionVectorDataFile() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ SimpleFileEntry base = createFileEntry("base", ADD);
+ when(scanner.readAllEntriesFromDataFiles(
+ snapshot, changedPartitions,
Collections.singleton("base")))
+ .thenReturn(Collections.singletonList(base));
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.singletonList(
+ createDvIndexEntry(
+ "dv", ADD,
Collections.singletonList("base"))),
+ Snapshot.CommitKind.OVERWRITE,
+ null,
+ false))
+ .containsExactly(base);
+ verify(scanner, never()).readAllEntriesFromChangedPartitions(snapshot,
changedPartitions);
+ }
+
+ @Test
+ void testDataEvolutionOverwriteSupplementsLegacyDeletedFile() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ BinaryRow partition = BinaryRow.singleColumn(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(partition);
+ Range changedRange = new Range(10, 19);
+
+ ManifestEntry added = manifestEntry(ADD, "added", 10L, changedRange);
+ ManifestEntry deleted = manifestEntry(DELETE, "legacy", null, null);
+ SimpleFileEntry rangeBase =
+ createFileEntryWithRowId("range-base", ADD, partition, 0, 10L,
10L);
+ SimpleFileEntry legacyBase = createFileEntry("legacy", ADD);
+ when(scanner.readAllEntriesFromChangedRowRanges(
+ snapshot, changedPartitions,
Collections.singletonList(changedRange)))
+ .thenReturn(Collections.singletonList(rangeBase));
+ when(scanner.readAllEntriesFromDataFiles(
+ snapshot, changedPartitions,
Collections.singleton("legacy")))
+ .thenReturn(Collections.singletonList(legacyBase));
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Arrays.asList(added, deleted),
+ Collections.emptyList(),
+ Snapshot.CommitKind.OVERWRITE,
+ null,
+ false))
+ .containsExactly(rangeBase, legacyBase);
+ verify(scanner, never()).readAllEntriesFromChangedPartitions(snapshot,
changedPartitions);
+ }
+
+ @Test
+ void testDataEvolutionOverwriteFallsBackWithoutSelectors() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ List<SimpleFileEntry> expected =
Collections.singletonList(createFileEntry("base", ADD));
+ when(scanner.readAllEntriesFromChangedPartitions(snapshot,
changedPartitions))
+ .thenReturn(expected);
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.OVERWRITE,
+ null,
+ false))
+ .isSameAs(expected);
+ }
+
+ @Test
+ void testDataEvolutionAppendUsesDefaultPartitionScan() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
createConflictDetection(scanner, true, false);
+ Snapshot snapshot = snapshot(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ List<SimpleFileEntry> expected =
Collections.singletonList(createFileEntry("base", ADD));
+ when(scanner.readAllEntriesFromChangedPartitions(snapshot,
changedPartitions))
+ .thenReturn(expected);
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.APPEND,
+ null,
+ false))
+ .isSameAs(expected);
+ }
+
+ @Test
+ void testAppendScansChangedPartitions() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ ConflictDetection detection = createConflictDetection(scanner, false,
false);
+ Snapshot snapshot = snapshot(1);
+ List<BinaryRow> changedPartitions =
Collections.singletonList(BinaryRow.singleColumn(1));
+ List<SimpleFileEntry> expected =
Collections.singletonList(createFileEntry("base", ADD));
+ when(scanner.readAllEntriesFromChangedPartitions(snapshot,
changedPartitions))
+ .thenReturn(expected);
+
+ assertThat(
+ detection.scanBaseDataFiles(
+ snapshot,
+ changedPartitions,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Snapshot.CommitKind.APPEND,
+ null,
+ false))
+ .isSameAs(expected);
+ verify(scanner).readAllEntriesFromChangedPartitions(snapshot,
changedPartitions);
+ }
+
+ @Test
+ void testAppendSkipsDataEvolutionConflictChecks() {
+ ConflictDetection detection = createConflictDetection(null, false,
false);
+
+ Optional<RuntimeException> exception =
+ detection.checkConflicts(
+ snapshot(1),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonList(
+ createGlobalIndexEntry("idx", ADD,
BinaryRow.EMPTY_ROW, 0, 149)),
+ null,
+ Snapshot.CommitKind.APPEND);
+
+ assertThat(exception).isEmpty();
+ }
+
@Test
public void testBuildBaseEntriesWithDV() {
{
@@ -314,6 +689,23 @@ class ConflictDetectionTest {
null);
}
+ private SimpleFileEntry createLevelFileEntry(String fileName, int level) {
+ return new SimpleFileEntry(
+ ADD,
+ EMPTY_ROW,
+ 0,
+ 1,
+ level,
+ fileName,
+ Collections.emptyList(),
+ null,
+ EMPTY_ROW,
+ EMPTY_ROW,
+ null,
+ 0L,
+ null);
+ }
+
private SimpleFileEntryWithDV createFileEntryWithDV(
String fileName, FileKind kind, @Nullable String dvFileName) {
return new SimpleFileEntryWithDV(createFileEntry(fileName, kind),
dvFileName);
@@ -369,7 +761,7 @@ class ConflictDetectionTest {
@Test
void testShouldBeOverwriteCommit() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> addOnlyEntries = new ArrayList<>();
addOnlyEntries.add(createFileEntry("f1", ADD));
@@ -400,7 +792,7 @@ class ConflictDetectionTest {
@Test
void testMaterializeDvRowIdCheckOnlyAppliesToCompactCommit() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
detection.setRowIdCheckFromSnapshotForMaterializeDvCompaction(1L);
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.APPEND)).isFalse();
@@ -409,17 +801,98 @@ class ConflictDetectionTest {
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.ADD)).isTrue();
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.DELETE)).isFalse();
+ RowIdConflictChecker checker =
+ detection.createRowIdConflictChecker(
+ mock(SchemaManager.class),
+ Arrays.asList(
+ manifestEntry(DELETE, "deleted", 10L, new
Range(10, 19)),
+ manifestEntry(ADD, "added", 30L, new Range(30,
39)),
+ manifestEntry(DELETE, "dedicated.blob", 50L,
new Range(50, 59)),
+ manifestEntry(
+ DELETE, "dedicated.vector.data", 70L,
new Range(70, 79))),
+ Snapshot.CommitKind.COMPACT);
+ assertThat(checker).isNotNull();
+ assertThat(
+ checker.conflictsWith(
+ manifestEntry(ADD, "historical", 15L, new
Range(15, 24)).file()))
+ .isTrue();
+ assertThat(
+ checker.conflictsWith(
+ manifestEntry(ADD, "historical", 35L, new
Range(35, 44)).file()))
+ .isFalse();
+ assertThat(
+ checker.conflictsWith(
+ manifestEntry(ADD, "historical", 55L, new
Range(55, 64)).file()))
+ .isFalse();
+ assertThat(
+ checker.conflictsWith(
+ manifestEntry(ADD, "historical", 75L, new
Range(75, 84)).file()))
+ .isFalse();
+
detection.setRowIdCheckFromSnapshot(1L);
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.APPEND)).isTrue();
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.COMPACT)).isTrue();
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.ADD)).isTrue();
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.DELETE)).isTrue();
+ assertThat(
+ detection.createRowIdConflictChecker(
+ mock(SchemaManager.class),
+ Collections.singletonList(manifestEntry(ADD,
"added", null, null)),
+ Snapshot.CommitKind.APPEND))
+ .isInstanceOf(RowIdColumnConflictChecker.class);
detection.setRowIdCheckFromSnapshot(null);
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.APPEND)).isFalse();
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.COMPACT)).isFalse();
}
+ @Test
+ void testMaterializeRowIdCheckSkipsCompactSnapshotsAndHistoricalDeletes() {
+ CommitScanner scanner = mock(CommitScanner.class);
+ SnapshotManager snapshotManager = mock(SnapshotManager.class);
+ DataEvolutionConflictDetection detection =
+ (DataEvolutionConflictDetection)
+ createConflictDetection(scanner, true, false, false,
snapshotManager);
+ detection.setRowIdCheckFromSnapshotForMaterializeDvCompaction(1L);
+
+ Snapshot checkSnapshot = mock(Snapshot.class);
+ Snapshot compactSnapshot = mock(Snapshot.class);
+ Snapshot appendSnapshot = mock(Snapshot.class);
+ Snapshot latestSnapshot = mock(Snapshot.class);
+ when(checkSnapshot.nextRowId()).thenReturn(20L);
+
when(compactSnapshot.commitKind()).thenReturn(Snapshot.CommitKind.COMPACT);
+
when(appendSnapshot.commitKind()).thenReturn(Snapshot.CommitKind.APPEND);
+ when(latestSnapshot.id()).thenReturn(3L);
+ when(latestSnapshot.commitUser()).thenReturn("other-user");
+ when(snapshotManager.snapshot(1)).thenReturn(checkSnapshot);
+ when(snapshotManager.snapshot(2)).thenReturn(compactSnapshot);
+ when(snapshotManager.snapshot(3)).thenReturn(appendSnapshot);
+
+ ManifestEntry historicalDelete =
+ manifestEntry(DELETE, "historical", 10L, new Range(10, 19));
+ ManifestEntry historicalAdd = manifestEntry(ADD, "non-conflicting",
10L, new Range(10, 19));
+ when(scanner.readIncrementalEntries(appendSnapshot,
Collections.emptyList()))
+ .thenReturn(Arrays.asList(historicalDelete, historicalAdd));
+ RowIdConflictChecker checker = mock(RowIdConflictChecker.class);
+ when(checker.isEmpty()).thenReturn(false);
+ when(checker.conflictsWith(historicalDelete.file())).thenReturn(true);
+ when(checker.conflictsWith(historicalAdd.file())).thenReturn(false);
+
+ assertThat(
+ detection.checkConflicts(
+ latestSnapshot,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ checker,
+ Snapshot.CommitKind.COMPACT))
+ .isEmpty();
+ verify(scanner, never()).readIncrementalEntries(compactSnapshot,
Collections.emptyList());
+ verify(scanner).readIncrementalEntries(appendSnapshot,
Collections.emptyList());
+ verify(checker, never()).conflictsWith(historicalDelete.file());
+ verify(checker).conflictsWith(historicalAdd.file());
+ }
+
@Test
void testChangedPartitionsIncludesGlobalIndexFiles() {
BinaryRow partition = BinaryRow.singleColumn(1);
@@ -434,7 +907,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceNoConflict() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 100L));
@@ -450,7 +923,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceBaseFileRemoved() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
@@ -468,7 +941,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceBaseFileRewritten() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f2", ADD, 0L, 200L));
@@ -487,7 +960,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceNormalFileRejectsAdjacentDataFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
@@ -505,7 +978,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 4L));
@@ -521,7 +994,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceDedicatedFileRejectsAdjacentDataFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
@@ -539,7 +1012,7 @@ class ConflictDetectionTest {
@Test
void
testCheckRowIdExistenceDedicatedFileRejectsRangeNotCoveredByOneDataFile() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
@@ -556,7 +1029,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceDedicatedFileIgnoresBaseDedicatedFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
baseEntries.add(createFileEntryWithRowId("old.blob", ADD, 0L, 2L));
@@ -573,7 +1046,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceSkipsNewlyAppendedFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
// nextRowId=100: files with firstRowId >= 100 are newly appended, not
references
List<SimpleFileEntry> baseEntries = new ArrayList<>();
@@ -593,7 +1066,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceSkipsNonPreAssigned() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
@@ -608,7 +1081,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceSkipsDeleteEntries() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
@@ -623,7 +1096,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceSkipsWhenNextRowIdNull() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries = new ArrayList<>();
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
@@ -637,7 +1110,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceCompactAllowsAdjacentNormalRanges() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries =
Arrays.asList(
@@ -654,7 +1127,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceCompactAllowsBlobAcrossAdjacentNormalRanges() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries =
Arrays.asList(
@@ -671,7 +1144,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdExistenceCompactRejectsStaleRangeAfterReassign() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
List<SimpleFileEntry> baseEntries =
Arrays.asList(
@@ -681,15 +1154,20 @@ class ConflictDetectionTest {
Collections.singletonList(createFileEntryWithRowId("compacted", ADD, 0L, 4L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(
- baseEntries, deltaEntries, 14L,
Snapshot.CommitKind.COMPACT);
+ detection.checkConflicts(
+ snapshot(1),
+ baseEntries,
+ deltaEntries,
+ Collections.emptyList(),
+ null,
+ Snapshot.CommitKind.COMPACT);
assertThat(result).isPresent();
assertThat(result.get()).hasMessageContaining("Row ID existence
conflict");
}
@Test
void testCheckRowIdExistenceCompactDoesNotMergeAcrossPartitions() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
BinaryRow partition0 = BinaryRow.singleColumn(0);
BinaryRow partition1 = BinaryRow.singleColumn(1);
@@ -710,7 +1188,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdRangeConflictsReportsDedicatedFileSpanningDataFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -732,9 +1210,30 @@ class ConflictDetectionTest {
.hasMessageContaining("f2");
}
+ @Test
+ void testCheckRowIdRangeConflictsRejectsOverlappingNormalFiles() {
+ DataEvolutionConflictDetection detection = createConflictDetection();
+
+ Optional<RuntimeException> exception =
+ detection.checkConflicts(
+ snapshot(1),
+
Collections.singletonList(createFileEntryWithRowId("base", ADD, 0L, 5L)),
+ Collections.singletonList(
+ createFileEntryWithRowId("compacted", ADD, 2L,
2L)),
+ Collections.emptyList(),
+ null,
+ Snapshot.CommitKind.COMPACT);
+
+ assertThat(exception).isPresent();
+ assertThat(exception.get())
+ .hasMessageContaining("multiple 'MERGE INTO' and 'COMPACT'
operations")
+ .hasMessageContaining("base")
+ .hasMessageContaining("compacted");
+ }
+
@Test
void testCheckRowIdRangeConflictsAllowsAdjacentDataFiles() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -752,7 +1251,7 @@ class ConflictDetectionTest {
@Test
void testCheckRowIdRangeConflictsAllowsDedicatedFileCoveredByOneDataFile()
{
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -796,7 +1295,7 @@ class ConflictDetectionTest {
@Test
void testCheckGlobalIndexRowIdExistenceNoConflict() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -815,7 +1314,7 @@ class ConflictDetectionTest {
@Test
void testCheckGlobalIndexRowIdExistenceBaseFileRemoved() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -836,7 +1335,7 @@ class ConflictDetectionTest {
@Test
void testCheckGlobalIndexRowIdExistenceByPartitionAndBucket() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
BinaryRow partition0 = BinaryRow.singleColumn(0);
BinaryRow partition1 = BinaryRow.singleColumn(1);
@@ -869,7 +1368,7 @@ class ConflictDetectionTest {
@Test
void testCheckGlobalIndexRowIdExistenceSkipsDeleteIndexEntry() {
- ConflictDetection detection = createConflictDetection();
+ DataEvolutionConflictDetection detection = createConflictDetection();
Optional<RuntimeException> exception =
detection.checkConflicts(
@@ -884,20 +1383,60 @@ class ConflictDetectionTest {
assertThat(exception).isNotPresent();
}
- private ConflictDetection createConflictDetection() {
- return new ConflictDetection(
+ private DataEvolutionConflictDetection createConflictDetection() {
+ return (DataEvolutionConflictDetection) createConflictDetection(null,
true, false);
+ }
+
+ private ConflictDetection createConflictDetection(
+ @Nullable CommitScanner scanner,
+ boolean dataEvolutionEnabled,
+ boolean primaryKeyTable) {
+ return createConflictDetection(scanner, dataEvolutionEnabled,
primaryKeyTable, false, null);
+ }
+
+ private ConflictDetection createConflictDetection(
+ @Nullable CommitScanner scanner,
+ boolean dataEvolutionEnabled,
+ boolean primaryKeyTable,
+ boolean pkClusteringOverride,
+ @Nullable SnapshotManager snapshotManager) {
+ return ConflictDetection.create(
"test-table",
"test-user",
RowType.of(),
null,
- null,
+ primaryKeyTable ? (left, right) -> 0 : null,
BucketMode.HASH_FIXED,
false,
- true,
- false,
+ dataEvolutionEnabled,
+ pkClusteringOverride,
null,
- null,
- null);
+ snapshotManager,
+ scanner);
+ }
+
+ private CommitFailRetryResult commitFailRetryResult(
+ @Nullable Snapshot latestSnapshot, @Nullable List<SimpleFileEntry>
baseDataFiles) {
+ return (CommitFailRetryResult)
+ RetryCommitResult.forCommitFail(
+ latestSnapshot,
+ baseDataFiles,
+ new RuntimeException("expected test retry"),
+ null);
+ }
+
+ private ManifestEntry manifestEntry(
+ FileKind kind, String fileName, @Nullable Long firstRowId,
@Nullable Range rowIdRange) {
+ ManifestEntry entry = mock(ManifestEntry.class);
+ DataFileMeta file = mock(DataFileMeta.class);
+ when(entry.kind()).thenReturn(kind);
+ when(entry.file()).thenReturn(file);
+ when(file.fileName()).thenReturn(fileName);
+ when(file.firstRowId()).thenReturn(firstRowId);
+ if (rowIdRange != null) {
+ when(file.nonNullRowIdRange()).thenReturn(rowIdRange);
+ }
+ return entry;
}
private Snapshot snapshot(long id) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
index 9e52c920b5..384f3fe2a3 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -81,6 +81,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static
org.apache.paimon.errors.ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
@@ -720,6 +721,93 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
.doesNotContainAnyElementsOf(oldAnchorFiles);
}
+ @Test
+ public void testStaleMaterializeRejectedAfterConcurrentUpdate() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ Snapshot materializeSnapshot = table.latestSnapshot().get();
+ List<CommitMessage> staleMaterializeMessages =
+ prepareMaterializeDeletionVectors(table, materializeSnapshot,
null);
+
+ RowType writeType =
table.rowType().project(Collections.singletonList("f2"));
+ List<CommitMessage> concurrentUpdateMessages = new ArrayList<>();
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType)) {
+ for (int rowId = batch * 5; rowId < batch * 5 + 5; rowId++) {
+
write.write(GenericRow.of(BinaryString.fromString("concurrent-" + rowId)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ setFirstRowId(messages, batch * 5L);
+ concurrentUpdateMessages.addAll(messages);
+ }
+ }
+ commit(table, concurrentUpdateMessages);
+
+ List<String> concurrentValues = expectedProjectedStrings("concurrent",
FULL_RANGE);
+
assertThat(readProjectedStrings(table.newReadBuilder().withProjection(new int[]
{2})))
+ .containsExactlyElementsOf(concurrentValues);
+ long updateSnapshotId = table.latestSnapshot().get().id();
+
+ assertThatThrownBy(
+ () ->
+ commitMaterializeDeletionVectors(
+ table,
+ materializeSnapshot,
+ staleMaterializeMessages,
+ "test-stale-materialize"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE);
+
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(updateSnapshotId);
+
assertThat(readProjectedStrings(table.newReadBuilder().withProjection(new int[]
{2})))
+ .containsExactlyElementsOf(concurrentValues);
+ }
+
+ @Test
+ public void testStaleMaterializeAllowsNonOverlappingConcurrentUpdate()
throws Exception {
+ FileStoreTable table =
+
createPartitionedReassignTable("non_overlapping_materialize_update_table",
false);
+ writePartitionRows(table, "a", 0, 1, 2, 3, 4);
+ writePartitionRows(table, "b", 5, 6, 7, 8, 9);
+
+ BinaryRow partitionA = partition(table, "a");
+ BinaryRow partitionB = partition(table, "b");
+ commitDeletionVectors(
+ table, partitionA, Collections.singletonList(new DvSpec(new
Range(0, 4), 1)));
+
+ Snapshot materializeSnapshot = table.latestSnapshot().get();
+ List<CommitMessage> staleMaterializeMessages =
+ prepareMaterializeDeletionVectors(table, materializeSnapshot,
null);
+
+ writePartialStrings(table, "b", 5L, 5, 6, 7, 8, 9);
+ long updateSnapshotId = table.latestSnapshot().get().id();
+
+ commitMaterializeDeletionVectors(
+ table, materializeSnapshot, staleMaterializeMessages,
"test-stale-materialize");
+
+
assertThat(table.latestSnapshot().get().id()).isGreaterThan(updateSnapshotId);
+ assertThat(table.latestSnapshot().get().commitKind())
+ .isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(anchorFilesByRange(table, partitionA).keySet())
+ .containsExactly(new Range(10, 13));
+ assertThat(anchorFilesByRange(table,
partitionB).keySet()).containsExactly(new Range(5, 9));
+ assertThat(readPartitionedRows(table))
+ .containsExactlyInAnyOrder(
+ "a|0|base-0",
+ "a|2|base-2",
+ "a|3|base-3",
+ "a|4|base-4",
+ "b|5|updated-5",
+ "b|6|updated-6",
+ "b|7|updated-7",
+ "b|8|updated-8",
+ "b|9|updated-9");
+ }
+
@Test
public void testMaterializeKeepsSiblingDeletionVectorsInTouchedIndexFile()
throws Exception {
createTableDefault();
@@ -1133,6 +1221,14 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
private void materializeDeletionVectors(FileStoreTable table, Integer
deletionFilesPerBatch)
throws Exception {
Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> commitMessages =
+ prepareMaterializeDeletionVectors(table, snapshot,
deletionFilesPerBatch);
+ commitMaterializeDeletionVectors(table, snapshot, commitMessages,
"test-materialize-dv");
+ }
+
+ private List<CommitMessage> prepareMaterializeDeletionVectors(
+ FileStoreTable table, Snapshot snapshot, Integer
deletionFilesPerBatch)
+ throws Exception {
DataEvolutionDeletionVectorMaterializeCoordinator coordinator =
deletionFilesPerBatch == null
? new
DataEvolutionDeletionVectorMaterializeCoordinator(
@@ -1154,7 +1250,16 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
commitMessages.addAll(
new DataEvolutionCompactionCommitPreparation(table, snapshot)
.prepare(commitMessages));
- try (TableCommitImpl commit = table.newCommit("test-materialize-dv")) {
+ return commitMessages;
+ }
+
+ private void commitMaterializeDeletionVectors(
+ FileStoreTable table,
+ Snapshot snapshot,
+ List<CommitMessage> commitMessages,
+ String commitUser)
+ throws Exception {
+ try (TableCommitImpl commit = table.newCommit(commitUser)) {
commit.rowIdCheckConflictForMaterializeDvCompaction(snapshot.id())
.commit(commitMessages);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index beec22b2b0..c90c69c8cb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -44,6 +44,7 @@ import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.BatchWriteBuilderImpl;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
@@ -71,6 +72,7 @@ import java.util.OptionalLong;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import static
org.apache.paimon.errors.ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE;
import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
@@ -1175,6 +1177,39 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
Arrays.asList("0|base-0", "10|updated-10",
"11|updated-11", "20|base-20"));
}
+ @Test
+ public void testConcurrentSameColumnPartialUpdateConflict() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ long firstRowId = writeFullRows(table, 10, 11);
+ long readSnapshotId = table.latestSnapshot().get().id();
+
+ RowType writeType =
table.rowType().project(Collections.singletonList("f2"));
+ BatchWriteBuilderImpl staleBuilder = (BatchWriteBuilderImpl)
table.newBatchWriteBuilder();
+ List<CommitMessage> staleMessages;
+ try (BatchTableWrite write =
staleBuilder.newWrite().withWriteType(writeType)) {
+ write.write(GenericRow.of(BinaryString.fromString("stale-10")));
+ write.write(GenericRow.of(BinaryString.fromString("stale-11")));
+ staleMessages = write.prepareCommit();
+ setFirstRowId(staleMessages, firstRowId);
+ }
+
+ updateF2(table, firstRowId, 100, 101);
+ long concurrentSnapshotId = table.latestSnapshot().get().id();
+ staleBuilder.rowIdCheckConflict(readSnapshotId);
+
+ assertThatThrownBy(
+ () -> {
+ try (BatchTableCommit commit =
staleBuilder.newCommit()) {
+ commit.commit(staleMessages);
+ }
+ })
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(concurrentSnapshotId);
+
assertThat(readF0AndF2(table)).isEqualTo(Arrays.asList("10|updated-100",
"11|updated-101"));
+ }
+
@Test
public void
testCompactPreservesConcurrentPartialUpdateWithinCandidateRange() throws
Exception {
createTableDefault();
