This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 702900d458 [core] Improve row ID reassignment under concurrent writes
(#9439)
702900d458 is described below
commit 702900d458947172887194bc866eeb06ebb1f033
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 28 11:30:18 2026 +0800
[core] Improve row ID reassignment under concurrent writes (#9439)
---
.../DataEvolutionRowIdReassigner.java | 144 ++++++++++++++-------
.../paimon/operation/FileStoreCommitImpl.java | 6 +
.../DataEvolutionRowIdReassignerTest.java | 71 ++++++++++
3 files changed, 175 insertions(+), 46 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 361b9a812a..7c49757f07 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
@@ -43,6 +43,7 @@ import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RetryWaiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,7 +69,6 @@ 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;
@@ -135,52 +135,29 @@ public class DataEvolutionRowIdReassigner {
return Result.skipped(
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);
- }
-
- 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;
- }
-
- throw new IllegalStateException("Unreachable retry state while
reassigning row IDs.");
+ CommittedAssignment committed =
+ commitAssignmentWithRetry(
+ optionalPlan.get(), latest, manifestFile,
manifestList, commitUser);
+ Assignment assignment = committed.assignment;
+ CommitAssignmentResult commitResult = committed.commitResult;
+ 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);
}
private Optional<AssignmentPlan> planAssignment(List<ManifestFileMeta>
manifestMetas) {
@@ -358,6 +335,65 @@ public class DataEvolutionRowIdReassigner {
return partitionPredicate != null;
}
+ private CommittedAssignment commitAssignmentWithRetry(
+ AssignmentPlan initialAssignmentPlan,
+ Snapshot initialSnapshot,
+ ManifestFile manifestFile,
+ ManifestList manifestList,
+ String commitUser) {
+ AssignmentPlan assignmentPlan = initialAssignmentPlan;
+ Snapshot latest = initialSnapshot;
+ int retryCount = 0;
+ long startMillis = System.currentTimeMillis();
+ CoreOptions options = table.coreOptions();
+ RetryWaiter retryWaiter =
+ new RetryWaiter(options.commitMinRetryWait(),
options.commitMaxRetryWait());
+
+ while (true) {
+ Snapshot observedLatest = table.snapshotManager().latestSnapshot();
+ checkState(
+ observedLatest != null,
+ "Latest snapshot disappeared while reassigning row IDs.");
+ if (observedLatest.id() > latest.id()) {
+ assignmentPlan =
+ advanceAssignmentPlan(
+ assignmentPlan, latest, observedLatest,
manifestFile, manifestList);
+ latest = observedLatest;
+ }
+
+ Assignment assignment = assignmentPlan.createAssignment(latest);
+ CommitAssignmentResult commitResult =
+ commitAssignment(assignment, manifestFile, manifestList,
commitUser);
+ if (commitResult.success) {
+ return new CommittedAssignment(assignment, commitResult);
+ }
+
+ if (System.currentTimeMillis() - startMillis >
options.commitTimeout()
+ || retryCount >= options.commitMaxRetries()) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to reassign row IDs after %s millis
with %s retries because newer snapshots kept being committed.",
+ System.currentTimeMillis() - startMillis,
retryCount));
+ }
+
+ 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(),
+ retryCount + 1,
+ options.commitMaxRetries());
+ retryWaiter.retryWait(retryCount);
+ retryCount++;
+ latest = newLatest;
+ }
+ }
+
private CommitAssignmentResult commitAssignment(
Assignment assignment,
ManifestFile manifestFile,
@@ -556,15 +592,21 @@ public class DataEvolutionRowIdReassigner {
Map<String, List<ManifestFileMeta>> rewrittenManifestMetas,
ManifestList manifestList) {
List<ManifestFileMeta> baseManifestMetas = new ArrayList<>();
+ Set<String> unmatchedReplacements = new
HashSet<>(rewrittenManifestMetas.keySet());
for (ManifestFileMeta manifestMeta : manifestMetas) {
List<ManifestFileMeta> replacement =
rewrittenManifestMetas.get(manifestMeta.fileName());
if (replacement == null) {
baseManifestMetas.add(manifestMeta);
} else {
+ unmatchedReplacements.remove(manifestMeta.fileName());
baseManifestMetas.addAll(replacement);
}
}
+ checkState(
+ unmatchedReplacements.isEmpty(),
+ "Cannot replace planned manifests %s because they are not in
the current manifest list.",
+ unmatchedReplacements);
return manifestList.write(baseManifestMetas);
}
@@ -860,6 +902,16 @@ public class DataEvolutionRowIdReassigner {
}
}
+ private static class CommittedAssignment {
+ private final Assignment assignment;
+ private final CommitAssignmentResult commitResult;
+
+ private CommittedAssignment(Assignment assignment,
CommitAssignmentResult commitResult) {
+ this.assignment = assignment;
+ this.commitResult = commitResult;
+ }
+ }
+
private static class CommitAssignmentResult {
private final boolean success;
private final long fileCount;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index cd4919b9ce..2809101551 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1372,6 +1372,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latest.nextRowId());
}
+ /**
+ * Replaces the manifest lists exactly as supplied, without manifest
compaction or sorting.
+ *
+ * <p>This is intended for metadata-only operations which have already
produced their final
+ * manifest layout and must commit it without invoking {@link
ManifestFileMerger}.
+ */
public boolean replaceManifestList(
Snapshot latest,
long totalRecordCount,
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 08b35b7bdf..af0178bb27 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
@@ -976,6 +976,40 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsEntry("pt=d/", Collections.singletonList(6L));
}
+ @Test
+ public void testReassignUsesConfiguredRetryBudget() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ Map<String, String> retryOptions = new HashMap<>();
+ retryOptions.put(CoreOptions.COMMIT_MAX_RETRIES.key(), "3");
+ retryOptions.put(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "0ms");
+ retryOptions.put(CoreOptions.COMMIT_MAX_RETRY_WAIT.key(), "0ms");
+ FileStoreTable configured = table.copy(retryOptions);
+
+ AtomicInteger beforeCommits = new AtomicInteger();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ configured,
+ partitionPredicate(configured, "a"),
+ () -> {
+ int attempt =
beforeCommits.getAndIncrement();
+ if (attempt < 3) {
+ try {
+ writeOneRow(
+ configured, "new-" +
attempt, 100 + attempt);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign("test-reassign-configured-retries");
+
+ assertThat(beforeCommits).hasValue(4);
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.fileCount).isEqualTo(3L);
+ assertThat(result.rowCount).isEqualTo(3L);
+
assertThat(rowIdsByPartition(configured).get("pt=a/")).containsExactly(8L, 9L,
10L);
+ }
+
@Test
public void
testReassignPartitionFilterAfterConcurrentAppendOutsideFilter() throws
Exception {
FileStoreTable table = createTableWithInterleavedPartitions();
@@ -1287,6 +1321,43 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(afterManifestFiles).doesNotContainAnyElementsOf(affectedManifests);
}
+ @Test
+ public void testReassignDoesNotCompactManifests() throws Exception {
+ testReassignSkipsManifestOptimization(false);
+ }
+
+ @Test
+ public void testReassignDoesNotSortManifests() throws Exception {
+ testReassignSkipsManifestOptimization(true);
+ }
+
+ private void testReassignSkipsManifestOptimization(boolean
manifestSortEnabled)
+ throws Exception {
+ FileStoreTable table = createTableWithPartiallyOverlappedPartitions();
+ Map<String, Set<String>> partitionsByManifest =
currentPartitionsByManifest(table);
+ List<String> unaffectedManifests = new ArrayList<>();
+ for (Map.Entry<String, Set<String>> entry :
partitionsByManifest.entrySet()) {
+ if (!entry.getValue().contains("pt=a/")) {
+ unaffectedManifests.add(entry.getKey());
+ }
+ }
+ assertThat(unaffectedManifests).isNotEmpty();
+
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.MANIFEST_SORT_ENABLED.key(),
Boolean.toString(manifestSortEnabled));
+ options.put(CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), "1");
+ options.put(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
"1B");
+ FileStoreTable configured = table.copy(options);
+
+ new DataEvolutionRowIdReassigner(configured)
+ .reassign(
+ manifestSortEnabled
+ ? "test-reassign-with-manifest-sort"
+ : "test-reassign-with-manifest-compaction");
+
+
assertThat(dataManifestFileNames(configured)).containsAll(unaffectedManifests);
+ }
+
@Test
public void testSkipWhenPartitionRowIdsAreContiguous() throws Exception {
createTableDefault();
