[common][oss] Add strict batch deletion to FileIO

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -217,6 +217,20 @@ default FileStatus[] listDirectories(Path path) throws IOException {
      */
     boolean exists(Path path) throws IOException;
 
+    /**
+     * Deletes files in provider batches when supported.
+     *
+     * <p>{@code false} means that no storage access was made and callers may fall back to
+     * individual deletes. Once an implementation accesses storage, it must either delete every file
+     * (missing files count as deleted) and return {@code true}, or throw an exception.
+     *
+     * @param files files from the same URI scheme and authority
+     * @since 2.1
+     */
+    default boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        return files.isEmpty();
+    }
+
     /**
      * Delete a file.
      *

paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java

@@ -24,6 +24,7 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.List;
 
 /**
  * A {@link FileIO} for plugin jar. {@link FileIO} is serializable, so plugin FileIO should be
@@ -78,6 +79,14 @@ public boolean exists(Path path) throws IOException {
         return wrap(() -> fileIO(path).exists(path));
     }
 
+    @Override
+    public boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        if (files.isEmpty()) {
+            return true;
+        }
+        return wrap(() -> fileIO(files.get(0)).deleteFilesInBatch(files));
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return wrap(() -> fileIO(path).delete(path, recursive));

paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java

@@ -26,12 +26,15 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.net.URI;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
 import static org.apache.paimon.options.CatalogOptions.RESOLVING_FILE_IO_ENABLED;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /**
  * An implementation of {@link FileIO} that supports multiple file system schemas. It dynamically
@@ -94,6 +97,26 @@ public boolean exists(Path path) throws IOException {
         return wrap(() -> fileIO(path).exists(path));
     }
 
+    @Override
+    public boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        checkArgument(files != null, "Batch delete files must not be null.");
+        if (files.isEmpty()) {
+            return true;
+        }
+
+        Path first = files.get(0);
+        checkArgument(first != null, "Batch delete file must not be null.");
+        URI provider = first.toUri();
+        for (Path file : files) {
+            checkArgument(
+                    file != null
+                            && Objects.equals(provider.getScheme(), file.toUri().getScheme())
+                            && Objects.equals(provider.getAuthority(), file.toUri().getAuthority()),
+                    "Batch delete files must use the same URI scheme and authority.");
+        }
+        return wrap(() -> fileIO(first).deleteFilesInBatch(files));
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return wrap(() -> fileIO(path).delete(path, recursive));

paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java

@@ -38,6 +38,7 @@
 import java.time.Duration;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -165,6 +166,11 @@ public boolean exists(Path path) throws IOException {
         return delegate.exists(path);
     }
 
+    @Override
+    public boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        return delegate.deleteFilesInBatch(files);
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return delegate.delete(path, recursive);

paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -47,6 +47,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -145,6 +146,11 @@ public boolean exists(Path path) throws IOException {
         return fileIO().exists(path);
     }
 
+    @Override
+    public boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        return fileIO().deleteFilesInBatch(files);
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return fileIO().delete(path, recursive);

paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java

@@ -29,13 +29,15 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -184,4 +186,14 @@ public void testTryToWriteAtomicReachesResolvedOverride() throws IOException {
         // the interface default would have written a temp file and renamed it instead
         verify(delegate, never()).rename(any(), any());
     }
+
+    @Test
+    public void testBatchDeleteRejectsMixedProviders() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () ->
+                        resolvingFileIO.deleteFilesInBatch(
+                                Arrays.asList(
+                                        new Path("file:///table/a"), new Path("hdfs:///table/b"))));
+    }
 }

paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java

@@ -37,8 +37,11 @@
 import com.aliyun.oss.internal.OSSHeaders;
 import com.aliyun.oss.internal.OSSMultipartOperation;
 import com.aliyun.oss.internal.OSSObjectOperation;
+import com.aliyun.oss.internal.OSSUtils;
 import com.aliyun.oss.model.CopyObjectRequest;
 import com.aliyun.oss.model.CopyObjectResult;
+import com.aliyun.oss.model.DeleteObjectsRequest;
+import com.aliyun.oss.model.DeleteObjectsResult;
 import com.aliyun.oss.model.InitiateMultipartUploadRequest;
 import com.aliyun.oss.model.InitiateMultipartUploadResult;
 import com.aliyun.oss.model.ObjectMetadata;
@@ -57,9 +60,13 @@
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
@@ -73,6 +80,8 @@ public class OSSFileIO extends HadoopCompliantFileIO implements HadoopOptionsPro
 
     private static final Logger LOG = LoggerFactory.getLogger(OSSFileIO.class);
 
+    private static final int MAX_BATCH_DELETE_SIZE = 1000;
+
     /**
      * In order to simplify, we make paimon oss configuration keys same with hadoop oss module. So,
      * we add all configuration key with prefix `fs.oss` in paimon conf to hadoop conf.
@@ -138,6 +147,38 @@ public boolean isObjectStore() {
         return true;
     }
 
+    @Override
+    public boolean deleteFilesInBatch(List<Path> files) throws IOException {
+        checkArgument(files != null, "Batch delete files must not be null.");
+        if (files.isEmpty()) {
+            return true;
+        }
+
+        List<String> keys = validateBatch(files);
+        String bucket = files.get(0).toUri().getHost();
+        OSSClient client;
+        try {
+            client = ossClient(files.get(0));
+        } catch (Exception e) {
+            throw new IOException("Failed to create OSS client for batch delete.", e);
+        }
+
+        for (int start = 0; start < keys.size(); start += MAX_BATCH_DELETE_SIZE) {
+            List<String> batch =
+                    keys.subList(start, Math.min(start + MAX_BATCH_DELETE_SIZE, keys.size()));
+            DeleteObjectsRequest request =
+                    new DeleteObjectsRequest(bucket).withKeys(batch).withQuiet(false);
+            DeleteObjectsResult response;
+            try {
+                response = client.deleteObjects(request);
+            } catch (Exception e) {
+                throw new IOException("Failed to delete OSS object batch.", e);
+            }
+            validateResponse(batch, response);
+        }
+        return true;
+    }
+
     @Override
     public void configure(CatalogContext context) {
         allowCache = context.options().get(FILE_IO_ALLOW_CACHE);
@@ -285,6 +326,55 @@ OSSClient ossClient(Path path) throws Exception {
         return getOssClient((AliyunOSSFileSystem) getFileSystem(path(path)));
     }
 
+    private static List<String> validateBatch(List<Path> files) {
+        List<String> keys = new ArrayList<>(files.size());
+        Set<String> uniqueKeys = new HashSet<>();
+        String bucket = null;
+        for (Path file : files) {
+            checkArgument(file != null, "Batch delete file must not be null.");
+            URI uri = file.toUri();
+            checkArgument("oss".equals(uri.getScheme()), "Batch delete only supports OSS paths.");
+            String host = uri.getHost();
+            checkArgument(
+                    host != null && host.equals(uri.getAuthority()),
+                    "Batch delete OSS authority must contain only a bucket.");
+            OSSUtils.ensureBucketNameValid(host);
+            if (bucket == null) {
+                bucket = host;
+            } else {
+                checkArgument(
+                        bucket.equals(host), "Batch delete files must use the same OSS bucket.");
+            }
+
+            String path = uri.getPath();
+            checkArgument(
+                    path != null && path.length() > 1,
+                    "Batch delete OSS object key must not be empty.");
+            String key = path.substring(1);
+            OSSUtils.ensureObjectKeyValid(key);
+            checkArgument(
+                    uniqueKeys.add(key), "Batch delete object keys must not contain duplicates.");
+            keys.add(key);
+        }
+        return keys;
+    }
+
+    private static void validateResponse(List<String> requestedKeys, DeleteObjectsResult response)
+            throws IOException {
+        if (response == null || response.getDeletedObjects() == null) {
+            throw new IOException("OSS batch delete returned no acknowledgement.");
+        }
+
+        List<String> deletedObjects = response.getDeletedObjects();
+        if (deletedObjects.size() != requestedKeys.size()) {
+            throw new IOException("OSS batch delete returned an incomplete acknowledgement.");
+        }
+
+        if (!new HashSet<>(requestedKeys).equals(new HashSet<>(deletedObjects))) {
+            throw new IOException("OSS batch delete returned an invalid acknowledgement.");
+        }
+    }
+
     @Override
     public void close() {
         if (!allowCache) {