Core: Bulk deletion in RemoveSnapshots

The current implementation uses the deleteFile() of the FileIO even if it
supports bulk operations. Even though the user of the RemoveSnapshots API can
provide a custom Consumer to perform bulk deletion, Iceberg can be clever enough
itself to find out if bulk deletion is possible on the FileIO.

core/src/main/java/org/apache/iceberg/BulkDeleteConsumer.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.util.List;
+import java.util.function.Consumer;
+import org.apache.iceberg.io.SupportsBulkOperations;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * Consumer class to collect file paths one by one and perform a bulk deletion on them. Not thread
+ * safe.
+ */
+public class BulkDeleteConsumer implements Consumer<String> {
+  private final List<String> files = Lists.newArrayList();
+
+  private final SupportsBulkOperations ops;
+
+  public BulkDeleteConsumer(SupportsBulkOperations ops) {
+    this.ops = ops;
+  }
+
+  @Override
+  public void accept(String file) {
+    files.add(file);
+  }
+
+  public void consumeAll() {
+    if (files.isEmpty()) {
+      return;
+    }
+
+    ops.deleteFiles(files);
+
+    files.clear();
+  }
+}

core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java

@@ -23,6 +23,7 @@
 import java.util.function.Consumer;
 import org.apache.iceberg.avro.Avro;
 import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.io.BulkDeletionFailureException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -75,14 +76,25 @@ protected CloseableIterable<ManifestFile> readManifests(Snapshot snapshot) {
   }
 
   protected void deleteFiles(Set<String> pathsToDelete, String fileType) {
-    Tasks.foreach(pathsToDelete)
-        .executeWith(deleteExecutorService)
-        .retry(3)
-        .stopRetryOn(NotFoundException.class)
-        .suppressFailureWhenFinished()
-        .onFailure(
-            (file, thrown) -> LOG.warn("Delete failed for {} file: {}", fileType, file, thrown))
-        .run(deleteFunc::accept);
+    if (deleteFunc instanceof BulkDeleteConsumer) {
+      pathsToDelete.forEach(deleteFunc);
+
+      try {
+        ((BulkDeleteConsumer) deleteFunc).consumeAll();
+      } catch (BulkDeletionFailureException e) {
+        LOG.warn("Bulk deletion failed for {} file(s).", e.numberFailedObjects(), e);
+      }
+    } else {
+      Tasks.foreach(pathsToDelete)
+          .executeWith(deleteExecutorService)
+          .retry(3)
+          .stopRetryOn(NotFoundException.class)
+          .stopOnFailure()
+          .suppressFailureWhenFinished()
+          .onFailure(
+              (file, thrown) -> LOG.warn("Delete failed for {} file: {}", fileType, file, thrown))
+          .run(deleteFunc::accept);
+    }
   }
 
   protected boolean hasAnyStatisticsFiles(TableMetadata tableMetadata) {

core/src/main/java/org/apache/iceberg/RemoveSnapshots.java

@@ -43,6 +43,7 @@
 import java.util.function.Consumer;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.io.SupportsBulkOperations;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -105,6 +106,10 @@ public void accept(String file) {
 
     this.defaultMaxRefAgeMs =
         PropertyUtil.propertyAsLong(base.properties(), MAX_REF_AGE_MS, MAX_REF_AGE_MS_DEFAULT);
+
+    if (ops.io() instanceof SupportsBulkOperations) {
+      this.deleteFunc = new BulkDeleteConsumer((SupportsBulkOperations) ops.io());
+    }
   }
 
   @Override