Skip to content

[flink] Support table batching for orphan file cleanup #9247

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhuyaogai wants to merge 4 commits into
base: master
Choose a base branch
from
+134 −36
Open

[flink] Support table batching for orphan file cleanup #9247

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2639012
[flink] Support table batching for orphan file cleanup
zhuyaogai Aug 16, 2026
0a27a8f
Merge remote-tracking branch 'origin/master' into issue-7155
zhuyaogai Aug 16, 2026
8022bb0
Merge branch 'apache:master' into issue-7155
zhuyaogai Aug 16, 2026
026cf05
[flink] Distinguish orphan file cleanup batch jobs
zhuyaogai Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions docs/docs/flink/procedures.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,13 +407,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', mode => 'mode') <br/><br/>
CALL [catalog.]sys.remove_orphan_files(`table` => 'identifier', older_than => 'olderThan', dry_run => 'dryRun', mode => 'mode', table_batch_size => 'tableBatchSize') <br/><br/>

@JingsongLi JingsongLi Aug 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe max_table_number?

@zhuyaogai zhuyaogai Aug 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This parameter controls the number of tables included in each batch (each submitted Flink job), rather than limiting the total number of tables to clean.

For example, 23 tables with a value of 10 will be processed in three batches: 10, 10, and 3. All 23 tables will still be cleaned.

max_table_number might be interpreted as a limit on the total number of tables to process. I think table_batch_size better reflects the current semantics. Would max_tables_per_batch be clearer?

@zhuyaogai zhuyaogai Aug 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @JingsongLi, could you elaborate a bit on what you have in mind with max_table_number?

The current design submits all matched tables in sequential Flink jobs, and this parameter controls how many tables are included in each job.

I'm not sure whether your suggestion is only about the parameter name, or whether you have different semantics or an alternative design in mind. I'd like to understand your intention before changing the API.

-- Use indexed argument<br/>
CALL [catalog.]sys.remove_orphan_files('identifier')<br/>
CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan')<br/>
CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun')<br/>
CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun','parallelism')<br/>
CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun','parallelism','mode')
CALL [catalog.]sys.remove_orphan_files('identifier', 'olderThan', 'dryRun','parallelism','mode','tableBatchSize')
</td>
<td>
To remove the orphan data files and metadata files. Arguments:
Expand All @@ -424,12 +424,14 @@ All available procedures are listed below.
<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>
<li>tableBatchSize: The maximum number of tables cleaned by each distributed Flink job. Default is 10.</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 => false, parallelism => 5)<br/><br/>
CALL sys.remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5, mode => 'local')
CALL sys.remove_orphan_files(`table` => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => false, parallelism => 5, mode => 'local')<br/><br/>
CALL sys.remove_orphan_files(`table` => 'default.*', older_than => '2023-10-31 12:00:00', table_batch_size => 5)
</td>
</tr>
<tr>
Expand Down
15 changes: 14 additions & 1 deletion ...link-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,18 @@ public String[] call(
Integer parallelism,
String mode)
throws Exception {
return call(procedureContext, tableId, olderThan, dryRun, parallelism, mode, null);
}

public String[] call(
ProcedureContext procedureContext,
String tableId,
String olderThan,
boolean dryRun,
Integer parallelism,
String mode,
Integer tableBatchSize)
throws Exception {
Identifier identifier = Identifier.fromString(tableId);
String databaseName = identifier.getDatabaseName();
String tableName = identifier.getObjectName();
Expand All @@ -99,7 +111,8 @@ public String[] call(
dryRun,
parallelism,
databaseName,
tableName);
tableName,
tableBatchSize);
break;
case "LOCAL":
cleanOrphanFilesResult =
Expand Down
107 changes: 79 additions & 28 deletions ...imon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,11 @@
import static org.apache.flink.util.Preconditions.checkState;
import static org.apache.paimon.utils.Preconditions.checkArgument;

/** Flink {@link OrphanFilesClean}, it will submit a job for a table. */
/** Flink {@link OrphanFilesClean}, it will submit jobs in table batches. */
public class FlinkOrphanFilesClean extends OrphanFilesClean {

public static final int DEFAULT_TABLE_BATCH_SIZE = 10;

protected static final Logger LOG = LoggerFactory.getLogger(FlinkOrphanFilesClean.class);

@Nullable protected final Integer parallelism;
Expand Down Expand Up @@ -436,49 +438,98 @@ public static CleanOrphanFilesResult executeDatabaseOrphanFiles(
String databaseName,
@Nullable String tableName)
throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
return executeDatabaseOrphanFiles(
env,
catalog,
olderThanMillis,
dryRun,
parallelism,
databaseName,
tableName,
DEFAULT_TABLE_BATCH_SIZE);
}

