This is an automated email from the ASF dual-hosted git repository.
leaves12138 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 1886d2f4e2 [core] Improve row id reassign retry reuse (#8506)
1886d2f4e2 is described below
commit 1886d2f4e297cbb3e2f72b041387037488aebba7
Author: YeJunHao <[email protected]>
AuthorDate: Sat Jul 11 13:05:59 2026 +0800
[core] Improve row id reassign retry reuse (#8506)
---
.../DataEvolutionRowIdReassigner.java | 744 +++++++-----
.../append/dataevolution/RowRangeMappingIndex.java | 20 +
.../DataEvolutionRowIdReassignerTest.java | 1198 +++++++++++++++++++-
.../dataevolution/RowRangeMappingIndexTest.java | 29 +
4 files changed, 1689 insertions(+), 302 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 b7261639b5..d81d98ed70 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
@@ -20,6 +20,7 @@ package org.apache.paimon.append.dataevolution;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.codegen.CodeGenUtils;
import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
@@ -39,6 +40,7 @@ import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RangeHelper;
@@ -69,9 +71,11 @@ public class DataEvolutionRowIdReassigner {
private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionRowIdReassigner.class);
private static final String COMMIT_USER_PREFIX = "reassign-row-id";
+ private static final int MAX_COMMIT_ATTEMPTS = 3;
private final FileStoreTable table;
private final @Nullable PartitionPredicate partitionPredicate;
+ private final Runnable beforeCommit;
public DataEvolutionRowIdReassigner(FileStoreTable table) {
this(table, null);
@@ -79,8 +83,17 @@ public class DataEvolutionRowIdReassigner {
public DataEvolutionRowIdReassigner(
FileStoreTable table, @Nullable PartitionPredicate
partitionPredicate) {
+ this(table, partitionPredicate, () -> {});
+ }
+
+ @VisibleForTesting
+ DataEvolutionRowIdReassigner(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionPredicate,
+ Runnable beforeCommit) {
this.table = table;
this.partitionPredicate = partitionPredicate;
+ this.beforeCommit = beforeCommit;
}
public Result reassign() {
@@ -116,86 +129,77 @@ public class DataEvolutionRowIdReassigner {
ManifestFile manifestFile =
table.store().manifestFileFactory().create();
ManifestList manifestList =
table.store().manifestListFactory().create();
- List<ManifestFileMeta> manifestMetas =
manifestList.readDataManifests(latest);
- AssignmentPlan assignment = planAssignment(manifestMetas,
manifestFile, nextRowId);
- if (!assignment.hasCurrentFiles) {
- return Result.skipped(
- latest.id(),
- nextRowId,
- partitionFilterEnabled()
- ? "partition filter matches no current files"
- : "table has no current files");
- }
- if (assignment.reassignedFileCount == 0) {
+ Optional<AssignmentPlan> optionalPlan =
+ planAssignment(manifestList.readDataManifests(latest),
manifestFile);
+ if (!optionalPlan.isPresent()) {
LOG.info(
- "Skip reassigning row IDs for table {} because partition
row IDs are already contiguous.",
+ "Skip reassigning row IDs for table {} because no
partition requires reassignment.",
table.name());
return Result.skipped(
- latest.id(), nextRowId, "partition row IDs are already
contiguous");
+ latest.id(), nextRowId, "no partition requires row-id
reassignment");
}
+ AssignmentPlan assignmentPlan = optionalPlan.get();
+
+ for (int attempt = 1; attempt <= MAX_COMMIT_ATTEMPTS; attempt++) {
+ Assignment assignment = assignmentPlan.createAssignment(latest);
+ CommitAssignmentResult commitResult =
+ commitAssignment(assignment, manifestFile, manifestList,
commitUser);
+ if (commitResult.success) {
+ LOG.info(
+ "Reassigned row IDs for table {} from {} to {},
partitions={}, files={}, rows={}.",
+ table.name(),
+ assignment.firstAssignedRowId,
+ assignment.nextRowId,
+ assignment.rowIdMappings.size(),
+ commitResult.fileCount,
+ assignment.logicalRowCount());
+ return new Result(
+ assignment.snapshot.id(),
+ assignment.snapshot.id() + 1,
+ commitResult.fileCount,
+ assignment.logicalRowCount(),
+ commitResult.indexFileCount,
+ assignment.firstAssignedRowId,
+ assignment.nextRowId);
+ }
- Pair<String, Long> baseManifestList =
- writeBaseManifestList(
- manifestMetas, assignment.rewrittenManifestMetas,
manifestList);
- Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
-
- RewrittenIndexManifest rewrittenIndexManifest =
rewriteIndexManifest(latest, assignment);
-
- try (FileStoreCommitImpl commit =
- (FileStoreCommitImpl) table.store().newCommit(commitUser,
table)) {
- boolean success =
- commit.replaceManifestList(
- latest,
- latest.totalRecordCount(),
- baseManifestList,
- deltaManifestList,
- rewrittenIndexManifest.indexManifest,
- assignment.nextRowId);
- if (!success) {
+ if (attempt == MAX_COMMIT_ATTEMPTS) {
throw new RuntimeException(
"Failed to reassign row IDs because a newer snapshot
has been committed.");
}
+
+ Snapshot newLatest = table.snapshotManager().latestSnapshot();
+ checkState(newLatest != null, "Latest snapshot disappeared while
reassigning row IDs.");
+ assignmentPlan =
+ advanceAssignmentPlan(
+ assignmentPlan, latest, newLatest, manifestFile,
manifestList);
+ LOG.info(
+ "Failed to commit row-id reassignment for table {} based
on snapshot {} because snapshot {} has been committed. Retrying {}/{} with the
updated assignment plan.",
+ table.name(),
+ latest.id(),
+ newLatest.id(),
+ attempt + 1,
+ MAX_COMMIT_ATTEMPTS);
+ latest = newLatest;
}
- LOG.info(
- "Reassigned row IDs for table {} from {} to {}, partitions={},
files={}, rows={}.",
- table.name(),
- nextRowId,
- assignment.nextRowId,
- assignment.rowIdMappings.size(),
- assignment.reassignedFileCount,
- assignment.logicalRowCount);
- return new Result(
- latest.id(),
- latest.id() + 1,
- assignment.reassignedFileCount,
- assignment.logicalRowCount,
- rewrittenIndexManifest.indexFileCount,
- nextRowId,
- assignment.nextRowId);
+ throw new IllegalStateException("Unreachable retry state while
reassigning row IDs.");
}
- private AssignmentPlan planAssignment(
- List<ManifestFileMeta> manifestMetas, ManifestFile manifestFile,
long firstRowId) {
+ private Optional<AssignmentPlan> planAssignment(
+ List<ManifestFileMeta> manifestMetas, ManifestFile manifestFile) {
List<List<ManifestFileMeta>> manifestGroups =
manifestGroupsByPartition(manifestMetas);
- Map<String, List<ManifestFileMeta>> rewrittenManifestMetas = new
HashMap<>();
- Map<BinaryRow, RowRangeMappingIndex> rowIdMappings = new
LinkedHashMap<>();
- long nextRowId = firstRowId;
- long logicalRowCount = 0;
- long reassignedFileCount = 0;
- boolean hasCurrentFiles = false;
+ Map<BinaryRow, List<ManifestEntry>> entriesToReassignByPartition = new
LinkedHashMap<>();
for (List<ManifestFileMeta> manifestGroup : manifestGroups) {
if (skipManifestGroupByPartitionFilter(manifestGroup)) {
continue;
}
- CurrentManifest currentManifest = currentManifest(manifestGroup,
manifestFile);
- List<ManifestEntry> currentEntries = currentManifest.entries();
+ List<ManifestEntry> currentEntries = currentEntries(manifestGroup,
manifestFile);
if (currentEntries.isEmpty()) {
continue;
}
- hasCurrentFiles = true;
Map<BinaryRow, List<ManifestEntry>> entriesByPartition =
entriesByPartition(currentEntries);
@@ -204,42 +208,26 @@ public class DataEvolutionRowIdReassigner {
continue;
}
- Assignment groupAssignment =
- assign(entriesByPartition, partitionsToReassign,
nextRowId);
- nextRowId = groupAssignment.nextRowId;
- logicalRowCount += groupAssignment.logicalRowCount;
- reassignedFileCount += groupAssignment.reassignedFileCount;
- for (Map.Entry<BinaryRow, RowRangeMappingIndex> mapping :
- groupAssignment.rowIdMappings.entrySet()) {
- RowRangeMappingIndex previous =
- rowIdMappings.put(mapping.getKey(),
mapping.getValue());
- checkState(
- previous == null,
- "Partition %s appears in multiple manifest groups.",
-
table.store().pathFactory().getPartitionString(mapping.getKey()));
- }
-
- Map<String, List<ManifestFileMeta>> groupRewrittenManifestMetas =
- writeManifestReplacements(
- currentManifest, groupAssignment,
partitionsToReassign, manifestFile);
- for (Map.Entry<String, List<ManifestFileMeta>> rewritten :
- groupRewrittenManifestMetas.entrySet()) {
- List<ManifestFileMeta> previous =
- rewrittenManifestMetas.put(rewritten.getKey(),
rewritten.getValue());
- checkState(
- previous == null,
- "Manifest file %s appears in multiple manifest
groups.",
- rewritten.getKey());
+ for (ManifestEntry entry : currentEntries) {
+ if (partitionsToReassign.contains(entry.partition())) {
+ entriesToReassignByPartition
+ .computeIfAbsent(entry.partition(), ignored -> new
ArrayList<>())
+ .add(entry);
+ }
}
}
- return new AssignmentPlan(
- rewrittenManifestMetas,
- rowIdMappings,
- nextRowId,
- logicalRowCount,
- reassignedFileCount,
- hasCurrentFiles);
+ if (entriesToReassignByPartition.isEmpty()) {
+ return Optional.empty();
+ }
+
+ RelativeRowIdMappings relativeRowIdMappings =
+ createRelativeRowIdMappings(entriesToReassignByPartition);
+ return Optional.of(
+ new AssignmentPlan(
+ findManifestMetasToRewrite(
+ manifestMetas, relativeRowIdMappings,
manifestFile),
+ relativeRowIdMappings));
}
private List<List<ManifestFileMeta>> manifestGroupsByPartition(
@@ -272,6 +260,7 @@ public class DataEvolutionRowIdReassigner {
manifestMeta,
manifestMeta.partitionStats().minValues(),
manifestMeta.partitionStats().maxValues(),
+ containsNullPartition(manifestMeta,
partitionFieldCount),
i));
}
Collections.sort(
@@ -304,6 +293,32 @@ public class DataEvolutionRowIdReassigner {
}
groupedManifestRanges.add(currentGroup);
+ // Partition min/max excludes nulls, so null-bearing ranges need an
extra shared group.
+ List<PartitionManifestRange> nullPartitionGroup = new ArrayList<>();
+ int nullPartitionGroupIndex = -1;
+ for (int i = 0; i < groupedManifestRanges.size(); ) {
+ List<PartitionManifestRange> group = groupedManifestRanges.get(i);
+ boolean containsNullPartition = false;
+ for (PartitionManifestRange range : group) {
+ if (range.containsNullPartition) {
+ containsNullPartition = true;
+ break;
+ }
+ }
+ if (containsNullPartition) {
+ if (nullPartitionGroupIndex < 0) {
+ nullPartitionGroupIndex = i;
+ }
+ nullPartitionGroup.addAll(group);
+ groupedManifestRanges.remove(i);
+ } else {
+ i++;
+ }
+ }
+ if (!nullPartitionGroup.isEmpty()) {
+ groupedManifestRanges.add(nullPartitionGroupIndex,
nullPartitionGroup);
+ }
+
List<List<ManifestFileMeta>> groups = new ArrayList<>();
for (List<PartitionManifestRange> group : groupedManifestRanges) {
Collections.sort(group, Comparator.comparingInt(left ->
left.originalIndex));
@@ -347,27 +362,35 @@ public class DataEvolutionRowIdReassigner {
&& partitionStats.nullCounts().size() == partitionFieldCount;
}
- private CurrentManifest currentManifest(
+ private boolean containsNullPartition(ManifestFileMeta manifestMeta, int
partitionFieldCount) {
+ for (int i = 0; i < partitionFieldCount; i++) {
+ if (manifestMeta.partitionStats().nullCounts().getLong(i) != 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private List<ManifestEntry> currentEntries(
List<ManifestFileMeta> manifestMetas, ManifestFile manifestFile) {
Set<FileEntry.Identifier> deletedIdentifiers =
deletedIdentifiers(manifestFile, manifestMetas);
- List<SourcedManifestEntry> currentEntries = new ArrayList<>();
+ List<ManifestEntry> currentEntries = new ArrayList<>();
for (ManifestFileMeta manifestMeta : manifestMetas) {
if (manifestMeta.numAddedFiles() <= 0) {
continue;
}
- List<ManifestEntry> entries =
- manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
+ List<ManifestEntry> entries =
readPlanningManifestEntries(manifestFile, manifestMeta);
for (ManifestEntry entry : entries) {
if (entry.kind() == FileKind.ADD
&& partitionIncluded(entry.partition())
&& !deletedIdentifiers.contains(entry.identifier())) {
- currentEntries.add(new SourcedManifestEntry(manifestMeta,
entry));
+ currentEntries.add(entry);
}
}
}
- return new CurrentManifest(manifestMetas, currentEntries);
+ return currentEntries;
}
private Set<FileEntry.Identifier> deletedIdentifiers(
@@ -377,8 +400,7 @@ public class DataEvolutionRowIdReassigner {
if (manifestMeta.numDeletedFiles() <= 0) {
continue;
}
- List<ManifestEntry> entries =
- manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
+ List<ManifestEntry> entries =
readPlanningManifestEntries(manifestFile, manifestMeta);
for (ManifestEntry entry : entries) {
if (entry.kind() == FileKind.DELETE &&
partitionIncluded(entry.partition())) {
deletedIdentifiers.add(entry.identifier());
@@ -396,6 +418,184 @@ public class DataEvolutionRowIdReassigner {
return partitionPredicate != null;
}
+ private CommitAssignmentResult commitAssignment(
+ Assignment assignment,
+ ManifestFile manifestFile,
+ ManifestList manifestList,
+ String commitUser) {
+ RewrittenDataManifests rewrittenDataManifests =
+ writeManifestReplacements(assignment, manifestFile);
+ Pair<String, Long> baseManifestList =
+ writeBaseManifestList(
+ manifestList.readDataManifests(assignment.snapshot),
+ rewrittenDataManifests.manifestMetas,
+ manifestList);
+ Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
+ RewrittenIndexManifest rewrittenIndexManifest =
rewriteIndexManifest(assignment);
+
+ boolean success;
+ try (FileStoreCommitImpl commit =
+ (FileStoreCommitImpl) table.store().newCommit(commitUser,
table)) {
+ beforeCommit.run();
+ success =
+ commit.replaceManifestList(
+ assignment.snapshot,
+ assignment.snapshot.totalRecordCount(),
+ baseManifestList,
+ deltaManifestList,
+ rewrittenIndexManifest.indexManifest,
+ assignment.nextRowId);
+ }
+ return new CommitAssignmentResult(
+ success, rewrittenDataManifests.fileCount,
rewrittenIndexManifest.indexFileCount);
+ }
+
+ private AssignmentPlan advanceAssignmentPlan(
+ AssignmentPlan assignmentPlan,
+ Snapshot previous,
+ Snapshot latest,
+ ManifestFile manifestFile,
+ ManifestList manifestList) {
+ checkState(
+ latest.id() > previous.id(),
+ "Cannot advance row-id assignment from snapshot %s to %s.",
+ previous.id(),
+ latest.id());
+
+ Map<String, ManifestFileMeta> manifestMetasToRewrite = new
LinkedHashMap<>();
+ for (ManifestFileMeta manifestMeta :
assignmentPlan.manifestMetasToRewrite) {
+ manifestMetasToRewrite.put(manifestMeta.fileName(), manifestMeta);
+ }
+ for (long id = previous.id() + 1; id <= latest.id(); id++) {
+ Snapshot snapshot;
+ try {
+ snapshot = table.snapshotManager().tryGetSnapshot(id);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Abort row-id reassignment because snapshot %s
cannot be read.",
+ id),
+ e);
+ }
+
+ if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT
+ || snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE)
{
+ throw new RuntimeException(
+ String.format(
+ "Abort row-id reassignment because %s snapshot
%s was committed after snapshot %s.",
+ snapshot.commitKind(), snapshot.id(),
previous.id()));
+ }
+ if (snapshot.commitKind() == Snapshot.CommitKind.ANALYZE) {
+ continue;
+ }
+ checkState(
+ snapshot.commitKind() == Snapshot.CommitKind.APPEND,
+ "Unsupported snapshot kind %s while advancing row-id
assignment.",
+ snapshot.commitKind());
+
+ for (ManifestFileMeta manifestMeta :
manifestList.readDeltaManifests(snapshot)) {
+ boolean needsReassign = false;
+ for (ManifestEntry entry :
+ readPlanningManifestEntries(manifestFile,
manifestMeta)) {
+ checkState(
+ entry.kind() == FileKind.ADD,
+ "APPEND snapshot %s contains non-ADD manifest
entry %s.",
+ snapshot.id(),
+ entry);
+ if (appendedEntryNeedsReassign(assignmentPlan, entry)) {
+ needsReassign = true;
+ }
+ }
+ if (needsReassign) {
+ manifestMetasToRewrite.put(manifestMeta.fileName(),
manifestMeta);
+ }
+ }
+ }
+
+ List<ManifestFileMeta> latestManifestMetas =
manifestList.readDataManifests(latest);
+ Set<String> latestManifestFiles = new HashSet<>();
+ for (ManifestFileMeta manifestMeta : latestManifestMetas) {
+ latestManifestFiles.add(manifestMeta.fileName());
+ }
+ for (String plannedManifestFile : manifestMetasToRewrite.keySet()) {
+ checkState(
+ latestManifestFiles.contains(plannedManifestFile),
+ "Cannot advance row-id assignment because planned manifest
%s no longer exists after APPEND manifest merge.",
+ plannedManifestFile);
+ }
+ return new AssignmentPlan(
+ new ArrayList<>(manifestMetasToRewrite.values()),
+ assignmentPlan.relativeRowIdMappings);
+ }
+
+ private boolean appendedEntryNeedsReassign(
+ AssignmentPlan assignmentPlan, ManifestEntry appendedEntry) {
+ RowRangeMappingIndex mapping =
+
assignmentPlan.relativeRowIdMappings.mappings.get(appendedEntry.partition());
+ if (mapping == null) {
+ return false;
+ }
+
+ Range appendedRange = appendedEntry.file().nonNullRowIdRange();
+ if (mapping.map(appendedRange).isPresent()) {
+ return true;
+ }
+
+ checkState(
+ !mapping.overlaps(appendedRange),
+ "Cannot advance row-id assignment because appended row-id
range %s partially overlaps planned ranges in partition %s.",
+ appendedRange,
+ appendedEntry.partition());
+ return false;
+ }
+
+ private boolean manifestGroupMayContainPartitions(
+ List<ManifestFileMeta> manifestGroup, PartitionPredicate
effectivePartitionPredicate) {
+ int partitionFieldCount =
table.schema().logicalPartitionType().getFieldCount();
+ for (ManifestFileMeta manifestMeta : manifestGroup) {
+ if (!containsPartitionStats(manifestMeta, partitionFieldCount)) {
+ return true;
+ }
+
+ SimpleStats partitionStats = manifestMeta.partitionStats();
+ if (effectivePartitionPredicate.test(
+ manifestMeta.numAddedFiles() +
manifestMeta.numDeletedFiles(),
+ partitionStats.minValues(),
+ partitionStats.maxValues(),
+ partitionStats.nullCounts())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private List<ManifestFileMeta> findManifestMetasToRewrite(
+ List<ManifestFileMeta> manifestMetas,
+ RelativeRowIdMappings relativeRowIdMappings,
+ ManifestFile manifestFile) {
+ PartitionPredicate plannedPartitionPredicate =
+ PartitionPredicate.fromMultiple(
+ table.schema().logicalPartitionType(),
+ relativeRowIdMappings.mappings.keySet());
+ checkState(plannedPartitionPredicate != null, "Planned partition
predicate is null.");
+ List<ManifestFileMeta> result = new ArrayList<>();
+ for (ManifestFileMeta manifestMeta : manifestMetas) {
+ if (!manifestGroupMayContainPartitions(
+ Collections.singletonList(manifestMeta),
plannedPartitionPredicate)) {
+ continue;
+ }
+ for (ManifestEntry entry :
readPlanningManifestEntries(manifestFile, manifestMeta)) {
+ RowRangeMappingIndex mapping =
+ relativeRowIdMappings.mappings.get(entry.partition());
+ if (mapping != null &&
mapping.map(entry.file().nonNullRowIdRange()).isPresent()) {
+ result.add(manifestMeta);
+ break;
+ }
+ }
+ }
+ return result;
+ }
+
private Pair<String, Long> writeBaseManifestList(
List<ManifestFileMeta> manifestMetas,
Map<String, List<ManifestFileMeta>> rewrittenManifestMetas,
@@ -413,77 +613,48 @@ public class DataEvolutionRowIdReassigner {
return manifestList.write(baseManifestMetas);
}
- private Map<String, List<ManifestFileMeta>> writeManifestReplacements(
- CurrentManifest currentManifest,
- Assignment assignment,
- Set<BinaryRow> partitionsToReassign,
- ManifestFile manifestFile) {
- Set<String> manifestsToRewrite =
- manifestsToRewrite(currentManifest.currentEntries,
partitionsToReassign);
- Map<FileEntry.Identifier, ManifestEntry> reassignedEntries =
- entriesByIdentifier(assignment.entries);
-
+ private RewrittenDataManifests writeManifestReplacements(
+ Assignment assignment, ManifestFile manifestFile) {
Map<String, List<ManifestFileMeta>> rewrittenManifestMetas = new
HashMap<>();
- for (ManifestFileMeta manifestMeta : currentManifest.manifestMetas) {
- if (!manifestsToRewrite.contains(manifestMeta.fileName())) {
- continue;
- }
-
- List<ManifestEntry> rewrittenEntries = new ArrayList<>();
+ long fileCount = 0L;
+ for (ManifestFileMeta manifestMeta :
assignment.manifestMetasToRewrite) {
List<ManifestEntry> entries =
manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
- for (ManifestEntry entry : entries) {
- if (entry.kind() == FileKind.ADD) {
- ManifestEntry reassignedEntry =
reassignedEntries.get(entry.identifier());
- if (reassignedEntry != null) {
- entry = reassignedEntry;
+ long reassignedAddFileCount = 0L;
+ boolean hasRewrittenEntry = false;
+ for (int i = 0; i < entries.size(); i++) {
+ ManifestEntry entry = entries.get(i);
+ RowRangeMappingIndex mapping =
assignment.rowIdMappings.get(entry.partition());
+ if (mapping == null) {
+ continue;
+ }
+ Optional<Range> reassignedRange =
mapping.map(entry.file().nonNullRowIdRange());
+ if (reassignedRange.isPresent()) {
+ validatePlanningEntry(entry);
+ entries.set(i,
entry.assignFirstRowId(reassignedRange.get().from));
+ hasRewrittenEntry = true;
+ if (entry.kind() == FileKind.ADD) {
+ reassignedAddFileCount++;
}
}
- rewrittenEntries.add(entry);
- }
- rewrittenManifestMetas.put(
- manifestMeta.fileName(),
manifestFile.write(rewrittenEntries));
- }
- return rewrittenManifestMetas;
- }
-
- private Set<String> manifestsToRewrite(
- List<SourcedManifestEntry> currentEntries, Set<BinaryRow>
partitionsToReassign) {
- Set<String> manifestsToRewrite = new HashSet<>();
- for (SourcedManifestEntry currentEntry : currentEntries) {
- if (partitionsToReassign.contains(currentEntry.entry.partition()))
{
- manifestsToRewrite.add(currentEntry.manifest.fileName());
}
+ checkState(
+ hasRewrittenEntry,
+ "Cannot find entries to reassign in planned manifest %s.",
+ manifestMeta.fileName());
+ rewrittenManifestMetas.put(manifestMeta.fileName(),
manifestFile.write(entries));
+ fileCount += reassignedAddFileCount;
}
- return manifestsToRewrite;
- }
-
- private Map<FileEntry.Identifier, ManifestEntry> entriesByIdentifier(
- List<ManifestEntry> entries) {
- Map<FileEntry.Identifier, ManifestEntry> result = new HashMap<>();
- for (ManifestEntry entry : entries) {
- ManifestEntry previous = result.put(entry.identifier(), entry);
- checkState(previous == null, "Duplicate current manifest entry for
file %s.", entry);
- }
- return result;
+ return new RewrittenDataManifests(rewrittenManifestMetas, fileCount);
}
private Map<BinaryRow, List<ManifestEntry>>
entriesByPartition(List<ManifestEntry> entries) {
- List<ManifestEntry> sorted = new ArrayList<>(entries);
- Collections.sort(sorted, entryComparator());
+ Comparator<ManifestEntry> comparator = entryComparator();
+ Collections.sort(entries, comparator);
Map<BinaryRow, List<ManifestEntry>> entriesByPartition = new
LinkedHashMap<>();
- for (ManifestEntry entry : sorted) {
- List<String> writeCols = entry.file().writeCols();
- checkState(
- writeCols == null ||
!writeCols.contains(SpecialFields.ROW_ID.name()),
- "Cannot reassign row IDs for file '%s' because it
physically stores the row-id field.",
- entry.file().fileName());
- checkState(
- entry.file().firstRowId() != null,
- "File '%s' in table '%s' does not have first row id.",
- entry.file().fileName(),
- table.name());
+ for (ManifestEntry entry : entries) {
+ validatePlanningEntry(entry);
entriesByPartition
.computeIfAbsent(entry.partition(), k -> new ArrayList<>())
.add(entry);
@@ -491,6 +662,19 @@ public class DataEvolutionRowIdReassigner {
return entriesByPartition;
}
+ private void validatePlanningEntry(ManifestEntry entry) {
+ List<String> writeCols = entry.file().writeCols();
+ checkState(
+ writeCols == null ||
!writeCols.contains(SpecialFields.ROW_ID.name()),
+ "Cannot reassign row IDs for file '%s' because it physically
stores the row-id field.",
+ entry.file().fileName());
+ checkState(
+ entry.file().firstRowId() != null,
+ "File '%s' in table '%s' does not have first row id.",
+ entry.file().fileName(),
+ table.name());
+ }
+
private Set<BinaryRow> partitionsToReassign(
Map<BinaryRow, List<ManifestEntry>> entriesByPartition) {
Set<BinaryRow> partitionsToReassign = new HashSet<>();
@@ -503,6 +687,9 @@ public class DataEvolutionRowIdReassigner {
}
private boolean partitionRowIdsAreContiguous(List<ManifestEntry> entries) {
+ for (ManifestEntry entry : entries) {
+ validatePlanningEntry(entry);
+ }
List<Range> logicalRanges = logicalRanges(entries);
if (logicalRanges.size() <= 1) {
return true;
@@ -536,59 +723,42 @@ public class DataEvolutionRowIdReassigner {
return logicalRanges;
}
- private Assignment assign(
- Map<BinaryRow, List<ManifestEntry>> entriesByPartition,
- Set<BinaryRow> partitionsToReassign,
- long firstRowId) {
- List<ManifestEntry> entries = new ArrayList<>();
- Map<BinaryRow, RowRangeMappingIndex> rowIdMappings = new
LinkedHashMap<>();
- long nextRowId = firstRowId;
- long logicalRowCount = 0;
- long reassignedFileCount = 0;
- for (Map.Entry<BinaryRow, List<ManifestEntry>> entry :
entriesByPartition.entrySet()) {
- if (partitionsToReassign.contains(entry.getKey())) {
- long partitionFirstRowId = nextRowId;
- PartitionAssignment partitionAssignment =
- assignPartition(entry.getValue(), nextRowId);
- entries.addAll(partitionAssignment.entries);
- rowIdMappings.put(entry.getKey(),
partitionAssignment.rowIdMappings);
- nextRowId = partitionAssignment.nextRowId;
- logicalRowCount += nextRowId - partitionFirstRowId;
- reassignedFileCount += partitionAssignment.entries.size();
+ private RelativeRowIdMappings createRelativeRowIdMappings(
+ Map<BinaryRow, List<ManifestEntry>> entriesByPartition) {
+ Map<BinaryRow, List<Range>> logicalRangesByPartition = new
LinkedHashMap<>();
+ for (Map.Entry<BinaryRow, List<ManifestEntry>> partition :
entriesByPartition.entrySet()) {
+ for (ManifestEntry entry : partition.getValue()) {
+ validatePlanningEntry(entry);
}
+ logicalRangesByPartition.put(partition.getKey(),
logicalRanges(partition.getValue()));
}
-
- return new Assignment(
- entries, rowIdMappings, nextRowId, logicalRowCount,
reassignedFileCount);
+ return createRelativeRowIdMappingsFromRanges(logicalRangesByPartition);
}
- private PartitionAssignment assignPartition(List<ManifestEntry> entries,
long firstRowId) {
- RangeHelper<ManifestEntry> rangeHelper =
- new RangeHelper<>(entry -> entry.file().nonNullRowIdRange());
- List<List<ManifestEntry>> groups =
rangeHelper.mergeOverlappingRanges(entries);
-
- List<ManifestEntry> reassigned = new ArrayList<>(entries.size());
- List<RowRangeMappingIndex.Mapping> mappings = new ArrayList<>();
- long nextRowId = firstRowId;
-
- for (List<ManifestEntry> group : groups) {
- Collections.sort(group, entryComparatorWithoutPartition());
- Range oldLogicalRange = oldLogicalRange(group);
- mappings.add(
- RowRangeMappingIndex.mapping(
- oldLogicalRange.from, oldLogicalRange.to,
nextRowId));
-
- for (ManifestEntry entry : group) {
- long oldFirstRowId = entry.file().nonNullFirstRowId();
- long newFirstRowId = nextRowId + oldFirstRowId -
oldLogicalRange.from;
- reassigned.add(entry.assignFirstRowId(newFirstRowId));
+ private RelativeRowIdMappings createRelativeRowIdMappingsFromRanges(
+ Map<BinaryRow, List<Range>> logicalRangesByPartition) {
+ List<BinaryRow> partitions = new
ArrayList<>(logicalRangesByPartition.keySet());
+ RecordComparator partitionComparator = partitionComparator();
+ Collections.sort(partitions, partitionComparator);
+
+ Map<BinaryRow, RowRangeMappingIndex> result = new LinkedHashMap<>();
+ long nextOffset = 0L;
+ for (BinaryRow partition : partitions) {
+ List<Range> ranges = new
ArrayList<>(logicalRangesByPartition.get(partition));
+ Collections.sort(
+ ranges,
+ (left, right) -> {
+ int compare = Long.compare(left.from, right.from);
+ return compare == 0 ? Long.compare(left.to, right.to)
: compare;
+ });
+ List<RowRangeMappingIndex.Mapping> mappings = new
ArrayList<>(ranges.size());
+ for (Range range : ranges) {
+ mappings.add(RowRangeMappingIndex.mapping(range.from,
range.to, nextOffset));
+ nextOffset = Math.addExact(nextOffset, range.count());
}
-
- nextRowId += oldLogicalRange.count();
+ result.put(partition, RowRangeMappingIndex.create(mappings));
}
-
- return new PartitionAssignment(
- reassigned, RowRangeMappingIndex.create(mappings), nextRowId);
+ return new RelativeRowIdMappings(result, nextOffset);
}
private Range oldLogicalRange(List<ManifestEntry> group) {
@@ -637,14 +807,14 @@ public class DataEvolutionRowIdReassigner {
return new Range(min, max);
}
- private RewrittenIndexManifest rewriteIndexManifest(
- Snapshot latest, AssignmentPlan assignment) {
- if (latest.indexManifest() == null) {
+ private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment)
{
+ if (assignment.snapshot.indexManifest() == null) {
return new RewrittenIndexManifest(null, 0);
}
IndexManifestFile indexManifestFile =
table.store().indexManifestFileFactory().create();
- List<IndexManifestEntry> indexEntries =
indexManifestFile.read(latest.indexManifest());
+ List<IndexManifestEntry> indexEntries =
+ indexManifestFile.read(assignment.snapshot.indexManifest());
if (indexEntries.isEmpty()) {
return new RewrittenIndexManifest(null, 0);
}
@@ -655,7 +825,7 @@ public class DataEvolutionRowIdReassigner {
checkState(
entry.kind() == FileKind.ADD,
"Index manifest '%s' contains non-current entry %s.",
- latest.indexManifest(),
+ assignment.snapshot.indexManifest(),
entry);
IndexFileMeta indexFile = entry.indexFile();
@@ -668,6 +838,10 @@ public class DataEvolutionRowIdReassigner {
Optional<Range> newRange =
mappingIndex.map(globalIndex.rowRange());
if (!newRange.isPresent()) {
+ if (!mappingIndex.overlaps(globalIndex.rowRange())) {
+ rewritten.add(entry);
+ continue;
+ }
LOG.warn(
"Drop global index file '{}' from table {} during
row-id reassignment because its row range {} cannot be rewritten safely.",
indexFile.fileName(),
@@ -702,6 +876,18 @@ public class DataEvolutionRowIdReassigner {
indexManifestFile.writeWithoutRolling(rewritten),
globalIndexFileCount);
}
+ private List<ManifestEntry> readPlanningManifestEntries(
+ ManifestFile manifestFile, ManifestFileMeta manifestMeta) {
+ return manifestFile.read(
+ manifestMeta.fileName(),
+ manifestMeta.fileSize(),
+ partitionPredicate,
+ null,
+ Filter.alwaysTrue(),
+ entry -> partitionPredicate == null ||
partitionPredicate.test(entry.partition()),
+ ManifestEntry::copyWithoutStats);
+ }
+
private Comparator<ManifestEntry> entryComparator() {
RecordComparator partitionComparator = partitionComparator();
Comparator<ManifestEntry> withoutPartition =
entryComparatorWithoutPartition();
@@ -842,77 +1028,95 @@ public class DataEvolutionRowIdReassigner {
}
}
- private static class Assignment {
- private final List<ManifestEntry> entries;
- private final Map<BinaryRow, RowRangeMappingIndex> rowIdMappings;
- private final long nextRowId;
- private final long logicalRowCount;
- private final long reassignedFileCount;
+ private static class RelativeRowIdMappings {
+ private final Map<BinaryRow, RowRangeMappingIndex> mappings;
+ private final long totalOffset;
- private Assignment(
- List<ManifestEntry> entries,
- Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
- long nextRowId,
- long logicalRowCount,
- long reassignedFileCount) {
- this.entries = entries;
- this.rowIdMappings = rowIdMappings;
- this.nextRowId = nextRowId;
- this.logicalRowCount = logicalRowCount;
- this.reassignedFileCount = reassignedFileCount;
+ private RelativeRowIdMappings(
+ Map<BinaryRow, RowRangeMappingIndex> mappings, long
totalOffset) {
+ this.mappings = Collections.unmodifiableMap(new
LinkedHashMap<>(mappings));
+ this.totalOffset = totalOffset;
}
}
private static class AssignmentPlan {
- private final Map<String, List<ManifestFileMeta>>
rewrittenManifestMetas;
+ private final List<ManifestFileMeta> manifestMetasToRewrite;
+ private final RelativeRowIdMappings relativeRowIdMappings;
+
+ private AssignmentPlan(
+ List<ManifestFileMeta> manifestMetasToRewrite,
+ RelativeRowIdMappings relativeRowIdMappings) {
+ this.manifestMetasToRewrite = new
ArrayList<>(manifestMetasToRewrite);
+ this.relativeRowIdMappings = relativeRowIdMappings;
+ }
+
+ private Assignment createAssignment(Snapshot snapshot) {
+ Long firstAssignedRowId = snapshot.nextRowId();
+ checkState(
+ firstAssignedRowId != null,
+ "Next row id cannot be null for snapshot %s.",
+ snapshot.id());
+ Map<BinaryRow, RowRangeMappingIndex> absoluteRowIdMappings = new
LinkedHashMap<>();
+ for (Map.Entry<BinaryRow, RowRangeMappingIndex> mapping :
+ relativeRowIdMappings.mappings.entrySet()) {
+ absoluteRowIdMappings.put(
+ mapping.getKey(),
mapping.getValue().shiftNewStarts(firstAssignedRowId));
+ }
+ return new Assignment(
+ snapshot,
+ manifestMetasToRewrite,
+ absoluteRowIdMappings,
+ firstAssignedRowId,
+ Math.addExact(firstAssignedRowId,
relativeRowIdMappings.totalOffset));
+ }
+ }
+
+ private static class Assignment {
+ private final Snapshot snapshot;
+ private final List<ManifestFileMeta> manifestMetasToRewrite;
private final Map<BinaryRow, RowRangeMappingIndex> rowIdMappings;
+ private final long firstAssignedRowId;
private final long nextRowId;
- private final long logicalRowCount;
- private final long reassignedFileCount;
- private final boolean hasCurrentFiles;
- private AssignmentPlan(
- Map<String, List<ManifestFileMeta>> rewrittenManifestMetas,
+ private Assignment(
+ Snapshot snapshot,
+ List<ManifestFileMeta> manifestMetasToRewrite,
Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
- long nextRowId,
- long logicalRowCount,
- long reassignedFileCount,
- boolean hasCurrentFiles) {
- this.rewrittenManifestMetas = rewrittenManifestMetas;
- this.rowIdMappings = rowIdMappings;
+ long firstAssignedRowId,
+ long nextRowId) {
+ this.snapshot = snapshot;
+ this.manifestMetasToRewrite =
+ Collections.unmodifiableList(new
ArrayList<>(manifestMetasToRewrite));
+ this.rowIdMappings = Collections.unmodifiableMap(new
LinkedHashMap<>(rowIdMappings));
+ this.firstAssignedRowId = firstAssignedRowId;
this.nextRowId = nextRowId;
- this.logicalRowCount = logicalRowCount;
- this.reassignedFileCount = reassignedFileCount;
- this.hasCurrentFiles = hasCurrentFiles;
+ }
+
+ private long logicalRowCount() {
+ return nextRowId - firstAssignedRowId;
}
}
- private static class CurrentManifest {
- private final List<ManifestFileMeta> manifestMetas;
- private final List<SourcedManifestEntry> currentEntries;
+ private static class RewrittenDataManifests {
+ private final Map<String, List<ManifestFileMeta>> manifestMetas;
+ private final long fileCount;
- private CurrentManifest(
- List<ManifestFileMeta> manifestMetas,
List<SourcedManifestEntry> currentEntries) {
+ private RewrittenDataManifests(
+ Map<String, List<ManifestFileMeta>> manifestMetas, long
fileCount) {
this.manifestMetas = manifestMetas;
- this.currentEntries = currentEntries;
- }
-
- private List<ManifestEntry> entries() {
- List<ManifestEntry> entries = new
ArrayList<>(currentEntries.size());
- for (SourcedManifestEntry currentEntry : currentEntries) {
- entries.add(currentEntry.entry);
- }
- return entries;
+ this.fileCount = fileCount;
}
}
- private static class SourcedManifestEntry {
- private final ManifestFileMeta manifest;
- private final ManifestEntry entry;
+ private static class CommitAssignmentResult {
+ private final boolean success;
+ private final long fileCount;
+ private final long indexFileCount;
- private SourcedManifestEntry(ManifestFileMeta manifest, ManifestEntry
entry) {
- this.manifest = manifest;
- this.entry = entry;
+ private CommitAssignmentResult(boolean success, long fileCount, long
indexFileCount) {
+ this.success = success;
+ this.fileCount = fileCount;
+ this.indexFileCount = indexFileCount;
}
}
@@ -920,30 +1124,20 @@ public class DataEvolutionRowIdReassigner {
private final ManifestFileMeta manifest;
private final BinaryRow minPartition;
private final BinaryRow maxPartition;
+ private final boolean containsNullPartition;
private final int originalIndex;
private PartitionManifestRange(
ManifestFileMeta manifest,
BinaryRow minPartition,
BinaryRow maxPartition,
+ boolean containsNullPartition,
int originalIndex) {
this.manifest = manifest;
this.minPartition = minPartition;
this.maxPartition = maxPartition;
+ this.containsNullPartition = containsNullPartition;
this.originalIndex = originalIndex;
}
}
-
- private static class PartitionAssignment {
- private final List<ManifestEntry> entries;
- private final RowRangeMappingIndex rowIdMappings;
- private final long nextRowId;
-
- private PartitionAssignment(
- List<ManifestEntry> entries, RowRangeMappingIndex
rowIdMappings, long nextRowId) {
- this.entries = entries;
- this.rowIdMappings = rowIdMappings;
- this.nextRowId = nextRowId;
- }
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
index a24515c905..7977c9295c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
@@ -70,6 +70,18 @@ final class RowRangeMappingIndex {
return new Mapping(oldStart, oldEnd, newStart);
}
+ RowRangeMappingIndex shiftNewStarts(long offset) {
+ List<Mapping> shifted = new ArrayList<>(mappings.size());
+ for (Mapping mapping : mappings) {
+ shifted.add(
+ mapping(
+ mapping.oldStart,
+ mapping.oldEnd,
+ Math.addExact(mapping.newStart, offset)));
+ }
+ return create(shifted);
+ }
+
Optional<Range> map(Range oldRange) {
checkArgument(oldRange != null, "Old row range cannot be null.");
checkArgument(oldRange.from <= oldRange.to, "Invalid old row range
%s.", oldRange);
@@ -108,6 +120,14 @@ final class RowRangeMappingIndex {
return Optional.of(new Range(newFrom, newTo));
}
+ boolean overlaps(Range oldRange) {
+ checkArgument(oldRange != null, "Old row range cannot be null.");
+ checkArgument(oldRange.from <= oldRange.to, "Invalid old row range
%s.", oldRange);
+
+ int index = lowerBound(oldEnds, oldRange.from);
+ return index < mappings.size() && mappings.get(index).oldStart <=
oldRange.to;
+ }
+
private static int lowerBound(long[] sorted, long target) {
int left = 0;
int right = sorted.length;
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 ffdc12539a..77cb9802f3 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
@@ -26,11 +26,15 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
@@ -46,25 +50,28 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SnapshotManager;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -72,21 +79,26 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link DataEvolutionRowIdReassigner}. */
public class DataEvolutionRowIdReassignerTest extends TableTestBase {
private static final int LARGE_ENTRY_COUNT = 10_000;
private static final int LARGE_MANIFEST_FILE_COUNT = 20;
- private static final int ONE_MILLION_ENTRY_COUNT = 1_000_000;
- private static final int ONE_MILLION_MANIFEST_FILE_COUNT = 200;
private static final long LARGE_ENTRY_ROW_COUNT = 2L;
private static final long RANDOM_PARTITION_SEED = 20260519L;
private static final String LARGE_FILE_PREFIX = "large-partition-";
private static final String LARGE_FILE_SUFFIX = ".parquet";
private static final int LARGE_PARTITION_ID_WIDTH = 7;
+ private static final String[] DATA_INTEGRITY_PARTITIONS = {"a", "b", "c",
"d"};
+ private static final int DATA_INTEGRITY_FILES_PER_PARTITION = 6;
+ private static final long SCRAMBLED_ROW_ID_GAP = 100_000_000L;
+ private static final long SCRAMBLED_ROW_ID_SEED = 20260710L;
@Override
protected Schema schemaDefault() {
@@ -136,6 +148,493 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(10L);
}
+ @Test
+ public void testReassignNullPartitionAcrossManifestGroups() throws
Exception {
+ createTableDefault();
+ FileStoreTable table =
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.ROW_TRACKING_PARTITION_GROUP_ON_COMMIT.key(),
+ "false"));
+ writeRowsInPartitions(table, null, 0, "a", 1);
+ writeOneRow(table, "z", 2);
+ writeOneRow(table, null, 3);
+
+ BinaryRow nullPartition =
+ new
InternalRowSerializer(table.schema().logicalPartitionType())
+ .toBinaryRow(GenericRow.of((Object) null));
+ String nullPartitionPath =
table.store().pathFactory().getPartitionString(nullPartition);
+ Map<String, List<Long>> beforeRowIds = rowIdsByPartition(table);
+ assertThat(beforeRowIds)
+ .containsEntry(nullPartitionPath, Arrays.asList(1L, 3L))
+ .containsEntry("pt=a/", Collections.singletonList(0L))
+ .containsEntry("pt=z/", Collections.singletonList(2L));
+
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(table)
+ .reassign("test-reassign-null-partition-row-id");
+
+ assertThat(result.firstAssignedRowId).isEqualTo(4L);
+ assertThat(result.nextRowId).isEqualTo(6L);
+ assertThat(result.fileCount).isEqualTo(2L);
+ assertThat(result.rowCount).isEqualTo(2L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry(nullPartitionPath, Arrays.asList(4L, 5L))
+ .containsEntry("pt=a/", Collections.singletonList(0L))
+ .containsEntry("pt=z/", Collections.singletonList(2L));
+ }
+
+ @Test
+ public void testReassignRetriesWithLatestNextRowIdAfterConcurrentAppend()
throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+ Map<String, SimpleStats> valueStatsBefore = valueStatsByFile(table);
+ assertThat(before.nextRowId()).isEqualTo(5L);
+ assertThat(valueStatsBefore.values())
+ .allSatisfy(stats ->
assertThat(stats).isNotEqualTo(SimpleStats.EMPTY_STATS));
+
+ AtomicBoolean appended = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "c", 100);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-conflict-retry");
+
+ assertThat(appended).isTrue();
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(11L);
+ assertThat(result.fileCount).isEqualTo(5L);
+ assertThat(result.rowCount).isEqualTo(5L);
+ 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(valueStatsByFile(table)).containsAllEntriesOf(valueStatsBefore);
+
assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(11L);
+ }
+
+ @Test
+ public void testReassignLeavesDisjointConcurrentAppendOutOfPlan() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean appended = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "a", 100);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-append-planned-partition");
+
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(11L);
+ assertThat(result.fileCount).isEqualTo(5L);
+ assertThat(result.rowCount).isEqualTo(5L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(5L, 6L, 7L, 8L))
+ .containsEntry("pt=b/", Arrays.asList(9L, 10L));
+ }
+
+ @Test
+ public void testReassignAbortsForPartiallyOverlappingConcurrentAppend()
throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean appended = new AtomicBoolean();
+ assertThatThrownBy(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if
(appended.compareAndSet(false, true)) {
+ try {
+
writePartialPayload(
+ table,
+ "a",
+ 4L,
+
"overlap-4",
+
"overlap-5");
+ } catch (Exception e) {
+ throw new
RuntimeException(e);
+ }
+ }
+ })
+
.reassign("test-reassign-partially-overlapping-append"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("partially overlaps planned range");
+
+ assertThat(appended).isTrue();
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
+ }
+
+ @Test
+ public void testReassignAbortsWhenConcurrentAppendMergesPlannedManifests()
throws Exception {
+ FileStoreTable originalTable = createTableWithInterleavedPartitions();
+ List<String> plannedManifestFiles =
dataManifestFileNames(originalTable);
+ assertThat(plannedManifestFiles).hasSizeGreaterThan(1);
+
+ FileStoreTable table = withManifestMergeOnNextAppend(originalTable);
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean merged = new AtomicBoolean();
+ assertThatThrownBy(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if
(merged.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table,
"c", 100);
+ Snapshot
appendSnapshot =
+
table.snapshotManager()
+
.latestSnapshot();
+
assertThat(appendSnapshot.commitKind())
+ .isEqualTo(
+
Snapshot.CommitKind
+
.APPEND);
+
assertThat(dataManifestFileNames(table))
+
.doesNotContainAnyElementsOf(
+
plannedManifestFiles);
+ } catch (Exception e) {
+ throw new
RuntimeException(e);
+ }
+ }
+ })
+
.reassign("test-reassign-append-manifest-merge"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("planned manifest")
+ .hasMessageContaining("no longer exists");
+
+ assertThat(merged).isTrue();
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(0L, 2L, 4L))
+ .containsEntry("pt=b/", Arrays.asList(1L, 3L))
+ .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");
+ }
+
+ @Test
+ public void testReassignKeepsHistoricalAddDeleteEntriesConsistent() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ FileEntry.Identifier compactedFile = compactOneFile(table, "pt=a/");
+ long oldFirstRowId = assertAddDeleteEntriesConsistent(table,
compactedFile);
+
+ new DataEvolutionRowIdReassigner(table)
+ .reassign("test-reassign-historical-add-delete-without-retry");
+
+ assertThat(assertAddDeleteEntriesConsistent(table, compactedFile))
+ .isNotEqualTo(oldFirstRowId);
+ assertThat(readTableRows(table))
+ .containsExactly("0|a|v0", "1|b|v1", "2|a|v2", "3|b|v3",
"4|a|v4");
+ }
+
+ @Test
+ public void
testReassignKeepsHistoricalAddDeleteEntriesConsistentAfterRetry() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ FileEntry.Identifier compactedFile = compactOneFile(table, "pt=a/");
+ long oldFirstRowId = assertAddDeleteEntriesConsistent(table,
compactedFile);
+
+ AtomicBoolean appended = new AtomicBoolean();
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "c", 100);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-historical-add-delete");
+
+ assertThat(appended).isTrue();
+ assertThat(assertAddDeleteEntriesConsistent(table, compactedFile))
+ .isNotEqualTo(oldFirstRowId);
+ assertThat(readTableRows(table))
+ .containsExactly("0|a|v0", "100|c|v100", "1|b|v1", "2|a|v2",
"3|b|v3", "4|a|v4");
+ }
+
+ @Test
+ public void testReassignRetriesAfterConcurrentPartialColumnAppend() throws
Exception {
+ createTableDefault();
+ FileStoreTable originalTable = getTableDefault();
+ writeRows(originalTable, "a", 0, 1);
+ writeOneRow(originalTable, "b", 2);
+ writeRows(originalTable, "a", 3, 4);
+ FileStoreTable table = originalTable;
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean appended = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writePartialPayload(
+ table, "a", 0L,
"updated-0", "updated-1");
+ List<ManifestEntry>
sharedRangeEntries =
+
currentEntriesAtRange(table, "pt=a/", 0L, 1L);
+
assertThat(sharedRangeEntries).hasSize(2);
+ assertThat(
+ sharedRangeEntries
+ .get(0)
+ .file()
+
.maxSequenceNumber())
+ .isNotEqualTo(
+ sharedRangeEntries
+ .get(1)
+ .file()
+
.maxSequenceNumber());
+ assertThat(
+
table.snapshotManager()
+
.latestSnapshot()
+
.nextRowId())
+ .isEqualTo(5L);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-partial-column-append");
+
+ assertThat(appended).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(5L);
+ assertThat(result.nextRowId).isEqualTo(9L);
+ assertThat(result.fileCount).isEqualTo(3L);
+ assertThat(result.rowCount).isEqualTo(4L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(5L, 5L, 7L))
+ .containsEntry("pt=b/", Collections.singletonList(2L));
+ assertThat(readTableRows(table))
+ .containsExactly("0|a|updated-0", "1|a|updated-1", "2|b|v2",
"3|a|v3", "4|a|v4");
+ }
+
+ @Test
+ public void testReassignAdvancesAcrossRepeatedAppendConflicts() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicInteger beforeCommits = new AtomicInteger();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ try {
+ int attempt =
beforeCommits.getAndIncrement();
+ if (attempt == 0) {
+ writeOneRow(table, "c", 100);
+ writeOneRow(table, "d", 101);
+ } else if (attempt == 1) {
+ writeOneRow(table, "a", 102);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .reassign("test-reassign-multiple-append-conflicts");
+
+ assertThat(beforeCommits).hasValue(3);
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 3);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 4);
+ assertThat(result.firstAssignedRowId).isEqualTo(8L);
+ assertThat(result.nextRowId).isEqualTo(13L);
+ assertThat(result.fileCount).isEqualTo(5L);
+ assertThat(result.rowCount).isEqualTo(5L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(7L, 8L, 9L, 10L))
+ .containsEntry("pt=b/", Arrays.asList(11L, 12L))
+ .containsEntry("pt=c/", Collections.singletonList(5L))
+ .containsEntry("pt=d/", Collections.singletonList(6L));
+ }
+
+ @Test
+ public void
testReassignPartitionFilterAfterConcurrentAppendOutsideFilter() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean appended = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ partitionPredicate(table, "a"),
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "c", 100);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+
.reassign("test-reassign-filter-after-append-conflict");
+
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(9L);
+ assertThat(result.fileCount).isEqualTo(3L);
+ assertThat(result.rowCount).isEqualTo(3L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(6L, 7L, 8L))
+ .containsEntry("pt=b/", Arrays.asList(1L, 3L))
+ .containsEntry("pt=c/", Collections.singletonList(5L));
+ }
+
+ @Test
+ public void testReassignDoesNotExpandPlanForNewlyNonContiguousPartition()
throws Exception {
+ FileStoreTable table = createTableWithPartiallyOverlappedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean appended = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (appended.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "c", 100);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-append-new-partition");
+
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(8L);
+ assertThat(result.fileCount).isEqualTo(2L);
+ assertThat(result.rowCount).isEqualTo(2L);
+ assertThat(expandedRowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(6L, 7L))
+ .containsEntry("pt=b/", Collections.singletonList(3L))
+ .containsEntry("pt=c/", Arrays.asList(0L, 1L, 5L));
+ }
+
+ @Test
+ public void testReassignAbortsAfterConcurrentOverwrite() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean overwritten = new AtomicBoolean();
+ assertThatThrownBy(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if
(overwritten.compareAndSet(false, true)) {
+ try {
+
overwriteOneRow(table, "c", 100);
+ } catch (Exception e) {
+ throw new
RuntimeException(e);
+ }
+ }
+ })
+
.reassign("test-reassign-overwrite-conflict"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("OVERWRITE snapshot");
+
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id()
+ 1);
+ assertThat(table.snapshotManager().latestSnapshot().commitKind())
+ .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();
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean analyzed = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (analyzed.compareAndSet(false, true)) {
+ try {
+ updateStatistics(table);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-analyze-conflict");
+
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(5L);
+ assertThat(result.nextRowId).isEqualTo(10L);
+
assertThat(table.snapshotManager().latestSnapshot().statistics()).isNotNull();
+ assertThat(table.snapshotManager().latestSnapshot().commitKind())
+ .isEqualTo(Snapshot.CommitKind.OVERWRITE);
+ }
+
@Test
public void testReassignMultiRowFilesByPartition() throws Exception {
createTableDefault();
@@ -301,7 +800,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
new
DataEvolutionRowIdReassigner(table).reassign("test-skip-row-id");
assertThat(result.reassigned).isFalse();
- assertThat(result.skipReason).isEqualTo("partition row IDs are already
contiguous");
+ assertThat(result.skipReason).isEqualTo("no partition requires row-id
reassignment");
assertThat(result.previousSnapshotId).isEqualTo(3L);
assertThat(result.newSnapshotId).isEqualTo(3L);
assertThat(result.firstAssignedRowId).isEqualTo(5L);
@@ -388,6 +887,47 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(readPayloads(table, predicate)).containsExactly("v4");
}
+ @Test
+ public void testReassignKeepsConcurrentDisjointGlobalIndexRange() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ createBTreeIndex(table);
+ Snapshot before = table.snapshotManager().latestSnapshot();
+
+ AtomicBoolean indexed = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+ if (indexed.compareAndSet(false, true)) {
+ try {
+ writeOneRow(table, "a", 100);
+ appendGlobalIndexRange(table,
"pt=a/", new Range(5, 5));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-disjoint-global-index");
+
+ assertThat(indexed).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 2);
+ assertThat(result.firstAssignedRowId).isEqualTo(6L);
+ assertThat(result.nextRowId).isEqualTo(11L);
+ assertThat(result.indexFileCount).isEqualTo(5L);
+ assertThat(globalIndexRanges(table))
+ .containsExactly(
+ new Range(5, 5),
+ new Range(6, 6),
+ new Range(7, 7),
+ new Range(8, 8),
+ new Range(9, 9),
+ new Range(10, 10));
+ assertThat(readTableRows(table))
+ .containsExactly("0|a|v0", "100|a|v100", "1|b|v1", "2|a|v2",
"3|b|v3", "4|a|v4");
+ }
+
@Test
public void
testDropUnsafeGlobalIndexEntryWhenRangeCannotBeRewrittenByMetadataOnly()
throws Exception {
@@ -430,15 +970,141 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
@Test
- public void testReassignManyOutOfOrderPartitionEntries() throws Exception {
- verifyReassignOutOfOrderPartitionEntries(LARGE_ENTRY_COUNT,
LARGE_MANIFEST_FILE_COUNT);
+ public void testFullReassignPreservesDataInTableWithLargeRowIdGaps()
throws Exception {
+ FileStoreTable table = createDataIntegrityTable();
+ long scrambledNextRowId = scrambleCurrentRowIds(table);
+ createBTreeIndex(table);
+
+ Snapshot before = table.snapshotManager().latestSnapshot();
+ List<String> expectedRows = readTableRows(table);
+ Map<String, List<Object>> expectedFileMetadata =
fileMetadataWithoutRowId(table);
+ Map<String, SimpleStats> expectedValueStats = valueStatsByFile(table);
+ Map<String, Long> rowCountsByPartition = rowCountsByPartition(table);
+ Map<String, List<Long>> scrambledRowIds = rowIdsByPartition(table);
+ int expectedFileCount = expectedFileMetadata.size();
+ int expectedIndexFileCount = globalIndexRanges(table).size();
+
+ assertThat(before.nextRowId()).isEqualTo(scrambledNextRowId);
+ assertThat(before.totalRecordCount()).isEqualTo(expectedRows.size());
+ assertThat(expectedFileCount)
+ .isEqualTo(DATA_INTEGRITY_PARTITIONS.length *
DATA_INTEGRITY_FILES_PER_PARTITION);
+ assertThat(expectedIndexFileCount).isGreaterThan(0);
+ assertPartitionsHaveLargeRowIdGaps(scrambledRowIds);
+ assertTableDataQueryable(table, expectedRows);
+
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(table)
+ .reassign("test-full-reassign-data-integrity");
+
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id());
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.firstAssignedRowId).isEqualTo(scrambledNextRowId);
+ assertThat(result.fileCount).isEqualTo(expectedFileCount);
+ assertThat(result.rowCount).isEqualTo(expectedRows.size());
+ assertThat(result.indexFileCount).isEqualTo(expectedIndexFileCount);
+
+ long expectedNextRowId =
+ assertContiguousRowIds(
+ expandedRowIdsByPartition(table),
+ rowCountsByPartition,
+ Arrays.asList(DATA_INTEGRITY_PARTITIONS),
+ scrambledNextRowId);
+ assertThat(result.nextRowId).isEqualTo(expectedNextRowId);
+ assertThat(table.snapshotManager().latestSnapshot().nextRowId())
+ .isEqualTo(expectedNextRowId);
+ assertThat(table.snapshotManager().latestSnapshot().totalRecordCount())
+ .isEqualTo(before.totalRecordCount());
+
assertThat(fileMetadataWithoutRowId(table)).isEqualTo(expectedFileMetadata);
+ assertThat(valueStatsByFile(table)).isEqualTo(expectedValueStats);
+ List<Range> reassignedIndexRanges = globalIndexRanges(table);
+ assertThat(reassignedIndexRanges).hasSize(expectedIndexFileCount);
+ for (Range range : reassignedIndexRanges) {
+ assertThat(range.from).isGreaterThanOrEqualTo(scrambledNextRowId);
+ assertThat(range.to).isLessThan(expectedNextRowId);
+ }
+ assertTableDataQueryable(table, expectedRows);
}
- @Disabled("Writes one million manifest entries; run manually for large
metadata stress tests.")
@Test
- public void testReassignOneMillionOutOfOrderPartitionEntries() throws
Exception {
- verifyReassignOutOfOrderPartitionEntries(
- ONE_MILLION_ENTRY_COUNT, ONE_MILLION_MANIFEST_FILE_COUNT);
+ public void testPartitionReassignPreservesDataAndOtherPartitionRowIds()
throws Exception {
+ FileStoreTable table = createDataIntegrityTable();
+ long scrambledNextRowId = scrambleCurrentRowIds(table);
+ createBTreeIndex(table);
+
+ Snapshot before = table.snapshotManager().latestSnapshot();
+ List<String> expectedRows = readTableRows(table);
+ Map<String, List<Object>> expectedFileMetadata =
fileMetadataWithoutRowId(table);
+ Map<String, SimpleStats> expectedValueStats = valueStatsByFile(table);
+ Map<String, Long> rowCountsByPartition = rowCountsByPartition(table);
+ Map<String, List<Long>> beforeFirstRowIds = rowIdsByPartition(table);
+ Map<String, List<Long>> beforeExpandedRowIds =
expandedRowIdsByPartition(table);
+ Map<String, List<Range>> beforeIndexRanges =
globalIndexRangesByPartition(table);
+ String reassignedPartition = "c";
+ String reassignedPartitionPath = "pt=" + reassignedPartition + "/";
+
+ assertThat(before.nextRowId()).isEqualTo(scrambledNextRowId);
+ assertThat(before.totalRecordCount()).isEqualTo(expectedRows.size());
+ assertPartitionsHaveLargeRowIdGaps(beforeFirstRowIds);
+ assertTableDataQueryable(table, expectedRows);
+
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table, partitionPredicate(table,
reassignedPartition))
+ .reassign("test-partition-reassign-data-integrity");
+
+ long reassignedRowCount =
rowCountsByPartition.get(reassignedPartitionPath);
+ long expectedNextRowId = scrambledNextRowId + reassignedRowCount;
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(before.id());
+ assertThat(result.newSnapshotId).isEqualTo(before.id() + 1);
+ assertThat(result.firstAssignedRowId).isEqualTo(scrambledNextRowId);
+ assertThat(result.nextRowId).isEqualTo(expectedNextRowId);
+
assertThat(result.fileCount).isEqualTo(DATA_INTEGRITY_FILES_PER_PARTITION);
+ assertThat(result.rowCount).isEqualTo(reassignedRowCount);
+ assertThat(result.indexFileCount)
+
.isEqualTo(beforeIndexRanges.get(reassignedPartitionPath).size());
+
+ Map<String, List<Long>> afterFirstRowIds = rowIdsByPartition(table);
+ Map<String, List<Long>> afterExpandedRowIds =
expandedRowIdsByPartition(table);
+ Map<String, List<Range>> afterIndexRanges =
globalIndexRangesByPartition(table);
+ assertContiguousRowIds(
+ afterExpandedRowIds,
+ rowCountsByPartition,
+ Collections.singletonList(reassignedPartition),
+ scrambledNextRowId);
+ assertThat(afterFirstRowIds.get(reassignedPartitionPath))
+ .isNotEqualTo(beforeFirstRowIds.get(reassignedPartitionPath));
+ for (String partition : DATA_INTEGRITY_PARTITIONS) {
+ String partitionPath = "pt=" + partition + "/";
+ if (!partition.equals(reassignedPartition)) {
+ assertThat(afterFirstRowIds.get(partitionPath))
+ .isEqualTo(beforeFirstRowIds.get(partitionPath));
+ assertThat(afterExpandedRowIds.get(partitionPath))
+ .isEqualTo(beforeExpandedRowIds.get(partitionPath));
+ assertThat(afterIndexRanges.get(partitionPath))
+ .isEqualTo(beforeIndexRanges.get(partitionPath));
+ }
+ }
+ assertThat(afterIndexRanges.get(reassignedPartitionPath))
+ .hasSameSizeAs(beforeIndexRanges.get(reassignedPartitionPath));
+ for (Range range : afterIndexRanges.get(reassignedPartitionPath)) {
+ assertThat(range.from).isGreaterThanOrEqualTo(scrambledNextRowId);
+ assertThat(range.to).isLessThan(expectedNextRowId);
+ }
+
+ assertThat(table.snapshotManager().latestSnapshot().nextRowId())
+ .isEqualTo(expectedNextRowId);
+ assertThat(table.snapshotManager().latestSnapshot().totalRecordCount())
+ .isEqualTo(before.totalRecordCount());
+
assertThat(fileMetadataWithoutRowId(table)).isEqualTo(expectedFileMetadata);
+ assertThat(valueStatsByFile(table)).isEqualTo(expectedValueStats);
+ assertTableDataQueryable(table, expectedRows);
+ }
+
+ @Test
+ public void testReassignManyOutOfOrderPartitionEntries() throws Exception {
+ verifyReassignOutOfOrderPartitionEntries(LARGE_ENTRY_COUNT,
LARGE_MANIFEST_FILE_COUNT);
}
private FileStoreTable createTableWithInterleavedPartitions() throws
Exception {
@@ -462,6 +1128,88 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return table;
}
+ private FileStoreTable createDataIntegrityTable() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ int nextId = 0;
+ for (int file = 0; file < DATA_INTEGRITY_FILES_PER_PARTITION; file++) {
+ for (int partition = 0; partition <
DATA_INTEGRITY_PARTITIONS.length; partition++) {
+ int rowCount = 3 + (file + partition) % 5;
+ int[] ids = new int[rowCount];
+ for (int row = 0; row < rowCount; row++) {
+ ids[row] = nextId++;
+ }
+ writeRows(table, DATA_INTEGRITY_PARTITIONS[partition], ids);
+ }
+ }
+ return table;
+ }
+
+ private long scrambleCurrentRowIds(FileStoreTable table) throws Exception {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ assertThat(latest.indexManifest()).isNull();
+ List<ManifestEntry> currentEntries = currentEntries(table);
+ List<Long> rowIdSlots = new ArrayList<>(currentEntries.size());
+ for (int i = 0; i < currentEntries.size(); i++) {
+ rowIdSlots.add((i + 1L) * SCRAMBLED_ROW_ID_GAP);
+ }
+ Collections.shuffle(rowIdSlots, new Random(SCRAMBLED_ROW_ID_SEED));
+
+ Map<FileEntry.Identifier, Long> assignments = new HashMap<>();
+ for (int i = 0; i < currentEntries.size(); i++) {
+ assignments.put(currentEntries.get(i).identifier(),
rowIdSlots.get(i));
+ }
+
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ ManifestList manifestList =
table.store().manifestListFactory().create();
+ List<ManifestFileMeta> rewrittenManifestMetas = new ArrayList<>();
+ int rewrittenEntryCount = 0;
+ for (ManifestFileMeta manifestMeta :
manifestList.readDataManifests(latest)) {
+ List<ManifestEntry> entries =
+ manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
+ boolean rewritten = false;
+ for (int i = 0; i < entries.size(); i++) {
+ ManifestEntry entry = entries.get(i);
+ Long firstRowId =
+ entry.kind() == FileKind.ADD ?
assignments.get(entry.identifier()) : null;
+ if (firstRowId != null) {
+ entries.set(i, entry.assignFirstRowId(firstRowId));
+ rewritten = true;
+ rewrittenEntryCount++;
+ }
+ }
+ if (rewritten) {
+ rewrittenManifestMetas.addAll(manifestFile.write(entries));
+ } else {
+ rewrittenManifestMetas.add(manifestMeta);
+ }
+ }
+ assertThat(rewrittenEntryCount).isEqualTo(currentEntries.size());
+
+ Pair<String, Long> baseManifestList =
manifestList.write(rewrittenManifestMetas);
+ Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
+ long scrambledNextRowId = (currentEntries.size() + 10L) *
SCRAMBLED_ROW_ID_GAP;
+ try (FileStoreCommitImpl commit =
+ (FileStoreCommitImpl)
table.store().newCommit("test-scramble-row-ids", table)) {
+ assertThat(
+ commit.replaceManifestList(
+ latest,
+ latest.totalRecordCount(),
+ baseManifestList,
+ deltaManifestList,
+ null,
+ scrambledNextRowId))
+ .isTrue();
+ }
+
+ Map<FileEntry.Identifier, Long> committedAssignments = new HashMap<>();
+ for (ManifestEntry entry : currentEntries(table)) {
+ committedAssignments.put(entry.identifier(),
entry.file().nonNullFirstRowId());
+ }
+ assertThat(committedAssignments).isEqualTo(assignments);
+ return scrambledNextRowId;
+ }
+
private void verifyReassignOutOfOrderPartitionEntries(int entryCount, int
manifestFileCount)
throws Exception {
assertThat(entryCount % 2).isEqualTo(0);
@@ -791,16 +1539,184 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
try (BatchTableWrite write = builder.newWrite();
BatchTableCommit commit = builder.newCommit()) {
for (int id : ids) {
+ write.write(row(partition, id));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void writePartialPayload(
+ FileStoreTable table, String partition, long firstRowId, String...
payloads)
+ throws Exception {
+ RowType writeType = table.rowType().project(Arrays.asList("pt",
"payload"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (String payload : payloads) {
write.write(
GenericRow.of(
BinaryString.fromString(partition),
- id,
- BinaryString.fromString("v" + id)));
+ BinaryString.fromString(payload)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ commit.commit(messages);
+ }
+ }
+
+ private FileStoreTable withManifestMergeOnNextAppend(FileStoreTable table)
{
+ Map<String, String> mergeOptions = new HashMap<>();
+ mergeOptions.put(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1GB");
+ mergeOptions.put(CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), "2");
+ mergeOptions.put(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
"1GB");
+ mergeOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "false");
+ return table.copy(mergeOptions);
+ }
+
+ private void assignFirstRowId(List<CommitMessage> messages, long
firstRowId) {
+ for (CommitMessage message : messages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ List<DataFileMeta> files =
+ new
ArrayList<>(commitMessage.newFilesIncrement().newFiles());
+ commitMessage.newFilesIncrement().newFiles().clear();
+ for (DataFileMeta file : files) {
+
commitMessage.newFilesIncrement().newFiles().add(file.assignFirstRowId(firstRowId));
}
+ }
+ }
+
+ private void writeRowsInPartitions(
+ FileStoreTable table,
+ String firstPartition,
+ int firstId,
+ String secondPartition,
+ int secondId)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(row(firstPartition, firstId));
+ write.write(row(secondPartition, secondId));
commit.commit(write.prepareCommit());
}
}
+ private GenericRow row(String partition, int id) {
+ return GenericRow.of(
+ partition == null ? null : BinaryString.fromString(partition),
+ id,
+ BinaryString.fromString("v" + id));
+ }
+
+ private void overwriteOneRow(FileStoreTable table, String partition, int
id) throws Exception {
+ BatchWriteBuilder builder =
table.newBatchWriteBuilder().withOverwrite();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ id,
+ BinaryString.fromString("v" + id)));
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void compactManifests(FileStoreTable table) throws Exception {
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.compactManifests();
+ }
+ }
+
+ private FileEntry.Identifier compactOneFile(FileStoreTable table, String
partitionPath)
+ throws Exception {
+ ManifestEntry compactBefore =
+ currentEntries(table).stream()
+ .filter(
+ entry ->
+ table.store()
+ .pathFactory()
+
.getPartitionString(entry.partition())
+ .equals(partitionPath))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Cannot find file in partition
" + partitionPath));
+ DataFilePathFactory pathFactory =
+ table.store()
+ .pathFactory()
+ .createDataFilePathFactory(
+ compactBefore.partition(),
compactBefore.bucket());
+ Path compactAfterPath = pathFactory.newPath();
+ table.fileIO().copyFile(pathFactory.toPath(compactBefore.file()),
compactAfterPath, false);
+ DataFileMeta compactAfter =
compactBefore.file().rename(compactAfterPath.getName());
+ CommitMessage message =
+ new CommitMessageImpl(
+ compactBefore.partition(),
+ compactBefore.bucket(),
+ compactBefore.totalBuckets(),
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+
Collections.singletonList(compactBefore.file()),
+ Collections.singletonList(compactAfter),
+ Collections.emptyList()));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ return compactBefore.identifier();
+ }
+
+ private long assertAddDeleteEntriesConsistent(
+ FileStoreTable table, FileEntry.Identifier identifier) {
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ ManifestList manifestList =
table.store().manifestListFactory().create();
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ List<ManifestEntry> matchingEntries = new ArrayList<>();
+ Set<String> matchingManifestFiles = new HashSet<>();
+ for (ManifestFileMeta manifestMeta :
manifestList.readDataManifests(latest)) {
+ for (ManifestEntry entry :
+ manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize())) {
+ if (entry.identifier().equals(identifier)) {
+ matchingEntries.add(entry);
+ matchingManifestFiles.add(manifestMeta.fileName());
+ }
+ }
+ }
+
+ assertThat(matchingEntries).hasSize(2);
+ assertThat(matchingManifestFiles).hasSize(2);
+ ManifestEntry add =
+ matchingEntries.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .findFirst()
+ .orElseThrow(
+ () -> new AssertionError("Missing ADD entry
for " + identifier));
+ ManifestEntry delete =
+ matchingEntries.stream()
+ .filter(entry -> entry.kind() == FileKind.DELETE)
+ .findFirst()
+ .orElseThrow(
+ () -> new AssertionError("Missing DELETE entry
for " + identifier));
+ assertThat(delete.partition()).isEqualTo(add.partition());
+ assertThat(delete.bucket()).isEqualTo(add.bucket());
+ assertThat(delete.totalBuckets()).isEqualTo(add.totalBuckets());
+ assertThat(delete.file()).isEqualTo(add.file());
+ return add.file().nonNullFirstRowId();
+ }
+
+ private void updateStatistics(FileStoreTable table) throws Exception {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ Statistics statistics =
+ new Statistics(
+ latest.id(),
+ latest.schemaId(),
+ latest.totalRecordCount(),
+ latest.totalRecordCount() * 100L);
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.updateStatistics(statistics);
+ }
+ }
+
private void writeUnpartitionedRows(FileStoreTable table, int... ids)
throws Exception {
BatchWriteBuilder builder = table.newBatchWriteBuilder();
try (BatchTableWrite write = builder.newWrite();
@@ -819,15 +1735,100 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
table.coreOptions().partitionDefaultName());
}
+ private List<ManifestEntry> currentEntries(FileStoreTable table) {
+ return table.store()
+ .newScan()
+ .withSnapshot(table.snapshotManager().latestSnapshot())
+ .plan()
+ .files();
+ }
+
+ private List<ManifestEntry> currentEntriesAtRange(
+ FileStoreTable table, String partitionPath, long firstRowId, long
lastRowId) {
+ List<ManifestEntry> result = new ArrayList<>();
+ for (ManifestEntry entry : currentEntries(table)) {
+ String partition =
table.store().pathFactory().getPartitionString(entry.partition());
+ Range range = entry.file().nonNullRowIdRange();
+ if (partitionPath.equals(partition)
+ && range.from == firstRowId
+ && range.to == lastRowId) {
+ result.add(entry);
+ }
+ }
+ return result;
+ }
+
+ private Map<String, Long> rowCountsByPartition(FileStoreTable table) {
+ Map<String, Long> result = new LinkedHashMap<>();
+ for (ManifestEntry entry : currentEntries(table)) {
+ String partition =
table.store().pathFactory().getPartitionString(entry.partition());
+ result.merge(partition, entry.file().rowCount(), Long::sum);
+ }
+ return result;
+ }
+
+ private Map<String, List<Object>> fileMetadataWithoutRowId(FileStoreTable
table) {
+ Map<String, List<Object>> result = new LinkedHashMap<>();
+ for (ManifestEntry entry : currentEntries(table)) {
+ DataFileMeta file = entry.file();
+ String partition =
table.store().pathFactory().getPartitionString(entry.partition());
+ String key = partition + entry.bucket() + "/" + file.fileName();
+ List<Object> metadata =
+ Arrays.asList(
+ entry.kind(),
+ entry.totalBuckets(),
+ file.fileSize(),
+ file.rowCount(),
+ file.minSequenceNumber(),
+ file.maxSequenceNumber(),
+ file.schemaId(),
+ file.level(),
+ file.extraFiles(),
+ file.deleteRowCount(),
+ file.fileSource(),
+ file.valueStatsCols(),
+ file.externalPath(),
+ file.writeCols());
+ assertThat(result.put(key, metadata)).isNull();
+ }
+ return result;
+ }
+
+ private void assertPartitionsHaveLargeRowIdGaps(
+ Map<String, List<Long>> firstRowIdsByPartition) {
+
assertThat(firstRowIdsByPartition).hasSize(DATA_INTEGRITY_PARTITIONS.length);
+ for (String partition : DATA_INTEGRITY_PARTITIONS) {
+ List<Long> firstRowIds = firstRowIdsByPartition.get("pt=" +
partition + "/");
+
assertThat(firstRowIds).hasSize(DATA_INTEGRITY_FILES_PER_PARTITION);
+ for (int i = 1; i < firstRowIds.size(); i++) {
+ assertThat(firstRowIds.get(i) - firstRowIds.get(i - 1))
+ .isGreaterThanOrEqualTo(SCRAMBLED_ROW_ID_GAP);
+ }
+ }
+ }
+
+ private long assertContiguousRowIds(
+ Map<String, List<Long>> rowIdsByPartition,
+ Map<String, Long> rowCountsByPartition,
+ List<String> partitions,
+ long firstRowId) {
+ long nextRowId = firstRowId;
+ for (String partition : partitions) {
+ String partitionPath = "pt=" + partition + "/";
+ long rowCount = rowCountsByPartition.get(partitionPath);
+ List<Long> rowIds = rowIdsByPartition.get(partitionPath);
+ assertThat(rowIds).hasSize(Math.toIntExact(rowCount));
+ for (int i = 0; i < rowIds.size(); i++) {
+ assertThat(rowIds.get(i)).isEqualTo(nextRowId + i);
+ }
+ nextRowId += rowCount;
+ }
+ return nextRowId;
+ }
+
private Map<String, List<Long>> rowIdsByPartition(FileStoreTable table) {
- List<ManifestEntry> entries =
- table.store()
- .newScan()
- .withSnapshot(table.snapshotManager().latestSnapshot())
- .plan()
- .files();
Map<String, List<Long>> result = new LinkedHashMap<>();
- for (ManifestEntry entry : entries) {
+ for (ManifestEntry entry : currentEntries(table)) {
String partition =
table.store().pathFactory().getPartitionString(entry.partition());
result.computeIfAbsent(partition, k -> new ArrayList<>())
.add(entry.file().nonNullFirstRowId());
@@ -838,15 +1839,18 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return result;
}
+ private Map<String, SimpleStats> valueStatsByFile(FileStoreTable table) {
+ Map<String, SimpleStats> result = new LinkedHashMap<>();
+ for (ManifestEntry entry : currentEntries(table)) {
+ SimpleStats previous = result.put(entry.file().fileName(),
entry.file().valueStats());
+ assertThat(previous).isNull();
+ }
+ return result;
+ }
+
private Map<String, List<Long>> expandedRowIdsByPartition(FileStoreTable
table) {
- List<ManifestEntry> entries =
- table.store()
- .newScan()
- .withSnapshot(table.snapshotManager().latestSnapshot())
- .plan()
- .files();
Map<String, List<Long>> result = new LinkedHashMap<>();
- for (ManifestEntry entry : entries) {
+ for (ManifestEntry entry : currentEntries(table)) {
String partition =
table.store().pathFactory().getPartitionString(entry.partition());
List<Long> rowIds = result.computeIfAbsent(partition, k -> new
ArrayList<>());
Range range = entry.file().nonNullRowIdRange();
@@ -971,6 +1975,52 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
}
String staleIndexManifest =
indexManifestFile.writeWithoutRolling(rewritten);
+ replaceLatestSnapshotIndexManifest(table, latest, staleIndexManifest);
+ }
+
+ private void appendGlobalIndexRange(FileStoreTable table, String
partition, Range rowRange)
+ throws Exception {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ IndexManifestFile indexManifestFile =
table.store().indexManifestFileFactory().create();
+ List<IndexManifestEntry> entries =
indexManifestFile.read(latest.indexManifest());
+ IndexManifestEntry template = null;
+ for (IndexManifestEntry entry : entries) {
+ if (table.store()
+ .pathFactory()
+ .getPartitionString(entry.partition())
+ .equals(partition)) {
+ template = entry;
+ break;
+ }
+ }
+ assertThat(template).isNotNull();
+ IndexFileMeta indexFile = template.indexFile();
+ GlobalIndexMeta globalIndex = indexFile.globalIndexMeta();
+ assertThat(globalIndex).isNotNull();
+ entries.add(
+ new IndexManifestEntry(
+ template.kind(),
+ template.partition(),
+ template.bucket(),
+ new IndexFileMeta(
+ indexFile.indexType(),
+ indexFile.fileName(),
+ indexFile.fileSize(),
+ indexFile.rowCount(),
+ indexFile.dvRanges(),
+ indexFile.externalPath(),
+ new GlobalIndexMeta(
+ rowRange.from,
+ rowRange.to,
+ globalIndex.indexFieldId(),
+ globalIndex.extraFieldIds(),
+ globalIndex.indexMeta()))));
+ replaceLatestSnapshotIndexManifest(
+ table, latest, indexManifestFile.writeWithoutRolling(entries));
+ }
+
+ private void replaceLatestSnapshotIndexManifest(
+ FileStoreTable table, Snapshot latest, String indexManifest)
throws Exception {
Snapshot staleSnapshot =
new Snapshot(
latest.version(),
@@ -982,7 +2032,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
latest.deltaManifestListSize(),
latest.changelogManifestList(),
latest.changelogManifestListSize(),
- staleIndexManifest,
+ indexManifest,
latest.commitUser(),
latest.commitIdentifier(),
latest.commitKind(),
@@ -1025,6 +2075,100 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
return ranges;
}
+ private Map<String, List<Range>>
globalIndexRangesByPartition(FileStoreTable table) {
+ Map<String, List<Range>> result = new LinkedHashMap<>();
+ List<IndexManifestEntry> entries =
+ table.store()
+ .indexManifestFileFactory()
+ .create()
+
.read(table.snapshotManager().latestSnapshot().indexManifest());
+ for (IndexManifestEntry entry : entries) {
+ GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
+ if (globalIndex == null) {
+ continue;
+ }
+ String partition =
table.store().pathFactory().getPartitionString(entry.partition());
+ result.computeIfAbsent(partition, ignored -> new ArrayList<>())
+ .add(globalIndex.rowRange());
+ }
+ for (List<Range> ranges : result.values()) {
+ Collections.sort(
+ ranges,
+ (left, right) -> {
+ int compare = Long.compare(left.from, right.from);
+ return compare == 0 ? Long.compare(left.to, right.to)
: compare;
+ });
+ }
+ return result;
+ }
+
+ private void assertTableDataQueryable(FileStoreTable table, List<String>
expectedRows)
+ throws Exception {
+
assertThat(readTableRows(table)).containsExactlyElementsOf(expectedRows);
+
+ PredicateBuilder predicateBuilder = new
PredicateBuilder(table.rowType());
+ int partitionField = table.rowType().getFieldIndex("pt");
+ for (String partition : DATA_INTEGRITY_PARTITIONS) {
+ Predicate predicate =
+ predicateBuilder.equal(partitionField,
BinaryString.fromString(partition));
+ assertThat(readTableRows(table, predicate))
+ .containsExactlyElementsOf(rowsForPartition(expectedRows,
partition));
+ }
+
+ int idField = table.rowType().getFieldIndex("id");
+ int[] pointIds = {0, expectedRows.size() / 2, expectedRows.size() - 1};
+ for (int id : pointIds) {
+ Predicate predicate = predicateBuilder.equal(idField, id);
+ assertThat(readTableRows(table,
predicate)).containsAll(rowsForId(expectedRows, id));
+ }
+ }
+
+ private List<String> rowsForPartition(List<String> rows, String partition)
{
+ List<String> result = new ArrayList<>();
+ for (String row : rows) {
+ if (row.split("\\|", 3)[1].equals(partition)) {
+ result.add(row);
+ }
+ }
+ return result;
+ }
+
+ private List<String> rowsForId(List<String> rows, int id) {
+ List<String> result = new ArrayList<>();
+ String prefix = id + "|";
+ for (String row : rows) {
+ if (row.startsWith(prefix)) {
+ result.add(row);
+ }
+ }
+ return result;
+ }
+
+ private List<String> readTableRows(FileStoreTable table) throws Exception {
+ return readTableRows(table.newReadBuilder());
+ }
+
+ private List<String> readTableRows(FileStoreTable table, Predicate
predicate) throws Exception {
+ return readTableRows(table.newReadBuilder().withFilter(predicate));
+ }
+
+ private List<String> readTableRows(ReadBuilder readBuilder) throws
Exception {
+ List<String> result = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(
+ row -> {
+ InternalRow projected = row;
+ String partition =
projected.getString(0).toString();
+ int id = projected.getInt(1);
+ String payload = projected.getString(2).toString();
+ result.add(id + "|" + partition + "|" + payload);
+ });
+ Collections.sort(result);
+ return result;
+ }
+
private List<String> readPayloads(FileStoreTable table, Predicate
predicate) throws Exception {
ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
List<String> result = new ArrayList<>();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
index baeae7d34e..266574106c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
@@ -71,4 +71,33 @@ public class RowRangeMappingIndexTest {
assertThat(index.map(new Range(12, 17))).isEmpty();
}
+
+ @Test
+ public void testRelativeMappingCanBeShiftedToAbsoluteRowIds() {
+ RowRangeMappingIndex relative =
+ RowRangeMappingIndex.create(
+ Arrays.asList(
+ RowRangeMappingIndex.mapping(10, 14, 0),
+ RowRangeMappingIndex.mapping(20, 24, 5)));
+
+ RowRangeMappingIndex absolute = relative.shiftNewStarts(100L);
+ assertThat(absolute.map(new Range(10, 14))).hasValue(new Range(100,
104));
+ assertThat(absolute.map(new Range(20, 24))).hasValue(new Range(105,
109));
+ }
+
+ @Test
+ public void testOverlaps() {
+ RowRangeMappingIndex index =
+ RowRangeMappingIndex.create(
+ Arrays.asList(
+ RowRangeMappingIndex.mapping(10, 14, 100),
+ RowRangeMappingIndex.mapping(20, 24, 105)));
+
+ assertThat(index.overlaps(new Range(5, 9))).isFalse();
+ assertThat(index.overlaps(new Range(5, 10))).isTrue();
+ assertThat(index.overlaps(new Range(15, 19))).isFalse();
+ assertThat(index.overlaps(new Range(14, 20))).isTrue();
+ assertThat(index.overlaps(new Range(24, 30))).isTrue();
+ assertThat(index.overlaps(new Range(25, 30))).isFalse();
+ }
}
