[core] Introduce local cleanup for unreferenced managed BLOB packs.

docs/docs/primary-key-table/blob-storage.md

@@ -183,9 +183,8 @@ participates in aggregation or retraction, even when its sequence value is older
 the field for both newer and older retract records.
 
 Managed BLOB partial updates externalize each non-null scalar BLOB, array element, or map value into a
-`.managed.blob` pack. Empty collections and collections containing only null values write no payload. BLOB garbage
-collection for orphaned packs is not implemented yet; repeated updates can leave unreachable storage until a future
-collector is available.
+`.managed.blob` pack. Empty collections and collections containing only null values write no payload. Unreachable packs
+from repeated updates are reclaimed by `remove_orphan_blobs` after they are older than `older_than`.
 
 `blob-view-field` columns store serialized view structs inline. Reads resolve upstream blob bytes through the catalog
 when `blob-view.resolve.enabled` is true (default). Append upstream tables used by `sys.blob_view(...)` must enable
@@ -240,16 +239,33 @@ extra files because more than one retained data file can reference the same pack
 
 ## Garbage Collection
 
-Garbage collection of unreferenced `.managed.blob` packs is not implemented yet. Updates, deletes, compaction, or an
-ambiguous writer failure can therefore leave payload packs that are no longer reachable from current rows.
-
-The ordinary orphan-file cleaner intentionally preserves all `.managed.blob` files. This fail-safe behavior prevents it
-from deleting a payload that is still reachable from a snapshot, tag, branch, or another retained root, but it also
-means unused BLOB storage can grow until a root-aware BLOB garbage collector is available.
-
-A future collector must compute reachability across all retained roots and treat a missing, corrupt, or unsupported
-`.blobref` sidecar as unsafe to delete. An empty, valid sidecar is different from a missing sidecar: it explicitly states
-that the data file references no managed payload pack.
+Unreferenced `.managed.blob` packs are reclaimed by `LocalManagedBlobOrphanFilesClean`.
+The cleaner reads every retained data file's `.blobref` sidecar across snapshots, tags, and
+branches, then deletes packs that are not referenced and whose modification time is earlier than the absolute
+`older_than` cutoff (1 day before the run starts by default).
+`remove_orphan_files` never deletes `.managed.blob` packs.
+
+This cleanup is best-effort. It lists snapshots, collects used packs twice, and aborts the run (deletes
+nothing) if the snapshot topology or used-pack set changed between those collections. That shrinks the
+window in which a concurrent commit can change reachability. Standard Paimon compaction does not make a pack
+that was unreachable at the final collection reachable afterward: it only reuses packs referenced by its
+input data files, and deletion-conflict detection rejects a stale compact whose inputs have already been
+removed. Under these standard compaction invariants, no separate commit lease is required for that
+compaction path.
+
+`older_than` provides a grace period for packs created by a writer but not yet referenced by a committed
+snapshot. Standard writers create new UUID-named packs; choose a cutoff far enough behind the current time
+for writes, commits, and retries to finish. The one-day default assumes those operations complete within one
+day. Writers that publish references to pre-existing old packs, or commit implementations that bypass normal
+deletion-conflict detection, are outside this safety model.
+
+A missing, corrupt, or unsupported `.blobref` sidecar on a data file that still exists is unsafe: that run skips
+deleting every `.managed.blob` file. ADD entries left in unmerged manifests after snapshot expire, whose data files
+are already gone, are ignored. An empty, valid sidecar is different from a missing sidecar: it explicitly states that
+the data file references no managed payload pack.
+
+Snapshot expiration still deletes only the data file and its `.blobref` extra file. Pack bytes are reclaimed on the
+next managed blob orphan cleanup run after they become unreachable.
 
 ## Reference Metadata
 

paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java

@@ -0,0 +1,273 @@
+/*
+ * 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.
+ *
+ * <p>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.
+     *
+     * <p>Orphan cleanup resolves sidecars itself and calls {@link #fromSidecar(Path, Path)}, so
+     * this whole-entry variant currently has no production caller. It is retained as the
+     * entry-level reachability oracle for tests and for snapshot expiration, which needs to walk
+     * {@code extraFiles} rather than pre-resolved sidecar paths.
+     */
+    public Result fromDataFile(Path dataFile, List<String> 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 {
+            return Result.of(readWithRetry(sidecar));
+        } catch (IOException e) {
+            LOG.warn(
+                    "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.",
+                    sidecar,
+                    e);
+            return Result.unsafe();
+        }
+    }
+
+    /**
+     * Reads one resolved sidecar while preserving data-file-aware orphan cleanup semantics. An
+     * unreadable sidecar is ignored only when its data file is already gone.
+     */
+    public Result fromSidecar(Path dataFile, Path sidecar) {
+        try {
+            return Result.of(readWithRetry(sidecar));
+        } catch (IOException e) {
+            if (!checkDataFileExists(dataFile)) {
+                LOG.debug(
+                        "Ignore unreadable blobref {} because data file {} is already gone.",
+                        sidecar,
+                        dataFile);
+                return Result.empty();
+            }
+            LOG.warn(
+                    "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.",
+                    sidecar,
+                    e);
+            return Result.unsafe();
+        }
+    }
+
+    private List<Reference> 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.<Reference>emptySet(), false);
+        private static final Result UNSAFE = new Result(Collections.<Reference>emptySet(), true);
+
+        private final Set<Reference> referenced;
+        private final boolean unsafe;
+
+        private Result(Set<Reference> 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<Reference> refs) {
+            if (refs == null || refs.isEmpty()) {
+                return empty();
+            }
+            return new Result(Collections.unmodifiableSet(new HashSet<>(refs)), false);
+        }
+
+        public Set<Reference> 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<Reference> 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 + '}';
+        }
+    }
+}

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) {