Core: Delegate bulk and prefix operations in ResolvingFileIO

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.io;
 
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -30,15 +31,22 @@
 import org.apache.iceberg.hadoop.SerializableConfiguration;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.PeekingIterator;
 import org.apache.iceberg.util.SerializableMap;
 import org.apache.iceberg.util.SerializableSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** FileIO implementation that uses location scheme to choose the correct FileIO implementation. */
-public class ResolvingFileIO implements FileIO, HadoopConfigurable {
+/**
+ * FileIO implementation that uses location scheme to choose the correct FileIO implementation.
+ * Delegate FileIO implementations should support the mixin interfaces {@link
+ * SupportsPrefixOperations} and {@link SupportsBulkOperations}.
+ */
+public class ResolvingFileIO
+    implements FileIO, SupportsPrefixOperations, SupportsBulkOperations, HadoopConfigurable {
   private static final Logger LOG = LoggerFactory.getLogger(ResolvingFileIO.class);
   private static final String FALLBACK_IMPL = "org.apache.iceberg.hadoop.HadoopFileIO";
   private static final String S3_FILE_IO_IMPL = "org.apache.iceberg.aws.s3.S3FileIO";
@@ -85,6 +93,44 @@ public void deleteFile(String location) {
     io(location).deleteFile(location);
   }
 
+  @Override
+  public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+    // peek at the first element to determine the type of FileIO
+    Iterator<String> originalIterator = pathsToDelete.iterator();
+    if (!originalIterator.hasNext()) {
+      return;
+    }
+
+    PeekingIterator<String> iterator = Iterators.peekingIterator(originalIterator);
+    FileIO fileIO = io(iterator.peek());
+    if (!(fileIO instanceof SupportsPrefixOperations)) {
+      throw new UnsupportedOperationException(
+          "FileIO doesn't support bulk operations: " + fileIO.getClass().getName());
+    }
+
+    ((SupportsBulkOperations) fileIO).deleteFiles(() -> iterator);
+  }
+
+  @Override
+  public Iterable<FileInfo> listPrefix(String prefix) {
+    FileIO fileIO = io(prefix);
+    if (!(fileIO instanceof SupportsPrefixOperations)) {
+      throw new UnsupportedOperationException(
+          "FileIO doesn't support prefix operations: " + fileIO.getClass().getName());
+    }
+    return ((SupportsPrefixOperations) fileIO).listPrefix(prefix);
+  }
+
+  @Override
+  public void deletePrefix(String prefix) {
+    FileIO fileIO = io(prefix);
+    if (!(fileIO instanceof SupportsPrefixOperations)) {
+      throw new UnsupportedOperationException(
+          "FileIO doesn't support prefix operations: " + fileIO.getClass().getName());
+    }
+    ((SupportsPrefixOperations) fileIO).deletePrefix(prefix);
+  }
+
   @Override
   public Map<String, String> properties() {
     return properties.immutableMap();