[core][flink][spark] Introduce remove_orphan_blobs to clean orphan blobs.

docs/docs/flink/procedures.md

@@ -424,6 +424,7 @@ All available procedures are listed below.
             <li>dryRun: when true, view only orphan files, don't actually remove files. Default is false.</li>
             <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
             <li>mode: The mode of remove orphan clean procedure (local or distributed) . By default is distributed.</li>
+            <li>Note: this procedure does not delete primary-key <code>.managed.blob</code> packs. Use <code>remove_orphan_blobs</code>.</li>
       </td>
       <td>CALL sys.remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
           CALL sys.remove_orphan_files(`table` => 'default.*', older_than => '2023-10-31 12:00:00')<br/><br/>
@@ -432,6 +433,35 @@ All available procedures are listed below.
           CALL sys.remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5, mode => 'local')
       </td>
    </tr>
+   <tr>
+      <td>remove_orphan_blobs</td>
+      <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.remove_orphan_blobs(`table` => 'identifier', older_than => 'olderThan', dry_run => 'dryRun', mode => 'mode') <br/><br/>
+         -- Use indexed argument<br/>
+         CALL [catalog.]sys.remove_orphan_blobs('identifier')<br/>
+         CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan')<br/>
+         CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun')<br/>
+         CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun','parallelism')<br/>
+         CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun','parallelism','mode')
+      </td>
+      <td>
+         To remove unreferenced primary-key <code>.managed.blob</code> packs. Arguments:
+            <li>table: the target table identifier. Cannot be empty, you can use database_name.* to clean whole database.</li>
+            <li>olderThan: to avoid deleting newly written packs, this procedure only
+               deletes packs older than 1 day by default. This argument can modify the interval.
+            </li>
+            <li>dryRun: when true, view only orphan packs, don't actually remove files. Default is false.</li>
+            <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
+            <li>mode: The mode of remove orphan blob procedure (local or distributed). By default is distributed.</li>
+      </td>
+      <td>CALL sys.remove_orphan_blobs(`table` => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL sys.remove_orphan_blobs(`table` => 'default.*', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL sys.remove_orphan_blobs(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true)<br/><br/>
+          CALL sys.remove_orphan_blobs(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5)<br/><br/>
+          CALL sys.remove_orphan_blobs(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5, mode => 'local')
+      </td>
+   </tr>
    <tr>
       <td>remove_unexisting_files</td>
       <td>

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,26 @@ 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 removed by [`remove_orphan_blobs`](../flink/procedures)
+(local, Flink, or Spark). The procedure reads every retained data file's `.blobref` sidecar across snapshots, tags, and
+branches, then deletes packs that are not referenced and older than `older_than` (1 day 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 compaction can reuse a pack after the first scan. There is still no commit lease.
+Compaction reuses pack bytes and does not refresh pack modification time, so `older_than` does not fence
+a compact that commits after the second collection and before delete. Keep a non-zero `older_than`; the
+one-day default makes this window unlikely in ordinary jobs, but it is not a logical guarantee for very
+old packs, long-running compaction, or `older_than` set to now.
+
+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 `remove_orphan_blobs` run after they become unreachable.
 
 ## Reference Metadata
 

docs/docs/spark/procedures.md

@@ -300,6 +300,7 @@ This section introduce all available spark procedures about paimon.
             <li>dry_run: when true, view only orphan files, don't actually remove files. Default is false.</li>
             <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
             <li>mode: The mode of remove orphan clean procedure (local or distributed) . By default is distributed.</li>
+            <li>Note: this procedure does not delete primary-key <code>.managed.blob</code> packs. Use <code>remove_orphan_blobs</code>.</li>
       </td>
       <td>
           CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
@@ -309,6 +310,24 @@ This section introduce all available spark procedures about paimon.
           CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true, parallelism => 5, mode => 'local')
       </td>
     </tr>
+    <tr>
+      <td>remove_orphan_blobs</td>
+      <td>
+         To remove unreferenced primary-key <code>.managed.blob</code> packs. Arguments:
+            <li>table: the target table identifier. Cannot be empty, you can use database_name.* to clean whole database.</li>
+            <li>older_than: to avoid deleting newly written packs, this procedure only deletes packs older than 1 day by default. This argument can modify the interval.</li>
+            <li>dry_run: when true, view only orphan packs, don't actually remove files. Default is false.</li>
+            <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
+            <li>mode: The mode of remove orphan blob procedure (local or distributed). By default is distributed.</li>
+      </td>
+      <td>
+          CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL sys.remove_orphan_blobs(table => 'default.*', older_than => '2023-10-31 12:00:00')<br/><br/>
+          CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true)<br/><br/>
+          CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5)<br/><br/>
+          CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5, mode => 'local')
+      </td>
+    </tr>
     <tr>
       <td>remove_unexisting_files</td>
       <td>

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.
+ *
+ * <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.
+     */
+    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 {
+            List<Reference> 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<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) {