Skip to content

[core] Support row ID reassignment across concurrent compaction #10780

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 9 commits into from
Oct 26, 2026
Merged

[core] Support row ID reassignment across concurrent compaction #10780

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b3b85f1
[core] Support row ID reassignment across concurrent compaction
leaves12138 Oct 23, 2026
03ce7ed
[core] Simplify row ID reassignment planning calls
leaves12138 Oct 26, 2026
3d27a63
[core] Remove reflection from row ID reassignment tests
leaves12138 Oct 26, 2026
e3ec25d
[core] Propagate row ID reassignment commit exceptions
leaves12138 Oct 26, 2026
74b09a1
[core] Limit reassignment changes to COMPACT advancement
leaves12138 Oct 26, 2026
a987f48
[core] Reuse existing mappings or reject concurrent compaction
leaves12138 Oct 26, 2026
83f79da
[core] Allow reassignment across deletion-vector materialization
leaves12138 Oct 26, 2026
c1432e7
[core] Simplify concurrent compaction reassignment tests
leaves12138 Oct 26, 2026
07ac3db
[core] Inline row ID reassignment test factory
leaves12138 Oct 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 12 additions & 13 deletions ...re/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -487,8 +487,7 @@ private AssignmentPlan advanceAssignmentPlan(
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.",
Expand All @@ -498,14 +497,15 @@ private AssignmentPlan advanceAssignmentPlan(
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);
Expand Down Expand Up @@ -562,11 +562,11 @@ private boolean manifestContainsMappedEntry(
}
}

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(
Expand All @@ -577,11 +577,11 @@ private boolean appendedManifestNeedsReassign(
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;
}
}
Expand All @@ -594,8 +594,7 @@ private boolean appendedManifestNeedsReassign(
}
}

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) {
Expand All @@ -609,7 +608,7 @@ private boolean appendedEntryNeedsReassign(

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;
Expand Down
Loading
Loading