[flink] Support remove_orphan_blobs action and procedure.

docs/docs/flink/procedures.md

@@ -72,5 +72,5 @@ defaults are specific to the procedure.
 | [Table Operations](./procedures/table-operations) | [`merge_into`](./procedures/table-operations#merge_into), [`data_evolution_merge_into`](./procedures/table-operations#data_evolution_merge_into), [`migrate_database`](./procedures/table-operations#migrate_database), [`migrate_table`](./procedures/table-operations#migrate_table), [`clone`](./procedures/table-operations#clone), [`copy_files`](./procedures/table-operations#copy_files), [`alter_column_default_value`](./procedures/table-operations#alter_column_default_value), [`drop_partition`](./procedures/table-operations#drop_partition), [`mark_partition_done`](./procedures/table-operations#mark_partition_done) |
 | [Indexes and Search](./procedures/indexes) | [`create_global_index`](./procedures/indexes#create_global_index), [`drop_global_index`](./procedures/indexes#drop_global_index), [`full_text_search`](./procedures/indexes#full_text_search), [`vector_search`](./procedures/indexes#vector_search), [`rewrite_file_index`](./procedures/indexes#rewrite_file_index) |
 | [Consumers and Query Service](./procedures/consumers) | [`reset_consumer`](./procedures/consumers#reset_consumer), [`clear_consumers`](./procedures/consumers#clear_consumers), [`query_service`](./procedures/consumers#query_service) |
-| [Cleanup and Repair](./procedures/repair) | [`remove_orphan_files`](./procedures/repair#remove_orphan_files), [`remove_unexisting_files`](./procedures/repair#remove_unexisting_files), [`remove_unexisting_manifests`](./procedures/repair#remove_unexisting_manifests), [`repair`](./procedures/repair#repair), [`repair_earliest_snapshot`](./procedures/repair#repair_earliest_snapshot) |
+| [Cleanup and Repair](./procedures/repair) | [`remove_orphan_files`](./procedures/repair#remove_orphan_files), [`remove_orphan_blobs`](./procedures/repair#remove_orphan_blobs), [`remove_unexisting_files`](./procedures/repair#remove_unexisting_files), [`remove_unexisting_manifests`](./procedures/repair#remove_unexisting_manifests), [`repair`](./procedures/repair#repair), [`repair_earliest_snapshot`](./procedures/repair#repair_earliest_snapshot) |
 | [Views and Functions](./procedures/catalog) | [`alter_view_dialect`](./procedures/catalog#alter_view_dialect), [`create_function`](./procedures/catalog#create_function), [`alter_function`](./procedures/catalog#alter_function), [`drop_function`](./procedures/catalog#drop_function) |

docs/docs/flink/procedures/repair.md

@@ -90,6 +90,73 @@ CALL sys.remove_orphan_files(
 );
 ```
 
+This procedure does not delete primary-key `.managed.blob` packs. Use [`remove_orphan_blobs`](#remove_orphan_blobs).
+
+## remove_orphan_blobs
+
+Remove unreferenced primary-key `.managed.blob` packs.
+
+**Arguments**
+
+- `table`: the target table identifier. Cannot be empty, you can use `database_name.*` to clean the whole database.
+
+- `olderThan`: 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.
+
+- `dryRun`: 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`: per-table concurrency. In `distributed` mode this is the Flink task parallelism of each table job. 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 after another, 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`: The mode of remove orphan blob procedure (`local` or `distributed`). By default is `distributed`.
+
+**Syntax**
+
+```sql
+-- Use named argument
+CALL [catalog.]sys.remove_orphan_blobs(
+    `table` => 'identifier',
+    older_than => 'olderThan',
+    dry_run => 'dryRun',
+    parallelism => parallelism,
+    mode => 'mode'
+);
+
+-- Use indexed argument
+CALL [catalog.]sys.remove_orphan_blobs('identifier');
+
+CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan');
+
+CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun');
+
+CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun','parallelism');
+
+CALL [catalog.]sys.remove_orphan_blobs('identifier', 'olderThan', 'dryRun','parallelism','mode');
+```
+
+**Example**
+
+```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.apache.org/docs/master/api/java/org/apache/paimon/flink/action/RemoveUnexistingFilesAction.html) for detailed use cases. Arguments:

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

@@ -259,8 +259,9 @@ extra files because more than one retained data file can reference the same pack
 ## Garbage Collection
 
 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).
+Local cleanup is `LocalManagedBlobOrphanFilesClean`. Spark exposes the same cleanup as
+[`remove_orphan_blobs`](../spark/procedures/maintenance#remove_orphan_blobs);
+Flink exposes it as [`remove_orphan_blobs`](../flink/procedures/repair#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).

paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java

@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.procedure;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Locale;
+
+import static org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean.validateParallelism;
+import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
+
+/**
+ * Remove orphan managed BLOB packs procedure. Usage:
+ *
+ * <pre><code>
+ *  CALL sys.remove_orphan_blobs('tableId')
+ *
+ *  CALL sys.remove_orphan_blobs('tableId', '2023-12-31 23:59:59')
+ *
+ *  CALL sys.remove_orphan_blobs('databaseName.*', '2023-12-31 23:59:59')
+ * </code></pre>
+ */
+public class RemoveOrphanBlobsProcedure extends ProcedureBase {
+
+    public static final String IDENTIFIER = "remove_orphan_blobs";
+
+    public String[] call(ProcedureContext procedureContext, String tableId) throws Exception {
+        return call(procedureContext, tableId, "");
+    }
+
+    public String[] call(ProcedureContext procedureContext, String tableId, String olderThan)
+            throws Exception {
+        return call(procedureContext, tableId, olderThan, false);
+    }
+
+    public String[] call(
+            ProcedureContext procedureContext, String tableId, String olderThan, boolean dryRun)
+            throws Exception {
+        return call(procedureContext, tableId, olderThan, dryRun, null);
+    }
+
+    public String[] call(
+            ProcedureContext procedureContext,
+            String tableId,
+            String olderThan,
+            boolean dryRun,
+            Integer parallelism)
+            throws Exception {
+        return call(procedureContext, tableId, olderThan, dryRun, parallelism, null);
+    }
+
+    public String[] call(
+            ProcedureContext procedureContext,
+            String tableId,
+            String olderThan,
+            boolean dryRun,
+            Integer parallelism,
+            String mode)
+            throws Exception {
+        validateParallelism(parallelism);
+        Identifier identifier = Identifier.fromString(tableId);
+        String databaseName = identifier.getDatabaseName();
+        String tableName = identifier.getObjectName();
+        if (mode == null) {
+            mode = "DISTRIBUTED";
+        }
+
+        CleanOrphanFilesResult result;
+        try {
+            switch (mode.toUpperCase(Locale.ROOT)) {
+                case "DISTRIBUTED":
+                    result =
+                            FlinkManagedBlobOrphanFilesClean.executeDatabase(
+                                    procedureContext.getExecutionEnvironment(),
+                                    catalog,
+                                    olderThanMillis(olderThan),
+                                    dryRun,
+                                    parallelism,
+                                    databaseName,
+                                    tableName);
+                    break;
+                case "LOCAL":
+                    result =
+                            LocalManagedBlobOrphanFilesClean.executeDatabase(
+                                    catalog,
+                                    databaseName,
+                                    tableName,
+                                    olderThanMillis(olderThan),
+                                    parallelism,
+                                    dryRun);
+                    break;
+                default:
+                    throw new IllegalArgumentException(
+                            "Unknown mode: "
+                                    + mode
+                                    + ". Only 'DISTRIBUTED' and 'LOCAL' are supported.");
+            }
+            return new String[] {
+                String.valueOf(result.getDeletedFileCount()),
+                String.valueOf(result.getDeletedFileTotalLenInBytes())
+            };
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+}

paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.flink.action.RemoveOrphanBlobsAction;
+import org.apache.paimon.flink.action.RemoveOrphanBlobsActionITCaseBase;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink 1.18. */
+public class RemoveOrphanBlobsActionITCase extends RemoveOrphanBlobsActionITCaseBase {
+
+    protected boolean supportNamedArgument() {
+        return false;
+    }
+}

paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.flink.action.RemoveOrphanBlobsAction;
+import org.apache.paimon.flink.action.RemoveOrphanBlobsActionITCaseBase;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink 1.19. */
+public class RemoveOrphanBlobsActionITCase extends RemoveOrphanBlobsActionITCaseBase {}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+import static org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean.executeDatabase;
+import static org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean.validateParallelism;
+import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
+
+/** Action to remove unreferenced primary-key managed BLOB packs. */
+public class RemoveOrphanBlobsAction extends ActionBase {
+
+    private final String databaseName;
+    @Nullable private final String tableName;
+    @Nullable private final Integer parallelism;
+
+    private String olderThan = null;
+    private boolean dryRun = false;
+
+    public RemoveOrphanBlobsAction(
+            String databaseName,
+            @Nullable String tableName,
+            @Nullable String parallelism,
+            Map<String, String> catalogConfig) {
+        super(catalogConfig);
+        this.databaseName = databaseName;
+        this.tableName = tableName;
+        this.parallelism = parseParallelism(parallelism);
+    }
+
+    public void olderThan(String olderThan) {
+        this.olderThan = olderThan;
+    }
+
+    public void dryRun() {
+        this.dryRun = true;
+    }
+
+    @Override
+    public void run() throws Exception {
+        executeDatabase(
+                env,
+                catalog,
+                olderThanMillis(olderThan),
+                dryRun,
+                parallelism,
+                databaseName,
+                tableName);
+    }
+
+    @Nullable
+    static Integer parseParallelism(@Nullable String parallelism) {
+        if (parallelism == null) {
+            return null;
+        }
+
+        final int parsed;
+        try {
+            parsed = Integer.parseInt(parallelism);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Parallelism must be a positive integer, but was '%s'.", parallelism),
+                    e);
+        }
+        validateParallelism(parsed);
+        return parsed;
+    }
+}