From b8fd6724c51478e0356e0f9f0e98baaac4750551 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 16:57:57 +0800 Subject: [PATCH 01/10] [core] Snapshot expiration deletes data and changelog files in batches --- .../org/apache/paimon/AbstractFileStore.java | 9 +- .../paimon/operation/ChangelogDeletion.java | 8 +- .../paimon/operation/FileDeletionBase.java | 17 ++-- .../paimon/operation/SnapshotDeletion.java | 8 +- .../apache/paimon/operation/TagDeletion.java | 8 +- .../paimon/table/ExpireSnapshotsImpl.java | 35 +++++-- .../paimon/operation/ExpireSnapshotsTest.java | 97 ++++++++++++++++++- .../paimon/operation/FileDeletionTest.java | 14 +-- 8 files changed, 140 insertions(+), 56 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 25c64029adf7..1bf2a59e3f9d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -344,8 +344,7 @@ public SnapshotDeletion newSnapshotDeletion() { newStatsFileHandler(), options.changelogProducer() != CoreOptions.ChangelogProducer.NONE, options.cleanEmptyDirectories(), - options.fileOperationThreadNum(), - options.scanManifestParallelism()); + options.fileOperationThreadNum()); } @Override @@ -358,8 +357,7 @@ public ChangelogDeletion newChangelogDeletion() { newIndexFileHandler(), newStatsFileHandler(), options.cleanEmptyDirectories(), - options.fileOperationThreadNum(), - options.scanManifestParallelism()); + options.fileOperationThreadNum()); } @Override @@ -377,8 +375,7 @@ public TagDeletion newTagDeletion() { newIndexFileHandler(), newStatsFileHandler(), options.cleanEmptyDirectories(), - options.fileOperationThreadNum(), - options.scanManifestParallelism()); + options.fileOperationThreadNum()); } public abstract Comparator 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 9689f272e2eb..ba64eeecabce 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,8 +31,6 @@ 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; @@ -51,8 +49,7 @@ public ChangelogDeletion( IndexFileHandler indexFileHandler, StatsFileHandler statsFileHandler, boolean cleanEmptyDirectories, - int fileOperationThreadNum, - @Nullable Integer manifestReadParallelism) { + int fileOperationThreadNum) { super( fileIO, pathFactory, @@ -61,8 +58,7 @@ public ChangelogDeletion( indexFileHandler, statsFileHandler, cleanEmptyDirectories, - fileOperationThreadNum, - manifestReadParallelism); + fileOperationThreadNum); } @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 724c0354cdaa..44c18d76f762 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,8 +44,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nullable; - import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; @@ -83,7 +81,6 @@ public abstract class FileDeletionBase { private final Executor fileExecutor; private final int fileOperationParallelism; - @Nullable private final Integer manifestReadParallelism; protected boolean changelogDecoupled; @@ -101,8 +98,7 @@ public FileDeletionBase( IndexFileHandler indexFileHandler, StatsFileHandler statsFileHandler, boolean cleanEmptyDirectories, - int fileOperationThreadNum, - @Nullable Integer manifestReadParallelism) { + int fileOperationThreadNum) { this.fileIO = fileIO; this.pathFactory = pathFactory; this.manifestFile = manifestFile; @@ -116,13 +112,16 @@ public FileDeletionBase( fileOperationThreadNum > 0 ? fileOperationThreadNum : Runtime.getRuntime().availableProcessors(); - this.manifestReadParallelism = manifestReadParallelism; } public Executor fileExecutor() { return fileExecutor; } + public int fileOperationParallelism() { + return fileOperationParallelism; + } + /** * Clean data files that will not be used anymore in the snapshot. * @@ -262,7 +261,7 @@ public List planAddedInChangelogManifest(T snapshot) { } }, manifests, - manifestReadParallelism); + fileOperationParallelism); List dataFiles = new ArrayList<>(); DataFilePathFactories factories = new DataFilePathFactories(pathFactory); @@ -281,7 +280,7 @@ private Iterable readExpireFileEntries(List m return ManifestReadThreadPool.sequentialBatchedExecute( manifest -> manifestFile.readExpireFileEntries(manifest.fileName()), manifests, - manifestReadParallelism); + fileOperationParallelism); } public void cleanDataFiles(Collection dataFiles) { @@ -468,7 +467,7 @@ public Set manifestSkippingSet(List skippingSnapshots) { futures.add( CompletableFuture.supplyAsync( () -> manifestSkippingSet(skippingSnapshot), - ManifestReadThreadPool.getExecutorService(manifestReadParallelism))); + ManifestReadThreadPool.getExecutorService(fileOperationParallelism))); } Set 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 00bb731d03a4..f97ec0474d36 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,8 +29,6 @@ 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; @@ -49,8 +47,7 @@ public SnapshotDeletion( StatsFileHandler statsFileHandler, boolean produceChangelog, boolean cleanEmptyDirectories, - int fileOperationThreadNum, - @Nullable Integer manifestReadParallelism) { + int fileOperationThreadNum) { super( fileIO, pathFactory, @@ -59,8 +56,7 @@ public SnapshotDeletion( indexFileHandler, statsFileHandler, cleanEmptyDirectories, - fileOperationThreadNum, - manifestReadParallelism); + fileOperationThreadNum); 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 e6f9d9d6730c..ff3e05f9aa4c 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,8 +35,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nullable; - import java.io.IOException; import java.util.Collection; import java.util.HashMap; @@ -59,8 +57,7 @@ public TagDeletion( IndexFileHandler indexFileHandler, StatsFileHandler statsFileHandler, boolean cleanEmptyDirectories, - int fileOperationThreadNum, - @Nullable Integer manifestReadParallelism) { + int fileOperationThreadNum) { super( fileIO, pathFactory, @@ -69,8 +66,7 @@ public TagDeletion( indexFileHandler, statsFileHandler, cleanEmptyDirectories, - fileOperationThreadNum, - manifestReadParallelism); + fileOperationThreadNum); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 32124bfc2aac..f7c170301ec2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -42,9 +42,11 @@ import java.util.Collection; import java.util.Comparator; import java.util.HashMap; +import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Queue; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -67,6 +69,7 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { private final ConsumerManager consumerManager; private final SnapshotDeletion snapshotDeletion; private final Executor fileExecutor; + private final int fileOperationParallelism; private final TagManager tagManager; private ExpireConfig expireConfig; @@ -88,6 +91,7 @@ public ExpireSnapshotsImpl( this.tagManager = tagManager; this.expireConfig = ExpireConfig.builder().build(); this.fileExecutor = snapshotDeletion.fileExecutor(); + this.fileOperationParallelism = snapshotDeletion.fileOperationParallelism(); } @VisibleForTesting @@ -207,12 +211,11 @@ private int innerExpireUntil(long earliestId, long endExclusiveId) // delete merge tree files // deleted merge tree files in a snapshot are not used by the next snapshot, so the range of // id should be (beginInclusiveId, endExclusiveId] - snapshotDeletion.cleanDataFiles( - collectDataFilesToDelete(snapshotsIncludingEnd, taggedSnapshots, beginInclusiveId)); + cleanDataFiles(snapshotsIncludingEnd, taggedSnapshots, beginInclusiveId); // delete changelog files if (!expireConfig.isChangelogDecoupled()) { - snapshotDeletion.cleanDataFiles(collectChangelogFilesToDelete(snapshotsExcludingEnd)); + planAndCleanChangelogFiles(snapshotsExcludingEnd); } // data files and changelog files in bucket directories has been deleted @@ -262,7 +265,7 @@ private int innerExpireUntil(long earliestId, long endExclusiveId) return snapshotsExcludingEnd.size(); } - private Collection collectDataFilesToDelete( + private void cleanDataFiles( List snapshotsIncludingEnd, List taggedSnapshots, long beginInclusiveId) @@ -287,7 +290,7 @@ private Collection collectDataFilesToDelete( Map>> skippers = collectTagSkippers(tags.values()); Predicate deleteAll = entry -> false; - List>> futures = new ArrayList<>(); + Queue>> futures = new LinkedList<>(); for (Snapshot snapshot : snapshotsIncludingEnd) { long id = snapshot.id(); if (id == beginInclusiveId) { @@ -315,8 +318,11 @@ private Collection collectDataFilesToDelete( snapshotDeletion.planDeletedInDeltaManifest( snapshot, skipper.get()), fileExecutor)); + if (futures.size() >= fileOperationParallelism) { + collectAndClean(futures); + } } - return flatten(getAll(futures)); + collectAndClean(futures); } private Map>> collectTagSkippers( @@ -350,9 +356,9 @@ private Map>> collectTagSkippers( return skippers; } - private Collection collectChangelogFilesToDelete(List snapshots) + private void planAndCleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { - List>> futures = new ArrayList<>(); + Queue>> futures = new LinkedList<>(); for (Snapshot snapshot : snapshots) { if (LOG.isDebugEnabled()) { LOG.debug("Ready to delete changelog files from snapshot #{}", snapshot.id()); @@ -362,9 +368,20 @@ private Collection collectChangelogFilesToDelete(List snapshots) CompletableFuture.supplyAsync( () -> snapshotDeletion.planAddedInChangelogManifest(snapshot), fileExecutor)); + if (futures.size() >= fileOperationParallelism) { + collectAndClean(futures); + } } } - return flatten(getAll(futures)); + collectAndClean(futures); + } + + private void collectAndClean(Queue>> futures) + throws ExecutionException, InterruptedException { + while (!futures.isEmpty()) { + CompletableFuture> future = futures.remove(); + snapshotDeletion.cleanDataFiles(future.get()); + } } private Collection collectManifestDeletionTasks( 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 946144af5801..6dbc5b325eff 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 @@ -672,6 +672,37 @@ public void testExpirePlansDataFilesConcurrently() throws Exception { store.assertCleaned(); } + @Test + public void testExpireConsumesDataFilePlansInBoundedBatches() throws Exception { + int fileOperationParallelism = 2; + store.options() + .toConfiguration() + .set(CoreOptions.FILE_OPERATION_THREAD_NUM, fileOperationParallelism); + + List allData = new ArrayList<>(); + List snapshotPositions = new ArrayList<>(); + commit(8, allData, snapshotPositions); + int latestSnapshotId = requireNonNull(snapshotManager.latestSnapshotId()).intValue(); + for (int i = 1; i <= latestSnapshotId; i++) { + rewriteSnapshotTime(i, 0); + } + + TrackingSnapshotDeletion snapshotDeletion = new TrackingSnapshotDeletion(store); + ExpireSnapshotsImpl expire = + newExpireWithSnapshotDeletion(store, snapshotManager, snapshotDeletion); + expire.config(expireAllButLatestConfig()); + expire.setCurrentTimeMillis(() -> 1000L); + + expire.expire(); + + assertThat(snapshotDeletion.maxPendingDataFilePlans()) + .isLessThanOrEqualTo(fileOperationParallelism); + assertThat(snapshotDeletion.pendingDataFilePlans()).isZero(); + assertThat(snapshotDeletion.dataFileCleanCalls()).isGreaterThan(1); + assertSnapshot(latestSnapshotId, allData, snapshotPositions); + store.assertCleaned(); + } + @Test public void testExpirePlansChangelogFilesConcurrently() throws Exception { TestFileStore inputStore = createStore(CoreOptions.ChangelogProducer.INPUT); @@ -1353,8 +1384,7 @@ private BlockingSnapshotDeletion( store.newStatsFileHandler(), store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, store.options().cleanEmptyDirectories(), - store.options().fileOperationThreadNum(), - store.options().scanManifestParallelism()); + store.options().fileOperationThreadNum()); this.minBlockedSnapshotId = minBlockedSnapshotId; this.maxBlockedSnapshotId = maxBlockedSnapshotId; } @@ -1427,8 +1457,7 @@ private CapturingSnapshotDeletion(TestFileStore store) { store.newStatsFileHandler(), store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, store.options().cleanEmptyDirectories(), - store.options().fileOperationThreadNum(), - store.options().scanManifestParallelism()); + store.options().fileOperationThreadNum()); } @Override @@ -1451,6 +1480,66 @@ private void reset() { } } + private static class TrackingSnapshotDeletion extends SnapshotDeletion { + + private final AtomicInteger pendingDataFilePlans = new AtomicInteger(); + private final AtomicInteger maxPendingDataFilePlans = new AtomicInteger(); + private final AtomicInteger dataFileCleanCalls = new AtomicInteger(); + + private TrackingSnapshotDeletion(TestFileStore store) { + super( + store.fileIO(), + store.pathFactory(), + store.manifestFileFactory().create(), + store.manifestListFactory().create(), + store.newIndexFileHandler(), + store.newStatsFileHandler(), + store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, + store.options().cleanEmptyDirectories(), + store.options().fileOperationThreadNum()); + } + + @Override + public List planDeletedInDeltaManifest( + Snapshot snapshot, Predicate skipper) { + List paths = super.planDeletedInDeltaManifest(snapshot, skipper); + int pending = pendingDataFilePlans.incrementAndGet(); + maxPendingDataFilePlans.accumulateAndGet(pending, Math::max); + return new TrackedDataFiles(paths); + } + + @Override + public void cleanDataFiles(Collection dataFiles) { + try { + super.cleanDataFiles(dataFiles); + } finally { + if (dataFiles instanceof TrackedDataFiles) { + pendingDataFilePlans.decrementAndGet(); + dataFileCleanCalls.incrementAndGet(); + } + } + } + + private int pendingDataFilePlans() { + return pendingDataFilePlans.get(); + } + + private int maxPendingDataFilePlans() { + return maxPendingDataFilePlans.get(); + } + + private int dataFileCleanCalls() { + return dataFileCleanCalls.get(); + } + } + + private static class TrackedDataFiles extends ArrayList { + + private TrackedDataFiles(Collection paths) { + super(paths); + } + } + private static class ConcurrentCallTracker { private final String timeoutMessage; 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 49345b0e983a..ee4eccd1d545 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,8 +56,6 @@ 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; @@ -750,8 +748,7 @@ public void testDataFileSkippingSetException() throws Exception { store.newStatsFileHandler(), store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, store.options().cleanEmptyDirectories(), - store.options().fileOperationThreadNum(), - store.options().scanManifestParallelism()); + store.options().fileOperationThreadNum()); ExpireSnapshots expireSnapshots = new ExpireSnapshotsImpl( @@ -816,8 +813,7 @@ public void testManifestFileSkippingSetException() throws Exception { store.newStatsFileHandler(), store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, store.options().cleanEmptyDirectories(), - store.options().fileOperationThreadNum(), - store.options().scanManifestParallelism()); + store.options().fileOperationThreadNum()); ExpireSnapshots expireSnapshots = new ExpireSnapshotsImpl( snapshotManager, changelogManager, snapshotDeletion, tagManager); @@ -956,8 +952,7 @@ public TestSnapshotDeletion( StatsFileHandler statsFileHandler, boolean produceChangelog, boolean cleanEmptyDirectories, - int deleteFileThreadNum, - @Nullable Integer scanManifestParallelism) { + int deleteFileThreadNum) { super( fileIO, pathFactory, @@ -967,8 +962,7 @@ public TestSnapshotDeletion( statsFileHandler, produceChangelog, cleanEmptyDirectories, - deleteFileThreadNum, - scanManifestParallelism); + deleteFileThreadNum); } @Override From cc334f270ecd2ceec598afdb112ce575b2875af0 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 17:07:34 +0800 Subject: [PATCH 02/10] fix --- .../paimon/operation/ExpireSnapshotsTest.java | 91 ------------------- 1 file changed, 91 deletions(-) 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 6dbc5b325eff..46c7626f1108 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 @@ -672,37 +672,6 @@ public void testExpirePlansDataFilesConcurrently() throws Exception { store.assertCleaned(); } - @Test - public void testExpireConsumesDataFilePlansInBoundedBatches() throws Exception { - int fileOperationParallelism = 2; - store.options() - .toConfiguration() - .set(CoreOptions.FILE_OPERATION_THREAD_NUM, fileOperationParallelism); - - List allData = new ArrayList<>(); - List snapshotPositions = new ArrayList<>(); - commit(8, allData, snapshotPositions); - int latestSnapshotId = requireNonNull(snapshotManager.latestSnapshotId()).intValue(); - for (int i = 1; i <= latestSnapshotId; i++) { - rewriteSnapshotTime(i, 0); - } - - TrackingSnapshotDeletion snapshotDeletion = new TrackingSnapshotDeletion(store); - ExpireSnapshotsImpl expire = - newExpireWithSnapshotDeletion(store, snapshotManager, snapshotDeletion); - expire.config(expireAllButLatestConfig()); - expire.setCurrentTimeMillis(() -> 1000L); - - expire.expire(); - - assertThat(snapshotDeletion.maxPendingDataFilePlans()) - .isLessThanOrEqualTo(fileOperationParallelism); - assertThat(snapshotDeletion.pendingDataFilePlans()).isZero(); - assertThat(snapshotDeletion.dataFileCleanCalls()).isGreaterThan(1); - assertSnapshot(latestSnapshotId, allData, snapshotPositions); - store.assertCleaned(); - } - @Test public void testExpirePlansChangelogFilesConcurrently() throws Exception { TestFileStore inputStore = createStore(CoreOptions.ChangelogProducer.INPUT); @@ -1480,66 +1449,6 @@ private void reset() { } } - private static class TrackingSnapshotDeletion extends SnapshotDeletion { - - private final AtomicInteger pendingDataFilePlans = new AtomicInteger(); - private final AtomicInteger maxPendingDataFilePlans = new AtomicInteger(); - private final AtomicInteger dataFileCleanCalls = new AtomicInteger(); - - private TrackingSnapshotDeletion(TestFileStore store) { - super( - store.fileIO(), - store.pathFactory(), - store.manifestFileFactory().create(), - store.manifestListFactory().create(), - store.newIndexFileHandler(), - store.newStatsFileHandler(), - store.options().changelogProducer() != CoreOptions.ChangelogProducer.NONE, - store.options().cleanEmptyDirectories(), - store.options().fileOperationThreadNum()); - } - - @Override - public List planDeletedInDeltaManifest( - Snapshot snapshot, Predicate skipper) { - List paths = super.planDeletedInDeltaManifest(snapshot, skipper); - int pending = pendingDataFilePlans.incrementAndGet(); - maxPendingDataFilePlans.accumulateAndGet(pending, Math::max); - return new TrackedDataFiles(paths); - } - - @Override - public void cleanDataFiles(Collection dataFiles) { - try { - super.cleanDataFiles(dataFiles); - } finally { - if (dataFiles instanceof TrackedDataFiles) { - pendingDataFilePlans.decrementAndGet(); - dataFileCleanCalls.incrementAndGet(); - } - } - } - - private int pendingDataFilePlans() { - return pendingDataFilePlans.get(); - } - - private int maxPendingDataFilePlans() { - return maxPendingDataFilePlans.get(); - } - - private int dataFileCleanCalls() { - return dataFileCleanCalls.get(); - } - } - - private static class TrackedDataFiles extends ArrayList { - - private TrackedDataFiles(Collection paths) { - super(paths); - } - } - private static class ConcurrentCallTracker { private final String timeoutMessage; From 3ca61cd5e9cb42fb95cdaa39b75a93219274b7d0 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 17:27:58 +0800 Subject: [PATCH 03/10] fix --- .../java/org/apache/paimon/AbstractFileStore.java | 9 ++++++--- .../apache/paimon/operation/ChangelogDeletion.java | 8 ++++++-- .../apache/paimon/operation/FileDeletionBase.java | 13 +++++++++---- .../apache/paimon/operation/SnapshotDeletion.java | 8 ++++++-- .../org/apache/paimon/operation/TagDeletion.java | 8 ++++++-- .../paimon/operation/ExpireSnapshotsTest.java | 6 ++++-- .../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 1bf2a59e3f9d..25c64029adf7 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 @@ public SnapshotDeletion newSnapshotDeletion() { newStatsFileHandler(), options.changelogProducer() != CoreOptions.ChangelogProducer.NONE, options.cleanEmptyDirectories(), - options.fileOperationThreadNum()); + options.fileOperationThreadNum(), + options.scanManifestParallelism()); } @Override @@ -357,7 +358,8 @@ public ChangelogDeletion newChangelogDeletion() { newIndexFileHandler(), newStatsFileHandler(), options.cleanEmptyDirectories(), - options.fileOperationThreadNum()); + options.fileOperationThreadNum(), + options.scanManifestParallelism()); } @Override @@ -375,7 +377,8 @@ public TagDeletion newTagDeletion() { newIndexFileHandler(), newStatsFileHandler(), options.cleanEmptyDirectories(), - options.fileOperationThreadNum()); + options.fileOperationThreadNum(), + options.scanManifestParallelism()); } public abstract Comparator 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 ba64eeecabce..9689f272e2eb 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.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 ChangelogDeletion( IndexFileHandler indexFileHandler, StatsFileHandler statsFileHandler, boolean cleanEmptyDirectories, - int fileOperationThreadNum) { + int fileOperationThreadNum, + @Nullable Integer manifestReadParallelism) { super( fileIO, pathFactory, @@ -58,7 +61,8 @@ public ChangelogDeletion( 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 44c18d76f762..6fc67c49bce1 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.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 { private final Executor fileExecutor; private final int fileOperationParallelism; + @Nullable private final Integer manifestReadParallelism; protected boolean changelogDecoupled; @@ -98,7 +101,8 @@ public FileDeletionBase( 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 FileDeletionBase( fileOperationThreadNum > 0 ? fileOperationThreadNum : Runtime.getRuntime().availableProcessors(); + this.manifestReadParallelism = manifestReadParallelism; } public Executor fileExecutor() { @@ -261,7 +266,7 @@ public List planAddedInChangelogManifest(T snapshot) { } }, manifests, - fileOperationParallelism); + manifestReadParallelism); List dataFiles = new ArrayList<>(); DataFilePathFactories factories = new DataFilePathFactories(pathFactory); @@ -280,7 +285,7 @@ private Iterable readExpireFileEntries(List m return ManifestReadThreadPool.sequentialBatchedExecute( manifest -> manifestFile.readExpireFileEntries(manifest.fileName()), manifests, - fileOperationParallelism); + manifestReadParallelism); } public void cleanDataFiles(Collection dataFiles) { @@ -467,7 +472,7 @@ public Set manifestSkippingSet(List skippingSnapshots) { futures.add( CompletableFuture.supplyAsync( () -> manifestSkippingSet(skippingSnapshot), - ManifestReadThreadPool.getExecutorService(fileOperationParallelism))); + ManifestReadThreadPool.getExecutorService(manifestReadParallelism))); } Set 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 f97ec0474d36..00bb731d03a4 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.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 SnapshotDeletion( StatsFileHandler statsFileHandler, boolean produceChangelog, boolean cleanEmptyDirectories, - int fileOperationThreadNum) { + int fileOperationThreadNum, + @Nullable Integer manifestReadParallelism) { super( fileIO, pathFactory, @@ -56,7 +59,8 @@ public SnapshotDeletion( 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 ff3e05f9aa4c..e6f9d9d6730c 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.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 TagDeletion( IndexFileHandler indexFileHandler, StatsFileHandler statsFileHandler, boolean cleanEmptyDirectories, - int fileOperationThreadNum) { + int fileOperationThreadNum, + @Nullable Integer manifestReadParallelism) { super( fileIO, pathFactory, @@ -66,7 +69,8 @@ public TagDeletion( 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 46c7626f1108..946144af5801 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 @@ private BlockingSnapshotDeletion( 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 @@ private CapturingSnapshotDeletion(TestFileStore store) { 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 ee4eccd1d545..49345b0e983a 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.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 void testDataFileSkippingSetException() throws Exception { 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 void testManifestFileSkippingSetException() throws Exception { 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 TestSnapshotDeletion( StatsFileHandler statsFileHandler, boolean produceChangelog, boolean cleanEmptyDirectories, - int deleteFileThreadNum) { + int deleteFileThreadNum, + @Nullable Integer scanManifestParallelism) { super( fileIO, pathFactory, @@ -962,7 +967,8 @@ public TestSnapshotDeletion( statsFileHandler, produceChangelog, cleanEmptyDirectories, - deleteFileThreadNum); + deleteFileThreadNum, + scanManifestParallelism); } @Override From cf2648d73d8d43dc1889b269e58ce50c8d97fbd8 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 17:28:56 +0800 Subject: [PATCH 04/10] fix --- .../java/org/apache/paimon/table/ExpireSnapshotsImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index f7c170301ec2..3a7198d7cd92 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -215,7 +215,7 @@ private int innerExpireUntil(long earliestId, long endExclusiveId) // delete changelog files if (!expireConfig.isChangelogDecoupled()) { - planAndCleanChangelogFiles(snapshotsExcludingEnd); + cleanChangelogFiles(snapshotsExcludingEnd); } // data files and changelog files in bucket directories has been deleted @@ -356,7 +356,7 @@ private Map>> collectTagSkippers( return skippers; } - private void planAndCleanChangelogFiles(List snapshots) + private void cleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { Queue>> futures = new LinkedList<>(); for (Snapshot snapshot : snapshots) { From b6fe8e58e1d553fb4d6cd55c210d601899c5e422 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 18:34:04 +0800 Subject: [PATCH 05/10] fix --- .../paimon/table/AbstractFileStoreTable.java | 3 ++- .../paimon/table/ExpireSnapshotsImpl.java | 16 +++++++++++----- .../java/org/apache/paimon/TestFileStore.java | 6 ++++-- .../paimon/operation/ExpireSnapshotsTest.java | 12 +++++++++--- .../paimon/operation/FileDeletionTest.java | 18 +++++++++++++++--- 5 files changed, 41 insertions(+), 14 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 88ee3f6a4b65..97058f2b8e03 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -450,7 +450,8 @@ public ExpireSnapshots newExpireSnapshots() { snapshotManager(), changelogManager(), store().newSnapshotDeletion(), - store().newTagManager()); + store().newTagManager(), + store().options().scanManifestParallelism()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 3a7198d7cd92..c51e60b672b2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -35,6 +35,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.FileNotFoundException; import java.io.IOException; import java.io.UncheckedIOException; @@ -69,8 +71,8 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { private final ConsumerManager consumerManager; private final SnapshotDeletion snapshotDeletion; private final Executor fileExecutor; - private final int fileOperationParallelism; private final TagManager tagManager; + private final int filesExpireBatchSize; private ExpireConfig expireConfig; private Supplier currentTimeMillis = System::currentTimeMillis; @@ -79,7 +81,8 @@ public ExpireSnapshotsImpl( SnapshotManager snapshotManager, ChangelogManager changelogManager, SnapshotDeletion snapshotDeletion, - TagManager tagManager) { + TagManager tagManager, + @Nullable Integer scanManifestParallelism) { this.snapshotManager = snapshotManager; this.changelogManager = changelogManager; this.consumerManager = @@ -91,7 +94,10 @@ public ExpireSnapshotsImpl( this.tagManager = tagManager; this.expireConfig = ExpireConfig.builder().build(); this.fileExecutor = snapshotDeletion.fileExecutor(); - this.fileOperationParallelism = snapshotDeletion.fileOperationParallelism(); + this.filesExpireBatchSize = + scanManifestParallelism == null + ? Runtime.getRuntime().availableProcessors() + : scanManifestParallelism; } @VisibleForTesting @@ -318,7 +324,7 @@ private void cleanDataFiles( snapshotDeletion.planDeletedInDeltaManifest( snapshot, skipper.get()), fileExecutor)); - if (futures.size() >= fileOperationParallelism) { + if (futures.size() >= filesExpireBatchSize) { collectAndClean(futures); } } @@ -368,7 +374,7 @@ private void cleanChangelogFiles(List snapshots) CompletableFuture.supplyAsync( () -> snapshotDeletion.planAddedInChangelogManifest(snapshot), fileExecutor)); - if (futures.size() >= fileOperationParallelism) { + if (futures.size() >= filesExpireBatchSize) { collectAndClean(futures); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java index b670ffa48f0b..178905917c07 100644 --- a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java +++ b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java @@ -175,7 +175,8 @@ public ExpireSnapshots newExpire(int numRetainedMin, int numRetainedMax, long mi snapshotManager(), changelogManager(), newSnapshotDeletion(), - new TagManager(fileIO, options.path())) + new TagManager(fileIO, options.path()), + null) .config( ExpireConfig.builder() .snapshotRetainMax(numRetainedMax) @@ -189,7 +190,8 @@ public ExpireSnapshots newExpire(ExpireConfig expireConfig) { snapshotManager(), changelogManager(), newSnapshotDeletion(), - new TagManager(fileIO, options.path())) + new TagManager(fileIO, options.path()), + null) .config(expireConfig); } 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 946144af5801..0d2538e82671 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 @@ -635,7 +635,8 @@ public void testExpireCollectsSnapshotsConcurrently() throws Exception { blockingSnapshotManager, changelogManager, store.newSnapshotDeletion(), - store.newTagManager()); + store.newTagManager(), + store.options().scanManifestParallelism()); expire.expireUntil(1, latestSnapshotId); @@ -961,7 +962,8 @@ public void testExpireWithTimeDoesNotReadProtectedRange() throws Exception { failingSnapshotManager, changelogManager, store.newSnapshotDeletion(), - store.newTagManager()); + store.newTagManager(), + store.options().scanManifestParallelism()); expire.config(config); expire.setCurrentTimeMillis(() -> 6000L); @@ -1217,7 +1219,11 @@ private ExpireSnapshotsImpl newExpireWithSnapshotDeletion( SnapshotManager snapshotManager, SnapshotDeletion snapshotDeletion) { return new ExpireSnapshotsImpl( - snapshotManager, store.changelogManager(), snapshotDeletion, store.newTagManager()); + snapshotManager, + store.changelogManager(), + snapshotDeletion, + store.newTagManager(), + store.options().scanManifestParallelism()); } private void rewriteSnapshotTime(long snapshotId, long newTimeMillis) throws IOException { 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 49345b0e983a..31ba29a2b9df 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 @@ -682,7 +682,11 @@ public void testExpireWithDeletingTags() throws Exception { // result: exist A & B (because of tag2) ExpireSnapshots expireSnapshots = new ExpireSnapshotsImpl( - snapshotManager, changelogManager, store.newSnapshotDeletion(), tagManager); + snapshotManager, + changelogManager, + store.newSnapshotDeletion(), + tagManager, + store.options().scanManifestParallelism()); expireSnapshots .config( ExpireConfig.builder() @@ -755,7 +759,11 @@ public void testDataFileSkippingSetException() throws Exception { ExpireSnapshots expireSnapshots = new ExpireSnapshotsImpl( - snapshotManager, changelogManager, snapshotDeletion, tagManager); + snapshotManager, + changelogManager, + snapshotDeletion, + tagManager, + store.options().scanManifestParallelism()); snapshotDeletion.readMergedDataFilesThrowException = true; expireSnapshots .config( @@ -820,7 +828,11 @@ public void testManifestFileSkippingSetException() throws Exception { store.options().scanManifestParallelism()); ExpireSnapshots expireSnapshots = new ExpireSnapshotsImpl( - snapshotManager, changelogManager, snapshotDeletion, tagManager); + snapshotManager, + changelogManager, + snapshotDeletion, + tagManager, + store.options().scanManifestParallelism()); snapshotDeletion.manifestSkippingSetThrowException = true; expireSnapshots .config( From ee436d770400a4b8cd22aeedf724c03ce0e5578a Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 18:44:57 +0800 Subject: [PATCH 06/10] fix --- .../paimon/table/ExpireSnapshotsImpl.java | 50 +++++++++++-------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index c51e60b672b2..130313120c15 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -44,11 +44,9 @@ import java.util.Collection; import java.util.Comparator; import java.util.HashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.Queue; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -95,7 +93,7 @@ public ExpireSnapshotsImpl( this.expireConfig = ExpireConfig.builder().build(); this.fileExecutor = snapshotDeletion.fileExecutor(); this.filesExpireBatchSize = - scanManifestParallelism == null + scanManifestParallelism == null || scanManifestParallelism <= 0 ? Runtime.getRuntime().availableProcessors() : scanManifestParallelism; } @@ -276,6 +274,21 @@ private void cleanDataFiles( List taggedSnapshots, long beginInclusiveId) throws ExecutionException, InterruptedException { + for (int from = 0; from < snapshotsIncludingEnd.size(); from += filesExpireBatchSize) { + int to = Math.min(from + filesExpireBatchSize, snapshotsIncludingEnd.size()); + snapshotDeletion.cleanDataFiles( + collectDataFilesToDelete( + snapshotsIncludingEnd.subList(from, to), + taggedSnapshots, + beginInclusiveId)); + } + } + + private Collection collectDataFilesToDelete( + List snapshotsIncludingEnd, + List taggedSnapshots, + long beginInclusiveId) + throws ExecutionException, InterruptedException { Map tagIdBySnapshotId = new HashMap<>(); Map tags = new HashMap<>(); int tagIndex = -1; @@ -296,7 +309,7 @@ private void cleanDataFiles( Map>> skippers = collectTagSkippers(tags.values()); Predicate deleteAll = entry -> false; - Queue>> futures = new LinkedList<>(); + List>> futures = new ArrayList<>(); for (Snapshot snapshot : snapshotsIncludingEnd) { long id = snapshot.id(); if (id == beginInclusiveId) { @@ -324,11 +337,8 @@ private void cleanDataFiles( snapshotDeletion.planDeletedInDeltaManifest( snapshot, skipper.get()), fileExecutor)); - if (futures.size() >= filesExpireBatchSize) { - collectAndClean(futures); - } } - collectAndClean(futures); + return flatten(getAll(futures)); } private Map>> collectTagSkippers( @@ -364,7 +374,16 @@ private Map>> collectTagSkippers( private void cleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { - Queue>> futures = new LinkedList<>(); + for (int from = 0; from < snapshots.size(); from += filesExpireBatchSize) { + int to = Math.min(from + filesExpireBatchSize, snapshots.size()); + snapshotDeletion.cleanDataFiles( + collectChangelogFilesToDelete(snapshots.subList(from, to))); + } + } + + private Collection collectChangelogFilesToDelete(List snapshots) + throws ExecutionException, InterruptedException { + List>> futures = new ArrayList<>(); for (Snapshot snapshot : snapshots) { if (LOG.isDebugEnabled()) { LOG.debug("Ready to delete changelog files from snapshot #{}", snapshot.id()); @@ -374,20 +393,9 @@ private void cleanChangelogFiles(List snapshots) CompletableFuture.supplyAsync( () -> snapshotDeletion.planAddedInChangelogManifest(snapshot), fileExecutor)); - if (futures.size() >= filesExpireBatchSize) { - collectAndClean(futures); - } } } - collectAndClean(futures); - } - - private void collectAndClean(Queue>> futures) - throws ExecutionException, InterruptedException { - while (!futures.isEmpty()) { - CompletableFuture> future = futures.remove(); - snapshotDeletion.cleanDataFiles(future.get()); - } + return flatten(getAll(futures)); } private Collection collectManifestDeletionTasks( From 0615b14bff228524180d8c7b75e8f140d8b82652 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 18:47:31 +0800 Subject: [PATCH 07/10] fix --- .../java/org/apache/paimon/operation/FileDeletionBase.java | 4 ---- 1 file changed, 4 deletions(-) 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 6fc67c49bce1..724c0354cdaa 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 @@ -123,10 +123,6 @@ public Executor fileExecutor() { return fileExecutor; } - public int fileOperationParallelism() { - return fileOperationParallelism; - } - /** * Clean data files that will not be used anymore in the snapshot. * From 4f1555bc885d825572dc667d40878ec8dc8e16e0 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 18:48:48 +0800 Subject: [PATCH 08/10] fix --- .../org/apache/paimon/table/ExpireSnapshotsImpl.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 130313120c15..23e35d4cc55f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -70,7 +70,7 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { private final SnapshotDeletion snapshotDeletion; private final Executor fileExecutor; private final TagManager tagManager; - private final int filesExpireBatchSize; + private final int snapshotExpireBatchSize; private ExpireConfig expireConfig; private Supplier currentTimeMillis = System::currentTimeMillis; @@ -92,7 +92,7 @@ public ExpireSnapshotsImpl( this.tagManager = tagManager; this.expireConfig = ExpireConfig.builder().build(); this.fileExecutor = snapshotDeletion.fileExecutor(); - this.filesExpireBatchSize = + this.snapshotExpireBatchSize = scanManifestParallelism == null || scanManifestParallelism <= 0 ? Runtime.getRuntime().availableProcessors() : scanManifestParallelism; @@ -274,8 +274,8 @@ private void cleanDataFiles( List taggedSnapshots, long beginInclusiveId) throws ExecutionException, InterruptedException { - for (int from = 0; from < snapshotsIncludingEnd.size(); from += filesExpireBatchSize) { - int to = Math.min(from + filesExpireBatchSize, snapshotsIncludingEnd.size()); + for (int from = 0; from < snapshotsIncludingEnd.size(); from += snapshotExpireBatchSize) { + int to = Math.min(from + snapshotExpireBatchSize, snapshotsIncludingEnd.size()); snapshotDeletion.cleanDataFiles( collectDataFilesToDelete( snapshotsIncludingEnd.subList(from, to), @@ -374,8 +374,8 @@ private Map>> collectTagSkippers( private void cleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { - for (int from = 0; from < snapshots.size(); from += filesExpireBatchSize) { - int to = Math.min(from + filesExpireBatchSize, snapshots.size()); + for (int from = 0; from < snapshots.size(); from += snapshotExpireBatchSize) { + int to = Math.min(from + snapshotExpireBatchSize, snapshots.size()); snapshotDeletion.cleanDataFiles( collectChangelogFilesToDelete(snapshots.subList(from, to))); } From 560523333bd79acf0aeabf89cb0853bba977ff15 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 18:57:51 +0800 Subject: [PATCH 09/10] fix --- .../paimon/table/ExpireSnapshotsImpl.java | 41 +++++++------------ 1 file changed, 15 insertions(+), 26 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 23e35d4cc55f..7c04bd1deb00 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -274,21 +274,6 @@ private void cleanDataFiles( List taggedSnapshots, long beginInclusiveId) throws ExecutionException, InterruptedException { - for (int from = 0; from < snapshotsIncludingEnd.size(); from += snapshotExpireBatchSize) { - int to = Math.min(from + snapshotExpireBatchSize, snapshotsIncludingEnd.size()); - snapshotDeletion.cleanDataFiles( - collectDataFilesToDelete( - snapshotsIncludingEnd.subList(from, to), - taggedSnapshots, - beginInclusiveId)); - } - } - - private Collection collectDataFilesToDelete( - List snapshotsIncludingEnd, - List taggedSnapshots, - long beginInclusiveId) - throws ExecutionException, InterruptedException { Map tagIdBySnapshotId = new HashMap<>(); Map tags = new HashMap<>(); int tagIndex = -1; @@ -337,8 +322,11 @@ private Collection collectDataFilesToDelete( snapshotDeletion.planDeletedInDeltaManifest( snapshot, skipper.get()), fileExecutor)); + if (futures.size() >= snapshotExpireBatchSize) { + collectAndClean(futures); + } } - return flatten(getAll(futures)); + collectAndClean(futures); } private Map>> collectTagSkippers( @@ -374,15 +362,6 @@ private Map>> collectTagSkippers( private void cleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { - for (int from = 0; from < snapshots.size(); from += snapshotExpireBatchSize) { - int to = Math.min(from + snapshotExpireBatchSize, snapshots.size()); - snapshotDeletion.cleanDataFiles( - collectChangelogFilesToDelete(snapshots.subList(from, to))); - } - } - - private Collection collectChangelogFilesToDelete(List snapshots) - throws ExecutionException, InterruptedException { List>> futures = new ArrayList<>(); for (Snapshot snapshot : snapshots) { if (LOG.isDebugEnabled()) { @@ -393,9 +372,19 @@ private Collection collectChangelogFilesToDelete(List snapshots) CompletableFuture.supplyAsync( () -> snapshotDeletion.planAddedInChangelogManifest(snapshot), fileExecutor)); + if (futures.size() >= snapshotExpireBatchSize) { + collectAndClean(futures); + } } } - return flatten(getAll(futures)); + collectAndClean(futures); + } + + private void collectAndClean(List>> futures) + throws ExecutionException, InterruptedException { + Collection paths = flatten(getAll(futures)); + futures.clear(); + snapshotDeletion.cleanDataFiles(paths); } private Collection collectManifestDeletionTasks( From ba1db8041266bd9e908c8e482caea05d5bff0935 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 26 Aug 2026 20:11:35 +0800 Subject: [PATCH 10/10] fix --- .../paimon/operation/FileDeletionBase.java | 5 ----- .../paimon/table/ExpireSnapshotsImpl.java | 19 +++++++++++-------- 2 files changed, 11 insertions(+), 13 deletions(-) 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 724c0354cdaa..a0545c87e484 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 @@ -82,7 +82,6 @@ public abstract class FileDeletionBase { protected final Map> deletionBuckets; private final Executor fileExecutor; - private final int fileOperationParallelism; @Nullable private final Integer manifestReadParallelism; protected boolean changelogDecoupled; @@ -112,10 +111,6 @@ public FileDeletionBase( this.cleanEmptyDirectories = cleanEmptyDirectories; this.deletionBuckets = new ConcurrentHashMap<>(); this.fileExecutor = FileOperationThreadPool.getExecutorService(fileOperationThreadNum); - this.fileOperationParallelism = - fileOperationThreadNum > 0 - ? fileOperationThreadNum - : Runtime.getRuntime().availableProcessors(); this.manifestReadParallelism = manifestReadParallelism; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 7c04bd1deb00..b7cabb4d32f3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -295,6 +295,7 @@ private void cleanDataFiles( collectTagSkippers(tags.values()); Predicate deleteAll = entry -> false; List>> futures = new ArrayList<>(); + int plannedSnapshots = 0; for (Snapshot snapshot : snapshotsIncludingEnd) { long id = snapshot.id(); if (id == beginInclusiveId) { @@ -315,18 +316,18 @@ private void cleanDataFiles( id); continue; } - futures.add( CompletableFuture.supplyAsync( () -> snapshotDeletion.planDeletedInDeltaManifest( snapshot, skipper.get()), fileExecutor)); - if (futures.size() >= snapshotExpireBatchSize) { - collectAndClean(futures); + if (++plannedSnapshots >= snapshotExpireBatchSize) { + cleanBatch(futures); + plannedSnapshots = 0; } } - collectAndClean(futures); + cleanBatch(futures); } private Map>> collectTagSkippers( @@ -363,6 +364,7 @@ private Map>> collectTagSkippers( private void cleanChangelogFiles(List snapshots) throws ExecutionException, InterruptedException { List>> futures = new ArrayList<>(); + int plannedSnapshots = 0; for (Snapshot snapshot : snapshots) { if (LOG.isDebugEnabled()) { LOG.debug("Ready to delete changelog files from snapshot #{}", snapshot.id()); @@ -372,15 +374,16 @@ private void cleanChangelogFiles(List snapshots) CompletableFuture.supplyAsync( () -> snapshotDeletion.planAddedInChangelogManifest(snapshot), fileExecutor)); - if (futures.size() >= snapshotExpireBatchSize) { - collectAndClean(futures); + if (++plannedSnapshots >= snapshotExpireBatchSize) { + cleanBatch(futures); + plannedSnapshots = 0; } } } - collectAndClean(futures); + cleanBatch(futures); } - private void collectAndClean(List>> futures) + private void cleanBatch(List>> futures) throws ExecutionException, InterruptedException { Collection paths = flatten(getAll(futures)); futures.clear();