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 4f0a475b8 [spark][flink] Introduce orphan file cleaning local and
distributed mode (#4285)
4f0a475b8 is described below
commit 4f0a475b8ec4dba00bfca5a5c05a2f8bcec4995b
Author: bknbkn <[email protected]>
AuthorDate: Tue Oct 8 17:13:51 2024 +0800
[spark][flink] Introduce orphan file cleaning local and distributed mode
(#4285)
---
docs/content/flink/procedures.md | 9 ++-
docs/content/spark/procedures.md | 4 +-
.../paimon/operation/LocalOrphanFilesClean.java | 26 +++++++--
.../procedure/RemoveOrphanFilesProcedure.java | 57 +++++++++++++-----
.../action/RemoveOrphanFilesActionITCase.java | 67 ++++++++++++++++++++++
.../procedure/RemoveOrphanFilesProcedure.java | 57 +++++++++++++-----
.../procedure/RemoveOrphanFilesProcedureTest.scala | 47 +++++++++++++++
7 files changed, 231 insertions(+), 36 deletions(-)
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
index 430b42c3b..ce8c8043a 100644
--- a/docs/content/flink/procedures.md
+++ b/docs/content/flink/procedures.md
@@ -268,12 +268,13 @@ All available procedures are listed below.
<td>remove_orphan_files</td>
<td>
-- Use named argument<br/>
- CALL [catalog.]sys.remove_orphan_files(`table` => 'identifier',
older_than => 'olderThan', dry_run => 'dryRun') <br/><br/>
+ CALL [catalog.]sys.remove_orphan_files(`table` => 'identifier',
older_than => 'olderThan', dry_run => 'dryRun', mode => 'mode') <br/><br/>
-- Use indexed argument<br/>
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')<br/><br/>
- CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan',
'dryRun','parallelism')
+ CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan',
'dryRun','parallelism')<br/><br/>
+ CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan',
'dryRun','parallelism','mode')
</td>
<td>
To remove the orphan data files and metadata files. Arguments:
@@ -283,11 +284,13 @@ All available procedures are listed below.
</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>
+ <li>mode: The mode of remove orphan clean procedure (local or
distributed) . By default is distributed.</li>
</td>
<td>CALL remove_orphan_files(`table` => 'default.T', older_than =>
'2023-10-31 12:00:00')<br/><br/>
CALL remove_orphan_files(`table` => 'default.*', older_than =>
'2023-10-31 12:00:00')<br/><br/>
CALL remove_orphan_files(`table` => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => true)<br/><br/>
- CALL remove_orphan_files(`table` => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => false, parallelism => '5')
+ CALL remove_orphan_files(`table` => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => false, parallelism => '5')<br/><br/>
+ CALL remove_orphan_files(`table` => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => false, parallelism => '5', mode => 'local')
</td>
</tr>
<tr>
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 1557fd6a7..d0d8f2397 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -188,12 +188,14 @@ This section introduce all available spark procedures
about paimon.
<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>
+ <li>mode: The mode of remove orphan clean procedure (local or
distributed) . By default is distributed.</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)<br/><br/>
- CALL sys.remove_orphan_files(table => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => true, parallelism => '5')
+ CALL sys.remove_orphan_files(table => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => true, parallelism => '5')<br/><br/>
+ CALL sys.remove_orphan_files(table => 'default.T', older_than =>
'2023-10-31 12:00:00', dry_run => true, parallelism => '5', mode => 'local')
</td>
</tr>
<tr>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
b/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
index 91b23b9a5..a17434faa 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
@@ -56,15 +56,13 @@ import static
org.apache.paimon.utils.ThreadPoolUtils.randomlyExecute;
/**
* Local {@link OrphanFilesClean}, it will use thread pool to execute deletion.
*
- * <p>Note that, this class is not used any more since each engine should
implement its own
- * distributed one. See `FlinkOrphanFilesClean` and `SparkOrphanFilesClean`.
+ * <p>Note that, this class will be used when the orphan clean mode is local,
else orphan clean will
+ * use distributed one. See `FlinkOrphanFilesClean` and
`SparkOrphanFilesClean`.
*/
public class LocalOrphanFilesClean extends OrphanFilesClean {
private final ThreadPoolExecutor executor;
- private static final int SHOW_LIMIT = 200;
-
private final List<Path> deleteFiles;
public LocalOrphanFilesClean(FileStoreTable table) {
@@ -220,7 +218,23 @@ public class LocalOrphanFilesClean extends
OrphanFilesClean {
return orphanFilesCleans;
}
- public static String[] executeOrphanFilesClean(List<LocalOrphanFilesClean>
tableCleans) {
+ public static long executeDatabaseOrphanFiles(
+ Catalog catalog,
+ String databaseName,
+ @Nullable String tableName,
+ long olderThanMillis,
+ SerializableConsumer<Path> fileCleaner,
+ @Nullable Integer parallelism)
+ throws Catalog.DatabaseNotExistException,
Catalog.TableNotExistException {
+ List<LocalOrphanFilesClean> tableCleans =
+ createOrphanFilesCleans(
+ catalog,
+ databaseName,
+ tableName,
+ olderThanMillis,
+ fileCleaner,
+ parallelism);
+
ExecutorService executorService =
Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
List<Future<List<Path>>> tasks = new ArrayList<>();
@@ -241,6 +255,6 @@ public class LocalOrphanFilesClean extends OrphanFilesClean
{
}
executorService.shutdownNow();
- return showDeletedFiles(cleanOrphanFiles, SHOW_LIMIT).toArray(new
String[0]);
+ return cleanOrphanFiles.size();
}
}
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 9bc20350d..10ad878e0 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
@@ -20,12 +20,15 @@ package org.apache.paimon.flink.procedure;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.orphan.FlinkOrphanFilesClean;
+import org.apache.paimon.operation.LocalOrphanFilesClean;
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.operation.OrphanFilesClean.createFileCleaner;
import static org.apache.paimon.operation.OrphanFilesClean.olderThanMillis;
@@ -55,29 +58,57 @@ public class RemoveOrphanFilesProcedure extends
ProcedureBase {
type = @DataTypeHint("STRING"),
isOptional = true),
@ArgumentHint(name = "dry_run", type =
@DataTypeHint("BOOLEAN"), isOptional = true),
- @ArgumentHint(name = "parallelism", type =
@DataTypeHint("INT"), 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)
+ Integer parallelism,
+ String mode)
throws Exception {
Identifier identifier = Identifier.fromString(tableId);
String databaseName = identifier.getDatabaseName();
String tableName = identifier.getObjectName();
-
- long deleted =
- FlinkOrphanFilesClean.executeDatabaseOrphanFiles(
- procedureContext.getExecutionEnvironment(),
- catalog,
- olderThanMillis(olderThan),
- createFileCleaner(catalog, dryRun),
- parallelism,
- databaseName,
- tableName);
- return new String[] {String.valueOf(deleted)};
+ if (mode == null) {
+ mode = "DISTRIBUTED";
+ }
+ long deletedFiles;
+ try {
+ switch (mode.toUpperCase(Locale.ROOT)) {
+ case "DISTRIBUTED":
+ deletedFiles =
+ FlinkOrphanFilesClean.executeDatabaseOrphanFiles(
+ procedureContext.getExecutionEnvironment(),
+ catalog,
+ olderThanMillis(olderThan),
+ createFileCleaner(catalog, dryRun),
+ parallelism,
+ databaseName,
+ tableName);
+ break;
+ case "LOCAL":
+ deletedFiles =
+ LocalOrphanFilesClean.executeDatabaseOrphanFiles(
+ catalog,
+ databaseName,
+ tableName,
+ olderThanMillis(olderThan),
+ createFileCleaner(catalog, dryRun),
+ parallelism);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown mode: "
+ + mode
+ + ". Only 'DISTRIBUTED' and 'LOCAL' are
supported.");
+ }
+ return new String[] {String.valueOf(deletedFiles)};
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
@Override
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 82c09c667..938a8ce1b 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
@@ -271,4 +271,71 @@ public class RemoveOrphanFilesActionITCase extends
ActionITCaseBase {
ImmutableList<Row> actualDeleteFile =
ImmutableList.copyOf(executeSQL(procedure));
assertThat(actualDeleteFile).containsOnly(Row.of("4"));
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testRunWithMode(boolean isNamedArgument) throws Exception {
+ createTableAndWriteData(tableName);
+
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ "remove_orphan_files",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName));
+ RemoveOrphanFilesAction action1 =
createAction(RemoveOrphanFilesAction.class, args);
+ assertThatCode(action1::run).doesNotThrowAnyException();
+
+ args.add("--older_than");
+ args.add("2023-12-31 23:59:59");
+ RemoveOrphanFilesAction action2 =
createAction(RemoveOrphanFilesAction.class, args);
+ assertThatCode(action2::run).doesNotThrowAnyException();
+
+ String withoutOlderThan =
+ String.format(
+ isNamedArgument
+ ? "CALL sys.remove_orphan_files(`table` =>
'%s.%s')"
+ : "CALL sys.remove_orphan_files('%s.%s')",
+ database,
+ tableName);
+ CloseableIterator<Row> withoutOlderThanCollect =
executeSQL(withoutOlderThan);
+
assertThat(ImmutableList.copyOf(withoutOlderThanCollect)).containsOnly(Row.of("0"));
+
+ String withLocalMode =
+ String.format(
+ isNamedArgument
+ ? "CALL sys.remove_orphan_files(`table` =>
'%s.%s', older_than => '2999-12-31 23:59:59', dry_run => true, parallelism =>
5, mode => 'local')"
+ : "CALL sys.remove_orphan_files('%s.%s',
'2999-12-31 23:59:59', true, 5, 'local')",
+ database,
+ tableName);
+ ImmutableList<Row> actualLocalRunDeleteFile =
+ ImmutableList.copyOf(executeSQL(withLocalMode));
+ assertThat(actualLocalRunDeleteFile).containsOnly(Row.of("2"));
+
+ String withDistributedMode =
+ String.format(
+ isNamedArgument
+ ? "CALL sys.remove_orphan_files(`table` =>
'%s.%s', older_than => '2999-12-31 23:59:59', dry_run => true, parallelism =>
5, mode => 'distributed')"
+ : "CALL sys.remove_orphan_files('%s.%s',
'2999-12-31 23:59:59', true, 5, 'distributed')",
+ database,
+ tableName);
+ ImmutableList<Row> actualDistributedRunDeleteFile =
+ ImmutableList.copyOf(executeSQL(withDistributedMode));
+ assertThat(actualDistributedRunDeleteFile).containsOnly(Row.of("2"));
+
+ String withInvalidMode =
+ String.format(
+ isNamedArgument
+ ? "CALL sys.remove_orphan_files(`table` =>
'%s.%s', older_than => '2999-12-31 23:59:59', dry_run => true, parallelism =>
5, mode => 'unknown')"
+ : "CALL sys.remove_orphan_files('%s.%s',
'2999-12-31 23:59:59', true, 5, 'unknown')",
+ database,
+ tableName);
+ assertThatCode(() -> executeSQL(withInvalidMode))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unknown mode");
+ }
}
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 4f442fbae..293e84ca1 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
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.procedure;
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.operation.LocalOrphanFilesClean;
import org.apache.paimon.operation.OrphanFilesClean;
import org.apache.paimon.spark.catalog.WithPaimonCatalog;
import org.apache.paimon.spark.orphan.SparkOrphanFilesClean;
@@ -32,6 +33,8 @@ import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Locale;
+
import static org.apache.spark.sql.types.DataTypes.BooleanType;
import static org.apache.spark.sql.types.DataTypes.IntegerType;
import static org.apache.spark.sql.types.DataTypes.LongType;
@@ -56,7 +59,8 @@ public class RemoveOrphanFilesProcedure extends BaseProcedure
{
ProcedureParameter.required("table", StringType),
ProcedureParameter.optional("older_than", StringType),
ProcedureParameter.optional("dry_run", BooleanType),
- ProcedureParameter.optional("parallelism", IntegerType)
+ ProcedureParameter.optional("parallelism", IntegerType),
+ ProcedureParameter.optional("mode", StringType)
};
private static final StructType OUTPUT_TYPE =
@@ -98,18 +102,45 @@ public class RemoveOrphanFilesProcedure extends
BaseProcedure {
LOG.info("identifier is {}.", identifier);
Catalog catalog = ((WithPaimonCatalog) tableCatalog()).paimonCatalog();
- long deletedFiles =
- SparkOrphanFilesClean.executeDatabaseOrphanFiles(
- catalog,
- identifier.getDatabaseName(),
- identifier.getTableName(),
- OrphanFilesClean.olderThanMillis(
- args.isNullAt(1) ? null : args.getString(1)),
- OrphanFilesClean.createFileCleaner(
- catalog, !args.isNullAt(2) &&
args.getBoolean(2)),
- args.isNullAt(3) ? null : args.getInt(3));
-
- return new InternalRow[] {newInternalRow(deletedFiles)};
+ String mode = args.isNullAt(4) ? "DISTRIBUTED" : args.getString(4);
+
+ long deletedFiles;
+ try {
+ switch (mode.toUpperCase(Locale.ROOT)) {
+ case "LOCAL":
+ deletedFiles =
+ LocalOrphanFilesClean.executeDatabaseOrphanFiles(
+ catalog,
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ OrphanFilesClean.olderThanMillis(
+ args.isNullAt(1) ? null :
args.getString(1)),
+ OrphanFilesClean.createFileCleaner(
+ catalog, !args.isNullAt(2) &&
args.getBoolean(2)),
+ args.isNullAt(3) ? null : args.getInt(3));
+ break;
+ case "DISTRIBUTED":
+ deletedFiles =
+ SparkOrphanFilesClean.executeDatabaseOrphanFiles(
+ catalog,
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ OrphanFilesClean.olderThanMillis(
+ args.isNullAt(1) ? null :
args.getString(1)),
+ OrphanFilesClean.createFileCleaner(
+ catalog, !args.isNullAt(2) &&
args.getBoolean(2)),
+ args.isNullAt(3) ? null : args.getInt(3));
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown mode: "
+ + mode
+ + ". Only 'DISTRIBUTED' and 'LOCAL' are
supported.");
+ }
+ return new InternalRow[] {newInternalRow(deletedFiles)};
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
public static ProcedureBuilder builder() {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedureTest.scala
index c414515f1..d9d738112 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedureTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedureTest.scala
@@ -172,4 +172,51 @@ class RemoveOrphanFilesProcedureTest extends
PaimonSparkTestBase {
checkAnswer(spark.sql(s"CALL sys.remove_orphan_files(table => 'test.*')"),
Row(0) :: Nil)
}
+ test("Paimon procedure: remove orphan files with mode") {
+ spark.sql(s"""
+ |CREATE TABLE T (id STRING, name STRING)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO T VALUES ('1', 'a'), ('2', 'b')")
+
+ val table = loadTable("T")
+ val fileIO = table.fileIO()
+ val tablePath = table.location()
+
+ val orphanFile1 = new Path(tablePath, ORPHAN_FILE_1)
+ val orphanFile2 = new Path(tablePath, ORPHAN_FILE_2)
+
+ fileIO.tryToWriteAtomic(orphanFile1, "a")
+ Thread.sleep(2000)
+ fileIO.tryToWriteAtomic(orphanFile2, "b")
+
+ // by default, no file deleted
+ checkAnswer(spark.sql(s"CALL sys.remove_orphan_files(table => 'T')"),
Row(0) :: Nil)
+
+ val orphanFile2ModTime =
fileIO.getFileStatus(orphanFile2).getModificationTime
+ val older_than1 = DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(
+ orphanFile2ModTime -
+ TimeUnit.SECONDS.toMillis(1)),
+ 3)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.remove_orphan_files(table => 'T', older_than =>
'$older_than1', mode => 'diSTributed')"),
+ Row(1) :: Nil)
+
+ val older_than2 = DateTimeUtils.formatLocalDateTime(
+ DateTimeUtils.toLocalDateTime(System.currentTimeMillis()),
+ 3)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.remove_orphan_files(table => 'T', older_than =>
'$older_than2', mode => 'local')"),
+ Row(1) :: Nil)
+
+ checkAnswer(spark.sql(s"CALL sys.remove_orphan_files(table => 'T')"),
Row(0) :: Nil)
+ }
+
}
