JingsongLi commented on code in PR #9247:
URL: https://github.com/apache/paimon/pull/9247#discussion_r3818156848
##########
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:
I don't quite understand the execution process here—is it concurrent
execution? Why is it described in terms of batches?
--
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]
