zhuyaogai commented on code in PR #9247:
URL: https://github.com/apache/paimon/pull/9247#discussion_r3823768467
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java:
##########
@@ -436,40 +438,85 @@ 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);
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
