Share single FormatChecksumStrategy instance per shard between engine and store

sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LucenePlugin.java

@@ -22,7 +22,6 @@
 import org.opensearch.index.engine.exec.EngineReaderManager;
 import org.opensearch.index.engine.exec.commit.Committer;
 import org.opensearch.index.engine.exec.commit.CommitterFactory;
-import org.opensearch.index.store.FormatChecksumStrategy;
 import org.opensearch.plugins.EnginePlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.SearchBackEndPlugin;
@@ -66,15 +65,11 @@ public DataFormat getDataFormat() {
      * Requires the committer to be a {@link LuceneCommitter}.
      *
      * @param indexingEngineConfig the engine configuration containing committer, mapper service, and store
-     * @param checksumStrategy     the checksum strategy for the format (unused by Lucene)
      * @return a new Lucene indexing execution engine
      * @throws IllegalStateException if the committer is not a {@link LuceneCommitter}
      */
     @Override
-    public IndexingExecutionEngine<?, ?> indexingEngine(
-        IndexingEngineConfig indexingEngineConfig,
-        FormatChecksumStrategy checksumStrategy
-    ) {
+    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig indexingEngineConfig) {
         Committer committer = indexingEngineConfig.committer();
         if (committer instanceof LuceneCommitter luceneCommitter) {
             return new LuceneIndexingExecutionEngine(

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

@@ -94,6 +94,7 @@ private CommitterConfig createCommitterConfig() throws IOException {
             null,
             null,
             null,
+            null,
             null
         );
         return new CommitterConfig(engineConfig);

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

@@ -123,6 +123,7 @@ private LuceneCommitter createCommitter() throws IOException {
             null,
             null,
             null,
+            null,
             null
         );
         CommitterConfig settings = new CommitterConfig(engineConfig);

sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeMergeIT.java

@@ -41,6 +41,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 /**
  * Integration tests for composite merge operations across single and multiple data format engines.
@@ -66,8 +67,8 @@ public MockParquetDataFormatPlugin() {
         }
 
         @Override
-        public Map<String, DataFormatDescriptor> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) {
-            return Map.of("parquet", new DataFormatDescriptor("parquet", new PrecomputedChecksumStrategy()));
+        public Map<String, Supplier<DataFormatDescriptor>> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) {
+            return Map.of("parquet", () -> new DataFormatDescriptor("parquet", new PrecomputedChecksumStrategy()));
         }
 
         @Override

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

@@ -20,14 +20,14 @@
 import org.opensearch.index.engine.dataformat.DataFormatRegistry;
 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;
 
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 
 /**
  * Sandbox plugin that provides a {@link CompositeIndexingExecutionEngine} for
@@ -92,35 +92,33 @@ public DataFormat getDataFormat() {
     }
 
     @Override
-    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy) {
-        Map<String, FormatChecksumStrategy> strategies = new HashMap<>();
-        for (Map.Entry<String, DataFormatDescriptor> entry : getFormatDescriptors(settings.indexSettings(), settings.registry())
-            .entrySet()) {
-            strategies.put(entry.getKey(), entry.getValue().getChecksumStrategy());
-        }
+    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings) {
         return new CompositeIndexingExecutionEngine(
             settings.indexSettings(),
             settings.mapperService(),
             settings.committer(),
             settings.registry(),
             settings.store(),
-            strategies
+            settings.checksumStrategies()
         );
     }
 
     @Override
-    public Map<String, DataFormatDescriptor> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry dataFormatRegistry) {
+    public Map<String, Supplier<DataFormatDescriptor>> getFormatDescriptors(
+        IndexSettings indexSettings,
+        DataFormatRegistry dataFormatRegistry
+    ) {
         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<>();
+        Map<String, Supplier<DataFormatDescriptor>> descriptors = new HashMap<>();
         if (primaryFormatName != null) {
-            descriptors.putAll(dataFormatRegistry.getFormatDescriptors(indexSettings));
+            descriptors.putAll(dataFormatRegistry.getFormatDescriptors(indexSettings, dataFormatRegistry.format(primaryFormatName)));
         }
         for (String secondaryName : secondaryFormatNames) {
             if (secondaryName != null) {
-                descriptors.putAll(dataFormatRegistry.getFormatDescriptors(indexSettings));
+                descriptors.putAll(dataFormatRegistry.getFormatDescriptors(indexSettings, dataFormatRegistry.format(secondaryName)));
             }
         }
         return Map.copyOf(descriptors);

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

@@ -115,7 +115,14 @@ public CompositeIndexingExecutionEngine(
         validateFormatsRegistered(dataFormatRegistry, primaryFormatName, secondaryFormatNames);
 
         Map<String, FormatChecksumStrategy> strategies = checksumStrategies != null ? checksumStrategies : Map.of();
-        IndexingEngineConfig engineSettings = new IndexingEngineConfig(committer, mapperService, indexSettings, store, dataFormatRegistry);
+        IndexingEngineConfig engineSettings = new IndexingEngineConfig(
+            committer,
+            mapperService,
+            indexSettings,
+            store,
+            dataFormatRegistry,
+            strategies
+        );
 
         List<DataFormat> allFormats = new ArrayList<>();
         DataFormat primaryFormat = dataFormatRegistry.format(primaryFormatName);

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

@@ -11,11 +11,13 @@
 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.DataFormatRegistry;
 import org.opensearch.test.OpenSearchTestCase;
 
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -60,24 +62,27 @@ public void testGetFormatDescriptorsDelegatestoPlugins() {
         IndexSettings indexSettings = new IndexSettings(indexMetadata, Settings.EMPTY);
 
         DataFormatRegistry registry = mock(DataFormatRegistry.class);
-        when(registry.format("parquet")).thenReturn(CompositeTestHelper.stubFormat("parquet", 2, java.util.Set.of()));
-        when(registry.getFormatDescriptors(indexSettings)).thenReturn(
+        DataFormat parquetFormat = CompositeTestHelper.stubFormat("parquet", 2, java.util.Set.of());
+        when(registry.format("parquet")).thenReturn(parquetFormat);
+        when(registry.format("lucene")).thenReturn(CompositeTestHelper.stubFormat("lucene", 1, java.util.Set.of()));
+        when(registry.getFormatDescriptors(indexSettings, parquetFormat)).thenReturn(
             Map.of(
                 "parquet",
-                new org.opensearch.index.engine.dataformat.DataFormatDescriptor(
-                    "parquet",
-                    new org.opensearch.index.store.checksum.GenericCRC32ChecksumHandler()
-                )
+                (Supplier<
+                    org.opensearch.index.engine.dataformat.DataFormatDescriptor>) () -> new org.opensearch.index.engine.dataformat.DataFormatDescriptor(
+                        "parquet",
+                        new org.opensearch.index.store.checksum.GenericCRC32ChecksumHandler()
+                    )
             )
         );
 
-        Map<String, org.opensearch.index.engine.dataformat.DataFormatDescriptor> descriptors = plugin.getFormatDescriptors(
+        Map<String, Supplier<org.opensearch.index.engine.dataformat.DataFormatDescriptor>> descriptors = plugin.getFormatDescriptors(
             indexSettings,
             registry
         );
         assertEquals(1, descriptors.size());
         assertTrue(descriptors.containsKey("parquet"));
-        assertEquals("parquet", descriptors.get("parquet").getFormatName());
+        assertEquals("parquet", descriptors.get("parquet").get().getFormatName());
     }
 
     public void testGetFormatDescriptorsEmptyWhenNoPluginsMatch() {
@@ -94,7 +99,7 @@ public void testGetFormatDescriptorsEmptyWhenNoPluginsMatch() {
             .build();
         IndexSettings indexSettings = new IndexSettings(indexMetadata, Settings.EMPTY);
 
-        Map<String, org.opensearch.index.engine.dataformat.DataFormatDescriptor> descriptors = plugin.getFormatDescriptors(
+        Map<String, Supplier<org.opensearch.index.engine.dataformat.DataFormatDescriptor>> descriptors = plugin.getFormatDescriptors(
             indexSettings,
             registry
         );

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

@@ -76,7 +76,7 @@ public void testConstructorThrowsWhenSecondaryFormatNotRegistered() {
         when(registry.getRegisteredFormats()).thenReturn(Set.of(CompositeTestHelper.stubFormat("lucene", 1, Set.of())));
         when(registry.getIndexingEngine(any(), any())).thenAnswer(invocation -> {
             DataFormatPlugin plugin = CompositeTestHelper.stubPlugin("lucene", 1);
-            return plugin.indexingEngine(null, null);
+            return plugin.indexingEngine(null);
         });
 
         Settings settings = Settings.builder()

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

@@ -31,7 +31,6 @@
 import org.opensearch.index.engine.exec.commit.Committer;
 import org.opensearch.index.engine.exec.commit.IndexStoreProvider;
 import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
-import org.opensearch.index.store.FormatChecksumStrategy;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -72,7 +71,7 @@ static CompositeIndexingExecutionEngine createStubEngine(String primaryName, Str
         when(registry.getIndexingEngine(any(), any())).thenAnswer(invocation -> {
             DataFormat format = invocation.getArgument(1);
             DataFormatPlugin plugin = plugins.get(format.name());
-            return plugin.indexingEngine(null, null);
+            return plugin.indexingEngine(null);
         });
 
         Settings.Builder settingsBuilder = Settings.builder()
@@ -101,7 +100,7 @@ public DataFormat getDataFormat() {
             }
 
             @Override
-            public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy) {
+            public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings) {
                 return new StubIndexingExecutionEngine(format);
             }
         };
@@ -116,7 +115,7 @@ public DataFormat getDataFormat() {
             }
 
             @Override
-            public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy) {
+            public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig settings) {
                 return new StubIndexingExecutionEngine(format);
             }
         };

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java

@@ -24,7 +24,6 @@
 import org.opensearch.index.engine.dataformat.DataFormatRegistry;
 import org.opensearch.index.engine.dataformat.IndexingEngineConfig;
 import org.opensearch.index.engine.dataformat.IndexingExecutionEngine;
-import org.opensearch.index.store.FormatChecksumStrategy;
 import org.opensearch.index.store.PrecomputedChecksumStrategy;
 import org.opensearch.parquet.engine.ParquetDataFormat;
 import org.opensearch.parquet.engine.ParquetIndexingEngine;
@@ -52,10 +51,9 @@
  * {@link #createComponents} and passes them to the per-shard
  * {@link ParquetIndexingEngine} instances created in {@link #indexingEngine}.
  *
- * <p>The descriptor provides a {@link PrecomputedChecksumStrategy} that the directory
- * holds at construction time. The {@link ParquetIndexingEngine} receives the same
- * strategy instance from the directory via
- * {@link org.opensearch.index.store.DataFormatAwareStoreDirectory#getChecksumStrategy},
+ * <p>The descriptor provides a {@link PrecomputedChecksumStrategy} that is created once
+ * per shard during initialization. The same strategy instance is shared between the
+ * directory and the {@link ParquetIndexingEngine} via the checksum strategies map,
  * so pre-computed CRC32 values registered during write are directly visible to the
  * upload path — no post-construction wiring needed.
  *
@@ -99,23 +97,23 @@ public DataFormat getDataFormat() {
     }
 
     @Override
-    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig engineConfig, FormatChecksumStrategy checksumStrategy) {
+    public IndexingExecutionEngine<?, ?> indexingEngine(IndexingEngineConfig engineConfig) {
         return new ParquetIndexingEngine(
             settings,
             dataFormat,
             engineConfig.store().shardPath(),
             () -> ArrowSchemaBuilder.getSchema(engineConfig.mapperService()),
             engineConfig.indexSettings(),
             threadPool,
-            checksumStrategy
+            engineConfig.checksumStrategies().get(ParquetDataFormat.PARQUET_DATA_FORMAT_NAME)
         );
     }
 
     @Override
-    public Map<String, DataFormatDescriptor> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) {
+    public Map<String, Supplier<DataFormatDescriptor>> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) {
         return Map.of(
             ParquetDataFormat.PARQUET_DATA_FORMAT_NAME,
-            new DataFormatDescriptor(ParquetDataFormat.PARQUET_DATA_FORMAT_NAME, new PrecomputedChecksumStrategy())
+            () -> new DataFormatDescriptor(ParquetDataFormat.PARQUET_DATA_FORMAT_NAME, new PrecomputedChecksumStrategy())
         );
     }
 

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -734,6 +734,7 @@ public static final IndexShard newIndexShard(
             clusterService.getClusterApplierService(),
             MergedSegmentPublisher.EMPTY,
             ReferencedSegmentsPublisher.EMPTY,
+            Collections.emptyMap(),
             null // TODO
         );
     }

server/src/main/java/org/opensearch/index/IndexService.java

@@ -100,6 +100,7 @@
 import org.opensearch.index.similarity.SimilarityService;
 import org.opensearch.index.store.DataFormatAwareStoreDirectory;
 import org.opensearch.index.store.DataFormatAwareStoreDirectoryFactory;
+import org.opensearch.index.store.FormatChecksumStrategy;
 import org.opensearch.index.store.RemoteSegmentStoreDirectoryFactory;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.remote.filecache.FileCache;
@@ -773,6 +774,10 @@ protected void closeInternal() {
             }
 
             Directory directory = null;
+            Map<String, FormatChecksumStrategy> checksumStrategies = Collections.emptyMap();
+            if (this.indexSettings.isPluggableDataFormatEnabled() && dataFormatRegistry != null) {
+                checksumStrategies = dataFormatRegistry.createChecksumStrategies(this.indexSettings);
+            }
             if (FeatureFlags.isEnabled(FeatureFlags.WRITABLE_WARM_INDEX_SETTING) &&
             // TODO : Need to remove this check after support for hot indices is added in Composite Directory
                 this.indexSettings.isWarmIndex()) {
@@ -788,7 +793,7 @@ protected void closeInternal() {
                 directory = directoryFactory.newDirectory(this.indexSettings, path);
             } else {
                 // Will be enabled in case of formatAware indices.
-                directory = createDataFormatAwareStoreDirectory(shardId, path);
+                directory = createDataFormatAwareStoreDirectory(shardId, path, checksumStrategies);
             }
             store = storeFactory.newStore(
                 shardId,
@@ -839,6 +844,7 @@ protected void closeInternal() {
                 clusterService.getClusterApplierService(),
                 this.indexSettings.isSegRepEnabledOrRemoteNode() ? mergedSegmentPublisher : null,
                 this.indexSettings.isSegRepEnabledOrRemoteNode() ? referencedSegmentsPublisher : null,
+                checksumStrategies,
                 dataFormatRegistry
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
@@ -1344,15 +1350,19 @@ public boolean isForceExecution() {
      * Creates DataFormatAwareStoreDirectory using the factory if available, otherwise fallback to Store's internal creation.
      * This method centralizes the directory creation logic and enables plugin-based format discovery.
      */
-    private DataFormatAwareStoreDirectory createDataFormatAwareStoreDirectory(ShardId shardId, ShardPath shardPath) throws IOException {
+    private DataFormatAwareStoreDirectory createDataFormatAwareStoreDirectory(
+        ShardId shardId,
+        ShardPath shardPath,
+        Map<String, FormatChecksumStrategy> checksumStrategies
+    ) throws IOException {
         if (dataFormatAwareStoreDirectoryFactory != null) {
             logger.debug("Using DataFormatAwareStoreDirectoryFactory to create directory for shard path: {}", shardPath);
             return dataFormatAwareStoreDirectoryFactory.newDataFormatAwareStoreDirectory(
                 indexSettings,
                 shardId,
                 shardPath,
                 directoryFactory,
-                dataFormatRegistry
+                checksumStrategies
             );
         }
 

server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java

@@ -238,7 +238,8 @@ public DataFormatAwareEngine(EngineConfig engineConfig) {
                     config().getMapperService(),
                     config().getIndexSettings(),
                     config().getStore(),
-                    registry
+                    registry,
+                    config().getChecksumStrategies()
                 ),
                 registry.format(config().getIndexSettings().pluggableDataFormat())
             );