This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1069106dd5 [flink] Support remove_orphan_blobs action and procedure.
(#10014)
1069106dd5 is described below
commit 1069106dd59beb7d4572230f2c680ebd3aa17ec7
Author: Wenchao Wu <[email protected]>
AuthorDate: Tue Sep 22 11:28:20 2026 +0800
[flink] Support remove_orphan_blobs action and procedure. (#10014)
---
docs/docs/flink/procedures.md | 2 +-
docs/docs/flink/procedures/repair.md | 67 ++
docs/docs/primary-key-table/blob-storage.md | 5 +-
.../procedure/RemoveOrphanBlobsProcedure.java | 132 +++
.../flink/RemoveOrphanBlobsActionITCase.java | 30 +
.../flink/RemoveOrphanBlobsActionITCase.java | 25 +
.../flink/action/RemoveOrphanBlobsAction.java | 88 ++
.../action/RemoveOrphanBlobsActionFactory.java | 98 +++
.../orphan/FlinkManagedBlobOrphanFilesClean.java | 965 +++++++++++++++++++++
.../paimon/flink/orphan/FlinkOrphanFilesClean.java | 4 +
.../procedure/RemoveOrphanBlobsProcedure.java | 120 +++
.../services/org.apache.paimon.factories.Factory | 2 +
.../paimon/flink/action/ActionJobCoverageTest.java | 5 +-
.../action/RemoveOrphanBlobsActionITCase.java | 22 +
.../action/RemoveOrphanBlobsActionITCaseBase.java | 935 ++++++++++++++++++++
.../action/RemoveOrphanFilesActionITCaseBase.java | 24 +
16 files changed, 2520 insertions(+), 4 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index 831813994d..b248d8ab1c 100644
--- a/docs/docs/flink/procedures.md
+++ b/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-op [...]
| [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) |
diff --git a/docs/docs/flink/procedures/repair.md
b/docs/docs/flink/procedures/repair.md
index 84effe5f2b..cbbf919ec9 100644
--- a/docs/docs/flink/procedures/repair.md
+++ b/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:
diff --git a/docs/docs/primary-key-table/blob-storage.md
b/docs/docs/primary-key-table/blob-storage.md
index e7fd8f21f3..dff5e8f83c 100644
--- a/docs/docs/primary-key-table/blob-storage.md
+++ b/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).
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
new file mode 100644
index 0000000000..50ee675832
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 0000000000..c987d9d6c8
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
b/paimon-flink/paimon-flink-1.19/src/test/java/org/apache/paimon/flink/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 0000000000..16fd381355
--- /dev/null
+++
b/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 {}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsAction.java
new file mode 100644
index 0000000000..700d0c7146
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java
new file mode 100644
index 0000000000..ed39315e69
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionFactory.java
@@ -0,0 +1,98 @@
+/*
+ * 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 java.util.Optional;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory to create {@link RemoveOrphanBlobsAction}. */
+public class RemoveOrphanBlobsActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "remove_orphan_blobs";
+ private static final String OLDER_THAN = "older_than";
+ private static final String DRY_RUN = "dry_run";
+ private static final String PARALLELISM = "parallelism";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ boolean dryRun = false;
+ if (params.has(DRY_RUN)) {
+ String dryRunValue = params.get(DRY_RUN);
+ checkArgument(
+ "true".equalsIgnoreCase(dryRunValue) ||
"false".equalsIgnoreCase(dryRunValue),
+ "Argument 'dry_run' must be either 'true' or 'false', but
was '%s'.",
+ dryRunValue);
+ dryRun = Boolean.parseBoolean(dryRunValue);
+ }
+
+ RemoveOrphanBlobsAction action =
+ new RemoveOrphanBlobsAction(
+ params.getRequired(DATABASE),
+ params.get(TABLE),
+ params.get(PARALLELISM),
+ catalogConfigMap(params));
+
+ if (params.has(OLDER_THAN)) {
+ action.olderThan(params.get(OLDER_THAN));
+ }
+
+ if (dryRun) {
+ action.dryRun();
+ }
+
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println(
+ "Action \"remove_orphan_blobs\" removes unreferenced
primary-key managed BLOB packs.");
+ System.out.println();
+ System.out.println("Syntax:");
+ System.out.println(
+ " remove_orphan_blobs \\\n"
+ + "--warehouse <warehouse_path> \\\n"
+ + "--database <database_name> \\\n"
+ + "--table <table_name> \\\n"
+ + "[--older_than <timestamp>] \\\n"
+ + "[--dry_run <false/true>] \\\n"
+ + "[--parallelism <positive_integer>]");
+ System.out.println();
+ System.out.println(
+ "To avoid deleting newly written packs, the default cutoff is
1 day before the action starts. "
+ + "'--older_than' sets the absolute cutoff timestamp;
only packs with an earlier modification time are eligible. "
+ + "<timestamp> format: yyyy-MM-dd HH:mm:ss");
+ System.out.println();
+ System.out.println(
+ "When '--dry_run true', calculate the orphan pack count and
total bytes without deleting files. Default is false.");
+ System.out.println();
+ System.out.println(
+ "'--parallelism' controls the parallelism of each table
cleanup job. It must be greater than 0.");
+ System.out.println();
+ System.out.println(
+ "If the table is null or *, all managed BLOB packs in all
tables under the db will be cleaned up.");
+ System.out.println();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java
new file mode 100644
index 0000000000..e5498176b3
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java
@@ -0,0 +1,965 @@
+/*
+ * 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.orphan;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.flink.utils.BoundedTwoInputOperator;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean;
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean.SidecarWorkItem;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.DataFilePathFactories;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.apache.flink.api.common.BatchShuffleMode;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.operators.InputSelection;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static
org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Flink {@link ManagedBlobOrphanFilesClean}. */
+public class FlinkManagedBlobOrphanFilesClean extends
ManagedBlobOrphanFilesClean {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(FlinkManagedBlobOrphanFilesClean.class);
+
+ @Nullable private final Integer parallelism;
+
+ public FlinkManagedBlobOrphanFilesClean(
+ FileStoreTable table,
+ long olderThanMillis,
+ boolean dryRun,
+ @Nullable Integer parallelism) {
+ super(table, olderThanMillis, dryRun);
+ validateParallelism(parallelism);
+ this.parallelism = parallelism;
+ }
+
+ @Nullable
+ public DataStream<CleanOrphanFilesResult>
doClean(StreamExecutionEnvironment env) {
+ configureEnv(env);
+ FrozenUsedPacks frozen = freezeUsedPacks(env);
+ if (frozen.abort) {
+ return env.fromCollection(
+ Collections.singletonList(new CleanOrphanFilesResult(0,
0)),
+ TypeInformation.of(CleanOrphanFilesResult.class));
+ }
+ return deleteUnused(env, frozen.used);
+ }
+
+ private void configureEnv(StreamExecutionEnvironment env) {
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH);
+ flinkConf.set(ExecutionOptions.SORT_INPUTS, false);
+ flinkConf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false);
+ // Bounded two-input operators fully consume the first input before
the second. A pipelined
+ // shuffle can deadlock that build/probe order, so force blocking
exchanges.
+ flinkConf.set(ExecutionOptions.BATCH_SHUFFLE_MODE,
BatchShuffleMode.ALL_EXCHANGES_BLOCKING);
+ if (parallelism != null) {
+ flinkConf.set(CoreOptions.DEFAULT_PARALLELISM, parallelism);
+ }
+
flinkConf.setString("execution.batch.adaptive.auto-parallelism.enabled",
"false");
+ env.configure(flinkConf);
+ }
+
+ private FrozenUsedPacks freezeUsedPacks(StreamExecutionEnvironment env) {
+ List<String> topologyBefore;
+ try {
+ topologyBefore = snapshotTopology();
+ } catch (java.io.IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ List<String> branches = validBranches();
+ final OutputTag<Boolean> firstMarkSkipGcTag =
+ new OutputTag<Boolean>("first-managed-blob-mark-skip") {};
+ SingleOutputStreamOperator<Tuple2<String, String>> firstManifestLists =
+ env.fromCollection(branches)
+ .name("branch-source")
+ .process(
+ new ProcessFunction<String, Tuple2<String,
String>>() {
+ @Override
+ public void processElement(
+ String branch,
+ ProcessFunction<String,
Tuple2<String, String>>.Context
+ ctx,
+ Collector<Tuple2<String, String>>
out)
+ throws Exception {
+ emitManifestLists(branch,
out::collect);
+ }
+ })
+ .name("collect-first-mark-manifest-lists");
+
+ SingleOutputStreamOperator<String> usedPacks =
+ collectUsedPacks(firstManifestLists, firstMarkSkipGcTag,
"first");
+
+ DataStream<Boolean> firstMarkCompleted = markCompletion(usedPacks,
"first");
+ SingleOutputStreamOperator<Tuple2<String, String>> secondManifestLists
=
+ firstMarkCompleted
+ .transform(
+ "wait-before-second-managed-blob-mark",
+ Types.TUPLE(Types.STRING, Types.STRING),
+ new BoundedOneInputOperator<Boolean,
Tuple2<String, String>>() {
+
+ @Override
+ public void
processElement(StreamRecord<Boolean> element) {}
+
+ @Override
+ public void endInput() throws Exception {
+ for (String branch : branches) {
+ emitManifestLists(
+ branch,
+ manifestList ->
+ output.collect(
+ new
StreamRecord<>(
+
manifestList)));
+ }
+ }
+ })
+ .forceNonParallel();
+
+ final OutputTag<Boolean> secondMarkSkipGcTag =
+ new OutputTag<Boolean>("second-managed-blob-mark-skip") {};
+ SingleOutputStreamOperator<String> usedPacks2 =
+ collectUsedPacks(secondManifestLists, secondMarkSkipGcTag,
"second");
+
+ DataStream<Boolean> usedPacksChanged = compareUsedPacks(usedPacks,
usedPacks2);
+ DataStream<Boolean> topologyChanged =
+ markCompletion(usedPacks2, "second")
+ .transform(
+ "check-managed-blob-snapshot-topology",
+ TypeInformation.of(Boolean.class),
+ new BoundedOneInputOperator<Boolean,
Boolean>() {
+
+ @Override
+ public void
processElement(StreamRecord<Boolean> element) {}
+
+ @Override
+ public void endInput() throws Exception {
+ List<String> topologyAfter =
snapshotTopology();
+ if
(!topologyBefore.equals(topologyAfter)) {
+ LOG.warn(
+ "Skip managed blob pack GC
for table {} because snapshot topology changed during used-pack collection.",
+ table.fullName());
+ output.collect(new
StreamRecord<>(Boolean.TRUE));
+ }
+ }
+ })
+ .forceNonParallel();
+
+ DataStream<Boolean> skipGc =
+ usedPacks
+ .getSideOutput(firstMarkSkipGcTag)
+ .union(
+ usedPacks2.getSideOutput(secondMarkSkipGcTag),
+ usedPacksChanged,
+ topologyChanged);
+
+ // executeAndCollect materializes abort flags and used names together.
Deletion re-sources
+ // the frozen names and must not reconnect the mark DAG: a lost
used-pack shuffle with a
+ // surviving sidecar shuffle can drop a reused pack and still agree.
Cost: client memory
+ // grows with the used-pack set. An OOM fails the job instead of
deleting live packs.
+ DataStream<String> freezeRows =
+ usedPacks2.union(
+ skipGc.map(
+ new MapFunction<Boolean, String>() {
+ @Override
+ public String map(Boolean value) {
+ return SKIP_MANAGED_BLOB_GC;
+ }
+ })
+ .returns(STRING_TYPE_INFO));
+
+ HashSet<String> used = new HashSet<>();
+ boolean abort = false;
+ try (CloseableIterator<String> iterator =
+ freezeRows.executeAndCollect("FreezeManagedBlobUsedPacks")) {
+ while (iterator.hasNext()) {
+ String name = iterator.next();
+ if (SKIP_MANAGED_BLOB_GC.equals(name)) {
+ abort = true;
+ } else {
+ used.add(name);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ if (abort) {
+ LOG.warn(
+ "Skip managed blob pack GC for table {} because sidecars,
manifests, or snapshot topology cannot be trusted, or the used pack set changed
during collection.",
+ table.fullName());
+ } else {
+ LOG.info(
+ "Frozen {} used managed blob pack names for table {}
before deletion.",
+ used.size(),
+ table.fullName());
+ }
+ return new FrozenUsedPacks(abort, used);
+ }
+
+ private DataStream<CleanOrphanFilesResult> deleteUnused(
+ StreamExecutionEnvironment env, Set<String> frozenUsed) {
+ configureEnv(env);
+ final OutputTag<Boolean> candidateSkipGcTag =
+ new OutputTag<Boolean>("candidate-managed-blob-skip") {};
+ SingleOutputStreamOperator<Tuple3<String, String, Long>> candidates =
+ env.fromCollection(Collections.singletonList(1),
TypeInformation.of(Integer.class))
+ .process(
+ new ProcessFunction<Integer, String>() {
+ @Override
+ public void processElement(
+ Integer i,
+ ProcessFunction<Integer,
String>.Context ctx,
+ Collector<String> out) {
+ FileStorePathFactory pathFactory =
+ table.store().pathFactory();
+ listPaimonFileDirs(
+ table.fullName(),
+
pathFactory.manifestPath().toString(),
+
pathFactory.indexPath().toString(),
+
pathFactory.statisticsPath().toString(),
+
pathFactory.dataFilePath().toString(),
+ partitionKeysNum,
+
table.coreOptions().dataFileExternalPaths())
+ .stream()
+ .map(Path::toUri)
+ .map(Object::toString)
+ .forEach(out::collect);
+ }
+ })
+ .name("list-dirs")
+ .forceNonParallel()
+ .process(
+ new ProcessFunction<String, Tuple3<String,
String, Long>>() {
+ @Override
+ public void processElement(
+ String dir,
+ ProcessFunction<String,
Tuple3<String, String, Long>>
+ .Context
+ ctx,
+ Collector<Tuple3<String, String,
Long>> out) {
+ for (FileStatus file :
tryBestListingDirs(new Path(dir))) {
+ if (!file.isDir()
+ && oldEnough(file)
+ && isManagedBlobPackName(
+
file.getPath().getName())) {
+ Optional<String> identity =
+
FlinkManagedBlobOrphanFilesClean.this
+
.packIdentityForCleanup(
+
file.getPath());
+ if (identity.isPresent()) {
+ out.collect(
+ Tuple3.of(
+
identity.get(),
+
file.getPath().toString(),
+
file.getLen()));
+ } else {
+ LOG.warn(
+ "Cannot safely
identify candidate managed blob pack {}. Skip pack GC this run.",
+ file.getPath());
+
ctx.output(candidateSkipGcTag, Boolean.TRUE);
+ }
+ }
+ }
+ }
+ })
+ .name("collect-candidate-packs");
+
+ DataStream<Tuple3<String, String, Long>> unused =
+ unusedCandidates(env, frozenUsed, candidates);
+
+ DataStream<Boolean> skipGc =
candidates.getSideOutput(candidateSkipGcTag);
+
+ final OutputTag<Path> emptyDirTag = new
OutputTag<Path>("empty-managed-blob-dir") {};
+ SingleOutputStreamOperator<CleanOrphanFilesResult> cleaned =
+ unused.keyBy(candidate -> candidate.f0)
+ .connect(skipGc.broadcast())
+ .transform(
+ "clean-unused-managed-blobs",
+
TypeInformation.of(CleanOrphanFilesResult.class),
+ new BoundedTwoInputOperator<
+ Tuple3<String, String, Long>,
+ Boolean,
+ CleanOrphanFilesResult>() {
+
+ private boolean skipEnded;
+ private boolean skipGc;
+ private long emittedFilesCount;
+ private long emittedFilesLen;
+ private final Set<String>
completedCandidates = new HashSet<>();
+
+ @Override
+ public InputSelection nextSelection() {
+ return skipEnded
+ ? InputSelection.FIRST
+ : InputSelection.SECOND;
+ }
+
+ @Override
+ public void endInput(int inputId) {
+ switch (inputId) {
+ case 2:
+ checkState(!skipEnded, "Should
not skip ended.");
+ skipEnded = true;
+ LOG.info("Managed blob GC skip
flag: {}", skipGc);
+ break;
+ case 1:
+ checkState(skipEnded, "Should
skip ended.");
+ output.collect(
+ new StreamRecord<>(
+ new
CleanOrphanFilesResult(
+
emittedFilesCount,
+
emittedFilesLen)));
+ break;
+ }
+ }
+
+ @Override
+ public void processElement1(
+ StreamRecord<Tuple3<String,
String, Long>> element) {
+ checkState(skipEnded, "Should skip
ended.");
+ if (skipGc) {
+ return;
+ }
+ Tuple3<String, String, Long> candidate
= element.getValue();
+ if
(completedCandidates.contains(candidate.f0)) {
+ return;
+ }
+ Path path = new Path(candidate.f1);
+ if (cleanPack(path)) {
+
completedCandidates.add(candidate.f0);
+ emittedFilesCount++;
+ emittedFilesLen += candidate.f2;
+ Path parent = path.getParent();
+ if (parent != null
+ && parent.toString()
+
.contains(BUCKET_PATH_PREFIX)) {
+ output.collect(
+ emptyDirTag, new
StreamRecord<>(parent));
+ }
+ LOG.info("Cleaned managed blob
pack: {}", path);
+ }
+ }
+
+ @Override
+ public void
processElement2(StreamRecord<Boolean> element) {
+ skipGc = true;
+ }
+ });
+
+ cleaned.getSideOutput(emptyDirTag)
+ .transform(
+ "clean-empty-dirs",
+ STRING_TYPE_INFO,
+ new BoundedOneInputOperator<Path, String>() {
+
+ private final Set<Path> bucketDirs = new
HashSet<>();
+
+ @Override
+ public void processElement(StreamRecord<Path>
element) {
+ bucketDirs.add(element.getValue());
+ }
+
+ @Override
+ public void endInput() {
+ tryCleanDataDirectory(bucketDirs,
partitionKeysNum + 1);
+ }
+ })
+ .forceNonParallel()
+ .sinkTo(new DiscardingSink<>())
+ .name("end")
+ .setParallelism(1)
+ .setMaxParallelism(1);
+
+ return cleaned;
+ }
+
+ // Re-source frozen used identities so deletion does not reconnect the
mark DAG. Keep the source
+ // serial so each worker does not deserialize the full used set; the keyed
anti-join retains
+ // only its key partition.
+ private DataStream<Tuple3<String, String, Long>> unusedCandidates(
+ StreamExecutionEnvironment env,
+ Set<String> frozenUsed,
+ DataStream<Tuple3<String, String, Long>> candidates) {
+ if (frozenUsed.isEmpty()) {
+ return candidates;
+ }
+ DataStream<String> frozenNames =
+ env.fromCollection(new ArrayList<>(frozenUsed),
STRING_TYPE_INFO)
+ .name("frozen-used-managed-blob-packs")
+ .setParallelism(1);
+ SingleOutputStreamOperator<Tuple3<String, String, Long>> unused =
+ frozenNames
+ .keyBy(name -> name)
+ .connect(candidates.keyBy(candidate -> candidate.f0))
+ .transform(
+ "join-used-and-candidate-packs",
+ Types.TUPLE(Types.STRING, Types.STRING,
Types.LONG),
+ new UnusedCandidateJoinOperator());
+ if (parallelism != null) {
+ unused.setParallelism(parallelism);
+ }
+ return unused;
+ }
+
+ protected SingleOutputStreamOperator<String> collectUsedPacks(
+ DataStream<Tuple2<String, String>> manifestLists,
+ OutputTag<Boolean> skipGcTag,
+ String markName) {
+ SingleOutputStreamOperator<Tuple2<String, String>>
distinctManifestLists =
+ deduplicateNamedFiles(manifestLists, markName +
"-managed-blob-manifest-lists");
+
+ SingleOutputStreamOperator<Tuple2<String, String>> manifests =
+ distinctManifestLists
+ .process(
+ new ProcessFunction<
+ Tuple2<String, String>, Tuple2<String,
String>>() {
+
+ private transient Map<String,
ManifestList> readers;
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(OpenContext openContext) {
+ open(new Configuration());
+ }
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(Configuration parameters)
{
+ readers = new HashMap<>();
+ }
+
+ @Override
+ public void processElement(
+ Tuple2<String, String>
branchAndList,
+ ProcessFunction<
+
Tuple2<String, String>,
+
Tuple2<String, String>>
+ .Context
+ ctx,
+ Collector<Tuple2<String, String>>
out)
+ throws Exception {
+ String branch = branchAndList.f0;
+ ManifestList reader =
readers.get(branch);
+ if (reader == null) {
+ reader =
+
table.switchToBranch(branch)
+ .store()
+
.manifestListFactory()
+ .create();
+ readers.put(branch, reader);
+ }
+ final ManifestList manifestListReader
= reader;
+ List<ManifestFileMeta> listed =
+ retryReadingFiles(
+ () ->
+
manifestListReader
+
.readWithIOException(
+
branchAndList.f1),
+ null);
+ if (listed == null) {
+ LOG.warn(
+ "Manifest list {} is
missing while collecting used managed blob packs. Skip pack GC this run.",
+ branchAndList.f1);
+ out.collect(Tuple2.of(branch,
SKIP_MANAGED_BLOB_GC));
+ return;
+ }
+ for (ManifestFileMeta meta : listed) {
+ out.collect(Tuple2.of(branch,
meta.fileName()));
+ }
+ }
+ })
+ .name("read-" + markName +
"-managed-blob-manifest-lists");
+ if (parallelism != null) {
+ manifests.setParallelism(parallelism);
+ }
+
+ SingleOutputStreamOperator<Tuple2<String, String>> distinctManifests =
+ deduplicateNamedFiles(manifests, markName +
"-managed-blob-manifests");
+ final OutputTag<String> manifestSkipGcTag =
+ new OutputTag<String>(markName +
"-managed-blob-manifest-skip") {};
+ SingleOutputStreamOperator<SidecarWorkItem> sidecars =
+ distinctManifests
+ .process(
+ new ProcessFunction<Tuple2<String, String>,
SidecarWorkItem>() {
+
+ private transient Map<String,
ManifestFile> manifestFiles;
+ private transient Map<String,
DataFilePathFactories>
+ pathFactories;
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(OpenContext openContext) {
+ open(new Configuration());
+ }
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(Configuration parameters)
{
+ manifestFiles = new HashMap<>();
+ pathFactories = new HashMap<>();
+ }
+
+ @Override
+ public void processElement(
+ Tuple2<String, String>
branchAndManifest,
+ ProcessFunction<Tuple2<String,
String>, SidecarWorkItem>
+ .Context
+ ctx,
+ Collector<SidecarWorkItem> out)
+ throws Exception {
+ if
(SKIP_MANAGED_BLOB_GC.equals(branchAndManifest.f1)) {
+ ctx.output(manifestSkipGcTag,
SKIP_MANAGED_BLOB_GC);
+ return;
+ }
+
+ String branch = branchAndManifest.f0;
+ ManifestFile manifestFile =
manifestFiles.get(branch);
+ DataFilePathFactories factories =
pathFactories.get(branch);
+ if (manifestFile == null) {
+ FileStoreTable branchTable =
+
table.switchToBranch(branch);
+ manifestFile =
+ branchTable
+ .store()
+
.manifestFileFactory()
+ .create();
+ factories =
+ new DataFilePathFactories(
+
branchTable.store().pathFactory());
+ manifestFiles.put(branch,
manifestFile);
+ pathFactories.put(branch,
factories);
+ }
+
+ final ManifestFile manifestReader =
manifestFile;
+ List<ManifestEntry> entries =
+ retryReadingFiles(
+ () ->
+
manifestReader.readWithIOException(
+
branchAndManifest.f1),
+ null);
+ if (entries == null) {
+ LOG.warn(
+ "Manifest {} is missing
while collecting used managed blob packs. Skip pack GC this run.",
+ branchAndManifest.f1);
+ ctx.output(manifestSkipGcTag,
SKIP_MANAGED_BLOB_GC);
+ return;
+ }
+ for (ManifestEntry entry : entries) {
+ for (SidecarWorkItem workItem :
+
FlinkManagedBlobOrphanFilesClean.this
+
.createSidecarWorkItems(
+ entry,
+
factories.get(
+
entry.partition(),
+
entry.bucket()))) {
+ out.collect(workItem);
+ }
+ }
+ }
+ })
+ .name("collect-" + markName + "-managed-blob-sidecars")
+ .returns(TypeInformation.of(SidecarWorkItem.class));
+ if (parallelism != null) {
+ sidecars.setParallelism(parallelism);
+ }
+
+ SingleOutputStreamOperator<SidecarWorkItem> distinctSidecars =
+ sidecars.keyBy(SidecarWorkItem::dedupIdentity)
+ .transform(
+ "deduplicate-" + markName +
"-managed-blob-sidecars",
+ TypeInformation.of(SidecarWorkItem.class),
+ new BoundedOneInputOperator<SidecarWorkItem,
SidecarWorkItem>() {
+
+ private final Set<String> identities = new
HashSet<>();
+
+ @Override
+ public void processElement(
+ StreamRecord<SidecarWorkItem>
element) {
+ if
(identities.add(element.getValue().dedupIdentity())) {
+ output.collect(element);
+ }
+ }
+
+ @Override
+ public void endInput() {}
+ });
+ if (parallelism != null) {
+ distinctSidecars.setParallelism(parallelism);
+ }
+
+ SingleOutputStreamOperator<String> marked =
+ distinctSidecars
+ .process(
+ new ProcessFunction<SidecarWorkItem, String>()
{
+
+ private transient
ManagedBlobOrphanFilesClean.ReachabilityScan
+ scan;
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(OpenContext openContext) {
+ open(new Configuration());
+ }
+
+ // @Override is skipped for compatibility
between Flink versions
+ public void open(Configuration parameters)
{
+ scan =
+
FlinkManagedBlobOrphanFilesClean.this
+ .newReachabilityScan();
+ }
+
+ @Override
+ public void processElement(
+ SidecarWorkItem workItem,
+ ProcessFunction<SidecarWorkItem,
String>.Context ctx,
+ Collector<String> out) {
+
FlinkManagedBlobOrphanFilesClean.this.emitUsedPacks(
+ workItem, scan, out::collect);
+ }
+ })
+ .name(markName + "-managed-blob-mark")
+ .returns(STRING_TYPE_INFO);
+ if (parallelism != null) {
+ marked.setParallelism(parallelism);
+ }
+
+ DataStream<String> markedAndManifestSkips =
+ marked.union(sidecars.getSideOutput(manifestSkipGcTag));
+ return markedAndManifestSkips
+ .keyBy(identity -> identity)
+ .transform(
+ "deduplicate-" + markName + "-managed-blob-mark",
+ STRING_TYPE_INFO,
+ new BoundedOneInputOperator<String, String>() {
+
+ private final Set<String> identities = new
HashSet<>();
+
+ @Override
+ public void processElement(StreamRecord<String>
element) {
+ identities.add(element.getValue());
+ }
+
+ @Override
+ public void endInput() {
+ for (String identity : identities) {
+ if (SKIP_MANAGED_BLOB_GC.equals(identity))
{
+ output.collect(skipGcTag, new
StreamRecord<>(Boolean.TRUE));
+ } else {
+ output.collect(new
StreamRecord<>(identity));
+ }
+ }
+ }
+ });
+ }
+
+ private SingleOutputStreamOperator<Tuple2<String, String>>
deduplicateNamedFiles(
+ DataStream<Tuple2<String, String>> files, String operatorName) {
+ SingleOutputStreamOperator<Tuple2<String, String>> deduplicated =
+ files.keyBy(file -> namedFileIdentity(file.f0, file.f1))
+ .transform(
+ "deduplicate-" + operatorName,
+ Types.TUPLE(Types.STRING, Types.STRING),
+ new BoundedOneInputOperator<
+ Tuple2<String, String>, Tuple2<String,
String>>() {
+
+ private final Set<String> identities = new
HashSet<>();
+
+ @Override
+ public void processElement(
+ StreamRecord<Tuple2<String,
String>> element) {
+ Tuple2<String, String> file =
element.getValue();
+ if
(identities.add(namedFileIdentity(file.f0, file.f1))) {
+ output.collect(
+ new StreamRecord<>(
+ Tuple2.of(file.f0,
file.f1)));
+ }
+ }
+
+ @Override
+ public void endInput() {}
+ });
+ if (parallelism != null) {
+ deduplicated.setParallelism(parallelism);
+ }
+ return deduplicated;
+ }
+
+ private static String namedFileIdentity(String branch, String fileName) {
+ return branch + '\0' + fileName;
+ }
+
+ private void emitManifestLists(String branch, Consumer<Tuple2<String,
String>> manifestLists)
+ throws Exception {
+ for (Snapshot snapshot : safelyGetAllSnapshots(branch)) {
+ emitManifestList(branch, snapshot.changelogManifestList(),
manifestLists);
+ emitManifestList(branch, snapshot.deltaManifestList(),
manifestLists);
+ emitManifestList(branch, snapshot.baseManifestList(),
manifestLists);
+ }
+ }
+
+ private static void emitManifestList(
+ String branch,
+ @Nullable String manifestList,
+ Consumer<Tuple2<String, String>> manifestLists) {
+ if (manifestList != null) {
+ manifestLists.accept(Tuple2.of(branch, manifestList));
+ }
+ }
+
+ private DataStream<Boolean> markCompletion(DataStream<String> usedPacks,
String markName) {
+ return usedPacks.transform(
+ markName + "-managed-blob-mark-completion",
+ TypeInformation.of(Boolean.class),
+ new BoundedOneInputOperator<String, Boolean>() {
+ @Override
+ public void processElement(StreamRecord<String> element) {}
+
+ @Override
+ public void endInput() {
+ output.collect(new StreamRecord<>(Boolean.TRUE));
+ }
+ });
+ }
+
+ private DataStream<Boolean> compareUsedPacks(
+ DataStream<String> usedPacks, DataStream<String> usedPacks2) {
+ return usedPacks
+ .keyBy(identity -> identity)
+ .connect(usedPacks2.keyBy(identity -> identity))
+ .transform(
+ "compare-managed-blob-marks",
+ TypeInformation.of(Boolean.class),
+ new BoundedTwoInputOperator<String, String, Boolean>()
{
+
+ private boolean firstMarkEnded;
+ private boolean marksDiffer;
+ private final Set<String> firstMark = new
HashSet<>();
+
+ @Override
+ public InputSelection nextSelection() {
+ return firstMarkEnded
+ ? InputSelection.SECOND
+ : InputSelection.FIRST;
+ }
+
+ @Override
+ public void endInput(int inputId) {
+ switch (inputId) {
+ case 1:
+ checkState(
+ !firstMarkEnded,
+ "Should not have finished the
first mark.");
+ firstMarkEnded = true;
+ break;
+ case 2:
+ checkState(
+ firstMarkEnded,
+ "Should have finished the
first mark.");
+ if (marksDiffer ||
!firstMark.isEmpty()) {
+ LOG.warn(
+ "Skip managed blob pack GC
for table {} because the used pack set changed during used-pack collection.",
+ table.fullName());
+ output.collect(new
StreamRecord<>(Boolean.TRUE));
+ }
+ break;
+ }
+ }
+
+ @Override
+ public void processElement1(StreamRecord<String>
element) {
+ firstMark.add(element.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<String>
element) {
+ checkState(firstMarkEnded, "Should have
finished the first mark.");
+ if (!firstMark.remove(element.getValue())) {
+ marksDiffer = true;
+ }
+ }
+ });
+ }
+
+ protected boolean cleanPack(Path path) {
+ return cleanManagedBlobFileIdempotently(path);
+ }
+
+ public static CleanOrphanFilesResult executeDatabase(
+ StreamExecutionEnvironment env,
+ Catalog catalog,
+ long olderThanMillis,
+ boolean dryRun,
+ @Nullable Integer parallelism,
+ String databaseName,
+ @Nullable String tableName)
+ throws Catalog.DatabaseNotExistException,
Catalog.TableNotExistException {
+ validateParallelism(parallelism);
+ List<String> tableNames = Collections.singletonList(tableName);
+ if (tableName == null || "*".equals(tableName)) {
+ tableNames = catalog.listTables(databaseName);
+ }
+
+ // Freeze+delete must finish for one table before the next table's
operators are added to
+ // the same environment. Unioning table graphs would let a later
freeze executeAndCollect
+ // also run an earlier table's uncommitted deletion DAG.
+ long deletedFilesCount = 0;
+ long deletedFilesLenInBytes = 0;
+ for (String t : tableNames) {
+ Identifier identifier = new Identifier(databaseName, t);
+ Table table = catalog.getTable(identifier);
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports remove-orphan-blobs action.
The table type is '%s'.",
+ table.getClass().getName());
+ DataStream<CleanOrphanFilesResult> clean =
+ new FlinkManagedBlobOrphanFilesClean(
+ (FileStoreTable) table, olderThanMillis,
dryRun, parallelism)
+ .doClean(env);
+ if (clean != null) {
+ CleanOrphanFilesResult one = sum(clean);
+ deletedFilesCount += one.getDeletedFileCount();
+ deletedFilesLenInBytes += one.getDeletedFileTotalLenInBytes();
+ }
+ }
+ return new CleanOrphanFilesResult(deletedFilesCount,
deletedFilesLenInBytes);
+ }
+
+ private static CleanOrphanFilesResult
sum(DataStream<CleanOrphanFilesResult> deleted) {
+ long deletedFilesCount = 0;
+ long deletedFilesLenInBytes = 0;
+ if (deleted != null) {
+ try (CloseableIterator<CleanOrphanFilesResult> iterator =
+
deleted.global().executeAndCollect("ManagedBlobOrphanFilesClean")) {
+ while (iterator.hasNext()) {
+ CleanOrphanFilesResult cleanOrphanFilesResult =
iterator.next();
+ deletedFilesCount +=
cleanOrphanFilesResult.getDeletedFileCount();
+ deletedFilesLenInBytes +=
+
cleanOrphanFilesResult.getDeletedFileTotalLenInBytes();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return new CleanOrphanFilesResult(deletedFilesCount,
deletedFilesLenInBytes);
+ }
+
+ private static final class FrozenUsedPacks {
+ private final boolean abort;
+ private final Set<String> used;
+
+ private FrozenUsedPacks(boolean abort, Set<String> used) {
+ this.abort = abort;
+ this.used = used;
+ }
+ }
+
+ // Keyed bounded anti-join: build used identities from the
lineage-detached source, then emit
+ // candidates that are not in this task's key partition.
+ private static final class UnusedCandidateJoinOperator
+ extends BoundedTwoInputOperator<
+ String, Tuple3<String, String, Long>, Tuple3<String,
String, Long>> {
+
+ private boolean buildEnd;
+ private final Set<String> used = new HashSet<>();
+
+ @Override
+ public InputSelection nextSelection() {
+ return buildEnd ? InputSelection.SECOND : InputSelection.FIRST;
+ }
+
+ @Override
+ public void endInput(int inputId) {
+ switch (inputId) {
+ case 1:
+ checkState(!buildEnd, "Should not build ended.");
+ LOG.info("Finish build phase for frozen used managed blob
packs.");
+ buildEnd = true;
+ break;
+ case 2:
+ checkState(buildEnd, "Should build ended.");
+ LOG.info("Finish probe phase for managed blob
candidates.");
+ break;
+ }
+ }
+
+ @Override
+ public void processElement1(StreamRecord<String> element) {
+ used.add(element.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<Tuple3<String, String, Long>>
element) {
+ checkState(buildEnd, "Should build ended.");
+ Tuple3<String, String, Long> candidate = element.getValue();
+ if (!used.contains(candidate.f0)) {
+ output.collect(element);
+ }
+ }
+ }
+
+ public static void validateParallelism(@Nullable Integer parallelism) {
+ checkArgument(
+ parallelism == null || parallelism > 0,
+ "Parallelism must be greater than 0, but was %s.",
+ parallelism);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
index 3ce2bf82f8..1337016c88 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
@@ -33,6 +33,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.flink.api.common.BatchShuffleMode;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -92,6 +93,9 @@ public class FlinkOrphanFilesClean extends OrphanFilesClean {
flinkConf.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH);
flinkConf.set(ExecutionOptions.SORT_INPUTS, false);
flinkConf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false);
+ // The used/candidate join below fully consumes its first input before
reading the second.
+ // A pipelined shuffle can deadlock that build/probe order, so force
blocking exchanges.
+ flinkConf.set(ExecutionOptions.BATCH_SHUFFLE_MODE,
BatchShuffleMode.ALL_EXCHANGES_BLOCKING);
if (parallelism != null) {
flinkConf.set(CoreOptions.DEFAULT_PARALLELISM, parallelism);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
new file mode 100644
index 0000000000..09d0e16586
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanBlobsProcedure.java
@@ -0,0 +1,120 @@
+/*
+ * 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.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+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";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(
+ name = "older_than",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(name = "dry_run", type =
@DataTypeHint("BOOLEAN"), isOptional = true),
+ @ArgumentHint(name = "parallelism", type =
@DataTypeHint("INT"), isOptional = true),
+ @ArgumentHint(name = "mode", type = @DataTypeHint("STRING"),
isOptional = true)
+ })
+ 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 != null && dryRun,
+ parallelism,
+ databaseName,
+ tableName);
+ break;
+ case "LOCAL":
+ result =
+ LocalManagedBlobOrphanFilesClean.executeDatabase(
+ catalog,
+ databaseName,
+ tableName,
+ olderThanMillis(olderThan),
+ parallelism,
+ dryRun != null && 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;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 188b50550e..288431c31a 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -33,6 +33,7 @@ org.apache.paimon.flink.action.ResetConsumerActionFactory
org.apache.paimon.flink.action.MigrateTableActionFactory
org.apache.paimon.flink.action.MigrateDatabaseActionFactory
org.apache.paimon.flink.action.RemoveOrphanFilesActionFactory
+org.apache.paimon.flink.action.RemoveOrphanBlobsActionFactory
org.apache.paimon.flink.action.QueryServiceActionFactory
org.apache.paimon.flink.action.ExpirePartitionsActionFactory
org.apache.paimon.flink.action.MarkPartitionDoneActionFactory
@@ -78,6 +79,7 @@ org.apache.paimon.flink.procedure.RollbackToWatermarkProcedure
org.apache.paimon.flink.procedure.MigrateTableProcedure
org.apache.paimon.flink.procedure.MigrateDatabaseProcedure
org.apache.paimon.flink.procedure.RemoveOrphanFilesProcedure
+org.apache.paimon.flink.procedure.RemoveOrphanBlobsProcedure
org.apache.paimon.flink.procedure.QueryServiceProcedure
org.apache.paimon.flink.procedure.ExpireSnapshotsProcedure
org.apache.paimon.flink.procedure.ExpireChangelogsProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
index 4e447e3972..f5c77c54e0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionJobCoverageTest.java
@@ -120,7 +120,10 @@ public class ActionJobCoverageTest {
"<init>"),
Tuple2.of(
"org/apache/paimon/flink/orphan/FlinkOrphanFilesClean",
- "executeDatabaseOrphanFiles"));
+ "executeDatabaseOrphanFiles"),
+ Tuple2.of(
+
"org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean",
+ "executeDatabase"));
private static final List<Tuple2<String, String>>
VALID_OWNER_PATTERN_AND_NAMES =
Collections.singletonList(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java
new file mode 100644
index 0000000000..196ffea58e
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCase.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/** IT cases for {@link RemoveOrphanBlobsAction} in Flink Common. */
+public class RemoveOrphanBlobsActionITCase extends
RemoveOrphanBlobsActionITCaseBase {}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java
new file mode 100644
index 0000000000..7c54c31070
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanBlobsActionITCaseBase.java
@@ -0,0 +1,935 @@
+/*
+ * 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 org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.blob.ManagedBlobReferenceFile;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.flink.orphan.FlinkManagedBlobOrphanFilesClean;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
+import org.apache.paimon.operation.ManagedBlobOrphanFilesClean;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.DataFilePathFactories;
+import org.apache.paimon.utils.DateTimeUtils;
+import org.apache.paimon.utils.TraceableFileIO;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+
+import org.apache.flink.api.common.BatchShuffleMode;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.OutputTag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
+import static
org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for {@link RemoveOrphanBlobsAction}. */
+public abstract class RemoveOrphanBlobsActionITCaseBase extends
ActionITCaseBase {
+
+ private static final AtomicBoolean FAIL_AFTER_FIRST_DELETE = new
AtomicBoolean();
+ private static final List<CleanOrphanFilesResult> FAILOVER_CLEANUP_RESULTS
=
+ Collections.synchronizedList(new
ArrayList<CleanOrphanFilesResult>());
+ private static final AtomicInteger MARK_PASS = new AtomicInteger();
+ private static final List<String> FROZEN_USED_NAMES =
+ Collections.synchronizedList(new ArrayList<String>());
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testDeleteUnreferencedManagedBlobPack(String mode) throws
Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ List<Path> referenced =
+ filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ referenced.removeIf(p -> orphan.getName().equals(p.getName()));
+ assertThat(referenced).isNotEmpty();
+
+ ImmutableList.copyOf(executeSQL(removeOrphanBlobsCall(mode)));
+
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testMissingManagedBlobSidecarSkipsPackGc(String mode) throws
Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphanPack = new Path(bucketPath(table), "orphan.managed.blob");
+ Path orphanOther = new Path(bucketPath(table), "orphan.txt");
+ table.fileIO().newOutputStream(orphanPack, false).close();
+ table.fileIO().writeFile(orphanOther, "x", true);
+ Thread.sleep(2000);
+
+ List<Path> referenced =
+ filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ referenced.removeIf(p -> orphanPack.getName().equals(p.getName()));
+ assertThat(referenced).isNotEmpty();
+ deleteFilesWithSuffix(table,
ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX);
+
+ ImmutableList.copyOf(executeSQL(removeOrphanBlobsCall(mode)));
+
+ assertThat(table.fileIO().exists(orphanPack)).isTrue();
+ assertThat(table.fileIO().exists(orphanOther)).isTrue();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @Test
+ public void testDistributedDeleteWithEmptyUsedPackSet() throws Exception {
+ FileStoreTable table = createManagedBlobTable();
+ Path bucket = bucketPath(table);
+ table.fileIO().mkdirs(bucket);
+ Path orphan = new Path(bucket, "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ ImmutableList.copyOf(executeSQL(removeOrphanBlobsCall("distributed")));
+
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ }
+
+ @Test
+ public void testDistributedUnresolvableRelativeCandidateSkipsGc() throws
Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ UnresolvableRelativeListingFileIO.reset();
+ FileStoreTable relativeListingTable =
+ FileStoreTableFactory.create(
+ new UnresolvableRelativeListingFileIO(),
table.location(), table.schema());
+ StreamExecutionEnvironment env =
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(5).build();
+ FlinkManagedBlobOrphanFilesClean cleaner =
+ new FlinkManagedBlobOrphanFilesClean(
+ relativeListingTable, Long.MAX_VALUE, false, 5);
+ List<CleanOrphanFilesResult> cleanResults = new ArrayList<>();
+ try (CloseableIterator<CleanOrphanFilesResult> results =
+ cleaner.doClean(env).executeAndCollect()) {
+ while (results.hasNext()) {
+ cleanResults.add(results.next());
+ }
+ }
+
+
assertThat(UnresolvableRelativeListingFileIO.resolveAttempts()).isPositive();
+ assertThat(cleanResults).isNotEmpty();
+ assertThat(cleanResults)
+ .allSatisfy(
+ result -> {
+ assertThat(result.getDeletedFileCount()).isZero();
+
assertThat(result.getDeletedFileTotalLenInBytes()).isZero();
+ });
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ }
+
+ @Test
+ public void testDistributedMarkReadsSharedMetadataOncePerPass() throws
Exception {
+ FileStoreTable table = createManagedBlobTable();
+ StreamWriteBuilder writeBuilder =
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+ writeData(rowData(1, BinaryString.fromString("a"), new BlobData(new
byte[] {1, 2})));
+ writeData(rowData(2, BinaryString.fromString("b"), new BlobData(new
byte[] {3, 4})));
+ write.close();
+ commit.close();
+ write = null;
+ commit = null;
+ try (TableCommitImpl manifestCommit =
table.newCommit("manifest-compact-test")) {
+ manifestCommit.compactManifests();
+ }
+ assertSharedManifestMetadata(table);
+ int parallelism = assertSharedSidecarAcrossManifestPartitions(table);
+
+ CountingManifestFileIO.reset();
+ FileStoreTable countingTable =
+ FileStoreTableFactory.create(
+ new CountingManifestFileIO(), table.location(),
table.schema());
+ StreamExecutionEnvironment env =
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(parallelism).build();
+ FlinkManagedBlobOrphanFilesClean cleaner =
+ new FlinkManagedBlobOrphanFilesClean(
+ countingTable, Long.MAX_VALUE, true, parallelism);
+ try (CloseableIterator<CleanOrphanFilesResult> results =
+ cleaner.doClean(env).executeAndCollect()) {
+ while (results.hasNext()) {
+ results.next();
+ }
+ }
+
+ assertThat(CountingManifestFileIO.readCounts()).isNotEmpty();
+ assertThat(CountingManifestFileIO.readCounts().keySet())
+ .anyMatch(path ->
path.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX));
+
assertThat(CountingManifestFileIO.readCounts().values()).allMatch(count ->
count == 2);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testDistributedCleanupDoesNotHangWithPipelinedShuffle() throws
Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path bucket = bucketPath(table);
+ List<Path> referenced =
+ filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ assertThat(referenced).isNotEmpty();
+ for (int i = 0; i < 256; i++) {
+ table.fileIO()
+ .newOutputStream(new Path(bucket, "orphan-" + i +
".managed.blob"), false)
+ .close();
+ }
+ Thread.sleep(2000);
+
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .batchMode()
+ .parallelism(2)
+ .setConf(
+ ExecutionOptions.BATCH_SHUFFLE_MODE,
+ BatchShuffleMode.ALL_EXCHANGES_PIPELINED)
+ .build();
+ FlinkManagedBlobOrphanFilesClean cleaner =
+ new FlinkManagedBlobOrphanFilesClean(table, Long.MAX_VALUE,
false, 2);
+ DataStream<CleanOrphanFilesResult> clean = cleaner.doClean(env);
+
assertThat(env.getConfiguration().get(ExecutionOptions.BATCH_SHUFFLE_MODE))
+ .isEqualTo(BatchShuffleMode.ALL_EXCHANGES_BLOCKING);
+ long deleted = 0;
+ try (CloseableIterator<CleanOrphanFilesResult> results =
clean.executeAndCollect()) {
+ while (results.hasNext()) {
+ deleted += results.next().getDeletedFileCount();
+ }
+ }
+ assertThat(deleted).isEqualTo(256);
+ assertThat(filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX))
+ .containsExactlyInAnyOrderElementsOf(referenced);
+ }
+
+ @Test
+ public void
testDistributedDeletionAccountingSurvivesCleanupTaskRestartAfterDelete()
+ throws Exception {
+ FileStoreTable table = createManagedBlobTable();
+ Path bucket = bucketPath(table);
+ table.fileIO().mkdirs(bucket);
+ Path first = new Path(bucket, "first.managed.blob");
+ Path second = new Path(bucket, "second.managed.blob");
+ String firstContent = "one";
+ String secondContent = "second";
+ table.fileIO().writeFile(first, firstContent, false);
+ table.fileIO().writeFile(second, secondContent, false);
+ long expectedDeletedBytes = firstContent.length() +
secondContent.length();
+ Thread.sleep(2000);
+
+ FAIL_AFTER_FIRST_DELETE.set(true);
+ FAILOVER_CLEANUP_RESULTS.clear();
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .batchMode()
+ .parallelism(1)
+ .allowRestart()
+ .build();
+ FlinkManagedBlobOrphanFilesClean cleaner = new
FailAfterFirstDeleteCleaner(table);
+
+ // executeAndCollect() uses an uncheckpointed collect buffer that
throws "Job restarted"
+ // on failover, so collect operator output through a MiniCluster-local
sink instead.
+ cleaner.doClean(env).map(new CollectCleanupResult()).sinkTo(new
DiscardingSink<>());
+ env.execute();
+
+ long deleted = 0;
+ long deletedBytes = 0;
+ for (CleanOrphanFilesResult result : FAILOVER_CLEANUP_RESULTS) {
+ deleted += result.getDeletedFileCount();
+ deletedBytes += result.getDeletedFileTotalLenInBytes();
+ }
+
+ assertThat(FAIL_AFTER_FIRST_DELETE).isFalse();
+ assertThat(deleted).isEqualTo(2);
+ assertThat(deletedBytes).isEqualTo(expectedDeletedBytes);
+ assertThat(table.fileIO().exists(first)).isFalse();
+ assertThat(table.fileIO().exists(second)).isFalse();
+ }
+
+ @Test
+ public void testDistributedCanonicalAliasCandidatesAreCountedOnce() throws
Exception {
+ FileStoreTable table = createManagedBlobTable();
+ Path bucket = bucketPath(table);
+ table.fileIO().mkdirs(bucket);
+ Path orphan = new Path(bucket, "duplicate.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ FileStoreTable duplicateListingTable =
+ FileStoreTableFactory.create(
+ new CanonicalAliasListingFileIO(), table.location(),
table.schema());
+ StreamExecutionEnvironment env =
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(2).build();
+ FlinkManagedBlobOrphanFilesClean cleaner =
+ new FlinkManagedBlobOrphanFilesClean(
+ duplicateListingTable, Long.MAX_VALUE, false, 2);
+
+ long deleted = 0;
+ try (CloseableIterator<CleanOrphanFilesResult> results =
+ cleaner.doClean(env).executeAndCollect()) {
+ while (results.hasNext()) {
+ deleted += results.next().getDeletedFileCount();
+ }
+ }
+
+ assertThat(deleted).isEqualTo(1);
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testActionFactoryDryRunParsing(boolean dryRun) throws
Exception {
+ FileStoreTable table = createManagedBlobTable();
+ Path bucket = bucketPath(table);
+ table.fileIO().mkdirs(bucket);
+ Path orphan = new Path(bucket, "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ createAction(
+ RemoveOrphanBlobsAction.class,
+ "remove_orphan_blobs",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--older_than",
+ currentTimestamp(),
+ "--dry_run",
+ String.valueOf(dryRun),
+ "--parallelism",
+ "2")
+ .run();
+
+ assertThat(table.fileIO().exists(orphan)).isEqualTo(dryRun);
+ }
+
+ @Test
+ public void testActionFactoryRejectsInvalidDryRun() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ RemoveOrphanBlobsAction.class,
+ "remove_orphan_blobs",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--dry_run",
+ "ture"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Argument 'dry_run' must be either 'true' or 'false',
but was 'ture'.");
+ }
+
+ @Test
+ public void testActionFactoryRejectsNonPositiveParallelism() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ RemoveOrphanBlobsAction.class,
+ "remove_orphan_blobs",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--parallelism",
+ "0"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Parallelism must be greater than 0, but
was 0.");
+ }
+
+ @Test
+ public void testProcedureRejectsNonPositiveParallelism() {
+ String call =
+ supportNamedArgument()
+ ? String.format(
+ "CALL sys.remove_orphan_blobs(`table` =>
'%s.%s', parallelism => 0)",
+ database, tableName)
+ : String.format(
+ "CALL sys.remove_orphan_blobs('%s.%s', '',
false, 0)",
+ database, tableName);
+
+ assertThatThrownBy(() -> ImmutableList.copyOf(executeSQL(call)))
+ .satisfies(
+ anyCauseMatches(
+ IllegalArgumentException.class,
+ "Parallelism must be greater than 0, but was
0."));
+ }
+
+ @Test
+ public void testFrozenUsedMarkIsNotRecomputedDuringDeletion() throws
Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ List<Path> referenced =
+ filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ assertThat(referenced).isNotEmpty();
+ Path orphan = new Path(bucketPath(table), "orphan.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+
+ MARK_PASS.set(0);
+ FROZEN_USED_NAMES.clear();
+ for (Path pack : referenced) {
+
FROZEN_USED_NAMES.add(ManagedBlobOrphanFilesClean.packIdentity(pack));
+ }
+ StreamExecutionEnvironment env =
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(2).build();
+ FlinkManagedBlobOrphanFilesClean cleaner = new
FreezeAfterTwoMarksCleaner(table);
+
+ long deleted = 0;
+ try (CloseableIterator<CleanOrphanFilesResult> results =
+ cleaner.doClean(env).executeAndCollect()) {
+ while (results.hasNext()) {
+ deleted += results.next().getDeletedFileCount();
+ }
+ }
+
+ assertThat(MARK_PASS.get()).isEqualTo(2);
+ assertThat(deleted).isEqualTo(1);
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @Test
+ public void testUsedPackSetChangeSkipsGc() throws Exception {
+ FileStoreTable table = createManagedBlobTableAndWrite();
+ Path orphan = new Path(bucketPath(table),
"used-set-changed.managed.blob");
+ table.fileIO().newOutputStream(orphan, false).close();
+ Thread.sleep(2000);
+ List<Path> referenced =
+ filesWithSuffix(table,
ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX);
+ referenced.removeIf(path -> orphan.getName().equals(path.getName()));
+ assertThat(referenced).isNotEmpty();
+
+ MARK_PASS.set(0);
+ StreamExecutionEnvironment env =
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(2).build();
+ FlinkManagedBlobOrphanFilesClean cleaner = new
ChangingUsedMarkCleaner(table);
+
+ long deleted = 0;
+ try (CloseableIterator<CleanOrphanFilesResult> results =
+ cleaner.doClean(env).executeAndCollect()) {
+ while (results.hasNext()) {
+ deleted += results.next().getDeletedFileCount();
+ }
+ }
+
+ assertThat(deleted).isZero();
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ for (Path pack : referenced) {
+ assertThat(table.fileIO().exists(pack)).isTrue();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testRemoveDatabaseOrphanBlobs(String mode) throws Exception {
+ FileStoreTable table1 = createManagedBlobTableAndWrite("T1");
+ FileStoreTable table2 = createManagedBlobTableAndWrite("T2");
+ try {
+ Path orphan1 = new Path(bucketPath(table1), "orphan.managed.blob");
+ Path orphan2 = new Path(bucketPath(table2), "orphan.managed.blob");
+ table1.fileIO().writeFile(orphan1, "abc", false);
+ table2.fileIO().writeFile(orphan2, "abc", false);
+ Thread.sleep(2000);
+
+ String olderThan = currentTimestamp();
+ String call =
+ supportNamedArgument()
+ ? String.format(
+ "CALL sys.remove_orphan_blobs(`table` =>
'%s.*', older_than => '%s', mode => '%s')",
+ database, olderThan, mode)
+ : String.format(
+ "CALL sys.remove_orphan_blobs('%s.*',
'%s', false, 5, '%s')",
+ database, olderThan, mode);
+ assertThat(ImmutableList.copyOf(executeSQL(call)))
+ .containsExactly(Row.of("2"), Row.of("6"));
+ assertThat(table1.fileIO().exists(orphan1)).isFalse();
+ assertThat(table2.fileIO().exists(orphan2)).isFalse();
+ } finally {
+ catalog.dropTable(Identifier.create(database, "T1"), true);
+ catalog.dropTable(Identifier.create(database, "T2"), true);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"local", "distributed"})
+ public void testProcedureDryRun(String mode) throws Exception {
+ FileStoreTable table = createManagedBlobTable();
+ Path bucket = bucketPath(table);
+ table.fileIO().mkdirs(bucket);
+ Path orphan = new Path(bucket, "orphan.managed.blob");
+ table.fileIO().writeFile(orphan, "abc", false);
+ Thread.sleep(2000);
+
+ String olderThan = currentTimestamp();
+ String call =
+ supportNamedArgument()
+ ? String.format(
+ "CALL sys.remove_orphan_blobs(`table` =>
'%s.%s', older_than => '%s', dry_run => true, mode => '%s')",
+ database, tableName, olderThan, mode)
+ : String.format(
+ "CALL sys.remove_orphan_blobs('%s.%s', '%s',
true, 5, '%s')",
+ database, tableName, olderThan, mode);
+ assertThat(ImmutableList.copyOf(executeSQL(call)))
+ .containsExactly(Row.of("1"), Row.of("3"));
+ assertThat(table.fileIO().exists(orphan)).isTrue();
+ }
+
+ private FileStoreTable createManagedBlobTableAndWrite() throws Exception {
+ return createManagedBlobTableAndWrite(tableName);
+ }
+
+ private FileStoreTable createManagedBlobTableAndWrite(String name) throws
Exception {
+ FileStoreTable table = createManagedBlobTable(name);
+ StreamWriteBuilder writeBuilder =
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+ writeData(rowData(1, BinaryString.fromString("a"), new BlobData(new
byte[] {1, 2})));
+ write.close();
+ commit.close();
+ write = null;
+ commit = null;
+ return table;
+ }
+
+ private FileStoreTable createManagedBlobTable() throws Exception {
+ return createManagedBlobTable(tableName);
+ }
+
+ private FileStoreTable createManagedBlobTable(String name) throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "none");
+ options.put("bucket", "1");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING(),
DataTypes.BLOB()},
+ new String[] {"id", "name", "payload"});
+ return createFileStoreTable(
+ name,
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyList(),
+ options);
+ }
+
+ private static SingleOutputStreamOperator<String> emitUsedNames(
+ StreamExecutionEnvironment env, List<String> names) {
+ List<String> source = names.isEmpty() ? Collections.singletonList("")
: names;
+ return env.fromCollection(source, TypeInformation.of(String.class))
+ .process(
+ new ProcessFunction<String, String>() {
+ @Override
+ public void processElement(
+ String name, Context ctx,
Collector<String> out) {
+ if (!names.isEmpty()) {
+ out.collect(name);
+ }
+ }
+ });
+ }
+
+ private String removeOrphanBlobsCall(String mode) {
+ String olderThan = currentTimestamp();
+ if (supportNamedArgument()) {
+ return String.format(
+ "CALL sys.remove_orphan_blobs(`table` => '%s.%s',
older_than => '%s', parallelism => 5, mode => '%s')",
+ database, tableName, olderThan, mode);
+ }
+ return String.format(
+ "CALL sys.remove_orphan_blobs('%s.%s', '%s', false, 5, '%s')",
+ database, tableName, olderThan, mode);
+ }
+
+ private static String currentTimestamp() {
+ return DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), 3);
+ }
+
+ private static Path bucketPath(FileStoreTable table) {
+ return table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0);
+ }
+
+ private static List<Path> filesWithSuffix(FileStoreTable table, String
suffix)
+ throws IOException {
+ List<Path> result = new ArrayList<>();
+ FileStatus[] statuses = table.fileIO().listStatus(bucketPath(table));
+ if (statuses == null) {
+ return result;
+ }
+ for (FileStatus status : statuses) {
+ if (status.getPath().getName().endsWith(suffix)) {
+ result.add(status.getPath());
+ }
+ }
+ return result;
+ }
+
+ private static void deleteFilesWithSuffix(FileStoreTable table, String
suffix)
+ throws IOException {
+ for (Path path : filesWithSuffix(table, suffix)) {
+ table.fileIO().deleteQuietly(path);
+ }
+ }
+
+ private static void assertSharedManifestMetadata(FileStoreTable table)
throws IOException {
+ ManifestList manifestList =
table.store().manifestListFactory().create();
+ Map<String, Integer> references = new HashMap<>();
+ Set<String> readLists = new HashSet<>();
+ Iterator<Snapshot> snapshots = table.snapshotManager().snapshots();
+ while (snapshots.hasNext()) {
+ Snapshot snapshot = snapshots.next();
+ String[] listNames = {
+ snapshot.changelogManifestList(),
+ snapshot.deltaManifestList(),
+ snapshot.baseManifestList()
+ };
+ for (String listName : listNames) {
+ if (listName == null) {
+ continue;
+ }
+ references.merge("list:" + listName, 1, Integer::sum);
+ if (readLists.add(listName)) {
+ for (ManifestFileMeta meta :
manifestList.readWithIOException(listName)) {
+ references.merge("manifest:" + meta.fileName(), 1,
Integer::sum);
+ }
+ }
+ }
+ }
+ assertThat(references.values()).anyMatch(count -> count > 1);
+ }
+
+ private static int
assertSharedSidecarAcrossManifestPartitions(FileStoreTable table)
+ throws IOException {
+ ManifestList manifestList =
table.store().manifestListFactory().create();
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ DataFilePathFactories pathFactories =
+ new DataFilePathFactories(table.store().pathFactory());
+ Set<String> readLists = new HashSet<>();
+ Set<String> readManifests = new HashSet<>();
+ Map<String, Set<String>> sidecarManifests = new HashMap<>();
+ Iterator<Snapshot> snapshots = table.snapshotManager().snapshots();
+ while (snapshots.hasNext()) {
+ Snapshot snapshot = snapshots.next();
+ String[] listNames = {
+ snapshot.changelogManifestList(),
+ snapshot.deltaManifestList(),
+ snapshot.baseManifestList()
+ };
+ for (String listName : listNames) {
+ if (listName == null || !readLists.add(listName)) {
+ continue;
+ }
+ for (ManifestFileMeta meta :
manifestList.readWithIOException(listName)) {
+ if (!readManifests.add(meta.fileName())) {
+ continue;
+ }
+ for (ManifestEntry entry :
manifestFile.readWithIOException(meta.fileName())) {
+ if (entry.kind() != FileKind.ADD ||
entry.file().extraFiles() == null) {
+ continue;
+ }
+ Path dataFile =
+ pathFactories.get(entry.partition(),
entry.bucket()).toPath(entry);
+ for (String extraFile : entry.file().extraFiles()) {
+ if (extraFile != null
+ && extraFile.endsWith(
+
ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ String sidecar =
+ new Path(dataFile.getParent(),
extraFile)
+ .toUri()
+ .normalize()
+ .toString();
+ sidecarManifests
+ .computeIfAbsent(sidecar, ignored ->
new HashSet<>())
+ .add(meta.fileName());
+ }
+ }
+ }
+ }
+ }
+ }
+
+ assertThat(sidecarManifests.values()).anyMatch(manifests ->
manifests.size() > 1);
+ for (int parallelism = 2; parallelism <= 16; parallelism++) {
+ int maxParallelism =
KeyGroupRangeAssignment.computeDefaultMaxParallelism(parallelism);
+ for (Set<String> manifests : sidecarManifests.values()) {
+ Set<Integer> partitions = new HashSet<>();
+ for (String manifest : manifests) {
+ partitions.add(
+
KeyGroupRangeAssignment.assignKeyToParallelOperator(
+ DEFAULT_MAIN_BRANCH + '\0' + manifest,
+ maxParallelism,
+ parallelism));
+ }
+ if (partitions.size() > 1) {
+ return parallelism;
+ }
+ }
+ }
+ throw new AssertionError(
+ "Shared sidecars did not span manifest-reader partitions for
parallelism 2-16.");
+ }
+
+ private static class CountingManifestFileIO extends LocalFileIO {
+
+ private static final Map<String, AtomicInteger> READ_COUNTS = new
ConcurrentHashMap<>();
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ String name = path.getName();
+ if (name.startsWith("manifest-list-")
+ || (name.startsWith("manifest-") &&
!name.startsWith("manifest-list-"))
+ ||
name.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) {
+ READ_COUNTS
+ .computeIfAbsent(path.toUri().getPath(), ignored ->
new AtomicInteger())
+ .incrementAndGet();
+ }
+ return super.newInputStream(path);
+ }
+
+ private static Map<String, Integer> readCounts() {
+ Map<String, Integer> result = new HashMap<>();
+ READ_COUNTS.forEach((path, count) -> result.put(path,
count.get()));
+ return result;
+ }
+
+ private static void reset() {
+ READ_COUNTS.clear();
+ }
+ }
+
+ private static class UnresolvableRelativeListingFileIO extends
TraceableFileIO {
+
+ private static final AtomicInteger RESOLVE_ATTEMPTS = new
AtomicInteger();
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ FileStatus[] statuses = super.listStatus(path);
+ if (statuses == null) {
+ return null;
+ }
+ FileStatus[] relative = new FileStatus[statuses.length];
+ for (int i = 0; i < statuses.length; i++) {
+ FileStatus status = statuses[i];
+ relative[i] =
+ status.getPath()
+ .getName()
+
.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)
+ ? withPath(status, new
Path(status.getPath().getName()))
+ : status;
+ }
+ return relative;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ if (!path.toUri().getPath().startsWith("/")) {
+ RESOLVE_ATTEMPTS.incrementAndGet();
+ throw new IOException("Cannot resolve relative path " + path);
+ }
+ return super.getFileStatus(path);
+ }
+
+ private static int resolveAttempts() {
+ return RESOLVE_ATTEMPTS.get();
+ }
+
+ private static void reset() {
+ RESOLVE_ATTEMPTS.set(0);
+ }
+
+ private static FileStatus withPath(FileStatus status, Path path) {
+ return new FileStatus() {
+ @Override
+ public long getLen() {
+ return status.getLen();
+ }
+
+ @Override
+ public boolean isDir() {
+ return status.isDir();
+ }
+
+ @Override
+ public Path getPath() {
+ return path;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return status.getModificationTime();
+ }
+ };
+ }
+ }
+
+ private static class CanonicalAliasListingFileIO extends TraceableFileIO {
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ FileStatus[] statuses = super.listStatus(path);
+ if (statuses == null) {
+ return null;
+ }
+ List<FileStatus> duplicated = new ArrayList<>();
+ for (FileStatus status : statuses) {
+ duplicated.add(status);
+ if (status.getPath()
+ .getName()
+
.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) {
+ Path alias =
+ new Path(
+ "hdfs://duplicate-listing"
+ +
status.getPath().toUri().getPath());
+
duplicated.add(UnresolvableRelativeListingFileIO.withPath(status, alias));
+ }
+ }
+ return duplicated.toArray(new FileStatus[0]);
+ }
+ }
+
+ private static class FreezeAfterTwoMarksCleaner extends
FlinkManagedBlobOrphanFilesClean {
+
+ private FreezeAfterTwoMarksCleaner(FileStoreTable table) {
+ super(table, Long.MAX_VALUE, false, 2);
+ }
+
+ @Override
+ protected SingleOutputStreamOperator<String> collectUsedPacks(
+ DataStream<Tuple2<String, String>> manifestLists,
+ OutputTag<Boolean> skipGcTag,
+ String markName) {
+ int pass = MARK_PASS.incrementAndGet();
+ List<String> names =
+ pass <= 2 ? new ArrayList<String>(FROZEN_USED_NAMES) :
Collections.emptyList();
+ return emitUsedNames(manifestLists.getExecutionEnvironment(),
names);
+ }
+ }
+
+ private static class ChangingUsedMarkCleaner extends
FlinkManagedBlobOrphanFilesClean {
+
+ private ChangingUsedMarkCleaner(FileStoreTable table) {
+ super(table, Long.MAX_VALUE, false, 2);
+ }
+
+ @Override
+ protected SingleOutputStreamOperator<String> collectUsedPacks(
+ DataStream<Tuple2<String, String>> manifestLists,
+ OutputTag<Boolean> skipGcTag,
+ String markName) {
+ String name = MARK_PASS.incrementAndGet() == 1 ? "first-pack" :
"second-pack";
+ return emitUsedNames(
+ manifestLists.getExecutionEnvironment(),
Collections.singletonList(name));
+ }
+ }
+
+ private static class FailAfterFirstDeleteCleaner extends
FlinkManagedBlobOrphanFilesClean {
+
+ private FailAfterFirstDeleteCleaner(FileStoreTable table) {
+ super(table, Long.MAX_VALUE, false, 1);
+ }
+
+ @Override
+ protected boolean cleanPack(Path path) {
+ boolean cleaned = super.cleanPack(path);
+ if (cleaned && FAIL_AFTER_FIRST_DELETE.compareAndSet(true, false))
{
+ throw new RuntimeException("Injected failure after delete.");
+ }
+ return cleaned;
+ }
+ }
+
+ private static class CollectCleanupResult
+ implements MapFunction<CleanOrphanFilesResult,
CleanOrphanFilesResult> {
+
+ @Override
+ public CleanOrphanFilesResult map(CleanOrphanFilesResult value) {
+ FAILOVER_CLEANUP_RESULTS.add(value);
+ return value;
+ }
+ }
+
+ protected boolean supportNamedArgument() {
+ return true;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCaseBase.java
index d65a2713a0..eafa3383d8 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCaseBase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCaseBase.java
@@ -21,9 +21,11 @@ package org.apache.paimon.flink.action;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.orphan.FlinkOrphanFilesClean;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.operation.CleanOrphanFilesResult;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.SchemaChange;
@@ -41,6 +43,10 @@ import org.apache.paimon.utils.DateTimeUtils;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+import org.apache.flink.api.common.BatchShuffleMode;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
@@ -512,6 +518,24 @@ public abstract class RemoveOrphanFilesActionITCaseBase
extends ActionITCaseBase
assertThat(fileIO.exists(new Path(nonEmptyPath,
"guard.txt"))).isTrue();
}
+ @Test
+ public void testDistributedCleanupForcesBlockingShuffle() throws Exception
{
+ FileStoreTable table = createTableAndWriteData(tableName);
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .batchMode()
+ .parallelism(2)
+ .setConf(
+ ExecutionOptions.BATCH_SHUFFLE_MODE,
+ BatchShuffleMode.ALL_EXCHANGES_PIPELINED)
+ .build();
+ DataStream<CleanOrphanFilesResult> clean =
+ new FlinkOrphanFilesClean(table, Long.MAX_VALUE, false,
2).doOrphanClean(env);
+
assertThat(env.getConfiguration().get(ExecutionOptions.BATCH_SHUFFLE_MODE))
+ .isEqualTo(BatchShuffleMode.ALL_EXCHANGES_BLOCKING);
+ assertThat(clean).isNotNull();
+ }
+
protected boolean supportNamedArgument() {
return true;
}
