Simplify & Optimise IndexDeleter locks

sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java

@@ -288,6 +288,11 @@ public java.util.List<org.opensearch.index.engine.exec.coord.CatalogSnapshot> li
 
             @Override
             public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {}
+
+            @Override
+            public boolean isCommitManagedFile(String fileName) {
+                return false;
+            }
         };
 
         Map<String, DataFormatPlugin> plugins = new HashMap<>();
@@ -356,6 +361,11 @@ public java.util.List<org.opensearch.index.engine.exec.coord.CatalogSnapshot> li
 
             @Override
             public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {}
+
+            @Override
+            public boolean isCommitManagedFile(String fileName) {
+                return false;
+            }
         };
 
         Map<String, DataFormatPlugin> plugins = new HashMap<>();
@@ -427,6 +437,11 @@ public java.util.List<org.opensearch.index.engine.exec.coord.CatalogSnapshot> li
 
         @Override
         public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {}
+
+        @Override
+        public boolean isCommitManagedFile(String fileName) {
+            return false;
+        }
     }
 
     private IndexSettings createIndexSettings(String primaryFormat) {

sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java

@@ -263,5 +263,10 @@ public java.util.List<org.opensearch.index.engine.exec.coord.CatalogSnapshot> li
 
         @Override
         public void deleteCommit(org.opensearch.index.engine.exec.coord.CatalogSnapshot snapshot) {}
+
+        @Override
+        public boolean isCommitManagedFile(String fileName) {
+            return false;
+        }
     }
 }

server/src/main/java/org/opensearch/index/engine/exec/CommitFileManager.java

@@ -33,10 +33,12 @@ public interface CommitFileManager {
     /**
      * Returns true if the given file is managed by the commit mechanism
      * (e.g., segments_N, write.lock) and should not be treated as an orphan.
+     * <p>
+     * Implementations MUST return true for all files they manage. Returning false
+     * for a commit-managed file will cause {@code IndexFileDeleter} to treat it
+     * as an orphan and delete it on startup, leading to data loss.
      *
      * @param fileName the file name to check
      */
-    default boolean isCommitManagedFile(String fileName) {
-        return false;
-    }
+    boolean isCommitManagedFile(String fileName);
 }

server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshot.java

@@ -18,6 +18,9 @@
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -55,6 +58,12 @@ public abstract class CatalogSnapshot implements Writeable, Cloneable {
 
     private final AbstractRefCounted refCounter;
 
+    /**
+     * Lazily computed cache of file names grouped by format. Computed once on first access
+     * since segments are immutable after construction.
+     */
+    private volatile Map<String, Collection<String>> filesByFormatCache;
+
     protected CatalogSnapshot(String name, long generation, long version) {
         this.generation = generation;
         this.version = version;
@@ -177,6 +186,35 @@ int refCount() {
      */
     public abstract String serializeToString() throws IOException;
 
+    /**
+     * Returns all file names grouped by data format name. The result is computed once
+     * and cached for the lifetime of this snapshot (segments are immutable after construction).
+     * <p>
+     * Used by {@code IndexFileDeleter} for ref-count bookkeeping, avoiding repeated
+     * iteration over segments on every add/remove call.
+     *
+     * @return unmodifiable map of format name to unmodifiable set of file names
+     */
+    public Map<String, Collection<String>> getFilesByFormat() {
+        Map<String, Collection<String>> cached = this.filesByFormatCache;
+        if (cached != null) {
+            return cached;
+        }
+        Map<String, Set<String>> result = new HashMap<>();
+        for (Segment segment : getSegments()) {
+            for (Map.Entry<String, WriterFileSet> entry : segment.dfGroupedSearchableFiles().entrySet()) {
+                result.computeIfAbsent(entry.getKey(), k -> new HashSet<>()).addAll(entry.getValue().files());
+            }
+        }
+        Map<String, Collection<String>> unmodifiable = new HashMap<>();
+        for (Map.Entry<String, Set<String>> entry : result.entrySet()) {
+            unmodifiable.put(entry.getKey(), Collections.unmodifiableSet(entry.getValue()));
+        }
+        cached = Collections.unmodifiableMap(unmodifiable);
+        this.filesByFormatCache = cached;
+        return cached;
+    }
+
     /**
      * Creates a clone without acquiring a reference count.
      * Used for Lucene compatibility where clone is required.