This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 bb453c283 [core] add parallelism arguement for remove orphan files
(#4044)
bb453c283 is described below
commit bb453c28332ecc4aa41f3db762315d254d9947ec
Author: LsomeYeah <[email protected]>
AuthorDate: Wed Sep 11 15:31:45 2024 +0800
[core] add parallelism arguement for remove orphan files (#4044)
---
docs/content/flink/procedures.md | 7 ++++--
docs/content/maintenance/manage-snapshots.md | 3 ++-
docs/content/spark/procedures.md | 4 +++-
.../apache/paimon/operation/OrphanFilesClean.java | 20 ++++++++++-------
.../procedure/RemoveOrphanFilesProcedure.java | 26 +++++++++++++++++++++-
.../flink/action/RemoveOrphanFilesAction.java | 6 +++--
.../action/RemoveOrphanFilesActionFactory.java | 12 +++++++++-
.../procedure/RemoveOrphanFilesProcedure.java | 25 ++++++++++++++++++---
.../action/RemoveOrphanFilesActionITCase.java | 10 +++++++++
.../procedure/RemoveOrphanFilesProcedure.java | 18 ++++++++++++++-
10 files changed, 111 insertions(+), 20 deletions(-)
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
index 6c66f5411..c3018600c 100644
--- a/docs/content/flink/procedures.md
+++ b/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>
diff --git a/docs/content/maintenance/manage-snapshots.md
b/docs/content/maintenance/manage-snapshots.md
index f7fb94d01..d25124008 100644
--- a/docs/content/maintenance/manage-snapshots.md
+++ b/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
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 249ede0e8..085e5866f 100644
--- a/docs/content/spark/procedures.md
+++ b/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>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
index dd57f269c..f5a32583f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
+++
b/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 class OrphanFilesClean {
} catch (IOException ignored) {
}
};
+ this.executor =
+ createCachedThreadPool(
+ table.coreOptions().deleteFileThreadNum(),
"ORPHAN_FILES_CLEAN");
}
public OrphanFilesClean olderThan(String timestamp) {
@@ -204,7 +205,7 @@ public class OrphanFilesClean {
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 @@ public class OrphanFilesClean {
.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 @@ public class OrphanFilesClean {
// 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 class OrphanFilesClean {
}
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 class OrphanFilesClean {
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'.",
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
index 03a498606..47e2a56c0 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
+++
b/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 class RemoveOrphanFilesProcedure extends
ProcedureBase {
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));
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
index c95db99d5..9b369a0dc 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
@@ -37,11 +37,13 @@ public class RemoveOrphanFilesAction extends ActionBase {
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) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
index 769453d4a..6ac6e7280 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
+++
b/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 class RemoveOrphanFilesActionFactory implements
ActionFactory {
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);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
index 1d6e024b8..b7a857e45 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
+++
b/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.DataTypeHint;
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 class RemoveOrphanFilesProcedure extends
ProcedureBase {
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));
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
index 844ade694..b0bd7d0d6 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCase.java
@@ -185,6 +185,11 @@ public class RemoveOrphanFilesActionITCase extends
ActionITCaseBase {
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 class RemoveOrphanFilesActionITCase extends
ActionITCaseBase {
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
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
index a22ec01a6..6c977e0da 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
+++
b/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 org.slf4j.LoggerFactory;
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 class RemoveOrphanFilesProcedure extends
BaseProcedure {
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 class RemoveOrphanFilesProcedure extends
BaseProcedure {
tableCleans =
OrphanFilesClean.createOrphanFilesCleans(
((WithPaimonCatalog)
tableCatalog()).paimonCatalog(),
+ dynamicOptions,
identifier.getDatabaseName(),
identifier.getObjectName());
} catch (Exception e) {
