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 c049c02c1a [core] Parallelize row ID reassignment manifest rewrites
(#9441)
c049c02c1a is described below
commit c049c02c1a4cbc3a94f0030b8205e2e8ebeac16b
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 28 13:26:09 2026 +0800
[core] Parallelize row ID reassignment manifest rewrites (#9441)
---
.../DataEvolutionRowIdReassigner.java | 117 ++++++++++++++++-----
.../DataEvolutionRowIdReassignerTest.java | 95 +++++++++++++++++
2 files changed, 185 insertions(+), 27 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 7c49757f07..c7e5e77a69 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
@@ -44,6 +44,7 @@ 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.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,7 +61,11 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import static java.util.Collections.singletonList;
+import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecuteCloseable;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
@@ -73,6 +78,7 @@ public class DataEvolutionRowIdReassigner {
private final FileStoreTable table;
private final @Nullable PartitionPredicate partitionPredicate;
private final Runnable beforeCommit;
+ private final Consumer<ManifestFileMeta> beforeManifestRewrite;
public DataEvolutionRowIdReassigner(FileStoreTable table) {
this(table, null);
@@ -88,9 +94,19 @@ public class DataEvolutionRowIdReassigner {
FileStoreTable table,
@Nullable PartitionPredicate partitionPredicate,
Runnable beforeCommit) {
+ this(table, partitionPredicate, beforeCommit, manifest -> {});
+ }
+
+ @VisibleForTesting
+ DataEvolutionRowIdReassigner(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionPredicate,
+ Runnable beforeCommit,
+ Consumer<ManifestFileMeta> beforeManifestRewrite) {
this.table = table;
this.partitionPredicate = partitionPredicate;
this.beforeCommit = beforeCommit;
+ this.beforeManifestRewrite = beforeManifestRewrite;
}
public Result reassign() {
@@ -612,39 +628,71 @@ public class DataEvolutionRowIdReassigner {
private RewrittenDataManifests writeManifestReplacements(
Assignment assignment, ManifestFile manifestFile) {
- Map<String, List<ManifestFileMeta>> rewrittenManifestMetas = new
HashMap<>();
- long fileCount = 0L;
- for (ManifestFileMeta manifestMeta :
assignment.manifestMetasToRewrite) {
- List<ManifestEntry> entries =
- manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
- 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++;
- }
+ Integer parallelism = table.coreOptions().scanManifestParallelism();
+ List<RewrittenDataManifest> rewritten =
+ new ArrayList<>(assignment.manifestMetasToRewrite.size());
+ if (assignment.manifestMetasToRewrite.size() == 1
+ || (parallelism != null && parallelism == 1)) {
+ for (ManifestFileMeta manifestMeta :
assignment.manifestMetasToRewrite) {
+ rewritten.add(rewriteDataManifest(assignment, manifestFile,
manifestMeta));
+ }
+ } else {
+ Function<ManifestFileMeta, List<RewrittenDataManifest>> rewriter =
+ manifestMeta ->
+ singletonList(
+ rewriteDataManifest(
+ assignment,
+
table.store().manifestFileFactory().create(),
+ manifestMeta));
+ try (CloseableBatchIterator<RewrittenDataManifest> results =
+ sequentialBatchedExecuteCloseable(
+ rewriter, assignment.manifestMetasToRewrite,
parallelism)) {
+ while (results.hasNext()) {
+ rewritten.add(results.next());
}
}
- checkState(
- hasRewrittenEntry,
- "Cannot find entries to reassign in planned manifest %s.",
- manifestMeta.fileName());
- rewrittenManifestMetas.put(manifestMeta.fileName(),
manifestFile.write(entries));
- fileCount += reassignedAddFileCount;
+ }
+
+ Map<String, List<ManifestFileMeta>> rewrittenManifestMetas = new
LinkedHashMap<>();
+ long fileCount = 0L;
+ for (RewrittenDataManifest manifest : rewritten) {
+ rewrittenManifestMetas.put(manifest.originalFileName,
manifest.replacements);
+ fileCount += manifest.reassignedAddFileCount;
}
return new RewrittenDataManifests(rewrittenManifestMetas, fileCount);
}
+ private RewrittenDataManifest rewriteDataManifest(
+ Assignment assignment, ManifestFile manifestFile, ManifestFileMeta
manifestMeta) {
+ beforeManifestRewrite.accept(manifestMeta);
+ List<ManifestEntry> entries =
+ manifestFile.read(manifestMeta.fileName(),
manifestMeta.fileSize());
+ 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++;
+ }
+ }
+ }
+ checkState(
+ hasRewrittenEntry,
+ "Cannot find entries to reassign in planned manifest %s.",
+ manifestMeta.fileName());
+ return new RewrittenDataManifest(
+ manifestMeta.fileName(), manifestFile.write(entries),
reassignedAddFileCount);
+ }
+
private void validatePlanningEntry(ManifestEntry entry) {
List<String> writeCols = entry.file().writeCols();
checkState(
@@ -902,6 +950,21 @@ public class DataEvolutionRowIdReassigner {
}
}
+ private static class RewrittenDataManifest {
+ private final String originalFileName;
+ private final List<ManifestFileMeta> replacements;
+ private final long reassignedAddFileCount;
+
+ private RewrittenDataManifest(
+ String originalFileName,
+ List<ManifestFileMeta> replacements,
+ long reassignedAddFileCount) {
+ this.originalFileName = originalFileName;
+ this.replacements = replacements;
+ this.reassignedAddFileCount = reassignedAddFileCount;
+ }
+ }
+
private static class CommittedAssignment {
private final Assignment assignment;
private final CommitAssignmentResult commitResult;
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 af0178bb27..0b6fbb6141 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
@@ -87,6 +87,11 @@ import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -1321,6 +1326,96 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(afterManifestFiles).doesNotContainAnyElementsOf(affectedManifests);
}
+ @Test
+ public void testReassignRewritesDataManifestsInParallel() throws Exception
{
+ FileStoreTable originalTable = createTableWithInterleavedPartitions();
+ assertThat(dataManifestFileNames(originalTable)).hasSizeGreaterThan(1);
+ FileStoreTable table =
+ originalTable.copy(
+
Collections.singletonMap(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "2"));
+
+ CountDownLatch twoRewritesStarted = new CountDownLatch(2);
+ CountDownLatch releaseRewrites = new CountDownLatch(1);
+ AtomicInteger startedRewriteCount = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<DataEvolutionRowIdReassigner.Result> future =
+ executor.submit(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {},
+ ignored -> {
+
startedRewriteCount.incrementAndGet();
+
twoRewritesStarted.countDown();
+ try {
+ if
(!releaseRewrites.await(
+ 30,
TimeUnit.SECONDS)) {
+ throw new
AssertionError(
+ "Timed out
waiting to release manifest rewrites.");
+ }
+ } catch
(InterruptedException e) {
+
Thread.currentThread().interrupt();
+ throw new
RuntimeException(e);
+ }
+ })
+
.reassign("test-parallel-manifest-rewrite"));
+
+ boolean rewritesOverlapped;
+ int startedBeforeRelease;
+ try {
+ rewritesOverlapped = twoRewritesStarted.await(10,
TimeUnit.SECONDS);
+ startedBeforeRelease = startedRewriteCount.get();
+ } finally {
+ releaseRewrites.countDown();
+ }
+
+ try {
+ DataEvolutionRowIdReassigner.Result result = future.get(30,
TimeUnit.SECONDS);
+ assertThat(result.fileCount).isEqualTo(5L);
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(5L, 6L, 7L))
+ .containsEntry("pt=b/", Arrays.asList(8L, 9L));
+ } finally {
+ executor.shutdownNow();
+ }
+ assertThat(rewritesOverlapped).isTrue();
+ assertThat(startedBeforeRelease).isEqualTo(2);
+ }
+
+ @Test
+ public void testReassignDoesNotCommitWhenParallelManifestRewriteFails()
throws Exception {
+ FileStoreTable originalTable = createTableWithInterleavedPartitions();
+ FileStoreTable table =
+ originalTable.copy(
+
Collections.singletonMap(CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), "2"));
+ Snapshot before = table.snapshotManager().latestSnapshot();
+ AtomicBoolean failureInjected = new AtomicBoolean();
+
+ assertThatThrownBy(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {},
+ ignored -> {
+ if
(failureInjected.compareAndSet(
+ false, true)) {
+ throw new
IllegalStateException(
+ "Injected
manifest rewrite failure.");
+ }
+ })
+
.reassign("test-failed-parallel-manifest-rewrite"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Injected manifest rewrite failure.");
+
+ assertThat(failureInjected).isTrue();
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(before.id());
+ assertThat(rowIdsByPartition(table))
+ .containsEntry("pt=a/", Arrays.asList(0L, 2L, 4L))
+ .containsEntry("pt=b/", Arrays.asList(1L, 3L));
+ }
+
@Test
public void testReassignDoesNotCompactManifests() throws Exception {
testReassignSkipsManifestOptimization(false);
