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 a0226d338c [core] Support row ID reassignment across concurrent
compaction (#10780)
a0226d338c is described below
commit a0226d338cfa936ad65e45ed0687da16fbd452ba
Author: YeJunHao <[email protected]>
AuthorDate: Mon Oct 26 16:40:06 2026 +0800
[core] Support row ID reassignment across concurrent compaction (#10780)
---
.../DataEvolutionRowIdReassigner.java | 25 +-
.../DataEvolutionRowIdReassignerTest.java | 726 +--------------------
.../table/DataEvolutionDeletionVectorTest.java | 55 +-
3 files changed, 91 insertions(+), 715 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
index e357decb6b..6a6d87dc46 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
@@ -487,8 +487,7 @@ public class DataEvolutionRowIdReassigner {
e);
}
- if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT
- || snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE)
{
+ if (snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE) {
throw new RuntimeException(
String.format(
"Abort row-id reassignment because %s snapshot
%s was committed after snapshot %s.",
@@ -498,14 +497,15 @@ public class DataEvolutionRowIdReassigner {
continue;
}
checkState(
- snapshot.commitKind() == Snapshot.CommitKind.APPEND,
+ snapshot.commitKind() == Snapshot.CommitKind.APPEND
+ || snapshot.commitKind() ==
Snapshot.CommitKind.COMPACT,
"Unsupported snapshot kind %s while advancing row-id
assignment.",
snapshot.commitKind());
for (ManifestFileMeta manifestMeta :
manifestList.readDeltaManifests(snapshot)) {
boolean needsReassign =
- appendedManifestNeedsReassign(
- assignmentPlan, manifestFile, manifestMeta,
snapshot.id());
+ deltaManifestNeedsReassign(
+ assignmentPlan, manifestFile, manifestMeta,
snapshot);
newManifestNeedsReassign.put(manifestMeta.fileName(),
needsReassign);
if (needsReassign) {
manifestMetasToRewrite.put(manifestMeta.fileName(),
manifestMeta);
@@ -562,11 +562,11 @@ public class DataEvolutionRowIdReassigner {
}
}
- private boolean appendedManifestNeedsReassign(
+ private boolean deltaManifestNeedsReassign(
AssignmentPlan assignmentPlan,
ManifestFile manifestFile,
ManifestFileMeta manifestMeta,
- long appendSnapshotId) {
+ Snapshot snapshot) {
boolean needsReassign = false;
try (CloseableIterator<ProjectedManifestEntry> entries =
manifestFile.scan(
@@ -577,11 +577,11 @@ public class DataEvolutionRowIdReassigner {
continue;
}
checkState(
- entry.isAdd(),
+ snapshot.commitKind() != Snapshot.CommitKind.APPEND ||
entry.isAdd(),
"APPEND snapshot %s contains a non-ADD entry in
manifest %s.",
- appendSnapshotId,
+ snapshot.id(),
manifestMeta.fileName());
- if (appendedEntryNeedsReassign(assignmentPlan, entry)) {
+ if (entryNeedsReassign(assignmentPlan, entry)) {
needsReassign = true;
}
}
@@ -594,8 +594,7 @@ public class DataEvolutionRowIdReassigner {
}
}
- private boolean appendedEntryNeedsReassign(
- AssignmentPlan assignmentPlan, ManifestEntry appendedEntry) {
+ private boolean entryNeedsReassign(AssignmentPlan assignmentPlan,
ManifestEntry appendedEntry) {
RowRangeMappingIndex mapping =
assignmentPlan.relativeRowIdMappings.mappings.get(appendedEntry.partition());
if (mapping == null) {
@@ -609,7 +608,7 @@ public class DataEvolutionRowIdReassigner {
checkState(
!mapping.overlaps(appendedRange),
- "Cannot advance row-id assignment because appended row-id
range %s partially overlaps planned ranges in partition %s.",
+ "Cannot advance row-id assignment because row-id range %s
partially overlaps planned ranges in partition %s.",
appendedRange,
appendedEntry.partition());
return false;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index abf14e67b1..b0ad384968 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -94,8 +94,6 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.io.OutputStream;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -106,7 +104,6 @@ import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
@@ -247,408 +244,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return schemaBuilder.build();
}
- @Test
- public void testCompactAndLegacyPlansMatchOnSameHistoricalSnapshot()
throws Exception {
- FileStoreTable table = createTableWithInterleavedPartitions();
- FileEntry.Identifier historicalFile = compactOneFile(table, "pt=a/");
- assertAddDeleteEntriesConsistent(table, historicalFile);
-
- Snapshot snapshot = table.snapshotManager().latestSnapshot();
- ManifestList manifestList =
table.store().manifestListFactory().create();
- List<ManifestFileMeta> manifestMetas =
manifestList.readDataManifests(snapshot);
- List<String> beforeManifestFiles = dataManifestFileNames(table);
- int[] historicalManifestOrdinals =
- manifestOrdinalsContaining(table, manifestMetas,
historicalFile);
- assertThat(historicalManifestOrdinals).hasSize(2);
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
-
assertThat(compactPlan.get().manifestOrdinals).contains(historicalManifestOrdinals);
-
-
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(snapshot.id());
-
assertThat(dataManifestFileNames(table)).containsExactlyElementsOf(beforeManifestFiles);
- }
-
- @Test
- public void testCompactAndLegacyEmptyPlansMatch() throws Exception {
- createTableDefault();
- FileStoreTable table = getTableDefault();
- writeRows(table, "a", 0, 1);
- writeRows(table, "a", 2);
- writeRows(table, "b", 3, 4);
-
- Snapshot snapshot = table.snapshotManager().latestSnapshot();
- ManifestList manifestList =
table.store().manifestListFactory().create();
- List<ManifestFileMeta> manifestMetas =
manifestList.readDataManifests(snapshot);
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isEmpty();
-
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(snapshot.id());
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchWithPartitionPredicate() throws
Exception {
- FileStoreTable table = createTableWithInterleavedPartitions();
- Snapshot snapshot = table.snapshotManager().latestSnapshot();
- ManifestList manifestList =
table.store().manifestListFactory().create();
- List<ManifestFileMeta> manifestMetas =
manifestList.readDataManifests(snapshot);
- PartitionPredicate onlyA = partitionPredicate(table, "a");
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, onlyA);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, onlyA);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
- assertThat(compactPlan.get().partitionMappings).hasSize(1);
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchWithoutPartitionStats() throws
Exception {
- FileStoreTable table = createTableWithInterleavedPartitions();
- Snapshot snapshot = table.snapshotManager().latestSnapshot();
- List<ManifestFileMeta> manifestMetas =
-
table.store().manifestListFactory().create().readDataManifests(snapshot);
- List<ManifestFileMeta> metasWithoutPartitionStats = new
ArrayList<>(manifestMetas.size());
- for (ManifestFileMeta manifestMeta : manifestMetas) {
-
metasWithoutPartitionStats.add(withoutPartitionStats(manifestMeta));
- }
-
- Optional<AssignmentPlanView> compactPlan =
- compactPlanView(table, metasWithoutPartitionStats, null);
- Optional<AssignmentPlanView> legacyPlan =
- legacyPlanView(table, metasWithoutPartitionStats, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchForSpecialFileRanges() throws
Exception {
- createTableDefault();
- FileStoreTable table = getTableDefault();
- InternalRowSerializer partitionSerializer =
- new
InternalRowSerializer(table.schema().logicalPartitionType());
- BinaryRow partitionA =
-
partitionSerializer.toBinaryRow(GenericRow.of(BinaryString.fromString("a"))).copy();
- BinaryRow partitionB =
-
partitionSerializer.toBinaryRow(GenericRow.of(BinaryString.fromString("b"))).copy();
- BinaryRow partitionC =
-
partitionSerializer.toBinaryRow(GenericRow.of(BinaryString.fromString("c"))).copy();
-
- List<ManifestEntry> entries =
- Arrays.asList(
- planningManifestEntry(partitionA,
"ordinary-a-0.parquet", 10L, 10L),
- planningManifestEntry(partitionA, "ordinary-a-0.blob",
12L, 4L),
- planningManifestEntry(partitionA,
"ordinary-a-0.vector.bin", 16L, 2L),
- planningManifestEntry(partitionA,
"ordinary-a-1.parquet", 30L, 10L),
- planningManifestEntry(partitionB,
"ordinary-b-0.parquet", 0L, 5L),
- planningManifestEntry(partitionB,
"ordinary-b-1.parquet", 5L, 5L),
- planningManifestEntry(partitionC, "special-c-0.blob",
50L, 6L),
- planningManifestEntry(partitionC,
"special-c-0.vector.bin", 54L, 7L),
- planningManifestEntry(partitionC, "special-c-1.blob",
80L, 3L),
- planningManifestEntry(partitionC,
"special-c-1.vector.bin", 82L, 4L));
- List<ManifestFileMeta> manifestMetas =
- table.store().manifestFileFactory().create().write(entries);
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
- assertThat(compactPlan.get().partitionMappings).hasSize(2);
- PartitionMappingView ordinaryMapping =
compactPlan.get().partitionMappings.get(0);
- assertThat(ordinaryMapping.oldStarts).containsExactly(10L, 30L);
- assertThat(ordinaryMapping.oldEnds).containsExactly(19L, 39L);
- assertThat(ordinaryMapping.newStarts).containsExactly(0L, 10L);
- PartitionMappingView allSpecialMapping =
compactPlan.get().partitionMappings.get(1);
- assertThat(allSpecialMapping.oldStarts).containsExactly(50L, 80L);
- assertThat(allSpecialMapping.oldEnds).containsExactly(60L, 85L);
- assertThat(allSpecialMapping.newStarts).containsExactly(20L, 31L);
- assertThat(compactPlan.get().totalOffset).isEqualTo(37L);
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchForRawNaNPartitionPayloads()
throws Exception {
- Schema.Builder schemaBuilder = Schema.newBuilder();
- schemaBuilder.column("float_pt", DataTypes.FLOAT());
- schemaBuilder.column("double_pt", DataTypes.DOUBLE());
- schemaBuilder.column("id", DataTypes.INT());
- schemaBuilder.partitionKeys("float_pt", "double_pt");
- schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
- schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
- catalog.createTable(identifier(), schemaBuilder.build(), true);
- FileStoreTable table = getTableDefault();
-
- InternalRowSerializer partitionSerializer =
- new
InternalRowSerializer(table.schema().logicalPartitionType());
- float floatNaN1 = Float.intBitsToFloat(0x7fc00001);
- float floatNaN2 = Float.intBitsToFloat(0x7fc00002);
- double doubleNaN1 = Double.longBitsToDouble(0x7ff8000000000001L);
- double doubleNaN2 = Double.longBitsToDouble(0x7ff8000000000002L);
- BinaryRow doubleScanFirst =
- partitionSerializer.toBinaryRow(GenericRow.of(0.0f,
doubleNaN1)).copy();
- BinaryRow doubleLegacyFirst =
- partitionSerializer.toBinaryRow(GenericRow.of(0.0f,
doubleNaN2)).copy();
- BinaryRow floatScanFirst =
- partitionSerializer.toBinaryRow(GenericRow.of(floatNaN1,
0.0d)).copy();
- BinaryRow floatLegacyFirst =
- partitionSerializer.toBinaryRow(GenericRow.of(floatNaN2,
0.0d)).copy();
-
-
assertThat(doubleScanFirst.toBytes()).isNotEqualTo(doubleLegacyFirst.toBytes());
-
assertThat(floatScanFirst.toBytes()).isNotEqualTo(floatLegacyFirst.toBytes());
-
- List<ManifestEntry> entries =
- Arrays.asList(
- planningManifestEntry(doubleScanFirst,
"double-scan-100.parquet", 100L, 1L),
- planningManifestEntry(
- doubleLegacyFirst, "double-legacy-10.parquet",
10L, 1L),
- planningManifestEntry(doubleScanFirst,
"double-scan-102.parquet", 102L, 1L),
- planningManifestEntry(
- doubleLegacyFirst, "double-legacy-12.parquet",
12L, 1L),
- planningManifestEntry(floatScanFirst,
"float-scan-300.parquet", 300L, 1L),
- planningManifestEntry(
- floatLegacyFirst, "float-legacy-200.parquet",
200L, 1L),
- planningManifestEntry(floatScanFirst,
"float-scan-302.parquet", 302L, 1L),
- planningManifestEntry(
- floatLegacyFirst, "float-legacy-202.parquet",
202L, 1L));
- List<ManifestFileMeta> manifestMetas =
- table.store().manifestFileFactory().create().write(entries);
- assertThat(manifestMetas).hasSize(1);
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
-
- assertThat(compactPlan).isPresent();
- AssignmentPlanView plan = compactPlan.get();
- assertThat(plan.manifestOrdinals).containsExactly(0);
- assertThat(plan.partitionMappings).hasSize(4);
-
- PartitionMappingView doubleLegacyMapping =
plan.partitionMappings.get(0);
-
assertThat(doubleLegacyMapping.partitionBytes).containsExactly(doubleLegacyFirst.toBytes());
- assertThat(doubleLegacyMapping.oldStarts).containsExactly(10L, 12L);
- assertThat(doubleLegacyMapping.oldEnds).containsExactly(10L, 12L);
- assertThat(doubleLegacyMapping.newStarts).containsExactly(0L, 1L);
-
- PartitionMappingView doubleScanMapping = plan.partitionMappings.get(1);
-
assertThat(doubleScanMapping.partitionBytes).containsExactly(doubleScanFirst.toBytes());
- assertThat(doubleScanMapping.oldStarts).containsExactly(100L, 102L);
- assertThat(doubleScanMapping.oldEnds).containsExactly(100L, 102L);
- assertThat(doubleScanMapping.newStarts).containsExactly(2L, 3L);
-
- PartitionMappingView floatLegacyMapping =
plan.partitionMappings.get(2);
-
assertThat(floatLegacyMapping.partitionBytes).containsExactly(floatLegacyFirst.toBytes());
- assertThat(floatLegacyMapping.oldStarts).containsExactly(200L, 202L);
- assertThat(floatLegacyMapping.oldEnds).containsExactly(200L, 202L);
- assertThat(floatLegacyMapping.newStarts).containsExactly(4L, 5L);
-
- PartitionMappingView floatScanMapping = plan.partitionMappings.get(3);
-
assertThat(floatScanMapping.partitionBytes).containsExactly(floatScanFirst.toBytes());
- assertThat(floatScanMapping.oldStarts).containsExactly(300L, 302L);
- assertThat(floatScanMapping.oldEnds).containsExactly(300L, 302L);
- assertThat(floatScanMapping.newStarts).containsExactly(6L, 7L);
- assertThat(plan.totalOffset).isEqualTo(8L);
- }
-
- @Test
- public void testDeleteEntryAtEndDoesNotAffectEqualPartitionOrdering()
throws Exception {
- Schema.Builder schemaBuilder = Schema.newBuilder();
- schemaBuilder.column("float_pt", DataTypes.FLOAT());
- schemaBuilder.column("id", DataTypes.INT());
- schemaBuilder.partitionKeys("float_pt");
- schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
- schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
- catalog.createTable(identifier(), schemaBuilder.build(), true);
- FileStoreTable table = getTableDefault();
-
- InternalRowSerializer partitionSerializer =
- new
InternalRowSerializer(table.schema().logicalPartitionType());
- BinaryRow laterPartition =
- partitionSerializer
-
.toBinaryRow(GenericRow.of(Float.intBitsToFloat(0x7fc00001)))
- .copy();
- BinaryRow earlierPartition =
- partitionSerializer
-
.toBinaryRow(GenericRow.of(Float.intBitsToFloat(0x7fc00002)))
- .copy();
-
assertThat(laterPartition.toBytes()).isNotEqualTo(earlierPartition.toBytes());
-
- DataFileMeta deletedFile =
- planningDataFile("deleted-0.parquet", 0L, 1,
Collections.emptyList(), null, null);
- List<ManifestEntry> entries =
- Arrays.asList(
- ManifestEntry.create(FileKind.ADD, laterPartition, 0,
1, deletedFile),
- planningManifestEntry(earlierPartition,
"earlier-live-10.parquet", 10L, 1L),
- planningManifestEntry(earlierPartition,
"earlier-live-12.parquet", 12L, 1L),
- planningManifestEntry(laterPartition,
"later-live-100.parquet", 100L, 1L),
- planningManifestEntry(laterPartition,
"later-live-102.parquet", 102L, 1L),
- // Keep DELETE after its ADD, matching manifest
compaction ordering.
- ManifestEntry.create(FileKind.DELETE, laterPartition,
0, 1, deletedFile));
- ManifestFile manifestFile =
table.store().manifestFileFactory().create();
- List<ManifestFileMeta> manifestMetas = manifestFile.write(entries);
- assertThat(manifestMetas).hasSize(1);
- assertThat(
- manifestFile.read(
- manifestMetas.get(0).fileName(),
manifestMetas.get(0).fileSize()))
- .extracting(ManifestEntry::kind)
- .containsExactly(
- FileKind.ADD,
- FileKind.ADD,
- FileKind.ADD,
- FileKind.ADD,
- FileKind.ADD,
- FileKind.DELETE);
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
-
- assertThat(compactPlan).isPresent();
- AssignmentPlanView plan = compactPlan.get();
- assertThat(plan.partitionMappings).hasSize(2);
- PartitionMappingView earlierMapping = plan.partitionMappings.get(0);
-
assertThat(earlierMapping.partitionBytes).containsExactly(earlierPartition.toBytes());
- assertThat(earlierMapping.oldStarts).containsExactly(10L, 12L);
- PartitionMappingView laterMapping = plan.partitionMappings.get(1);
-
assertThat(laterMapping.partitionBytes).containsExactly(laterPartition.toBytes());
- assertThat(laterMapping.oldStarts).containsExactly(100L, 102L);
- assertThat(plan.totalOffset).isEqualTo(4L);
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchForComplexDeleteIdentifier()
throws Exception {
- createTableDefault();
- FileStoreTable table = getTableDefault();
- BinaryRow partition =
- new
InternalRowSerializer(table.schema().logicalPartitionType())
-
.toBinaryRow(GenericRow.of(BinaryString.fromString("a")))
- .copy();
- DataFileMeta deletedFile =
- DataFileMeta.forAppend(
- "deleted.parquet",
- 1L,
- 1L,
- SimpleStats.EMPTY_STATS,
- 7L,
- 9L,
- 0L,
- Arrays.asList("extra-0", "extra-1"),
- new byte[] {1, 2, 3, 4},
- FileSource.APPEND,
- null,
- "/external/deleted.parquet",
- 1L,
- null)
- .upgrade(3);
- ManifestFile manifestFile =
table.store().manifestFileFactory().create();
- List<ManifestFileMeta> manifestMetas = new ArrayList<>();
- manifestMetas.addAll(
- manifestFile.write(
- Arrays.asList(
- planningManifestEntry(partition,
"current-0.parquet", 0L, 1L),
- ManifestEntry.create(FileKind.ADD, partition,
7, 8, deletedFile),
- planningManifestEntry(partition,
"current-2.parquet", 2L, 1L))));
- manifestMetas.addAll(
- manifestFile.write(
- Collections.singletonList(
- ManifestEntry.create(
- FileKind.DELETE, partition, 7, 8,
deletedFile))));
-
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
- assertThat(compactPlan.get().partitionMappings).hasSize(1);
- PartitionMappingView mapping =
compactPlan.get().partitionMappings.get(0);
- assertThat(mapping.oldStarts).containsExactly(0L, 2L);
- assertThat(mapping.oldEnds).containsExactly(0L, 2L);
- assertThat(mapping.newStarts).containsExactly(0L, 1L);
- }
-
- @Test
- public void testCompactAndLegacyPlansMatchForDeleteIdentifierBoundaries()
throws Exception {
- createTableDefault();
- FileStoreTable table = getTableDefault();
- InternalRowSerializer partitionSerializer =
- new
InternalRowSerializer(table.schema().logicalPartitionType());
- BinaryRow partitionA =
-
partitionSerializer.toBinaryRow(GenericRow.of(BinaryString.fromString("a"))).copy();
- BinaryRow partitionB =
-
partitionSerializer.toBinaryRow(GenericRow.of(BinaryString.fromString("b"))).copy();
-
- List<String> extraFiles = Arrays.asList("extra-0", "extra-1");
- byte[] embeddedIndex = {1, 2, 3, 4};
- String externalPath = "/external/identity.parquet";
- DataFileMeta identityFile =
- planningDataFile(
- "identity.parquet", 2L, 3, extraFiles, embeddedIndex,
externalPath);
- DataFileMeta totalBucketsFile =
- planningDataFile(
- "total-buckets.parquet", 6L, 3, extraFiles,
embeddedIndex, externalPath);
-
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(planningManifestEntry(partitionA, "anchor-0.parquet", 0L,
1L));
- entries.add(ManifestEntry.create(FileKind.ADD, partitionA, 7, 8,
identityFile));
- entries.add(planningManifestEntry(partitionA, "anchor-4.parquet", 4L,
1L));
- entries.add(ManifestEntry.create(FileKind.ADD, partitionA, 9, 16,
totalBucketsFile));
-
- // Each DELETE differs from identityFile in exactly one Identifier
field.
- entries.add(ManifestEntry.create(FileKind.DELETE, partitionB, 7, 8,
identityFile));
- entries.add(ManifestEntry.create(FileKind.DELETE, partitionA, 6, 8,
identityFile));
- List<DataFileMeta> fileFieldNearMisses =
- Arrays.asList(
- identityFile.upgrade(4),
- planningDataFile(
- "different-name.parquet",
- 2L,
- 3,
- extraFiles,
- embeddedIndex,
- externalPath),
- planningDataFile(
- "identity.parquet",
- 2L,
- 3,
- Arrays.asList("extra-1", "extra-0"),
- embeddedIndex,
- externalPath),
- planningDataFile(
- "identity.parquet",
- 2L,
- 3,
- extraFiles,
- new byte[] {1, 2, 3, 5},
- externalPath),
- planningDataFile(
- "identity.parquet",
- 2L,
- 3,
- extraFiles,
- embeddedIndex,
- "/external/different.parquet"));
- for (DataFileMeta nearMiss : fileFieldNearMisses) {
- entries.add(ManifestEntry.create(FileKind.DELETE, partitionA, 7,
8, nearMiss));
- }
-
- // totalBuckets is deliberately different but is not part of
FileEntry.Identifier.
- entries.add(ManifestEntry.create(FileKind.DELETE, partitionA, 9, 32,
totalBucketsFile));
-
- List<ManifestFileMeta> manifestMetas =
- table.store().manifestFileFactory().create().write(entries);
- Optional<AssignmentPlanView> compactPlan = compactPlanView(table,
manifestMetas, null);
- Optional<AssignmentPlanView> legacyPlan = legacyPlanView(table,
manifestMetas, null);
-
- assertPlanViewsEqual(compactPlan, legacyPlan);
- assertThat(compactPlan).isPresent();
- assertThat(compactPlan.get().partitionMappings).hasSize(1);
- PartitionMappingView mapping =
compactPlan.get().partitionMappings.get(0);
- assertThat(mapping.oldStarts).containsExactly(0L, 2L, 4L);
- assertThat(mapping.oldEnds).containsExactly(0L, 2L, 4L);
- assertThat(mapping.newStarts).containsExactly(0L, 1L, 2L);
- assertThat(compactPlan.get().totalOffset).isEqualTo(3L);
- }
-
@Test
public void testReassignRowIdsByPartition() throws Exception {
FileStoreTable table = createTableWithInterleavedPartitions();
@@ -1032,7 +627,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
}
- @SuppressWarnings("unchecked")
private void assertPersistedPlan(FileStoreTable table) throws Exception {
Snapshot snapshot =
Snapshot.fromJson(table.snapshotManager().latestSnapshot().toJson());
assertThat(snapshot.properties())
@@ -1049,8 +643,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(assignment.firstAssignedRowId())
.isEqualTo(table.snapshotManager().snapshot(snapshot.id() -
1).nextRowId());
assertThat(assignment.nextRowId()).isEqualTo(snapshot.nextRowId());
- Map<BinaryRow, RowRangeMappingIndex> mappings =
- (Map<BinaryRow, RowRangeMappingIndex>) fieldValue(assignment,
"rowIdMappings");
Map<String, ManifestEntry> previous = new HashMap<>();
for (ManifestEntry entry :
table.store().newScan().withSnapshot(snapshot.id() -
1).plan().files()) {
@@ -1059,16 +651,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
for (ManifestEntry entry : currentEntries(table)) {
Range oldRange =
previous.get(entry.file().fileName()).file().nonNullRowIdRange();
Range newRange = entry.file().nonNullRowIdRange();
- RowRangeMappingIndex mapping = mappings.get(entry.partition());
- if (oldRange.equals(newRange)) {
- if (mapping != null) {
- assertThat(mapping.map(oldRange)).isEmpty();
- assertThat(mapping.overlaps(oldRange)).isFalse();
- }
- } else {
- assertThat(mapping).isNotNull();
- assertThat(mapping.map(oldRange)).hasValue(newRange);
- }
+ assertThat(assignment.mapRowRange(entry.partition(),
oldRange)).isEqualTo(newRange);
}
}
@@ -1095,13 +678,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsEntry("pt=a/", Collections.singletonList(0L))
.containsEntry("pt=z/", Collections.singletonList(2L));
- Snapshot planningSnapshot = table.snapshotManager().latestSnapshot();
- List<ManifestFileMeta> planningManifests =
-
table.store().manifestListFactory().create().readDataManifests(planningSnapshot);
- assertPlanViewsEqual(
- compactPlanView(table, planningManifests, null),
- legacyPlanView(table, planningManifests, null));
-
DataEvolutionRowIdReassigner.Result result =
new DataEvolutionRowIdReassigner(table)
.reassign("test-reassign-null-partition-row-id");
@@ -1195,8 +771,10 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsEntry("pt=b/", Arrays.asList(9L, 10L));
}
- @Test
- public void testReassignAbortsForPartiallyOverlappingConcurrentAppend()
throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testReassignAbortsForPartiallyOverlappingConcurrentAppend(boolean compact)
+ throws Exception {
FileStoreTable table = createTableWithInterleavedPartitions();
Snapshot before = table.snapshotManager().latestSnapshot();
@@ -1209,6 +787,9 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
() -> {
if
(appended.compareAndSet(false, true)) {
try {
+ if (compact) {
+
compactManifests(table);
+ }
writePartialPayload(
table,
"a",
@@ -1225,7 +806,8 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.hasMessageContaining("partially overlaps planned range");
assertThat(appended).isTrue();
-
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
+ assertThat(table.snapshotManager().latestSnapshot().id())
+ .isEqualTo(before.id() + (compact ? 2 : 1));
}
@Test
@@ -1290,11 +872,14 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsExactly("0|a|v0", "1|b|v1", "2|a|v2", "3|b|v3",
"4|a|v4");
}
- @Test
- public void
testReassignKeepsHistoricalAddDeleteEntriesConsistentAfterRetry() throws
Exception {
+ @ParameterizedTest
+ @ValueSource(strings = {"none", "data", "manifest"})
+ public void testReassignKeepsHistoricalEntriesConsistentAfterRetry(String
compaction)
+ throws Exception {
FileStoreTable table = createTableWithInterleavedPartitions();
FileEntry.Identifier compactedFile = compactOneFile(table, "pt=a/");
long oldFirstRowId = assertAddDeleteEntriesConsistent(table,
compactedFile);
+ List<String> plannedManifestFiles = dataManifestFileNames(table);
AtomicBoolean appended = new AtomicBoolean();
new DataEvolutionRowIdReassigner(
@@ -1303,6 +888,13 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
() -> {
if (appended.compareAndSet(false, true)) {
try {
+ if (compaction.equals("data")) {
+ compactOneFile(table, "pt=a/");
+ } else if (compaction.equals("manifest")) {
+ compactManifests(table);
+
assertThat(dataManifestFileNames(table))
+
.doesNotContainAnyElementsOf(plannedManifestFiles);
+ }
writeOneRow(table, "c", 100);
} catch (Exception e) {
throw new RuntimeException(e);
@@ -1312,8 +904,14 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.reassign("test-reassign-historical-add-delete");
assertThat(appended).isTrue();
- assertThat(assertAddDeleteEntriesConsistent(table, compactedFile))
- .isNotEqualTo(oldFirstRowId);
+ if (!compaction.equals("manifest")) {
+ assertThat(assertAddDeleteEntriesConsistent(table, compactedFile))
+ .isNotEqualTo(oldFirstRowId);
+ }
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(6L, 7L, 8L))
+ .containsEntry("pt=b/", Arrays.asList(9L, 10L))
+ .containsEntry("pt=c/", Collections.singletonList(5L));
assertThat(readTableRows(table))
.containsExactly("0|a|v0", "100|c|v100", "1|b|v1", "2|a|v2",
"3|b|v3", "4|a|v4");
}
@@ -1655,35 +1253,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.isEqualTo(Snapshot.CommitKind.OVERWRITE);
}
- @Test
- public void testReassignAbortsAfterConcurrentManifestCompaction() throws
Exception {
- FileStoreTable table = createTableWithInterleavedPartitions();
- Snapshot before = table.snapshotManager().latestSnapshot();
-
- AtomicBoolean compacted = new AtomicBoolean();
- assertThatThrownBy(
- () ->
- new DataEvolutionRowIdReassigner(
- table,
- null,
- () -> {
- if
(compacted.compareAndSet(false, true)) {
- try {
-
compactManifests(table);
- } catch (Exception e) {
- throw new
RuntimeException(e);
- }
- }
- })
-
.reassign("test-reassign-compact-conflict"))
- .isInstanceOf(RuntimeException.class)
- .hasMessageContaining("COMPACT snapshot");
-
-
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
- assertThat(table.snapshotManager().latestSnapshot().commitKind())
- .isEqualTo(Snapshot.CommitKind.COMPACT);
- }
-
@Test
public void testReassignContinuesAfterConcurrentAnalyze() throws Exception
{
FileStoreTable table = createTableWithInterleavedPartitions();
@@ -1724,13 +1293,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(6L);
- Snapshot planningSnapshot = table.snapshotManager().latestSnapshot();
- List<ManifestFileMeta> planningManifests =
-
table.store().manifestListFactory().create().readDataManifests(planningSnapshot);
- assertPlanViewsEqual(
- compactPlanView(table, planningManifests, null),
- legacyPlanView(table, planningManifests, null));
-
DataEvolutionRowIdReassigner.Result result =
new
DataEvolutionRowIdReassigner(table).reassign("test-reassign-multi-row");
@@ -2568,6 +2130,11 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
verifyReassignOutOfOrderPartitionEntries(LARGE_ENTRY_COUNT,
LARGE_MANIFEST_FILE_COUNT);
}
+ public static DataEvolutionRowIdReassigner reassignerWithBeforeCommit(
+ FileStoreTable table, Runnable beforeCommit) {
+ return new DataEvolutionRowIdReassigner(table, null, beforeCommit);
+ }
+
private FileStoreTable createTableWithInterleavedPartitions() throws
Exception {
createTableDefault();
FileStoreTable table = getTableDefault();
@@ -2687,12 +2254,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
ExpectedLargeManifestAssignment expectedAssignment =
expectedLargeManifestAssignment(table, before, entryCount);
- List<ManifestFileMeta> planningManifests =
-
table.store().manifestListFactory().create().readDataManifests(before);
- assertPlanViewsEqual(
- compactPlanView(table, planningManifests, null),
- legacyPlanView(table, planningManifests, null));
-
DataEvolutionRowIdReassigner.Result result =
new DataEvolutionRowIdReassigner(table)
.reassign("test-reassign-large-out-of-order-row-id");
@@ -2856,49 +2417,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
}
- private DataFileMeta planningDataFile(
- String fileName,
- long firstRowId,
- int level,
- List<String> extraFiles,
- byte[] embeddedIndex,
- String externalPath) {
- return DataFileMeta.forAppend(
- fileName,
- 1L,
- 1L,
- SimpleStats.EMPTY_STATS,
- 0L,
- 0L,
- 0L,
- extraFiles,
- embeddedIndex,
- FileSource.APPEND,
- null,
- externalPath,
- firstRowId,
- null)
- .upgrade(level);
- }
-
- private ManifestFileMeta withoutPartitionStats(ManifestFileMeta
manifestMeta) {
- return new ManifestFileMeta(
- manifestMeta.fileName(),
- manifestMeta.fileSize(),
- manifestMeta.numAddedFiles(),
- manifestMeta.numDeletedFiles(),
- null,
- manifestMeta.schemaId(),
- manifestMeta.minBucket(),
- manifestMeta.maxBucket(),
- manifestMeta.minLevel(),
- manifestMeta.maxLevel(),
- manifestMeta.minRowId(),
- manifestMeta.maxRowId(),
- null,
- null);
- }
-
private void assertReassignedOutOfOrderPartitionEntries(
FileStoreTable table,
int entryCount,
@@ -3794,180 +3312,6 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return result;
}
- private Optional<AssignmentPlanView> compactPlanView(
- FileStoreTable table,
- List<ManifestFileMeta> manifestMetas,
- PartitionPredicate partitionPredicate)
- throws Exception {
- Method planAssignment =
-
DataEvolutionRowIdReassigner.class.getDeclaredMethod("planAssignment",
List.class);
- planAssignment.setAccessible(true);
- Object result =
- planAssignment.invoke(
- new DataEvolutionRowIdReassigner(table,
partitionPredicate), manifestMetas);
- return assignmentPlanView(result, manifestMetas);
- }
-
- private Optional<AssignmentPlanView> legacyPlanView(
- FileStoreTable table,
- List<ManifestFileMeta> manifestMetas,
- PartitionPredicate partitionPredicate)
- throws Exception {
- Method planAssignment =
- LegacyDataEvolutionRowIdReassigner.class.getDeclaredMethod(
- "planAssignment", List.class, ManifestFile.class);
- planAssignment.setAccessible(true);
- Object result =
- planAssignment.invoke(
- new LegacyDataEvolutionRowIdReassigner(table,
partitionPredicate),
- manifestMetas,
- table.store().manifestFileFactory().create());
- return assignmentPlanView(result, manifestMetas);
- }
-
- private Optional<AssignmentPlanView> assignmentPlanView(
- Object optionalResult, List<ManifestFileMeta> manifestMetas)
throws Exception {
- assertThat(optionalResult).isInstanceOf(Optional.class);
- Optional<?> optionalPlan = (Optional<?>) optionalResult;
- if (!optionalPlan.isPresent()) {
- return Optional.empty();
- }
-
- Object assignmentPlan = optionalPlan.get();
- @SuppressWarnings("unchecked")
- List<ManifestFileMeta> rewrittenManifests =
- (List<ManifestFileMeta>) fieldValue(assignmentPlan,
"manifestMetasToRewrite");
- int[] manifestOrdinals = new int[rewrittenManifests.size()];
- for (int i = 0; i < rewrittenManifests.size(); i++) {
- manifestOrdinals[i] = manifestOrdinal(manifestMetas,
rewrittenManifests.get(i));
- }
-
- Object relativeMappings = fieldValue(assignmentPlan,
"relativeRowIdMappings");
- @SuppressWarnings("unchecked")
- Map<BinaryRow, RowRangeMappingIndex> mappings =
- (Map<BinaryRow, RowRangeMappingIndex>)
fieldValue(relativeMappings, "mappings");
- List<PartitionMappingView> partitionMappings = new
ArrayList<>(mappings.size());
- for (Map.Entry<BinaryRow, RowRangeMappingIndex> mapping :
mappings.entrySet()) {
- partitionMappings.add(
- new PartitionMappingView(
- mapping.getKey().toBytes(),
- longArrayField(mapping.getValue(), "oldStarts"),
- longArrayField(mapping.getValue(), "oldEnds"),
- longArrayField(mapping.getValue(), "newStarts"),
- (Long) fieldValue(mapping.getValue(),
"newStartOffset")));
- }
- long totalOffset = (Long) fieldValue(relativeMappings, "totalOffset");
- return Optional.of(
- new AssignmentPlanView(manifestOrdinals, partitionMappings,
totalOffset));
- }
-
- private void assertPlanViewsEqual(
- Optional<AssignmentPlanView> compact, Optional<AssignmentPlanView>
legacy) {
- assertThat(compact.isPresent()).isEqualTo(legacy.isPresent());
- if (!compact.isPresent()) {
- return;
- }
-
- AssignmentPlanView compactPlan = compact.get();
- AssignmentPlanView legacyPlan = legacy.get();
-
assertThat(compactPlan.manifestOrdinals).containsExactly(legacyPlan.manifestOrdinals);
- assertThat(compactPlan.totalOffset).isEqualTo(legacyPlan.totalOffset);
-
assertThat(compactPlan.partitionMappings).hasSameSizeAs(legacyPlan.partitionMappings);
- for (int i = 0; i < compactPlan.partitionMappings.size(); i++) {
- PartitionMappingView compactMapping =
compactPlan.partitionMappings.get(i);
- PartitionMappingView legacyMapping =
legacyPlan.partitionMappings.get(i);
-
assertThat(compactMapping.partitionBytes).containsExactly(legacyMapping.partitionBytes);
-
assertThat(compactMapping.oldStarts).containsExactly(legacyMapping.oldStarts);
-
assertThat(compactMapping.oldEnds).containsExactly(legacyMapping.oldEnds);
-
assertThat(compactMapping.newStarts).containsExactly(legacyMapping.newStarts);
-
assertThat(compactMapping.newStartOffset).isEqualTo(legacyMapping.newStartOffset);
- }
- }
-
- private int[] manifestOrdinalsContaining(
- FileStoreTable table,
- List<ManifestFileMeta> manifestMetas,
- FileEntry.Identifier identifier) {
- ManifestFile manifestFile =
table.store().manifestFileFactory().create();
- List<Integer> ordinals = new ArrayList<>();
- for (int i = 0; i < manifestMetas.size(); i++) {
- ManifestFileMeta manifestMeta = manifestMetas.get(i);
- for (ManifestEntry entry :
- manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize())) {
- if (entry.identifier().equals(identifier)) {
- ordinals.add(i);
- break;
- }
- }
- }
- int[] result = new int[ordinals.size()];
- for (int i = 0; i < ordinals.size(); i++) {
- result[i] = ordinals.get(i);
- }
- return result;
- }
-
- private static int manifestOrdinal(
- List<ManifestFileMeta> manifestMetas, ManifestFileMeta
plannedManifest) {
- for (int i = 0; i < manifestMetas.size(); i++) {
- if
(manifestMetas.get(i).fileName().equals(plannedManifest.fileName())) {
- return i;
- }
- }
- throw new AssertionError(
- "Planned manifest is not present in the source snapshot: "
- + plannedManifest.fileName());
- }
-
- private static long[] longArrayField(Object target, String fieldName)
throws Exception {
- long[] values = (long[]) fieldValue(target, fieldName);
- return Arrays.copyOf(values, values.length);
- }
-
- private static Object fieldValue(Object target, String fieldName) throws
Exception {
- Field field = target.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- return field.get(target);
- }
-
- private static class AssignmentPlanView {
-
- private final int[] manifestOrdinals;
- private final List<PartitionMappingView> partitionMappings;
- private final long totalOffset;
-
- private AssignmentPlanView(
- int[] manifestOrdinals,
- List<PartitionMappingView> partitionMappings,
- long totalOffset) {
- this.manifestOrdinals = manifestOrdinals;
- this.partitionMappings = partitionMappings;
- this.totalOffset = totalOffset;
- }
- }
-
- private static class PartitionMappingView {
-
- private final byte[] partitionBytes;
- private final long[] oldStarts;
- private final long[] oldEnds;
- private final long[] newStarts;
- private final long newStartOffset;
-
- private PartitionMappingView(
- byte[] partitionBytes,
- long[] oldStarts,
- long[] oldEnds,
- long[] newStarts,
- long newStartOffset) {
- this.partitionBytes = partitionBytes;
- this.oldStarts = oldStarts;
- this.oldEnds = oldEnds;
- this.newStarts = newStarts;
- this.newStartOffset = newStartOffset;
- }
- }
-
private static class ExpectedLargeManifestAssignment {
private final int[] entriesByPartition;
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 1ec1343ab1..a0c2afa26c 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
@@ -50,7 +50,6 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
-import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -77,7 +76,6 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
-import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -89,6 +87,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
+import static
org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignerTest.reassignerWithBeforeCommit;
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.format.blob.BlobFileFormat.isBlobFile;
@@ -189,6 +188,49 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
.containsExactly("a|0|0", "a|2|2", "b|4|4", "a|5|5", "a|7|7");
}
+ @Test
+ public void testRowIdReassignAcrossConcurrentDeletionMaterialization()
throws Exception {
+ FileStoreTable table =
createPartitionedReassignTable("reassign_materialize_dv", true);
+ writeDedicatedPartitionRows(table, "a", 0, 1, 2);
+ writeDedicatedPartitionRows(table, "b", 3, 4);
+ writeDedicatedPartitionRows(table, "a", 5, 6, 7);
+ writePartialStrings(table, "a", 0, 100, 101, 102);
+ commitDeletionVectors(
+ table,
+ partition(table, "a"),
+ Collections.singletonList(new DvSpec(new Range(0, 2), 1)));
+
+ AtomicBoolean materialized = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ reassignerWithBeforeCommit(
+ table,
+ () -> {
+ if (materialized.compareAndSet(false,
true)) {
+ try {
+ materializeDeletionVectors(table);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-concurrent-materialize");
+
+ assertThat(materialized).isTrue();
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.firstAssignedRowId).isEqualTo(10L);
+ assertThat(result.nextRowId).isEqualTo(16L);
+ // Materialized files retain their new IDs; surviving old files keep
the plan's offsets.
+ assertThat(readPartitionedRowsWithRowIds(table))
+ .containsExactly(
+ "b|3|base-3|3|3",
+ "b|4|base-4|4|4",
+ "a|0|updated-100|0|8",
+ "a|2|updated-102|2|9",
+ "a|5|base-5|5|13",
+ "a|6|base-6|6|14",
+ "a|7|base-7|7|15");
+ }
+
@Test
public void testRowIdReassignKeepsPartialWriteAndBlobFileOffsets() throws
Exception {
FileStoreTable table =
createPartitionedReassignTable("reassign_dedicated_dv_table", true);
@@ -1286,15 +1328,6 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
.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();
