You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/12/01 07:03:11 UTC
(incubator-paimon) 37/46: [core][flink] Introduce orphan files clean and corresponding flink action (#2357)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit ba87b172106d8504f2f35b2ba118f446e007fa4b
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Thu Nov 30 17:22:25 2023 +0800
[core][flink] Introduce orphan files clean and corresponding flink action (#2357)
---
docs/content/maintenance/understand-files.md | 38 ++
.../java/org/apache/paimon/AbstractFileStore.java | 5 +-
.../src/main/java/org/apache/paimon/FileStore.java | 6 +
.../src/main/java/org/apache/paimon/Snapshot.java | 11 +
.../org/apache/paimon/index/IndexFileHandler.java | 5 +
.../org/apache/paimon/io/DataFilePathFactory.java | 4 +-
.../apache/paimon/operation/OrphanFilesClean.java | 438 ++++++++++++++++++
.../org/apache/paimon/utils/SnapshotManager.java | 52 +++
.../java/org/apache/paimon/utils/TagManager.java | 25 +-
.../paimon/operation/OrphanFilesCleanTest.java | 511 +++++++++++++++++++++
.../flink/action/RemoveOrphanFilesAction.java | 56 +++
.../action/RemoveOrphanFilesActionFactory.java | 70 +++
.../procedure/RemoveOrphanFilesProcedure.java | 74 +++
.../services/org.apache.paimon.factories.Factory | 2 +
.../action/RemoveOrphanFilesActionITCase.java | 73 +++
15 files changed, 1358 insertions(+), 12 deletions(-)
diff --git a/docs/content/maintenance/understand-files.md b/docs/content/maintenance/understand-files.md
index ceb2712e7..03e8705c8 100644
--- a/docs/content/maintenance/understand-files.md
+++ b/docs/content/maintenance/understand-files.md
@@ -337,6 +337,44 @@ expiration looks like:
As a result, partition `20230503` to `20230510` are physically deleted.
+### Remove Orphan Files
+
+Paimon files are deleted physically only when expiring snapshots. However, it is possible that some unexpected errors occurred
+when deleting files, so that there may exist files that are not used by Paimon table (so-called "orphan files"). You can
+submit a `remove-orphan-files` job to clean them:
+
+{{< tabs "remove-orphan-files" >}}
+
+{{< tab "Flink" >}}
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ remove-orphan-files \
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--older-than <timestamp>]
+```
+
+To avoid deleting files that are newly added by other writing jobs, this action only deletes orphan files older than
+1 day by default. The interval can be modified by `--older-than`. For example:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-{{< version >}}.jar \
+ remove-orphan-files \
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table T \
+ --older-than '2023-10-31 12:00:00'
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
+
### Flink Stream Write
Finally, we will examine Flink Stream Write by utilizing the example
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 46451888e..cf6809b95 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -18,7 +18,6 @@
package org.apache.paimon;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.index.HashIndexFile;
@@ -92,7 +91,7 @@ public abstract class AbstractFileStore<T> implements FileStore<T> {
return new SnapshotManager(fileIO, options.path());
}
- @VisibleForTesting
+ @Override
public ManifestFile.Factory manifestFileFactory() {
return manifestFileFactory(false);
}
@@ -108,7 +107,7 @@ public abstract class AbstractFileStore<T> implements FileStore<T> {
forWrite ? writeManifestCache : null);
}
- @VisibleForTesting
+ @Override
public ManifestList.Factory manifestListFactory() {
return manifestListFactory(false);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/FileStore.java b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
index 93d7b5ba4..e67cf9f1e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/FileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
@@ -20,6 +20,8 @@ package org.apache.paimon;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.manifest.ManifestCacheFilter;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.operation.FileStoreExpire;
import org.apache.paimon.operation.FileStoreRead;
@@ -58,6 +60,10 @@ public interface FileStore<T> extends Serializable {
FileStoreScan newScan();
+ ManifestList.Factory manifestListFactory();
+
+ ManifestFile.Factory manifestFileFactory();
+
IndexFileHandler newIndexFileHandler();
FileStoreRead<T> newRead();
diff --git a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
index 0bbeda7d1..9792aca9a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
@@ -34,12 +34,14 @@ import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonPro
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
/**
* This file is the entrance to all data committed at some specific time point.
@@ -421,6 +423,15 @@ public class Snapshot {
}
}
+ public static Optional<Snapshot> safelyFromPath(FileIO fileIO, Path path) throws IOException {
+ try {
+ String json = fileIO.readFileUtf8(path);
+ return Optional.of(Snapshot.fromJson(json));
+ } catch (FileNotFoundException e) {
+ return Optional.empty();
+ }
+ }
+
@Override
public int hashCode() {
return Objects.hash(
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index efba5492b..73b0d3416 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -132,6 +132,11 @@ public class IndexFileHandler {
return indexManifestFile.read(indexManifest);
}
+ public List<IndexManifestEntry> readManifestWithIOException(String indexManifest)
+ throws IOException {
+ return indexManifestFile.readWithIOException(indexManifest);
+ }
+
public boolean existsIndexFile(IndexManifestEntry file) {
return hashIndex.exists(file.indexFile().fileName());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
index d4a49cc46..83aef4f84 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
@@ -34,6 +34,8 @@ public class DataFilePathFactory {
public static final String CHANGELOG_FILE_PREFIX = "changelog-";
+ public static final String BUCKET_PATH_PREFIX = "bucket-";
+
private final Path bucketDir;
private final String uuid;
@@ -71,7 +73,7 @@ public class DataFilePathFactory {
}
public static Path bucketPath(Path tablePath, String partition, int bucket) {
- return new Path(tablePath + "/" + partition + "/bucket-" + bucket);
+ return new Path(tablePath + "/" + partition + "/" + BUCKET_PATH_PREFIX + bucket);
}
public static String formatIdentifier(String fileName) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
new file mode 100644
index 000000000..81fb7f6d1
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -0,0 +1,438 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.DateTimeUtils;
+import org.apache.paimon.utils.FileUtils;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.io.DataFilePathFactory.BUCKET_PATH_PREFIX;
+
+/**
+ * To remove the data files and metadata files that are not used by table (so-called "orphan
+ * files").
+ *
+ * <p>It will ignore exception when listing all files because it's OK to not delete unread files.
+ *
+ * <p>To avoid deleting newly written files, it only deletes orphan files older than 1 day by
+ * default. The interval can be modified by {@link #olderThan}.
+ *
+ * <p>To avoid conflicting with snapshot expiration, tag deletion and rollback, it will skip the
+ * snapshot/tag when catching {@link FileNotFoundException} in the process of listing used files.
+ *
+ * <p>To avoid deleting files that are used but not read by mistaken, it will stop removing process
+ * when failed to read used files.
+ */
+public class OrphanFilesClean {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrphanFilesClean.class);
+
+ private static final int READ_FILE_RETRY_NUM = 3;
+ private static final int READ_FILE_RETRY_INTERVAL = 5;
+
+ private final SnapshotManager snapshotManager;
+ private final TagManager tagManager;
+ private final FileIO fileIO;
+ private final Path location;
+ private final int partitionKeysNum;
+ private final ManifestList manifestList;
+ private final ManifestFile manifestFile;
+ private final IndexFileHandler indexFileHandler;
+
+ // an estimated value of how many files were deleted
+ private int deletedFilesNum = 0;
+ private long olderThanMillis = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1);
+
+ public OrphanFilesClean(FileStoreTable table) {
+ this.snapshotManager = table.snapshotManager();
+ this.tagManager = table.tagManager();
+ this.fileIO = table.fileIO();
+ this.location = table.location();
+ this.partitionKeysNum = table.partitionKeys().size();
+
+ FileStore<?> store = table.store();
+ this.manifestList = store.manifestListFactory().create();
+ this.manifestFile = store.manifestFileFactory().create();
+ this.indexFileHandler = store.newIndexFileHandler();
+ }
+
+ public OrphanFilesClean olderThan(String timestamp) {
+ // The FileStatus#getModificationTime returns milliseconds
+ this.olderThanMillis =
+ DateTimeUtils.parseTimestampData(timestamp, 3, DateTimeUtils.LOCAL_TZ)
+ .getMillisecond();
+ return this;
+ }
+
+ public int clean() throws IOException, ExecutionException, InterruptedException {
+ if (snapshotManager.earliestSnapshotId() == null) {
+ LOG.info("No snapshot found, skip removing.");
+ return 0;
+ }
+
+ // specially handle the snapshot directory
+ List<Path> nonSnapshotFiles = snapshotManager.tryGetNonSnapshotFiles(this::oldEnough);
+ nonSnapshotFiles.forEach(this::deleteFileOrDirQuietly);
+ deletedFilesNum += nonSnapshotFiles.size();
+
+ Set<String> usedFiles = getUsedFiles();
+ Map<String, Path> candidates = getCandidateDeletingFiles();
+
+ Set<String> deleted = new HashSet<>(candidates.keySet());
+ deleted.removeAll(usedFiles);
+
+ for (String file : deleted) {
+ Path path = candidates.get(file);
+ deleteFileOrDirQuietly(path);
+ }
+ deletedFilesNum += deleted.size();
+
+ return deletedFilesNum;
+ }
+
+ /** Get all the files used by snapshots and tags. */
+ private Set<String> getUsedFiles()
+ throws IOException, ExecutionException, InterruptedException {
+ // safely get all snapshots to be read
+ Set<Snapshot> readSnapshots = new HashSet<>(snapshotManager.safelyGetAllSnapshots());
+ SortedMap<Snapshot, String> tags = tagManager.tags();
+ List<Snapshot> taggedSnapshots = new ArrayList<>(tags.keySet());
+ readSnapshots.addAll(taggedSnapshots);
+
+ return FileUtils.COMMON_IO_FORK_JOIN_POOL
+ .submit(
+ () ->
+ readSnapshots
+ .parallelStream()
+ .flatMap(
+ snapshot ->
+ getUsedFilesForSnapshot(snapshot).stream())
+ .collect(Collectors.toSet()))
+ .get();
+ }
+
+ /**
+ * Get all the candidate deleting files in the specified directories and filter them by
+ * olderThanMillis.
+ */
+ private Map<String, Path> getCandidateDeletingFiles() {
+ List<Path> fileDirs = listPaimonFileDirs();
+ try {
+ return FileUtils.COMMON_IO_FORK_JOIN_POOL
+ .submit(
+ () ->
+ fileDirs.parallelStream()
+ .flatMap(p -> tryBestListingDirs(p).stream())
+ .filter(this::oldEnough)
+ .map(FileStatus::getPath)
+ .collect(
+ Collectors.toMap(
+ Path::getName, Function.identity())))
+ .get();
+ } catch (ExecutionException | InterruptedException e) {
+ LOG.debug("Failed to get candidate deleting files.", e);
+ return Collections.emptyMap();
+ }
+ }
+
+ /**
+ * If getting null when reading some files, the snapshot/tag is being deleted, so just return an
+ * empty result.
+ */
+ private List<String> getUsedFilesForSnapshot(Snapshot snapshot) {
+ List<String> files = new ArrayList<>();
+ addManifestList(files, snapshot);
+
+ try {
+ // try to read manifests
+ List<ManifestFileMeta> manifestFileMetas =
+ retryReadingFiles(() -> readAllManifestsWithIOException(snapshot));
+ if (manifestFileMetas == null) {
+ return Collections.emptyList();
+ }
+ List<String> manifestFileName =
+ manifestFileMetas.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList());
+ files.addAll(manifestFileName);
+
+ // try to read data files
+ List<String> dataFiles = retryReadingDataFiles(manifestFileName);
+ if (dataFiles == null) {
+ return Collections.emptyList();
+ }
+ files.addAll(dataFiles);
+
+ // try to read index files
+ String indexManifest = snapshot.indexManifest();
+ if (indexManifest != null && indexFileHandler.existsManifest(indexManifest)) {
+ files.add(indexManifest);
+
+ List<IndexManifestEntry> indexManifestEntries =
+ retryReadingFiles(
+ () -> indexFileHandler.readManifestWithIOException(indexManifest));
+ if (indexManifestEntries == null) {
+ return Collections.emptyList();
+ }
+
+ indexManifestEntries.stream()
+ .map(IndexManifestEntry::indexFile)
+ .map(IndexFileMeta::fileName)
+ .forEach(files::add);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return files;
+ }
+
+ private void addManifestList(List<String> used, Snapshot snapshot) {
+ used.add(snapshot.baseManifestList());
+ used.add(snapshot.deltaManifestList());
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ used.add(changelogManifestList);
+ }
+ }
+
+ /**
+ * Retry reading files when {@link IOException} was thrown by the reader. If the exception is
+ * {@link FileNotFoundException}, return null. Finally, if retry times reaches the limits,
+ * rethrow the IOException.
+ */
+ @Nullable
+ private <T> T retryReadingFiles(ReaderWithIOException<T> reader) throws IOException {
+ int retryNumber = 0;
+ IOException caught = null;
+ while (retryNumber++ < READ_FILE_RETRY_NUM) {
+ try {
+ return reader.read();
+ } catch (FileNotFoundException e) {
+ return null;
+ } catch (IOException e) {
+ caught = e;
+ }
+ try {
+ TimeUnit.MILLISECONDS.sleep(READ_FILE_RETRY_INTERVAL);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ throw caught;
+ }
+
+ private List<ManifestFileMeta> readAllManifestsWithIOException(Snapshot snapshot)
+ throws IOException {
+ List<ManifestFileMeta> result = new ArrayList<>();
+
+ result.addAll(manifestList.readWithIOException(snapshot.baseManifestList()));
+ result.addAll(manifestList.readWithIOException(snapshot.deltaManifestList()));
+
+ String changelogManifestList = snapshot.changelogManifestList();
+ if (changelogManifestList != null) {
+ result.addAll(manifestList.readWithIOException(changelogManifestList));
+ }
+
+ return result;
+ }
+
+ @Nullable
+ private List<String> retryReadingDataFiles(List<String> manifestNames) throws IOException {
+ List<String> dataFiles = new ArrayList<>();
+ for (String manifestName : manifestNames) {
+ List<ManifestEntry> manifestEntries =
+ retryReadingFiles(() -> manifestFile.readWithIOException(manifestName));
+ if (manifestEntries == null) {
+ return null;
+ }
+
+ manifestEntries.stream()
+ .map(ManifestEntry::file)
+ .forEach(
+ f -> {
+ dataFiles.add(f.fileName());
+ dataFiles.addAll(f.extraFiles());
+ });
+ }
+ return dataFiles;
+ }
+
+ /** List directories that contains data files and manifest files. */
+ private List<Path> listPaimonFileDirs() {
+ List<Path> paimonFileDirs = new ArrayList<>();
+
+ paimonFileDirs.add(new Path(location, "manifest"));
+ paimonFileDirs.add(new Path(location, "index"));
+ paimonFileDirs.addAll(listAndCleanDataDirs(location, partitionKeysNum));
+
+ return paimonFileDirs;
+ }
+
+ /**
+ * If failed to list directory, just return an empty result because it's OK to not delete them.
+ */
+ private List<FileStatus> tryBestListingDirs(Path dir) {
+ try {
+ if (!fileIO.exists(dir)) {
+ return Collections.emptyList();
+ }
+
+ List<FileStatus> status =
+ retryReadingFiles(
+ () -> {
+ FileStatus[] s = fileIO.listStatus(dir);
+ return s == null ? Collections.emptyList() : Arrays.asList(s);
+ });
+ return status == null ? Collections.emptyList() : status;
+ } catch (IOException e) {
+ LOG.debug("Failed to list directory {}, skip it.", dir, e);
+ return Collections.emptyList();
+ }
+ }
+
+ private boolean oldEnough(FileStatus status) {
+ return status.getModificationTime() < olderThanMillis;
+ }
+
+ /**
+ * List directories that contains data files and may clean non Paimon data dirs/files. The
+ * argument level is used to control recursive depth.
+ */
+ private List<Path> listAndCleanDataDirs(Path dir, int level) {
+ List<FileStatus> dirs = tryBestListingDirs(dir);
+
+ if (level == 0) {
+ // return bucket paths
+ return filterAndCleanDataDirs(
+ dirs,
+ p -> p.getName().startsWith(BUCKET_PATH_PREFIX),
+ // if buckets are under partition, we can do clean
+ partitionKeysNum -> partitionKeysNum != 0);
+ }
+
+ List<Path> partitionPaths =
+ filterAndCleanDataDirs(
+ dirs,
+ p -> p.getName().contains("="),
+ // if partitions are under a parent partition, we can do clean
+ partitionKeysNum -> level != partitionKeysNum);
+
+ // dive into the next partition level
+ try {
+ return FileUtils.COMMON_IO_FORK_JOIN_POOL
+ .submit(
+ () ->
+ partitionPaths
+ .parallelStream()
+ .flatMap(
+ p ->
+ listAndCleanDataDirs(p, level - 1)
+ .stream())
+ .collect(Collectors.toList()))
+ .get();
+ } catch (ExecutionException | InterruptedException e) {
+ LOG.debug("Failed to list partition directory {}", dir, e);
+ return Collections.emptyList();
+ }
+ }
+
+ private List<Path> filterAndCleanDataDirs(
+ List<FileStatus> statuses, Predicate<Path> filter, Predicate<Integer> cleanCondition) {
+ List<Path> filtered = new ArrayList<>();
+ List<FileStatus> mayBeClean = new ArrayList<>();
+
+ for (FileStatus status : statuses) {
+ Path path = status.getPath();
+ if (filter.test(path)) {
+ filtered.add(path);
+ } else {
+ mayBeClean.add(status);
+ }
+ }
+
+ if (cleanCondition.test(partitionKeysNum)) {
+ mayBeClean.stream()
+ .filter(this::oldEnough)
+ .map(FileStatus::getPath)
+ .forEach(
+ p -> {
+ deleteFileOrDirQuietly(p);
+ deletedFilesNum++;
+ });
+ }
+
+ return filtered;
+ }
+
+ private void deleteFileOrDirQuietly(Path path) {
+ try {
+ if (fileIO.isDir(path)) {
+ fileIO.deleteDirectoryQuietly(path);
+ } else {
+ fileIO.deleteQuietly(path);
+ }
+ } catch (IOException ignored) {
+ }
+ }
+
+ /** A helper functional interface for method {@link #retryReadingFiles}. */
+ @FunctionalInterface
+ private interface ReaderWithIOException<T> {
+
+ T read() throws IOException;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index 80f4a11b0..96d45d456 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -20,14 +20,17 @@ package org.apache.paimon.utils;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
@@ -38,6 +41,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.BinaryOperator;
import java.util.function.Predicate;
+import java.util.stream.Collectors;
import static org.apache.paimon.utils.FileUtils.listVersionedFiles;
@@ -199,6 +203,54 @@ public class SnapshotManager implements Serializable {
.iterator();
}
+ /**
+ * If {@link FileNotFoundException} is thrown when reading the snapshot file, this snapshot may
+ * be deleted by other processes, so just skip this snapshot.
+ */
+ public List<Snapshot> safelyGetAllSnapshots() throws IOException {
+ List<Path> paths =
+ listVersionedFiles(fileIO, snapshotDirectory(), SNAPSHOT_PREFIX)
+ .map(this::snapshotPath)
+ .collect(Collectors.toList());
+
+ List<Snapshot> snapshots = new ArrayList<>();
+ for (Path path : paths) {
+ Snapshot.safelyFromPath(fileIO, path).ifPresent(snapshots::add);
+ }
+
+ return snapshots;
+ }
+
+ /**
+ * Try to get non snapshot files. If any error occurred, just ignore it and return an empty
+ * result.
+ */
+ public List<Path> tryGetNonSnapshotFiles(Predicate<FileStatus> fileStatusFilter) {
+ try {
+ FileStatus[] statuses = fileIO.listStatus(snapshotDirectory());
+ if (statuses == null) {
+ return Collections.emptyList();
+ }
+
+ return Arrays.stream(statuses)
+ .filter(fileStatusFilter)
+ .map(FileStatus::getPath)
+ .filter(nonSnapshotFileFilter())
+ .collect(Collectors.toList());
+ } catch (IOException ignored) {
+ return Collections.emptyList();
+ }
+ }
+
+ private Predicate<Path> nonSnapshotFileFilter() {
+ return path -> {
+ String name = path.getName();
+ return !name.startsWith(SNAPSHOT_PREFIX)
+ && !name.equals(EARLIEST)
+ && !name.equals(LATEST);
+ };
+ }
+
public Optional<Snapshot> latestSnapshotOfUser(String user) {
Long latestId = latestSnapshotId();
if (latestId == null) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
index 8a46b5f24..927dbd373 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
@@ -20,6 +20,7 @@ package org.apache.paimon.utils;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.TagDeletion;
@@ -34,6 +35,7 @@ import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.function.Predicate;
+import java.util.stream.Collectors;
import static org.apache.paimon.utils.FileUtils.listVersionedFileStatus;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -179,14 +181,21 @@ public class TagManager {
public SortedMap<Snapshot, String> tags() {
TreeMap<Snapshot, String> tags = new TreeMap<>(Comparator.comparingLong(Snapshot::id));
try {
- listVersionedFileStatus(fileIO, tagDirectory(), TAG_PREFIX)
- .forEach(
- status -> {
- Path path = status.getPath();
- tags.put(
- Snapshot.fromPath(fileIO, path),
- path.getName().substring(TAG_PREFIX.length()));
- });
+ List<Path> paths =
+ listVersionedFileStatus(fileIO, tagDirectory(), TAG_PREFIX)
+ .map(FileStatus::getPath)
+ .collect(Collectors.toList());
+
+ for (Path path : paths) {
+ // If the tag file is not found, it might be deleted by
+ // other processes, so just skip this tag
+ Snapshot.safelyFromPath(fileIO, path)
+ .ifPresent(
+ snapshot ->
+ tags.put(
+ snapshot,
+ path.getName().substring(TAG_PREFIX.length())));
+ }
} catch (IOException e) {
throw new RuntimeException(e);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
new file mode 100644
index 000000000..759deb909
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
@@ -0,0 +1,511 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.DataFormatTestUtil;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.mergetree.compact.ConcatRecordReader;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.SchemaUtils;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.DateTimeUtils;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.StringUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.io.DataFilePathFactory.BUCKET_PATH_PREFIX;
+import static org.apache.paimon.io.DataFilePathFactory.CHANGELOG_FILE_PREFIX;
+import static org.apache.paimon.io.DataFilePathFactory.DATA_FILE_PREFIX;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link OrphanFilesClean}. */
+public class OrphanFilesCleanTest {
+
+ private static final Random RANDOM = new Random(System.currentTimeMillis());
+
+ @TempDir private java.nio.file.Path tempDir;
+ private Path tablePath;
+ private FileIO fileIO;
+ private RowType rowType;
+ private FileStoreTable table;
+ private TableWriteImpl<?> write;
+ private TableCommitImpl commit;
+ private Path manifestDir;
+
+ private long incrementalIdentifier;
+ private List<Path> manuallyAddedFiles;
+
+ @BeforeEach
+ public void beforeEach() throws Exception {
+ tablePath = new Path(tempDir.toString());
+ fileIO = LocalFileIO.create();
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(), DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING()
+ },
+ new String[] {"pk", "part1", "part2", "value"});
+ table = createFileStoreTable(rowType);
+ String commitUser = UUID.randomUUID().toString();
+ write = table.newWrite(commitUser);
+ commit = table.newCommit(commitUser);
+ manifestDir = new Path(tablePath, "manifest");
+
+ incrementalIdentifier = 0;
+ manuallyAddedFiles = new ArrayList<>();
+ }
+
+ @AfterEach
+ public void afterEach() throws Exception {
+ write.close();
+ commit.close();
+ TestPojo.reset();
+ }
+
+ @Test
+ public void testNormallyRemoving() throws Throwable {
+ int commitTimes = 30;
+ List<List<TestPojo>> committedData = new ArrayList<>();
+ Map<Long, List<TestPojo>> snapshotData = new HashMap<>();
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ // generate data of first snapshot
+ List<TestPojo> data = generateData();
+ commit(data);
+ committedData.add(data);
+ recordSnapshotData(data, snapshotData, snapshotManager);
+
+ // randomly generate data
+ for (int i = 1; i <= commitTimes; i++) {
+ List<TestPojo> previous =
+ new ArrayList<>(snapshotData.get(snapshotManager.latestSnapshotId()));
+ // randomly update
+ if (RANDOM.nextBoolean()) {
+ List<TestPojo> toBeUpdated = randomlyPick(previous);
+ List<TestPojo> updateAfter = commitUpdate(toBeUpdated);
+ committedData.add(updateAfter);
+
+ previous.removeAll(toBeUpdated);
+ previous.addAll(updateAfter);
+ recordSnapshotData(previous, snapshotData, snapshotManager);
+ } else {
+ List<TestPojo> current = generateData();
+ commit(current);
+ committedData.add(current);
+
+ current.addAll(previous);
+ recordSnapshotData(current, snapshotData, snapshotManager);
+ }
+ }
+
+ // randomly create tags
+ List<String> allTags = new ArrayList<>();
+ int snapshotCount = (int) snapshotManager.snapshotCount();
+ for (int i = 1; i <= snapshotCount; i++) {
+ if (RANDOM.nextBoolean()) {
+ String tagName = "tag" + i;
+ table.createTag(tagName, i);
+ allTags.add(tagName);
+ }
+ }
+
+ // generate non used files
+ int shouldBeDeleted = 0;
+ int fileNum = RANDOM.nextInt(10);
+ fileNum = fileNum == 0 ? 1 : fileNum;
+
+ // snapshot
+ addNonUsedFiles(
+ new Path(tablePath, "snapshot"), fileNum, Collections.singletonList("UNKNOWN"));
+ shouldBeDeleted += fileNum;
+
+ // data files
+ shouldBeDeleted += randomlyAddNonUsedDataFiles();
+
+ // manifests
+ addNonUsedFiles(
+ manifestDir,
+ fileNum,
+ Arrays.asList("manifest-list-", "manifest-", "index-manifest-", "UNKNOWN-"));
+ shouldBeDeleted += fileNum;
+
+ // randomly expire snapshots
+ int expired = RANDOM.nextInt(snapshotCount / 2);
+ expired = expired == 0 ? 1 : expired;
+ Options expireOptions = new Options();
+ expireOptions.set(CoreOptions.SNAPSHOT_EXPIRE_LIMIT, snapshotCount);
+ expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, snapshotCount - expired);
+ expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, snapshotCount - expired);
+ table.copy(expireOptions.toMap()).store().newExpire().expire();
+
+ // randomly delete tags
+ List<String> deleteTags = Collections.emptyList();
+ if (!allTags.isEmpty()) {
+ deleteTags = randomlyPick(allTags);
+ for (String tagName : deleteTags) {
+ table.deleteTag(tagName);
+ }
+ }
+
+ // first check, nothing will be deleted because the default olderThan interval is 1 day
+ OrphanFilesClean orphanFilesClean = new OrphanFilesClean(table);
+ assertThat(orphanFilesClean.clean()).isEqualTo(0);
+
+ // second check
+ orphanFilesClean = new OrphanFilesClean(table);
+ setOlderThan(orphanFilesClean);
+ int deletedActual = orphanFilesClean.clean();
+ try {
+ validate(deletedActual, shouldBeDeleted, snapshotData);
+ } catch (Throwable t) {
+ String tableOptions = "Table options:\n" + table.options();
+
+ String committed = "Committed data:";
+ for (int i = 0; i < committedData.size(); i++) {
+ String insertValues =
+ committedData.get(i).stream()
+ .map(TestPojo::toInsertValueString)
+ .collect(Collectors.joining(","));
+ committed = String.format("%s\n%d:{%s}", committed, i, insertValues);
+ }
+
+ String snapshot = "Snapshot expired: " + expired;
+
+ String tag =
+ String.format(
+ "Tags: created{%s}; deleted{%s}",
+ String.join(",", allTags), String.join(",", deleteTags));
+
+ String addedFile =
+ "Manually added file:\n"
+ + manuallyAddedFiles.stream()
+ .map(Path::toString)
+ .collect(Collectors.joining("\n"));
+
+ throw new Exception(
+ String.format(
+ "%s\n%s\n%s\n%s\n%s",
+ tableOptions, committed, snapshot, tag, addedFile),
+ t);
+ }
+ }
+
+ private void validate(
+ int actualDeleted, int shouldBeDeleted, Map<Long, List<TestPojo>> snapshotData)
+ throws Exception {
+ assertThat(actualDeleted).isEqualTo(shouldBeDeleted);
+
+ Set<Snapshot> snapshots = new HashSet<>();
+ table.snapshotManager().snapshots().forEachRemaining(snapshots::add);
+ snapshots.addAll(table.tagManager().taggedSnapshots());
+ List<Snapshot> sorted =
+ snapshots.stream()
+ .sorted(Comparator.comparingLong(Snapshot::id))
+ .collect(Collectors.toList());
+
+ for (Snapshot snapshot : sorted) {
+ try {
+ validateSnapshot(snapshot, snapshotData.get(snapshot.id()));
+ } catch (Exception e) {
+ throw new Exception("Failed to validate snapshot " + snapshot.id(), e);
+ }
+ }
+ }
+
+ private void validateSnapshot(Snapshot snapshot, List<TestPojo> data) throws Exception {
+ List<Split> splits = table.newSnapshotReader().withSnapshot(snapshot).read().splits();
+ List<ConcatRecordReader.ReaderSupplier<InternalRow>> readers = new ArrayList<>();
+ for (Split split : splits) {
+ readers.add(() -> table.newRead().createReader(split));
+ }
+ RecordReader<InternalRow> recordReader = ConcatRecordReader.create(readers);
+ RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(recordReader);
+ List<String> result = new ArrayList<>();
+ while (iterator.hasNext()) {
+ InternalRow rowData = iterator.next();
+ result.add(DataFormatTestUtil.toStringNoRowKind(rowData, rowType));
+ }
+ iterator.close();
+
+ assertThat(result).containsExactlyInAnyOrderElementsOf(TestPojo.formatData(data));
+ }
+
+ /** Manually make a FileNotFoundException to simulate snapshot expire while clean. */
+ @Test
+ public void testAbnormallyRemoving() throws Exception {
+ // generate randomly number of snapshots
+ int num = RANDOM.nextInt(5) + 1;
+ for (int i = 0; i < num; i++) {
+ commit(generateData());
+ }
+
+ // randomly delete a manifest file of snapshot 1
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Snapshot snapshot1 = snapshotManager.snapshot(1);
+
+ List<Path> manifests = new ArrayList<>();
+ ManifestList manifestList = table.store().manifestListFactory().create();
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+ snapshot1
+ .allManifests(manifestList)
+ .forEach(m -> manifests.add(pathFactory.toManifestFilePath(m.fileName())));
+
+ Path manifest = manifests.get(RANDOM.nextInt(manifests.size()));
+ fileIO.deleteQuietly(manifest);
+
+ OrphanFilesClean orphanFilesClean = new OrphanFilesClean(table);
+ setOlderThan(orphanFilesClean);
+ assertThat(orphanFilesClean.clean()).isGreaterThan(0);
+ }
+
+ private void setOlderThan(OrphanFilesClean orphanFilesClean) {
+ String timestamp =
+ DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(
+ System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2)),
+ 3);
+ orphanFilesClean.olderThan(timestamp);
+ }
+
+ private List<TestPojo> generateData() {
+ int num = RANDOM.nextInt(6) + 5;
+ List<TestPojo> data = new ArrayList<>(num);
+ for (int i = 0; i < num; i++) {
+ data.add(TestPojo.next());
+ }
+ return data;
+ }
+
+ private void commit(List<TestPojo> data) throws Exception {
+ for (TestPojo d : data) {
+ write.write(d.toRow(RowKind.INSERT));
+ }
+ commit.commit(incrementalIdentifier, write.prepareCommit(true, incrementalIdentifier));
+ incrementalIdentifier++;
+ }
+
+ private List<TestPojo> commitUpdate(List<TestPojo> updates) throws Exception {
+ List<TestPojo> after = new ArrayList<>();
+ for (TestPojo u : updates) {
+ write.write(u.toRow(RowKind.UPDATE_BEFORE));
+ TestPojo updateAfter = u.copyWithNewValue();
+ after.add(updateAfter);
+ write.write(updateAfter.toRow(RowKind.UPDATE_AFTER));
+ }
+ commit.commit(incrementalIdentifier, write.prepareCommit(true, incrementalIdentifier));
+ incrementalIdentifier++;
+ return after;
+ }
+
+ private void recordSnapshotData(
+ List<TestPojo> data,
+ Map<Long, List<TestPojo>> snapshotData,
+ SnapshotManager snapshotManager) {
+ Snapshot latest = snapshotManager.latestSnapshot();
+ if (latest.commitKind() == Snapshot.CommitKind.COMPACT) {
+ snapshotData.put(latest.id() - 1, data);
+ }
+ snapshotData.put(latest.id(), data);
+ }
+
+ private int randomlyAddNonUsedDataFiles() throws IOException {
+ int addedFiles = 0;
+ List<Path> part1 = listSubDirs(tablePath, p -> p.getName().contains("="));
+ // add non used file at partition part1
+ List<Path> corruptedPartitions = randomlyPick(part1);
+ for (Path path : corruptedPartitions) {
+ addNonUsedFiles(path, 1, Collections.singletonList("UNKNOWN"));
+ }
+ addedFiles += corruptedPartitions.size();
+
+ List<Path> part2 = new ArrayList<>();
+ for (Path path : part1) {
+ part2.addAll(listSubDirs(path, p -> p.getName().contains("=")));
+ }
+
+ List<Path> buckets = new ArrayList<>();
+ for (Path path : part2) {
+ buckets.addAll(listSubDirs(path, p -> p.getName().startsWith(BUCKET_PATH_PREFIX)));
+ }
+
+ // add files
+ List<Path> corruptedBuckets = randomlyPick(buckets);
+ for (Path path : corruptedBuckets) {
+ addNonUsedFiles(
+ path, 1, Arrays.asList(DATA_FILE_PREFIX, CHANGELOG_FILE_PREFIX, "UNKNOWN-"));
+ }
+ addedFiles += corruptedBuckets.size();
+
+ return addedFiles;
+ }
+
+ private <T> List<T> randomlyPick(List<T> list) {
+ int num = RANDOM.nextInt(list.size());
+ num = num == 0 ? 1 : num;
+ List<T> copy = new ArrayList<>(list);
+ List<T> picked = new ArrayList<>();
+ for (int i = 0; i < num; i++) {
+ int index = RANDOM.nextInt(copy.size());
+ picked.add(copy.remove(index));
+ }
+ return picked;
+ }
+
+ private void addNonUsedFiles(Path dir, int fileNum, List<String> fileNamePrefix)
+ throws IOException {
+ for (int i = 0; i < fileNum; i++) {
+ String fileName =
+ fileNamePrefix.get(RANDOM.nextInt(fileNamePrefix.size())) + UUID.randomUUID();
+ Path file = new Path(dir, fileName);
+ if (RANDOM.nextBoolean()) {
+ fileIO.writeFileUtf8(file, "");
+ } else {
+ fileIO.mkdirs(file);
+ }
+ manuallyAddedFiles.add(file);
+ }
+ }
+
+ private List<Path> listSubDirs(Path root, Predicate<Path> filter) throws IOException {
+ return Arrays.stream(fileIO.listStatus(root))
+ .map(FileStatus::getPath)
+ .filter(filter)
+ .collect(Collectors.toList());
+ }
+
+ private static class TestPojo {
+
+ private static int increaseKey = 0;
+
+ private final int pk;
+ // 0-2
+ private final int part1;
+ // A-C
+ private final String part2;
+ private final String value;
+
+ public TestPojo(int pk, int part1, String part2, String value) {
+ this.pk = pk;
+ this.part1 = part1;
+ this.part2 = part2;
+ this.value = value;
+ }
+
+ public InternalRow toRow(RowKind rowKind) {
+ return GenericRow.ofKind(
+ rowKind,
+ pk,
+ part1,
+ BinaryString.fromString(part2),
+ BinaryString.fromString(value));
+ }
+
+ public String toInsertValueString() {
+ return String.format("(%d, %d, '%s', '%s')", pk, part1, part2, value);
+ }
+
+ public TestPojo copyWithNewValue() {
+ return new TestPojo(pk, part1, part2, randomValue());
+ }
+
+ public static void reset() {
+ increaseKey = 0;
+ }
+
+ public static TestPojo next() {
+ int pk = increaseKey++;
+ int part1 = RANDOM.nextInt(3);
+ char c = (char) (RANDOM.nextInt(3) + 'A');
+ String part2 = String.valueOf(c);
+ String value = randomValue();
+ return new TestPojo(pk, part1, part2, value);
+ }
+
+ public static List<String> formatData(List<TestPojo> data) {
+ return data.stream().map(TestPojo::format).collect(Collectors.toList());
+ }
+
+ private String format() {
+ return String.format("%d, %d, %s, %s", pk, part1, part2, value);
+ }
+ }
+
+ private static String randomValue() {
+ return StringUtils.getRandomString(RANDOM, 5, 20, 'a', 'z');
+ }
+
+ private FileStoreTable createFileStoreTable(RowType rowType) throws Exception {
+ Options conf = new Options();
+ conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(CoreOptions.BUCKET, RANDOM.nextInt(3) + 1);
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new SchemaManager(fileIO, tablePath),
+ new Schema(
+ rowType.getFields(),
+ Arrays.asList("part1", "part2"),
+ Arrays.asList("pk", "part1", "part2"),
+ conf.toMap(),
+ ""));
+ return FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
new file mode 100644
index 000000000..59ddf26c0
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
@@ -0,0 +1,56 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.operation.OrphanFilesClean;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Action to remove the orphan data files and metadata files. */
+public class RemoveOrphanFilesAction extends TableActionBase {
+
+ private final OrphanFilesClean orphanFilesClean;
+
+ public RemoveOrphanFilesAction(
+ String warehouse,
+ String databaseName,
+ String tableName,
+ Map<String, String> catalogConfig) {
+ super(warehouse, databaseName, tableName, catalogConfig);
+
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports remove-orphan-files action. The table type is '%s'.",
+ table.getClass().getName());
+ this.orphanFilesClean = new OrphanFilesClean((FileStoreTable) table);
+ }
+
+ public RemoveOrphanFilesAction olderThan(String timestamp) {
+ this.orphanFilesClean.olderThan(timestamp);
+ return this;
+ }
+
+ @Override
+ public void run() throws Exception {
+ orphanFilesClean.clean();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
new file mode 100644
index 000000000..4e0ca0316
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.action;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.utils.MultipleParameterTool;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Factory to create {@link RemoveOrphanFilesAction}. */
+public class RemoveOrphanFilesActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "remove-orphan-files";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterTool params) {
+ Tuple3<String, String, String> tablePath = getTablePath(params);
+ Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
+
+ RemoveOrphanFilesAction action =
+ new RemoveOrphanFilesAction(
+ tablePath.f0, tablePath.f1, tablePath.f2, catalogConfig);
+
+ if (params.has("older-than")) {
+ action.olderThan(params.get("olderThan"));
+ }
+
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println(
+ "Action \"remove-orphan-files\" remove files which are not used by any snapshots or tags of a Paimon table.");
+ System.out.println();
+
+ System.out.println("Syntax:");
+ System.out.println(
+ " remove-orphan-files --warehouse <warehouse-path> --database <database-name> "
+ + "--table <table-name> [--older-than <timestamp>]");
+
+ System.out.println();
+ System.out.println(
+ "To avoid deleting newly written files, this action only deletes orphan files older than 1 day by default. "
+ + "The interval can be modified by '--older-than'. <timestamp> format: yyyy-MM-dd HH:mm:ss");
+ System.out.println();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
new file mode 100644
index 000000000..3a1acac1b
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
@@ -0,0 +1,74 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.operation.OrphanFilesClean;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Remove orphan files procedure. Usage:
+ *
+ * <pre><code>
+ * -- use the default file delete interval
+ * CALL sys.remove_orphan_files('tableId')
+ *
+ * -- use custom file delete interval
+ * CALL sys.remove_orphan_files('tableId', '2023-12-31 23:59:59')
+ * </code></pre>
+ */
+public class RemoveOrphanFilesProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "remove_orphan_files";
+
+ public String[] call(ProcedureContext procedureContext, String tableId) throws Exception {
+ return call(procedureContext, tableId, "");
+ }
+
+ public String[] call(ProcedureContext procedureContext, String tableId, String olderThan)
+ throws Exception {
+ Identifier identifier = Identifier.fromString(tableId);
+ Table table = catalog.getTable(identifier);
+
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports remove-orphan-files action. The table type is '%s'.",
+ table.getClass().getName());
+
+ OrphanFilesClean orphanFilesClean = new OrphanFilesClean((FileStoreTable) table);
+ if (!StringUtils.isBlank(olderThan)) {
+ orphanFilesClean.olderThan(olderThan);
+ }
+
+ int deleted = orphanFilesClean.clean();
+
+ return new String[] {"Deleted=" + deleted};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index bac5cdd83..fb483b73c 100644
--- a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -24,6 +24,7 @@ org.apache.paimon.flink.action.CreateTagActionFactory
org.apache.paimon.flink.action.DeleteTagActionFactory
org.apache.paimon.flink.action.ResetConsumerActionFactory
org.apache.paimon.flink.action.MigrateTableActionFactory
+org.apache.paimon.flink.action.RemoveOrphanFilesActionFactory
### procedure factories
org.apache.paimon.flink.procedure.CompactDatabaseProcedure
@@ -36,3 +37,4 @@ org.apache.paimon.flink.procedure.ResetConsumerProcedure
org.apache.paimon.flink.procedure.RollbackToProcedure
org.apache.paimon.flink.procedure.MigrateTableProcedure
org.apache.paimon.flink.procedure.MigrateFileProcedure
+org.apache.paimon.flink.procedure.RemoveOrphanFilesProcedure
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
new file mode 100644
index 000000000..1c7af1f0f
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.action;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** IT cases for {@link RemoveOrphanFilesAction}. */
+public class RemoveOrphanFilesActionITCase extends ActionITCaseBase {
+
+ @Test
+ public void testRunWithoutException() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(), DataTypes.STRING()},
+ new String[] {"k", "v"});
+
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ Collections.emptyMap());
+
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+
+ writeData(rowData(1L, BinaryString.fromString("Hi")));
+
+ RemoveOrphanFilesAction action =
+ new RemoveOrphanFilesAction(warehouse, database, tableName, Collections.emptyMap());
+ assertThatCode(action::run).doesNotThrowAnyException();
+ assertThatCode(() -> action.olderThan("2023-12-31 23:59:59").run())
+ .doesNotThrowAnyException();
+
+ String withoutOlderThan =
+ String.format("CALL sys.remove_orphan_files('%s.%s')", database, tableName);
+ assertThatCode(() -> callProcedure(withoutOlderThan)).doesNotThrowAnyException();
+
+ String withOlderThan =
+ String.format(
+ "CALL sys.remove_orphan_files('%s.%s', '2023-12-31 23:59:59')",
+ database, tableName);
+ assertThatCode(() -> callProcedure(withOlderThan)).doesNotThrowAnyException();
+ }
+}