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 a777c1b2b4 [core] Add conflict detection for concurrent merge-into &
compaction (#8964)
a777c1b2b4 is described below
commit a777c1b2b49ca89e60b70e16715bf3ac483120ae
Author: Faiz <[email protected]>
AuthorDate: Mon Aug 3 19:05:46 2026 +0800
[core] Add conflict detection for concurrent merge-into & compaction (#8964)
---
.../apache/paimon/operation/FileStoreCommit.java | 3 +
.../paimon/operation/FileStoreCommitImpl.java | 54 +++++--
.../paimon/operation/commit/ConflictDetection.java | 46 +++++-
.../commit/RowIdColumnConflictChecker.java | 8 +-
.../operation/commit/RowIdConflictChecker.java | 35 +++++
.../commit/RowIdRangeConflictChecker.java | 56 ++++++++
.../apache/paimon/table/sink/InnerTableCommit.java | 3 +
.../apache/paimon/table/sink/TableCommitImpl.java | 7 +
.../operation/commit/ConflictDetectionTest.java | 22 +++
.../commit/RowIdRangeConflictCheckerTest.java | 83 +++++++++++
.../table/DataEvolutionDeletionVectorTest.java | 156 +++++++++++++++++++++
.../flink/sink/DataEvolutionTableCompactSink.java | 14 +-
.../paimon/spark/procedure/CompactProcedure.java | 12 ++
13 files changed, 478 insertions(+), 21 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index a9936833a9..b039ffb9e9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -46,6 +46,9 @@ public interface FileStoreCommit extends AutoCloseable {
FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
+ FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(
+ @Nullable Long rowIdCheckFromSnapshot);
+
FileStoreCommit withOperation(Snapshot.Operation operation);
/** Find out which committables need to be retried when recovering from
the failure. */
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 e07fcb443f..80f4ffba46 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
@@ -54,6 +54,8 @@ 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;
@@ -106,13 +108,16 @@ import java.util.stream.Collectors;
import static java.util.Collections.emptyList;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
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;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -264,6 +269,14 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return this;
}
+ @Override
+ public FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(
+ @Nullable Long rowIdCheckFromSnapshot) {
+
this.conflictDetection.setRowIdCheckFromSnapshotForMaterializeDvCompaction(
+ rowIdCheckFromSnapshot);
+ return this;
+ }
+
@Override
public FileStoreCommit withOperation(Snapshot.Operation operation) {
this.operation = operation;
@@ -346,7 +359,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
checkAppendFiles = true;
allowRollback = true;
}
- if (conflictDetection.hasRowIdCheckFromSnapshot()) {
+ if
(conflictDetection.shouldCheckRowIdFromSnapshot(commitKind)) {
checkAppendFiles = true;
allowRollback = true;
}
@@ -1010,14 +1023,35 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
.filter(entry ->
!baseIdentifiers.contains(entry.identifier()))
.collect(Collectors.toList());
}
- RowIdColumnConflictChecker rowIdColumnConflictChecker = null;
- if (conflictDetection.hasRowIdCheckFromSnapshot()) {
- rowIdColumnConflictChecker =
- RowIdColumnConflictChecker.fromDataFiles(
- schemaManager,
- deltaFiles.stream()
- .map(ManifestEntry::file)
- .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);
+ }
}
Optional<RuntimeException> exception =
conflictDetection.checkConflicts(
@@ -1025,7 +1059,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
baseDataFiles,
SimpleFileEntry.from(deltaFiles),
indexFiles,
- rowIdColumnConflictChecker,
+ rowIdConflictChecker,
commitKind);
if (exception.isPresent()) {
if (allowRollback && rollback != null) {
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 4cdc5f997b..99bae20226 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
@@ -67,6 +67,7 @@ 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;
@@ -101,6 +102,7 @@ public class ConflictDetection {
private @Nullable PartitionExpire partitionExpire;
private @Nullable Long rowIdCheckFromSnapshot = null;
+ private @Nullable RowIdConflictChecker.TriggerSource
rowIdConflictCheckTriggerSource = null;
public ConflictDetection(
String tableName,
@@ -130,11 +132,34 @@ public class ConflictDetection {
}
public void setRowIdCheckFromSnapshot(@Nullable Long
rowIdCheckFromSnapshot) {
+ setRowIdCheckFromSnapshot(
+ rowIdCheckFromSnapshot,
RowIdConflictChecker.TriggerSource.DATA_EVOLUTION_DML);
+ }
+
+ 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;
+ }
+
+ public boolean shouldCheckRowIdFromSnapshot(CommitKind commitKind) {
+ return rowIdCheckFromSnapshot != null
+ && (rowIdConflictCheckTriggerSource !=
MATERIALIZE_DV_COMPACTION
+ || commitKind == CommitKind.COMPACT);
}
- public boolean hasRowIdCheckFromSnapshot() {
- return rowIdCheckFromSnapshot != null;
+ public RowIdConflictChecker.TriggerSource
rowIdConflictCheckTriggerSource() {
+ checkState(
+ rowIdConflictCheckTriggerSource != null,
+ "Row ID conflict check trigger source is not set.");
+ return rowIdConflictCheckTriggerSource;
}
@Nullable
@@ -166,7 +191,7 @@ public class ConflictDetection {
List<SimpleFileEntry> baseEntries,
List<SimpleFileEntry> deltaEntries,
List<IndexManifestEntry> deltaIndexEntries,
- @Nullable RowIdColumnConflictChecker rowIdColumnConflictChecker,
+ @Nullable RowIdConflictChecker rowIdConflictChecker,
CommitKind commitKind) {
String baseCommitUser = latestSnapshot.commitUser();
if (deletionVectorsEnabled &&
bucketMode.equals(BucketMode.BUCKET_UNAWARE)) {
@@ -247,7 +272,7 @@ public class ConflictDetection {
}
return checkForRowIdFromSnapshot(
- latestSnapshot, deltaEntries, deltaIndexEntries,
rowIdColumnConflictChecker);
+ latestSnapshot, deltaEntries, deltaIndexEntries,
rowIdConflictChecker);
}
public <T extends FileEntry> Map<BinaryRow, Integer>
collectUncheckedFixedBucketPartitions(
@@ -580,14 +605,14 @@ public class ConflictDetection {
Snapshot latestSnapshot,
List<SimpleFileEntry> deltaEntries,
List<IndexManifestEntry> deltaIndexEntries,
- @Nullable RowIdColumnConflictChecker columnChecker) {
+ @Nullable RowIdConflictChecker conflictChecker) {
if (!dataEvolutionEnabled) {
return Optional.empty();
}
if (rowIdCheckFromSnapshot == null) {
return Optional.empty();
}
- if (columnChecker == null || columnChecker.isEmpty()) {
+ if (conflictChecker == null || conflictChecker.isEmpty()) {
return Optional.empty();
}
@@ -607,10 +632,13 @@ public class ConflictDetection {
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
- && columnChecker.conflictsWith(file)) {
+ && conflictChecker.conflictsWith(file)) {
LOG.debug(
"Data evolution row id conflict detected for table
{}, commit user {}, "
+ "snapshot {}, file {}.",
@@ -628,6 +656,10 @@ public class ConflictDetection {
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) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
index b2f8740f52..f5d00a1e89 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdColumnConflictChecker.java
@@ -46,7 +46,7 @@ import java.util.stream.Collectors;
* columns also overlap, return conflicting result.
* </ol>
*/
-public class RowIdColumnConflictChecker {
+public class RowIdColumnConflictChecker implements RowIdConflictChecker {
private final SchemaManager schemaManager;
private final List<WriteRange> writeRanges;
@@ -122,7 +122,8 @@ public class RowIdColumnConflictChecker {
return new Range(from, to);
}
- boolean isEmpty() {
+ @Override
+ public boolean isEmpty() {
return writeRanges.isEmpty();
}
@@ -134,7 +135,8 @@ public class RowIdColumnConflictChecker {
* @param file committed incremental data file
* @return true if conflict
*/
- boolean conflictsWith(DataFileMeta file) {
+ @Override
+ public boolean conflictsWith(DataFileMeta file) {
Long firstRowId = file.firstRowId();
if (firstRowId == null) {
return false;
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
new file mode 100644
index 0000000000..e5e9808946
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdConflictChecker.java
@@ -0,0 +1,35 @@
+/*
+ * 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.io.DataFileMeta;
+
+/** Detects row ID conflicts between committing files and a historical data
file. */
+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/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
new file mode 100644
index 0000000000..cf6bea0a51
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdRangeConflictChecker.java
@@ -0,0 +1,56 @@
+/*
+ * 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.io.DataFileMeta;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Detects row ID conflicts solely by row range overlap. */
+public class RowIdRangeConflictChecker implements RowIdConflictChecker {
+
+ private final RowRangeIndex rowRangeIndex;
+
+ private RowIdRangeConflictChecker(List<DataFileMeta> deltaFiles) {
+ this.rowRangeIndex =
+ RowRangeIndex.create(
+ deltaFiles.stream()
+ .filter(file -> file.firstRowId() != null)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList()));
+ }
+
+ public static RowIdRangeConflictChecker fromDataFiles(List<DataFileMeta>
deltaFiles) {
+ return new RowIdRangeConflictChecker(deltaFiles);
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return rowRangeIndex.ranges().isEmpty();
+ }
+
+ @Override
+ public boolean conflictsWith(DataFileMeta file) {
+ return file.firstRowId() != null
+ && rowRangeIndex.intersects(
+ file.nonNullRowIdRange().from,
file.nonNullRowIdRange().to);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
index f5bb81d667..43f98d0e79 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
@@ -58,6 +58,9 @@ public interface InnerTableCommit extends StreamTableCommit,
BatchTableCommit {
InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
+ InnerTableCommit rowIdCheckConflictForMaterializeDvCompaction(
+ @Nullable Long rowIdCheckFromSnapshot);
+
@Override
InnerTableCommit withMetricRegistry(MetricRegistry registry);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index b66cca67e3..7e47041db6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -182,6 +182,13 @@ public class TableCommitImpl implements InnerTableCommit {
return this;
}
+ @Override
+ public TableCommitImpl rowIdCheckConflictForMaterializeDvCompaction(
+ @Nullable Long rowIdCheckFromSnapshot) {
+
commit.rowIdCheckConflictForMaterializeDvCompaction(rowIdCheckFromSnapshot);
+ return this;
+ }
+
@Override
public TableCommitImpl withOperation(Snapshot.Operation operation) {
commit.withOperation(operation);
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 cfc2963d08..d73cfac0e0 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
@@ -398,6 +398,28 @@ class ConflictDetectionTest {
.isFalse();
}
+ @Test
+ void testMaterializeDvRowIdCheckOnlyAppliesToCompactCommit() {
+ ConflictDetection detection = createConflictDetection();
+
+ detection.setRowIdCheckFromSnapshotForMaterializeDvCompaction(1L);
+
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.APPEND)).isFalse();
+
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.OVERWRITE)).isFalse();
+
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.COMPACT)).isTrue();
+
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.ADD)).isTrue();
+
assertThat(detection.shouldCheckHistoricalRowIdEntry(FileKind.DELETE)).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();
+
+ detection.setRowIdCheckFromSnapshot(null);
+
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.APPEND)).isFalse();
+
assertThat(detection.shouldCheckRowIdFromSnapshot(Snapshot.CommitKind.COMPACT)).isFalse();
+ }
+
@Test
void testChangedPartitionsIncludesGlobalIndexFiles() {
BinaryRow partition = BinaryRow.singleColumn(1);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdRangeConflictCheckerTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdRangeConflictCheckerTest.java
new file mode 100644
index 0000000000..419410a265
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/RowIdRangeConflictCheckerTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.io.DataFileMeta;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RowIdRangeConflictCheckerTest {
+
+ @Test
+ void testDetectsOverlappingRowRanges() {
+ RowIdRangeConflictChecker checker =
+ RowIdRangeConflictChecker.fromDataFiles(
+ Arrays.asList(file("first", 0L, 5L), file("second",
10L, 5L)));
+
+ assertThat(checker.conflictsWith(file("same", 0L, 5L))).isTrue();
+ assertThat(checker.conflictsWith(file("overlap", 4L, 7L))).isTrue();
+ assertThat(checker.conflictsWith(file("contained", 11L, 2L))).isTrue();
+ }
+
+ @Test
+ void testAllowsDisjointAndAdjacentRowRanges() {
+ RowIdRangeConflictChecker checker =
+ RowIdRangeConflictChecker.fromDataFiles(
+ Collections.singletonList(file("current", 5L, 5L)));
+
+ assertThat(checker.conflictsWith(file("before", 0L, 5L))).isFalse();
+ assertThat(checker.conflictsWith(file("after", 10L, 5L))).isFalse();
+ }
+
+ @Test
+ void testIgnoresFilesWithoutRowIds() {
+ RowIdRangeConflictChecker checker =
+ RowIdRangeConflictChecker.fromDataFiles(
+ Collections.singletonList(file("current", null, 5L)));
+
+ assertThat(checker.isEmpty()).isTrue();
+ assertThat(checker.conflictsWith(file("historical", 0L,
5L))).isFalse();
+ }
+
+ private DataFileMeta file(String fileName, @Nullable Long firstRowId, long
rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 0L,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0L,
+ 0L,
+ 0L,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ firstRowId,
+ null);
+ }
+}
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 1de5d51a4b..df0b646cc8 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
@@ -53,6 +53,7 @@ import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.EndOfScanException;
@@ -79,6 +80,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;
@@ -598,6 +600,143 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
.doesNotContainAnyElementsOf(oldAnchorFiles);
}
+ @Test
+ public void testStaleMaterializeCompactionRejectedAfterConcurrentUpdate()
throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+
dynamicOptions.put(CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key(),
"true");
+ FileStoreTable compactTable = table.copy(dynamicOptions);
+ Snapshot compactSnapshot = compactTable.latestSnapshot().get();
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(compactTable, false,
false, compactSnapshot);
+ List<CommitMessage> staleMaterializeMessages = new ArrayList<>();
+ try {
+ while (true) {
+ for (DataEvolutionCompactTask task : coordinator.plan()) {
+ staleMaterializeMessages.add(
+ task.doCompact(compactTable,
"test-stale-materialize"));
+ }
+ }
+ } catch (EndOfScanException ignored) {
+ }
+ assertThat(staleMaterializeMessages).isNotEmpty();
+ staleMaterializeMessages.addAll(
+ new DataEvolutionCompactionCommitPreparation(compactTable,
compactSnapshot)
+ .prepare(staleMaterializeMessages));
+
+ 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);
+
+
assertThat(readProjectedStrings(table.newReadBuilder().withProjection(new int[]
{2})))
+
.containsExactlyElementsOf(expectedProjectedStrings("concurrent", FULL_RANGE));
+ long updateSnapshotId = table.latestSnapshot().get().id();
+
+ assertThatThrownBy(
+ () -> {
+ try (TableCommitImpl commit =
+
compactTable.newCommit("test-stale-materialize")) {
+
commit.rowIdCheckConflictForMaterializeDvCompaction(
+ compactSnapshot.id())
+ .commit(staleMaterializeMessages);
+ }
+ })
+ .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(expectedProjectedStrings("concurrent", FULL_RANGE));
+ }
+
+ @Test
+ public void
testStaleMaterializeCompactionAllowsNonOverlappingConcurrentUpdate()
+ 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");
+ assertThat(anchorFilesByRange(table,
partitionA).keySet()).containsExactly(new Range(0, 4));
+ assertThat(anchorFilesByRange(table,
partitionB).keySet()).containsExactly(new Range(5, 9));
+ commitDeletionVectors(
+ table, partitionA, Collections.singletonList(new DvSpec(new
Range(0, 4), 1)));
+
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+
dynamicOptions.put(CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key(),
"true");
+ FileStoreTable compactTable = table.copy(dynamicOptions);
+ Snapshot compactSnapshot = compactTable.latestSnapshot().get();
+ PartitionPredicate partitionPredicate =
+ PartitionPredicate.fromMaps(
+ table.schema().logicalPartitionType(),
+
Collections.singletonList(Collections.singletonMap("pt", "a")),
+ table.coreOptions().partitionDefaultName());
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(
+ compactTable, partitionPredicate, false, false,
compactSnapshot);
+ List<CommitMessage> staleMaterializeMessages = new ArrayList<>();
+ try {
+ while (true) {
+ for (DataEvolutionCompactTask task : coordinator.plan()) {
+ assertThat(task.type())
+
.isEqualTo(DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION);
+ staleMaterializeMessages.add(
+ task.doCompact(compactTable,
"test-stale-materialize"));
+ }
+ }
+ } catch (EndOfScanException ignored) {
+ }
+ assertThat(staleMaterializeMessages).isNotEmpty();
+ staleMaterializeMessages.addAll(
+ new DataEvolutionCompactionCommitPreparation(compactTable,
compactSnapshot)
+ .prepare(staleMaterializeMessages));
+
+ writePartialStrings(table, "b", 5L, 5, 6, 7, 8, 9);
+ long updateSnapshotId = table.latestSnapshot().get().id();
+
+ try (TableCommitImpl commit =
compactTable.newCommit("test-stale-materialize")) {
+
commit.rowIdCheckConflictForMaterializeDvCompaction(compactSnapshot.id())
+ .commit(staleMaterializeMessages);
+ }
+
+
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
testMaterializeCompactionUsesRemainingSizeForLargeDeletedRange() throws
Exception {
createTableDefault();
@@ -867,6 +1006,23 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
return rows;
}
+ private static List<String> readPartitionedRows(FileStoreTable table)
throws IOException {
+ ReadBuilder readBuilder = table.newReadBuilder();
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ rows.add(
+ row.getString(0)
+ + "|"
+ + row.getInt(1)
+ + "|"
+ + row.getString(2)));
+ }
+ return rows;
+ }
+
private static Map<String, Range> relativeFileRanges(
FileStoreTable table, BinaryRow partition) {
List<DataFileMeta> dataFiles = currentDataFiles(table, partition);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionTableCompactSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionTableCompactSink.java
index 5fc4320e4e..d57ecbe11b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionTableCompactSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionTableCompactSink.java
@@ -18,10 +18,12 @@
package org.apache.paimon.flink.sink;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
@@ -71,7 +73,17 @@ public class DataEvolutionTableCompactSink extends
FlinkSink<DataEvolutionCompac
@Override
protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
- return context -> new StoreCommitter(table,
table.newCommit(context.commitUser()), context);
+ return context -> {
+ TableCommitImpl commit = table.newCommit(context.commitUser());
+ if (shouldCheckMaterializeDvConflict(table.coreOptions())) {
+
commit.rowIdCheckConflictForMaterializeDvCompaction(snapshot.id());
+ }
+ return new StoreCommitter(table, commit, context);
+ };
+ }
+
+ static boolean shouldCheckMaterializeDvConflict(CoreOptions options) {
+ return options.deletionVectorsEnabled() &&
options.dataEvolutionCompactionRewriteRowIds();
}
@Override
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 13a71d773c..59a92cd269 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -536,6 +536,7 @@ public class CompactProcedure extends BaseProcedure {
LOG.info("Task plan is empty, no compact job to execute.");
continue;
}
+ boolean containsMaterializeDeletion =
containsMaterializeDeletion(compactionTasks);
DataEvolutionCompactTaskSerializer serializer =
new DataEvolutionCompactTaskSerializer();
@@ -576,6 +577,9 @@ public class CompactProcedure extends BaseProcedure {
List<byte[]> serializedMessages = new
ArrayList<>(commitMessageJavaRDD.collect());
try (TableCommitImpl commit = table.newCommit(commitUser)) {
+ if (containsMaterializeDeletion) {
+
commit.rowIdCheckConflictForMaterializeDvCompaction(snapshot.id());
+ }
List<CommitMessage> messages =
deserializeCommitMessagesAndReleaseSerializedBytes(
messageSerializerser, serializedMessages);
@@ -592,6 +596,14 @@ public class CompactProcedure extends BaseProcedure {
}
}
+ static boolean containsMaterializeDeletion(List<DataEvolutionCompactTask>
compactionTasks) {
+ return compactionTasks.stream()
+ .anyMatch(
+ task ->
+ task.type()
+ ==
DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION);
+ }
+
private static List<CommitMessage>
deserializeCommitMessagesAndReleaseSerializedBytes(
CommitMessageSerializer serializer, List<byte[]>
serializedMessages)
throws IOException {
