Increase S3 delete batch size to 1000

Matches batch size used in legacy FS io.trino.hdfs.s3.TrinoS3FileSystem#DELETE_BATCH_SIZE
1000 is the maximum batch size supported by S3 client
Reduces potential of getting throttled by S3

Author: raunaqmorarka

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3FileSystem.java

@@ -69,6 +69,8 @@
 final class S3FileSystem
         implements TrinoFileSystem
 {
+    static final int DELETE_BATCH_SIZE = 1000;
+
     private final Executor uploadExecutor;
     private final S3Client client;
     private final S3Presigner preSigner;
@@ -188,7 +190,7 @@ private void deleteObjects(Collection<Location> locations)
             String bucket = entry.getKey();
             Collection<String> allKeys = entry.getValue();
 
-            for (List<String> keys : partition(allKeys, 250)) {
+            for (List<String> keys : partition(allKeys, DELETE_BATCH_SIZE)) {
                 List<ObjectIdentifier> objects = keys.stream()
                         .map(key -> ObjectIdentifier.builder().key(key).build())
                         .toList();

lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3FileSystemMinIo.java

@@ -13,8 +13,10 @@
  */
 package io.trino.filesystem.s3;
 
+import com.google.common.io.Closer;
 import io.airlift.units.DataSize;
 import io.opentelemetry.api.OpenTelemetry;
+import io.trino.filesystem.Location;
 import io.trino.testing.containers.Minio;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
@@ -25,7 +27,10 @@
 
 import java.io.IOException;
 import java.net.URI;
+import java.util.List;
 
+import static io.trino.filesystem.s3.S3FileSystem.DELETE_BATCH_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class TestS3FileSystemMinIo
@@ -119,4 +124,23 @@ public void testListDirectories()
         // MinIO is not hierarchical but has hierarchical naming constraints. For example it's not possible to have two blobs "level0" and "level0/level1".
         testListDirectories(true);
     }
+
+    @Test
+    void testDeleteManyFiles()
+            throws IOException
+    {
+        try (Closer closer = Closer.create()) {
+            // create a large number of files to test batch deletion over multiple batches
+            // we run this test only on MinIO to avoid API costs and long execution time on AWS S3
+            List<TempBlob> blobs = randomBlobs(closer, DELETE_BATCH_SIZE + 100);
+            List<Location> locations = blobs.stream()
+                    .map(TempBlob::location)
+                    .toList();
+
+            getFileSystem().deleteFiles(locations);
+            for (Location location : locations) {
+                assertThat(getFileSystem().newInputFile(location).exists()).isFalse();
+            }
+        }
+    }
 }

lib/trino-filesystem/src/test/java/io/trino/filesystem/AbstractTestTrinoFileSystem.java

@@ -1214,7 +1214,7 @@ public void testListLexicographicalOrder()
             throws IOException
     {
         try (Closer closer = Closer.create()) {
-            List<TempBlob> blobs = randomBlobs(closer);
+            List<TempBlob> blobs = randomBlobs(closer, 100);
 
             List<Location> sortedLocations = blobs.stream()
                         .map(TempBlob::location)
@@ -1538,11 +1538,11 @@ protected TempBlob randomBlobLocation(String nameHint)
         return tempBlob;
     }
 
-    private List<TempBlob> randomBlobs(Closer closer)
+    protected List<TempBlob> randomBlobs(Closer closer, int count)
     {
         char[] chars = new char[] {'a', 'b', 'c', 'd', 'A', 'B', 'C', 'D'};
         ImmutableList.Builder<TempBlob> names = ImmutableList.builder();
-        for (int i = 0; i < 100; i++) {
+        for (int i = 0; i < count; i++) {
             StringBuilder name = new StringBuilder();
             for (int j = 0; j < 10; j++) {
                 name.append(chars[ThreadLocalRandom.current().nextInt(chars.length)]);