Add foundational classes for storage support for multi-format data support

modules/store-subdirectory/src/main/java/org/opensearch/plugin/store/subdirectory/SubdirectoryAwareStore.java

@@ -16,10 +16,8 @@
 import org.apache.lucene.index.SegmentInfos;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
-import org.apache.lucene.store.FilterDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
-import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.util.Version;
 import org.opensearch.common.lucene.Lucene;
 import org.opensearch.core.index.shard.ShardId;
@@ -31,20 +29,12 @@
 import org.opensearch.plugins.IndexStorePlugin;
 
 import java.io.IOException;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
-import java.util.Arrays;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * A store implementation that supports files organized in subdirectories.
@@ -252,109 +242,17 @@ private void computeFileMetadata(String fileName, Map<String, StoreFileMetadata>
 
     /**
      * A Lucene Directory implementation that handles files in subdirectories.
-     *
-     * This directory wrapper enables file operations across subdirectories within
-     * the shard data path. It resolves paths, creates necessary directory structures,
-     * and delegates actual file operations to appropriate filesystem locations.
+     * Extends the server's SubdirectoryAwareDirectory for backward compatibility.
      */
-    public static class SubdirectoryAwareDirectory extends FilterDirectory {
-        private static final Set<String> EXCLUDED_SUBDIRECTORIES = Set.of("index/", "translog/", "_state/");
-        private final ShardPath shardPath;
-
+    public static class SubdirectoryAwareDirectory extends org.opensearch.index.store.SubdirectoryAwareDirectory {
         /**
-         * Constructor for SubdirectoryAwareDirectory.
+         * Creates a new SubdirectoryAwareDirectory wrapping the given delegate.
          *
-         * @param delegate the delegate directory
-         * @param shardPath the shard path
+         * @param delegate  the underlying Lucene directory
+         * @param shardPath the shard path for resolving subdirectories
          */
         public SubdirectoryAwareDirectory(Directory delegate, ShardPath shardPath) {
-            super(delegate);
-            this.shardPath = shardPath;
-        }
-
-        @Override
-        public IndexInput openInput(String name, IOContext context) throws IOException {
-            return super.openInput(parseFilePath(name), context);
-        }
-
-        @Override
-        public IndexOutput createOutput(String name, IOContext context) throws IOException {
-            String targetFilePath = parseFilePath(name);
-            Path targetFile = Path.of(targetFilePath);
-            Files.createDirectories(targetFile.getParent());
-            return super.createOutput(targetFilePath, context);
-        }
-
-        @Override
-        public void deleteFile(String name) throws IOException {
-            super.deleteFile(parseFilePath(name));
-        }
-
-        @Override
-        public long fileLength(String name) throws IOException {
-            return super.fileLength(parseFilePath(name));
-        }
-
-        @Override
-        public void sync(Collection<String> names) throws IOException {
-            super.sync(names.stream().map(this::parseFilePath).collect(Collectors.toList()));
-        }
-
-        @Override
-        public void rename(String source, String dest) throws IOException {
-            super.rename(parseFilePath(source), parseFilePath(dest));
-        }
-
-        @Override
-        public String[] listAll() throws IOException {
-            // Get files from the delegate (regular index files)
-            String[] delegateFiles = super.listAll();
-
-            // Get subdirectory files by scanning all subdirectories
-            Set<String> allFiles = new HashSet<>(Arrays.asList(delegateFiles));
-            addSubdirectoryFiles(allFiles);
-
-            return allFiles.stream().sorted().toArray(String[]::new);
-        }
-
-        private void addSubdirectoryFiles(Set<String> allFiles) throws IOException {
-            Path dataPath = shardPath.getDataPath();
-            Files.walkFileTree(dataPath, new SimpleFileVisitor<>() {
-                @Override
-                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
-                    if (attrs.isRegularFile()) {
-                        Path relativePath = dataPath.relativize(file);
-                        // Only add files that are in subdirectories (have a parent directory)
-                        if (relativePath.getParent() != null) {
-                            String relativePathStr = relativePath.toString();
-                            // Exclude index dir (handled in super.listAll()), translog dir, and _state dir
-                            if (EXCLUDED_SUBDIRECTORIES.stream().noneMatch(relativePathStr::startsWith)) {
-                                allFiles.add(relativePathStr);
-                            }
-                        }
-                    }
-                    return FileVisitResult.CONTINUE;
-                }
-
-                @Override
-                public FileVisitResult visitFileFailed(Path file, IOException e) throws IOException {
-                    if (e instanceof NoSuchFileException) {
-                        logger.debug("Skipping inaccessible file during size estimation: {}", file);
-                        return FileVisitResult.CONTINUE;
-                    }
-                    throw e;
-                }
-            });
-        }
-
-        private String parseFilePath(String fileName) {
-            if (Path.of(fileName).getParent() != null) {
-                // File path (e.g., "subdirectory/segments_1" or "subdirectory/recovery.xxx.segments_1")
-                return shardPath.getDataPath().resolve(fileName).toString();
-            } else {
-                // Simple filename (e.g., "segments_1") - resolve relative to the shard's index directory
-                return shardPath.resolveIndex().resolve(fileName).toString();
-            }
+            super(delegate, shardPath);
         }
     }
 }

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -48,6 +48,7 @@ url = "2.0"
 tempfile = "3.0"
 chrono = "0.4"
 once_cell = "1.21.3"
