Support bulk deletion in batch file cleanup task

[danielhumanmod]

Summary

As a follow up improvement for #515

After introducing batch file clean up, we want to make use of the bulk deletion APIs provided by Iceberg, which essentially utilize the bulk deletion ability of many object storage services (e.g., S3, GCS, Azure Blob Storage) to achieve better performance and lower cost

Recommended Review Order

1. FileCleanupTaskHandler.java
2. BatchFileCleanupTaskHandler.java
3. TableCleanupTaskHandler.java
4. Unit Tests

Iceberg API Reference

// https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L194
    public static void deleteFiles(FileIO io, Iterable<String> files, String type, boolean concurrent) {
        if (io instanceof SupportsBulkOperations) {
            try {
                SupportsBulkOperations bulkIO = (SupportsBulkOperations)io;
                bulkIO.deleteFiles(files);
            } catch (RuntimeException e) {
                LOG.warn("Failed to bulk delete {} files", type, e);
            }
        } else if (concurrent) {
            deleteFiles(io, files, type);
        } else {
            files.forEach((file) -> deleteFile(io, file, type));
        }

[eric-maynard]

This LGTM, but it's important to note that we may not retry in the event that the service dies. Eventually, we should have Polaris try to drain the task queue for any tasks that failed the first time they were run

[danielhumanmod]

that's a good catch, do you mind if I follow up with a new PR to do this, it might involve refactoring on existing delete task

[adnanhemani]

@eric-maynard - I was looking at the currently existent tryDelete function regarding this comment. Is this comment something that also applies to that function?

[danielhumanmod]

@eric-maynard - I was looking at the currently existent tryDelete function regarding this comment. Is this comment something that also applies to that function?

No @adnanhemani , current PR mainly focusing on provide more efficient bulk deletion, but can not guarantee task got eventually executed.

For Eric's comment, we have a plan to fix that, please refer to #774

[adnanhemani]

@danielhumanmod - my question was that does the current tryDelete function also suffer from the same lack of guarantees that the task may not eventually retry. (Was a question for my learning to see if there was anything fundamentally different in this new function that was previously covered by the existing function or tasks in general, not accusatory or asking for any change). After reading #774 - it seems like the same comment apply to both tasks. Thanks!

[adnanhemani]

One nit and a clarifying question :)

[adnanhemani]

nit: this is now a single future, so maybe we should name this deleteFuture instead

[adnanhemani]

Not completely sure here, so am asking for context: why did we introduce this enum type here instead of keeping it as a record? Is this going to be extensible for something in the immediate future?

[danielhumanmod]

The record in line 114 is still there — we added this enum to specify what kind of file the BatchFileCleanupTask is cleaning up.

[adnanhemani]

That makes a lot more sense :) Thanks!