Batch deletes in remote segment store

server/src/main/java/org/opensearch/index/store/RemoteDirectory.java

@@ -184,6 +184,22 @@ public void deleteFile(String name) throws IOException {
         blobContainer.deleteBlobsIgnoringIfNotExists(Collections.singletonList(name));
     }
 
+    /**
+     * Removes multiple existing files in the directory in a batch operation.
+     *
+     * <p>This method will not throw an exception when a file doesn't exist and simply ignores missing files.
+     * This is consistent with the behavior of {@link #deleteFile(String)}.
+     *
+     * @param names the collection of filenames to delete.
+     * @throws IOException if the files exist but could not be deleted.
+     */
+    public void deleteFiles(List<String> names) throws IOException {
+        if (names == null || names.isEmpty()) {
+            return;
+        }
+        blobContainer.deleteBlobsIgnoringIfNotExists(names);
+    }
+
     /**
      * Creates and returns a new instance of {@link RemoteIndexOutput} which will be used to copy files to the remote
      * store.

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java

@@ -1013,28 +1013,35 @@ public void deleteStaleSegments(int lastNMetadataFilesToKeep) throws IOException
                 .stream()
                 .map(metadata -> metadata.uploadedFilename)
                 .collect(Collectors.toSet());
-            AtomicBoolean deletionSuccessful = new AtomicBoolean(true);
-            staleSegmentRemoteFilenames.stream()
+
+            // Collect all files to delete for this metadata file
+            List<String> filesToDelete = staleSegmentRemoteFilenames.stream()
                 .filter(file -> activeSegmentRemoteFilenames.contains(file) == false)
                 .filter(file -> deletedSegmentFiles.contains(file) == false)
-                .forEach(file -> {
-                    try {
-                        remoteDataDirectory.deleteFile(file);
-                        deletedSegmentFiles.add(file);
-                        if (!activeSegmentFilesMetadataMap.containsKey(getLocalSegmentFilename(file))) {
-                            segmentsUploadedToRemoteStore.remove(getLocalSegmentFilename(file));
-                        }
-                    } catch (NoSuchFileException e) {
-                        logger.info("Segment file {} corresponding to metadata file {} does not exist in remote", file, metadataFile);
-                    } catch (IOException e) {
-                        deletionSuccessful.set(false);
-                        logger.warn(
-                            "Exception while deleting segment file {} corresponding to metadata file {}. Deletion will be re-tried",
-                            file,
-                            metadataFile
-                        );
+                .collect(Collectors.toList());
+
+            AtomicBoolean deletionSuccessful = new AtomicBoolean(true);
+            try {
+                // Batch delete all stale segment files
+                remoteDataDirectory.deleteFiles(filesToDelete);
+                deletedSegmentFiles.addAll(filesToDelete);
+
+                // Update cache after successful batch deletion
+                for (String file : filesToDelete) {
+                    if (!activeSegmentFilesMetadataMap.containsKey(getLocalSegmentFilename(file))) {
+                        segmentsUploadedToRemoteStore.remove(getLocalSegmentFilename(file));
                     }
-                });
+                }
+            } catch (IOException e) {
+                deletionSuccessful.set(false);
+                logger.warn(
+                    () -> new ParameterizedMessage(
+                        "Exception while deleting segment files corresponding to metadata file {}. Deletion will be re-tried",
+                        metadataFile
+                    ),
+                    e
+                );
+            }
             if (deletionSuccessful.get()) {
                 logger.debug("Deleting stale metadata file {} from remote segment store", metadataFile);
                 remoteMetadataDirectory.deleteFile(metadataFile);

server/src/test/java/org/opensearch/index/store/BaseRemoteSegmentStoreDirectoryTests.java

@@ -113,9 +113,9 @@ protected Map<String, Map<String, String>> populateMetadata() throws IOException
             metadataFilename,
             getDummyMetadata("_0", 1),
             metadataFilename2,
-            getDummyMetadata("_0", 1),
+            getDummyMetadata("_0", 2),
             metadataFilename3,
-            getDummyMetadata("_0", 1)
+            getDummyMetadata("_0", 3)
         );
 
         when(remoteMetadataDirectory.getBlobStream(metadataFilename)).thenAnswer(

server/src/test/java/org/opensearch/index/store/RemoteDirectoryTests.java

@@ -198,6 +198,61 @@ public void testDeleteFileException() throws IOException {
         assertThrows(IOException.class, () -> remoteDirectory.deleteFile("segment_1"));
     }
 
+    /**
+     *
+     * Tests that deleteFiles successfully deletes multiple files from the remote store.
+     */
+    public void testDeleteFiles() throws IOException {
+        List<String> filesToDelete = List.of("segment_1", "segment_2", "segment_3");
+
+        remoteDirectory.deleteFiles(filesToDelete);
+
+        verify(blobContainer).deleteBlobsIgnoringIfNotExists(filesToDelete);
+    }
+
+    /**
+     *
+     * Tests that deleteFiles handles empty collection gracefully without attempting any deletions.
+     */
+    public void testDeleteFilesEmptyCollection() throws IOException {
+        remoteDirectory.deleteFiles(Collections.emptyList());
+
+        verify(blobContainer, times(0)).deleteBlobsIgnoringIfNotExists(any());
+    }
+
+    /**
+     *
+     * Tests that deleteFiles handles null collection gracefully without attempting any deletions.
+     */
+    public void testDeleteFilesNullCollection() throws IOException {
+        remoteDirectory.deleteFiles(null);
+        verify(blobContainer, times(0)).deleteBlobsIgnoringIfNotExists(any());
+    }
+
+    /**
+     *
+     * Tests that deleteFiles completes successfully even when some files don't exist.
+     * The underlying deleteBlobsIgnoringIfNotExists should handle non-existent files gracefully.
+     */
+    public void testDeleteFilesWithNonExistentFiles() throws IOException {
+        List<String> filesToDelete = List.of("segment_1", "non_existent", "segment_2");
+
+        remoteDirectory.deleteFiles(filesToDelete);
+
+        verify(blobContainer).deleteBlobsIgnoringIfNotExists(filesToDelete);
+    }
+
+    /**
+     *
+     * Tests that deleteFiles propagates IOException when the underlying blob container operation fails.
+     */
+    public void testDeleteFilesException() throws IOException {
+        List<String> filesToDelete = List.of("segment_1", "segment_2");
+        doThrow(new IOException("Error writing to blob store")).when(blobContainer).deleteBlobsIgnoringIfNotExists(filesToDelete);
+
+        assertThrows(IOException.class, () -> remoteDirectory.deleteFiles(filesToDelete));
+    }
+
     public void testCreateOutput() {
         IndexOutput indexOutput = remoteDirectory.createOutput("segment_1", IOContext.DEFAULT);
         assertTrue(indexOutput instanceof RemoteIndexOutput);