This is an automated email from the ASF dual-hosted git repository.
yuzelin 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 679878566c [core] Manifest read pool in expiration use
scanManifestParallelism (#9395)
679878566c is described below
commit 679878566c615ddf0f212b19ec8ff6c3832f4f74
Author: yuzelin <[email protected]>
AuthorDate: Wed Aug 26 15:02:29 2026 +0800
[core] Manifest read pool in expiration use scanManifestParallelism (#9395)
---
.../src/main/java/org/apache/paimon/AbstractFileStore.java | 9 ++++++---
.../org/apache/paimon/operation/ChangelogDeletion.java | 8 ++++++--
.../java/org/apache/paimon/operation/FileDeletionBase.java | 13 +++++++++----
.../java/org/apache/paimon/operation/SnapshotDeletion.java | 8 ++++++--
.../main/java/org/apache/paimon/operation/TagDeletion.java | 8 ++++++--
.../org/apache/paimon/operation/ExpireSnapshotsTest.java | 6 ++++--
.../java/org/apache/paimon/operation/FileDeletionTest.java | 14 ++++++++++----
7 files changed, 47 insertions(+), 19 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index 1bf2a59e3f..25c64029ad 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -344,7 +344,8 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
newStatsFileHandler(),
options.changelogProducer() !=
CoreOptions.ChangelogProducer.NONE,
options.cleanEmptyDirectories(),
- options.fileOperationThreadNum());
+ options.fileOperationThreadNum(),
+ options.scanManifestParallelism());
}
@Override
@@ -357,7 +358,8 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
newIndexFileHandler(),
newStatsFileHandler(),
options.cleanEmptyDirectories(),
- options.fileOperationThreadNum());
+ options.fileOperationThreadNum(),
+ options.scanManifestParallelism());
}
@Override
@@ -375,7 +377,8 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
newIndexFileHandler(),
newStatsFileHandler(),
options.cleanEmptyDirectories(),
- options.fileOperationThreadNum());
+ options.fileOperationThreadNum(),
+ options.scanManifestParallelism());
}
public abstract Comparator<InternalRow> newKeyComparator();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
index ba64eeecab..9689f272e2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
@@ -31,6 +31,8 @@ import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.FileStorePathFactory;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashSet;
@@ -49,7 +51,8 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int fileOperationThreadNum) {
+ int fileOperationThreadNum,
+ @Nullable Integer manifestReadParallelism) {
super(
fileIO,
pathFactory,
@@ -58,7 +61,8 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- fileOperationThreadNum);
+ fileOperationThreadNum,
+ manifestReadParallelism);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
index 50222d2c16..724c0354cd 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
@@ -44,6 +44,8 @@ import org.apache.paimon.utils.SnapshotManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
@@ -81,6 +83,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
private final Executor fileExecutor;
private final int fileOperationParallelism;
+ @Nullable private final Integer manifestReadParallelism;
protected boolean changelogDecoupled;
@@ -98,7 +101,8 @@ public abstract class FileDeletionBase<T extends Snapshot> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int fileOperationThreadNum) {
+ int fileOperationThreadNum,
+ @Nullable Integer manifestReadParallelism) {
this.fileIO = fileIO;
this.pathFactory = pathFactory;
this.manifestFile = manifestFile;
@@ -112,6 +116,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
fileOperationThreadNum > 0
? fileOperationThreadNum
: Runtime.getRuntime().availableProcessors();
+ this.manifestReadParallelism = manifestReadParallelism;
}
public Executor fileExecutor() {
@@ -257,7 +262,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
}
},
manifests,
- fileOperationParallelism);
+ manifestReadParallelism);
List<Path> dataFiles = new ArrayList<>();
DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
@@ -276,7 +281,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
return ManifestReadThreadPool.sequentialBatchedExecute(
manifest ->
manifestFile.readExpireFileEntries(manifest.fileName()),
manifests,
- fileOperationParallelism);
+ manifestReadParallelism);
}
public void cleanDataFiles(Collection<Path> dataFiles) {
@@ -463,7 +468,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
futures.add(
CompletableFuture.supplyAsync(
() -> manifestSkippingSet(skippingSnapshot),
-
ManifestReadThreadPool.getExecutorService(fileOperationParallelism)));
+
ManifestReadThreadPool.getExecutorService(manifestReadParallelism)));
}
Set<String> skippingSet = new HashSet<>();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java
index f97ec0474d..00bb731d03 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/SnapshotDeletion.java
@@ -29,6 +29,8 @@ import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.FileStorePathFactory;
+import javax.annotation.Nullable;
+
import java.util.List;
import java.util.Set;
import java.util.function.Predicate;
@@ -47,7 +49,8 @@ public class SnapshotDeletion extends
FileDeletionBase<Snapshot> {
StatsFileHandler statsFileHandler,
boolean produceChangelog,
boolean cleanEmptyDirectories,
- int fileOperationThreadNum) {
+ int fileOperationThreadNum,
+ @Nullable Integer manifestReadParallelism) {
super(
fileIO,
pathFactory,
@@ -56,7 +59,8 @@ public class SnapshotDeletion extends
FileDeletionBase<Snapshot> {
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- fileOperationThreadNum);
+ fileOperationThreadNum,
+ manifestReadParallelism);
this.produceChangelog = produceChangelog;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
index ff3e05f9aa..e6f9d9d673 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java
@@ -35,6 +35,8 @@ import org.apache.paimon.utils.FileStorePathFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
@@ -57,7 +59,8 @@ public class TagDeletion extends FileDeletionBase<Snapshot> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int fileOperationThreadNum) {
+ int fileOperationThreadNum,
+ @Nullable Integer manifestReadParallelism) {
super(
fileIO,
pathFactory,
@@ -66,7 +69,8 @@ public class TagDeletion extends FileDeletionBase<Snapshot> {
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- fileOperationThreadNum);
+ fileOperationThreadNum,
+ manifestReadParallelism);
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
index 46c7626f11..946144af58 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
@@ -1353,7 +1353,8 @@ public class ExpireSnapshotsTest {
store.newStatsFileHandler(),
store.options().changelogProducer() !=
CoreOptions.ChangelogProducer.NONE,
store.options().cleanEmptyDirectories(),
- store.options().fileOperationThreadNum());
+ store.options().fileOperationThreadNum(),
+ store.options().scanManifestParallelism());
this.minBlockedSnapshotId = minBlockedSnapshotId;
this.maxBlockedSnapshotId = maxBlockedSnapshotId;
}
@@ -1426,7 +1427,8 @@ public class ExpireSnapshotsTest {
store.newStatsFileHandler(),
store.options().changelogProducer() !=
CoreOptions.ChangelogProducer.NONE,
store.options().cleanEmptyDirectories(),
- store.options().fileOperationThreadNum());
+ store.options().fileOperationThreadNum(),
+ store.options().scanManifestParallelism());
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
index ee4eccd1d5..49345b0e98 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
@@ -56,6 +56,8 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
@@ -748,7 +750,8 @@ public class FileDeletionTest {
store.newStatsFileHandler(),
store.options().changelogProducer() !=
CoreOptions.ChangelogProducer.NONE,
store.options().cleanEmptyDirectories(),
- store.options().fileOperationThreadNum());
+ store.options().fileOperationThreadNum(),
+ store.options().scanManifestParallelism());
ExpireSnapshots expireSnapshots =
new ExpireSnapshotsImpl(
@@ -813,7 +816,8 @@ public class FileDeletionTest {
store.newStatsFileHandler(),
store.options().changelogProducer() !=
CoreOptions.ChangelogProducer.NONE,
store.options().cleanEmptyDirectories(),
- store.options().fileOperationThreadNum());
+ store.options().fileOperationThreadNum(),
+ store.options().scanManifestParallelism());
ExpireSnapshots expireSnapshots =
new ExpireSnapshotsImpl(
snapshotManager, changelogManager, snapshotDeletion,
tagManager);
@@ -952,7 +956,8 @@ public class FileDeletionTest {
StatsFileHandler statsFileHandler,
boolean produceChangelog,
boolean cleanEmptyDirectories,
- int deleteFileThreadNum) {
+ int deleteFileThreadNum,
+ @Nullable Integer scanManifestParallelism) {
super(
fileIO,
pathFactory,
@@ -962,7 +967,8 @@ public class FileDeletionTest {
statsFileHandler,
produceChangelog,
cleanEmptyDirectories,
- deleteFileThreadNum);
+ deleteFileThreadNum,
+ scanManifestParallelism);
}
@Override