public static CleanOrphanFilesResult executeDatabaseOrphanFiles(
StreamExecutionEnvironment env,
Catalog catalog,
long olderThanMillis,
boolean dryRun,
@Nullable Integer parallelism,
String databaseName,
@Nullable String tableName,
@Nullable Integer tableBatchSize)
throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
List<String> tableNames = Collections.singletonList(tableName);
if (tableName == null || "*".equals(tableName)) {
tableNames = catalog.listTables(databaseName);
}

List<DataStream<CleanOrphanFilesResult>> orphanFilesCleans =
new ArrayList<>(tableNames.size());
for (String t : tableNames) {
Identifier identifier = new Identifier(databaseName, t);
Table table = catalog.getTable(identifier);
checkArgument(
table instanceof FileStoreTable,
"Only FileStoreTable supports remove-orphan-files action. The table type is '%s'.",
table.getClass().getName());

DataStream<CleanOrphanFilesResult> clean =
new FlinkOrphanFilesClean(
(FileStoreTable) table, olderThanMillis, dryRun, parallelism)
.doOrphanClean(env);
if (clean != null) {
orphanFilesCleans.add(clean);
int batchSize = tableBatchSize == null ? DEFAULT_TABLE_BATCH_SIZE : tableBatchSize;
checkArgument(batchSize > 0, "Table batch size must be greater than 0.");

long deletedFilesCount = 0;
long deletedFilesLenInBytes = 0;
for (int start = 0; start < tableNames.size(); start += batchSize) {
int end = Math.min(start + batchSize, tableNames.size());
int batchNumber = start / batchSize + 1;
int tableCount = end - start;
long batchStart = System.currentTimeMillis();
LOG.info(
"Starting orphan files clean batch #{} with {} tables.",
batchNumber,
tableCount);

List<DataStream<CleanOrphanFilesResult>> orphanFilesCleans =
new ArrayList<>(tableCount);
for (String t : tableNames.subList(start, end)) {
Identifier identifier = new Identifier(databaseName, t);
Table table = catalog.getTable(identifier);
checkArgument(
table instanceof FileStoreTable,
"Only FileStoreTable supports remove-orphan-files action. The table type is '%s'.",
table.getClass().getName());

DataStream<CleanOrphanFilesResult> clean =
new FlinkOrphanFilesClean(
(FileStoreTable) table,
olderThanMillis,
dryRun,
parallelism)
.doOrphanClean(env);
if (clean != null) {
orphanFilesCleans.add(clean);
}
}
}

DataStream<CleanOrphanFilesResult> result = null;
for (DataStream<CleanOrphanFilesResult> clean : orphanFilesCleans) {
if (result == null) {
result = clean;
} else {
result = result.union(clean);
DataStream<CleanOrphanFilesResult> result = null;
for (DataStream<CleanOrphanFilesResult> clean : orphanFilesCleans) {
result = result == null ? clean : result.union(clean);

@JingsongLi JingsongLi Aug 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't quite understand the execution process here—is it concurrent execution? Why is it described in terms of batches?

@zhuyaogai zhuyaogai Aug 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, tables within one batch are added to the same Flink JobGraph through union, so their cleanup pipelines may execute concurrently.

Different batches are executed sequentially. sum(result) calls executeAndCollect() and consumes the iterator until the current Flink job finishes. Only then does the outer loop build and submit the next job.

Therefore, the parameter limits the number of tables included in each submitted Flink job, rather than directly limiting task concurrency. Does this execution model match what you had in mind, or would you prefer a different way to control resource usage?

}

CleanOrphanFilesResult batchResult = executeAndAggregateResults(result, batchNumber);
deletedFilesCount += batchResult.getDeletedFileCount();
deletedFilesLenInBytes += batchResult.getDeletedFileTotalLenInBytes();
LOG.info(
"Finished orphan files clean batch #{} with {} tables in {} ms.",
batchNumber,
tableCount,
System.currentTimeMillis() - batchStart);
}

return sum(result);
return new CleanOrphanFilesResult(deletedFilesCount, deletedFilesLenInBytes);
}

private static CleanOrphanFilesResult sum(DataStream<CleanOrphanFilesResult> deleted) {
private static CleanOrphanFilesResult executeAndAggregateResults(
DataStream<CleanOrphanFilesResult> cleanResults, int batchNumber) {
long deletedFilesCount = 0;
long deletedFilesLenInBytes = 0;
if (deleted != null) {
if (cleanResults != null) {
try {
CloseableIterator<CleanOrphanFilesResult> iterator =
deleted.global().executeAndCollect("OrphanFilesClean");
cleanResults
.global()
.executeAndCollect(
String.format("OrphanFilesClean-Batch-%d", batchNumber));
while (iterator.hasNext()) {
CleanOrphanFilesResult cleanOrphanFilesResult = iterator.next();
deletedFilesCount += cleanOrphanFilesResult.getDeletedFileCount();
Expand Down
12 changes: 9 additions & 3 deletions ...nk-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,15 +59,20 @@ public class RemoveOrphanFilesProcedure extends ProcedureBase {
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)
@ArgumentHint(name = "mode", type = @DataTypeHint("STRING"), isOptional = true),
@ArgumentHint(
name = "table_batch_size",
type = @DataTypeHint("INT"),
isOptional = true)
})
public String[] call(
ProcedureContext procedureContext,
String tableId,
String olderThan,
Boolean dryRun,
Integer parallelism,
String mode)
String mode,
Integer tableBatchSize)
throws Exception {
Identifier identifier = Identifier.fromString(tableId);
String databaseName = identifier.getDatabaseName();
Expand All @@ -87,7 +92,8 @@ public String[] call(
dryRun != null && dryRun,
parallelism,
databaseName,
tableName);
tableName,
tableBatchSize);
break;
case "LOCAL":
cleanOrphanFilesResult =
Expand Down
28 changes: 27 additions & 1 deletion ...ommon/src/test/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionITCaseBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@

import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;

import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -69,6 +71,15 @@ public abstract class RemoveOrphanFilesActionITCaseBase extends ActionITCaseBase
private static final String ORPHAN_FILE_1 = "bucket-0/orphan_file1";
private static final String ORPHAN_FILE_2 = "bucket-0/orphan_file2";

private long countFinishedOrphanFilesCleanJobs() throws Exception {
try (ClusterClient<?> client = MINI_CLUSTER_EXTENSION.createRestClusterClient()) {
return client.listJobs().get().stream()
.filter(job -> job.getJobName().startsWith("OrphanFilesClean-Batch-"))
.filter(job -> job.getJobState() == JobStatus.FINISHED)
.count();
}
}

private FileStoreTable createTableAndWriteData(String tableName) throws Exception {
RowType rowType =
RowType.of(
Expand Down Expand Up @@ -248,8 +259,24 @@ public void testRemoveDatabaseOrphanFilesITCase(boolean isNamedArgument) throws
database,
"*",
olderThan);
long defaultBatchJobCountBefore = countFinishedOrphanFilesCleanJobs();
ImmutableList<Row> actualDryRunDeleteFile = ImmutableList.copyOf(executeSQL(withDryRun));
assertThat(actualDryRunDeleteFile).containsOnly(Row.of("4"));
assertThat(countFinishedOrphanFilesCleanJobs() - defaultBatchJobCountBefore).isEqualTo(1);

String withBatchSize =
String.format(
isNamedArgument
? "CALL sys.remove_orphan_files(`table` => '%s.%s', older_than => '%s', dry_run => true, table_batch_size => 1)"
: "CALL sys.remove_orphan_files('%s.%s', '%s', true, 5, 'distributed', 1)",
database,
"*",
olderThan);
long configuredBatchJobCountBefore = countFinishedOrphanFilesCleanJobs();
ImmutableList<Row> actualBatchDeleteFile = ImmutableList.copyOf(executeSQL(withBatchSize));
assertThat(actualBatchDeleteFile).containsOnly(Row.of("4"));
assertThat(countFinishedOrphanFilesCleanJobs() - configuredBatchJobCountBefore)
.isEqualTo(2);

String withOlderThan =
String.format(
Expand All @@ -260,7 +287,6 @@ public void testRemoveDatabaseOrphanFilesITCase(boolean isNamedArgument) throws
"*",
olderThan);
ImmutableList<Row> actualDeleteFile = ImmutableList.copyOf(executeSQL(withOlderThan));

assertThat(actualDeleteFile).containsOnly(Row.of("4"));
}

Expand Down
Loading