+crc32fast = "1.4"
 parking_lot = "0.12.5"
 lazy_static = "1.4.0"
 criterion = { version = "0.5", features = ["async_tokio"] }

sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/LuceneReaderManagerTests.java

@@ -134,12 +134,27 @@ public CatalogSnapshot cloneNoAcquire() {
             }
 
             @Override
-            public void setUserData(Map<String, String> userData) {}
+            public void setUserData(Map<String, String> userData, boolean commitData) {}
 
             @Override
             public CatalogSnapshot clone() {
                 return this;
             }
+
+            @Override
+            public int getFormatVersionForFile(String file) {
+                return 0;
+            }
+
+            @Override
+            public byte[] serialize() throws IOException {
+                return new byte[0];
+            }
+
+            @Override
+            public Collection<String> getFiles(boolean includeSegmentsFile) {
+                return List.of();
+            }
         };
     }
 

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/DefaultPlanExecutorTests.java

@@ -292,13 +292,28 @@ public String serializeToString() {
         }
 
         @Override
-        public void setUserData(Map<String, String> userData) {}
+        public void setUserData(Map<String, String> userData, boolean commitData) {}
 
         @Override
         public MockCatalogSnapshot clone() {
             return new MockCatalogSnapshot(generation, segments, format);
         }
 
+        @Override
+        public int getFormatVersionForFile(String file) {
+            return 0;
+        }
+
+        @Override
+        public byte[] serialize() throws IOException {
+            return new byte[0];
+        }
+
+        @Override
+        public Collection<String> getFiles(boolean includeSegmentsFile) {
+            return List.of();
+        }
+
         @Override
         protected void closeInternal() {}
     }

sandbox/plugins/composite-engine/README.md

