[flink] Support table batching for orphan file cleanup - #9247
Conversation
# Conflicts: # docs/docs/flink/procedures.md
| <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/> |
There was a problem hiding this comment.
Maybe max_table_number?
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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.
| result = result.union(clean); | ||
| DataStream<CleanOrphanFilesResult> result = null; | ||
| for (DataStream<CleanOrphanFilesResult> clean : orphanFilesCleans) { | ||
| result = result == null ? clean : result.union(clean); |
There was a problem hiding this comment.
I don't quite understand the execution process here—is it concurrent execution? Why is it described in terms of batches?
There was a problem hiding this 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?
2 participants
Purpose
Fixes #7155.
Previously, orphan file cleanup for all matched tables was unioned into a single Flink batch job. In batch mode, Flink does not put all job vertices into the same SlotSharingGroup by default, so tasks belonging to different groups cannot reuse the same slots.
As each table creates its own cleanup pipeline, running many tables in one job can require a large number of slots. Therefore, the actual resource usage may be much higher than the configured cleanup parallelism and can exhaust cluster resources.
This change processes tables in sequential batches and introduces a
table_batch_sizeoption, which defaults to 10. Only a limited number of table cleanup pipelines are included in each batch, bounding the slot resources required by a single job. Batch progress and elapsed time are also logged.Tests
Added integration test coverage for database-level orphan file cleanup with a configured batch size.