Fix sync method to account for block files in Composite Directory

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

@@ -42,6 +42,7 @@
 import java.util.stream.Stream;
 
 import static org.opensearch.index.store.remote.utils.FileTypeUtils.BLOCK_FILE_IDENTIFIER;
+import static org.opensearch.index.store.remote.utils.FileTypeUtils.isBlockFile;
 import static org.apache.lucene.index.IndexFileNames.SEGMENTS;
 
 /**
@@ -105,6 +106,24 @@ protected List<String> listBlockFiles(String fileName) throws IOException {
             .collect(Collectors.toList());
     }
 
+    /**
+     * Returns a list of names of all block files stored in the local directory for a given set of files,
+     * including the original file names itself if present.
+     *
+     * @param fileNames The set of files to search for, along with its associated block files.
+     * @return A list of file names, including the original file (if present) and all its block files.
+     * @throws IOException in case of I/O error while listing files.
+     */
+    protected List<String> listBlockFiles(String[] fileNames) throws IOException {
+        Set<String> files = Set.of(fileNames);
+        return Stream.of(listLocalFiles())
+            .filter(
+                file -> files.contains(file)
+                    || (isBlockFile(file) && files.contains(file.substring(0, file.indexOf(BLOCK_FILE_IDENTIFIER))))
+            )
+            .collect(Collectors.toList());
+    }
+
     /**
      * Returns names of all files stored in this directory in sorted order
      * Does not include locally stored block files (having _block_ in their names) and files pending deletion
@@ -249,8 +268,15 @@ public IndexOutput createOutput(String name, IOContext context) throws IOExcepti
     public void sync(Collection<String> names) throws IOException {
         ensureOpen();
         logger.trace("Composite Directory[{}]: sync() called {}", this::toString, () -> names);
-        Collection<String> remoteFiles = Arrays.asList(getRemoteFiles());
-        Collection<String> filesToSync = names.stream().filter(name -> remoteFiles.contains(name) == false).collect(Collectors.toList());
+        Set<String> remoteFiles = Set.of(getRemoteFiles());
+        Set<String> localFiles = Arrays.stream(listLocalFiles())
+            .map(file -> isBlockFile(file) ? file.substring(0, file.indexOf(BLOCK_FILE_IDENTIFIER)) : file)
+            .collect(Collectors.toSet());
+        String[] fullFilesToSync = names.stream().filter(name -> remoteFiles.contains(name) == false).toArray(String[]::new);
+        for (String fullFileToSync : fullFilesToSync) {
+            if (localFiles.contains(fullFileToSync) == false) throw new NoSuchFileException("Unable to sync file " + fullFileToSync);
+        }
+        List<String> filesToSync = listBlockFiles(fullFilesToSync);
         logger.trace("Composite Directory[{}]: Synced files : {}", this::toString, () -> filesToSync);
         localDirectory.sync(filesToSync);
     }

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

@@ -132,6 +132,12 @@ public void testSync() throws IOException {
         // All the files in the below list are present either locally or on remote, so sync should work as expected
         Collection<String> names = List.of("_0.cfe", "_0.cfs", "_0.si", "_1.cfe", "_2.cfe", "segments_1");
         compositeDirectory.sync(names);
+        // Deleting file _1.cfe and then adding its blocks locally so that full file is not present but block files are present in local
+        // State of _1.cfe file after these operations - not present in remote, full file not present locally but blocks present in local
+        compositeDirectory.deleteFile("_1.cfe");
+        addFilesToDirectory(new String[] { "_1.cfe_block_0", "_1.cfe_block_2" });
+        // Sync should work as expected since blocks are present in local
+        compositeDirectory.sync(List.of("_1.cfe"));
         // Below list contains a non-existent file, hence will throw an error
         Collection<String> names1 = List.of("_0.cfe", "_0.cfs", "_0.si", "_1.cfe", "_2.cfe", "segments_1", "non_existent_file");
         assertThrows(NoSuchFileException.class, () -> compositeDirectory.sync(names1));