diff --git a/docs/docs/primary-key-table/blob-storage.md b/docs/docs/primary-key-table/blob-storage.md
index 2b20099434d3..e7fd8f21f3e1 100644
--- a/docs/docs/primary-key-table/blob-storage.md
+++ b/docs/docs/primary-key-table/blob-storage.md
@@ -258,7 +258,9 @@ extra files because more than one retained data file can reference the same pack
 
 ## Garbage Collection
 
-Unreferenced `.managed.blob` packs are reclaimed by `LocalManagedBlobOrphanFilesClean`.
+Unreferenced `.managed.blob` packs are reclaimed by managed blob orphan cleanup.
+Local cleanup is `LocalManagedBlobOrphanFilesClean`; Spark exposes the same cleanup as
+[`remove_orphan_blobs`](../spark/procedures/maintenance#remove_orphan_blobs).
 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).
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 02fa0485df1d..045f4bce9229 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -67,6 +67,7 @@ Choose a group, then use the page contents to jump to a procedure:
 [`expire_snapshots`](./procedures/maintenance#expire_snapshots),
 [`expire_partitions`](./procedures/maintenance#expire_partitions),
 [`remove_orphan_files`](./procedures/maintenance#remove_orphan_files),
+[`remove_orphan_blobs`](./procedures/maintenance#remove_orphan_blobs),
 [`remove_unexisting_files`](./procedures/maintenance#remove_unexisting_files),
 [`purge_files`](./procedures/maintenance#purge_files),
 [`repair`](./procedures/maintenance#repair),
diff --git a/docs/docs/spark/procedures/maintenance.md b/docs/docs/spark/procedures/maintenance.md
index 1244848028fc..9b230e737428 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -233,6 +233,8 @@ Remove the orphan data files and metadata files.
 - `parallelism` (`INT`, optional): The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.
 - `mode` (`STRING`, optional): The mode of remove orphan clean procedure (local or distributed) . By default is distributed.
 
+This procedure does not delete primary-key `.managed.blob` packs. Use [`remove_orphan_blobs`](#remove_orphan_blobs).
+
 ```sql
 CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00');
 
@@ -256,6 +258,41 @@ CALL sys.remove_orphan_files(
 );
 ```
 
+## remove_orphan_blobs
+
+Remove unreferenced primary-key `.managed.blob` packs.
+
+**Arguments**
+
+- `table` (`STRING`, required): the target table identifier. Use `database_name.*` to process the whole database.
+- `older_than` (`STRING`, optional): an absolute timestamp cutoff. Only packs whose modification time is earlier than this timestamp are candidates. The default cutoff is 1 day before the procedure starts.
+- `dry_run` (`BOOLEAN`, optional): when true, calculate the candidate file count and total bytes without deleting files. The procedure returns aggregate counts, not individual pack paths. Default is false.
+- `parallelism` (`INT`, optional): per-table concurrency. In `distributed` mode this is the Spark task parallelism of each table job (default: the larger of Spark's default parallelism and `spark.sql.shuffle.partitions`). In `local` mode this is the per-table file-operation thread limit (default: the number of processors available to the Java virtual machine). For `database_name.*`, `distributed` mode runs tables one Spark job at a time, so cluster concurrency stays within this per-table value; `local` mode may run several tables at once, so total threads can exceed this value.
+- `mode` (`STRING`, optional): The mode of remove orphan blob procedure (`local` or `distributed`). By default is `distributed`.
+
+```sql
+CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00');
+
+CALL sys.remove_orphan_blobs(table => 'default.*', older_than => '2023-10-31 12:00:00');
+
+CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true);
+
+CALL sys.remove_orphan_blobs(
+  table => 'default.T',
+  older_than => '2023-10-31 12:00:00',
+  dry_run => false,
+  parallelism => 5
+);
+
+CALL sys.remove_orphan_blobs(
+  table => 'default.T',
+  older_than => '2023-10-31 12:00:00',
+  dry_run => false,
+  parallelism => 5,
+  mode => 'local'
+);
+```
+
 ## remove_unexisting_files
 
 Procedure to remove unexisting data files from manifest entries. See [Java docs](https://paimon.apac
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 866a54e802a9..672099d29349 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
@@ -53,6 +53,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;
@@ -120,6 +121,7 @@ private static Map<String, Supplier<ProcedureBuilder>> 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..7d52f24ef0ef
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java
@@ -0,0 +1,171 @@
+/*
+ * 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:
+ *
+ * <pre><code>
+ *  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'])
+ * </code></pre>
+ */
+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);
+
+        if (parallelism != null) {
+            Preconditions.checkArgument(
+                    parallelism > 0,
+                    "Parallelism must be greater than 0, but was %s.",
+                    parallelism);
+        }
+
+        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<RemoveOrphanBlobsProcedure>() {
+            @Override
+            public RemoveOrphanBlobsProcedure doBuild() {
+                return new RemoveOrphanBlobsProcedure(tableCatalog());
+            }
+        };
+    }
+
+    @Override
+    public String description() {
+        return "RemoveOrphanBlobsProcedure";
+    }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java
new file mode 100644
index 000000000000..67cf3386cdba
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java
@@ -0,0 +1,43 @@
+/*
+ * 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.fs.Path;
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.function.Consumer;
+
+/** Java bridge for FileIO-aware managed blob candidate identities. */
+abstract class SparkManagedBlobOrphanFilesCleanBase extends ManagedBlobOrphanFilesClean {
+
+    SparkManagedBlobOrphanFilesCleanBase(
+            FileStoreTable table, long olderThanMillis, boolean dryRun) {
+        super(table, olderThanMillis, dryRun);
+    }
+
+    final String packIdentityForCandidate(Path path) {
+        return packIdentityForCleanup(path).orElse(SKIP_MANAGED_BLOB_GC);
+    }
+
+    final void emitUsedPacksForSpark(
+            SidecarWorkItem workItem, ReachabilityScan scan, Consumer<String> used) {
+        emitUsedPacks(workItem, scan, used);
+    }
+}
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..3bcdd8a3f851
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala
@@ -0,0 +1,382 @@
+/*
+ * 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, Identifier}
+import org.apache.paimon.fs.Path
+import org.apache.paimon.manifest.{ManifestFile, ManifestFileMeta, ManifestList}
+import org.apache.paimon.operation.{CleanOrphanFilesResult, ManagedBlobOrphanFilesClean}
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean.SidecarWorkItem
+import org.apache.paimon.operation.OrphanFilesClean.retryReadingFiles
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.utils.DataFilePathFactories
+import org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX
+import org.apache.paimon.utils.Preconditions
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{functions, DataFrame, Dataset, PaimonSparkSession, SparkSession}
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.storage.StorageLevel
+
+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 SparkManagedBlobOrphanFilesCleanBase(specifiedTable, specifiedOlderThanMillis, dryRunPara)
+  with SQLConfHelper
+  with Logging {
+
+  def doClean(): (Dataset[(Long, Long)], Seq[Dataset[_]]) = {
+    import spark.implicits._
+
+    SparkManagedBlobOrphanFilesClean.checkParallelism(parallelism)
+    val cached = new mutable.ArrayBuffer[Dataset[_]]()
+    try {
+      val topologyBefore = snapshotTopology()
+      val usedPacks = collectUsedPacksDf().persist(StorageLevel.MEMORY_AND_DISK)
+      cached += usedPacks
+      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
+                  val parent = path.getParent
+                  (
+                    packIdentityForCandidate(path),
+                    path.toString,
+                    file.getLen,
+                    if (parent == null) "" else parent.toString)
+              }
+        }
+        .toDF("name", "path", "len", "dataDir")
+        .dropDuplicates("name")
+        .repartition(parallelism)
+        .persist(StorageLevel.MEMORY_AND_DISK)
+      cached += candidates
+      val candidateSkipGc = candidates
+        .filter($"name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+        .limit(1)
+        .collect()
+        .nonEmpty
+      val canonicalCandidates = candidates
+        .filter($"name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+
+      betweenUsedCollections()
+      val usedPacks2 = collectUsedPacksDf().persist(StorageLevel.MEMORY_AND_DISK)
+      cached += usedPacks2
+      val skipGc2 = usedPacks2
+        .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+        .limit(1)
+        .collect()
+        .nonEmpty
+      val topologyAfter = snapshotTopology()
+      val used1Packs =
+        usedPacks.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+      val used2Packs =
+        usedPacks2.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+
+      val topologyChanged = topologyBefore != topologyAfter
+      val usedSetDifferences = used1Packs
+        .toDF()
+        .except(used2Packs.toDF())
+        .union(used2Packs.toDF().except(used1Packs.toDF()))
+      val usedSetChanged = usedSetDifferences.limit(1).collect().nonEmpty
+      val frozenAbort =
+        skipGc || skipGc2 || candidateSkipGc || topologyChanged || usedSetChanged
+
+      // Freeze every abort already observed by an action, and also retain dynamic gates so a cache
+      // miss that discovers a new unsafe mark cannot drop a live pack from the join.
+      // usedSetDifferences is in this lineage, but if mark caches are lost both passes recompute
+      // from the current filesystem and almost always agree, so the two-collection race check
+      // does not survive recomputation. The SKIP-marker gate still does.
+      val abortKeys = spark
+        .range(if (frozenAbort) 1L else 0L)
+        .select(functions.lit(1).as("abort_key"))
+        .union(abortKeyDf(usedPacks, "used_name"))
+        .union(abortKeyDf(usedPacks2, "used_name"))
+        .union(abortKeyDf(candidates, "name"))
+        .union(usedSetDifferences
+          .limit(1)
+          .select(functions.lit(1).as("abort_key")))
+        .distinct()
+      if (frozenAbort) {
+        val reason =
+          if (usedSetChanged) {
+            "the used pack set changed during collection"
+          } else {
+            "sidecars, manifests, or candidate identities cannot be trusted, or snapshot topology changed during collection"
+          }
+        logWarning(s"Skip managed blob pack GC for table ${table.fullName()} because $reason.")
+      }
+
+      val unused =
+        canonicalCandidates.join(used2Packs.toDF(), $"name" === $"used_name", "left_anti")
+      val toDelete = unused
+        .withColumn("abort_key", functions.lit(1))
+        .join(abortKeys, Seq("abort_key"), "left_anti")
+        .drop("abort_key")
+
+      val deleted: Dataset[(Long, Long)] = toDelete
+        .repartition(parallelism, $"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)
+              if (cleanManagedBlobFileIdempotently(deletedPath)) {
+                deletedFilesLenInBytes += fileInfo.getLong(2)
+                logInfo(s"Cleaned managed blob pack: $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, cached.toSeq)
+    } catch {
+      case t: Throwable =>
+        cached.foreach(_.unpersist())
+        throw t
+    }
+  }
+
+  private[procedure] def abortKeyDf(source: Dataset[_], column: String): DataFrame = {
+    source
+      .filter(functions.col(column) === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+      .limit(1)
+      .select(functions.lit(1).as("abort_key"))
+  }
+
+  private[procedure] def collectUsedPacksDf(): DataFrame = {
+    import spark.implicits._
+    val branches = validBranches()
+    val maxBranchParallelism = Math.min(branches.size(), parallelism)
+    val manifestLists = spark.sparkContext
+      .parallelize(branches.asScala.toSeq, maxBranchParallelism)
+      .flatMap {
+        branch =>
+          safelyGetAllSnapshots(branch).asScala.flatMap {
+            snapshot =>
+              Seq(
+                snapshot.changelogManifestList(),
+                snapshot.deltaManifestList(),
+                snapshot.baseManifestList())
+                .filter(_ != null)
+                .map((branch, _))
+          }
+      }
+      .distinct(parallelism)
+
+    val manifests = manifestLists
+      .mapPartitions {
+        lists =>
+          val branchManifestLists = new util.HashMap[String, ManifestList]()
+          lists.flatMap {
+            case (branch, listName) =>
+              val manifestList = branchManifestLists.computeIfAbsent(
+                branch,
+                (key: String) =>
+                  specifiedTable.switchToBranch(key).store.manifestListFactory.create)
+              val metas = retryReadingFiles[java.util.List[ManifestFileMeta]](
+                () => manifestList.readWithIOException(listName),
+                null)
+              if (metas == null) {
+                logWarning(
+                  s"Manifest list $listName is missing while collecting used managed blob packs. Skip pack GC this run.")
+                Iterator.single((true, branch, listName))
+              } else {
+                metas.asScala.iterator.map(meta => (false, branch, meta.fileName()))
+              }
+          }
+      }
+      .distinct(parallelism)
+
+    val sidecarWorkItems = manifests
+      .mapPartitions {
+        records =>
+          val branchManifestFiles = new util.HashMap[String, ManifestFile]()
+          val branchPathFactories = new util.HashMap[String, DataFilePathFactories]()
+          records.flatMap {
+            case (unsafe, _, _) if unsafe =>
+              Iterator.single(
+                (ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC, null: SidecarWorkItem))
+            case (_, branch, manifestName) =>
+              val branchTable = specifiedTable.switchToBranch(branch)
+              val manifestFile = branchManifestFiles.computeIfAbsent(
+                branch,
+                (_: String) => branchTable.store.manifestFileFactory.create)
+              val pathFactories = branchPathFactories.computeIfAbsent(
+                branch,
+                (_: String) => new DataFilePathFactories(branchTable.store.pathFactory))
+              val entries =
+                retryReadingFiles(() => manifestFile.readWithIOException(manifestName), null)
+              if (entries == null) {
+                logWarning(
+                  s"Manifest $manifestName is missing while collecting used managed blob packs. Skip pack GC this run.")
+                Iterator.single(
+                  (ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC, null: SidecarWorkItem))
+              } else {
+                entries.asScala.iterator.flatMap {
+                  entry =>
+                    createSidecarWorkItems(
+                      entry,
+                      pathFactories.get(entry.partition(), entry.bucket())).asScala.iterator
+                      .map(workItem => (workItem.dedupIdentity(), workItem))
+                }
+              }
+          }
+      }
+      .distinct(parallelism)
+
+    val rawUsedPackNames = sidecarWorkItems
+      .mapPartitions {
+        records =>
+          val scan = newReachabilityScan()
+          records.flatMap {
+            case (_, null) =>
+              Iterator.single(ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC)
+            case (_, workItem) =>
+              val names = new util.ArrayList[String]()
+              emitUsedPacksForSpark(
+                workItem,
+                scan,
+                new Consumer[String] {
+                  override def accept(name: String): Unit = names.add(name)
+                })
+              names.iterator().asScala
+          }
+      }
+    distinctUsedPackNames(rawUsedPackNames)
+  }
+
+  private[procedure] def distinctUsedPackNames(raw: RDD[String]): DataFrame = {
+    import spark.implicits._
+    raw.distinct(parallelism).toDF("used_name")
+  }
+
+}
+
+object SparkManagedBlobOrphanFilesClean extends SQLConfHelper {
+
+  private def checkParallelism(parallelism: Int): Unit = {
+    Preconditions.checkArgument(
+      parallelism > 0,
+      "Parallelism must be greater than 0, but was %s.",
+      Int.box(parallelism))
+  }
+
+  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()
+    }
+    checkParallelism(parallelism)
+
+    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)
+    }
+    var deletedFilesCount = 0L
+    var deletedFilesLenInBytes = 0L
+    // Run one table at a time and unpersist its marks before the next. Unlike
+    // SparkOrphanFilesClean, abort gating is per-table, so unioning every table's
+    // deletion Dataset into one job would keep all mark caches alive for the whole
+    // database and mix independent skip/used-set signals.
+    tables.foreach {
+      table =>
+        val (tableDeleted, tableCached) = new SparkManagedBlobOrphanFilesClean(
+          table,
+          olderThanMillis,
+          parallelism,
+          dryRun,
+          spark
+        ).doClean()
+        try {
+          val result = tableDeleted
+            .toDF("deletedFilesCount", "deletedFilesLenInBytes")
+            .agg(functions.sum("deletedFilesCount"), functions.sum("deletedFilesLenInBytes"))
+            .head()
+          assert(result.schema.size == 2, result.schema)
+          if (!result.isNullAt(0)) {
+            deletedFilesCount += result.getLong(0)
+            deletedFilesLenInBytes += result.getLong(1)
+          }
+        } finally {
+          tableCached.foreach(_.unpersist())
+        }
+    }
+    new CleanOrphanFilesResult(deletedFilesCount, deletedFilesLenInBytes)
+  }
+}
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..e5cd754be006
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala
@@ -0,0 +1,693 @@
+/*
+ * 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.catalog.Identifier
+import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.fs.{FileStatus, Path, SeekableInputStream}
+import org.apache.paimon.fs.local.LocalFileIO
+import org.apache.paimon.manifest.FileKind
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.{FileStoreTable, FileStoreTableFactory}
+import org.apache.paimon.utils.{DataFilePathFactories, DateTimeUtils, TraceableFileIO}
+
+import org.apache.spark.sql.Row
+
+import java.io.{File, IOException}
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
+
+import scala.collection.JavaConverters._
+
+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 = writeOrphanPack(table, "orphan.managed.blob")
+        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)
+        checkAnswer(
+          spark.sql(
+            s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', mode => '$mode')"),
+          Row(1, 3) :: Nil)
+
+        assert(!table.fileIO().exists(orphan))
+        referenced.foreach(pack => assert(table.fileIO().exists(pack)))
+      }
+
+      test(s"Paimon procedure: dry run remove orphan blobs ($mode)") {
+        createManagedBlobTable()
+        spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+
+        val table = loadTable("T")
+        val orphan = writeOrphanPack(table, "orphan.managed.blob")
+        Thread.sleep(2000)
+
+        val olderThan = DateTimeUtils.formatLocalDateTime(
+          DateTimeUtils.toLocalDateTime(System.currentTimeMillis()),
+          3)
+        checkAnswer(
+          spark.sql(
+            s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', dry_run => true, mode => '$mode')"),
+          Row(1, 3) :: Nil)
+        assert(table.fileIO().exists(orphan))
+      }
+
+      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)))
+      }
+  }
+
+  test("Paimon procedure: preserve distributed deletion parallelism") {
+    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)
+
+    withSQLConf("spark.sql.shuffle.partitions" -> "7") {
+      val cleaner =
+        SparkManagedBlobOrphanFilesClean(table, System.currentTimeMillis(), 2, true, spark)
+      val (deleted, cached) = cleaner.doClean()
+      try {
+        assert(deleted.rdd.getNumPartitions == 2, deleted.queryExecution.executedPlan)
+      } finally {
+        cached.foreach(_.unpersist())
+      }
+    }
+  }
+
+  Seq("local", "distributed").foreach {
+    mode =>
+      test(s"Paimon procedure: reject non-positive parallelism ($mode)") {
+        createManagedBlobTable()
+
+        val error = intercept[Exception] {
+          spark
+            .sql(s"CALL sys.remove_orphan_blobs(table => 'T', parallelism => 0, mode => '$mode')")
+            .collect()
+        }
+        assert(causeMessages(error).contains("Parallelism must be greater than 0, but was 0."))
+      }
+  }
+
+  Seq("local", "distributed").foreach {
+    mode =>
+      test(s"Paimon procedure: remove database orphan blobs ($mode)") {
+        createManagedBlobTable("T1")
+        spark.sql("INSERT INTO T1 VALUES (1, 'a', X'0102')")
+        createManagedBlobTable("T2")
+        spark.sql("INSERT INTO T2 VALUES (1, 'a', X'0102')")
+
+        try {
+          val table1 = loadTable("T1")
+          val table2 = loadTable("T2")
+          val orphan1 = writeOrphanPack(table1, "orphan.managed.blob")
+          val orphan2 = writeOrphanPack(table2, "orphan.managed.blob")
+          Thread.sleep(2000)
+
+          val olderThan = DateTimeUtils.formatLocalDateTime(
+            DateTimeUtils.toLocalDateTime(System.currentTimeMillis()),
+            3)
+          // beforeEach only drops T. Row(2, 6) is the exact sum of these two 3-byte orphans.
+          checkAnswer(
+            spark.sql(
+              s"CALL sys.remove_orphan_blobs(table => 'test.*', older_than => '$olderThan', mode => '$mode')"),
+            Row(2, 6) :: Nil)
+          assert(!table1.fileIO().exists(orphan1))
+          assert(!table2.fileIO().exists(orphan2))
+        } finally {
+          spark.sql("DROP TABLE IF EXISTS T1")
+          spark.sql("DROP TABLE IF EXISTS T2")
+        }
+      }
+  }
+
+  test("Paimon procedure: release cached marks after distributed failure") {
+    createManagedBlobTable()
+    spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+
+    val persistentBefore = spark.sparkContext.getPersistentRDDs.keySet
+    val cleaner = new SparkManagedBlobOrphanFilesClean(
+      loadTable("T"),
+      System.currentTimeMillis(),
+      2,
+      true,
+      spark) {
+      override protected def betweenUsedCollections(): Unit =
+        throw new RuntimeException("Expected failure after the first mark.")
+    }
+
+    val error = intercept[RuntimeException] {
+      cleaner.doClean()
+    }
+    assert(error.getMessage == "Expected failure after the first mark.")
+    assert(spark.sparkContext.getPersistentRDDs.keySet == persistentBefore)
+  }
+
+  test("Paimon procedure: log and freeze used pack set changes") {
+    val sparkSession = spark
+    import sparkSession.implicits._
+    createManagedBlobTable()
+
+    val table = loadTable("T")
+    val orphan = new Path(bucketPath(table), "used-set-changed.managed.blob")
+    table.fileIO().mkdirs(orphan.getParent)
+    table.fileIO().newOutputStream(orphan, false).close()
+    val markPass = new AtomicInteger()
+    val warnings = new scala.collection.mutable.ArrayBuffer[String]()
+    val cleaner = new SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, false, spark) {
+      override private[procedure] def collectUsedPacksDf(): org.apache.spark.sql.DataFrame =
+        Seq(if (markPass.getAndIncrement() == 0) "first-pack" else "second-pack")
+          .toDF("used_name")
+
+      override protected def logWarning(msg: => String): Unit = warnings += msg
+    }
+
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      assert(deleted.collect().map(_._1).sum == 0)
+    } finally {
+      cached.foreach(_.unpersist())
+    }
+    assert(table.fileIO().exists(orphan))
+    assert(warnings.exists(_.contains("used pack set changed during collection")))
+  }
+
+  test("Paimon procedure: skip cleanup when candidate identity cannot be canonicalized") {
+    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()
+    val relativeTable = FileStoreTableFactory.create(
+      new RemoveOrphanBlobsProcedureTest.RelativeListingFileIO,
+      table.location(),
+      table.schema())
+    val cleaner = SparkManagedBlobOrphanFilesClean(relativeTable, Long.MaxValue, 2, false, spark)
+
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      assert(deleted.collect().map(_._1).sum == 0)
+    } finally {
+      cached.foreach(_.unpersist(true))
+    }
+    assert(table.fileIO().exists(orphan))
+  }
+
+  test("Paimon procedure: dynamic abort gate survives cache recomputation") {
+    val sparkSession = spark
+    import sparkSession.implicits._
+    createManagedBlobTable()
+    val cleaner = SparkManagedBlobOrphanFilesClean(loadTable("T"), Long.MaxValue, 2, true, spark)
+    RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+    val dynamicMark = spark
+      .range(1)
+      .map {
+        _ =>
+          if (RemoveOrphanBlobsProcedureTest.sidecarReadsFailing) {
+            ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC
+          } else {
+            "safe-pack"
+          }
+      }
+      .toDF("used_name")
+      .persist()
+    val guardedDeletion = spark
+      .range(1)
+      .withColumn("abort_key", org.apache.spark.sql.functions.lit(1))
+      .join(cleaner.abortKeyDf(dynamicMark, "used_name"), Seq("abort_key"), "left_anti")
+
+    try {
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+      assert(dynamicMark.collect().map(_.getString(0)).sameElements(Array("safe-pack")))
+      dynamicMark.unpersist(true)
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(true)
+      assert(guardedDeletion.count() == 0)
+    } finally {
+      dynamicMark.unpersist(true)
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+    }
+  }
+
+  test("Paimon procedure: observed unsafe marks survive cache recomputation") {
+    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()
+    val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)
+      .filterNot(_.getName == orphan.getName)
+    assert(referenced.nonEmpty)
+
+    RemoveOrphanBlobsProcedureTest.failSidecarReads.set(true)
+    val failingTable = FileStoreTableFactory.create(
+      new RemoveOrphanBlobsProcedureTest.FailingSidecarFileIO,
+      table.location(),
+      table.schema())
+    val cleaner =
+      SparkManagedBlobOrphanFilesClean(failingTable, Long.MaxValue, 2, false, spark)
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      cached.foreach(_.unpersist(true))
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+      assert(deleted.collect().map(_._1).sum == 0)
+    } finally {
+      cached.foreach(_.unpersist())
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+    }
+    assert(table.fileIO().exists(orphan))
+    referenced.foreach(pack => assert(table.fileIO().exists(pack)))
+  }
+
+  test("Paimon procedure: unpersisted unsafe marks abort actual deletion") {
+    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()
+    val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)
+      .filterNot(_.getName == orphan.getName)
+    assert(referenced.nonEmpty)
+
+    RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+    val failingTable = FileStoreTableFactory.create(
+      new RemoveOrphanBlobsProcedureTest.FailingSidecarFileIO,
+      table.location(),
+      table.schema())
+    val cleaner =
+      SparkManagedBlobOrphanFilesClean(failingTable, Long.MaxValue, 2, false, spark)
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      RemoveOrphanBlobsProcedureTest.failedSidecarReadCount.set(0)
+      cached.foreach(_.unpersist(true))
+      clearShuffleOutputs()
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(true)
+      assert(deleted.collect().map(_._1).sum == 0)
+      assert(RemoveOrphanBlobsProcedureTest.failedSidecarReadCount.get() > 0)
+    } finally {
+      cached.foreach(_.unpersist())
+      RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false)
+    }
+    assert(table.fileIO().exists(orphan))
+    referenced.foreach(pack => assert(table.fileIO().exists(pack)))
+  }
+
+  test("Paimon procedure: distinct used pack identities before caching the mark") {
+    createManagedBlobTable()
+    spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+    spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')")
+    spark.sql("CALL sys.compact(table => 'T')")
+
+    val table = loadTable("T")
+    val cleaner = SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, true, spark)
+    val identity = cleaner.collectUsedPacksDf().head().getString(0)
+    val raw = spark.sparkContext.parallelize(Seq(identity, identity), 2)
+    assert(raw.count() == 2)
+    assert(raw.distinct().count() == 1)
+    assert(
+      cleaner
+        .distinctUsedPackNames(raw)
+        .collect()
+        .map(_.getString(0))
+        .sameElements(Array(identity)))
+  }
+
+  test("Paimon procedure: distinct used pack identities from duplicate sidecars") {
+    val sparkSession = spark
+    import sparkSession.implicits._
+    createManagedBlobTable()
+    spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+    spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')")
+
+    val table = loadTable("T")
+    val sidecars = filesWithSuffix(table, ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)
+    assert(sidecars.size >= 2, s"Expected two managed BLOB sidecars: $sidecars")
+
+    val firstSidecar = sidecars.head
+    val secondSidecar = sidecars.tail.head
+    val firstReferences = ManagedBlobReferenceFile.read(table.fileIO(), firstSidecar)
+    assert(!firstReferences.isEmpty)
+    table.fileIO().delete(secondSidecar, false)
+    ManagedBlobReferenceFile.write(table.fileIO(), secondSidecar, firstReferences)
+
+    val sharedPack = ManagedBlobOrphanFilesClean.packIdentity(firstReferences.get(0).toPath())
+    val sidecarsReferencingSharedPack = Seq(firstSidecar, secondSidecar).count {
+      sidecar =>
+        ManagedBlobReferenceFile
+          .read(table.fileIO(), sidecar)
+          .asScala
+          .exists(
+            reference => ManagedBlobOrphanFilesClean.packIdentity(reference.toPath()) == sharedPack)
+    }
+    assert(sidecarsReferencingSharedPack == 2)
+
+    val cleaner = SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, true, spark)
+    val used = cleaner.collectUsedPacksDf()
+    assert(used.filter($"used_name" === sharedPack).count() == 1)
+  }
+
+  test("Paimon procedure: canonical alias candidates are counted once") {
+    createManagedBlobTable()
+    val table = loadTable("T")
+    val orphan = new Path(bucketPath(table), "duplicate.managed.blob")
+    table.fileIO().mkdirs(orphan.getParent)
+    table.fileIO().newOutputStream(orphan, false).close()
+
+    val duplicateListingTable = FileStoreTableFactory.create(
+      new RemoveOrphanBlobsProcedureTest.CanonicalAliasListingFileIO,
+      table.location(),
+      table.schema())
+    val aliases = duplicateListingTable.fileIO
+      .listStatus(orphan.getParent)
+      .filter(_.getPath.getName == orphan.getName)
+    assert(aliases.map(_.getPath.toString).distinct.length == 2)
+    assert(
+      aliases
+        .map(status => ManagedBlobOrphanFilesClean.packIdentity(status.getPath))
+        .distinct
+        .length == 1)
+    val cleaner =
+      SparkManagedBlobOrphanFilesClean(duplicateListingTable, Long.MaxValue, 2, false, spark)
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      assert(deleted.collect().map(_._1).sum == 1)
+    } finally {
+      cached.foreach(_.unpersist())
+    }
+    assert(!table.fileIO().exists(orphan))
+  }
+
+  test("Paimon procedure: deduplicate shared sidecar globally in each mark pass") {
+    createManagedBlobTable()
+    spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')")
+    spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')")
+
+    val table = loadTable("T")
+    val commit = table.newBatchWriteBuilder().newCommit()
+    try {
+      commit.compactManifests()
+    } finally {
+      commit.close()
+    }
+
+    val occurrences = sidecarManifestOccurrences(table)
+    val (sidecar, containingManifests) = occurrences
+      .find(_._2.distinct.size >= 2)
+      .getOrElse(fail(s"No sidecar is shared by distinct manifests: $occurrences"))
+    val parallelism = (2 to 64)
+      .find {
+        candidate =>
+          containingManifests
+            .map {
+              manifest =>
+                Math.floorMod(
+                  (false, Identifier.DEFAULT_MAIN_BRANCH, manifest).hashCode(),
+                  candidate)
+            }
+            .distinct
+            .size >= 2
+      }
+      .getOrElse(fail(s"Shared manifests cannot be assigned to different partitions: $occurrences"))
+
+    RemoveOrphanBlobsProcedureTest.resetSidecarReadCounts()
+    val countingTable = FileStoreTableFactory.create(
+      new RemoveOrphanBlobsProcedureTest.CountingSidecarFileIO,
+      table.location(),
+      table.schema())
+    val cleaner =
+      SparkManagedBlobOrphanFilesClean(countingTable, Long.MaxValue, parallelism, true, spark)
+
+    val (deleted, cached) = cleaner.doClean()
+    try {
+      deleted.collect()
+    } finally {
+      cached.foreach(_.unpersist())
+    }
+    assert(RemoveOrphanBlobsProcedureTest.sidecarReadCount(sidecar) == 2)
+  }
+
+  private def createManagedBlobTable(name: String = "T"): Unit = {
+    spark.sql(s"""
+                 |CREATE TABLE $name (id INT, name STRING, payload BINARY)
+                 |USING PAIMON
+                 |TBLPROPERTIES (
+                 |  'primary-key'='id',
+                 |  'bucket'='1',
+                 |  'changelog-producer'='none',
+                 |  'blob-field'='payload')
+                 |""".stripMargin)
+  }
+
+  private def writeOrphanPack(
+      table: FileStoreTable,
+      fileName: String,
+      payload: Array[Byte] = Array[Byte](1, 2, 3)): Path = {
+    val orphan = new Path(bucketPath(table), fileName)
+    val out = table.fileIO().newOutputStream(orphan, false)
+    try {
+      out.write(payload)
+    } finally {
+      out.close()
+    }
+    orphan
+  }
+
+  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))
+    }
+  }
+
+  private def sidecarManifestOccurrences(table: FileStoreTable): Map[String, Seq[String]] = {
+    val manifestList = table.store().manifestListFactory().create()
+    val manifestFile = table.store().manifestFileFactory().create()
+    val pathFactories = new DataFilePathFactories(table.store().pathFactory())
+    val manifests = table
+      .snapshotManager()
+      .safelyGetAllSnapshots()
+      .asScala
+      .flatMap(snapshot => manifestList.readDataManifests(snapshot).asScala)
+      .map(_.fileName())
+      .distinct
+    manifests
+      .flatMap {
+        manifest =>
+          manifestFile
+            .read(manifest)
+            .asScala
+            .filter(_.kind() == FileKind.ADD)
+            .flatMap {
+              entry =>
+                val dataFile = pathFactories
+                  .get(entry.partition(), entry.bucket())
+                  .toPath(entry)
+                Option(entry.file().extraFiles()).toSeq
+                  .flatMap(_.asScala)
+                  .filter(_.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX))
+                  .map(extra => (new Path(dataFile.getParent, extra).toUri.getPath, manifest))
+            }
+      }
+      .groupBy(_._1)
+      .map { case (sidecar, values) => (sidecar, values.map(_._2).toList) }
+  }
+
+  private def causeMessages(error: Throwable): String = {
+    Iterator
+      .iterate(error)(_.getCause)
+      .takeWhile(_ != null)
+      .flatMap(e => Option(e.getMessage))
+      .mkString("\n")
+  }
+
+  // Drop Spark shuffle map outputs so a later action recomputes `deleted` instead of replaying
+  // cached shuffle blocks. Used to prove the abort gate still holds after cache eviction. Relies
+  // on Spark's private MapOutputTracker.unregisterAllMapAndMergeOutput.
+  private def clearShuffleOutputs(): Unit = {
+    val sparkEnv = spark.sparkContext.getClass.getMethod("env").invoke(spark.sparkContext)
+    val mapOutputTracker =
+      sparkEnv.getClass.getMethod("mapOutputTracker").invoke(sparkEnv)
+    val shuffleStatuses = mapOutputTracker.getClass
+      .getMethod("shuffleStatuses")
+      .invoke(mapOutputTracker)
+      .asInstanceOf[scala.collection.Map[Int, _]]
+    val unregister = mapOutputTracker.getClass
+      .getMethod("unregisterAllMapAndMergeOutput", Integer.TYPE)
+    shuffleStatuses.keys.toSeq.foreach(id => unregister.invoke(mapOutputTracker, Int.box(id)))
+  }
+}
+
+private object RemoveOrphanBlobsProcedureTest {
+
+  private val sidecarReadCounts = new ConcurrentHashMap[String, AtomicInteger]()
+  private val failSidecarReads = new AtomicBoolean()
+  private val failedSidecarReadCount = new AtomicInteger()
+
+  private def sidecarReadsFailing: Boolean = failSidecarReads.get()
+
+  private def resetSidecarReadCounts(): Unit = sidecarReadCounts.clear()
+
+  private def sidecarReadCount(path: String): Int =
+    Option(sidecarReadCounts.get(path)).map(_.get()).getOrElse(0)
+
+  private class CountingSidecarFileIO extends LocalFileIO {
+
+    override def newInputStream(path: Path): SeekableInputStream = {
+      if (path.getName.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+        sidecarReadCounts
+          .computeIfAbsent(path.toUri.getPath, (_: String) => new AtomicInteger())
+          .incrementAndGet()
+      }
+      super.newInputStream(path)
+    }
+  }
+
+  private class FailingSidecarFileIO extends LocalFileIO {
+
+    override def newInputStream(path: Path): SeekableInputStream = {
+      if (
+        failSidecarReads.get() && path.getName.endsWith(
+          ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)
+      ) {
+        failedSidecarReadCount.incrementAndGet()
+        throw new IOException("Injected sidecar read failure.")
+      }
+      super.newInputStream(path)
+    }
+  }
+
+  private class CanonicalAliasListingFileIO extends TraceableFileIO {
+
+    override def listStatus(path: Path): Array[FileStatus] = {
+      val statuses = super.listStatus(path)
+      if (statuses == null) {
+        null
+      } else {
+        statuses.flatMap {
+          status =>
+            if (status.getPath.getName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) {
+              val alias =
+                new Path(s"hdfs://duplicate-listing${status.getPath.toUri.getPath}")
+              Array(status, withPath(status, alias))
+            } else {
+              Array(status)
+            }
+        }
+      }
+    }
+
+    private def withPath(status: FileStatus, path: Path): FileStatus = new FileStatus {
+      override def getLen: Long = status.getLen
+
+      override def isDir: Boolean = status.isDir
+
+      override def getPath: Path = path
+
+      override def getModificationTime: Long = status.getModificationTime
+    }
+  }
+
+  private class RelativeListingFileIO extends TraceableFileIO {
+
+    override def listStatus(path: Path): Array[FileStatus] = {
+      val statuses = super.listStatus(path)
+      if (statuses == null) {
+        null
+      } else {
+        statuses.map {
+          status =>
+            if (status.getPath.getName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) {
+              withPath(status, new Path(status.getPath.getName))
+            } else {
+              status
+            }
+        }
+      }
+    }
+
+    override def getFileStatus(path: Path): FileStatus = {
+      val status = super.getFileStatus(path)
+      if (new File(path.toUri.getPath).isAbsolute) {
+        status
+      } else {
+        withPath(status, path)
+      }
+    }
+
+    private def withPath(status: FileStatus, path: Path): FileStatus = new FileStatus {
+      override def getLen: Long = status.getLen
+
+      override def isDir: Boolean = status.isDir
+
+      override def getPath: Path = path
+
+      override def getModificationTime: Long = status.getModificationTime
+    }
+  }
+}