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 8d26890f1a [core] detect concurrent row-id reassignment and
data-evolution compaction (#8773)
8d26890f1a is described below
commit 8d26890f1abbd75fe73105d380c1b95a1d768fb2
Author: Faiz <[email protected]>
AuthorDate: Tue Jul 21 15:44:18 2026 +0800
[core] detect concurrent row-id reassignment and data-evolution compaction
(#8773)
---
.../paimon/operation/FileStoreCommitImpl.java | 7 +-
.../paimon/operation/commit/ConflictDetection.java | 98 ++++--
.../operation/commit/ConflictDetectionTest.java | 121 ++++++-
.../table/DataEvolutionDeletionVectorTest.java | 365 ++++++++++++++++++++-
4 files changed, 545 insertions(+), 46 deletions(-)
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 07505fa3e4..b440320212 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
@@ -891,6 +891,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
// Check if the commit has been completed. At this point, there will
be no more repeated
// commits and just return success
+ boolean hasOverwriteSinceLastAttempt = false;
if (retryResult instanceof CommitFailRetryResult && latestSnapshot !=
null) {
CommitFailRetryResult commitFailRetry = (CommitFailRetryResult)
retryResult;
Map<Long, Snapshot> snapshotCache = new HashMap<>();
@@ -903,6 +904,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
}
for (long i = startCheckSnapshot; i <= latestSnapshot.id(); i++) {
Snapshot snapshot = snapshotCache.computeIfAbsent(i,
snapshotManager::snapshot);
+ hasOverwriteSinceLastAttempt |= snapshot.commitKind() ==
CommitKind.OVERWRITE;
if (snapshot.commitUser().equals(commitUser)
&& snapshot.commitIdentifier() == identifier
&& snapshot.commitKind() == commitKind) {
@@ -959,9 +961,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
retryResult instanceof CommitFailRetryResult
? (CommitFailRetryResult) retryResult
: null;
+ // 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 (commitFailRetry != null
&& commitFailRetry.latestSnapshot != null
- && commitFailRetry.baseDataFiles != null) {
+ && commitFailRetry.baseDataFiles != null
+ && !hasOverwriteSinceLastAttempt) {
baseDataFiles = new ArrayList<>(commitFailRetry.baseDataFiles);
List<SimpleFileEntry> incremental =
scanner.readIncrementalChanges(
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 a679b9d091..34c5a29a46 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
@@ -228,12 +228,10 @@ public class ConflictDetection {
return exception;
}
- if (commitKind != CommitKind.COMPACT) {
- Long nextRowId = latestSnapshot.nextRowId();
- exception = checkRowIdExistence(baseEntries, deltaEntries,
nextRowId);
- if (exception.isPresent()) {
- return exception;
- }
+ Long nextRowId = latestSnapshot.nextRowId();
+ exception = checkRowIdExistence(baseEntries, deltaEntries, nextRowId,
commitKind);
+ if (exception.isPresent()) {
+ return exception;
}
exception = checkRowIdRangeConflicts(commitKind, mergedEntries);
@@ -690,11 +688,66 @@ public class ConflictDetection {
Optional<RuntimeException> checkRowIdExistence(
List<SimpleFileEntry> baseEntries,
List<SimpleFileEntry> deltaEntries,
- @Nullable Long nextRowId) {
+ @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(
@@ -709,13 +762,6 @@ public class ConflictDetection {
return Optional.empty();
}
- List<SimpleFileEntry> existingDataFiles =
- baseEntries.stream()
- .filter(
- base ->
- base.firstRowId() != null
- &&
!dedicatedStorageFile(base.fileName()))
- .collect(Collectors.toList());
RowRangeIndex existingIndex = rowRangeIndex(existingDataFiles, false);
for (SimpleFileEntry entry : filesToCheck) {
@@ -725,23 +771,23 @@ public class ConflictDetection {
? existingIndex.contains(rowRange)
: existingIndex.containsExactly(rowRange);
if (!exists) {
- return Optional.of(
- new RuntimeException(
- String.format(
- "Row ID existence conflict: file '%s'
references "
- + "firstRowId=%d, rowCount=%d
in bucket %d, "
- + "but no matching file exists
in the current snapshot. "
- + "The referenced file may
have been rewritten by a "
- + "concurrent compaction or
removed by an overwrite.",
- entry.fileName(),
- entry.firstRowId(),
- entry.rowCount(),
- entry.bucket())));
+ return Optional.of(rowIdExistenceConflict(entry));
}
}
return Optional.empty();
}
+ private RuntimeException rowIdExistenceConflict(SimpleFileEntry entry) {
+ return new RuntimeException(
+ String.format(
+ "Row ID existence conflict: file '%s' references "
+ + "firstRowId=%d, rowCount=%d in bucket %d, "
+ + "but no matching file exists in the current
snapshot. "
+ + "The referenced file may have been rewritten
by a "
+ + "concurrent compaction or removed by an
overwrite.",
+ entry.fileName(), entry.firstRowId(),
entry.rowCount(), entry.bucket()));
+ }
+
private static boolean dedicatedStorageFile(String fileName) {
return isBlobFile(fileName) || isVectorStoreFile(fileName);
}
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 5be15f0dcc..cfc2963d08 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
@@ -420,7 +420,10 @@ class ConflictDetectionTest {
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 100L));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
100L)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
}
@Test
@@ -433,7 +436,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 100L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 100L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -449,7 +453,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 100L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 200L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 200L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -466,7 +471,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 4L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -481,7 +487,10 @@ class ConflictDetectionTest {
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
4L)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
}
@Test
@@ -496,7 +505,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 4L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -512,7 +522,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 3L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 3L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 3L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -528,7 +539,8 @@ class ConflictDetectionTest {
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
Optional<RuntimeException> result =
- detection.checkRowIdExistence(baseEntries, deltaEntries, 2L);
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 2L,
Snapshot.CommitKind.APPEND);
assertThat(result).isPresent();
assertThat(result.get().getMessage()).contains("Row ID existence
conflict");
}
@@ -547,7 +559,10 @@ class ConflictDetectionTest {
// newly appended file (firstRowId=100 >= nextRowId=100), should be
skipped
deltaEntries.add(createFileEntryWithRowId("new1", ADD, 100L, 50L));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
100L)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
}
@Test
@@ -559,7 +574,10 @@ class ConflictDetectionTest {
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
deltaEntries.add(createFileEntry("f1", ADD));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
100L)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
}
@Test
@@ -571,7 +589,10 @@ class ConflictDetectionTest {
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
deltaEntries.add(createFileEntryWithRowId("f1", DELETE, 0L, 100L));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
100L)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 100L,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
}
@Test
@@ -582,7 +603,83 @@ class ConflictDetectionTest {
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 100L));
- assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries,
null)).isEmpty();
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, null,
Snapshot.CommitKind.APPEND))
+ .isEmpty();
+ }
+
+ @Test
+ void testCheckRowIdExistenceCompactAllowsAdjacentNormalRanges() {
+ ConflictDetection detection = createConflictDetection();
+
+ List<SimpleFileEntry> baseEntries =
+ Arrays.asList(
+ createFileEntryWithRowId("f1", ADD, 0L, 2L),
+ createFileEntryWithRowId("f2", ADD, 2L, 2L));
+ List<SimpleFileEntry> deltaEntries =
+
Collections.singletonList(createFileEntryWithRowId("compacted", ADD, 0L, 4L));
+
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.COMPACT))
+ .isEmpty();
+ }
+
+ @Test
+ void testCheckRowIdExistenceCompactAllowsBlobAcrossAdjacentNormalRanges() {
+ ConflictDetection detection = createConflictDetection();
+
+ List<SimpleFileEntry> baseEntries =
+ Arrays.asList(
+ createFileEntryWithRowId("f1", ADD, 0L, 2L),
+ createFileEntryWithRowId("f2", ADD, 2L, 2L));
+ List<SimpleFileEntry> deltaEntries =
+
Collections.singletonList(createFileEntryWithRowId("compacted.blob", ADD, 0L,
4L));
+
+ assertThat(
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.COMPACT))
+ .isEmpty();
+ }
+
+ @Test
+ void testCheckRowIdExistenceCompactRejectsStaleRangeAfterReassign() {
+ ConflictDetection detection = createConflictDetection();
+
+ List<SimpleFileEntry> baseEntries =
+ Arrays.asList(
+ createFileEntryWithRowId("f1", ADD, 10L, 2L),
+ createFileEntryWithRowId("f2", ADD, 12L, 2L));
+ List<SimpleFileEntry> deltaEntries =
+
Collections.singletonList(createFileEntryWithRowId("compacted", ADD, 0L, 4L));
+
+ Optional<RuntimeException> result =
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 14L,
Snapshot.CommitKind.COMPACT);
+ assertThat(result).isPresent();
+ assertThat(result.get()).hasMessageContaining("Row ID existence
conflict");
+ }
+
+ @Test
+ void testCheckRowIdExistenceCompactDoesNotMergeAcrossPartitions() {
+ ConflictDetection detection = createConflictDetection();
+ BinaryRow partition0 = BinaryRow.singleColumn(0);
+ BinaryRow partition1 = BinaryRow.singleColumn(1);
+
+ List<SimpleFileEntry> baseEntries =
+ Arrays.asList(
+ createFileEntryWithRowId("f1", ADD, partition0, 0, 0L,
2L),
+ createFileEntryWithRowId("f2", ADD, partition1, 0, 2L,
2L));
+ List<SimpleFileEntry> deltaEntries =
+ Collections.singletonList(
+ createFileEntryWithRowId("compacted", ADD, partition0,
0, 0L, 4L));
+
+ Optional<RuntimeException> result =
+ detection.checkRowIdExistence(
+ baseEntries, deltaEntries, 4L,
Snapshot.CommitKind.COMPACT);
+ assertThat(result).isPresent();
+ assertThat(result.get()).hasMessageContaining("Row ID existence
conflict");
}
@Test
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 325079dbe7..1de5d51a4b 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
@@ -23,12 +23,14 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
+import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassigner;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BinaryVector;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
@@ -42,6 +44,7 @@ import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -64,6 +67,7 @@ import org.apache.paimon.utils.RangeHelper;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -71,6 +75,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
@@ -79,6 +84,7 @@ import static
org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
/** Tests filename-anchored deletion vectors for data evolution tables. */
public class DataEvolutionDeletionVectorTest extends DataEvolutionTestBase {
@@ -120,6 +126,194 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
assertReadMatrix(getTableDefault(), "base");
}
+ @Test
+ public void testRowIdReassignKeepsAndMergesDeletionVectors() throws
Exception {
+ FileStoreTable table =
createPartitionedReassignTable("reassign_dv_table", false);
+ writePartitionRows(table, "a", 0, 1, 2);
+ writePartitionRows(table, "b", 3, 4);
+ writePartitionRows(table, "a", 5, 6, 7);
+
+ BinaryRow partitionA = partition(table, "a");
+ BinaryRow partitionB = partition(table, "b");
+ commitDeletionVectors(
+ table,
+ partitionA,
+ Arrays.asList(new DvSpec(new Range(0, 2), 1), new DvSpec(new
Range(5, 7), 6)));
+ commitDeletionVectors(
+ table, partitionB, Collections.singletonList(new DvSpec(new
Range(3, 4), 3)));
+
+ long historicalSnapshotId = table.latestSnapshot().get().id();
+ List<String> anchorFilesBefore =
liveDeletionVectorDataFileNames(table);
+ List<String> deletionVectorFilesBefore =
liveDeletionVectorIndexFileNames(table);
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly("a|0|0", "a|2|2", "b|4|4", "a|5|5", "a|7|7");
+
+ DataEvolutionRowIdReassigner.Result result =
+ new
DataEvolutionRowIdReassigner(table).reassign("test-reassign-dv");
+
+ assertThat(result.firstAssignedRowId).isEqualTo(8L);
+ assertThat(result.nextRowId).isEqualTo(14L);
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly("b|4|4", "a|0|8", "a|2|10", "a|5|11",
"a|7|13");
+
assertThat(liveDeletionVectorDataFileNames(table)).isEqualTo(anchorFilesBefore);
+
assertThat(liveDeletionVectorIndexFileNames(table)).isEqualTo(deletionVectorFilesBefore);
+
+ commitDeletionVectors(
+ table,
+ partitionA,
+ Arrays.asList(new DvSpec(new Range(8, 10), 8), new DvSpec(new
Range(11, 13), 13)));
+
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly("b|4|4", "a|2|10", "a|5|11");
+
assertThat(liveDeletionVectorDataFileNames(table)).isEqualTo(anchorFilesBefore);
+
+ FileStoreTable historicalTable =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(historicalSnapshotId)));
+ assertThat(readPartitionedRowsWithRowIds(historicalTable))
+ .containsExactly("a|0|0", "a|2|2", "b|4|4", "a|5|5", "a|7|7");
+ }
+
+ @Test
+ public void testRowIdReassignKeepsPartialWriteAndBlobFileOffsets() throws
Exception {
+ FileStoreTable table =
createPartitionedReassignTable("reassign_dedicated_dv_table", true);
+ writeDedicatedPartitionRows(table, "a", 0, 1, 2, 3, 4);
+ writeDedicatedPartitionRows(table, "b", 5, 6);
+ writeDedicatedPartitionRows(table, "a", 7, 8, 9, 10, 11);
+ writePartialStrings(table, "a", 0L, 0, 1, 2, 3, 4);
+
+ BinaryRow partitionA = partition(table, "a");
+ commitDeletionVectors(
+ table,
+ partitionA,
+ Arrays.asList(new DvSpec(new Range(0, 4), 1), new DvSpec(new
Range(7, 11), 8)));
+
+ Map<String, Range> relativeRangesBefore = relativeFileRanges(table,
partitionA);
+ assertThat(relativeRangesBefore.entrySet())
+ .anyMatch(
+ entry ->
+ BlobFileFormat.isBlobFile(entry.getKey())
+ && entry.getValue().count() < 5
+ && entry.getValue().from > 0);
+ assertThat(normalFilesByRange(table).get(new Range(0, 4)))
+ .hasSize(2)
+ .anyMatch(
+ file ->
+ file.writeCols().contains("f1")
+ && !file.writeCols().contains("f0"));
+
+ new
DataEvolutionRowIdReassigner(table).reassign("test-reassign-dedicated-dv");
+
+ assertThat(relativeFileRanges(table,
partitionA)).isEqualTo(relativeRangesBefore);
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly(
+ "b|5|base-5|5|5",
+ "b|6|base-6|6|6",
+ "a|0|updated-0|0|12",
+ "a|2|updated-2|2|14",
+ "a|3|updated-3|3|15",
+ "a|4|updated-4|4|16",
+ "a|7|base-7|7|17",
+ "a|9|base-9|9|19",
+ "a|10|base-10|10|20",
+ "a|11|base-11|11|21");
+ }
+
+ @Test
+ public void testRowIdReassignAbortsAfterConcurrentDeletionVectorCommit()
throws Exception {
+ FileStoreTable table =
+ createPartitionedReassignTable("reassign_concurrent_dv_table",
false);
+ writePartitionRows(table, "a", 0, 1, 2);
+ writePartitionRows(table, "b", 3, 4);
+ writePartitionRows(table, "a", 5, 6, 7);
+ Snapshot before = table.latestSnapshot().get();
+ BinaryRow partitionA = partition(table, "a");
+ AtomicBoolean committed = new AtomicBoolean();
+
+ DataEvolutionRowIdReassigner reassigner =
+ reassignerWithBeforeCommit(
+ table,
+ () -> {
+ if (committed.compareAndSet(false, true)) {
+ try {
+ commitDeletionVectors(
+ table,
+ partitionA,
+ Collections.singletonList(
+ new DvSpec(new Range(0,
2), 1)));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+
+ assertThatThrownBy(() ->
reassigner.reassign("test-reassign-concurrent-dv"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("OVERWRITE snapshot");
+
+ assertThat(committed).isTrue();
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(before.id() +
1);
+ assertThat(table.latestSnapshot().get().commitKind())
+ .isEqualTo(Snapshot.CommitKind.OVERWRITE);
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly("a|0|0", "a|2|2", "b|3|3", "b|4|4", "a|5|5",
"a|6|6", "a|7|7");
+ }
+
+ @Test
+ public void testStaleCompactionIsRejectedAfterRowIdReassign() throws
Exception {
+ FileStoreTable table =
createPartitionedReassignTable("stale_compaction_dv_table", false);
+ writePartitionRows(table, "a", 0, 1, 2);
+ writePartitionRows(table, "b", 3, 4);
+ writePartitionRows(table, "a", 5, 6, 7);
+ writePartialStrings(table, "a", 0L, 0, 1, 2);
+
+ BinaryRow partitionA = partition(table, "a");
+ commitDeletionVectors(
+ table, partitionA, Collections.singletonList(new DvSpec(new
Range(0, 2), 1)));
+
+ Snapshot compactSnapshot = table.latestSnapshot().get();
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ FileStoreTable compactTable = table.copy(dynamicOptions);
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(compactTable, false,
false, compactSnapshot);
+ List<CommitMessage> staleCompactMessages = new ArrayList<>();
+ try {
+ while (true) {
+ for (DataEvolutionCompactTask task : coordinator.plan()) {
+ staleCompactMessages.add(task.doCompact(compactTable,
"test-stale-compact"));
+ }
+ }
+ } catch (EndOfScanException ignored) {
+ }
+ assertThat(staleCompactMessages).isNotEmpty();
+ staleCompactMessages.addAll(
+ new DataEvolutionCompactionCommitPreparation(compactTable,
compactSnapshot)
+ .prepare(staleCompactMessages));
+
+ DataEvolutionRowIdReassigner.Result result =
+ new
DataEvolutionRowIdReassigner(table).reassign("test-before-stale-compact");
+ assertThat(result.firstAssignedRowId).isEqualTo(8L);
+ List<String> reassignedRows =
+ Arrays.asList("b|3|3", "b|4|4", "a|0|8", "a|2|10", "a|5|11",
"a|6|12", "a|7|13");
+
assertThat(readPartitionedRowsWithRowIds(table)).containsExactlyElementsOf(reassignedRows);
+ long reassignSnapshotId = table.latestSnapshot().get().id();
+
+ Throwable failure = catchThrowable(() -> commit(table,
staleCompactMessages));
+
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .as(
+ "stale compaction must not change reassigned row IDs;
commit failure: %s",
+ failure)
+ .containsExactlyElementsOf(reassignedRows);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(reassignSnapshotId);
+ assertThat(failure)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Row ID existence conflict");
+ }
+
@Test
public void testReadAfterAddingColumnAndDeletionVectors() throws Exception
{
// DVs with adding new columns.
@@ -568,6 +762,144 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
new Range(10, 14));
}
+ private FileStoreTable createPartitionedReassignTable(String tableName,
boolean dedicated)
+ throws Exception {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("pt", DataTypes.STRING());
+ schemaBuilder.column("f0", DataTypes.INT());
+ schemaBuilder.column("f1", DataTypes.STRING());
+ if (dedicated) {
+ schemaBuilder.column("f2", DataTypes.BLOB());
+ schemaBuilder.option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1
b");
+ schemaBuilder.option(CoreOptions.FILE_COMPRESSION.key(), "none");
+ }
+ schemaBuilder.partitionKeys("pt");
+ schemaBuilder.option(CoreOptions.TARGET_FILE_SIZE.key(), "128 MB");
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ catalog.createTable(identifier(tableName), schemaBuilder.build(),
false);
+ return getTable(identifier(tableName));
+ }
+
+ private void writePartitionRows(FileStoreTable table, String partition,
int... values)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int value : values) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ value,
+ BinaryString.fromString("base-" + value)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void writeDedicatedPartitionRows(FileStoreTable table, String
partition, int... values)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int value : values) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ value,
+ BinaryString.fromString("base-" + value),
+ new BlobData(new byte[] {(byte) value})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void writePartialStrings(
+ FileStoreTable table, String partition, long firstRowId, int...
values)
+ throws Exception {
+ RowType writeType = table.rowType().project(Arrays.asList("pt", "f1"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int value : values) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ BinaryString.fromString("updated-" + value)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ setFirstRowId(messages, firstRowId);
+ commit.commit(messages);
+ }
+ }
+
+ private static BinaryRow partition(FileStoreTable table, String value) {
+ return new InternalRowSerializer(table.schema().logicalPartitionType())
+ .toBinaryRow(GenericRow.of(BinaryString.fromString(value)));
+ }
+
+ private static List<String> readPartitionedRowsWithRowIds(FileStoreTable
table)
+ throws IOException {
+ RowType readType = SpecialFields.rowTypeWithRowId(table.rowType());
+ int rowIdIndex = table.rowType().getFieldCount();
+ boolean withBlob = rowIdIndex == 4;
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ String value = row.getString(0) + "|" + row.getInt(1);
+ if (withBlob) {
+ value +=
+ "|"
+ + row.getString(2)
+ + "|"
+ + (row.getBlob(3).toData()[0] &
0xFF);
+ }
+ rows.add(value + "|" + row.getLong(rowIdIndex));
+ });
+ }
+ rows.sort(
+ Comparator.comparingLong(
+ row ->
Long.parseLong(row.substring(row.lastIndexOf('|') + 1))));
+ return rows;
+ }
+
+ private static Map<String, Range> relativeFileRanges(
+ FileStoreTable table, BinaryRow partition) {
+ List<DataFileMeta> dataFiles = currentDataFiles(table, partition);
+ RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ Map<String, Range> result = new HashMap<>();
+ for (List<DataFileMeta> group :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
+ Range anchorRange = retrieveAnchorFile(group, file ->
file).nonNullRowIdRange();
+ for (DataFileMeta file : group) {
+ Range range = file.nonNullRowIdRange();
+ result.put(
+ file.fileName(),
+ new Range(range.from - anchorRange.from, range.to -
anchorRange.from));
+ }
+ }
+ return result;
+ }
+
+ private static List<DataFileMeta> currentDataFiles(FileStoreTable table,
BinaryRow partition) {
+ return table.store().newScan().plan().files().stream()
+ .filter(entry -> entry.partition().equals(partition))
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ }
+
+ private static DataEvolutionRowIdReassigner reassignerWithBeforeCommit(
+ FileStoreTable table, Runnable beforeCommit) throws Exception {
+ Constructor<DataEvolutionRowIdReassigner> constructor =
+ DataEvolutionRowIdReassigner.class.getDeclaredConstructor(
+ FileStoreTable.class, PartitionPredicate.class,
Runnable.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(table, null, beforeCommit);
+ }
+
@Override
protected Schema schemaDefault() {
Schema.Builder schemaBuilder = Schema.newBuilder();
@@ -671,12 +1003,18 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
private void commitDeletionVectors(FileStoreTable table, List<DvSpec>
deletionVectorSpecs)
throws Exception {
+ commitDeletionVectors(table, BinaryRow.EMPTY_ROW, deletionVectorSpecs);
+ }
+
+ private void commitDeletionVectors(
+ FileStoreTable table, BinaryRow partition, List<DvSpec>
deletionVectorSpecs)
+ throws Exception {
BaseAppendDeleteFileMaintainer maintainer =
BaseAppendDeleteFileMaintainer.forUnawareAppend(
table.store().newIndexFileHandler(),
table.latestSnapshot().get(),
- BinaryRow.EMPTY_ROW);
- Map<Range, String> anchorFiles = anchorFilesByRange(table);
+ partition);
+ Map<Range, String> anchorFiles = anchorFilesByRange(table, partition);
for (DvSpec spec : deletionVectorSpecs) {
DeletionVector deletionVector = new BitmapDeletionVector();
@@ -700,7 +1038,7 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
table,
Collections.singletonList(
new CommitMessageImpl(
- BinaryRow.EMPTY_ROW,
+ partition,
UNAWARE_BUCKET,
null,
new DataIncrement(
@@ -720,10 +1058,11 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
}
private Map<Range, String> anchorFilesByRange(FileStoreTable table) {
- List<DataFileMeta> dataFiles =
- table.store().newScan().plan().files().stream()
- .map(ManifestEntry::file)
- .collect(Collectors.toList());
+ return anchorFilesByRange(table, BinaryRow.EMPTY_ROW);
+ }
+
+ private Map<Range, String> anchorFilesByRange(FileStoreTable table,
BinaryRow partition) {
+ List<DataFileMeta> dataFiles = currentDataFiles(table, partition);
RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
Map<Range, String> result = new HashMap<>();
for (List<DataFileMeta> group :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
@@ -917,6 +1256,18 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
return result;
}
+ private static List<String>
liveDeletionVectorIndexFileNames(FileStoreTable table) {
+ List<String> result = new ArrayList<>();
+ for (IndexManifestEntry entry :
+ table.store()
+ .newIndexFileHandler()
+ .scan(table.latestSnapshot().get(),
DELETION_VECTORS_INDEX)) {
+ result.add(entry.indexFile().fileName());
+ }
+ Collections.sort(result);
+ return result;
+ }
+
private static List<String> readRows(ReadBuilder readBuilder,
TableScan.Plan plan)
throws IOException {
List<String> rows = new ArrayList<>();
