[core] add parallelism arguement for remove orphan files (#4044)

Author: LsomeYeah

docs/content/flink/procedures.md

@@ -214,7 +214,8 @@ All available procedures are listed below.
       <td>
          CALL [catalog.]sys.remove_orphan_files('identifier')<br/><br/>
          CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan')<br/><br/>
-         CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun')
+         CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun')<br/><br/>
+         CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun','parallelism')
       </td>
       <td>
          To remove the orphan data files and metadata files. Arguments:
@@ -223,10 +224,12 @@ All available procedures are listed below.
                deletes orphan files older than 1 day by default. This argument can modify the interval.
             </li>
             <li>dryRun: when true, view only orphan files, don't actually remove files. Default is false.</li>
+            <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
       </td>
       <td>CALL remove_orphan_files('default.T', '2023-10-31 12:00:00')<br/><br/>
           CALL remove_orphan_files('default.*', '2023-10-31 12:00:00')<br/><br/>
-          CALL remove_orphan_files('default.T', '2023-10-31 12:00:00', true)
+          CALL remove_orphan_files('default.T', '2023-10-31 12:00:00', true)<br/><br/>
+          CALL remove_orphan_files('default.T', '2023-10-31 12:00:00', false, '5')
       </td>
    </tr>
    <tr>

docs/content/maintenance/manage-snapshots.md

@@ -314,7 +314,8 @@ CALL sys.remove_orphan_files(table => "my_db.*", [older_than => "2023-10-31 12:0
     --database <database-name> \ 
     --table <table-name> \
     [--older_than <timestamp>] \
-    [--dry_run <false/true>] 
+    [--dry_run <false/true>] \
+    [--parallelism <parallelism>]
 ```
 
 To avoid deleting files that are newly added by other writing jobs, this action only deletes orphan files older than

docs/content/spark/procedures.md

@@ -161,11 +161,13 @@ This section introduce all available spark procedures about paimon.
             <li>table: the target table identifier. Cannot be empty, you can use database_name.* to clean whole database.</li>
             <li>older_than: to avoid deleting newly written files, this procedure only deletes orphan files older than 1 day by default. This argument can modify the interval.</li>
             <li>dry_run: when true, view only orphan files, don't actually remove files. Default is false.</li>
+            <li>parallelism: The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine.</li>
       </td>
       <td>
           CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00')<br/><br/>
           CALL sys.remove_orphan_files(table => 'default.*', older_than => '2023-10-31 12:00:00')<br/><br/>
-          CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true)
+          CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true)<br/><br/>
+          CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true, parallelism => '5')
       </td>
     </tr>
     <tr>

paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java

@@ -95,9 +95,7 @@ public class OrphanFilesClean {
 
     private static final Logger LOG = LoggerFactory.getLogger(OrphanFilesClean.class);
 
-    private static final ThreadPoolExecutor EXECUTOR =
-            createCachedThreadPool(
-                    Runtime.getRuntime().availableProcessors(), "ORPHAN_FILES_CLEAN");
+    private final ThreadPoolExecutor executor;
 
     private static final int READ_FILE_RETRY_NUM = 3;
     private static final int READ_FILE_RETRY_INTERVAL = 5;
@@ -130,6 +128,9 @@ public OrphanFilesClean(FileStoreTable table) {
                     } catch (IOException ignored) {
                     }
                 };
+        this.executor =
+                createCachedThreadPool(
+                        table.coreOptions().deleteFileThreadNum(), "ORPHAN_FILES_CLEAN");
     }
 
     public OrphanFilesClean olderThan(String timestamp) {
@@ -204,7 +205,7 @@ private Set<String> getUsedFiles(FileStoreTable branchTable) throws IOException
 
         return Sets.newHashSet(
                 randomlyExecute(
-                        EXECUTOR, snapshot -> getUsedFiles(branchTable, snapshot), readSnapshots));
+                        executor, snapshot -> getUsedFiles(branchTable, snapshot), readSnapshots));
     }
 
     private List<String> getUsedFiles(FileStoreTable branchTable, Snapshot snapshot) {
@@ -234,7 +235,7 @@ private Map<String, Path> getCandidateDeletingFiles() {
                                 .filter(this::oldEnough)
                                 .map(FileStatus::getPath)
                                 .collect(Collectors.toList());
-        Iterator<Path> allPaths = randomlyExecute(EXECUTOR, processor, fileDirs);
+        Iterator<Path> allPaths = randomlyExecute(executor, processor, fileDirs);
         Map<String, Path> result = new HashMap<>();
         while (allPaths.hasNext()) {
             Path next = allPaths.next();
@@ -525,7 +526,7 @@ private List<Path> listAndCleanDataDirs(Path dir, int level) {
 
         // dive into the next partition level
         return Lists.newArrayList(
-                randomlyExecute(EXECUTOR, p -> listAndCleanDataDirs(p, level - 1), partitionPaths));
+                randomlyExecute(executor, p -> listAndCleanDataDirs(p, level - 1), partitionPaths));
     }
 
     private List<Path> filterAndCleanDataDirs(
@@ -581,7 +582,10 @@ public static List<String> showDeletedFiles(List<Path> deleteFiles, int showLimi
     }
 
     public static List<OrphanFilesClean> createOrphanFilesCleans(
-            Catalog catalog, String databaseName, @Nullable String tableName)
+            Catalog catalog,
+            Map<String, String> tableConfig,
+            String databaseName,
+            @Nullable String tableName)
             throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
         List<OrphanFilesClean> orphanFilesCleans = new ArrayList<>();
         List<String> tableNames = Collections.singletonList(tableName);
@@ -591,7 +595,7 @@ public static List<OrphanFilesClean> createOrphanFilesCleans(
 
         for (String t : tableNames) {
             Identifier identifier = new Identifier(databaseName, t);
-            Table table = catalog.getTable(identifier);
+            Table table = catalog.getTable(identifier).copy(tableConfig);
             checkArgument(
                     table instanceof FileStoreTable,
                     "Only FileStoreTable supports remove-orphan-files action. The table type is '%s'.",

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

@@ -18,13 +18,17 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.operation.OrphanFilesClean;
 import org.apache.paimon.utils.StringUtils;
 
 import org.apache.flink.table.procedure.ProcedureContext;
 
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.paimon.operation.OrphanFilesClean.executeOrphanFilesClean;
 
@@ -58,12 +62,32 @@ public String[] call(ProcedureContext procedureContext, String tableId, String o
     public String[] call(
             ProcedureContext procedureContext, String tableId, String olderThan, boolean dryRun)
             throws Exception {
+        return call(procedureContext, tableId, olderThan, dryRun, "");
+    }
+
+    public String[] call(
+            ProcedureContext procedureContext,
+            String tableId,
+            String olderThan,
+            boolean dryRun,
+            String parallelism)
+            throws Exception {
         Identifier identifier = Identifier.fromString(tableId);
         String databaseName = identifier.getDatabaseName();
         String tableName = identifier.getObjectName();
 
+        Map<String, String> dynamicOptions =
+                !StringUtils.isNullOrWhitespaceOnly(parallelism)
+                        ? Collections.emptyMap()
+                        : new HashMap<String, String>() {
+                            {
+                                put(CoreOptions.DELETE_FILE_THREAD_NUM.key(), parallelism);
+                            }
+                        };
+
         List<OrphanFilesClean> tableCleans =
-                OrphanFilesClean.createOrphanFilesCleans(catalog, databaseName, tableName);
+                OrphanFilesClean.createOrphanFilesCleans(
+                        catalog, dynamicOptions, databaseName, tableName);
 
         if (!StringUtils.isNullOrWhitespaceOnly(olderThan)) {
             tableCleans.forEach(clean -> clean.olderThan(olderThan));

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

@@ -37,11 +37,13 @@ public RemoveOrphanFilesAction(
             String warehouse,
             String databaseName,
             @Nullable String tableName,
-            Map<String, String> catalogConfig)
+            Map<String, String> catalogConfig,
+            Map<String, String> dynamicOptions)
             throws Catalog.TableNotExistException, Catalog.DatabaseNotExistException {
         super(warehouse, catalogConfig);
         this.tableCleans =
-                OrphanFilesClean.createOrphanFilesCleans(catalog, databaseName, tableName);
+                OrphanFilesClean.createOrphanFilesCleans(
+                        catalog, dynamicOptions, databaseName, tableName);
     }
 
     public void olderThan(String olderThan) {

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

@@ -18,6 +18,9 @@
 
 package org.apache.paimon.flink.action;
 
+import org.apache.paimon.CoreOptions;
+
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 
@@ -29,6 +32,7 @@ public class RemoveOrphanFilesActionFactory implements ActionFactory {
     public static final String IDENTIFIER = "remove_orphan_files";
     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() {
@@ -44,10 +48,16 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
         String table = params.get(TABLE);
 
         Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
+        Map<String, String> dynamicOptions = new HashMap<>();
+        if (params.has(PARALLELISM)) {
+            dynamicOptions.put(CoreOptions.DELETE_FILE_THREAD_NUM.key(), params.get(PARALLELISM));
+        }
 
         RemoveOrphanFilesAction action;
         try {
-            action = new RemoveOrphanFilesAction(warehouse, database, table, catalogConfig);
+            action =
+                    new RemoveOrphanFilesAction(
+                            warehouse, database, table, catalogConfig, dynamicOptions);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.operation.OrphanFilesClean;
 import org.apache.paimon.utils.StringUtils;
@@ -27,7 +28,10 @@
 import org.apache.flink.table.annotation.ProcedureHint;
 import org.apache.flink.table.procedure.ProcedureContext;
 
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.paimon.operation.OrphanFilesClean.executeOrphanFilesClean;
 
@@ -56,13 +60,18 @@ public class RemoveOrphanFilesProcedure extends ProcedureBase {
                         name = "older_than",
                         type = @DataTypeHint("STRING"),
                         isOptional = true),
-                @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true)
+                @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true),
+                @ArgumentHint(
+                        name = "parallelism",
+                        type = @DataTypeHint("STRING"),
+                        isOptional = true)
             })
     public String[] call(
             ProcedureContext procedureContext,
             String tableId,
             String nullableOlderThan,
-            Boolean dryRun)
+            Boolean dryRun,
+            String parallelism)
             throws Exception {
         final String olderThan = notnull(nullableOlderThan);
         if (dryRun == null) {
@@ -73,8 +82,18 @@ public String[] call(
         String databaseName = identifier.getDatabaseName();
         String tableName = identifier.getObjectName();
 
+        Map<String, String> dynamicOptions =
+                !StringUtils.isNullOrWhitespaceOnly(parallelism)
+                        ? Collections.emptyMap()
+                        : new HashMap<String, String>() {
+                            {
+                                put(CoreOptions.DELETE_FILE_THREAD_NUM.key(), parallelism);
+                            }
+                        };
+
         List<OrphanFilesClean> tableCleans =
-                OrphanFilesClean.createOrphanFilesCleans(catalog, databaseName, tableName);
+                OrphanFilesClean.createOrphanFilesCleans(
+                        catalog, dynamicOptions, databaseName, tableName);
 
         if (!StringUtils.isNullOrWhitespaceOnly(olderThan)) {
             tableCleans.forEach(clean -> clean.olderThan(olderThan));

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java

@@ -185,6 +185,11 @@ public void testRemoveDatabaseOrphanFilesITCase(boolean isNamedArgument) throws
         RemoveOrphanFilesAction action2 = createAction(RemoveOrphanFilesAction.class, args);
         assertThatCode(action2::run).doesNotThrowAnyException();
 
+        args.add("--parallelism");
+        args.add("5");
+        RemoveOrphanFilesAction action3 = createAction(RemoveOrphanFilesAction.class, args);
+        assertThatCode(action3::run).doesNotThrowAnyException();
+
         String withoutOlderThan =
                 String.format(
                         isNamedArgument
@@ -195,6 +200,11 @@ public void testRemoveDatabaseOrphanFilesITCase(boolean isNamedArgument) throws
         CloseableIterator<Row> withoutOlderThanCollect = callProcedure(withoutOlderThan);
         assertThat(ImmutableList.copyOf(withoutOlderThanCollect).size()).isEqualTo(0);
 
+        String withParallelism =
+                String.format("CALL sys.remove_orphan_files('%s.%s','',true,'5')", database, "*");
+        CloseableIterator<Row> withParallelismCollect = callProcedure(withParallelism);
+        assertThat(ImmutableList.copyOf(withParallelismCollect).size()).isEqualTo(0);
+
         String withDryRun =
                 String.format(
                         isNamedArgument

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.operation.OrphanFilesClean;
 import org.apache.paimon.spark.catalog.WithPaimonCatalog;
 import org.apache.paimon.utils.Preconditions;
@@ -34,7 +35,10 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.paimon.operation.OrphanFilesClean.executeOrphanFilesClean;
 import static org.apache.spark.sql.types.DataTypes.BooleanType;
@@ -58,7 +62,8 @@ public class RemoveOrphanFilesProcedure extends BaseProcedure {
             new ProcedureParameter[] {
                 ProcedureParameter.required("table", StringType),
                 ProcedureParameter.optional("older_than", StringType),
-                ProcedureParameter.optional("dry_run", BooleanType)
+                ProcedureParameter.optional("dry_run", BooleanType),
+                ProcedureParameter.optional("parallelism", StringType)
             };
 
     private static final StructType OUTPUT_TYPE =
@@ -85,6 +90,16 @@ public StructType outputType() {
     public InternalRow[] call(InternalRow args) {
         org.apache.paimon.catalog.Identifier identifier;
         String tableId = args.getString(0);
+        String parallelism = args.isNullAt(3) ? null : args.getString(3);
+        Map<String, String> dynamicOptions =
+                !StringUtils.isNullOrWhitespaceOnly(parallelism)
+                        ? Collections.emptyMap()
+                        : new HashMap<String, String>() {
+                            {
+                                put(CoreOptions.DELETE_FILE_THREAD_NUM.key(), parallelism);
+                            }
+                        };
+
         Preconditions.checkArgument(
                 tableId != null && !tableId.isEmpty(),
                 "Cannot handle an empty tableId for argument %s",
@@ -104,6 +119,7 @@ public InternalRow[] call(InternalRow args) {
             tableCleans =
                     OrphanFilesClean.createOrphanFilesCleans(
                             ((WithPaimonCatalog) tableCatalog()).paimonCatalog(),
+                            dynamicOptions,
                             identifier.getDatabaseName(),
                             identifier.getObjectName());
         } catch (Exception e) {