@@ -23,7 +23,7 @@ Format plugins (e.g., Parquet) extend this plugin by declaring `extendedPlugins
 
 ## Key classes
 
-- **`CompositeEnginePlugin`** — The `ExtensiblePlugin` entry point. Discovers format plugins, validates settings, and creates the composite engine.
+- **`CompositeDataFormatPlugin`** — The `ExtensiblePlugin` entry point. Discovers format plugins, validates settings, and creates the composite engine.
 - **`CompositeIndexingExecutionEngine`** — Orchestrates indexing across primary and secondary format engines.
 - **`CompositeDataFormat`** — A `DataFormat` that wraps multiple per-format instances.
 - **`CompositeDocumentInput`** — Routes field additions to the appropriate per-format `DocumentInput` based on field type capabilities.

sandbox/plugins/composite-engine/build.gradle

@@ -8,7 +8,7 @@
 
 opensearchplugin {
   description = 'Composite indexing engine plugin that orchestrates multi-format indexing across multiple data format engines.'
-  classname = 'org.opensearch.composite.CompositeEnginePlugin'
+  classname = 'org.opensearch.composite.CompositeDataFormatPlugin'
 }
 
 dependencies {

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeEnginePlugin.java → sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java

@@ -12,10 +12,14 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.index.IndexSettings;
 import org.opensearch.index.engine.dataformat.DataFormat;
+import org.opensearch.index.engine.dataformat.DataFormatDescriptor;
 import org.opensearch.index.engine.dataformat.DataFormatPlugin;
 import org.opensearch.index.engine.dataformat.IndexingEngineConfig;
 import org.opensearch.index.engine.dataformat.IndexingExecutionEngine;
+import org.opensearch.index.store.FormatChecksumStrategy;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
 
@@ -44,9 +48,9 @@
  * @opensearch.experimental
  */
 @ExperimentalApi
-public class CompositeEnginePlugin extends Plugin implements ExtensiblePlugin, DataFormatPlugin {
+public class CompositeDataFormatPlugin extends Plugin implements ExtensiblePlugin, DataFormatPlugin {
 
-    private static final Logger logger = LogManager.getLogger(CompositeEnginePlugin.class);
+    private static final Logger logger = LogManager.getLogger(CompositeDataFormatPlugin.class);
 
     /**
      * Index setting that designates the primary data format for an index.
@@ -78,9 +82,10 @@ public class CompositeEnginePlugin extends Plugin implements ExtensiblePlugin, D
      * {@link DataFormat#priority()} is retained.
      */
     private volatile Map<String, DataFormatPlugin> dataFormatPlugins = Map.of();
+    private volatile Map<String, DataFormatDescriptor> lastDescriptors = Map.of();
 
     /** Creates a new composite engine plugin. */
-    public CompositeEnginePlugin() {}
+    public CompositeDataFormatPlugin() {}
 
     @Override
     public void loadExtensions(ExtensionLoader loader) {
@@ -135,16 +140,42 @@ public DataFormat getDataFormat() {
     }
 
     @Override
-    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings) {
+    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy) {
+        Map<String, FormatChecksumStrategy> strategies = new HashMap<>();
+        for (Map.Entry<String, DataFormatDescriptor> entry : lastDescriptors.entrySet()) {
+            strategies.put(entry.getKey(), entry.getValue().getChecksumStrategy());
+        }
         return new CompositeIndexingExecutionEngine(
             dataFormatPlugins,
             settings.indexSettings(),
             settings.mapperService(),
             settings.shardPath(),
-            settings.committer()
+            settings.committer(),
+            strategies
         );
     }
 
+    @Override
+    public Map<String, DataFormatDescriptor> getFormatDescriptors(IndexSettings indexSettings) {
+        Settings settings = indexSettings.getSettings();
+        String primaryFormatName = PRIMARY_DATA_FORMAT.get(settings);
+        List<String> secondaryFormatNames = SECONDARY_DATA_FORMATS.get(settings);
+
+        Map<String, DataFormatDescriptor> descriptors = new HashMap<>();
+        DataFormatPlugin primaryPlugin = dataFormatPlugins.get(primaryFormatName);
+        if (primaryPlugin != null) {
+            descriptors.putAll(primaryPlugin.getFormatDescriptors(indexSettings));
+        }
+        for (String secondaryName : secondaryFormatNames) {
+            DataFormatPlugin secondaryPlugin = dataFormatPlugins.get(secondaryName);
+            if (secondaryPlugin != null) {
+                descriptors.putAll(secondaryPlugin.getFormatDescriptors(indexSettings));
+            }
+        }
+        lastDescriptors = Map.copyOf(descriptors);
+        return lastDescriptors;
+    }
+
     /**
      * Returns the discovered data format plugins keyed by format name.
      *

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java

@@ -33,6 +33,7 @@
 import org.opensearch.index.engine.exec.coord.CatalogSnapshotManager;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.shard.ShardPath;
+import org.opensearch.index.store.FormatChecksumStrategy;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -85,11 +86,13 @@ public class CompositeIndexingExecutionEngine implements IndexingExecutionEngine
      * The writer pool is created internally and initialized with a writer supplier
      * that creates {@link CompositeWriter} instances bound to this engine.
      *
-     * @param dataFormatPlugins the discovered data format plugins keyed by format name
-     * @param indexSettings the index settings containing composite configuration
-     * @param mapperService the mapper service for field mapping resolution
-     * @param shardPath the shard path for file storage
-     * @param committer the committer for durable catalog snapshot persistence during flush
+     * @param dataFormatPlugins   the discovered data format plugins keyed by format name
+     * @param indexSettings       the index settings containing composite configuration
+     * @param mapperService       the mapper service for field mapping resolution
+     * @param shardPath           the shard path for file storage
+     * @param committer           the committer for durable catalog snapshot persistence during flush
+     * @param checksumStrategies  per-format checksum strategies from the directory, keyed by format name.
+     *                            May be null or empty if the directory is not yet available.
      * @throws IllegalArgumentException if any configured format is not registered
      * @throws IllegalStateException if committer is null
      */
@@ -98,7 +101,8 @@ public CompositeIndexingExecutionEngine(
         IndexSettings indexSettings,
         MapperService mapperService,
         ShardPath shardPath,
-        Committer committer
+        Committer committer,
+        Map<String, FormatChecksumStrategy> checksumStrategies
     ) {
         Objects.requireNonNull(dataFormatPlugins, "dataFormatPlugins must not be null");
         Objects.requireNonNull(indexSettings, "indexSettings must not be null");
@@ -108,22 +112,23 @@ public CompositeIndexingExecutionEngine(
 
         Settings settings = indexSettings.getSettings();
 
-        String primaryFormatName = CompositeEnginePlugin.PRIMARY_DATA_FORMAT.get(settings);
-        List<String> secondaryFormatNames = CompositeEnginePlugin.SECONDARY_DATA_FORMATS.get(settings);
+        String primaryFormatName = CompositeDataFormatPlugin.PRIMARY_DATA_FORMAT.get(settings);
+        List<String> secondaryFormatNames = CompositeDataFormatPlugin.SECONDARY_DATA_FORMATS.get(settings);
 
         validateFormatsRegistered(dataFormatPlugins, primaryFormatName, secondaryFormatNames);
 
+        Map<String, FormatChecksumStrategy> strategies = checksumStrategies != null ? checksumStrategies : Map.of();
         IndexingEngineConfig engineSettings = new IndexingEngineConfig(committer, mapperService, shardPath, indexSettings, null);
 
         List<DataFormat> allFormats = new ArrayList<>();
         DataFormatPlugin primaryPlugin = dataFormatPlugins.get(primaryFormatName);
-        this.primaryEngine = primaryPlugin.indexingEngine(engineSettings);
+        this.primaryEngine = primaryPlugin.indexingEngine(engineSettings, strategies.get(primaryFormatName));
         allFormats.add(primaryPlugin.getDataFormat());
 
         List<IndexingExecutionEngine<?, ?>> secondaries = new ArrayList<>();
         for (String secondaryName : secondaryFormatNames) {
             DataFormatPlugin secondaryPlugin = dataFormatPlugins.get(secondaryName);
-            secondaries.add(secondaryPlugin.indexingEngine(engineSettings));
+            secondaries.add(secondaryPlugin.indexingEngine(engineSettings, strategies.get(secondaryName)));
             allFormats.add(secondaryPlugin.getDataFormat());
         }
         this.secondaryEngines = Set.copyOf(secondaries);