[spark] Support remove_orphan_blobs procedure.

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

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

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

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

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";
+    }
+}

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);
+    }
+}