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 55f2be7386 [core] Parallelize snapshot expiration file IO (#8241)
55f2be7386 is described below
commit 55f2be73860624af7064c205f4bf47f236ad416d
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jun 16 13:08:11 2026 +0800
[core] Parallelize snapshot expiration file IO (#8241)
This PR parallelizes snapshot expiration planning and file IO work to
reduce object-store latency impact when expiring many snapshots. It
keeps final snapshot/changelog deletion ordering where required, while
moving expensive snapshot reads, manifest reads, tag reads, and file
deletion planning onto the existing file operation thread pool.
---
.../apache/paimon/operation/ChangelogDeletion.java | 28 +-
.../apache/paimon/operation/FileDeletionBase.java | 343 ++++++----
.../apache/paimon/operation/SnapshotDeletion.java | 37 +-
.../org/apache/paimon/operation/TagDeletion.java | 14 +-
.../apache/paimon/table/ExpireChangelogImpl.java | 4 +-
.../apache/paimon/table/ExpireSnapshotsImpl.java | 235 +++++--
.../paimon/utils/ManifestReadThreadPool.java | 6 +
.../paimon/operation/ExpireSnapshotsTest.java | 730 ++++++++++++++++++++-
.../paimon/utils/ManifestReadThreadPoolTest.java | 11 +
.../java/org/apache/paimon/utils/SlowFileIO.java | 159 +++++
.../services/org.apache.paimon.fs.FileIOLoader | 1 +
11 files changed, 1346 insertions(+), 222 deletions(-)
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 5a5f04fedc..ba64eeecab 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,7 +31,9 @@ import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.FileStorePathFactory;
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Predicate;
@@ -47,7 +49,7 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int deleteFileThreadNum) {
+ int fileOperationThreadNum) {
super(
fileIO,
pathFactory,
@@ -56,17 +58,15 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- deleteFileThreadNum);
+ fileOperationThreadNum);
}
@Override
- public void cleanUnusedDataFiles(Changelog changelog,
Predicate<ExpireFileEntry> skipper) {
+ public void cleanDeletedDataFiles(Changelog changelog,
Predicate<ExpireFileEntry> skipper) {
if (changelog.changelogManifestList() != null) {
- deleteAddedDataFiles(changelog.changelogManifestList());
+ cleanDataFiles(planAddedInChangelogManifest(changelog));
} else {
- if (manifestList.exists(changelog.deltaManifestList())) {
- cleanUnusedDataFiles(changelog.deltaManifestList(), skipper);
- }
+ cleanDataFiles(planDeletedInDeltaManifest(changelog, skipper));
}
}
@@ -127,4 +127,18 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
return skippingSet;
}
+
+ public void cleanUnusedManifestList(String manifestName, Set<String>
skippingSet) {
+ executeAll(planUnusedManifestList(manifestName, skippingSet));
+ }
+
+ public List<Runnable> planUnusedManifestList(String manifestName,
Set<String> skippingSet) {
+ Set<String> manifests = new LinkedHashSet<>();
+ collectUnusedManifestList(manifestName, skippingSet, manifests);
+ List<Runnable> tasks = new ArrayList<>();
+ for (String manifest : manifests) {
+ tasks.add(() -> manifestFile.delete(manifest));
+ }
+ return tasks;
+ }
}
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 4256a503c1..2d21b5743b 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
@@ -37,6 +37,7 @@ import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.DataFilePathFactories;
import org.apache.paimon.utils.FileOperationThreadPool;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.ManifestReadThreadPool;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
@@ -50,10 +51,13 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.Predicate;
@@ -76,6 +80,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
protected final Map<BinaryRow, Set<Integer>> deletionBuckets;
private final Executor fileExecutor;
+ private final int fileOperationParallelism;
protected boolean changelogDecoupled;
@@ -93,7 +98,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int deleteFileThreadNum) {
+ int fileOperationThreadNum) {
this.fileIO = fileIO;
this.pathFactory = pathFactory;
this.manifestFile = manifestFile;
@@ -101,8 +106,12 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
this.indexFileHandler = indexFileHandler;
this.statsFileHandler = statsFileHandler;
this.cleanEmptyDirectories = cleanEmptyDirectories;
- this.deletionBuckets = new HashMap<>();
- this.fileExecutor =
FileOperationThreadPool.getExecutorService(deleteFileThreadNum);
+ this.deletionBuckets = new ConcurrentHashMap<>();
+ this.fileExecutor =
FileOperationThreadPool.getExecutorService(fileOperationThreadNum);
+ this.fileOperationParallelism =
+ fileOperationThreadNum > 0
+ ? fileOperationThreadNum
+ : Runtime.getRuntime().availableProcessors();
}
public Executor fileExecutor() {
@@ -116,7 +125,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
* @param skipper if the test result of a data file is true, it will be
skipped when deleting;
* else it will be deleted
*/
- public abstract void cleanUnusedDataFiles(T snapshot,
Predicate<ExpireFileEntry> skipper);
+ public abstract void cleanDeletedDataFiles(T snapshot,
Predicate<ExpireFileEntry> skipper);
/**
* Clean metadata files that will not be used anymore of a snapshot,
including data manifests,
@@ -137,7 +146,6 @@ public abstract class FileDeletionBase<T extends Snapshot> {
return;
}
- // All directory paths are deduplicated and sorted by hierarchy level
Map<Integer, Set<Path>> deduplicate = new HashMap<>();
for (Map.Entry<BinaryRow, Set<Integer>> entry :
deletionBuckets.entrySet()) {
List<Path> toDeleteEmptyDirectory = new ArrayList<>();
@@ -145,7 +153,7 @@ public abstract class FileDeletionBase<T extends Snapshot> {
for (Integer bucket : entry.getValue()) {
toDeleteEmptyDirectory.add(pathFactory.bucketPath(entry.getKey(), bucket));
}
- deleteFiles(toDeleteEmptyDirectory, this::tryDeleteEmptyDirectory);
+ executeAll(toDeleteEmptyDirectory, this::tryDeleteEmptyDirectory);
List<Path> hierarchicalPaths =
pathFactory.getHierarchicalPartitionPath(entry.getKey());
int hierarchies = hierarchicalPaths.size();
@@ -172,36 +180,48 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
protected void recordDeletionBuckets(ExpireFileEntry entry) {
deletionBuckets
- .computeIfAbsent(entry.partition(), p -> new HashSet<>())
+ .computeIfAbsent(entry.partition(), p ->
ConcurrentHashMap.newKeySet())
.add(entry.bucket());
}
- public void cleanUnusedDataFiles(String manifestList,
Predicate<ExpireFileEntry> skipper) {
- // try read manifests
- List<ManifestFileMeta> manifests = tryReadManifestList(manifestList);
- List<ExpireFileEntry> manifestEntries;
+ /** Plan data files referenced by DELETE entries in the snapshot's delta
manifest list. */
+ public List<Path> planDeletedInDeltaManifest(T snapshot,
Predicate<ExpireFileEntry> skipper) {
+ String deltaManifestList = snapshot.deltaManifestList();
// data file path -> (original manifest entry, extra file paths)
Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete = new
HashMap<>();
- for (ManifestFileMeta manifest : manifests) {
- try {
- manifestEntries =
- manifestFile.readExpireFileEntries(
- manifest.fileName(), manifest.fileSize());
- } catch (Exception e) {
- // cancel deletion if any exception occurs
- LOG.warn("Failed to read some manifest files. Cancel
deletion.", e);
- return;
+ try {
+ Iterable<ExpireFileEntry> dataFileEntries =
+
readExpireFileEntries(tryReadManifestList(deltaManifestList));
+ // we cannot delete a data file directly when we meet a DELETE
entry, because that
+ // file might be upgraded
+ DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
+ for (ExpireFileEntry entry : dataFileEntries) {
+ DataFilePathFactory dataFilePathFactory =
+ factories.get(entry.partition(), entry.bucket());
+ Path dataFilePath = dataFilePathFactory.toPath(entry);
+ switch (entry.kind()) {
+ case ADD:
+ dataFileToDelete.remove(dataFilePath);
+ break;
+ case DELETE:
+ List<Path> extraFiles = new
ArrayList<>(entry.extraFiles().size());
+ for (String file : entry.extraFiles()) {
+
extraFiles.add(dataFilePathFactory.toAlignedPath(file, entry));
+ }
+ dataFileToDelete.put(dataFilePath, Pair.of(entry,
extraFiles));
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown value kind " + entry.kind().name());
+ }
}
-
- getDataFileToDelete(dataFileToDelete, manifestEntries);
+ } catch (Exception e) {
+ // cancel deletion if any exception occurs
+ LOG.warn("Failed to read some manifest files. Cancel deletion.",
e);
+ return Collections.emptyList();
}
- doCleanUnusedDataFile(dataFileToDelete, skipper);
- }
-
- protected void doCleanUnusedDataFile(
- Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete,
- Predicate<ExpireFileEntry> skipper) {
+ // apply skipper
List<Path> actualDataFileToDelete = new ArrayList<>();
dataFileToDelete.forEach(
(path, pair) -> {
@@ -215,79 +235,69 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
recordDeletionBuckets(entry);
}
});
- deleteFiles(actualDataFileToDelete, fileIO::deleteQuietly);
+ return actualDataFileToDelete;
}
- protected void getDataFileToDelete(
- Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete,
- List<ExpireFileEntry> dataFileEntries) {
- // we cannot delete a data file directly when we meet a DELETE entry,
because that
- // file might be upgraded
+ /** Plan data files referenced by ADD entries in the snapshot's changelog
manifest list. */
+ public List<Path> planAddedInChangelogManifest(T snapshot) {
+ List<ManifestFileMeta> manifests =
tryReadManifestList(snapshot.changelogManifestList());
+ Iterable<ExpireFileEntry> entries =
+ ManifestReadThreadPool.sequentialBatchedExecute(
+ manifest -> {
+ try {
+ return manifestFile.readExpireFileEntries(
+ manifest.fileName(),
manifest.fileSize());
+ } catch (Exception e) {
+ // We want to delete the data file, so just
ignore the unavailable
+ // files
+ LOG.info(
+ "Failed to read manifest {}. Ignore
it.",
+ manifest.fileName(),
+ e);
+ return Collections.emptyList();
+ }
+ },
+ manifests,
+ fileOperationParallelism);
+
+ List<Path> dataFiles = new ArrayList<>();
DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
- for (ExpireFileEntry entry : dataFileEntries) {
+ for (ExpireFileEntry entry : entries) {
DataFilePathFactory dataFilePathFactory =
factories.get(entry.partition(), entry.bucket());
- Path dataFilePath = dataFilePathFactory.toPath(entry);
- switch (entry.kind()) {
- case ADD:
- dataFileToDelete.remove(dataFilePath);
- break;
- case DELETE:
- List<Path> extraFiles = new
ArrayList<>(entry.extraFiles().size());
- for (String file : entry.extraFiles()) {
- extraFiles.add(dataFilePathFactory.toAlignedPath(file,
entry));
- }
- dataFileToDelete.put(dataFilePath, Pair.of(entry,
extraFiles));
- break;
- default:
- throw new UnsupportedOperationException(
- "Unknown value kind " + entry.kind().name());
+ if (entry.kind() == FileKind.ADD) {
+ dataFiles.add(dataFilePathFactory.toPath(entry));
+ recordDeletionBuckets(entry);
}
}
+ return dataFiles;
}
- /**
- * Delete added file in the manifest list files. Added files marked as
"ADD" in manifests.
- *
- * @param manifestListName name of manifest list
- */
- public void deleteAddedDataFiles(String manifestListName) {
- List<ManifestFileMeta> manifests =
tryReadManifestList(manifestListName);
- for (ManifestFileMeta manifest : manifests) {
- try {
- List<ExpireFileEntry> manifestEntries =
+ private Iterable<ExpireFileEntry>
readExpireFileEntries(List<ManifestFileMeta> manifests) {
+ return ManifestReadThreadPool.sequentialBatchedExecute(
+ manifest ->
manifestFile.readExpireFileEntries(
- manifest.fileName(), manifest.fileSize());
- deleteAddedDataFiles(manifestEntries);
- } catch (Exception e) {
- // We want to delete the data file, so just ignore the
unavailable files
- LOG.info("Failed to read manifest " + manifest.fileName() + ".
Ignore it.", e);
- }
- }
+ manifest.fileName(), manifest.fileSize()),
+ manifests,
+ fileOperationParallelism);
}
- private void deleteAddedDataFiles(List<ExpireFileEntry> manifestEntries) {
- List<Path> dataFileToDelete = new ArrayList<>();
- DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
- for (ExpireFileEntry entry : manifestEntries) {
- DataFilePathFactory dataFilePathFactory =
- factories.get(entry.partition(), entry.bucket());
- if (entry.kind() == FileKind.ADD) {
- dataFileToDelete.add(dataFilePathFactory.toPath(entry));
- recordDeletionBuckets(entry);
- }
- }
- deleteFiles(dataFileToDelete, fileIO::deleteQuietly);
+ public void cleanDataFiles(Collection<Path> dataFiles) {
+ executeAll(new LinkedHashSet<>(dataFiles), fileIO::deleteQuietly);
}
- public void cleanUnusedStatisticsManifests(Snapshot snapshot, Set<String>
skippingSet) {
- // clean statistics
- if (snapshot.statistics() != null &&
!skippingSet.contains(snapshot.statistics())) {
- statsFileHandler.deleteStats(snapshot.statistics());
+ private void collectUnusedStatisticsManifests(
+ Snapshot snapshot, Set<String> skippingSet, Set<String>
statistics) {
+ if (snapshot.statistics() != null &&
skippingSet.add(snapshot.statistics())) {
+ statistics.add(snapshot.statistics());
}
}
- public void cleanUnusedIndexManifests(Snapshot snapshot, Set<String>
skippingSet) {
+ private void collectUnusedIndexManifests(
+ Snapshot snapshot,
+ Set<String> skippingSet,
+ Set<IndexManifestEntry> indexFiles,
+ Set<String> indexManifests) {
// clean index manifests
String indexManifest = snapshot.indexManifest();
// check exists, it may have been deleted by other snapshots
@@ -300,39 +310,41 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
} catch (IOException e) {
throw new RuntimeException(e);
}
- indexManifestEntries.removeIf(
- entry ->
skippingSet.contains(entry.indexFile().fileName()));
- deleteFiles(indexManifestEntries,
indexFileHandler::deleteIndexFile);
+ for (IndexManifestEntry entry : indexManifestEntries) {
+ if (skippingSet.add(entry.indexFile().fileName())) {
+ indexFiles.add(entry);
+ }
+ }
- if (!skippingSet.contains(indexManifest)) {
- indexFileHandler.deleteManifest(indexManifest);
+ if (skippingSet.add(indexManifest)) {
+ indexManifests.add(indexManifest);
}
}
}
- public void cleanUnusedManifestList(String manifestName, Set<String>
skippingSet) {
- List<String> toDeleteManifests = new ArrayList<>();
+ protected void collectUnusedManifestList(
+ String manifestName, Set<String> skippingSet, Set<String>
manifests) {
List<ManifestFileMeta> toExpireManifests =
tryReadManifestList(manifestName);
for (ManifestFileMeta manifest : toExpireManifests) {
String fileName = manifest.fileName();
- if (!skippingSet.contains(fileName)) {
- toDeleteManifests.add(fileName);
- // to avoid other snapshots trying to delete again
- skippingSet.add(fileName);
+ if (skippingSet.add(fileName)) {
+ manifests.add(fileName);
}
}
- if (!skippingSet.contains(manifestName)) {
- toDeleteManifests.add(manifestName);
+ if (skippingSet.add(manifestName)) {
+ manifests.add(manifestName);
}
-
- deleteFiles(toDeleteManifests, manifestFile::delete);
}
- protected void cleanUnusedManifests(
+ protected List<Runnable> planManifestsCleaner(
Snapshot snapshot,
Set<String> skippingSet,
boolean deleteDataManifestLists,
boolean deleteChangelog) {
+ Set<String> manifests = new LinkedHashSet<>();
+ Set<IndexManifestEntry> indexFiles = new LinkedHashSet<>();
+ Set<String> indexManifests = new LinkedHashSet<>();
+ Set<String> statistics = new LinkedHashSet<>();
if (deleteDataManifestLists) {
// deleteDataManifestLists will be false
// with changelog decouple + none changelog producer.
@@ -341,14 +353,29 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
// For none changelog producer, changelog files are the level 0
files.
// Even if these files are not used by the earliest snapshot,
// we have to keep them as changelog, and clean then in
ChangelogDeletion.
- cleanUnusedManifestList(snapshot.baseManifestList(), skippingSet);
- cleanUnusedManifestList(snapshot.deltaManifestList(), skippingSet);
+ collectUnusedManifestList(snapshot.baseManifestList(),
skippingSet, manifests);
+ collectUnusedManifestList(snapshot.deltaManifestList(),
skippingSet, manifests);
}
if (deleteChangelog && snapshot.changelogManifestList() != null) {
- cleanUnusedManifestList(snapshot.changelogManifestList(),
skippingSet);
+ collectUnusedManifestList(snapshot.changelogManifestList(),
skippingSet, manifests);
}
- cleanUnusedIndexManifests(snapshot, skippingSet);
- cleanUnusedStatisticsManifests(snapshot, skippingSet);
+ collectUnusedIndexManifests(snapshot, skippingSet, indexFiles,
indexManifests);
+ collectUnusedStatisticsManifests(snapshot, skippingSet, statistics);
+
+ List<Runnable> tasks = new ArrayList<>();
+ for (String manifest : manifests) {
+ tasks.add(() -> manifestFile.delete(manifest));
+ }
+ for (IndexManifestEntry indexFile : indexFiles) {
+ tasks.add(() -> indexFileHandler.deleteIndexFile(indexFile));
+ }
+ for (String indexManifest : indexManifests) {
+ tasks.add(() -> indexFileHandler.deleteManifest(indexManifest));
+ }
+ for (String statistic : statistics) {
+ tasks.add(() -> statsFileHandler.deleteStats(statistic));
+ }
+ return tasks;
}
public Predicate<ExpireFileEntry> createDataFileSkipperForTags(
@@ -369,6 +396,12 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
return entry -> false;
}
+ public Predicate<ExpireFileEntry> createDataFileSkipperForTag(Snapshot
tag) throws Exception {
+ Map<BinaryRow, Map<Integer, Set<String>>> tagDataFiles = new
HashMap<>();
+ addMergedDataFiles(tagDataFiles, tag);
+ return entry -> containsDataFile(tagDataFiles, entry);
+ }
+
/**
* It is possible that a job was killed during expiration and some
manifest files have been
* deleted, so if the clean methods need to get manifests of a snapshot to
be cleaned, we should
@@ -403,12 +436,7 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
protected Collection<ExpireFileEntry>
readMergedDataFiles(List<ManifestFileMeta> manifests)
throws IOException {
Map<Identifier, ExpireFileEntry> map = new HashMap<>();
- for (ManifestFileMeta manifest : manifests) {
- List<ExpireFileEntry> entries =
- manifestFile.readExpireFileEntries(manifest.fileName(),
manifest.fileSize());
- FileEntry.mergeEntries(entries, map);
- }
-
+ FileEntry.mergeEntries(readExpireFileEntries(manifests), map);
return map.values();
}
@@ -425,30 +453,59 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
}
public Set<String> manifestSkippingSet(List<Snapshot> skippingSnapshots) {
- Set<String> skippingSet = new HashSet<>();
+ if (skippingSnapshots.size() <= 1) {
+ Set<String> skippingSet = new HashSet<>();
+ for (Snapshot skippingSnapshot : skippingSnapshots) {
+ skippingSet.addAll(manifestSkippingSet(skippingSnapshot));
+ }
+ return skippingSet;
+ }
+ List<CompletableFuture<Set<String>>> futures = new ArrayList<>();
for (Snapshot skippingSnapshot : skippingSnapshots) {
- // data manifests
- skippingSet.add(skippingSnapshot.baseManifestList());
- skippingSet.add(skippingSnapshot.deltaManifestList());
- manifestList.readDataManifests(skippingSnapshot).stream()
- .map(ManifestFileMeta::fileName)
- .forEach(skippingSet::add);
+ futures.add(
+ CompletableFuture.supplyAsync(
+ () -> manifestSkippingSet(skippingSnapshot),
+
ManifestReadThreadPool.getExecutorService(fileOperationParallelism)));
+ }
- // index manifests
- String indexManifest = skippingSnapshot.indexManifest();
- if (indexManifest != null) {
- skippingSet.add(indexManifest);
- indexFileHandler.readManifest(indexManifest).stream()
- .map(IndexManifestEntry::indexFile)
- .map(IndexFileMeta::fileName)
- .forEach(skippingSet::add);
+ Set<String> skippingSet = new HashSet<>();
+ for (CompletableFuture<Set<String>> future : futures) {
+ try {
+ skippingSet.addAll(future.get());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e.getCause());
}
+ }
+ return skippingSet;
+ }
- // statistics
- if (skippingSnapshot.statistics() != null) {
- skippingSet.add(skippingSnapshot.statistics());
- }
+ private Set<String> manifestSkippingSet(Snapshot skippingSnapshot) {
+ Set<String> skippingSet = new HashSet<>();
+
+ // data manifests
+ skippingSet.add(skippingSnapshot.baseManifestList());
+ skippingSet.add(skippingSnapshot.deltaManifestList());
+ manifestList.readDataManifests(skippingSnapshot).stream()
+ .map(ManifestFileMeta::fileName)
+ .forEach(skippingSet::add);
+
+ // index manifests
+ String indexManifest = skippingSnapshot.indexManifest();
+ if (indexManifest != null) {
+ skippingSet.add(indexManifest);
+ indexFileHandler.readManifest(indexManifest).stream()
+ .map(IndexManifestEntry::indexFile)
+ .map(IndexFileMeta::fileName)
+ .forEach(skippingSet::add);
+ }
+
+ // statistics
+ if (skippingSnapshot.statistics() != null) {
+ skippingSet.add(skippingSnapshot.statistics());
}
return skippingSet;
@@ -464,21 +521,31 @@ public abstract class FileDeletionBase<T extends
Snapshot> {
}
}
- protected <F> void deleteFiles(Collection<F> files, Consumer<F> deletion) {
- if (files.isEmpty()) {
+ protected <F> void executeAll(Collection<F> files, Consumer<F> consumer) {
+ List<Runnable> tasks = new ArrayList<>(files.size());
+ for (F f : files) {
+ tasks.add(() -> consumer.accept(f));
+ }
+ executeAll(tasks);
+ }
+
+ public void executeAll(Collection<Runnable> tasks) {
+ if (tasks.isEmpty()) {
return;
}
- List<CompletableFuture<Void>> deletionFutures = new
ArrayList<>(files.size());
- for (F file : files) {
- deletionFutures.add(
- CompletableFuture.runAsync(() -> deletion.accept(file),
fileExecutor));
+ List<CompletableFuture<Void>> futures = new ArrayList<>(tasks.size());
+ for (Runnable runnable : tasks) {
+ futures.add(CompletableFuture.runAsync(runnable, fileExecutor));
}
try {
- CompletableFuture.allOf(deletionFutures.toArray(new
CompletableFuture[0])).get();
- } catch (Exception e) {
+ CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e.getCause());
}
}
}
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 7d55b64c8e..f97ec0474d 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
@@ -19,7 +19,6 @@
package org.apache.paimon.operation;
import org.apache.paimon.Snapshot;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.IndexFileHandler;
@@ -29,11 +28,8 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.FileStorePathFactory;
-import org.apache.paimon.utils.Pair;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
@@ -51,7 +47,7 @@ public class SnapshotDeletion extends
FileDeletionBase<Snapshot> {
StatsFileHandler statsFileHandler,
boolean produceChangelog,
boolean cleanEmptyDirectories,
- int deleteFileThreadNum) {
+ int fileOperationThreadNum) {
super(
fileIO,
pathFactory,
@@ -60,41 +56,44 @@ public class SnapshotDeletion extends
FileDeletionBase<Snapshot> {
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- deleteFileThreadNum);
+ fileOperationThreadNum);
this.produceChangelog = produceChangelog;
}
@Override
- public void cleanUnusedDataFiles(Snapshot snapshot,
Predicate<ExpireFileEntry> skipper) {
+ public void cleanDeletedDataFiles(Snapshot snapshot,
Predicate<ExpireFileEntry> skipper) {
+ cleanDataFiles(planDeletedInDeltaManifest(snapshot, skipper));
+ }
+
+ @Override
+ public List<Path> planDeletedInDeltaManifest(
+ Snapshot snapshot, Predicate<ExpireFileEntry> skipper) {
+ Predicate<ExpireFileEntry> enriched = skipper;
if (changelogDecoupled && !produceChangelog) {
// Skip clean the 'APPEND' data files.If we do not have the file
source information
// eg: the old version table file, we just skip clean this here,
let it done by
// ExpireChangelogImpl
- Predicate<ExpireFileEntry> enriched =
+ enriched =
manifestEntry ->
skipper.test(manifestEntry)
||
(manifestEntry.fileSource().orElse(FileSource.APPEND)
== FileSource.APPEND);
- cleanUnusedDataFiles(snapshot.deltaManifestList(), enriched);
- } else {
- cleanUnusedDataFiles(snapshot.deltaManifestList(), skipper);
}
+ return super.planDeletedInDeltaManifest(snapshot, enriched);
}
@Override
public void cleanUnusedManifests(Snapshot snapshot, Set<String>
skippingSet) {
// delay clean the base and delta manifest lists when changelog
decoupled enabled
- cleanUnusedManifests(
+ executeAll(planManifestsCleaner(snapshot, skippingSet));
+ }
+
+ public List<Runnable> planManifestsCleaner(Snapshot snapshot, Set<String>
skippingSet) {
+ // delay clean the base and delta manifest lists when changelog
decoupled enabled
+ return planManifestsCleaner(
snapshot,
skippingSet,
!changelogDecoupled || produceChangelog,
!changelogDecoupled);
}
-
- @VisibleForTesting
- void cleanUnusedDataFile(List<ExpireFileEntry> dataFileLog) {
- Map<Path, Pair<ExpireFileEntry, List<Path>>> dataFileToDelete = new
HashMap<>();
- getDataFileToDelete(dataFileToDelete, dataFileLog);
- doCleanUnusedDataFile(dataFileToDelete, f -> false);
- }
}
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 569a1464cc..ff3e05f9aa 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
@@ -57,7 +57,7 @@ public class TagDeletion extends FileDeletionBase<Snapshot> {
IndexFileHandler indexFileHandler,
StatsFileHandler statsFileHandler,
boolean cleanEmptyDirectories,
- int deleteFileThreadNum) {
+ int fileOperationThreadNum) {
super(
fileIO,
pathFactory,
@@ -66,11 +66,11 @@ public class TagDeletion extends FileDeletionBase<Snapshot>
{
indexFileHandler,
statsFileHandler,
cleanEmptyDirectories,
- deleteFileThreadNum);
+ fileOperationThreadNum);
}
@Override
- public void cleanUnusedDataFiles(Snapshot taggedSnapshot,
Predicate<ExpireFileEntry> skipper) {
+ public void cleanDeletedDataFiles(Snapshot taggedSnapshot,
Predicate<ExpireFileEntry> skipper) {
Collection<ExpireFileEntry> manifestEntries;
try {
List<ManifestFileMeta> manifests =
@@ -96,13 +96,17 @@ public class TagDeletion extends FileDeletionBase<Snapshot>
{
recordDeletionBuckets(entry);
}
}
- deleteFiles(dataFileToDelete, fileIO::deleteQuietly);
+ executeAll(dataFileToDelete, fileIO::deleteQuietly);
+ }
+
+ public void cleanUnusedDataFiles(Snapshot taggedSnapshot,
Predicate<ExpireFileEntry> skipper) {
+ cleanDeletedDataFiles(taggedSnapshot, skipper);
}
@Override
public void cleanUnusedManifests(Snapshot taggedSnapshot, Set<String>
skippingSet) {
// doesn't clean changelog files because they are handled by
SnapshotDeletion
- cleanUnusedManifests(taggedSnapshot, skippingSet, true, false);
+ executeAll(planManifestsCleaner(taggedSnapshot, skippingSet, true,
false));
}
public Predicate<ExpireFileEntry> dataFileSkipper(List<Snapshot>
fromSnapshots)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
index 73f9c88b5f..b5d0af3954 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
@@ -191,7 +191,7 @@ public class ExpireChangelogImpl implements ExpireSnapshots
{
continue;
}
- changelogDeletion.cleanUnusedDataFiles(changelog, skipper);
+ changelogDeletion.cleanDeletedDataFiles(changelog, skipper);
changelogDeletion.cleanUnusedManifests(changelog,
manifestSkippSet);
changelogManager.fileIO().deleteQuietly(changelogManager.longLivedChangelogPath(id));
}
@@ -262,7 +262,7 @@ public class ExpireChangelogImpl implements ExpireSnapshots
{
continue;
}
- changelogDeletion.cleanUnusedDataFiles(changelog, skipper);
+ changelogDeletion.cleanDeletedDataFiles(changelog, skipper);
changelogDeletion.cleanUnusedManifests(changelog,
manifestSkippSet);
changelogManager.fileIO().deleteQuietly(changelogManager.longLivedChangelogPath(id));
}
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 ef3796f20d..32124bfc2a 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
@@ -22,9 +22,11 @@ import org.apache.paimon.Changelog;
import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ExpireFileEntry;
import org.apache.paimon.operation.SnapshotDeletion;
import org.apache.paimon.options.ExpireConfig;
+import org.apache.paimon.tag.Tag;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SnapshotManager;
@@ -37,11 +39,15 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.function.Predicate;
@@ -196,44 +202,17 @@ public class ExpireSnapshotsImpl implements
ExpireSnapshots {
long beginInclusiveId = snapshotsIncludingEnd.get(0).id();
// tags to create data file skipper
- List<Snapshot> taggedSnapshots = tagManager.taggedSnapshots();
+ List<Snapshot> taggedSnapshots = collectTaggedSnapshots();
// 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]
- for (Snapshot snapshot : snapshotsIncludingEnd) {
- long id = snapshot.id();
- if (id == beginInclusiveId) {
- continue;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Ready to delete merge tree files not used by
snapshot #{}", id);
- }
- // expire merge tree files and collect changed buckets
- Predicate<ExpireFileEntry> skipper;
- try {
- skipper =
snapshotDeletion.createDataFileSkipperForTags(taggedSnapshots, id);
- } catch (Exception e) {
- LOG.info(
- "Skip cleaning data files of snapshot '{}' due to
failed to build skipping set.",
- id,
- e);
- continue;
- }
-
- snapshotDeletion.cleanUnusedDataFiles(snapshot, skipper);
- }
+ snapshotDeletion.cleanDataFiles(
+ collectDataFilesToDelete(snapshotsIncludingEnd,
taggedSnapshots, beginInclusiveId));
// delete changelog files
if (!expireConfig.isChangelogDecoupled()) {
- for (Snapshot snapshot : snapshotsExcludingEnd) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Ready to delete changelog files from snapshot
#{}", snapshot.id());
- }
- if (snapshot.changelogManifestList() != null) {
-
snapshotDeletion.deleteAddedDataFiles(snapshot.changelogManifestList());
- }
- }
+
snapshotDeletion.cleanDataFiles(collectChangelogFilesToDelete(snapshotsExcludingEnd));
}
// data files and changelog files in bucket directories has been
deleted
@@ -254,17 +233,15 @@ public class ExpireSnapshotsImpl implements
ExpireSnapshots {
Set<String> skippingSet = null;
try {
- skippingSet = new
HashSet<>(snapshotDeletion.manifestSkippingSet(skippingSnapshots));
+ Set<String> builtSkippingSet = ConcurrentHashMap.newKeySet();
+
builtSkippingSet.addAll(snapshotDeletion.manifestSkippingSet(skippingSnapshots));
+ skippingSet = builtSkippingSet;
} catch (Exception e) {
LOG.info("Skip cleaning manifest files due to failed to build
skipping set.", e);
}
if (skippingSet != null) {
- for (Snapshot snapshot : snapshotsExcludingEnd) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Ready to delete manifests in snapshot #{}",
snapshot.id());
- }
- snapshotDeletion.cleanUnusedManifests(snapshot, skippingSet);
- }
+ snapshotDeletion.executeAll(
+ collectManifestDeletionTasks(snapshotsExcludingEnd,
skippingSet));
}
// delete snapshot file finally
@@ -285,6 +262,144 @@ public class ExpireSnapshotsImpl implements
ExpireSnapshots {
return snapshotsExcludingEnd.size();
}
+ private Collection<Path> collectDataFilesToDelete(
+ List<Snapshot> snapshotsIncludingEnd,
+ List<Snapshot> taggedSnapshots,
+ long beginInclusiveId)
+ throws ExecutionException, InterruptedException {
+ Map<Long, Long> tagIdBySnapshotId = new HashMap<>();
+ Map<Long, Snapshot> tags = new HashMap<>();
+ int tagIndex = -1;
+ for (Snapshot snapshot : snapshotsIncludingEnd) {
+ long id = snapshot.id();
+ if (id == beginInclusiveId) {
+ continue;
+ }
+
+ tagIndex = advancePreviousSnapshot(taggedSnapshots, tagIndex, id);
+ if (tagIndex >= 0) {
+ Snapshot tag = taggedSnapshots.get(tagIndex);
+ tagIdBySnapshotId.put(id, tag.id());
+ tags.put(tag.id(), tag);
+ }
+ }
+
+ Map<Long, Optional<Predicate<ExpireFileEntry>>> skippers =
+ collectTagSkippers(tags.values());
+ Predicate<ExpireFileEntry> deleteAll = entry -> false;
+ List<CompletableFuture<List<Path>>> futures = new ArrayList<>();
+ for (Snapshot snapshot : snapshotsIncludingEnd) {
+ long id = snapshot.id();
+ if (id == beginInclusiveId) {
+ continue;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ready to delete merge tree files not used by
snapshot #{}", id);
+ }
+
+ Long tagId = tagIdBySnapshotId.get(id);
+ Optional<Predicate<ExpireFileEntry>> skipper =
+ tagId == null
+ ? Optional.of(deleteAll)
+ : skippers.getOrDefault(tagId, Optional.empty());
+ if (!skipper.isPresent()) {
+ LOG.info(
+ "Skip cleaning data files of snapshot '{}' due to
failed to build skipping set.",
+ id);
+ continue;
+ }
+
+ futures.add(
+ CompletableFuture.supplyAsync(
+ () ->
+
snapshotDeletion.planDeletedInDeltaManifest(
+ snapshot, skipper.get()),
+ fileExecutor));
+ }
+ return flatten(getAll(futures));
+ }
+
+ private Map<Long, Optional<Predicate<ExpireFileEntry>>> collectTagSkippers(
+ Collection<Snapshot> tags) throws ExecutionException,
InterruptedException {
+ Map<Long, CompletableFuture<Optional<Predicate<ExpireFileEntry>>>>
futures =
+ new HashMap<>();
+ for (Snapshot tag : tags) {
+ futures.put(
+ tag.id(),
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Optional.of(
+
snapshotDeletion.createDataFileSkipperForTag(tag));
+ } catch (Exception e) {
+ LOG.info(
+ "Failed to build data file
skipping set for tag snapshot '{}'.",
+ tag.id(),
+ e);
+ return Optional.empty();
+ }
+ },
+ fileExecutor));
+ }
+
+ Map<Long, Optional<Predicate<ExpireFileEntry>>> skippers = new
HashMap<>();
+ for (Map.Entry<Long,
CompletableFuture<Optional<Predicate<ExpireFileEntry>>>> entry :
+ futures.entrySet()) {
+ skippers.put(entry.getKey(), entry.getValue().get());
+ }
+ return skippers;
+ }
+
+ private Collection<Path> collectChangelogFilesToDelete(List<Snapshot>
snapshots)
+ throws ExecutionException, InterruptedException {
+ List<CompletableFuture<List<Path>>> futures = new ArrayList<>();
+ for (Snapshot snapshot : snapshots) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ready to delete changelog files from snapshot #{}",
snapshot.id());
+ }
+ if (snapshot.changelogManifestList() != null) {
+ futures.add(
+ CompletableFuture.supplyAsync(
+ () ->
snapshotDeletion.planAddedInChangelogManifest(snapshot),
+ fileExecutor));
+ }
+ }
+ return flatten(getAll(futures));
+ }
+
+ private Collection<Runnable> collectManifestDeletionTasks(
+ List<Snapshot> snapshots, Set<String> skippingSet)
+ throws ExecutionException, InterruptedException {
+ List<CompletableFuture<List<Runnable>>> futures = new ArrayList<>();
+ for (Snapshot snapshot : snapshots) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ready to delete manifests in snapshot #{}",
snapshot.id());
+ }
+ futures.add(
+ CompletableFuture.supplyAsync(
+ () ->
snapshotDeletion.planManifestsCleaner(snapshot, skippingSet),
+ fileExecutor));
+ }
+ return flatten(getAll(futures));
+ }
+
+ private <T> List<T> getAll(List<CompletableFuture<T>> futures)
+ throws ExecutionException, InterruptedException {
+ List<T> result = new ArrayList<>();
+ for (CompletableFuture<T> future : futures) {
+ result.add(future.get());
+ }
+ return result;
+ }
+
+ private <T> List<T> flatten(List<? extends Collection<T>> collections) {
+ List<T> result = new ArrayList<>();
+ for (Collection<T> collection : collections) {
+ result.addAll(collection);
+ }
+ return result;
+ }
+
private List<Snapshot> collectSnapshots(long earliestId, long
endExclusiveId)
throws InterruptedException, ExecutionException {
List<CompletableFuture<Optional<Snapshot>>> futures = new
ArrayList<>();
@@ -309,6 +424,48 @@ public class ExpireSnapshotsImpl implements
ExpireSnapshots {
return snapshots;
}
+ private List<Snapshot> collectTaggedSnapshots()
+ throws InterruptedException, ExecutionException {
+ List<Path> tagPaths;
+ try {
+ tagPaths = tagManager.tagPaths(path -> true);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ List<CompletableFuture<Optional<Snapshot>>> futures = new
ArrayList<>();
+ for (Path path : tagPaths) {
+ futures.add(
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return Optional.of(
+
Tag.tryFromPath(snapshotManager.fileIO(), path)
+ .trimToSnapshot());
+ } catch (FileNotFoundException ignored) {
+ return Optional.empty();
+ }
+ },
+ fileExecutor));
+ }
+
+ List<Snapshot> snapshots = new ArrayList<>();
+ for (CompletableFuture<Optional<Snapshot>> future : futures) {
+ future.get().ifPresent(snapshots::add);
+ }
+ snapshots.sort(Comparator.comparingLong(Snapshot::id));
+ return snapshots;
+ }
+
+ private static int advancePreviousSnapshot(
+ List<Snapshot> sortedSnapshots, int currentIndex, long
targetSnapshotId) {
+ while (currentIndex + 1 < sortedSnapshots.size()
+ && sortedSnapshots.get(currentIndex + 1).id() <
targetSnapshotId) {
+ currentIndex++;
+ }
+ return currentIndex;
+ }
+
private void commitChangelog(Changelog changelog) {
try {
changelogManager.commitChangelog(changelog, changelog.id());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
index 16bd73ee80..7cf9778714 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
@@ -37,6 +37,7 @@ public class ManifestReadThreadPool {
createCachedThreadPool(Runtime.getRuntime().availableProcessors(),
THREAD_NAME);
public static synchronized ExecutorService getExecutorService(@Nullable
Integer threadNum) {
+ threadNum = normalizeThreadNum(threadNum);
if (threadNum == null || threadNum ==
executorService.getMaximumPoolSize()) {
return executorService;
}
@@ -54,6 +55,7 @@ public class ManifestReadThreadPool {
/** This method aims to parallel process tasks with memory control and
sequentially. */
public static <T, U> Iterable<T> sequentialBatchedExecute(
Function<U, List<T>> processor, List<U> input, @Nullable Integer
threadNum) {
+ threadNum = normalizeThreadNum(threadNum);
ExecutorService executor = getExecutorService(threadNum);
if (threadNum == null) {
threadNum =
@@ -64,6 +66,10 @@ public class ManifestReadThreadPool {
return ThreadPoolUtils.sequentialBatchedExecute(executor, processor,
input, threadNum);
}
+ private static @Nullable Integer normalizeThreadNum(@Nullable Integer
threadNum) {
+ return threadNum == null || threadNum <= 0 ? null : threadNum;
+ }
+
/** This method aims to parallel process tasks with randomly but return
values sequentially. */
public static <T, U> Iterator<T> randomlyExecuteSequentialReturn(
Function<U, List<T>> processor, List<U> input, @Nullable Integer
threadNum) {
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 0b2b0031c8..1da28a5bef 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
@@ -37,15 +37,18 @@ import org.apache.paimon.manifest.ExpireFileEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.options.ExpireConfig;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.ExpireSnapshots;
import org.apache.paimon.table.ExpireSnapshotsImpl;
import org.apache.paimon.utils.ChangelogManager;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.RecordWriter;
+import org.apache.paimon.utils.SlowFileIO;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -63,13 +66,18 @@ import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import static java.util.Objects.requireNonNull;
@@ -277,8 +285,8 @@ public class ExpireSnapshotsTest {
1,
EMPTY_ROW,
EMPTY_ROW,
- null,
- null,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
0,
1,
0,
@@ -296,9 +304,7 @@ public class ExpireSnapshotsTest {
ManifestEntry delete = ManifestEntry.create(FileKind.DELETE,
partition, 0, 1, dataFile);
// expire
- expire.snapshotDeletion()
- .cleanUnusedDataFile(
- Arrays.asList(ExpireFileEntry.from(add),
ExpireFileEntry.from(delete)));
+ cleanDeletedDataFiles(expire.snapshotDeletion(), Arrays.asList(add,
delete));
// check
assertThat(fileIO.exists(myDataFile)).isFalse();
@@ -340,8 +346,8 @@ public class ExpireSnapshotsTest {
1,
EMPTY_ROW,
EMPTY_ROW,
- null,
- null,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
0,
1,
0,
@@ -359,9 +365,7 @@ public class ExpireSnapshotsTest {
ManifestEntry delete = ManifestEntry.create(FileKind.DELETE,
partition, 0, 1, dataFile);
// expire
- expire.snapshotDeletion()
- .cleanUnusedDataFile(
- Arrays.asList(ExpireFileEntry.from(add),
ExpireFileEntry.from(delete)));
+ cleanDeletedDataFiles(expire.snapshotDeletion(), Arrays.asList(add,
delete));
// check
assertThat(fileIO.exists(myDataFile)).isFalse();
@@ -371,6 +375,54 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ private void cleanDeletedDataFiles(
+ SnapshotDeletion snapshotDeletion, List<ManifestEntry>
dataFileLog) {
+ List<ManifestFileMeta> manifests =
store.manifestFileFactory().create().write(dataFileLog);
+ String manifestList =
store.manifestListFactory().create().write(manifests).getLeft();
+ Snapshot snapshot = snapshotWithDeltaManifestList(manifestList);
+
+ snapshotDeletion.cleanDataFiles(
+ snapshotDeletion.planDeletedInDeltaManifest(snapshot, file ->
false));
+
+ for (ManifestFileMeta manifest : manifests) {
+
fileIO.deleteQuietly(store.pathFactory().toManifestFilePath(manifest.fileName()));
+ }
+
fileIO.deleteQuietly(store.pathFactory().toManifestListPath(manifestList));
+ }
+
+ private Snapshot snapshotWithDeltaManifestList(String manifestList) {
+ return snapshotWithManifestLists(manifestList, null);
+ }
+
+ private Snapshot snapshotWithChangelogManifestList(String manifestList) {
+ return snapshotWithManifestLists(null, manifestList);
+ }
+
+ private Snapshot snapshotWithManifestLists(
+ String deltaManifestList, String changelogManifestList) {
+ return new Snapshot(
+ 0,
+ 0L,
+ null,
+ null,
+ deltaManifestList,
+ null,
+ changelogManifestList,
+ null,
+ null,
+ "test",
+ 0L,
+ Snapshot.CommitKind.APPEND,
+ 0L,
+ 0L,
+ 0L,
+ null,
+ null,
+ null,
+ null,
+ null);
+ }
+
@Test
public void testNoSnapshot() throws IOException {
ExpireSnapshots expire = store.newExpire(1, 3, Long.MAX_VALUE);
@@ -560,6 +612,276 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ @Test
+ public void testExpireCollectsSnapshotsConcurrently() throws Exception {
+
store.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM, 4);
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(5, allData, snapshotPositions);
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= 5; i++) {
+ rewriteSnapshotTime(i, 0);
+ }
+
+ BlockingSnapshotManager blockingSnapshotManager =
+ new BlockingSnapshotManager(snapshotManager, 1, 5);
+ ExpireSnapshotsImpl expire =
+ new ExpireSnapshotsImpl(
+ blockingSnapshotManager,
+ changelogManager,
+ store.newSnapshotDeletion(),
+ store.newTagManager());
+
+ expire.expireUntil(1, latestSnapshotId);
+
+ assertThat(blockingSnapshotManager.maxActiveReads()).isGreaterThan(1);
+ assertSnapshot(latestSnapshotId, allData, snapshotPositions);
+ store.assertCleaned();
+ }
+
+ @Test
+ public void testExpirePlansDataFilesConcurrently() throws Exception {
+
store.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM, 4);
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(6, allData, snapshotPositions);
+ SnapshotManager snapshotManager = store.snapshotManager();
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ rewriteSnapshotTime(i, 0);
+ }
+
+ BlockingSnapshotDeletion snapshotDeletion =
+ new BlockingSnapshotDeletion(store, 2, latestSnapshotId);
+ snapshotDeletion.blockDataFilePlans();
+ ExpireSnapshotsImpl expire =
+ newExpireWithSnapshotDeletion(store, snapshotManager,
snapshotDeletion);
+ expire.config(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+
+ expire.expire();
+
+ assertThat(snapshotDeletion.maxActiveDataFilePlans()).isGreaterThan(1);
+ assertSnapshot(latestSnapshotId, allData, snapshotPositions);
+ store.assertCleaned();
+ }
+
+ @Test
+ public void testExpirePlansChangelogFilesConcurrently() throws Exception {
+ TestFileStore inputStore =
createStore(CoreOptions.ChangelogProducer.INPUT);
+
inputStore.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM,
4);
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(inputStore, 6, allData, snapshotPositions);
+ SnapshotManager snapshotManager = inputStore.snapshotManager();
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ rewriteSnapshotTime(inputStore.fileIO(), snapshotManager, i, 0);
+ }
+
+ Set<String> changelogManifestLists = new HashSet<>();
+ for (int i = 1; i < latestSnapshotId; i++) {
+ String changelogManifestList =
snapshotManager.snapshot(i).changelogManifestList();
+ if (changelogManifestList != null) {
+ changelogManifestLists.add(changelogManifestList);
+ }
+ }
+ assertThat(changelogManifestLists.size()).isGreaterThan(1);
+
+ BlockingSnapshotDeletion snapshotDeletion =
+ new BlockingSnapshotDeletion(inputStore, 1, latestSnapshotId -
1);
+ snapshotDeletion.blockChangelogPlans(changelogManifestLists);
+ ExpireSnapshotsImpl expire =
+ newExpireWithSnapshotDeletion(inputStore, snapshotManager,
snapshotDeletion);
+ expire.config(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+
+ expire.expire();
+
+
assertThat(snapshotDeletion.maxActiveChangelogPlans()).isGreaterThan(1);
+ assertSnapshot(inputStore, latestSnapshotId, allData,
snapshotPositions);
+ inputStore.assertCleaned();
+ }
+
+ @Test
+ public void testExpirePlansManifestsConcurrentlyWithSkippingSet() throws
Exception {
+
store.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM, 4);
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(6, allData, snapshotPositions);
+ SnapshotManager snapshotManager = store.snapshotManager();
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ rewriteSnapshotTime(i, 0);
+ }
+
+ BlockingSnapshotDeletion snapshotDeletion =
+ new BlockingSnapshotDeletion(store, 1, latestSnapshotId - 1);
+ snapshotDeletion.blockManifestPlans();
+ ExpireSnapshotsImpl expire =
+ newExpireWithSnapshotDeletion(store, snapshotManager,
snapshotDeletion);
+ expire.config(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+
+ expire.expire();
+
+ assertThat(snapshotDeletion.maxActiveManifestPlans()).isGreaterThan(1);
+ assertSnapshot(latestSnapshotId, allData, snapshotPositions);
+ store.assertCleaned();
+ }
+
+ @Test
+ public void
testExpireWithTagsAndConcurrentPlanningKeepsTaggedSnapshotsReadable()
+ throws Exception {
+
store.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM, 4);
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(8, allData, snapshotPositions);
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ rewriteSnapshotTime(i, 0);
+ }
+
+ TagManager tagManager = store.newTagManager();
+ tagManager.createTag(
+ snapshotManager.snapshot(3),
+ "tag3",
+ store.options().tagDefaultTimeRetained(),
+ Collections.emptyList(),
+ false);
+ tagManager.createTag(
+ snapshotManager.snapshot(6),
+ "tag6",
+ store.options().tagDefaultTimeRetained(),
+ Collections.emptyList(),
+ false);
+
+ ExpireSnapshotsImpl expire =
+ (ExpireSnapshotsImpl)
store.newExpire(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+ expire.expire();
+
+ for (int i = 1; i < latestSnapshotId; i++) {
+ assertThat(snapshotManager.snapshotExists(i)).isFalse();
+ }
+ assertSnapshot(latestSnapshotId, allData, snapshotPositions);
+ assertSnapshot(tagManager.getOrThrow("tag3").trimToSnapshot(),
allData, snapshotPositions);
+ assertSnapshot(tagManager.getOrThrow("tag6").trimToSnapshot(),
allData, snapshotPositions);
+ }
+
+ @Test
+ public void testExpireReadsTagsConcurrentlyWithObjectStoreFileIO() throws
Exception {
+ TestFileStore slowStore = createSlowStore();
+
slowStore.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM,
4);
+
+ try {
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(slowStore, 6, allData, snapshotPositions);
+
+ SnapshotManager snapshotManager = slowStore.snapshotManager();
+ int latestSnapshotId =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ TagManager tagManager = slowStore.newTagManager();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ tagManager.createTag(
+ snapshotManager.snapshot(i),
+ "tag" + i,
+ slowStore.options().tagDefaultTimeRetained(),
+ Collections.emptyList(),
+ false);
+ rewriteSnapshotTime(slowStore.fileIO(), snapshotManager, i, 0);
+ }
+
+ SlowFileIO.reset();
+ SlowFileIO.setDelayMillis(20);
+ ExpireSnapshotsImpl expire =
+ (ExpireSnapshotsImpl)
slowStore.newExpire(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+
+ expire.expire();
+
+ assertThat(SlowFileIO.delayedOperations()).isGreaterThan(0);
+ assertThat(SlowFileIO.maxActiveOperations()).isGreaterThan(1);
+ assertSnapshot(slowStore, latestSnapshotId, allData,
snapshotPositions);
+ } finally {
+ SlowFileIO.reset();
+ }
+ }
+
+ @Test
+ public void
testPlanUnusedDataFilesCancelsDeletionWhenManifestFileMissing() throws
Exception {
+ ManifestFileMeta missingManifest =
+ new ManifestFileMeta(
+ "missing-manifest",
+ 1,
+ 0,
+ 1,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+ String manifestList =
+ store.manifestListFactory()
+ .create()
+ .write(Arrays.asList(missingManifest))
+ .getLeft();
+
+ CapturingSnapshotDeletion snapshotDeletion = new
CapturingSnapshotDeletion(store);
+ assertThat(
+ snapshotDeletion.planDeletedInDeltaManifest(
+ snapshotWithDeltaManifestList(manifestList),
entry -> false))
+ .isEmpty();
+ }
+
+ @Test
+ public void testExpireWithSlowObjectStoreFileIO() throws Exception {
+ TestFileStore slowStore = createSlowStore();
+
slowStore.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM,
4);
+
+ try {
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ for (int i = 0; i < 8; i++) {
+ commit(slowStore, 3, allData, snapshotPositions);
+ }
+
+ SnapshotManager slowSnapshotManager = slowStore.snapshotManager();
+ int latestSnapshotId =
+
requireNonNull(slowSnapshotManager.latestSnapshotId()).intValue();
+ for (int i = 1; i <= latestSnapshotId; i++) {
+ rewriteSnapshotTime(slowStore.fileIO(), slowSnapshotManager,
i, 0);
+ }
+
+ SlowFileIO.reset();
+ SlowFileIO.setDelayMillis(20);
+ ExpireConfig config =
+ ExpireConfig.builder()
+ .snapshotRetainMin(1)
+ .snapshotRetainMax(Integer.MAX_VALUE)
+ .snapshotTimeRetain(Duration.ofMillis(1))
+ .build();
+ ExpireSnapshotsImpl expire = (ExpireSnapshotsImpl)
slowStore.newExpire(config);
+ expire.setCurrentTimeMillis(() -> 1000L);
+
+ expire.expire();
+
+ assertThat(SlowFileIO.delayedOperations()).isGreaterThan(0);
+ assertThat(SlowFileIO.maxActiveOperations()).isGreaterThan(1);
+ assertSnapshot(slowStore, latestSnapshotId, allData,
snapshotPositions);
+ } finally {
+ SlowFileIO.reset();
+ }
+ }
+
@Test
public void testExpireWithTimeProtectsEachSnapshot() throws Exception {
// Even with a small retainMin, each snapshot should be protected by
@@ -611,6 +933,45 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ @Test
+ public void testExpireWithTimeDoesNotReadProtectedRange() throws Exception
{
+ ExpireConfig config =
+ ExpireConfig.builder()
+ .snapshotRetainMin(1)
+ .snapshotRetainMax(Integer.MAX_VALUE)
+ .snapshotTimeRetain(Duration.ofMillis(5000))
+ .build();
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(6, allData, snapshotPositions);
+ for (int i = 1; i <= 6; i++) {
+ rewriteSnapshotTime(i, 0);
+ }
+ rewriteSnapshotTime(4, 2000);
+
+ FailingSnapshotManager failingSnapshotManager =
+ new FailingSnapshotManager(snapshotManager, 5);
+ ExpireSnapshotsImpl expire =
+ new ExpireSnapshotsImpl(
+ failingSnapshotManager,
+ changelogManager,
+ store.newSnapshotDeletion(),
+ store.newTagManager());
+ expire.config(config);
+ expire.setCurrentTimeMillis(() -> 6000L);
+
+ expire.expire();
+
+ assertThat(failingSnapshotManager.readFailedSnapshot()).isFalse();
+ assertThat(snapshotManager.snapshotExists(1)).isFalse();
+ assertThat(snapshotManager.snapshotExists(2)).isFalse();
+ for (int i = 3; i <= 6; i++) {
+ assertThat(snapshotManager.snapshotExists(i)).isTrue();
+ assertSnapshot(i, allData, snapshotPositions);
+ }
+ }
+
@Test
public void testExpireWithUpgradedFile() throws Exception {
// write & commit data
@@ -654,6 +1015,26 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ @Test
+ public void testExpireWithZeroFileOperationThreadNumCleansDataFiles()
throws Exception {
+
store.options().toConfiguration().set(CoreOptions.FILE_OPERATION_THREAD_NUM, 0);
+
+ List<KeyValue> data = FileStoreTestUtils.partitionedData(5, gen,
"0401", 8);
+ BinaryRow partition = gen.getPartition(data.get(0));
+ RecordWriter<KeyValue> writer = FileStoreTestUtils.writeData(store,
data, partition, 0);
+ Map<BinaryRow, Map<Integer, RecordWriter<KeyValue>>> writers =
+ Collections.singletonMap(partition,
Collections.singletonMap(0, writer));
+ FileStoreTestUtils.commitData(store, 0, writers);
+
+ writer.compact(true);
+ writer.sync();
+ FileStoreTestUtils.commitData(store, 1, writers);
+
+ store.newExpire(1, 1, Long.MAX_VALUE).expire();
+
+ store.assertCleaned();
+ }
+
@RepeatedTest(5)
public void testChangelogOutLivedSnapshot() throws Exception {
List<KeyValue> allData = new ArrayList<>();
@@ -779,6 +1160,14 @@ public class ExpireSnapshotsTest {
}
private TestFileStore createStore() {
+ return createStore(tempDir.toString());
+ }
+
+ private TestFileStore createSlowStore() {
+ return createStore(SlowFileIO.SCHEME + "://" + tempDir.toString());
+ }
+
+ private TestFileStore createStore(String root) {
ThreadLocalRandom random = ThreadLocalRandom.current();
CoreOptions.ChangelogProducer changelogProducer;
@@ -788,9 +1177,18 @@ public class ExpireSnapshotsTest {
changelogProducer = CoreOptions.ChangelogProducer.NONE;
}
+ return createStore(root, changelogProducer);
+ }
+
+ private TestFileStore createStore(CoreOptions.ChangelogProducer
changelogProducer) {
+ return createStore(tempDir.toString(), changelogProducer);
+ }
+
+ private TestFileStore createStore(
+ String root, CoreOptions.ChangelogProducer changelogProducer) {
return new TestFileStore.Builder(
"avro",
- tempDir.toString(),
+ root,
1,
TestKeyValueGenerator.DEFAULT_PART_TYPE,
TestKeyValueGenerator.KEY_TYPE,
@@ -802,7 +1200,29 @@ public class ExpireSnapshotsTest {
.build();
}
+ private ExpireConfig expireAllButLatestConfig() {
+ return ExpireConfig.builder()
+ .snapshotRetainMin(1)
+ .snapshotRetainMax(Integer.MAX_VALUE)
+ .snapshotTimeRetain(Duration.ofMillis(1))
+ .build();
+ }
+
+ private ExpireSnapshotsImpl newExpireWithSnapshotDeletion(
+ TestFileStore store,
+ SnapshotManager snapshotManager,
+ SnapshotDeletion snapshotDeletion) {
+ return new ExpireSnapshotsImpl(
+ snapshotManager, store.changelogManager(), snapshotDeletion,
store.newTagManager());
+ }
+
private void rewriteSnapshotTime(long snapshotId, long newTimeMillis)
throws IOException {
+ rewriteSnapshotTime(fileIO, snapshotManager, snapshotId,
newTimeMillis);
+ }
+
+ private void rewriteSnapshotTime(
+ FileIO fileIO, SnapshotManager snapshotManager, long snapshotId,
long newTimeMillis)
+ throws IOException {
String oldJson =
fileIO.readFileUtf8(snapshotManager.snapshotPath(snapshotId));
ObjectNode node = (ObjectNode)
JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(oldJson);
node.put("timeMillis", newTimeMillis);
@@ -811,8 +1231,275 @@ public class ExpireSnapshotsTest {
snapshotManager.invalidateCache();
}
+ private static class FailingSnapshotManager extends SnapshotManager {
+
+ private final long failedSnapshotId;
+ private boolean readFailedSnapshot;
+
+ private FailingSnapshotManager(SnapshotManager snapshotManager, long
failedSnapshotId) {
+ super(
+ snapshotManager.fileIO(),
+ snapshotManager.tablePath(),
+ snapshotManager.branch(),
+ null,
+ null);
+ this.failedSnapshotId = failedSnapshotId;
+ }
+
+ @Override
+ public Snapshot tryGetSnapshot(long snapshotId) throws
java.io.FileNotFoundException {
+ if (snapshotId == failedSnapshotId) {
+ readFailedSnapshot = true;
+ throw new RuntimeException("Unexpected snapshot read.");
+ }
+ return super.tryGetSnapshot(snapshotId);
+ }
+
+ private boolean readFailedSnapshot() {
+ return readFailedSnapshot;
+ }
+ }
+
+ private static class BlockingSnapshotManager extends SnapshotManager {
+
+ private final long minBlockedSnapshotId;
+ private final long maxBlockedSnapshotId;
+ private final CountDownLatch releaseReads = new CountDownLatch(1);
+ private final AtomicInteger activeReads = new AtomicInteger();
+ private final AtomicInteger maxActiveReads = new AtomicInteger();
+
+ private BlockingSnapshotManager(
+ SnapshotManager snapshotManager,
+ long minBlockedSnapshotId,
+ long maxBlockedSnapshotId) {
+ super(
+ snapshotManager.fileIO(),
+ snapshotManager.tablePath(),
+ snapshotManager.branch(),
+ null,
+ null);
+ this.minBlockedSnapshotId = minBlockedSnapshotId;
+ this.maxBlockedSnapshotId = maxBlockedSnapshotId;
+ }
+
+ @Override
+ public Snapshot tryGetSnapshot(long snapshotId) throws
java.io.FileNotFoundException {
+ if (snapshotId < minBlockedSnapshotId
+ || snapshotId > maxBlockedSnapshotId
+ || releaseReads.getCount() == 0) {
+ return super.tryGetSnapshot(snapshotId);
+ }
+
+ int active = activeReads.incrementAndGet();
+ updateMaxActiveReads(active);
+ if (active > 1) {
+ releaseReads.countDown();
+ }
+ try {
+ if (!releaseReads.await(5, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Snapshots were not read
concurrently.");
+ }
+ return super.tryGetSnapshot(snapshotId);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } finally {
+ activeReads.decrementAndGet();
+ }
+ }
+
+ private void updateMaxActiveReads(int active) {
+ int current;
+ do {
+ current = maxActiveReads.get();
+ if (active <= current) {
+ return;
+ }
+ } while (!maxActiveReads.compareAndSet(current, active));
+ }
+
+ private int maxActiveReads() {
+ return maxActiveReads.get();
+ }
+ }
+
+ private static class BlockingSnapshotDeletion extends SnapshotDeletion {
+
+ private final long minBlockedSnapshotId;
+ private final long maxBlockedSnapshotId;
+ private final ConcurrentCallTracker dataFilePlans =
+ new ConcurrentCallTracker("Data file plans were not created
concurrently.");
+ private final ConcurrentCallTracker changelogPlans =
+ new ConcurrentCallTracker("Changelog plans were not created
concurrently.");
+ private final ConcurrentCallTracker manifestPlans =
+ new ConcurrentCallTracker("Manifest plans were not created
concurrently.");
+
+ private boolean blockDataFilePlans;
+ private boolean blockManifestPlans;
+ private Set<String> blockedChangelogManifestLists =
Collections.emptySet();
+
+ private BlockingSnapshotDeletion(
+ TestFileStore store, long minBlockedSnapshotId, long
maxBlockedSnapshotId) {
+ 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());
+ this.minBlockedSnapshotId = minBlockedSnapshotId;
+ this.maxBlockedSnapshotId = maxBlockedSnapshotId;
+ }
+
+ private void blockDataFilePlans() {
+ blockDataFilePlans = true;
+ }
+
+ private void blockChangelogPlans(Set<String> changelogManifestLists) {
+ blockedChangelogManifestLists = changelogManifestLists;
+ }
+
+ private void blockManifestPlans() {
+ blockManifestPlans = true;
+ }
+
+ @Override
+ public List<Path> planDeletedInDeltaManifest(
+ Snapshot snapshot, Predicate<ExpireFileEntry> skipper) {
+ if (blockDataFilePlans && shouldBlock(snapshot.id())) {
+ dataFilePlans.awaitConcurrentCall();
+ }
+ return super.planDeletedInDeltaManifest(snapshot, skipper);
+ }
+
+ @Override
+ public List<Path> planAddedInChangelogManifest(Snapshot snapshot) {
+ if
(blockedChangelogManifestLists.contains(snapshot.changelogManifestList())) {
+ changelogPlans.awaitConcurrentCall();
+ }
+ return super.planAddedInChangelogManifest(snapshot);
+ }
+
+ @Override
+ public List<Runnable> planManifestsCleaner(Snapshot snapshot,
Set<String> skippingSet) {
+ if (blockManifestPlans && shouldBlock(snapshot.id())) {
+ manifestPlans.awaitConcurrentCall();
+ }
+ return super.planManifestsCleaner(snapshot, skippingSet);
+ }
+
+ private boolean shouldBlock(long snapshotId) {
+ return snapshotId >= minBlockedSnapshotId && snapshotId <=
maxBlockedSnapshotId;
+ }
+
+ private int maxActiveDataFilePlans() {
+ return dataFilePlans.maxActiveCalls();
+ }
+
+ private int maxActiveChangelogPlans() {
+ return changelogPlans.maxActiveCalls();
+ }
+
+ private int maxActiveManifestPlans() {
+ return manifestPlans.maxActiveCalls();
+ }
+ }
+
+ private static class CapturingSnapshotDeletion extends SnapshotDeletion {
+
+ private final List<List<Object>> deleteBatches = new ArrayList<>();
+
+ private CapturingSnapshotDeletion(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
+ protected <F> void executeAll(
+ Collection<F> files, java.util.function.Consumer<F> deletion) {
+ if (!files.isEmpty()) {
+ deleteBatches.add(new ArrayList<>(files));
+ }
+ }
+
+ private void assertDeleteBatchesDeduplicated() {
+ assertThat(deleteBatches).isNotEmpty();
+ for (List<Object> batch : deleteBatches) {
+ assertThat(new HashSet<>(batch)).hasSize(batch.size());
+ }
+ }
+
+ private void reset() {
+ deleteBatches.clear();
+ }
+ }
+
+ private static class ConcurrentCallTracker {
+
+ private final String timeoutMessage;
+ private final CountDownLatch releaseCalls = new CountDownLatch(1);
+ private final AtomicInteger activeCalls = new AtomicInteger();
+ private final AtomicInteger maxActiveCalls = new AtomicInteger();
+
+ private ConcurrentCallTracker(String timeoutMessage) {
+ this.timeoutMessage = timeoutMessage;
+ }
+
+ private void awaitConcurrentCall() {
+ int active = activeCalls.incrementAndGet();
+ updateMaxActiveCalls(active);
+ if (active > 1) {
+ releaseCalls.countDown();
+ }
+ try {
+ if (!releaseCalls.await(5, TimeUnit.SECONDS)) {
+ throw new RuntimeException(timeoutMessage);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } finally {
+ activeCalls.decrementAndGet();
+ }
+ }
+
+ private void updateMaxActiveCalls(int active) {
+ int current;
+ do {
+ current = maxActiveCalls.get();
+ if (active <= current) {
+ return;
+ }
+ } while (!maxActiveCalls.compareAndSet(current, active));
+ }
+
+ private int maxActiveCalls() {
+ return maxActiveCalls.get();
+ }
+ }
+
protected void commit(int numCommits, List<KeyValue> allData,
List<Integer> snapshotPositions)
throws Exception {
+ commit(store, numCommits, allData, snapshotPositions);
+ }
+
+ protected void commit(
+ TestFileStore store,
+ int numCommits,
+ List<KeyValue> allData,
+ List<Integer> snapshotPositions)
+ throws Exception {
for (int i = 0; i < numCommits; i++) {
int numRecords = ThreadLocalRandom.current().nextInt(100) + 1;
List<KeyValue> data = new ArrayList<>();
@@ -830,12 +1517,31 @@ public class ExpireSnapshotsTest {
protected void assertSnapshot(
int snapshotId, List<KeyValue> allData, List<Integer>
snapshotPositions)
throws Exception {
- assertSnapshot(snapshotManager.snapshot(snapshotId), allData,
snapshotPositions);
+ assertSnapshot(store, snapshotManager.snapshot(snapshotId), allData,
snapshotPositions);
+ }
+
+ protected void assertSnapshot(
+ TestFileStore store,
+ int snapshotId,
+ List<KeyValue> allData,
+ List<Integer> snapshotPositions)
+ throws Exception {
+ assertSnapshot(
+ store, store.snapshotManager().snapshot(snapshotId), allData,
snapshotPositions);
}
protected void assertSnapshot(
Snapshot snapshot, List<KeyValue> allData, List<Integer>
snapshotPositions)
throws Exception {
+ assertSnapshot(store, snapshot, allData, snapshotPositions);
+ }
+
+ protected void assertSnapshot(
+ TestFileStore store,
+ Snapshot snapshot,
+ List<KeyValue> allData,
+ List<Integer> snapshotPositions)
+ throws Exception {
int snapshotId = (int) snapshot.id();
Map<BinaryRow, BinaryRow> expected =
store.toKvMap(allData.subList(0,
snapshotPositions.get(snapshotId - 1)));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/ManifestReadThreadPoolTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/ManifestReadThreadPoolTest.java
index 8c131bdc52..7664660694 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/ManifestReadThreadPoolTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/ManifestReadThreadPoolTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import static java.util.Collections.emptyList;
@@ -128,6 +129,16 @@ public class ManifestReadThreadPoolTest {
re.forEach(i -> Assertions.assertThat(i).isEqualTo(2));
}
+ @Test
+ public void testNonPositiveThreadNumUsesDefaultExecutor() {
+ ExecutorService executor =
ManifestReadThreadPool.getExecutorService(0);
+
Assertions.assertThat(executor).isNotInstanceOf(SemaphoredDelegatingExecutor.class);
+
+ Iterable<Integer> re =
+ sequentialBatchedExecute(i -> singletonList(i + 1),
singletonList(1), 0);
+ re.forEach(i -> Assertions.assertThat(i).isEqualTo(2));
+ }
+
@Test
public void testDifferentQueueSizeWithFilterElement() {
for (int queueSize = 1; queueSize < 20; queueSize++) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SlowFileIO.java
b/paimon-core/src/test/java/org/apache/paimon/utils/SlowFileIO.java
new file mode 100644
index 0000000000..a8f68cd24d
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SlowFileIO.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileIOLoader;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** A local {@link FileIO} which adds configurable latency to each file
operation. */
+public class SlowFileIO extends LocalFileIO {
+
+ public static final String SCHEME = "slow-expire";
+
+ private static final AtomicLong DELAY_MILLIS = new AtomicLong();
+ private static final AtomicLong DELAYED_OPERATIONS = new AtomicLong();
+ private static final AtomicInteger ACTIVE_OPERATIONS = new AtomicInteger();
+ private static final AtomicInteger MAX_ACTIVE_OPERATIONS = new
AtomicInteger();
+
+ public static void reset() {
+ DELAY_MILLIS.set(0);
+ DELAYED_OPERATIONS.set(0);
+ ACTIVE_OPERATIONS.set(0);
+ MAX_ACTIVE_OPERATIONS.set(0);
+ }
+
+ public static void setDelayMillis(long delayMillis) {
+ DELAY_MILLIS.set(delayMillis);
+ }
+
+ public static long delayedOperations() {
+ return DELAYED_OPERATIONS.get();
+ }
+
+ public static int maxActiveOperations() {
+ return MAX_ACTIVE_OPERATIONS.get();
+ }
+
+ @Override
+ public boolean isObjectStore() {
+ return true;
+ }
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws IOException {
+ delay();
+ return super.newInputStream(path);
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean overwrite)
throws IOException {
+ delay();
+ return super.newOutputStream(path, overwrite);
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ delay();
+ return super.getFileStatus(path);
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ delay();
+ return super.listStatus(path);
+ }
+
+ @Override
+ public boolean exists(Path path) throws IOException {
+ delay();
+ return super.exists(path);
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) throws IOException {
+ delay();
+ return super.delete(path, recursive);
+ }
+
+ @Override
+ public boolean mkdirs(Path path) throws IOException {
+ delay();
+ return super.mkdirs(path);
+ }
+
+ @Override
+ public boolean rename(Path src, Path dst) throws IOException {
+ delay();
+ return super.rename(src, dst);
+ }
+
+ private void delay() throws IOException {
+ long delayMillis = DELAY_MILLIS.get();
+ if (delayMillis <= 0) {
+ return;
+ }
+
+ DELAYED_OPERATIONS.incrementAndGet();
+ int active = ACTIVE_OPERATIONS.incrementAndGet();
+ updateMaxActiveOperations(active);
+ try {
+ Thread.sleep(delayMillis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ } finally {
+ ACTIVE_OPERATIONS.decrementAndGet();
+ }
+ }
+
+ private static void updateMaxActiveOperations(int active) {
+ int current;
+ do {
+ current = MAX_ACTIVE_OPERATIONS.get();
+ if (active <= current) {
+ return;
+ }
+ } while (!MAX_ACTIVE_OPERATIONS.compareAndSet(current, active));
+ }
+
+ /** Loader for {@link SlowFileIO}. */
+ public static class Loader implements FileIOLoader {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String getScheme() {
+ return SCHEME;
+ }
+
+ @Override
+ public FileIO load(Path path) {
+ return new SlowFileIO();
+ }
+ }
+}
diff --git
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
index ae1021e74e..fd22396ea2 100644
---
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
+++
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
@@ -15,4 +15,5 @@
org.apache.paimon.utils.FailingFileIO$Loader
org.apache.paimon.utils.TraceableFileIO$Loader
+org.apache.paimon.utils.SlowFileIO$Loader
org.apache.paimon.rest.RESTFileIOTestLoader
