diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LucenePlugin.java b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LucenePlugin.java index 3c2de857d9449..88a3c569f53ae 100644 --- a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LucenePlugin.java +++ b/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( diff --git a/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneCommitterTests.java b/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneCommitterTests.java index 6a7d6c0844afd..05bdca5c79170 100644 --- a/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneCommitterTests.java +++ b/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); diff --git a/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngineTests.java b/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngineTests.java index 08b6c6027b855..a84d1c3b782b9 100644 --- a/sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngineTests.java +++ b/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); diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeMergeIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeMergeIT.java index 4faa5948d7c4e..3ceda94a96d06 100644 --- a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeMergeIT.java +++ b/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 getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) { - return Map.of("parquet", new DataFormatDescriptor("parquet", new PrecomputedChecksumStrategy())); + public Map> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) { + return Map.of("parquet", () -> new DataFormatDescriptor("parquet", new PrecomputedChecksumStrategy())); } @Override diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java index d1dc6463b396c..7ec61ceded267 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java @@ -20,7 +20,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.plugins.ExtensiblePlugin; import org.opensearch.plugins.Plugin; @@ -28,6 +27,7 @@ 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 strategies = new HashMap<>(); - for (Map.Entry 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 getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry dataFormatRegistry) { + public Map> getFormatDescriptors( + IndexSettings indexSettings, + DataFormatRegistry dataFormatRegistry + ) { Settings settings = indexSettings.getSettings(); String primaryFormatName = PRIMARY_DATA_FORMAT.get(settings); List secondaryFormatNames = SECONDARY_DATA_FORMATS.get(settings); - Map descriptors = new HashMap<>(); + Map> 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); diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java index 831ccfb0971c8..4dc8b3f8165b5 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java @@ -115,7 +115,14 @@ public CompositeIndexingExecutionEngine( validateFormatsRegistered(dataFormatRegistry, primaryFormatName, secondaryFormatNames); Map 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 allFormats = new ArrayList<>(); DataFormat primaryFormat = dataFormatRegistry.format(primaryFormatName); diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeDataFormatPluginTests.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeDataFormatPluginTests.java index 4e7dd4cdcea75..c48306473197e 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeDataFormatPluginTests.java +++ b/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 descriptors = plugin.getFormatDescriptors( + Map> 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 descriptors = plugin.getFormatDescriptors( + Map> descriptors = plugin.getFormatDescriptors( indexSettings, registry ); diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java index 59bd20c7a231b..fc6263f6f8b25 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeIndexingExecutionEngineTests.java +++ b/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() diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java index f15dadd01fc4a..236e52739bc66 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/CompositeTestHelper.java +++ b/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); } }; diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java index fc5da5742adf6..33bb655257a78 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java +++ b/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}. * - *

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}, + *

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,7 +97,7 @@ public DataFormat getDataFormat() { } @Override - public IndexingExecutionEngine indexingEngine(IndexingEngineConfig engineConfig, FormatChecksumStrategy checksumStrategy) { + public IndexingExecutionEngine indexingEngine(IndexingEngineConfig engineConfig) { return new ParquetIndexingEngine( settings, dataFormat, @@ -107,15 +105,15 @@ public DataFormat getDataFormat() { () -> ArrowSchemaBuilder.getSchema(engineConfig.mapperService()), engineConfig.indexSettings(), threadPool, - checksumStrategy + engineConfig.checksumStrategies().get(ParquetDataFormat.PARQUET_DATA_FORMAT_NAME) ); } @Override - public Map getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) { + public Map> 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()) ); } diff --git a/server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java b/server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java index 82a812cb4bb56..ea63e14cfcb3f 100644 --- a/server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java +++ b/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 ); } diff --git a/server/src/main/java/org/opensearch/index/IndexService.java b/server/src/main/java/org/opensearch/index/IndexService.java index 65bcfdcc565c5..12c611bcb2b15 100644 --- a/server/src/main/java/org/opensearch/index/IndexService.java +++ b/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 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,7 +1350,11 @@ 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 checksumStrategies + ) throws IOException { if (dataFormatAwareStoreDirectoryFactory != null) { logger.debug("Using DataFormatAwareStoreDirectoryFactory to create directory for shard path: {}", shardPath); return dataFormatAwareStoreDirectoryFactory.newDataFormatAwareStoreDirectory( @@ -1352,7 +1362,7 @@ private DataFormatAwareStoreDirectory createDataFormatAwareStoreDirectory(ShardI shardId, shardPath, directoryFactory, - dataFormatRegistry + checksumStrategies ); } diff --git a/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java b/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java index bfb919cba8e6d..0906953e68477 100644 --- a/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java +++ b/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()) ); diff --git a/server/src/main/java/org/opensearch/index/engine/EngineConfig.java b/server/src/main/java/org/opensearch/index/engine/EngineConfig.java index 6bf341852bfa1..78e319bfafc3b 100644 --- a/server/src/main/java/org/opensearch/index/engine/EngineConfig.java +++ b/server/src/main/java/org/opensearch/index/engine/EngineConfig.java @@ -62,6 +62,7 @@ import org.opensearch.index.mapper.ParsedDocument; import org.opensearch.index.merge.MergedSegmentTransferTracker; import org.opensearch.index.seqno.RetentionLeases; +import org.opensearch.index.store.FormatChecksumStrategy; import org.opensearch.index.store.Store; import org.opensearch.index.translog.InternalTranslogFactory; import org.opensearch.index.translog.TranslogConfig; @@ -70,8 +71,10 @@ import org.opensearch.indices.IndexingMemoryController; import org.opensearch.threadpool.ThreadPool; +import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.function.BooleanSupplier; @@ -123,6 +126,7 @@ public final class EngineConfig { private final DataFormatRegistry dataFormatRegistry; private final MapperService mapperService; private final CommitterFactory committerFactory; + private final Map checksumStrategies; /** * A supplier of the outstanding retention leases. This is used during merged operations to determine which operations that have been @@ -316,6 +320,7 @@ private EngineConfig(Builder builder) { this.dataFormatRegistry = builder.dataFormatRegistry; this.mapperService = builder.mapperService; this.committerFactory = builder.committerFactory; + this.checksumStrategies = builder.checksumStrategies; } /** @@ -655,6 +660,10 @@ public CommitterFactory getCommitterFactory() { return this.committerFactory; } + public Map getChecksumStrategies() { + return this.checksumStrategies; + } + /** * Builder for EngineConfig class * @@ -696,6 +705,7 @@ public static class Builder { private DataFormatRegistry dataFormatRegistry; private MapperService mapperService; private CommitterFactory committerFactory; + private Map checksumStrategies = Collections.emptyMap(); public Builder shardId(ShardId shardId) { this.shardId = shardId; @@ -867,6 +877,11 @@ public Builder committerFactory(CommitterFactory committerFactory) { return this; } + public Builder checksumStrategies(Map checksumStrategies) { + this.checksumStrategies = checksumStrategies; + return this; + } + public EngineConfig build() { return new EngineConfig(this); } diff --git a/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java b/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java index adbeee8ab29c6..b9d5be2ed5f2c 100644 --- a/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java +++ b/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java @@ -34,6 +34,7 @@ import org.opensearch.index.mapper.MapperService; import org.opensearch.index.merge.MergedSegmentTransferTracker; import org.opensearch.index.seqno.RetentionLeases; +import org.opensearch.index.store.FormatChecksumStrategy; import org.opensearch.index.store.Store; import org.opensearch.index.translog.TranslogConfig; import org.opensearch.index.translog.TranslogDeletionPolicyFactory; @@ -47,6 +48,7 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.function.BooleanSupplier; import java.util.function.LongSupplier; @@ -184,7 +186,8 @@ public EngineConfig newEngineConfig( ClusterApplierService clusterApplierService, MergedSegmentTransferTracker mergedSegmentTransferTracker, DataFormatRegistry dataFormatRegistry, - MapperService mapperService + MapperService mapperService, + Map checksumStrategies ) { CodecService codecServiceToUse = codecService; if (codecService == null && this.codecServiceFactory != null) { @@ -225,6 +228,7 @@ public EngineConfig newEngineConfig( .dataFormatRegistry(dataFormatRegistry) .mapperService(mapperService) .committerFactory(committerFactory) + .checksumStrategies(checksumStrategies) .build(); } diff --git a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatDescriptor.java b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatDescriptor.java index 0df1498a23b41..b88be06567401 100644 --- a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatDescriptor.java +++ b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatDescriptor.java @@ -19,7 +19,7 @@ *

The checksum strategy here is the default fallback — a full-file scan. * At runtime, the {@link IndexingExecutionEngine} may override this with a more * efficient strategy (e.g., {@link org.opensearch.index.store.PrecomputedChecksumStrategy}) - * via {@link org.opensearch.index.store.DataFormatAwareStoreDirectory#registerChecksumStrategy}. + * via the shared checksum strategies map created during shard initialization. * * @opensearch.experimental */ diff --git a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatPlugin.java b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatPlugin.java index ac34836f97e67..00835289bf122 100644 --- a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatPlugin.java +++ b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatPlugin.java @@ -10,9 +10,9 @@ import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.index.IndexSettings; -import org.opensearch.index.store.FormatChecksumStrategy; import java.util.Map; +import java.util.function.Supplier; /** * Plugin interface for providing custom data format implementations. @@ -35,23 +35,23 @@ public interface DataFormatPlugin { * Creates the indexing engine for the data format. This should be instantiated per shard. * * @param settings the engine initialization settings - * @param checksumStrategy the checksum strategy owned by the directory for this format, - * or null if not available. Engines that pre-compute checksums - * during write should register into this instance so the upload - * path can retrieve them in O(1). * @return the indexing execution engine instance */ - IndexingExecutionEngine indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy); + IndexingExecutionEngine indexingEngine(IndexingEngineConfig settings); /** - * Returns format descriptors for this plugin, filtered by the given index settings. - * Each entry maps a format name to its {@link DataFormatDescriptor} containing the - * default checksum strategy and format name. + * Returns format descriptor suppliers for this plugin, filtered by the given index settings. + * Each entry maps a format name to a {@link Supplier} of its {@link DataFormatDescriptor}, + * deferring descriptor object creation until the descriptor is actually needed. + * Callers that only need format names can use {@code keySet()} without triggering creation. * * @param indexSettings the index settings used to determine active formats - * @return map of format name to descriptor + * @return map of format name to descriptor supplier */ - default Map getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry dataFormatRegistry) { + default Map> getFormatDescriptors( + IndexSettings indexSettings, + DataFormatRegistry dataFormatRegistry + ) { return Map.of(); } } diff --git a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatRegistry.java b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatRegistry.java index 5a6254b0ce5ed..f61b47bc39b5c 100644 --- a/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatRegistry.java +++ b/server/src/main/java/org/opensearch/index/engine/dataformat/DataFormatRegistry.java @@ -19,11 +19,13 @@ import org.opensearch.plugins.SearchBackEndPlugin; import java.io.IOException; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Supplier; import java.util.stream.Collectors; /** @@ -35,9 +37,6 @@ @ExperimentalApi public class DataFormatRegistry { - /** Index setting name that specifies the active pluggable data format. */ - public static final String PLUGGABLE_DATAFORMAT_SETTING = "pluggable_dataformat"; - /** Map from data format to the plugin that provides its indexing engine. */ private final Map dataFormatPluginRegistry; @@ -97,10 +96,7 @@ public DataFormatRegistry(PluginsService pluginsService) { if (plugin == null) { throw new IllegalArgumentException("No plugin registered for DataFormat [" + format.name() + "]"); } - Map descriptors = plugin.getFormatDescriptors(settings.indexSettings(), this); - DataFormatDescriptor descriptor = descriptors.get(format.name()); - FormatChecksumStrategy checksumStrategy = descriptor != null ? descriptor.getChecksumStrategy() : null; - return plugin.indexingEngine(settings, checksumStrategy); + return plugin.indexingEngine(settings); } public DataFormat format(String name) { @@ -140,16 +136,17 @@ public Set getRegisteredFormats() { } /** - * Returns format descriptors for the active data format of the given index. + * Returns format descriptor suppliers for the active data format of the given index. * Resolves the data format from index settings via the {@code pluggable_dataformat} setting, * then delegates to {@link DataFormatPlugin#getFormatDescriptors(IndexSettings, DataFormatRegistry)}. + * Callers that only need format names can use {@code keySet()} without triggering descriptor creation. * * @param indexSettings the index settings used to determine the active data format - * @return unmodifiable map of format name to descriptor, or empty map if no pluggable data format is configured + * @return map of format name to descriptor supplier, or empty map if no pluggable data format is configured */ - public Map getFormatDescriptors(IndexSettings indexSettings) { - String dataformatName = indexSettings.getSettings().get(PLUGGABLE_DATAFORMAT_SETTING); - if (dataformatName != null) { + public Map> getFormatDescriptors(IndexSettings indexSettings) { + String dataformatName = indexSettings.pluggableDataFormat(); + if (dataformatName != null && dataformatName.isEmpty() == false) { DataFormat format = dataFormats.get(dataformatName); if (format != null) { DataFormatPlugin plugin = dataFormatPluginRegistry.get(format); @@ -161,6 +158,44 @@ public Map getFormatDescriptors(IndexSettings inde return Map.of(); } + /** + * Returns format descriptor suppliers for a specific data format, bypassing the + * {@code pluggable_dataformat} index setting lookup. This is used by composite + * plugins to resolve child format descriptors without recursion. + * + * @param indexSettings the index settings + * @param dataFormat the specific data format to get descriptors for + * @return map of format name to descriptor supplier, or empty map if the format is not registered + */ + public Map> getFormatDescriptors(IndexSettings indexSettings, DataFormat dataFormat) { + DataFormatPlugin plugin = dataFormatPluginRegistry.get(dataFormat); + if (plugin == null) { + return Map.of(); + } + return plugin.getFormatDescriptors(indexSettings, this); + } + + /** + * Creates checksum strategies for all formats of the given index, intended to be called + * once per shard during initialization. The returned map should be shared between the + * directory and the engine so that pre-computed checksums registered during write are + * visible to the upload path. + * + * @param indexSettings the index settings used to determine the active data format + * @return unmodifiable map of format name to checksum strategy + */ + public Map createChecksumStrategies(IndexSettings indexSettings) { + Map> descriptors = getFormatDescriptors(indexSettings); + Map strategies = new HashMap<>(); + for (Map.Entry> entry : descriptors.entrySet()) { + FormatChecksumStrategy strategy = entry.getValue().get().getChecksumStrategy(); + if (strategy != null) { + strategies.put(entry.getKey(), strategy); + } + } + return Collections.unmodifiableMap(strategies); + } + /** * Creates {@link EngineReaderManager} instances for all applicable data formats based on index settings/mappings. * Each reader manager is instantiated by applying the store provider and shard path to the factory registered diff --git a/server/src/main/java/org/opensearch/index/engine/dataformat/IndexingEngineConfig.java b/server/src/main/java/org/opensearch/index/engine/dataformat/IndexingEngineConfig.java index 0e417d9b5c3e7..e5cb8e58fe0e1 100644 --- a/server/src/main/java/org/opensearch/index/engine/dataformat/IndexingEngineConfig.java +++ b/server/src/main/java/org/opensearch/index/engine/dataformat/IndexingEngineConfig.java @@ -12,8 +12,11 @@ import org.opensearch.index.IndexSettings; import org.opensearch.index.engine.exec.commit.Committer; import org.opensearch.index.mapper.MapperService; +import org.opensearch.index.store.FormatChecksumStrategy; import org.opensearch.index.store.Store; +import java.util.Map; + /** * Initialization parameters for creating an {@link IndexingExecutionEngine} via * {@link DataFormatPlugin#indexingEngine}. Bundling parameters in a record avoids @@ -29,5 +32,5 @@ */ @ExperimentalApi public record IndexingEngineConfig(Committer committer, MapperService mapperService, IndexSettings indexSettings, Store store, - DataFormatRegistry registry) { + DataFormatRegistry registry, Map checksumStrategies) { } diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java index 042cdb0aba013..74b4b9f6d18be 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -184,6 +184,7 @@ import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.index.shard.PrimaryReplicaSyncer.ResyncTask; import org.opensearch.index.similarity.SimilarityService; +import org.opensearch.index.store.FormatChecksumStrategy; import org.opensearch.index.store.RemoteSegmentStoreDirectory; import org.opensearch.index.store.RemoteSegmentStoreDirectory.UploadedSegmentMetadata; import org.opensearch.index.store.RemoteStoreFileDownloader; @@ -416,6 +417,8 @@ Runnable getGlobalCheckpointSyncer() { private final DataFormatRegistry dataFormatRegistry; + private final Map checksumStrategies; + @InternalApi public IndexShard( final ShardRouting shardRouting, @@ -456,6 +459,7 @@ public IndexShard( final ClusterApplierService clusterApplierService, @Nullable final MergedSegmentPublisher mergedSegmentPublisher, @Nullable final ReferencedSegmentsPublisher referencedSegmentsPublisher, + final Map checksumStrategies, @Nullable final DataFormatRegistry dataFormatRegistry ) throws IOException { super(shardRouting.shardId(), indexSettings); @@ -611,6 +615,7 @@ public boolean shouldCache(Query query) { } } this.dataFormatRegistry = dataFormatRegistry; + this.checksumStrategies = checksumStrategies; } /** @@ -634,6 +639,10 @@ public Store store() { return this.store; } + public Map getChecksumStrategies() { + return checksumStrategies; + } + public boolean isMigratingToRemote() { // set it true only if shard is remote, but index setting doesn't say so return shardMigrationState == REMOTE_MIGRATING_UNSEEDED || shardMigrationState == REMOTE_MIGRATING_SEEDED; @@ -4523,7 +4532,8 @@ public void afterRefresh(boolean didRefresh) { clusterApplierService, mergedSegmentTransferTracker, dataFormatRegistry, - mapperService + mapperService, + checksumStrategies ); } diff --git a/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectory.java b/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectory.java index 24065799c537e..6413e84933b35 100644 --- a/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectory.java +++ b/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectory.java @@ -16,9 +16,6 @@ import org.apache.lucene.store.IndexInput; import org.apache.lucene.store.IndexOutput; import org.opensearch.common.annotation.PublicApi; -import org.opensearch.index.IndexSettings; -import org.opensearch.index.engine.dataformat.DataFormatDescriptor; -import org.opensearch.index.engine.dataformat.DataFormatRegistry; import org.opensearch.index.shard.ShardPath; import org.opensearch.index.store.checksum.GenericCRC32ChecksumHandler; import org.opensearch.index.store.checksum.LuceneChecksumHandler; @@ -81,32 +78,22 @@ public class DataFormatAwareStoreDirectory extends FilterDirectory { private static final FormatChecksumStrategy DEFAULT_CHECKSUM_STRATEGY = new GenericCRC32ChecksumHandler(); /** - * Constructs a DataFormatAwareStoreDirectory with a {@link DataFormatRegistry} for format-aware - * checksum calculation and other format-specific operations. + * Constructs a DataFormatAwareStoreDirectory with pre-built checksum strategies for + * format-aware checksum calculation and other format-specific operations. * * @param delegate the underlying FSDirectory (typically for <shard>/index/) * @param shardPath the shard path for resolving subdirectories - * @param dataFormatRegistry registry providing format-specific checksum handlers + * @param checksumStrategies pre-built checksum strategies keyed by format name */ - public DataFormatAwareStoreDirectory( - IndexSettings indexSettings, - Directory delegate, - ShardPath shardPath, - DataFormatRegistry dataFormatRegistry - ) { + public DataFormatAwareStoreDirectory(Directory delegate, ShardPath shardPath, Map checksumStrategies) { super(new SubdirectoryAwareDirectory(delegate, shardPath)); this.shardPath = shardPath; - Map descriptors = dataFormatRegistry.getFormatDescriptors(indexSettings); - this.checksumStrategies = new HashMap<>(); - for (Map.Entry entry : descriptors.entrySet()) { - this.checksumStrategies.put(entry.getKey(), entry.getValue().getChecksumStrategy()); - } + this.checksumStrategies = new HashMap<>(checksumStrategies); this.checksumStrategies.put(DEFAULT_FORMAT, new LuceneChecksumHandler()); - logger.debug( "Created DataFormatAwareStoreDirectory for shard {} with checksum strategies for formats: {}", shardPath.getShardId(), - checksumStrategies.keySet() + this.checksumStrategies.keySet() ); } @@ -246,24 +233,6 @@ public String calculateUploadChecksum(String name) throws IOException { return Long.toString(calculateChecksum(name)); } - /** - * Registers a {@link FormatChecksumStrategy} for a data format. - * Overrides any existing strategy - * - *

Use this to register strategies that support pre-computed checksums (e.g., - * {@link PrecomputedChecksumStrategy} for Parquet files whose CRC32 is computed - * during write by the Rust writer). - * - * @param format the data format name (e.g., "parquet") - * @param strategy the checksum strategy to use for this format - */ - public void registerChecksumStrategy(String format, FormatChecksumStrategy strategy) { - if (format != null && strategy != null) { - checksumStrategies.put(format, strategy); - logger.debug("Registered FormatChecksumStrategy for format [{}]", format); - } - } - /** * Returns the checksum strategy for the given format, or {@code null} if none is registered. * Engines use this to share the directory's strategy instance so that pre-computed diff --git a/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryFactory.java b/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryFactory.java index b633a00ca67eb..9c519e8001d93 100644 --- a/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryFactory.java +++ b/server/src/main/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryFactory.java @@ -11,11 +11,11 @@ import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.IndexSettings; -import org.opensearch.index.engine.dataformat.DataFormatRegistry; import org.opensearch.index.shard.ShardPath; import org.opensearch.plugins.IndexStorePlugin; import java.io.IOException; +import java.util.Map; /** * Factory interface for creating DataFormatAwareStoreDirectory instances. @@ -46,7 +46,7 @@ public interface DataFormatAwareStoreDirectoryFactory { * @param shardId the shard identifier * @param shardPath the path the shard is using for file storage * @param localDirectoryFactory the factory for creating the underlying local directory, respecting index store type configuration - * @param dataFormatRegistry registry of available data format plugins + * @param checksumStrategies pre-built checksum strategies keyed by format name * @return a new DataFormatAwareStoreDirectory instance supporting all discovered formats * @throws IOException if directory creation fails or resources cannot be allocated */ @@ -55,6 +55,6 @@ DataFormatAwareStoreDirectory newDataFormatAwareStoreDirectory( ShardId shardId, ShardPath shardPath, IndexStorePlugin.DirectoryFactory localDirectoryFactory, - DataFormatRegistry dataFormatRegistry + Map checksumStrategies ) throws IOException; } diff --git a/server/src/main/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactory.java b/server/src/main/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactory.java index 8e32942f5676d..8a53dfe696835 100644 --- a/server/src/main/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactory.java +++ b/server/src/main/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactory.java @@ -14,12 +14,12 @@ import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.IndexSettings; -import org.opensearch.index.engine.dataformat.DataFormatRegistry; import org.opensearch.index.shard.ShardPath; import org.opensearch.plugins.IndexStorePlugin; import java.io.IOException; import java.util.Locale; +import java.util.Map; /** * Default implementation of DataFormatAwareStoreDirectoryFactory that provides @@ -42,7 +42,7 @@ public class DefaultDataFormatAwareStoreDirectoryFactory implements DataFormatAw * @param shardId the shard identifier * @param shardPath the path the shard is using * @param localDirectoryFactory the factory for creating the underlying local directory - * @param dataFormatRegistry registry of available data format plugins + * @param checksumStrategies pre-built checksum strategies keyed by format name * @return a new DataFormatAwareStoreDirectory instance * @throws IOException if directory creation fails */ @@ -52,7 +52,7 @@ public DataFormatAwareStoreDirectory newDataFormatAwareStoreDirectory( ShardId shardId, ShardPath shardPath, IndexStorePlugin.DirectoryFactory localDirectoryFactory, - DataFormatRegistry dataFormatRegistry + Map checksumStrategies ) throws IOException { if (logger.isDebugEnabled()) { @@ -67,18 +67,13 @@ public DataFormatAwareStoreDirectory newDataFormatAwareStoreDirectory( // Delegate local directory creation to the configured DirectoryFactory Directory delegate = localDirectoryFactory.newDirectory(indexSettings, shardPath); - DataFormatAwareStoreDirectory directory = new DataFormatAwareStoreDirectory( - indexSettings, - delegate, - shardPath, - dataFormatRegistry - ); + DataFormatAwareStoreDirectory directory = new DataFormatAwareStoreDirectory(delegate, shardPath, checksumStrategies); if (logger.isDebugEnabled()) { logger.debug( "Successfully created DataFormatAwareStoreDirectory for shard: {} with registered formats: {}", shardPath.getShardId(), - dataFormatRegistry.getRegisteredFormats() + checksumStrategies.keySet() ); } diff --git a/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java b/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java index 3ec29f1c30841..28d7eccf9e64d 100644 --- a/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java +++ b/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java @@ -84,6 +84,7 @@ public void testCreateEngineConfigFromFactory() { null, null, null, + null, null ); @@ -197,6 +198,7 @@ public void testCreateCodecServiceFromFactory() { null, null, null, + null, null ); assertNotNull(config.getCodec()); diff --git a/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatPluginTests.java b/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatPluginTests.java index 18645af9421ac..69b2b220ec85d 100644 --- a/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatPluginTests.java +++ b/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatPluginTests.java @@ -87,9 +87,9 @@ public void testFullDataFormatLifecycle() throws IOException { mock(MapperService.class), new IndexSettings(IndexMetadata.builder("index").settings(settings).build(), settings), null, - null - ), - null + null, + Map.of() + ) ); assertEquals(format, engine.getDataFormat()); diff --git a/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatRegistryTests.java b/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatRegistryTests.java index 94ca8d727c56a..e9a144f78534e 100644 --- a/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatRegistryTests.java +++ b/server/src/test/java/org/opensearch/index/engine/dataformat/DataFormatRegistryTests.java @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.function.Supplier; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -146,7 +147,7 @@ public void testGetIndexingEngine() { DataFormatRegistry registry = new DataFormatRegistry(pluginsService); IndexingExecutionEngine engine = registry.getIndexingEngine( - new IndexingEngineConfig(null, mapperService, indexSettings, null, null), + new IndexingEngineConfig(null, mapperService, indexSettings, null, null, Map.of()), format ); assertNotNull(engine); @@ -162,7 +163,10 @@ public void testGetIndexingEngineForUnregisteredFormatThrows() { IllegalArgumentException e = expectThrows( IllegalArgumentException.class, - () -> registry.getIndexingEngine(new IndexingEngineConfig(null, mapperService, indexSettings, null, null), unregistered) + () -> registry.getIndexingEngine( + new IndexingEngineConfig(null, mapperService, indexSettings, null, null, Map.of()), + unregistered + ) ); assertTrue(e.getMessage().contains("unknown")); } @@ -286,4 +290,29 @@ public void testGetRegisteredFormatsIsUnmodifiable() { expectThrows(UnsupportedOperationException.class, () -> formats.add(new MockDataFormat("new", 1L, Set.of()))); } + + public void testGetFormatDescriptorsByDataFormatReturnsDescriptors() { + MockDataFormat format = new MockDataFormat("columnar", 100L, Set.of()); + MockDataFormatPlugin plugin = MockDataFormatPlugin.of(format); + MockSearchBackEndPlugin backEnd = new MockSearchBackEndPlugin(List.of("columnar")); + + when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of(plugin)); + when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn(List.of(backEnd)); + + DataFormatRegistry registry = new DataFormatRegistry(pluginsService); + + Map> descriptors = registry.getFormatDescriptors(indexSettings, format); + assertNotNull(descriptors); + } + + public void testGetFormatDescriptorsByDataFormatReturnsEmptyForUnregisteredFormat() { + when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of()); + when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn(List.of()); + + DataFormatRegistry registry = new DataFormatRegistry(pluginsService); + MockDataFormat unregistered = new MockDataFormat("unknown", 1L, Set.of()); + + Map> descriptors = registry.getFormatDescriptors(indexSettings, unregistered); + assertTrue(descriptors.isEmpty()); + } } diff --git a/server/src/test/java/org/opensearch/index/engine/dataformat/FormatChecksumStrategySharingTests.java b/server/src/test/java/org/opensearch/index/engine/dataformat/FormatChecksumStrategySharingTests.java new file mode 100644 index 0000000000000..4377cbaefa439 --- /dev/null +++ b/server/src/test/java/org/opensearch/index/engine/dataformat/FormatChecksumStrategySharingTests.java @@ -0,0 +1,224 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.index.engine.dataformat; + +import org.apache.lucene.store.FSDirectory; +import org.opensearch.Version; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.index.IndexSettings; +import org.opensearch.index.engine.dataformat.stub.MockDataFormat; +import org.opensearch.index.engine.dataformat.stub.MockSearchBackEndPlugin; +import org.opensearch.index.shard.ShardPath; +import org.opensearch.index.store.DataFormatAwareStoreDirectory; +import org.opensearch.index.store.FormatChecksumStrategy; +import org.opensearch.index.store.PrecomputedChecksumStrategy; +import org.opensearch.plugins.PluginsService; +import org.opensearch.plugins.SearchBackEndPlugin; +import org.opensearch.test.OpenSearchTestCase; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Supplier; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests that validate the FormatChecksumStrategy single-instance fix: + * strategies are created once per shard and shared between directory and engine. + */ +public class FormatChecksumStrategySharingTests extends OpenSearchTestCase { + + private static final String FORMAT_NAME = "test_format"; + + /** + * A DataFormatPlugin that returns a new PrecomputedChecksumStrategy on every + * getFormatDescriptors() call — reproducing the original bug pattern. + */ + private static class StrategyCreatingPlugin extends org.opensearch.plugins.Plugin implements DataFormatPlugin { + private final MockDataFormat format; + + StrategyCreatingPlugin(MockDataFormat format) { + this.format = format; + } + + @Override + public DataFormat getDataFormat() { + return format; + } + + @Override + public IndexingExecutionEngine indexingEngine(IndexingEngineConfig settings) { + return null; + } + + @Override + public Map> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) { + // Creates a NEW PrecomputedChecksumStrategy every call — this is the bug pattern + return Map.of(FORMAT_NAME, () -> new DataFormatDescriptor(FORMAT_NAME, new PrecomputedChecksumStrategy())); + } + } + + private DataFormatRegistry createRegistry(MockDataFormat format) { + StrategyCreatingPlugin plugin = new StrategyCreatingPlugin(format); + MockSearchBackEndPlugin backEnd = new MockSearchBackEndPlugin(List.of(format.name())); + PluginsService pluginsService = mock(PluginsService.class); + when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of(plugin)); + when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn(List.of(backEnd)); + return new DataFormatRegistry(pluginsService); + } + + private IndexSettings createIndexSettings(String indexName) { + Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) + .put(IndexMetadata.SETTING_INDEX_VERSION_CREATED.getKey(), Version.CURRENT) + .put(IndexSettings.PLUGGABLE_DATAFORMAT_VALUE_SETTING.getKey(), FORMAT_NAME) + .build(); + return new IndexSettings(IndexMetadata.builder(indexName).settings(settings).build(), settings); + } + + /** + * Verifies that createChecksumStrategies() returns the same strategy instance + * that both the directory and engine would share. + */ + public void testCreateChecksumStrategiesReturnsSameInstance() { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + IndexSettings indexSettings = createIndexSettings("test_index"); + + Map strategies = registry.createChecksumStrategies(indexSettings); + + assertNotNull(strategies.get(FORMAT_NAME)); + assertTrue(strategies.get(FORMAT_NAME) instanceof PrecomputedChecksumStrategy); + } + + /** + * Verifies that calling createChecksumStrategies() twice returns DIFFERENT + * instances (since getFormatDescriptors creates new ones each call). + * This confirms the fix must call it only once per shard. + */ + public void testMultipleCallsCreateDifferentInstances() { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + IndexSettings indexSettings = createIndexSettings("test_index"); + + Map first = registry.createChecksumStrategies(indexSettings); + Map second = registry.createChecksumStrategies(indexSettings); + + // Different calls produce different instances — this is WHY we must call it only once + assertNotSame(first.get(FORMAT_NAME), second.get(FORMAT_NAME)); + } + + /** + * Core test: checksum registered via the engine's strategy reference is visible + * from the directory's strategy reference when they share the same instance. + * This is the exact bug scenario that was broken before the fix. + */ + public void testChecksumVisibleAcrossSharedStrategy() throws IOException { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + IndexSettings indexSettings = createIndexSettings("test_index"); + + // Single call — same map shared by directory and engine + Map strategies = registry.createChecksumStrategies(indexSettings); + FormatChecksumStrategy sharedStrategy = strategies.get(FORMAT_NAME); + + long expectedChecksum = 3847291056L; + // Simulate engine registering a checksum during write + sharedStrategy.registerChecksum("_0_1.parquet", expectedChecksum, 1L); + + // Simulate directory reading the checksum during upload + Path tempDir = createTempDir(); + Path shardDataPath = tempDir.resolve("uuid").resolve("0"); + Files.createDirectories(shardDataPath.resolve(ShardPath.INDEX_FOLDER_NAME)); + ShardPath shardPath = new ShardPath(false, shardDataPath, shardDataPath, new ShardId("index", "uuid", 0)); + FSDirectory fsDir = FSDirectory.open(shardDataPath.resolve(ShardPath.INDEX_FOLDER_NAME)); + + DataFormatAwareStoreDirectory directory = new DataFormatAwareStoreDirectory(fsDir, shardPath, strategies); + + // The directory's strategy IS the same instance + FormatChecksumStrategy directoryStrategy = directory.getChecksumStrategy(FORMAT_NAME); + assertSame("Directory and engine must share the same strategy instance", sharedStrategy, directoryStrategy); + + // Verify the checksum registered by the engine is readable from the directory's strategy (O(1) lookup) + long actualChecksum = directoryStrategy.computeChecksum(fsDir, "_0_1.parquet"); + assertEquals("Checksum registered by engine must be visible via directory strategy", expectedChecksum, actualChecksum); + + directory.close(); + } + + /** + * Verifies that concurrent shard creation for different indices produces + * isolated strategy instances — no cross-index contamination. + */ + public void testDifferentIndicesGetIsolatedStrategies() { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + + IndexSettings indexSettingsA = createIndexSettings("index_a"); + IndexSettings indexSettingsB = createIndexSettings("index_b"); + + Map strategiesA = registry.createChecksumStrategies(indexSettingsA); + Map strategiesB = registry.createChecksumStrategies(indexSettingsB); + + // Different indices get different strategy instances + assertNotSame(strategiesA.get(FORMAT_NAME), strategiesB.get(FORMAT_NAME)); + + // Register checksum in index A's strategy + strategiesA.get(FORMAT_NAME).registerChecksum("_0.parquet", 12345L, 1L); + + // Index B's strategy should NOT see it + PrecomputedChecksumStrategy stratB = (PrecomputedChecksumStrategy) strategiesB.get(FORMAT_NAME); + // computeChecksum would fall back to file scan if not cached — but we can verify + // the cache is empty by checking that a different checksum isn't magically present + PrecomputedChecksumStrategy stratA = (PrecomputedChecksumStrategy) strategiesA.get(FORMAT_NAME); + assertNotSame(stratA, stratB); + } + + /** + * Verifies that the strategies map returned by createChecksumStrategies is unmodifiable. + */ + public void testCreateChecksumStrategiesReturnsUnmodifiableMap() { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + IndexSettings indexSettings = createIndexSettings("test_index"); + + Map strategies = registry.createChecksumStrategies(indexSettings); + + expectThrows(UnsupportedOperationException.class, () -> strategies.put("new_format", new PrecomputedChecksumStrategy())); + } + + /** + * Verifies that createChecksumStrategies returns empty map when no pluggable + * data format is configured. + */ + public void testCreateChecksumStrategiesEmptyWhenNoFormat() { + MockDataFormat format = new MockDataFormat(FORMAT_NAME, 100L, Set.of()); + DataFormatRegistry registry = createRegistry(format); + + // Index settings WITHOUT pluggable_dataformat setting + Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) + .put(IndexMetadata.SETTING_INDEX_VERSION_CREATED.getKey(), Version.CURRENT) + .build(); + IndexSettings indexSettings = new IndexSettings(IndexMetadata.builder("plain_index").settings(settings).build(), settings); + + Map strategies = registry.createChecksumStrategies(indexSettings); + + assertTrue(strategies.isEmpty()); + } +} diff --git a/server/src/test/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryTests.java index ba795396451b5..c8566ecbb7715 100644 --- a/server/src/test/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/DataFormatAwareStoreDirectoryTests.java @@ -13,17 +13,9 @@ import org.apache.lucene.store.IOContext; import org.apache.lucene.store.IndexInput; import org.apache.lucene.store.IndexOutput; -import org.opensearch.Version; -import org.opensearch.cluster.metadata.IndexMetadata; -import org.opensearch.common.settings.Settings; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; -import org.opensearch.index.IndexSettings; -import org.opensearch.index.engine.dataformat.DataFormatPlugin; -import org.opensearch.index.engine.dataformat.DataFormatRegistry; import org.opensearch.index.shard.ShardPath; -import org.opensearch.plugins.PluginsService; -import org.opensearch.plugins.SearchBackEndPlugin; import org.opensearch.test.OpenSearchTestCase; import org.junit.After; import org.junit.Before; @@ -34,12 +26,10 @@ import java.nio.file.Path; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.zip.CRC32; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - public class DataFormatAwareStoreDirectoryTests extends OpenSearchTestCase { private Path tempDir; @@ -64,20 +54,7 @@ public void setUp() throws Exception { ShardId sid = new ShardId(new Index("test-index", indexUUID), shardId); shardPath = new ShardPath(false, shardDataPath, shardDataPath, sid); - PluginsService pluginsService = mock(PluginsService.class); - when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of()); - when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn(List.of()); - DataFormatRegistry dataFormatRegistry = new DataFormatRegistry(pluginsService); - - // Create real IndexSettings (IndexSettings is final, cannot be mocked) - Settings settings = Settings.builder() - .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) - .put(IndexMetadata.SETTING_INDEX_UUID, indexUUID) - .build(); - IndexMetadata metadata = IndexMetadata.builder("test-index").settings(settings).numberOfShards(1).numberOfReplicas(0).build(); - IndexSettings indexSettings = new IndexSettings(metadata, Settings.EMPTY); - - dataFormatAwareStoreDirectory = new DataFormatAwareStoreDirectory(indexSettings, fsDirectory, shardPath, dataFormatRegistry); + dataFormatAwareStoreDirectory = new DataFormatAwareStoreDirectory(fsDirectory, shardPath, Map.of()); } @After diff --git a/server/src/test/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactoryTests.java b/server/src/test/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactoryTests.java index d1a47e9710661..b76ac1f3b3511 100644 --- a/server/src/test/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/DefaultDataFormatAwareStoreDirectoryFactoryTests.java @@ -14,22 +14,16 @@ import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.IndexSettings; -import org.opensearch.index.engine.dataformat.DataFormatPlugin; -import org.opensearch.index.engine.dataformat.DataFormatRegistry; import org.opensearch.index.shard.ShardPath; import org.opensearch.plugins.IndexStorePlugin; -import org.opensearch.plugins.PluginsService; -import org.opensearch.plugins.SearchBackEndPlugin; import org.opensearch.test.OpenSearchTestCase; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; -import java.util.List; +import java.util.Map; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_INDEX_UUID; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; /** * Unit tests for {@link DefaultDataFormatAwareStoreDirectoryFactory}. @@ -56,13 +50,6 @@ private IndexSettings createIndexSettings() { return new IndexSettings(metadata, Settings.EMPTY); } - private DataFormatRegistry createEmptyDataFormatRegistry() { - PluginsService pluginsService = mock(PluginsService.class); - when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of()); - when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn(List.of()); - return new DataFormatRegistry(pluginsService); - } - private IndexStorePlugin.DirectoryFactory createFsDirectoryFactory() { return new IndexStorePlugin.DirectoryFactory() { @Override @@ -86,7 +73,6 @@ public org.apache.lucene.store.Directory newFSDirectory( // ═══════════════════════════════════════════════════════════════ public void testNewDataFormatAwareStoreDirectory_CreatesSuccessfully() throws IOException { - DataFormatRegistry registry = createEmptyDataFormatRegistry(); DefaultDataFormatAwareStoreDirectoryFactory factory = new DefaultDataFormatAwareStoreDirectoryFactory(); Path tempDir = createTempDir(); ShardPath shardPath = createShardPath(tempDir); @@ -97,14 +83,13 @@ public void testNewDataFormatAwareStoreDirectory_CreatesSuccessfully() throws IO shardPath.getShardId(), shardPath, createFsDirectoryFactory(), - registry + Map.of() ); assertNotNull("Factory should create a non-null DataFormatAwareStoreDirectory", directory); } public void testNewDataFormatAwareStoreDirectory_HasCorrectShardPath() throws IOException { - DataFormatRegistry registry = createEmptyDataFormatRegistry(); DefaultDataFormatAwareStoreDirectoryFactory factory = new DefaultDataFormatAwareStoreDirectoryFactory(); Path tempDir = createTempDir(); ShardPath shardPath = createShardPath(tempDir); @@ -115,14 +100,13 @@ public void testNewDataFormatAwareStoreDirectory_HasCorrectShardPath() throws IO shardPath.getShardId(), shardPath, createFsDirectoryFactory(), - registry + Map.of() ); assertEquals(shardPath, directory.getShardPath()); } public void testNewDataFormatAwareStoreDirectory_CanListFiles() throws IOException { - DataFormatRegistry registry = createEmptyDataFormatRegistry(); DefaultDataFormatAwareStoreDirectoryFactory factory = new DefaultDataFormatAwareStoreDirectoryFactory(); Path tempDir = createTempDir(); ShardPath shardPath = createShardPath(tempDir); @@ -133,7 +117,7 @@ public void testNewDataFormatAwareStoreDirectory_CanListFiles() throws IOExcepti shardPath.getShardId(), shardPath, createFsDirectoryFactory(), - registry + Map.of() ); // Should not throw @@ -142,7 +126,6 @@ public void testNewDataFormatAwareStoreDirectory_CanListFiles() throws IOExcepti } public void testNewDataFormatAwareStoreDirectory_MultipleCalls_CreatesSeparateInstances() throws IOException { - DataFormatRegistry registry = createEmptyDataFormatRegistry(); DefaultDataFormatAwareStoreDirectoryFactory factory = new DefaultDataFormatAwareStoreDirectoryFactory(); Path tempDir1 = createTempDir(); Path tempDir2 = createTempDir(); @@ -155,14 +138,14 @@ public void testNewDataFormatAwareStoreDirectory_MultipleCalls_CreatesSeparateIn shardPath1.getShardId(), shardPath1, createFsDirectoryFactory(), - registry + Map.of() ); DataFormatAwareStoreDirectory dir2 = factory.newDataFormatAwareStoreDirectory( indexSettings, shardPath2.getShardId(), shardPath2, createFsDirectoryFactory(), - registry + Map.of() ); assertNotNull(dir1); @@ -171,7 +154,6 @@ public void testNewDataFormatAwareStoreDirectory_MultipleCalls_CreatesSeparateIn } public void testNewDataFormatAwareStoreDirectory_InvalidPath_ThrowsIOException() throws IOException { - DataFormatRegistry registry = createEmptyDataFormatRegistry(); DefaultDataFormatAwareStoreDirectoryFactory factory = new DefaultDataFormatAwareStoreDirectoryFactory(); IndexSettings indexSettings = createIndexSettings(); @@ -197,7 +179,7 @@ public void testNewDataFormatAwareStoreDirectory_InvalidPath_ThrowsIOException() invalidShardPath.getShardId(), invalidShardPath, createFsDirectoryFactory(), - registry + Map.of() ) ); assertTrue( diff --git a/server/src/test/java/org/opensearch/index/store/remote/DataFormatAwareRemoteDirectoryTests.java b/server/src/test/java/org/opensearch/index/store/remote/DataFormatAwareRemoteDirectoryTests.java index 62a571aab9a41..ec1a12db8504c 100644 --- a/server/src/test/java/org/opensearch/index/store/remote/DataFormatAwareRemoteDirectoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/remote/DataFormatAwareRemoteDirectoryTests.java @@ -50,6 +50,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import java.util.function.UnaryOperator; import org.mockito.Mockito; @@ -107,7 +108,7 @@ public void setUp() throws Exception { .build(); IndexSettings indexSettings = new IndexSettings(metadata, Settings.EMPTY); when(mockRegistry.getFormatDescriptors(any(IndexSettings.class))).thenReturn( - Map.of("parquet", new DataFormatDescriptor("parquet", new GenericCRC32ChecksumHandler())) + Map.of("parquet", (Supplier) () -> new DataFormatDescriptor("parquet", new GenericCRC32ChecksumHandler())) ); directory = new DataFormatAwareRemoteDirectory( diff --git a/test/framework/src/main/java/org/opensearch/index/engine/dataformat/stub/MockDataFormatPlugin.java b/test/framework/src/main/java/org/opensearch/index/engine/dataformat/stub/MockDataFormatPlugin.java index 021fef15f9969..82d5a7d929682 100644 --- a/test/framework/src/main/java/org/opensearch/index/engine/dataformat/stub/MockDataFormatPlugin.java +++ b/test/framework/src/main/java/org/opensearch/index/engine/dataformat/stub/MockDataFormatPlugin.java @@ -12,7 +12,6 @@ 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.Plugin; import java.util.Set; @@ -41,7 +40,7 @@ public DataFormat getDataFormat() { } @Override - public IndexingExecutionEngine indexingEngine(IndexingEngineConfig settings, FormatChecksumStrategy checksumStrategy) { + public IndexingExecutionEngine indexingEngine(IndexingEngineConfig settings) { return new MockIndexingExecutionEngine(dataFormat); } } diff --git a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java index b5408b3709e70..68e14e34624b8 100644 --- a/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java +++ b/test/framework/src/main/java/org/opensearch/index/shard/IndexShardTestCase.java @@ -807,6 +807,7 @@ protected IndexShard newShard( clusterService.getClusterApplierService(), mergedSegmentPublisher, ReferencedSegmentsPublisher.EMPTY, + Collections.emptyMap(), null // TODO ); indexShard.addShardFailureCallback(DEFAULT_SHARD_FAILURE_HANDLER);