| remove_unexisting_files |
diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java
new file mode 100644
index 000000000000..a30a0e991666
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java
@@ -0,0 +1,246 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Collects managed BLOB pack reachability from data-file {@code .blobref} sidecars.
+ *
+ * This collector does not scan snapshots or delete files. Callers such as orphan-file cleanup
+ * (and later snapshot expiration) supply data files and decide what to delete from {@link Result}.
+ */
+public class ManagedBlobReachabilityCollector {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ManagedBlobReachabilityCollector.class);
+
+ private static final int READ_RETRY_NUM = 3;
+ private static final int READ_RETRY_INTERVAL_MS = 5;
+
+ private final FileIO fileIO;
+
+ public ManagedBlobReachabilityCollector(FileIO fileIO) {
+ this.fileIO = fileIO;
+ }
+
+ /**
+ * Reads blobref extras of one data file. Extra files without a {@code .blobref} suffix are
+ * ignored. A listed sidecar that cannot be trusted marks the result unsafe, unless the data
+ * file itself is already gone: unmerged snapshot manifests can still contain {@code ADD}
+ * entries that snapshot expire has deleted, and those must not abort pack GC.
+ */
+ public Result fromDataFile(Path dataFile, List extraFiles) {
+ Result result = Result.empty();
+ if (extraFiles == null || extraFiles.isEmpty()) {
+ return result;
+ }
+ Path parent = dataFile.getParent();
+ Boolean dataFileExists = null;
+ for (String extra : extraFiles) {
+ if (extra == null || !extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ continue;
+ }
+ Path sidecar = new Path(parent, extra);
+ try {
+ result = result.merge(Result.of(readWithRetry(sidecar)));
+ } catch (IOException e) {
+ if (dataFileExists == null) {
+ dataFileExists = checkDataFileExists(dataFile);
+ }
+ if (!dataFileExists) {
+ LOG.debug(
+ "Ignore unreadable blobref {} because data file {} is already gone.",
+ sidecar,
+ dataFile);
+ continue;
+ }
+ LOG.warn(
+ "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.",
+ sidecar,
+ e);
+ return Result.unsafe();
+ }
+ }
+ return result;
+ }
+
+ private boolean checkDataFileExists(Path dataFile) {
+ try {
+ return fileIO.exists(dataFile);
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to check existence of {}, treat as present for managed blob GC.",
+ dataFile,
+ e);
+ return true;
+ }
+ }
+
+ /**
+ * Reads one sidecar. Missing, corrupt, or unsupported files are unsafe rather than thrown to
+ * the caller.
+ */
+ public Result fromSidecar(Path sidecar) {
+ try {
+ List references = readWithRetry(sidecar);
+ return Result.of(references);
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.",
+ sidecar,
+ e);
+ return Result.unsafe();
+ }
+ }
+
+ private List readWithRetry(Path sidecar) throws IOException {
+ IOException caught = null;
+ for (int retry = 0; retry < READ_RETRY_NUM; retry++) {
+ try {
+ return ManagedBlobReferenceFile.read(fileIO, sidecar);
+ } catch (FileNotFoundException e) {
+ throw e;
+ } catch (IOException e) {
+ caught = e;
+ }
+ try {
+ TimeUnit.MILLISECONDS.sleep(READ_RETRY_INTERVAL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while reading " + sidecar, e);
+ }
+ }
+ throw caught;
+ }
+
+ /** Reachability of managed BLOB packs from one or more data files. */
+ public static final class Result {
+
+ private static final Result EMPTY = new Result(Collections.emptySet(), false);
+ private static final Result UNSAFE = new Result(Collections.emptySet(), true);
+
+ private final Set referenced;
+ private final boolean unsafe;
+
+ private Result(Set referenced, boolean unsafe) {
+ this.referenced = referenced;
+ this.unsafe = unsafe;
+ }
+
+ public static Result empty() {
+ return EMPTY;
+ }
+
+ public static Result unsafe() {
+ return UNSAFE;
+ }
+
+ public static Result of(List refs) {
+ if (refs == null || refs.isEmpty()) {
+ return empty();
+ }
+ return new Result(Collections.unmodifiableSet(new HashSet<>(refs)), false);
+ }
+
+ public Set referenced() {
+ return referenced;
+ }
+
+ public boolean isUnsafe() {
+ return unsafe;
+ }
+
+ public boolean contains(Reference ref) {
+ return referenced.contains(ref);
+ }
+
+ public boolean containsPackName(String fileName) {
+ for (Reference reference : referenced) {
+ if (reference.relativePath().equals(fileName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public Result merge(Result other) {
+ if (other == null) {
+ return this;
+ }
+ boolean mergedUnsafe = unsafe || other.unsafe;
+ if (referenced.isEmpty() && other.referenced.isEmpty()) {
+ return mergedUnsafe ? unsafe() : empty();
+ }
+ Set refs;
+ if (referenced.isEmpty()) {
+ refs = other.referenced;
+ } else if (other.referenced.isEmpty()) {
+ refs = referenced;
+ } else {
+ refs = new HashSet<>(referenced);
+ refs.addAll(other.referenced);
+ refs = Collections.unmodifiableSet(refs);
+ }
+ if (mergedUnsafe == unsafe && refs == referenced) {
+ return this;
+ }
+ if (mergedUnsafe == other.unsafe && refs == other.referenced) {
+ return other;
+ }
+ return new Result(refs, mergedUnsafe);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Result result = (Result) o;
+ return unsafe == result.unsafe && Objects.equals(referenced, result.referenced);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(referenced, unsafe);
+ }
+
+ @Override
+ public String toString() {
+ return "Result{unsafe=" + unsafe + ", referenced=" + referenced + '}';
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
index f7b10512c65b..e540fb05cd94 100644
--- a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java
@@ -170,6 +170,10 @@ public String relativePath() {
return relativePath;
}
+ public Path toPath() {
+ return new Path(storageRootId, relativePath);
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java
new file mode 100644
index 000000000000..00e2134fe01a
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java
@@ -0,0 +1,238 @@
+/*
+ * 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.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
+import static org.apache.paimon.utils.ThreadPoolUtils.randomlyExecuteSequentialReturn;
+import static org.apache.paimon.utils.ThreadPoolUtils.randomlyOnlyExecute;
+
+/** Local {@link ManagedBlobOrphanFilesClean}. */
+public class LocalManagedBlobOrphanFilesClean extends ManagedBlobOrphanFilesClean {
+
+ private final ThreadPoolExecutor executor;
+ private final List deleteFiles = new ArrayList<>();
+ private final AtomicLong deletedFilesLenInBytes = new AtomicLong(0);
+
+ public LocalManagedBlobOrphanFilesClean(
+ FileStoreTable table, long olderThanMillis, boolean dryRun) {
+ super(table, olderThanMillis, dryRun);
+ this.executor =
+ createCachedThreadPool(
+ table.coreOptions().fileOperationThreadNum(),
+ "MANAGED_BLOB_ORPHAN_FILES_CLEAN");
+ }
+
+ public CleanOrphanFilesResult clean() throws IOException {
+ Map> candidates = getCandidatePacks();
+ if (candidates.isEmpty()) {
+ return new CleanOrphanFilesResult(0, 0, deleteFiles);
+ }
+
+ List topologyBefore = snapshotTopology();
+ Set usedPacks = collectUsedPacks();
+ betweenUsedCollections();
+ Set usedPacks2 = collectUsedPacks();
+ if (shouldAbortPackGc(topologyBefore, usedPacks, usedPacks2)) {
+ return new CleanOrphanFilesResult(0, 0, deleteFiles);
+ }
+
+ candidates.entrySet().stream()
+ .filter(e -> !usedPacks2.contains(e.getKey()))
+ .map(Map.Entry::getValue)
+ .forEach(
+ info -> {
+ deletedFilesLenInBytes.addAndGet(info.getRight());
+ cleanFile(info.getLeft());
+ deleteFiles.add(info.getLeft());
+ });
+
+ if (!dryRun) {
+ cleanEmptyDataDirectory(deleteFiles);
+ }
+ return new CleanOrphanFilesResult(
+ deleteFiles.size(), deletedFilesLenInBytes.get(), deleteFiles);
+ }
+
+ @Override
+ protected Set collectUsedPacks() {
+ return validBranches().stream()
+ .flatMap(branch -> getUsedPacks(branch).stream())
+ .collect(Collectors.toSet());
+ }
+
+ private Set getUsedPacks(String branch) {
+ Set used = ConcurrentHashMap.newKeySet();
+ try {
+ randomlyOnlyExecute(
+ executor,
+ snapshot -> {
+ try {
+ emitUsedPacks(branch, snapshot, used::add);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ safelyGetAllSnapshots(branch));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return used;
+ }
+
+ private Map> getCandidatePacks() {
+ List fileDirs = listPaimonFileDirs();
+ Iterator> packs =
+ randomlyExecuteSequentialReturn(executor, packLister(), fileDirs);
+ Map> result = new HashMap<>();
+ while (packs.hasNext()) {
+ Pair fileInfo = packs.next();
+ result.put(packIdentity(fileInfo.getLeft()), fileInfo);
+ }
+ return result;
+ }
+
+ private Function>> packLister() {
+ return path ->
+ tryBestListingDirs(path).stream()
+ .filter(status -> !status.isDir())
+ .filter(this::oldEnough)
+ .filter(status -> isManagedBlobPackName(status.getPath().getName()))
+ .map(status -> Pair.of(status.getPath(), status.getLen()))
+ .collect(Collectors.toList());
+ }
+
+ private void cleanEmptyDataDirectory(List deleted) {
+ if (deleted.isEmpty()) {
+ return;
+ }
+ Set bucketDirs =
+ deleted.stream()
+ .map(Path::getParent)
+ .filter(path -> path.toString().contains(BUCKET_PATH_PREFIX))
+ .collect(Collectors.toSet());
+ randomlyOnlyExecute(executor, this::tryDeleteEmptyDirectory, bucketDirs);
+ Set partitionDirs =
+ bucketDirs.stream().map(Path::getParent).collect(Collectors.toSet());
+ tryCleanDataDirectory(partitionDirs, partitionKeysNum);
+ }
+
+ public static List createCleans(
+ Catalog catalog,
+ String databaseName,
+ @Nullable String tableName,
+ long olderThanMillis,
+ @Nullable Integer parallelism,
+ boolean dryRun)
+ throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
+ List tableNames = Collections.singletonList(tableName);
+ if (tableName == null || "*".equals(tableName)) {
+ tableNames = catalog.listTables(databaseName);
+ }
+
+ Map dynamicOptions =
+ parallelism == null
+ ? Collections.emptyMap()
+ : new HashMap() {
+ {
+ put(
+ CoreOptions.FILE_OPERATION_THREAD_NUM.key(),
+ parallelism.toString());
+ }
+ };
+
+ List cleans = new ArrayList<>(tableNames.size());
+ for (String t : tableNames) {
+ Identifier identifier = new Identifier(databaseName, t);
+ Table table = catalog.getTable(identifier).copy(dynamicOptions);
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports remove-orphan-blobs action. The table type is '%s'.",
+ table.getClass().getName());
+ cleans.add(
+ new LocalManagedBlobOrphanFilesClean(
+ (FileStoreTable) table, olderThanMillis, dryRun));
+ }
+ return cleans;
+ }
+
+ public static CleanOrphanFilesResult executeDatabase(
+ Catalog catalog,
+ String databaseName,
+ @Nullable String tableName,
+ long olderThanMillis,
+ @Nullable Integer parallelism,
+ boolean dryRun)
+ throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
+ List tableCleans =
+ createCleans(
+ catalog, databaseName, tableName, olderThanMillis, parallelism, dryRun);
+ ExecutorService executorService =
+ Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+ List> tasks = new ArrayList<>(tableCleans.size());
+ for (LocalManagedBlobOrphanFilesClean clean : tableCleans) {
+ tasks.add(executorService.submit(clean::clean));
+ }
+
+ long deletedFileCount = 0;
+ long deletedFileTotalLenInBytes = 0;
+ for (Future task : tasks) {
+ try {
+ deletedFileCount += task.get().getDeletedFileCount();
+ deletedFileTotalLenInBytes += task.get().getDeletedFileTotalLenInBytes();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ executorService.shutdownNow();
+ return new CleanOrphanFilesResult(deletedFileCount, deletedFileTotalLenInBytes);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java
new file mode 100644
index 000000000000..c7e055bd1d27
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java
@@ -0,0 +1,261 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.blob.ManagedBlobReachabilityCollector;
+import org.apache.paimon.blob.ManagedBlobReachabilityCollector.Result;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.FileKind;
+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.DataFilePathFactories;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Consumer;
+
+/**
+ * Cleans unreferenced primary-key {@code .managed.blob} packs.
+ *
+ * Unlike {@link OrphanFilesClean}, this cleaner only lists and deletes managed BLOB packs. Pack
+ * reachability is collected from live {@link FileKind#ADD} data-file {@code .blobref} sidecars.
+ * Missing manifest lists or unreadable sidecars on a still-existing data file abort pack deletion
+ * for the rest of the run.
+ *
+ * Used packs are collected twice. If the snapshot topology or the used-pack set changes between
+ * those collections, this run deletes nothing. That shrinks the race with compaction reuse; it is
+ * not a commit lease.
+ */
+public abstract class ManagedBlobOrphanFilesClean extends OrphanFilesClean {
+
+ /**
+ * Marker emitted into the used-pack set when a {@code .blobref} sidecar or a required manifest
+ * cannot be trusted. Callers must skip deleting every {@code .managed.blob} pack.
+ */
+ public static final String SKIP_MANAGED_BLOB_GC = "__paimon_skip_managed_blob_gc__";
+
+ public ManagedBlobOrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) {
+ super(table, olderThanMillis, dryRun);
+ }
+
+ /**
+ * Join key for a managed pack. Identity is {@code storageRootId + relativePath}, reconstructed
+ * by {@link Reference#toPath()}. The FileIO scheme is omitted so a wrapper such as test {@code
+ * traceable:} still matches listed {@code file:} paths at the same location.
+ */
+ public static String packIdentity(Path packPath) {
+ URI uri = packPath.toUri();
+ String authority = uri.getAuthority();
+ String path = uri.getPath();
+ if (authority == null || authority.isEmpty()) {
+ return path;
+ }
+ return authority + path;
+ }
+
+ public static String packIdentity(Reference reference) {
+ return packIdentity(reference.toPath());
+ }
+
+ /**
+ * Sorted {@code branch:snapshotId} pairs over every valid branch. Used to abort pack GC when
+ * the snapshot set changes between the two used-pack collections.
+ */
+ protected List snapshotTopology() throws IOException {
+ List topology = new ArrayList<>();
+ for (String branch : validBranches()) {
+ for (Snapshot snapshot : safelyGetAllSnapshots(branch)) {
+ topology.add(branch + ":" + snapshot.id());
+ }
+ }
+ Collections.sort(topology);
+ return topology;
+ }
+
+ /**
+ * Collects used pack identities from every valid branch. Subclasses may override to
+ * parallelize.
+ */
+ protected Set collectUsedPacks() throws IOException {
+ Set used = new HashSet<>();
+ for (String branch : validBranches()) {
+ for (Snapshot snapshot : safelyGetAllSnapshots(branch)) {
+ emitUsedPacks(branch, snapshot, used::add);
+ }
+ }
+ return used;
+ }
+
+ /** Test hook between the two used-pack collections. Production cleaners leave this empty. */
+ protected void betweenUsedCollections() {}
+
+ /**
+ * Aborts this run when sidecars are untrusted, the snapshot topology changed, or the two
+ * used-pack collections disagree. Callers must not delete any pack when this returns true.
+ */
+ protected boolean shouldAbortPackGc(
+ List topologyBefore, Set used, Set used2) throws IOException {
+ if (used.contains(SKIP_MANAGED_BLOB_GC) || used2.contains(SKIP_MANAGED_BLOB_GC)) {
+ LOG.warn(
+ "Skip managed blob pack GC for table {} because some sidecars or manifests cannot be trusted.",
+ table.fullName());
+ return true;
+ }
+ List topologyAfter = snapshotTopology();
+ if (!topologyBefore.equals(topologyAfter)) {
+ LOG.warn(
+ "Skip managed blob pack GC for table {} because snapshot topology changed during used-pack collection.",
+ table.fullName());
+ return true;
+ }
+ if (!used.equals(used2)) {
+ LOG.warn(
+ "Skip managed blob pack GC for table {} because the used pack set changed during used-pack collection.",
+ table.fullName());
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Emits referenced pack identities from {@code entry}. Pack reachability is collected only from
+ * {@link FileKind#ADD} files: {@link FileKind#DELETE} entries remain in delta manifests after
+ * compaction, while snapshot expire may already have removed their {@code .blobref} sidecars.
+ */
+ protected void emitUsedPacks(
+ ManifestEntry entry, DataFilePathFactory pathFactory, Consumer used) {
+ if (entry.kind() != FileKind.ADD) {
+ return;
+ }
+ List extraFiles = entry.file().extraFiles();
+ Result reachability =
+ new ManagedBlobReachabilityCollector(fileIO)
+ .fromDataFile(pathFactory.toPath(entry), extraFiles);
+ if (reachability.isUnsafe()) {
+ used.accept(SKIP_MANAGED_BLOB_GC);
+ return;
+ }
+ for (Reference reference : reachability.referenced()) {
+ used.accept(packIdentity(reference));
+ }
+ }
+
+ /**
+ * Reads {@code manifestName} and emits used packs. A missing manifest is treated as unsafe:
+ * {@link java.io.FileNotFoundException} would otherwise look like an empty used set.
+ */
+ protected void emitUsedPacks(
+ String manifestName,
+ ManifestFile manifestFile,
+ DataFilePathFactories pathFactories,
+ Consumer used)
+ throws IOException {
+ List entries =
+ retryReadingFiles(() -> manifestFile.readWithIOException(manifestName), null);
+ if (entries == null) {
+ LOG.warn(
+ "Manifest {} is missing while collecting used managed blob packs. Skip pack GC this run.",
+ manifestName);
+ used.accept(SKIP_MANAGED_BLOB_GC);
+ return;
+ }
+ for (ManifestEntry entry : entries) {
+ emitUsedPacks(entry, pathFactories.get(entry.partition(), entry.bucket()), used);
+ }
+ }
+
+ /**
+ * Reads data manifests of {@code snapshot} and emits used packs. A missing manifest list is
+ * treated as unsafe for the same reason as a missing manifest.
+ */
+ protected void emitUsedPacks(String branch, Snapshot snapshot, Consumer used)
+ throws IOException {
+ FileStoreTable branchTable = table.switchToBranch(branch);
+ ManifestList manifestList = branchTable.store().manifestListFactory().create();
+ ManifestFile manifestFile = branchTable.store().manifestFileFactory().create();
+ DataFilePathFactories pathFactories =
+ new DataFilePathFactories(branchTable.store().pathFactory());
+ List metas = new ArrayList<>();
+ if (!addManifestList(manifestList, snapshot.changelogManifestList(), metas, used)
+ || !addManifestList(manifestList, snapshot.deltaManifestList(), metas, used)
+ || !addManifestList(manifestList, snapshot.baseManifestList(), metas, used)) {
+ return;
+ }
+ for (ManifestFileMeta meta : metas) {
+ emitUsedPacks(meta.fileName(), manifestFile, pathFactories, used);
+ }
+ }
+
+ private boolean addManifestList(
+ ManifestList manifestList,
+ String listFileName,
+ List metas,
+ Consumer used)
+ throws IOException {
+ if (listFileName == null) {
+ return true;
+ }
+ List listed =
+ retryReadingFiles(() -> manifestList.readWithIOException(listFileName), null);
+ if (listed == null) {
+ LOG.warn(
+ "Manifest list {} is missing while collecting used managed blob packs. Skip pack GC this run.",
+ listFileName);
+ used.accept(SKIP_MANAGED_BLOB_GC);
+ return false;
+ }
+ metas.addAll(listed);
+ return true;
+ }
+
+ /** Deletes a managed BLOB pack. Unlike {@link OrphanFilesClean}, packs are eligible here. */
+ @Override
+ protected void cleanFile(Path path) {
+ if (!dryRun) {
+ try {
+ if (fileIO.isDir(path)) {
+ LOG.error(
+ "Refusing to delete directory {} in managed blob orphan cleanup. "
+ + "This indicates a bug in candidate collection.",
+ path);
+ } else {
+ fileIO.deleteQuietly(path);
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to check whether {} is directory, skip deleting it.", path, e);
+ }
+ }
+ }
+
+ public static boolean isManagedBlobPackName(String fileName) {
+ return fileName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java
new file mode 100644
index 000000000000..dfcb16800d11
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.DataOutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ManagedBlobReachabilityCollector}. */
+class ManagedBlobReachabilityCollectorTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testEmptyExtraFiles() {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.emptyList());
+
+ assertThat(result.isUnsafe()).isFalse();
+ assertThat(result.referenced()).isEmpty();
+ }
+
+ @Test
+ void testEmptySidecar() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ ManagedBlobReferenceFile.write(fileIO, sidecar, Collections.emptyList());
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName()));
+
+ assertThat(result.isUnsafe()).isFalse();
+ assertThat(result.referenced()).isEmpty();
+ }
+
+ @Test
+ void testReferencedPacks() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ Reference first =
+ new Reference(
+ tempDir.resolve("bucket-0").toUri().toString(), "data-a.managed.blob");
+ Reference second =
+ new Reference(
+ tempDir.resolve("bucket-0").toUri().toString(), "data-b.managed.blob");
+ ManagedBlobReferenceFile.write(fileIO, sidecar, Arrays.asList(first, second));
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName()));
+
+ assertThat(result.isUnsafe()).isFalse();
+ assertThat(result.referenced()).containsExactlyInAnyOrder(first, second);
+ assertThat(result.contains(first)).isTrue();
+ assertThat(result.containsPackName("data-b.managed.blob")).isTrue();
+ assertThat(result.containsPackName("missing.managed.blob")).isFalse();
+ assertThat(first.toPath())
+ .isEqualTo(new Path(tempDir.resolve("bucket-0/data-a.managed.blob").toUri()));
+ }
+
+ @Test
+ void testMissingSidecarUnsafeWhenDataFileExists() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ fileIO.newOutputStream(dataFile, false).close();
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList("data.avro.blobref"));
+
+ assertThat(result.isUnsafe()).isTrue();
+ assertThat(result.referenced()).isEmpty();
+ }
+
+ @Test
+ void testMissingSidecarIgnoredWhenDataFileGone() {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("expired.avro").toUri());
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList("expired.avro.blobref"));
+
+ assertThat(result.isUnsafe()).isFalse();
+ assertThat(result.referenced()).isEmpty();
+ }
+
+ @Test
+ void testCorruptSidecarUnsafe() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ fileIO.newOutputStream(dataFile, false).close();
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ try (DataOutputStream out = new DataOutputStream(fileIO.newOutputStream(sidecar, false))) {
+ out.writeInt(0x50424C52);
+ out.writeByte(1);
+ out.writeInt(0);
+ out.writeInt(12345);
+ }
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName()));
+
+ assertThat(result.isUnsafe()).isTrue();
+ }
+
+ @Test
+ void testUnsupportedVersionUnsafe() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ fileIO.newOutputStream(dataFile, false).close();
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ try (DataOutputStream out = new DataOutputStream(fileIO.newOutputStream(sidecar, false))) {
+ out.writeInt(0x50424C52);
+ out.writeByte(99);
+ out.writeInt(0);
+ }
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result result =
+ collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName()));
+
+ assertThat(result.isUnsafe()).isTrue();
+ }
+
+ @Test
+ void testMergePropagatesUnsafe() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dataFile = new Path(tempDir.resolve("data.avro").toUri());
+ Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile);
+ Reference referenced =
+ new Reference(
+ tempDir.resolve("bucket-0").toUri().toString(), "data-a.managed.blob");
+ ManagedBlobReferenceFile.write(fileIO, sidecar, Collections.singletonList(referenced));
+ ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO);
+
+ ManagedBlobReachabilityCollector.Result safe =
+ collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName()));
+ ManagedBlobReachabilityCollector.Result merged =
+ safe.merge(ManagedBlobReachabilityCollector.Result.unsafe());
+
+ assertThat(merged.isUnsafe()).isTrue();
+ assertThat(merged.referenced()).containsExactly(referenced);
+ assertThat(
+ ManagedBlobReachabilityCollector.Result.empty()
+ .merge(ManagedBlobReachabilityCollector.Result.unsafe())
+ .isUnsafe())
+ .isTrue();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java
new file mode 100644
index 000000000000..4bc979b2ff60
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java
@@ -0,0 +1,563 @@
+/*
+ * 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.blob.ManagedBlobReachabilityCollector;
+import org.apache.paimon.blob.ManagedBlobReachabilityCollector.Result;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.FileKind;
+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.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.DataFilePathFactories;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+/** Tests orphan-file cleanup of unreferenced primary-key managed BLOB packs. */
+public class ManagedBlobOrphanFilesCleanTest extends TableTestBase {
+
+ @Test
+ public void testDeleteUnreferencedPack() throws Exception {
+ FileStoreTable table = createManagedBlobTable("orphan_pack");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2})));
+
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ List deleted = clean(table);
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob");
+ assertThat(managedBlobs(table)).isNotEmpty();
+ }
+
+ @Test
+ public void testKeepReferencedPack() throws Exception {
+ FileStoreTable table = createManagedBlobTable("keep_pack");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {9, 8, 7})));
+ List before = managedBlobs(table);
+ assertThat(before).isNotEmpty();
+
+ List deleted = clean(table);
+
+ assertThat(deleted).doesNotContainAnyElementsOf(before);
+ for (Path pack : before) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ assertThat(read(table)).hasSize(1);
+ assertThat(read(table).get(0).getBlob(2).toData())
+ .containsExactly((byte) 9, (byte) 8, (byte) 7);
+ }
+
+ @Test
+ public void testPackIdentityIgnoresFileIoScheme() {
+ Path listed = new Path("file:/tmp/table/bucket-0/data-a.managed.blob");
+ Path referenced = new Path("traceable:/tmp/table/bucket-0/data-a.managed.blob");
+ Path otherBucket = new Path("file:/tmp/table/bucket-1/data-a.managed.blob");
+ assertThat(ManagedBlobOrphanFilesClean.packIdentity(referenced))
+ .isEqualTo(ManagedBlobOrphanFilesClean.packIdentity(listed));
+ assertThat(ManagedBlobOrphanFilesClean.packIdentity(otherBucket))
+ .isNotEqualTo(ManagedBlobOrphanFilesClean.packIdentity(listed));
+ }
+
+ @Test
+ public void testEmptySidecarDoesNotBlockOthers() throws Exception {
+ FileStoreTable table = createManagedBlobTable("empty_sidecar");
+ write(table, GenericRow.of(1, BinaryString.fromString("a"), null));
+
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ List deleted = clean(table);
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob");
+ }
+
+ @Test
+ public void testMissingSidecarSkipsAllPacks() throws Exception {
+ FileStoreTable table = createManagedBlobTable("missing_sidecar");
+ write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1})));
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ deleteSidecars(table);
+ List referenced = managedBlobs(table);
+ referenced.remove(orphan);
+
+ clean(table);
+
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @Test
+ public void testCorruptSidecarSkipsAllPacks() throws Exception {
+ FileStoreTable table = createManagedBlobTable("corrupt_sidecar");
+ write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1})));
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ overwriteSidecars(
+ table,
+ out -> {
+ out.writeInt(0x50424C52);
+ out.writeByte(1);
+ out.writeInt(0);
+ out.writeInt(12345);
+ });
+
+ clean(table);
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ }
+
+ @Test
+ public void testUnsupportedVersionSkipsAllPacks() throws Exception {
+ FileStoreTable table = createManagedBlobTable("unsupported_sidecar");
+ write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1})));
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ overwriteSidecars(
+ table,
+ out -> {
+ out.writeInt(0x50424C52);
+ out.writeByte(99);
+ out.writeInt(0);
+ });
+
+ clean(table);
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ }
+
+ @Test
+ public void testUnreferencedAfterUpdateAndExpire() throws Exception {
+ FileStoreTable table = createManagedBlobTable("update_expire");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {1, 1})));
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {2, 2})));
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+
+ Map expire = new HashMap<>();
+ expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1");
+ expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1");
+ expire.put(CoreOptions.SNAPSHOT_EXPIRE_LIMIT.key(), "10");
+ try (org.apache.paimon.table.sink.TableCommitImpl commit =
+ table.copy(expire).newCommit("")) {
+ commit.expireSnapshots();
+ }
+
+ Set live = livePackNames(table);
+ assertThat(live).isNotEmpty();
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ List deleted = clean(table);
+
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob");
+ for (Path pack : managedBlobs(table)) {
+ assertThat(live).contains(pack.getName());
+ }
+ assertThat(read(table)).hasSize(1);
+ assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 2, (byte) 2);
+ }
+
+ /**
+ * Compaction can commit after orphan GC has listed snapshots. Expire then deletes
+ * compact-before data files and blobrefs while those snapshots' manifests are still readable. A
+ * used-file scan of the stale list therefore neither skips nor retains the reused pack.
+ * Production GC collects used packs twice and aborts when the used set or snapshot topology
+ * changes; this test keeps the scan-only interleaving. A compaction prepared from inputs that
+ * are later removed cannot fill the remaining window after the second collection because
+ * conflict detection rejects its stale commit; see {@link
+ * #testStaleCompactionCannotCommitAfterFinalMark()}.
+ */
+ @Test
+ public void testStaleSnapshotListMissesReusedPackAfterCompactBeforeDeleted() throws Exception {
+ FileStoreTable table = createManagedBlobTable("stale_list_compact");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {3, 3})));
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {4, 4})));
+ List listed = new ArrayList<>(table.snapshotManager().safelyGetAllSnapshots());
+ assertThat(listed).isNotEmpty();
+
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+ Set liveAfterCompact = livePackNames(table);
+ assertThat(liveAfterCompact).isNotEmpty();
+
+ List compactBefore =
+ table.store()
+ .newSnapshotDeletion()
+ .planDeletedInDeltaManifest(
+ table.snapshotManager().latestSnapshot(), entry -> false);
+ assertThat(compactBefore).isNotEmpty();
+ for (Path path : compactBefore) {
+ table.fileIO().deleteQuietly(path);
+ }
+
+ StaleScan stale = collectUsedPacks(table, listed);
+ assertThat(stale.skip).isFalse();
+ assertThat(stale.packs).doesNotContainAnyElementsOf(liveAfterCompact);
+ assertThat(read(table)).hasSize(1);
+ assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 4, (byte) 4);
+ }
+
+ @Test
+ public void testAbortWhenUsedSetChangesBetweenCollections() throws Exception {
+ FileStoreTable table = createManagedBlobTable("abort_used_change");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {3, 3})));
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {4, 4})));
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ List deleted =
+ new LocalManagedBlobOrphanFilesClean(
+ table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), false) {
+ @Override
+ protected void betweenUsedCollections() {
+ try {
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+ List compactBefore =
+ table.store()
+ .newSnapshotDeletion()
+ .planDeletedInDeltaManifest(
+ table.snapshotManager().latestSnapshot(),
+ entry -> false);
+ for (Path path : compactBefore) {
+ table.fileIO().deleteQuietly(path);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }.clean().getDeletedFilesPath();
+
+ assertThat(deleted).isEmpty();
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ Set live = livePackNames(table);
+ assertThat(live).isNotEmpty();
+ for (String name : live) {
+ assertThat(table.fileIO().exists(new Path(bucketPath(table), name))).isTrue();
+ }
+ assertThat(read(table)).hasSize(1);
+ assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 4, (byte) 4);
+ }
+
+ @Test
+ public void testStaleCompactionCannotCommitAfterFinalMark() throws Exception {
+ FileStoreTable table = createManagedBlobTable("stale_compaction_commit");
+ write(
+ table,
+ GenericRow.of(
+ 1, BinaryString.fromString("old-1"), new BlobData(new byte[] {1, 1})));
+ write(
+ table,
+ GenericRow.of(
+ 2, BinaryString.fromString("old-2"), new BlobData(new byte[] {2, 2})));
+ Set oldPacks = livePackNames(table);
+ assertThat(oldPacks).isNotEmpty();
+
+ BatchWriteBuilder staleBuilder = table.newBatchWriteBuilder();
+ List staleMessages;
+ try (BatchTableWrite staleWrite = staleBuilder.newWrite()) {
+ staleWrite.withIOManager(ioManager);
+ staleWrite.compact(BinaryRow.EMPTY_ROW, 0, true);
+ staleMessages = staleWrite.prepareCommit();
+ }
+ assertThat(staleMessages).isNotEmpty();
+
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("new-1"), new BlobData(new byte[] {3, 3})),
+ GenericRow.of(
+ 2, BinaryString.fromString("new-2"), new BlobData(new byte[] {4, 4})));
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+
+ Map expire = new HashMap<>();
+ expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1");
+ expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1");
+ expire.put(CoreOptions.SNAPSHOT_EXPIRE_LIMIT.key(), "10");
+ try (org.apache.paimon.table.sink.TableCommitImpl commit =
+ table.copy(expire).newCommit("")) {
+ commit.expireSnapshots();
+ }
+
+ Set currentPacks = livePackNames(table);
+ assertThat(currentPacks).isNotEmpty();
+ assertThat(currentPacks).doesNotContainAnyElementsOf(oldPacks);
+
+ AtomicBoolean commitAttempted = new AtomicBoolean();
+ AtomicReference commitFailure = new AtomicReference<>();
+ long snapshotIdBeforeClean = table.snapshotManager().latestSnapshotId();
+ try (BatchTableCommit staleCommit = staleBuilder.newCommit()) {
+ List deleted =
+ new LocalManagedBlobOrphanFilesClean(
+ table,
+ System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2),
+ false) {
+ @Override
+ protected void cleanFile(Path path) {
+ if (commitAttempted.compareAndSet(false, true)) {
+ commitFailure.set(
+ catchThrowable(() -> staleCommit.commit(staleMessages)));
+ }
+ super.cleanFile(path);
+ }
+ }.clean().getDeletedFilesPath();
+
+ assertThat(commitAttempted).isTrue();
+ assertThat(commitFailure.get())
+ .isNotNull()
+ .hasStackTraceContaining("File deletion conflicts detected");
+ assertThat(deleted).extracting(Path::getName).containsAnyElementsOf(oldPacks);
+ }
+
+ assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotIdBeforeClean);
+ for (String pack : currentPacks) {
+ assertThat(table.fileIO().exists(new Path(bucketPath(table), pack))).isTrue();
+ }
+ assertThat(read(table))
+ .extracting(row -> row.getString(1).toString())
+ .containsExactlyInAnyOrder("new-1", "new-2");
+ }
+
+ @Test
+ public void testJoinByFullPackPath() throws Exception {
+ FileStoreTable table = createManagedBlobTable("full_path_join");
+ write(
+ table,
+ GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2})));
+ List live = managedBlobs(table);
+ assertThat(live).isNotEmpty();
+ String liveName = live.get(0).getName();
+ Path otherBucket = new Path(bucketPath(table).getParent(), "bucket-1");
+ Path other = new Path(otherBucket, liveName);
+ table.fileIO().mkdirs(otherBucket);
+ table.fileIO().newOutputStream(other, false).close();
+
+ List deleted = clean(table);
+
+ assertThat(table.fileIO().exists(other)).isFalse();
+ assertThat(deleted).extracting(Path::getName).contains(liveName);
+ for (Path pack : live) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ private FileStoreTable createManagedBlobTable(String name) throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("payload", DataTypes.BLOB())
+ .primaryKey("id")
+ .option(CoreOptions.BLOB_FIELD.key(), "payload")
+ .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .build();
+ catalog.createTable(identifier(name), schema, true);
+ return getTable(identifier(name));
+ }
+
+ private static List clean(FileStoreTable table) throws Exception {
+ return new LocalManagedBlobOrphanFilesClean(
+ table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), false)
+ .clean()
+ .getDeletedFilesPath();
+ }
+
+ private static Path bucketPath(FileStoreTable table) {
+ return table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0);
+ }
+
+ private static Set livePackNames(FileStoreTable table) throws IOException {
+ Set names = new HashSet<>();
+ FileIO fileIO = table.fileIO();
+ DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory());
+ for (ManifestEntry entry : table.store().newScan().plan().files()) {
+ DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket());
+ DataFileMeta file = entry.file();
+ for (String extra : file.extraFiles()) {
+ if (!extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ continue;
+ }
+ Path sidecar = pathFactory.toAlignedPath(extra, file);
+ for (ManagedBlobReferenceFile.Reference ref :
+ ManagedBlobReferenceFile.read(fileIO, sidecar)) {
+ names.add(ref.relativePath());
+ }
+ }
+ }
+ return names;
+ }
+
+ private static List managedBlobs(FileStoreTable table) throws IOException {
+ List packs = new ArrayList<>();
+ FileStatus[] statuses = table.fileIO().listStatus(bucketPath(table));
+ if (statuses == null) {
+ return packs;
+ }
+ for (FileStatus status : statuses) {
+ if (status.getPath().getName().endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) {
+ packs.add(status.getPath());
+ }
+ }
+ return packs;
+ }
+
+ private static StaleScan collectUsedPacks(FileStoreTable table, Iterable snapshots)
+ throws IOException {
+ StaleScan scan = new StaleScan();
+ ManifestFile manifestFile = table.store().manifestFileFactory().create();
+ ManifestList manifestList = table.store().manifestListFactory().create();
+ DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory());
+ ManagedBlobReachabilityCollector collector =
+ new ManagedBlobReachabilityCollector(table.fileIO());
+ for (Snapshot snapshot : snapshots) {
+ List metas;
+ try {
+ metas = manifestList.readDataManifests(snapshot);
+ } catch (Exception e) {
+ scan.skip = true;
+ return scan;
+ }
+ for (ManifestFileMeta meta : metas) {
+ List entries;
+ try {
+ entries = manifestFile.read(meta.fileName());
+ } catch (Exception e) {
+ scan.skip = true;
+ return scan;
+ }
+ for (ManifestEntry entry : entries) {
+ if (entry.kind() != FileKind.ADD) {
+ continue;
+ }
+ Result result =
+ collector.fromDataFile(
+ factories.get(entry.partition(), entry.bucket()).toPath(entry),
+ entry.file().extraFiles());
+ if (result.isUnsafe()) {
+ scan.skip = true;
+ return scan;
+ }
+ for (Reference reference : result.referenced()) {
+ scan.packs.add(reference.relativePath());
+ }
+ }
+ }
+ }
+ return scan;
+ }
+
+ private static final class StaleScan {
+ private boolean skip;
+ private final Set packs = new HashSet<>();
+ }
+
+ private static void deleteSidecars(FileStoreTable table) throws IOException {
+ FileIO fileIO = table.fileIO();
+ DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory());
+ for (ManifestEntry entry : table.store().newScan().plan().files()) {
+ DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket());
+ DataFileMeta file = entry.file();
+ for (String extra : file.extraFiles()) {
+ if (extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ fileIO.deleteQuietly(pathFactory.toAlignedPath(extra, file));
+ }
+ }
+ }
+ }
+
+ private interface SidecarOverwriter {
+ void write(DataOutputStream out) throws IOException;
+ }
+
+ private static void overwriteSidecars(FileStoreTable table, SidecarOverwriter overwriter)
+ throws IOException {
+ FileIO fileIO = table.fileIO();
+ DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory());
+ for (ManifestEntry entry : table.store().newScan().plan().files()) {
+ DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket());
+ DataFileMeta file = entry.file();
+ for (String extra : file.extraFiles()) {
+ if (!extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ continue;
+ }
+ Path sidecar = pathFactory.toAlignedPath(extra, file);
+ fileIO.deleteQuietly(sidecar);
+ try (DataOutputStream out =
+ new DataOutputStream(fileIO.newOutputStream(sidecar, false))) {
+ overwriter.write(out);
+ }
+ }
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
new file mode 100644
index 000000000000..f684bd64a95b
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
@@ -0,0 +1,130 @@
+/*
+ * 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.flink.orphan.FlinkManagedBlobOrphanFilesClean;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Locale;
+
+import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
+
+/**
+ * Remove orphan managed BLOB packs procedure. Usage:
+ *
+ *
+ * CALL sys.remove_orphan_blobs('tableId')
+ *
+ * CALL sys.remove_orphan_blobs('tableId', '2023-12-31 23:59:59')
+ *
+ * CALL sys.remove_orphan_blobs('databaseName.*', '2023-12-31 23:59:59')
+ *
+ */
+public class RemoveOrphanBlobsProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "remove_orphan_blobs";
+
+ public String[] call(ProcedureContext procedureContext, String tableId) throws Exception {
+ return call(procedureContext, tableId, "");
+ }
+
+ public String[] call(ProcedureContext procedureContext, String tableId, String olderThan)
+ throws Exception {
+ return call(procedureContext, tableId, olderThan, false);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String tableId, String olderThan, boolean dryRun)
+ throws Exception {
+ return call(procedureContext, tableId, olderThan, dryRun, null);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String olderThan,
+ boolean dryRun,
+ Integer parallelism)
+ throws Exception {
+ return call(procedureContext, tableId, olderThan, dryRun, parallelism, null);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String olderThan,
+ boolean dryRun,
+ Integer parallelism,
+ String mode)
+ throws Exception {
+ Identifier identifier = Identifier.fromString(tableId);
+ String databaseName = identifier.getDatabaseName();
+ String tableName = identifier.getObjectName();
+ if (mode == null) {
+ mode = "DISTRIBUTED";
+ }
+
+ CleanOrphanFilesResult result;
+ try {
+ switch (mode.toUpperCase(Locale.ROOT)) {
+ case "DISTRIBUTED":
+ result =
+ FlinkManagedBlobOrphanFilesClean.executeDatabase(
+ procedureContext.getExecutionEnvironment(),
+ catalog,
+ olderThanMillis(olderThan),
+ dryRun,
+ parallelism,
+ databaseName,
+ tableName);
+ break;
+ case "LOCAL":
+ result =
+ LocalManagedBlobOrphanFilesClean.executeDatabase(
+ catalog,
+ databaseName,
+ tableName,
+ olderThanMillis(olderThan),
+ parallelism,
+ dryRun);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown mode: "
+ + mode
+ + ". Only 'DISTRIBUTED' and 'LOCAL' are supported.");
+ }
+ return new String[] {
+ String.valueOf(result.getDeletedFileCount()),
+ String.valueOf(result.getDeletedFileTotalLenInBytes())
+ };
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 000000000000..c987d9d6c859
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+import org.apache.paimon.flink.action.RemoveOrphanBlobsAction;
+import org.apache.paimon.flink.action.RemoveOrphanBlobsActionITCaseBase;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink 1.18. */
+public class RemoveOrphanBlobsActionITCase extends RemoveOrphanBlobsActionITCaseBase {
+
+ protected boolean supportNamedArgument() {
+ return false;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java b/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 000000000000..16fd381355eb
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+import org.apache.paimon.flink.action.RemoveOrphanBlobsAction;
+import org.apache.paimon.flink.action.RemoveOrphanBlobsActionITCaseBase;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink 1.19. */
+public class RemoveOrphanBlobsActionITCase extends RemoveOrphanBlobsActionITCaseBase {}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java
new file mode 100644
index 000000000000..30fab2ca218d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java
@@ -0,0 +1,68 @@
+/*
+ * 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 javax.annotation.Nullable;
+
+import java.util.Map;
+
+import static org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean.executeDatabase;
+import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
+
+/** Action to remove unreferenced primary-key managed BLOB packs. */
+public class RemoveOrphanBlobsAction extends ActionBase {
+
+ private final String databaseName;
+ @Nullable private final String tableName;
+ @Nullable private final String parallelism;
+
+ private String olderThan = null;
+ private boolean dryRun = false;
+
+ public RemoveOrphanBlobsAction(
+ String databaseName,
+ @Nullable String tableName,
+ @Nullable String parallelism,
+ Map catalogConfig) {
+ super(catalogConfig);
+ this.databaseName = databaseName;
+ this.tableName = tableName;
+ this.parallelism = parallelism;
+ }
+
+ public void olderThan(String olderThan) {
+ this.olderThan = olderThan;
+ }
+
+ public void dryRun() {
+ this.dryRun = true;
+ }
+
+ @Override
+ public void run() throws Exception {
+ executeDatabase(
+ env,
+ catalog,
+ olderThanMillis(olderThan),
+ dryRun,
+ parallelism == null ? null : Integer.parseInt(parallelism),
+ databaseName,
+ tableName);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java
new file mode 100644
index 000000000000..dc22a61852cd
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java
@@ -0,0 +1,81 @@
+/*
+ * 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 java.util.Optional;
+
+/** Factory to create {@link RemoveOrphanBlobsAction}. */
+public class RemoveOrphanBlobsActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "remove_orphan_blobs";
+ private static final String OLDER_THAN = "older_than";
+ private static final String DRY_RUN = "dry_run";
+ private static final String PARALLELISM = "parallelism";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional create(MultipleParameterToolAdapter params) {
+ RemoveOrphanBlobsAction action =
+ new RemoveOrphanBlobsAction(
+ params.getRequired(DATABASE),
+ params.get(TABLE),
+ params.get(PARALLELISM),
+ catalogConfigMap(params));
+
+ if (params.has(OLDER_THAN)) {
+ action.olderThan(params.get(OLDER_THAN));
+ }
+
+ if (params.has(DRY_RUN) && Boolean.parseBoolean(params.get(DRY_RUN))) {
+ action.dryRun();
+ }
+
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println(
+ "Action \"remove_orphan_blobs\" removes unreferenced primary-key managed BLOB packs.");
+ System.out.println();
+ System.out.println("Syntax:");
+ System.out.println(
+ " remove_orphan_blobs \\\n"
+ + "--warehouse \\\n"
+ + "--database \\\n"
+ + "--table \\\n"
+ + "[--older_than ] \\\n"
+ + "[--dry_run ]");
+ System.out.println();
+ System.out.println(
+ "To avoid deleting newly written packs, this action only deletes packs older than 1 day by default. "
+ + "The interval can be modified by '--older_than'. format: yyyy-MM-dd HH:mm:ss");
+ System.out.println();
+ System.out.println(
+ "When '--dry_run true', view only orphan packs, don't actually remove files. Default is false.");
+ System.out.println();
+ System.out.println(
+ "If the table is null or *, all managed BLOB packs in all tables under the db will be cleaned up.");
+ System.out.println();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java
new file mode 100644
index 000000000000..3ec3bd209697
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java
@@ -0,0 +1,427 @@
+/*
+ * 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.orphan;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.flink.utils.BoundedTwoInputOperator;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.operators.InputSelection;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Flink {@link ManagedBlobOrphanFilesClean}. */
+public class FlinkManagedBlobOrphanFilesClean extends ManagedBlobOrphanFilesClean {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(FlinkManagedBlobOrphanFilesClean.class);
+
+ @Nullable private final Integer parallelism;
+
+ public FlinkManagedBlobOrphanFilesClean(
+ FileStoreTable table,
+ long olderThanMillis,
+ boolean dryRun,
+ @Nullable Integer parallelism) {
+ super(table, olderThanMillis, dryRun);
+ this.parallelism = parallelism;
+ }
+
+ @Nullable
+ public DataStream doClean(StreamExecutionEnvironment env) {
+ List topologyBefore;
+ try {
+ topologyBefore = snapshotTopology();
+ } catch (java.io.IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ flinkConf.set(ExecutionOptions.SORT_INPUTS, false);
+ flinkConf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false);
+ if (parallelism != null) {
+ flinkConf.set(CoreOptions.DEFAULT_PARALLELISM, parallelism);
+ }
+ flinkConf.setString("execution.batch.adaptive.auto-parallelism.enabled", "false");
+ env.configure(flinkConf);
+
+ List branches = validBranches();
+ final OutputTag skipGcTag = new OutputTag("managed-blob-gc-skip") {};
+ SingleOutputStreamOperator usedPacks =
+ env.fromCollection(branches)
+ .name("branch-source")
+ .process(
+ new ProcessFunction() {
+ @Override
+ public void processElement(
+ String branch,
+ ProcessFunction.Context ctx,
+ Collector out)
+ throws Exception {
+ for (Snapshot snapshot : safelyGetAllSnapshots(branch)) {
+ emitUsedPacks(
+ branch,
+ snapshot,
+ identity -> {
+ if (SKIP_MANAGED_BLOB_GC.equals(identity)) {
+ ctx.output(skipGcTag, Boolean.TRUE);
+ } else {
+ out.collect(identity);
+ }
+ });
+ }
+ }
+ })
+ .name("collect-used-packs")
+ .returns(STRING_TYPE_INFO);
+
+ SingleOutputStreamOperator usedPacks2 =
+ usedPacks
+ .transform(
+ "re-mark-used-packs",
+ STRING_TYPE_INFO,
+ new BoundedOneInputOperator() {
+
+ private final Set used = new HashSet<>();
+
+ @Override
+ public void processElement(StreamRecord element) {
+ used.add(element.getValue());
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ Set used2 = collectUsedPacks();
+ if (shouldAbortPackGc(topologyBefore, used, used2)) {
+ output.collect(
+ skipGcTag, new StreamRecord<>(Boolean.TRUE));
+ return;
+ }
+ for (String identity : used2) {
+ if (!SKIP_MANAGED_BLOB_GC.equals(identity)) {
+ output.collect(new StreamRecord<>(identity));
+ }
+ }
+ }
+ })
+ .forceNonParallel();
+
+ SingleOutputStreamOperator> candidates =
+ env.fromCollection(Collections.singletonList(1), TypeInformation.of(Integer.class))
+ .process(
+ new ProcessFunction() {
+ @Override
+ public void processElement(
+ Integer i,
+ ProcessFunction.Context ctx,
+ Collector out) {
+ FileStorePathFactory pathFactory =
+ table.store().pathFactory();
+ listPaimonFileDirs(
+ table.fullName(),
+ pathFactory.manifestPath().toString(),
+ pathFactory.indexPath().toString(),
+ pathFactory.statisticsPath().toString(),
+ pathFactory.dataFilePath().toString(),
+ partitionKeysNum,
+ table.coreOptions().dataFileExternalPaths())
+ .stream()
+ .map(Path::toUri)
+ .map(Object::toString)
+ .forEach(out::collect);
+ }
+ })
+ .name("list-dirs")
+ .forceNonParallel()
+ .process(
+ new ProcessFunction>() {
+ @Override
+ public void processElement(
+ String dir,
+ ProcessFunction>.Context
+ ctx,
+ Collector> out) {
+ for (FileStatus file : tryBestListingDirs(new Path(dir))) {
+ if (!file.isDir()
+ && oldEnough(file)
+ && isManagedBlobPackName(
+ file.getPath().getName())) {
+ out.collect(
+ Tuple2.of(
+ file.getPath().toString(),
+ file.getLen()));
+ }
+ }
+ }
+ })
+ .name("collect-candidate-packs");
+
+ final OutputTag> unusedPackTag =
+ new OutputTag>("unused-managed-blob") {};
+
+ SingleOutputStreamOperator unusedJoin =
+ usedPacks2
+ .keyBy(identity -> identity)
+ .connect(
+ candidates.keyBy(
+ pathAndSize -> packIdentity(new Path(pathAndSize.f0))))
+ .transform(
+ "join-used-and-candidate-packs",
+ TypeInformation.of(CleanOrphanFilesResult.class),
+ new BoundedTwoInputOperator<
+ String, Tuple2, CleanOrphanFilesResult>() {
+
+ private boolean buildEnd;
+ private final Set used = new HashSet<>();
+
+ @Override
+ public InputSelection nextSelection() {
+ return buildEnd
+ ? InputSelection.SECOND
+ : InputSelection.FIRST;
+ }
+
+ @Override
+ public void endInput(int inputId) {
+ switch (inputId) {
+ case 1:
+ checkState(!buildEnd, "Should not build ended.");
+ buildEnd = true;
+ break;
+ case 2:
+ checkState(buildEnd, "Should build ended.");
+ output.collect(
+ new StreamRecord<>(
+ new CleanOrphanFilesResult(0, 0)));
+ break;
+ }
+ }
+
+ @Override
+ public void processElement1(StreamRecord element) {
+ used.add(element.getValue());
+ }
+
+ @Override
+ public void processElement2(
+ StreamRecord> element) {
+ checkState(buildEnd, "Should build ended.");
+ Tuple2 fileInfo = element.getValue();
+ if (!used.contains(packIdentity(new Path(fileInfo.f0)))) {
+ output.collect(
+ unusedPackTag, new StreamRecord<>(fileInfo));
+ }
+ }
+ });
+
+ DataStream skipGc =
+ usedPacks.getSideOutput(skipGcTag).union(usedPacks2.getSideOutput(skipGcTag));
+
+ final OutputTag emptyDirTag = new OutputTag("empty-managed-blob-dir") {};
+ SingleOutputStreamOperator cleaned =
+ unusedJoin
+ .getSideOutput(unusedPackTag)
+ .connect(skipGc.broadcast())
+ .transform(
+ "clean-unused-managed-blobs",
+ TypeInformation.of(CleanOrphanFilesResult.class),
+ new BoundedTwoInputOperator<
+ Tuple2, Boolean, CleanOrphanFilesResult>() {
+
+ private boolean skipEnded;
+ private boolean skipGc;
+ private long emittedFilesCount;
+ private long emittedFilesLen;
+
+ @Override
+ public InputSelection nextSelection() {
+ return skipEnded
+ ? InputSelection.FIRST
+ : InputSelection.SECOND;
+ }
+
+ @Override
+ public void endInput(int inputId) {
+ switch (inputId) {
+ case 2:
+ checkState(!skipEnded, "Should not skip ended.");
+ skipEnded = true;
+ LOG.info("Managed blob GC skip flag: {}", skipGc);
+ break;
+ case 1:
+ checkState(skipEnded, "Should skip ended.");
+ output.collect(
+ new StreamRecord<>(
+ new CleanOrphanFilesResult(
+ emittedFilesCount,
+ emittedFilesLen)));
+ break;
+ }
+ }
+
+ @Override
+ public void processElement1(
+ StreamRecord> element) {
+ checkState(skipEnded, "Should skip ended.");
+ if (skipGc) {
+ return;
+ }
+ Tuple2 fileInfo = element.getValue();
+ Path path = new Path(fileInfo.f0);
+ emittedFilesCount++;
+ emittedFilesLen += fileInfo.f1;
+ cleanFile(path);
+ Path parent = path.getParent();
+ if (parent != null
+ && parent.toString().contains(BUCKET_PATH_PREFIX)) {
+ output.collect(emptyDirTag, new StreamRecord<>(parent));
+ }
+ LOG.info("Dry clean: {}", path);
+ }
+
+ @Override
+ public void processElement2(StreamRecord element) {
+ skipGc = true;
+ }
+ });
+
+ cleaned.getSideOutput(emptyDirTag)
+ .transform(
+ "clean-empty-dirs",
+ STRING_TYPE_INFO,
+ new BoundedOneInputOperator() {
+
+ private final Set bucketDirs = new HashSet<>();
+
+ @Override
+ public void processElement(StreamRecord element) {
+ bucketDirs.add(element.getValue());
+ }
+
+ @Override
+ public void endInput() {
+ tryCleanDataDirectory(bucketDirs, partitionKeysNum + 1);
+ }
+ })
+ .forceNonParallel()
+ .sinkTo(new DiscardingSink<>())
+ .name("end")
+ .setParallelism(1)
+ .setMaxParallelism(1);
+
+ return cleaned;
+ }
+
+ public static CleanOrphanFilesResult executeDatabase(
+ StreamExecutionEnvironment env,
+ Catalog catalog,
+ long olderThanMillis,
+ boolean dryRun,
+ @Nullable Integer parallelism,
+ String databaseName,
+ @Nullable String tableName)
+ throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
+ List tableNames = Collections.singletonList(tableName);
+ if (tableName == null || "*".equals(tableName)) {
+ tableNames = catalog.listTables(databaseName);
+ }
+
+ List> cleans = new ArrayList<>(tableNames.size());
+ for (String t : tableNames) {
+ Identifier identifier = new Identifier(databaseName, t);
+ Table table = catalog.getTable(identifier);
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports remove-orphan-blobs action. The table type is '%s'.",
+ table.getClass().getName());
+ DataStream clean =
+ new FlinkManagedBlobOrphanFilesClean(
+ (FileStoreTable) table, olderThanMillis, dryRun, parallelism)
+ .doClean(env);
+ if (clean != null) {
+ cleans.add(clean);
+ }
+ }
+
+ DataStream result = null;
+ for (DataStream clean : cleans) {
+ result = result == null ? clean : result.union(clean);
+ }
+ return sum(result);
+ }
+
+ private static CleanOrphanFilesResult sum(DataStream deleted) {
+ long deletedFilesCount = 0;
+ long deletedFilesLenInBytes = 0;
+ if (deleted != null) {
+ try {
+ CloseableIterator iterator =
+ deleted.global().executeAndCollect("ManagedBlobOrphanFilesClean");
+ while (iterator.hasNext()) {
+ CleanOrphanFilesResult cleanOrphanFilesResult = iterator.next();
+ deletedFilesCount += cleanOrphanFilesResult.getDeletedFileCount();
+ deletedFilesLenInBytes +=
+ cleanOrphanFilesResult.getDeletedFileTotalLenInBytes();
+ }
+ iterator.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return new CleanOrphanFilesResult(deletedFilesCount, deletedFilesLenInBytes);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
new file mode 100644
index 000000000000..138558d69a15
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
@@ -0,0 +1,118 @@
+/*
+ * 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.flink.orphan.FlinkManagedBlobOrphanFilesClean;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Locale;
+
+import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
+
+/**
+ * Remove orphan managed BLOB packs procedure. Usage:
+ *
+ *
+ * CALL sys.remove_orphan_blobs('tableId')
+ *
+ * CALL sys.remove_orphan_blobs('tableId', '2023-12-31 23:59:59')
+ *
+ * CALL sys.remove_orphan_blobs('databaseName.*', '2023-12-31 23:59:59')
+ *
+ */
+public class RemoveOrphanBlobsProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "remove_orphan_blobs";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(
+ name = "older_than",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true),
+ @ArgumentHint(name = "parallelism", type = @DataTypeHint("INT"), isOptional = true),
+ @ArgumentHint(name = "mode", type = @DataTypeHint("STRING"), isOptional = true)
+ })
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String olderThan,
+ Boolean dryRun,
+ Integer parallelism,
+ String mode)
+ throws Exception {
+ Identifier identifier = Identifier.fromString(tableId);
+ String databaseName = identifier.getDatabaseName();
+ String tableName = identifier.getObjectName();
+ if (mode == null) {
+ mode = "DISTRIBUTED";
+ }
+ CleanOrphanFilesResult result;
+ try {
+ switch (mode.toUpperCase(Locale.ROOT)) {
+ case "DISTRIBUTED":
+ result =
+ FlinkManagedBlobOrphanFilesClean.executeDatabase(
+ procedureContext.getExecutionEnvironment(),
+ catalog,
+ olderThanMillis(olderThan),
+ dryRun != null && dryRun,
+ parallelism,
+ databaseName,
+ tableName);
+ break;
+ case "LOCAL":
+ result =
+ LocalManagedBlobOrphanFilesClean.executeDatabase(
+ catalog,
+ databaseName,
+ tableName,
+ olderThanMillis(olderThan),
+ parallelism,
+ dryRun != null && dryRun);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown mode: "
+ + mode
+ + ". Only 'DISTRIBUTED' and 'LOCAL' are supported.");
+ }
+ return new String[] {
+ String.valueOf(result.getDeletedFileCount()),
+ String.valueOf(result.getDeletedFileTotalLenInBytes())
+ };
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @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 882adc4e788b..8e3057e36d53 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
@@ -32,6 +32,7 @@ org.apache.paimon.flink.action.ResetConsumerActionFactory
org.apache.paimon.flink.action.MigrateTableActionFactory
org.apache.paimon.flink.action.MigrateDatabaseActionFactory
org.apache.paimon.flink.action.RemoveOrphanFilesActionFactory
+org.apache.paimon.flink.action.RemoveOrphanBlobsActionFactory
org.apache.paimon.flink.action.QueryServiceActionFactory
org.apache.paimon.flink.action.ExpirePartitionsActionFactory
org.apache.paimon.flink.action.MarkPartitionDoneActionFactory
@@ -77,6 +78,7 @@ org.apache.paimon.flink.procedure.RollbackToWatermarkProcedure
org.apache.paimon.flink.procedure.MigrateTableProcedure
org.apache.paimon.flink.procedure.MigrateDatabaseProcedure
org.apache.paimon.flink.procedure.RemoveOrphanFilesProcedure
+org.apache.paimon.flink.procedure.RemoveOrphanBlobsProcedure
org.apache.paimon.flink.procedure.QueryServiceProcedure
org.apache.paimon.flink.procedure.ExpireSnapshotsProcedure
org.apache.paimon.flink.procedure.ExpireChangelogsProcedure
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
index 4e447e39722a..f5c77c54e070 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
@@ -120,7 +120,10 @@ private static class EnvExecuteMethodVisitor extends ClassVisitor {
""),
Tuple2.of(
"org/apache/paimon/flink/orphan/FlinkOrphanFilesClean",
- "executeDatabaseOrphanFiles"));
+ "executeDatabaseOrphanFiles"),
+ Tuple2.of(
+ "org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean",
+ "executeDatabase"));
private static final List> VALID_OWNER_PATTERN_AND_NAMES =
Collections.singletonList(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 000000000000..196ffea58e0e
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink Common. */
+public class RemoveOrphanBlobsActionITCase extends RemoveOrphanBlobsActionITCaseBase {}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java
new file mode 100644
index 000000000000..3a4b562f6dd9
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java
@@ -0,0 +1,169 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+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.apache.paimon.utils.DateTimeUtils;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link RemoveOrphanBlobsAction}. */
+public abstract class RemoveOrphanBlobsActionITCaseBase extends ActionITCaseBase {
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testDeleteUnreferencedManagedBlobPack(String mode) throws Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ List referenced =
+ filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ referenced.removeIf(p -> orphan.getName().equals(p.getName()));
+ assertThat(referenced).isNotEmpty();
+
+ ImmutableList.copyOf(executeSQL(removeOrphanBlobsCall(mode)));
+
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testMissingManagedBlobSidecarSkipsPackGc(String mode) throws Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphanPack = new Path(bucketPath(table), "orphan.managed.blob");
+ Path orphanOther = new Path(bucketPath(table), "orphan.txt");
+ table.fileIO().newOutputStream(orphanPack, false).close();
+ table.fileIO().writeFile(orphanOther, "x", true);
+ Thread.sleep(2000);
+
+ List referenced =
+ filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ referenced.removeIf(p -> orphanPack.getName().equals(p.getName()));
+ assertThat(referenced).isNotEmpty();
+ deleteFilesWithSuffix(table, ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX);
+
+ ImmutableList.copyOf(executeSQL(removeOrphanBlobsCall(mode)));
+
+ assertThat(table.fileIO().exists(orphanPack)).isTrue();
+ assertThat(table.fileIO().exists(orphanOther)).isTrue();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ private FileStoreTable createManagedBlobTableAndWrite() throws Exception {
+ Map options = new HashMap<>();
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "none");
+ options.put("bucket", "1");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING(), DataTypes.BLOB()},
+ new String[] {"id", "name", "payload"});
+ FileStoreTable table =
+ createFileStoreTable(
+ tableName,
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyList(),
+ options);
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+ writeData(rowData(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2})));
+ write.close();
+ commit.close();
+ write = null;
+ commit = null;
+ return table;
+ }
+
+ private String removeOrphanBlobsCall(String mode) {
+ String olderThan =
+ DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), 3);
+ if (supportNamedArgument()) {
+ return String.format(
+ "CALL sys.remove_orphan_blobs(`table` => '%s.%s', older_than => '%s', mode => '%s')",
+ database, tableName, olderThan, mode);
+ }
+ return String.format(
+ "CALL sys.remove_orphan_blobs('%s.%s', '%s', false, 5, '%s')",
+ database, tableName, olderThan, mode);
+ }
+
+ private static Path bucketPath(FileStoreTable table) {
+ return table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0);
+ }
+
+ private static List filesWithSuffix(FileStoreTable table, String suffix)
+ throws IOException {
+ List result = new ArrayList<>();
+ FileStatus[] statuses = table.fileIO().listStatus(bucketPath(table));
+ if (statuses == null) {
+ return result;
+ }
+ for (FileStatus status : statuses) {
+ if (status.getPath().getName().endsWith(suffix)) {
+ result.add(status.getPath());
+ }
+ }
+ return result;
+ }
+
+ private static void deleteFilesWithSuffix(FileStoreTable table, String suffix)
+ throws IOException {
+ for (Path path : filesWithSuffix(table, suffix)) {
+ table.fileIO().deleteQuietly(path);
+ }
+ }
+
+ protected boolean supportNamedArgument() {
+ return true;
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index 60b5747e3db7..6687dbdaeb8a 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -48,6 +48,7 @@
import org.apache.paimon.spark.procedure.ProcedureBuilder;
import org.apache.paimon.spark.procedure.PurgeFilesProcedure;
import org.apache.paimon.spark.procedure.ReassignRowIdProcedure;
+import org.apache.paimon.spark.procedure.RemoveOrphanBlobsProcedure;
import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
import org.apache.paimon.spark.procedure.RemoveUnexistingFilesProcedure;
import org.apache.paimon.spark.procedure.RenameBranchProcedure;
@@ -114,6 +115,7 @@ private static Map> initProcedureBuilders() {
procedureBuilders.put("migrate_database", MigrateDatabaseProcedure::builder);
procedureBuilders.put("migrate_table", MigrateTableProcedure::builder);
procedureBuilders.put("remove_orphan_files", RemoveOrphanFilesProcedure::builder);
+ procedureBuilders.put("remove_orphan_blobs", RemoveOrphanBlobsProcedure::builder);
procedureBuilders.put("remove_unexisting_files", RemoveUnexistingFilesProcedure::builder);
procedureBuilders.put("expire_snapshots", ExpireSnapshotsProcedure::builder);
procedureBuilders.put("expire_partitions", ExpirePartitionsProcedure::builder);
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java
new file mode 100644
index 000000000000..57da6f8f648a
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java
@@ -0,0 +1,164 @@
+/*
+ * 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.spark.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean;
+import org.apache.paimon.operation.OrphanFilesClean;
+import org.apache.paimon.spark.catalog.WithPaimonCatalog;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Locale;
+
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+import static org.apache.spark.sql.types.DataTypes.LongType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Remove orphan managed BLOB packs procedure. Usage:
+ *
+ *
+ * CALL sys.remove_orphan_blobs(table => 'tableId', [older_than => '2023-10-31 12:00:00'])
+ *
+ * CALL sys.remove_orphan_blobs(table => 'databaseName.*', [older_than => '2023-10-31 12:00:00'])
+ *
+ */
+public class RemoveOrphanBlobsProcedure extends BaseProcedure {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RemoveOrphanBlobsProcedure.class.getName());
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.optional("older_than", StringType),
+ ProcedureParameter.optional("dry_run", BooleanType),
+ ProcedureParameter.optional("parallelism", IntegerType),
+ ProcedureParameter.optional("mode", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("deletedFileCount", LongType, true, Metadata.empty()),
+ new StructField(
+ "deletedFileTotalLenInBytes", LongType, true, Metadata.empty())
+ });
+
+ private RemoveOrphanBlobsProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ org.apache.paimon.catalog.Identifier identifier;
+ String tableId = args.getString(0);
+ String olderThan = args.isNullAt(1) ? null : args.getString(1);
+ boolean dryRun = !args.isNullAt(2) && args.getBoolean(2);
+ Integer parallelism = args.isNullAt(3) ? null : args.getInt(3);
+
+ Preconditions.checkArgument(
+ tableId != null && !tableId.isEmpty(),
+ "Cannot handle an empty tableId for argument %s",
+ PARAMETERS[0].name());
+
+ if (tableId.endsWith(".*")) {
+ identifier = org.apache.paimon.catalog.Identifier.fromString(tableId);
+ } else {
+ identifier =
+ org.apache.paimon.catalog.Identifier.fromString(
+ toIdentifier(args.getString(0), PARAMETERS[0].name()).toString());
+ }
+ LOG.info("identifier is {}.", identifier);
+
+ Catalog catalog = ((WithPaimonCatalog) tableCatalog()).paimonCatalog();
+ String mode = args.isNullAt(4) ? "DISTRIBUTED" : args.getString(4);
+
+ CleanOrphanFilesResult result;
+ try {
+ switch (mode.toUpperCase(Locale.ROOT)) {
+ case "LOCAL":
+ result =
+ LocalManagedBlobOrphanFilesClean.executeDatabase(
+ catalog,
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ OrphanFilesClean.olderThanMillis(olderThan),
+ parallelism,
+ dryRun);
+ break;
+ case "DISTRIBUTED":
+ result =
+ SparkManagedBlobOrphanFilesClean.executeDatabase(
+ catalog,
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ OrphanFilesClean.olderThanMillis(olderThan),
+ parallelism,
+ dryRun);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown mode: "
+ + mode
+ + ". Only 'DISTRIBUTED' and 'LOCAL' are supported.");
+ }
+
+ return new InternalRow[] {
+ newInternalRow(result.getDeletedFileCount(), result.getDeletedFileTotalLenInBytes())
+ };
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder() {
+ @Override
+ public RemoveOrphanBlobsProcedure doBuild() {
+ return new RemoveOrphanBlobsProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RemoveOrphanBlobsProcedure";
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala
new file mode 100644
index 000000000000..d3d2bd6d6b03
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.procedure
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.catalog.{Catalog, Identifier}
+import org.apache.paimon.fs.Path
+import org.apache.paimon.operation.{CleanOrphanFilesResult, ManagedBlobOrphanFilesClean}
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{functions, Dataset, PaimonSparkSession, SparkSession}
+import org.apache.spark.sql.catalyst.SQLConfHelper
+
+import java.util
+import java.util.function.Consumer
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+case class SparkManagedBlobOrphanFilesClean(
+ specifiedTable: FileStoreTable,
+ specifiedOlderThanMillis: Long,
+ parallelism: Int,
+ dryRunPara: Boolean,
+ @transient spark: SparkSession)
+ extends ManagedBlobOrphanFilesClean(specifiedTable, specifiedOlderThanMillis, dryRunPara)
+ with SQLConfHelper
+ with Logging {
+
+ def doClean(): (Dataset[(Long, Long)], Seq[Dataset[_]]) = {
+ import spark.implicits._
+
+ val topologyBefore = snapshotTopology()
+ val usedPacks = collectUsedPacksDf().cache()
+ val skipGc = usedPacks
+ .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+ .limit(1)
+ .collect()
+ .nonEmpty
+
+ val fileDirs = listPaimonFileDirs.asScala.map(_.toString).toSeq
+ val maxFileDirsParallelism = Math.min(Math.max(fileDirs.size, 1), parallelism)
+ val candidates = spark.sparkContext
+ .parallelize(fileDirs, maxFileDirsParallelism)
+ .flatMap {
+ dir =>
+ tryBestListingDirs(new Path(dir)).asScala
+ .filter(file => !file.isDir)
+ .filter(oldEnough)
+ .filter(file => ManagedBlobOrphanFilesClean.isManagedBlobPackName(file.getPath.getName))
+ .map {
+ file =>
+ val path = file.getPath
+ (
+ ManagedBlobOrphanFilesClean.packIdentity(path),
+ path.toString,
+ file.getLen,
+ path.getParent.toString)
+ }
+ }
+ .toDF("name", "path", "len", "dataDir")
+ .repartition(parallelism)
+
+ betweenUsedCollections()
+ val usedPacks2 = collectUsedPacksDf().cache()
+ val skipGc2 = usedPacks2
+ .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+ .limit(1)
+ .collect()
+ .nonEmpty
+ val topologyAfter = snapshotTopology()
+ val used1Packs: Dataset[_] =
+ usedPacks.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+ val used2Packs: Dataset[_] =
+ usedPacks2.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+ val usedChanged = used1Packs
+ .toDF()
+ .except(used2Packs.toDF())
+ .union(used2Packs.toDF().except(used1Packs.toDF()))
+ .limit(1)
+ .count() > 0
+ val abort = skipGc || skipGc2 || topologyBefore != topologyAfter || usedChanged
+ if (abort) {
+ logWarning(
+ s"Skip managed blob pack GC for table ${table.fullName()} because sidecars cannot be trusted or used packs changed during collection.")
+ }
+
+ val unused = candidates.join(used2Packs.toDF(), $"name" === $"used_name", "left_anti")
+ val toDelete = if (abort) unused.limit(0) else unused
+
+ val deleted: Dataset[(Long, Long)] = toDelete
+ .repartition($"dataDir")
+ .mapPartitions {
+ it =>
+ var deletedFilesCount = 0L
+ var deletedFilesLenInBytes = 0L
+ val dataDirs = new mutable.HashSet[String]()
+ while (it.hasNext) {
+ val fileInfo = it.next()
+ val pathToClean = fileInfo.getString(1)
+ val deletedPath = new Path(pathToClean)
+ deletedFilesLenInBytes += fileInfo.getLong(2)
+ cleanFile(deletedPath)
+ logInfo(s"Cleaned file: $pathToClean")
+ dataDirs.add(fileInfo.getString(3))
+ deletedFilesCount += 1
+ }
+ if (!dryRun) {
+ val bucketDirs = dataDirs
+ .filter(_.contains(BUCKET_PATH_PREFIX))
+ .map(new Path(_))
+ tryCleanDataDirectory(bucketDirs.asJava, partitionKeysNum + 1)
+ }
+ Iterator.single((deletedFilesCount, deletedFilesLenInBytes))
+ }
+
+ (deleted, Seq(usedPacks, usedPacks2))
+ }
+
+ private def collectUsedPacksDf(): Dataset[_] = {
+ import spark.implicits._
+ val branches = validBranches()
+ val maxBranchParallelism = Math.min(branches.size(), parallelism)
+ spark.sparkContext
+ .parallelize(branches.asScala.toSeq, maxBranchParallelism)
+ .flatMap {
+ branch => safelyGetAllSnapshots(branch).asScala.map(snapshot => (branch, snapshot.toJson))
+ }
+ .repartition(parallelism)
+ .flatMap {
+ case (branch, snapshotJson) =>
+ val names = new util.ArrayList[String]()
+ emitUsedPacks(
+ branch,
+ Snapshot.fromJson(snapshotJson),
+ new Consumer[String] {
+ override def accept(name: String): Unit = names.add(name)
+ })
+ names.asScala
+ }
+ .toDF("used_name")
+ }
+}
+
+object SparkManagedBlobOrphanFilesClean extends SQLConfHelper {
+ def executeDatabase(
+ catalog: Catalog,
+ databaseName: String,
+ tableName: String,
+ olderThanMillis: Long,
+ parallelismOpt: Integer,
+ dryRun: Boolean): CleanOrphanFilesResult = {
+ val spark = PaimonSparkSession.active
+ val parallelism = if (parallelismOpt == null) {
+ Math.max(spark.sparkContext.defaultParallelism, conf.numShufflePartitions)
+ } else {
+ parallelismOpt.intValue()
+ }
+
+ val tableNames = if (tableName == null || "*" == tableName) {
+ catalog.listTables(databaseName).asScala
+ } else {
+ tableName :: Nil
+ }
+ val tables = tableNames.map {
+ tableName =>
+ val identifier = new Identifier(databaseName, tableName)
+ val table = catalog.getTable(identifier)
+ assert(
+ table.isInstanceOf[FileStoreTable],
+ s"Only FileStoreTable supports remove-orphan-blobs action. The table type is '${table.getClass.getName}'.")
+ table.asInstanceOf[FileStoreTable]
+ }
+ if (tables.isEmpty) {
+ return new CleanOrphanFilesResult(0, 0)
+ }
+ val (deleted, waitToRelease) = tables.map {
+ table =>
+ new SparkManagedBlobOrphanFilesClean(
+ table,
+ olderThanMillis,
+ parallelism,
+ dryRun,
+ spark
+ ).doClean()
+ }.unzip
+ try {
+ val result = deleted
+ .reduce((l, r) => l.union(r))
+ .toDF("deletedFilesCount", "deletedFilesLenInBytes")
+ .agg(functions.sum("deletedFilesCount"), functions.sum("deletedFilesLenInBytes"))
+ .head()
+ assert(result.schema.size == 2, result.schema)
+ if (result.isNullAt(0)) {
+ new CleanOrphanFilesResult(0, 0)
+ } else {
+ new CleanOrphanFilesResult(result.getLong(0), result.getLong(1))
+ }
+ } finally {
+ waitToRelease.flatten.foreach(_.unpersist())
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala
new file mode 100644
index 000000000000..b13f3dbfe89e
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala
@@ -0,0 +1,108 @@
+/*
+ * 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.spark.procedure
+
+import org.apache.paimon.blob.ManagedBlobReferenceFile
+import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.fs.Path
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.utils.DateTimeUtils
+
+class RemoveOrphanBlobsProcedureTest extends PaimonSparkTestBase {
+
+ Seq("local", "distributed").foreach {
+ mode =>
+ test(s"Paimon procedure: remove unreferenced managed blob pack ($mode)") {
+ createManagedBlobTable()
+ spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+
+ val table = loadTable("T")
+ val orphan = new Path(bucketPath(table), "orphan.managed.blob")
+ table.fileIO().newOutputStream(orphan, false).close()
+ Thread.sleep(2000)
+
+ val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)
+ .filterNot(_.getName == orphan.getName)
+ assert(referenced.nonEmpty)
+
+ val olderThan = DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(System.currentTimeMillis()),
+ 3)
+ spark.sql(
+ s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', mode => '$mode')")
+
+ assert(!table.fileIO().exists(orphan))
+ referenced.foreach(pack => assert(table.fileIO().exists(pack)))
+ }
+
+ test(s"Paimon procedure: skip managed blob pack gc when sidecar missing ($mode)") {
+ createManagedBlobTable()
+ spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+
+ val table = loadTable("T")
+ val orphanPack = new Path(bucketPath(table), "orphan.managed.blob")
+ val orphanOther = new Path(bucketPath(table), "orphan.txt")
+ table.fileIO().newOutputStream(orphanPack, false).close()
+ table.fileIO().writeFile(orphanOther, "x", true)
+ Thread.sleep(2000)
+
+ val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)
+ .filterNot(_.getName == orphanPack.getName)
+ assert(referenced.nonEmpty)
+ filesWithSuffix(table, ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)
+ .foreach(table.fileIO().deleteQuietly)
+
+ val olderThan = DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(System.currentTimeMillis()),
+ 3)
+ spark.sql(
+ s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', mode => '$mode')")
+
+ assert(table.fileIO().exists(orphanPack))
+ assert(table.fileIO().exists(orphanOther))
+ referenced.foreach(pack => assert(table.fileIO().exists(pack)))
+ }
+ }
+
+ private def createManagedBlobTable(): Unit = {
+ spark.sql("""
+ |CREATE TABLE T (id INT, name STRING, payload BINARY)
+ |USING PAIMON
+ |TBLPROPERTIES (
+ | 'primary-key'='id',
+ | 'bucket'='1',
+ | 'changelog-producer'='none',
+ | 'blob-field'='payload')
+ |""".stripMargin)
+ }
+
+ private def bucketPath(table: FileStoreTable): Path = {
+ table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0)
+ }
+
+ private def filesWithSuffix(table: FileStoreTable, suffix: String): Seq[Path] = {
+ val statuses = table.fileIO().listStatus(bucketPath(table))
+ if (statuses == null) {
+ Seq.empty
+ } else {
+ statuses.map(_.getPath).filter(_.getName.endsWith(suffix))
+ }
+ }
+}
|