Split vacuum work definition and execution

Author: mdesmet

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/VacuumProcedure.java

@@ -24,6 +24,7 @@
 import io.trino.filesystem.Location;
 import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.TrinoInputFile;
 import io.trino.plugin.base.util.UncheckedCloseable;
 import io.trino.plugin.deltalake.DeltaLakeConfig;
 import io.trino.plugin.deltalake.DeltaLakeMetadata;
@@ -257,9 +258,8 @@ private void doVacuum(
             long transactionLogFiles = 0;
             long retainedKnownFiles = 0;
             long retainedUnknownFiles = 0;
-            long removedFiles = 0;
+            List<TrinoInputFile> filesToDelete = new ArrayList<>();
 
-            List<Location> filesToDelete = new ArrayList<>();
             FileIterator listing = fileSystem.listFiles(Location.of(tableLocation));
             while (listing.hasNext()) {
                 FileEntry entry = listing.next();
@@ -297,19 +297,20 @@ private void doVacuum(
                     retainedUnknownFiles++;
                     continue;
                 }
-
                 log.debug("[%s] deleting file [%s] with modification time %s", queryId, location, modificationTime);
-                filesToDelete.add(entry.location());
-                if (filesToDelete.size() == DELETE_BATCH_SIZE) {
-                    fileSystem.deleteFiles(filesToDelete);
-                    removedFiles += filesToDelete.size();
-                    filesToDelete.clear();
-                }
+
+                Location fileLocation = Location.of(location);
+                TrinoInputFile inputFile = fileSystem.newInputFile(fileLocation);
+                filesToDelete.add(inputFile);
             }
+            int totalFilesToDelete = filesToDelete.size();
+            int batchCount = (int) Math.ceil((double) totalFilesToDelete / DELETE_BATCH_SIZE);
+            for (int batchNumber = 0; batchNumber < batchCount; batchNumber++) {
+                int start = batchNumber * DELETE_BATCH_SIZE;
+                int end = Math.min(start + DELETE_BATCH_SIZE, totalFilesToDelete);
 
-            if (!filesToDelete.isEmpty()) {
-                fileSystem.deleteFiles(filesToDelete);
-                removedFiles += filesToDelete.size();
+                List<TrinoInputFile> batch = filesToDelete.subList(start, end);
+                fileSystem.deleteFiles(batch.stream().map(TrinoInputFile::location).collect(toImmutableList()));
             }
 
             log.info(
@@ -321,7 +322,7 @@ private void doVacuum(
                     transactionLogFiles,
                     retainedKnownFiles,
                     retainedUnknownFiles,
-                    removedFiles);
+                    totalFilesToDelete);
         }
     }
 }