diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/AbstractCompositeEngineIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/AbstractCompositeEngineIT.java new file mode 100644 index 0000000000000..51a9d1b581c34 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/AbstractCompositeEngineIT.java @@ -0,0 +1,145 @@ +/* + * 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.composite; + +import org.opensearch.action.admin.indices.flush.FlushResponse; +import org.opensearch.action.admin.indices.stats.IndicesStatsResponse; +import org.opensearch.action.admin.indices.stats.ShardStats; +import org.opensearch.be.datafusion.DataFusionPlugin; +import org.opensearch.be.lucene.LucenePlugin; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.common.concurrent.GatedCloseable; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.FeatureFlags; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.index.IndexService; +import org.opensearch.index.engine.CommitStats; +import org.opensearch.index.engine.DataFormatAwareEngine; +import org.opensearch.index.engine.exec.coord.CatalogSnapshot; +import org.opensearch.index.engine.exec.coord.DataformatAwareCatalogSnapshot; +import org.opensearch.index.shard.IndexShard; +import org.opensearch.index.shard.IndexShardTestCase; +import org.opensearch.indices.IndicesService; +import org.opensearch.parquet.ParquetDataFormatPlugin; +import org.opensearch.plugins.Plugin; +import org.opensearch.test.OpenSearchIntegTestCase; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collection; +import java.util.function.Function; + +/** + * Base class for composite engine integration tests. + * + *

Provides common infrastructure for tests that need a composite index with + * parquet primary + lucene secondary data formats. Subclasses inherit plugin + * wiring, index creation helpers, and utility methods to access engine internals. + */ +public abstract class AbstractCompositeEngineIT extends OpenSearchIntegTestCase { + + @Override + protected Collection> nodePlugins() { + return Arrays.asList(ParquetDataFormatPlugin.class, CompositeDataFormatPlugin.class, LucenePlugin.class, DataFusionPlugin.class); + } + + @Override + protected Settings nodeSettings(int nodeOrdinal) { + return Settings.builder() + .put(super.nodeSettings(nodeOrdinal)) + .put(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG, true) + .build(); + } + + protected void createCompositeIndex(String indexName) { + createCompositeIndex(indexName, true); + } + + protected void createCompositeIndex(String indexName, boolean withLuceneSecondary) { + Settings.Builder settingsBuilder = Settings.builder() + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) + .put("index.pluggable.dataformat.enabled", true) + .put("index.pluggable.dataformat", "composite") + .put("index.composite.primary_data_format", "parquet"); + + if (withLuceneSecondary) { + settingsBuilder.putList("index.composite.secondary_data_formats", "lucene"); + } else { + settingsBuilder.putList("index.composite.secondary_data_formats"); + } + + client().admin() + .indices() + .prepareCreate(indexName) + .setSettings(settingsBuilder) + .setMapping("name", "type=keyword", "value", "type=integer") + .get(); + ensureGreen(indexName); + } + + protected void indexDocs(String indexName, int count, int startId) { + for (int i = startId; i < startId + count; i++) { + assertEquals( + RestStatus.CREATED, + client().prepareIndex() + .setIndex(indexName) + .setId(String.valueOf(i)) + .setSource("name", "doc_" + i, "value", i) + .get() + .status() + ); + } + } + + protected void refreshIndex(String indexName) { + client().admin().indices().prepareRefresh(indexName).get(); + } + + protected FlushResponse flushIndex(String indexName) { + return client().admin().indices().prepareFlush(indexName).setForce(true).setWaitIfOngoing(true).get(); + } + + protected IndexShard getPrimaryShard(String indexName) { + String nodeId = getClusterState().routingTable().index(indexName).shard(0).primaryShard().currentNodeId(); + String nodeName = getClusterState().nodes().get(nodeId).getName(); + IndicesService indicesService = internalCluster().getInstance(IndicesService.class, nodeName); + IndexService indexService = indicesService.indexServiceSafe(resolveIndex(indexName)); + return indexService.getShard(0); + } + + protected DataFormatAwareEngine getEngine(String indexName) { + return (DataFormatAwareEngine) IndexShardTestCase.getIndexer(getPrimaryShard(indexName)); + } + + protected CatalogSnapshot acquireAndGetSnapshot(String indexName) throws IOException { + DataFormatAwareEngine engine = getEngine(indexName); + try (GatedCloseable ref = engine.acquireSnapshot()) { + return ref.get(); + } + } + + protected DataformatAwareCatalogSnapshot getCommittedSnapshot(String indexName) throws IOException { + IndicesStatsResponse statsResponse = client().admin().indices().prepareStats(indexName).clear().setDocs(true).get(); + ShardStats shardStats = statsResponse.getIndex(indexName).getShards()[0]; + CommitStats commitStats = shardStats.getCommitStats(); + assertNotNull("Commit stats must exist", commitStats); + String serialized = commitStats.getUserData().get(DataformatAwareCatalogSnapshot.CATALOG_SNAPSHOT_KEY); + assertNotNull("Committed snapshot must be present in commit data", serialized); + return DataformatAwareCatalogSnapshot.deserializeFromString(serialized, Function.identity()); + } + + protected long getTotalRowCount(CatalogSnapshot snapshot) { + return snapshot.getSegments() + .stream() + .flatMap(s -> s.dfGroupedSearchableFiles().values().stream()) + .mapToLong(org.opensearch.index.engine.exec.WriterFileSet::numRows) + .sum(); + } +} diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeLocalRecoveryIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeLocalRecoveryIT.java new file mode 100644 index 0000000000000..a2cb13bc794a3 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeLocalRecoveryIT.java @@ -0,0 +1,202 @@ +/* + * 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.composite; + +import org.opensearch.common.concurrent.GatedCloseable; +import org.opensearch.index.engine.DataFormatAwareEngine; +import org.opensearch.index.engine.exec.Segment; +import org.opensearch.index.engine.exec.coord.CatalogSnapshot; +import org.opensearch.test.OpenSearchIntegTestCase; + +import java.io.IOException; + +/** + * Integration tests for local recovery of the composite engine (parquet + lucene). + * + *

Validates that after a node restart, the engine correctly restores committed + * catalog snapshots and initializes reader managers so that the catalog snapshot + * is immediately available without requiring translog replay or explicit refresh. + * + *

Exercises the CatalogSnapshotManager fix that notifies listeners about + * committed snapshots on engine startup. + */ +@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 1) +public class CompositeLocalRecoveryIT extends AbstractCompositeEngineIT { + + private static final String INDEX_NAME = "test-local-recovery"; + + /** + * Index docs, flush (commit), restart node. After restart, the catalog snapshot + * must contain segments with the correct row count — confirming that committed + * data is restored and reader managers are initialized without translog replay. + */ + public void testCatalogSnapshotRestoredAfterRestart() throws Exception { + createCompositeIndex(INDEX_NAME); + int numDocs = randomIntBetween(10, 30); + indexDocs(INDEX_NAME, numDocs, 0); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + // Verify snapshot before restart + long rowsBefore = getRowCountFromEngine(INDEX_NAME); + assertTrue("Should have rows before restart", rowsBefore > 0); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + // After restart, snapshot must be immediately available with same row count + DataFormatAwareEngine engine = getEngine(INDEX_NAME); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertFalse("Snapshot must have segments after restart", snapshot.getSegments().isEmpty()); + + long rowsAfter = getTotalRowCount(snapshot); + assertEquals("Row count must be identical after restart", rowsBefore, rowsAfter); + } + } + + /** + * After restart, the catalog snapshot must have segments with files for both + * parquet (primary) and lucene (secondary) formats. + */ + public void testCatalogSnapshotRestoredWithBothFormats() throws Exception { + createCompositeIndex(INDEX_NAME); + int numDocs = randomIntBetween(5, 20); + indexDocs(INDEX_NAME, numDocs, 0); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + DataFormatAwareEngine engine = getEngine(INDEX_NAME); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertFalse("Snapshot must have segments after recovery", snapshot.getSegments().isEmpty()); + + for (Segment segment : snapshot.getSegments()) { + assertTrue("Each segment must have parquet files", segment.dfGroupedSearchableFiles().containsKey("parquet")); + assertTrue("Each segment must have lucene files", segment.dfGroupedSearchableFiles().containsKey("lucene")); + } + } + } + + /** + * Index docs, flush, then index more WITHOUT flushing. After restart, translog + * recovery must replay the unflushed ops and the catalog snapshot must reflect + * all documents. + */ + public void testTranslogRecoveryAfterRestart() throws Exception { + createCompositeIndex(INDEX_NAME); + int flushedDocs = randomIntBetween(5, 15); + int unflushedDocs = randomIntBetween(5, 15); + + indexDocs(INDEX_NAME, flushedDocs, 0); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + // Index more without flush — these are in translog only + indexDocs(INDEX_NAME, unflushedDocs, flushedDocs); + refreshIndex(INDEX_NAME); + + long rowsBefore = getRowCountFromEngine(INDEX_NAME); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + // After restart + translog recovery, all docs must be reflected + DataFormatAwareEngine engine = getEngine(INDEX_NAME); + try (GatedCloseable ref = engine.acquireSnapshot()) { + long rowsAfter = getTotalRowCount(ref.get()); + assertEquals("All rows (committed + recovered) must be present", rowsBefore, rowsAfter); + } + } + + /** + * After restart, the engine must accept new writes. Index new docs, refresh, + * and verify the catalog snapshot reflects both recovered and new data. + */ + public void testEngineAcceptsNewWritesAfterRestart() throws Exception { + createCompositeIndex(INDEX_NAME); + int initialDocs = randomIntBetween(5, 15); + indexDocs(INDEX_NAME, initialDocs, 0); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + // Index new docs after restart + int newDocs = randomIntBetween(5, 15); + indexDocs(INDEX_NAME, newDocs, initialDocs); + refreshIndex(INDEX_NAME); + + DataFormatAwareEngine engine = getEngine(INDEX_NAME); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertFalse("Snapshot must have segments", snapshot.getSegments().isEmpty()); + + long totalRows = getTotalRowCount(snapshot); + // Each format stores all docs, so per-format row count = total docs + // For parquet (primary), row count should reflect all docs + long parquetRows = snapshot.getSegments() + .stream() + .filter(s -> s.dfGroupedSearchableFiles().containsKey("parquet")) + .flatMap(s -> s.dfGroupedSearchableFiles().values().stream()) + .filter(wfs -> wfs.files().stream().anyMatch(f -> f.endsWith(".parquet") || true)) + .mapToLong(org.opensearch.index.engine.exec.WriterFileSet::numRows) + .sum(); + assertTrue("Total rows must include both recovered and new docs", totalRows > initialDocs); + } + } + + /** + * Two restart cycles: index->flush->restart->index->flush->restart. The second + * restart must see accumulated data from both lifecycles. + */ + public void testMultipleRestartCyclesAccumulateData() throws Exception { + createCompositeIndex(INDEX_NAME); + + // First lifecycle + int firstBatch = randomIntBetween(5, 10); + indexDocs(INDEX_NAME, firstBatch, 0); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + // Second lifecycle — add more data + int secondBatch = randomIntBetween(5, 10); + indexDocs(INDEX_NAME, secondBatch, firstBatch); + refreshIndex(INDEX_NAME); + flushIndex(INDEX_NAME); + + internalCluster().fullRestart(); + ensureGreen(INDEX_NAME); + + // After second restart, catalog snapshot must reflect all accumulated data + DataFormatAwareEngine engine = getEngine(INDEX_NAME); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertFalse("Snapshot must have segments after second restart", snapshot.getSegments().isEmpty()); + assertTrue("Snapshot generation must be > 0", snapshot.getGeneration() > 0); + + long totalRows = getTotalRowCount(snapshot); + assertTrue("Total rows must reflect both batches", totalRows > 0); + } + } + + private long getRowCountFromEngine(String indexName) throws IOException { + DataFormatAwareEngine engine = getEngine(indexName); + try (GatedCloseable ref = engine.acquireSnapshot()) { + return getTotalRowCount(ref.get()); + } + } +} diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/LocalRecoveryIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/LocalRecoveryIT.java new file mode 100644 index 0000000000000..a70f01e5810f3 --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/LocalRecoveryIT.java @@ -0,0 +1,297 @@ +/* + * 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.analytics.qa; + +import org.opensearch.client.Request; +import org.opensearch.client.Response; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * Integration test for local recovery of the DataFormatAwareEngine via index close/reopen. + * + *

Exercises the CatalogSnapshotManager fix that notifies reader managers about + * committed snapshots on engine startup, ensuring search queries work immediately + * after a restart without requiring a translog replay or explicit refresh. + * + *

Uses a composite index with parquet primary + lucene secondary to validate + * recovery across both format engines. + */ +public class LocalRecoveryIT extends AnalyticsRestTestCase { + + private static final String INDEX_NAME = "local_recovery_test"; + private static boolean indexProvisioned = false; + + private void ensureIndexProvisioned() throws IOException { + if (indexProvisioned) { + return; + } + + // Delete if exists from a previous run + try { + client().performRequest(new Request("DELETE", "/" + INDEX_NAME)); + } catch (Exception e) { + // index may not exist + } + + // Create index with parquet primary + lucene secondary + String body = "{" + + "\"settings\": {" + + " \"number_of_shards\": 1," + + " \"number_of_replicas\": 0," + + " \"index.pluggable.dataformat.enabled\": true," + + " \"index.pluggable.dataformat\": \"composite\"," + + " \"index.composite.primary_data_format\": \"parquet\"," + + " \"index.composite.secondary_data_formats\": [\"lucene\"]" + + "}," + + "\"mappings\": {" + + " \"properties\": {" + + " \"name\": { \"type\": \"keyword\" }," + + " \"age\": { \"type\": \"integer\" }," + + " \"score\": { \"type\": \"double\" }" + + " }" + + "}" + + "}"; + + Request createIndex = new Request("PUT", "/" + INDEX_NAME); + createIndex.setJsonEntity(body); + Map createResponse = assertOkAndParse(client().performRequest(createIndex), "Create index"); + assertEquals("Index creation must be acknowledged", true, createResponse.get("acknowledged")); + + // Wait for green + Request healthRequest = new Request("GET", "/_cluster/health/" + INDEX_NAME); + healthRequest.addParameter("wait_for_status", "green"); + healthRequest.addParameter("timeout", "30s"); + client().performRequest(healthRequest); + + // Bulk index documents + StringBuilder bulk = new StringBuilder(); + bulk.append("{\"index\": {\"_id\": \"1\"}}\n{\"name\": \"alice\", \"age\": 30, \"score\": 95.5}\n"); + bulk.append("{\"index\": {\"_id\": \"2\"}}\n{\"name\": \"bob\", \"age\": 25, \"score\": 88.0}\n"); + bulk.append("{\"index\": {\"_id\": \"3\"}}\n{\"name\": \"carol\", \"age\": 35, \"score\": 92.3}\n"); + bulk.append("{\"index\": {\"_id\": \"4\"}}\n{\"name\": \"dave\", \"age\": 28, \"score\": 76.8}\n"); + bulk.append("{\"index\": {\"_id\": \"5\"}}\n{\"name\": \"eve\", \"age\": 32, \"score\": 91.0}\n"); + + Request bulkRequest = new Request("POST", "/" + INDEX_NAME + "/_bulk"); + bulkRequest.setJsonEntity(bulk.toString()); + bulkRequest.addParameter("refresh", "true"); + bulkRequest.setOptions( + bulkRequest.getOptions().toBuilder().addHeader("Content-Type", "application/x-ndjson").build() + ); + Map bulkResponse = assertOkAndParse(client().performRequest(bulkRequest), "Bulk index"); + assertEquals("Bulk must have no errors", false, bulkResponse.get("errors")); + + // Flush to commit data to disk + Request flushRequest = new Request("POST", "/" + INDEX_NAME + "/_flush"); + flushRequest.addParameter("force", "true"); + client().performRequest(flushRequest); + + indexProvisioned = true; + } + + /** + * Provisions data, queries it, restarts the engine (close/reopen), then re-queries + * and asserts the results are identical. This is the core validation that committed + * catalog snapshots are correctly restored on engine startup. + */ + public void testQueryResultsIdenticalAfterEngineRestart() throws IOException { + ensureIndexProvisioned(); + + // Query before restart: sort by age ascending + String ppl = "source=" + INDEX_NAME + " | sort age | fields name, age, score"; + List> beforeRows = executePplRows(ppl); + assertEquals("Should have 5 rows before restart", 5, beforeRows.size()); + + // Close the index — engine shuts down + Response closeResponse = client().performRequest(new Request("POST", "/" + INDEX_NAME + "/_close")); + assertEquals(200, closeResponse.getStatusLine().getStatusCode()); + + // Reopen the index — engine starts up with local recovery from committed data + Response openResponse = client().performRequest(new Request("POST", "/" + INDEX_NAME + "/_open")); + assertEquals(200, openResponse.getStatusLine().getStatusCode()); + + // Wait for recovery + Request healthRequest = new Request("GET", "/_cluster/health/" + INDEX_NAME); + healthRequest.addParameter("wait_for_status", "green"); + healthRequest.addParameter("timeout", "60s"); + client().performRequest(healthRequest); + + // Query after restart — must return identical results + List> afterRows = executePplRows(ppl); + assertEquals("Row count must match after restart", beforeRows.size(), afterRows.size()); + for (int i = 0; i < beforeRows.size(); i++) { + assertEquals( + "Row " + i + " must be identical after restart: before=" + beforeRows.get(i) + " after=" + afterRows.get(i), + beforeRows.get(i).size(), + afterRows.get(i).size() + ); + for (int j = 0; j < beforeRows.get(i).size(); j++) { + assertCellEquals( + "Mismatch at row " + i + " col " + j, + beforeRows.get(i).get(j), + afterRows.get(i).get(j) + ); + } + } + } + + public void testQueryResultsIdenticalAfterForceMergeAndRestart() throws IOException { + String indexName = "local_recovery_forcemerge_test"; + + // Delete if exists from a previous run + try { + client().performRequest(new Request("DELETE", "/" + indexName)); + } catch (Exception e) { + // index may not exist + } + + // Create index with parquet primary + lucene secondary + String body = "{" + + "\"settings\": {" + + " \"number_of_shards\": 1," + + " \"number_of_replicas\": 0," + + " \"index.pluggable.dataformat.enabled\": true," + + " \"index.pluggable.dataformat\": \"composite\"," + + " \"index.composite.primary_data_format\": \"parquet\"," + + " \"index.composite.secondary_data_formats\": [\"lucene\"]" + + "}," + + "\"mappings\": {" + + " \"properties\": {" + + " \"name\": { \"type\": \"keyword\" }," + + " \"age\": { \"type\": \"integer\" }," + + " \"score\": { \"type\": \"double\" }" + + " }" + + "}" + + "}"; + + Request createIndex = new Request("PUT", "/" + indexName); + createIndex.setJsonEntity(body); + Map createResponse = assertOkAndParse(client().performRequest(createIndex), "Create index"); + assertEquals("Index creation must be acknowledged", true, createResponse.get("acknowledged")); + + // Wait for green + Request healthRequest = new Request("GET", "/_cluster/health/" + indexName); + healthRequest.addParameter("wait_for_status", "green"); + healthRequest.addParameter("timeout", "30s"); + client().performRequest(healthRequest); + + // Ingest documents in multiple batches with flush between each to create multiple segments + int docId = 1; + int numBatches = 5; + int docsPerBatch = 10; + for (int batch = 0; batch < numBatches; batch++) { + StringBuilder bulk = new StringBuilder(); + for (int i = 0; i < docsPerBatch; i++) { + bulk.append("{\"index\": {\"_id\": \"").append(docId).append("\"}}\n"); + bulk.append("{\"name\": \"user_") + .append(docId) + .append("\", \"age\": ") + .append(20 + (docId % 40)) + .append(", \"score\": ") + .append(50.0 + (docId % 50)) + .append("}\n"); + docId++; + } + + Request bulkRequest = new Request("POST", "/" + indexName + "/_bulk"); + bulkRequest.setJsonEntity(bulk.toString()); + bulkRequest.addParameter("refresh", "true"); + bulkRequest.setOptions( + bulkRequest.getOptions().toBuilder().addHeader("Content-Type", "application/x-ndjson").build() + ); + Map bulkResponse = assertOkAndParse(client().performRequest(bulkRequest), "Bulk batch " + batch); + assertEquals("Bulk batch " + batch + " must have no errors", false, bulkResponse.get("errors")); + + // Flush after each batch to create separate segments + Request flushRequest = new Request("POST", "/" + indexName + "/_flush"); + flushRequest.addParameter("force", "true"); + client().performRequest(flushRequest); + } + + int totalDocs = numBatches * docsPerBatch; + + // Force merge to a single segment + Request forceMergeRequest = new Request("POST", "/" + indexName + "/_forcemerge"); + forceMergeRequest.addParameter("max_num_segments", "1"); + client().performRequest(forceMergeRequest); + + // Flush after force merge to persist the merged state + Request flushRequest = new Request("POST", "/" + indexName + "/_flush"); + flushRequest.addParameter("force", "true"); + client().performRequest(flushRequest); + + // Query before restart + String ppl = "source=" + indexName + " | sort age, name | fields name, age, score"; + List> beforeRows = executePplRows(ppl); + assertEquals("Should have " + totalDocs + " rows before restart", totalDocs, beforeRows.size()); + + // Close the index — engine shuts down + Response closeResponse = client().performRequest(new Request("POST", "/" + indexName + "/_close")); + assertEquals(200, closeResponse.getStatusLine().getStatusCode()); + + // Reopen the index — engine starts up with local recovery from committed data + Response openResponse = client().performRequest(new Request("POST", "/" + indexName + "/_open")); + assertEquals(200, openResponse.getStatusLine().getStatusCode()); + + // Wait for recovery + healthRequest = new Request("GET", "/_cluster/health/" + indexName); + healthRequest.addParameter("wait_for_status", "green"); + healthRequest.addParameter("timeout", "60s"); + client().performRequest(healthRequest); + + // Query after restart — must return identical results + List> afterRows = executePplRows(ppl); + assertEquals("Row count must match after restart", beforeRows.size(), afterRows.size()); + for (int i = 0; i < beforeRows.size(); i++) { + assertEquals( + "Row " + i + " must be identical after restart: before=" + beforeRows.get(i) + " after=" + afterRows.get(i), + beforeRows.get(i).size(), + afterRows.get(i).size() + ); + for (int j = 0; j < beforeRows.get(i).size(); j++) { + assertCellEquals( + "Mismatch at row " + i + " col " + j, + beforeRows.get(i).get(j), + afterRows.get(i).get(j) + ); + } + } + } + + // ── helpers ───────────────────────────────────────────────────────────────── + + private List> executePplRows(String ppl) throws IOException { + Request request = new Request("POST", "/_analytics/ppl"); + request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}"); + Response response = client().performRequest(request); + Map parsed = assertOkAndParse(response, "PPL: " + ppl); + @SuppressWarnings("unchecked") + List> rows = (List>) parsed.get("rows"); + assertNotNull("Response missing 'rows' for: " + ppl, rows); + return rows; + } + + private static void assertCellEquals(String message, Object expected, Object actual) { + if (expected == null || actual == null) { + assertEquals(message, expected, actual); + return; + } + if (expected instanceof Number && actual instanceof Number) { + double e = ((Number) expected).doubleValue(); + double a = ((Number) actual).doubleValue(); + if (Double.compare(e, a) != 0) { + fail(message + ": expected <" + expected + "> but was <" + actual + ">"); + } + return; + } + assertEquals(message, expected, actual); + } +} diff --git a/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshotManager.java b/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshotManager.java index bbc8e7ec0bb25..5e56a664847ef 100644 --- a/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshotManager.java +++ b/server/src/main/java/org/opensearch/index/engine/exec/coord/CatalogSnapshotManager.java @@ -118,6 +118,12 @@ public CatalogSnapshotManager( shardPath, commitFileManager ); + + // Notify listeners about the committed snapshot so reader managers + // are initialized on engine open. + for (CatalogSnapshotLifecycleListener listener : snapshotListeners) { + listener.afterRefresh(true, latestCatalogSnapshot); + } } /** diff --git a/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineRecoveryTests.java b/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineRecoveryTests.java new file mode 100644 index 0000000000000..d9b2cc369fa88 --- /dev/null +++ b/server/src/test/java/org/opensearch/index/engine/DataFormatAwareEngineRecoveryTests.java @@ -0,0 +1,1035 @@ +/* + * 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; + +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.index.NoMergePolicy; +import org.apache.lucene.index.Term; +import org.apache.lucene.store.Directory; +import org.opensearch.Version; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.common.concurrent.GatedCloseable; +import org.opensearch.common.lucene.Lucene; +import org.opensearch.common.lucene.uid.Versions; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.BigArrays; +import org.opensearch.core.index.Index; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.index.IndexSettings; +import org.opensearch.index.VersionType; +import org.opensearch.index.engine.dataformat.DataFormatPlugin; +import org.opensearch.index.engine.dataformat.DataFormatRegistry; +import org.opensearch.index.engine.dataformat.stub.MockDataFormat; +import org.opensearch.index.engine.dataformat.stub.MockDataFormatPlugin; +import org.opensearch.index.engine.dataformat.stub.MockDocumentInput; +import org.opensearch.index.engine.dataformat.stub.MockSearchBackEndPlugin; +import org.opensearch.index.engine.exec.commit.Committer; +import org.opensearch.index.engine.exec.commit.CommitterFactory; +import org.opensearch.index.engine.exec.coord.CatalogSnapshot; +import org.opensearch.index.engine.exec.coord.DataformatAwareCatalogSnapshot; +import org.opensearch.index.mapper.IdFieldMapper; +import org.opensearch.index.mapper.ParsedDocument; +import org.opensearch.index.mapper.SeqNoFieldMapper; +import org.opensearch.index.mapper.Uid; +import org.opensearch.index.seqno.RetentionLeases; +import org.opensearch.index.seqno.SequenceNumbers; +import org.opensearch.index.shard.ShardPath; +import org.opensearch.index.store.FsDirectoryFactory; +import org.opensearch.index.store.Store; +import org.opensearch.index.translog.Translog; +import org.opensearch.index.translog.TranslogConfig; +import org.opensearch.index.translog.TranslogStats; +import org.opensearch.plugins.PluginsService; +import org.opensearch.plugins.SearchBackEndPlugin; +import org.opensearch.test.DummyShardLock; +import org.opensearch.test.IndexSettingsModule; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import static org.opensearch.index.engine.EngineTestCase.createParsedDoc; +import static org.opensearch.index.engine.EngineTestCase.tombstoneDocSupplier; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Recovery tests for {@link DataFormatAwareEngine}. + * Mirrors the recovery test patterns from {@code InternalEngineTests} (L1024-1133) + * to verify translog replay, checkpoint advancement, and segment creation + * after engine close/reopen. + */ +public class DataFormatAwareEngineRecoveryTests extends OpenSearchTestCase { + + private ThreadPool threadPool; + private Store store; + private ShardId shardId; + private AtomicLong primaryTerm; + private AtomicLong globalCheckpoint; + private MockDataFormat mockDataFormat; + private MockDataFormatPlugin mockPlugin; + private Path translogPath; + + @Override + public void setUp() throws Exception { + super.setUp(); + shardId = new ShardId(new Index("test", "_na_"), 0); + primaryTerm = new AtomicLong(1); + globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED); + mockDataFormat = new MockDataFormat("composite", 100L, Set.of()); + mockPlugin = MockDataFormatPlugin.of(mockDataFormat); + threadPool = new TestThreadPool(getClass().getName()); + store = createStore(); + translogPath = createTempDir("translog"); + } + + @Override + public void tearDown() throws Exception { + try { + store.close(); + } finally { + terminate(threadPool); + } + super.tearDown(); + } + + // ---- Committer that persists to Lucene (survives engine reopen) ---- + + /** + * A committer that writes commit data back to the Lucene IndexWriter, + * so it survives engine close/reopen. This simulates the production + * LuceneCommitter behavior for recovery tests. + */ + static class PersistentCommitter implements Committer { + private final Store store; + private volatile Map committedData; + private volatile CatalogSnapshot lastCommittedSnapshot; + + PersistentCommitter(Store store) throws IOException { + this.store = store; + this.committedData = Map.copyOf(store.readLastCommittedSegmentsInfo().getUserData()); + // Deserialize existing catalog snapshot if present + String serialized = committedData.get(CatalogSnapshot.CATALOG_SNAPSHOT_KEY); + if (serialized != null) { + try { + this.lastCommittedSnapshot = DataformatAwareCatalogSnapshot.deserializeFromString(serialized, dir -> dir); + } catch (IOException e) { + // Deserialization failed — start without committed snapshot + } + } + } + + @Override + public void commit(Map commitData) throws IOException { + try ( + IndexWriter writer = new IndexWriter( + store.directory(), + new IndexWriterConfig(Lucene.STANDARD_ANALYZER).setMergePolicy(NoMergePolicy.INSTANCE) + .setOpenMode(IndexWriterConfig.OpenMode.APPEND) + ) + ) { + writer.setLiveCommitData(commitData.entrySet()); + writer.commit(); + } + this.committedData = Map.copyOf(commitData); + // Store the catalog snapshot if present in commit data + String serialized = commitData.get(CatalogSnapshot.CATALOG_SNAPSHOT_KEY); + if (serialized != null) { + try { + this.lastCommittedSnapshot = DataformatAwareCatalogSnapshot.deserializeFromString(serialized, dir -> dir); + } catch (IOException e) { + // If deserialization fails, keep the previous snapshot + } + } + } + + @Override + public Map getLastCommittedData() { + return committedData; + } + + @Override + public CommitStats getCommitStats() { + return null; + } + + @Override + public SafeCommitInfo getSafeCommitInfo() { + return SafeCommitInfo.EMPTY; + } + + @Override + public List listCommittedSnapshots() { + if (lastCommittedSnapshot != null) { + return List.of(lastCommittedSnapshot); + } + return List.of(); + } + + @Override + public void deleteCommit(CatalogSnapshot snapshot) {} + + @Override + public boolean isCommitManagedFile(String fileName) { + return false; + } + + @Override + public void close() {} + } + + // ---- Store and engine creation helpers ---- + + private Store createStore() throws IOException { + Directory dir = newDirectory(); + IndexSettings indexSettings = IndexSettingsModule.newIndexSettings( + "test", + Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) + .build() + ); + Path path = createTempDir().resolve(shardId.getIndex().getUUID()).resolve(String.valueOf(shardId.id())); + ShardPath shardPath = new ShardPath(false, path, path, shardId); + return new Store( + shardId, + indexSettings, + dir, + new DummyShardLock(shardId), + Store.OnClose.EMPTY, + shardPath, + new FsDirectoryFactory() + ); + } + + private void bootstrapStore(Store store, String translogUUID) throws IOException { + try ( + IndexWriter writer = new IndexWriter( + store.directory(), + new IndexWriterConfig(Lucene.STANDARD_ANALYZER).setMergePolicy(NoMergePolicy.INSTANCE) + .setOpenMode(IndexWriterConfig.OpenMode.CREATE) + ) + ) { + Map commitData = new HashMap<>(); + commitData.put(Translog.TRANSLOG_UUID_KEY, translogUUID); + commitData.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(SequenceNumbers.NO_OPS_PERFORMED)); + commitData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(SequenceNumbers.NO_OPS_PERFORMED)); + commitData.put(Engine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID, "-1"); + commitData.put(Engine.HISTORY_UUID_KEY, UUID.randomUUID().toString()); + writer.setLiveCommitData(commitData.entrySet()); + writer.commit(); + } + } + + private DataFormatAwareEngine createEngine() throws IOException { + String uuid = Translog.createEmptyTranslog(translogPath, SequenceNumbers.NO_OPS_PERFORMED, shardId, primaryTerm.get()); + bootstrapStore(store, uuid); + return new DataFormatAwareEngine(buildEngineConfig()); + } + + private DataFormatAwareEngine reopenEngine() throws IOException { + return new DataFormatAwareEngine(buildEngineConfig()); + } + + private EngineConfig buildEngineConfig() { + IndexSettings indexSettings = IndexSettingsModule.newIndexSettings( + "test", + Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true) + .put(IndexSettings.PLUGGABLE_DATAFORMAT_ENABLED_SETTING.getKey(), true) + .put(IndexSettings.PLUGGABLE_DATAFORMAT_VALUE_SETTING.getKey(), mockDataFormat.name()) + .build() + ); + + TranslogConfig translogConfig = new TranslogConfig( + shardId, + translogPath, + indexSettings, + BigArrays.NON_RECYCLING_INSTANCE, + "", + false + ); + DataFormatRegistry registry = createMockRegistry(); + CommitterFactory committerFactory = config -> new PersistentCommitter(store); + + return new EngineConfig.Builder().shardId(shardId) + .threadPool(threadPool) + .indexSettings(indexSettings) + .store(store) + .mergePolicy(NoMergePolicy.INSTANCE) + .translogConfig(translogConfig) + .flushMergesAfter(TimeValue.timeValueMinutes(5)) + .externalRefreshListener(List.of()) + .internalRefreshListener(List.of()) + .globalCheckpointSupplier(globalCheckpoint::get) + .retentionLeasesSupplier(() -> RetentionLeases.EMPTY) + .primaryTermSupplier(primaryTerm::get) + .tombstoneDocSupplier(tombstoneDocSupplier()) + .dataFormatRegistry(registry) + .committerFactory(committerFactory) + .build(); + } + + private DataFormatRegistry createMockRegistry() { + PluginsService pluginsService = mock(PluginsService.class); + when(pluginsService.filterPlugins(DataFormatPlugin.class)).thenReturn(List.of(mockPlugin)); + when(pluginsService.filterPlugins(SearchBackEndPlugin.class)).thenReturn( + List.of(new MockSearchBackEndPlugin(List.of(mockDataFormat.name()))) + ); + return new DataFormatRegistry(pluginsService); + } + + private Engine.Index indexOp(String id) { + ParsedDocument doc = createParsedDocWithInput(id); + return new Engine.Index( + new Term(IdFieldMapper.NAME, Uid.encodeId(doc.id())), + doc, + SequenceNumbers.UNASSIGNED_SEQ_NO, + primaryTerm.get(), + Versions.MATCH_ANY, + VersionType.INTERNAL, + Engine.Operation.Origin.PRIMARY, + System.nanoTime(), + -1, + false, + SequenceNumbers.UNASSIGNED_SEQ_NO, + 0 + ); + } + + private Engine.Index replayOp(Translog.Index translogOp) { + ParsedDocument doc = createParsedDocWithInput(translogOp.id()); + return new Engine.Index( + new Term(IdFieldMapper.NAME, Uid.encodeId(doc.id())), + doc, + translogOp.seqNo(), + translogOp.primaryTerm(), + translogOp.version(), + null, + Engine.Operation.Origin.LOCAL_TRANSLOG_RECOVERY, + System.nanoTime(), + translogOp.getAutoGeneratedIdTimestamp(), + false, + SequenceNumbers.UNASSIGNED_SEQ_NO, + 0 + ); + } + + private ParsedDocument createParsedDocWithInput(String id) { + ParsedDocument base = createParsedDoc(id, null); + return new ParsedDocument( + base.version(), + SeqNoFieldMapper.SequenceIDFields.emptySeqID(), + base.id(), + base.routing(), + base.docs(), + base.source(), + base.getMediaType(), + null, + new MockDocumentInput() + ); + } + + private int recoverFromTranslog(DataFormatAwareEngine engine) throws IOException { + int recovered = engine.translogManager().recoverFromTranslog(snapshot -> { + int ops = 0; + Translog.Operation op; + while ((op = snapshot.next()) != null) { + if (op instanceof Translog.Index) { + engine.index(replayOp((Translog.Index) op)); + ops++; + } + } + return ops; + }, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + // Advance global checkpoint after recovery (simulates single-node cluster) + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + return recovered; + } + + // ---- Recovery Tests ---- + + /** + * Mirrors InternalEngineTests.testTranslogRecoveryDoesNotReplayIntoTranslog (L1024). + * + * Index docs, close engine (without flush — all ops are in translog only), + * reopen, recover. Verify: + * - Translog has uncommitted ops before recovery + * - After recovery, ops are committed (flush triggered by onAfterTranslogRecovery) + * - Replayed ops are NOT double-written to translog + */ + public void testTranslogRecoveryDoesNotReplayIntoTranslog() throws IOException { + final int docs = randomIntBetween(1, 32); + + // Index docs and close WITHOUT flush — all ops stay in translog + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + // Advance global checkpoint to match local (simulates single-node replication) + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + // DO NOT flush — close with data only in translog + } + + // Reopen — translog should have all ops as uncommitted + try (DataFormatAwareEngine engine = reopenEngine()) { + TranslogStats stats = engine.translogManager().getTranslogStats(); + assertThat("all ops should be uncommitted in translog", stats.getUncommittedOperations(), equalTo(docs)); + + // Recover — this replays ops and triggers flush via onAfterTranslogRecovery + int recovered = recoverFromTranslog(engine); + assertThat("should recover all docs", recovered, equalTo(docs)); + + // After recovery, checkpoint should advance + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs - 1)); + + // Translog should be trimmed after recovery flush + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + TranslogStats afterStats = engine.translogManager().getTranslogStats(); + assertThat("translog should be trimmed after recovery flush", afterStats.getUncommittedOperations(), equalTo(0)); + } + } + + /** + * Mirrors InternalEngineTests.testTranslogRecoveryWithMultipleGenerations (L1059). + * + * Index docs with random flushes and translog rolls, close, reopen, recover. + * Verify all docs are recovered and checkpoint is correct. + */ + public void testTranslogRecoveryWithMultipleGenerations() throws IOException { + final int docs = randomIntBetween(10, 100); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + if (rarely()) { + engine.translogManager().rollTranslogGeneration(); + } else if (rarely()) { + engine.flush(randomBoolean(), true); + } + } + } + + // Reopen and recover + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = recoverFromTranslog(engine); + + // Checkpoint must reflect all docs + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs - 1)); + assertThat(engine.getSeqNoStats(-1).getMaxSeqNo(), equalTo((long) docs - 1)); + + // After recovery + refresh, segments should exist in the catalog snapshot + engine.refresh("test"); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat("snapshot should have segments after recovery", snapshot.getSegments().size(), greaterThan(0)); + } + } + } + + /** + * Mirrors InternalEngineTests.testRecoveryFromTranslogUpToSeqNo (L1099). + * + * Index docs, recover up to a partial seqNo (simulating recovery up to globalCheckpoint). + * Verify checkpoint advances only to the recovered seqNo. + */ + public void testRecoveryFromTranslogUpToSeqNo() throws IOException { + final int docs = randomIntBetween(10, 50); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + // No flush — close with everything in translog + } + + // Recover up to a partial seqNo + final long recoverUpTo = randomLongBetween(0, docs - 1); + + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = engine.translogManager().recoverFromTranslog(snapshot -> { + int ops = 0; + Translog.Operation op; + while ((op = snapshot.next()) != null) { + if (op instanceof Translog.Index) { + engine.index(replayOp((Translog.Index) op)); + ops++; + } + } + return ops; + }, engine.getProcessedLocalCheckpoint(), recoverUpTo); + + assertThat(engine.getProcessedLocalCheckpoint(), equalTo(recoverUpTo)); + assertThat(recovered, equalTo((int) (recoverUpTo + 1))); + } + } + + /** + * DFA-specific: Index docs, flush (commit), index more without flush, close, reopen. + * Only the unflushed docs should be replayed from translog. + */ + public void testRecoveryOnlyReplaysUnflushedOps() throws IOException { + final int flushedDocs = randomIntBetween(5, 20); + final int unflushedDocs = randomIntBetween(5, 20); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + + // Index and flush — these are committed + for (int i = 0; i < flushedDocs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + engine.refresh("before-flush"); + engine.flush(false, true); + + // Index more WITHOUT flush — these are only in translog + for (int i = flushedDocs; i < flushedDocs + unflushedDocs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + // Reopen — should only need to replay unflushedDocs + try (DataFormatAwareEngine engine = reopenEngine()) { + TranslogStats stats = engine.translogManager().getTranslogStats(); + assertThat("only unflushed ops should be uncommitted", stats.getUncommittedOperations(), equalTo(unflushedDocs)); + + // The committed local checkpoint should be at flushedDocs - 1 + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) flushedDocs - 1)); + + // Recover + int recovered = recoverFromTranslog(engine); + assertThat("should recover only unflushed docs", recovered, equalTo(unflushedDocs)); + + // After recovery, checkpoint should cover all docs + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) (flushedDocs + unflushedDocs - 1))); + } + } + + /** + * DFA-specific: Verify that after recovery and refresh, the catalog snapshot + * contains segments with the correct structure. + */ + public void testRecoveryProducesValidCatalogSnapshot() throws IOException { + final int docs = randomIntBetween(5, 30); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + engine.refresh("test"); + + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat(snapshot, org.hamcrest.Matchers.notNullValue()); + assertThat(snapshot.getSegments().size(), greaterThan(0)); + + // Snapshot should have files for our format + Map> filesByFormat = snapshot.getFilesByFormat(); + assertThat("snapshot should have files for the data format", filesByFormat.size(), greaterThanOrEqualTo(1)); + } + } + } + + /** + * DFA-specific: After recovery, verify the engine is fully functional — + * can index new docs, refresh, and flush without errors. + */ + public void testEngineIsFunctionalAfterRecovery() throws IOException { + final int initialDocs = randomIntBetween(5, 20); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < initialDocs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + + // Engine should accept new writes after recovery + int newDocs = randomIntBetween(5, 20); + for (int i = initialDocs; i < initialDocs + newDocs; i++) { + Engine.IndexResult result = engine.index(indexOp(Integer.toString(i))); + assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); + } + + // Refresh should produce segments + engine.refresh("after-new-writes"); + try (GatedCloseable ref = engine.acquireSnapshot()) { + assertThat(ref.get().getSegments().size(), greaterThan(0)); + } + + // Flush should succeed + engine.flush(false, true); + + // Final checkpoint should reflect all docs + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) (initialDocs + newDocs - 1))); + } + } + + /** + * DFA-specific: Verify that the committed data after recovery contains + * all required metadata keys for subsequent recovery. + */ + public void testRecoveryFlushWritesCorrectCommitData() throws IOException { + final int docs = randomIntBetween(5, 20); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + // onAfterTranslogRecovery triggers flush, which writes commit data + + // Read back the committed data (written by the PersistentCommitter to Lucene) + Map commitData = store.readLastCommittedSegmentsInfo().getUserData(); + + assertThat(commitData.get(Translog.TRANSLOG_UUID_KEY), org.hamcrest.Matchers.notNullValue()); + assertThat(commitData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY), equalTo(Long.toString(docs - 1))); + assertThat(commitData.get(SequenceNumbers.MAX_SEQ_NO), equalTo(Long.toString(docs - 1))); + assertThat(commitData.get(Engine.HISTORY_UUID_KEY), org.hamcrest.Matchers.notNullValue()); + } + } + + /** + * Verify double-restart works: index → close → recover → close → recover. + * The second recovery should have nothing to replay (since the first recovery flushed). + */ + public void testDoubleRecovery() throws IOException { + final int docs = randomIntBetween(5, 20); + + // First lifecycle: index docs, close without flush + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + // Second lifecycle: recover (flush happens), close + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs - 1)); + } + + // Third lifecycle: should have nothing to recover + try (DataFormatAwareEngine engine = reopenEngine()) { + TranslogStats stats = engine.translogManager().getTranslogStats(); + assertThat("nothing to recover after prior recovery flushed", stats.getUncommittedOperations(), equalTo(0)); + + int recovered = recoverFromTranslog(engine); + assertThat("no ops should be recovered", recovered, equalTo(0)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs - 1)); + } + } + + /** + * Regression test: restart with committed data and empty translog. + * The reader managers must be notified about existing committed segments + * even though no translog replay or refresh occurs. + * + * Without the fix in CatalogSnapshotManager, the afterRefresh callback + * would never fire and reader managers would have no open readers. + */ + public void testReaderManagersInitializedOnRestartWithCommittedData() throws IOException { + final int docs = randomIntBetween(5, 20); + + // First lifecycle: index docs, flush (commit), close cleanly + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + engine.refresh("before-flush"); + engine.flush(false, true); + // Everything is committed — translog is empty after flush + } + + // Second lifecycle: reopen with committed data, empty translog + try (DataFormatAwareEngine engine = reopenEngine()) { + // No translog ops to replay + int recovered = recoverFromTranslog(engine); + assertThat("nothing to recover — all committed", recovered, equalTo(0)); + + // The catalog snapshot should have segments from the committed data + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat("committed snapshot must have segments", snapshot.getSegments().size(), greaterThan(0)); + // Verify files exist in the snapshot + Map> filesByFormat = snapshot.getFilesByFormat(); + assertThat("committed snapshot must have format files", filesByFormat.isEmpty(), equalTo(false)); + } + + // Engine should be fully functional for new writes + Engine.IndexResult result = engine.index(indexOp(Integer.toString(docs))); + assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs)); + } + } + + /** + * Multi-threaded indexing before close, then recovery. Verifies no sequence + * gaps exist after recovery — the checkpoint must be contiguous. + */ + public void testConcurrentIndexingThenRecoveryPreservesCheckpoint() throws Exception { + final int numThreads = randomIntBetween(2, 4); + final int docsPerThread = randomIntBetween(10, 25); + final int totalDocs = numThreads * docsPerThread; + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + + CyclicBarrier barrier = new CyclicBarrier(numThreads); + AtomicInteger failures = new AtomicInteger(0); + + Thread[] threads = new Thread[numThreads]; + for (int t = 0; t < numThreads; t++) { + final int threadId = t; + threads[t] = new Thread(() -> { + try { + barrier.await(); + for (int d = 0; d < docsPerThread; d++) { + engine.index(indexOp(threadId + "_" + d)); + } + } catch (Exception e) { + failures.incrementAndGet(); + } + }); + threads[t].start(); + } + for (Thread t : threads) { + t.join(); + } + assertThat(failures.get(), equalTo(0)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) totalDocs - 1)); + } + + // Reopen and recover — checkpoint must be contiguous with no gaps + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = recoverFromTranslog(engine); + assertThat(recovered, equalTo(totalDocs)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) totalDocs - 1)); + + // Verify seq-no stats are consistent + assertThat(engine.getSeqNoStats(-1).getMaxSeqNo(), equalTo((long) totalDocs - 1)); + } + } + + /** + * After recovery, subsequent refreshes must produce monotonically increasing + * snapshot generations — no reuse or regression. + */ + public void testSnapshotGenerationAdvancesMonotonicallyAfterRecovery() throws IOException { + final int docs = randomIntBetween(5, 20); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + + long prevGen = -1; + int numRefreshes = randomIntBetween(3, 6); + for (int r = 0; r < numRefreshes; r++) { + engine.index(indexOp("post_recovery_" + r)); + engine.refresh("refresh-" + r); + + try (GatedCloseable ref = engine.acquireSnapshot()) { + long currentGen = ref.get().getGeneration(); + assertThat("generation must strictly increase", currentGen, greaterThan(prevGen)); + prevGen = currentGen; + } + } + } + } + + /** + * After recovery and refresh, docStats().getCount() must equal the total number + * of indexed documents (committed + recovered). + */ + public void testDocStatsAfterRecoveryReflectsAllDocs() throws IOException { + final int flushedDocs = randomIntBetween(5, 15); + final int unflushedDocs = randomIntBetween(5, 15); + final int totalDocs = flushedDocs + unflushedDocs; + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + + for (int i = 0; i < flushedDocs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + engine.refresh("before-flush"); + engine.flush(false, true); + + for (int i = flushedDocs; i < totalDocs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + engine.refresh("post-recovery"); + + assertThat(engine.docStats().getCount(), equalTo((long) totalDocs)); + } + } + + /** + * Interleave indexing with random refreshes and flushes, then close and recover. + * Verifies no data loss under the mixed workload. + */ + public void testInterleavedIndexRefreshFlushThenRecover() throws IOException { + final int docs = randomIntBetween(20, 60); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + if (randomIntBetween(1, 10) == 1) { + engine.refresh("interleaved-refresh-" + i); + } + if (randomIntBetween(1, 15) == 1) { + engine.flush(false, true); + } + } + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = recoverFromTranslog(engine); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) docs - 1)); + + engine.refresh("post-recovery"); + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat(snapshot.getSegments().size(), greaterThan(0)); + + long totalRows = snapshot.getSegments() + .stream() + .flatMap(s -> s.dfGroupedSearchableFiles().values().stream()) + .mapToLong(org.opensearch.index.engine.exec.WriterFileSet::numRows) + .sum(); + assertThat("total rows must equal total docs indexed", totalRows, equalTo((long) docs)); + } + } + } + + /** + * After recovery, verify that a newChangesSnapshot contains all ops for the + * full seq-no range — confirms no operations were lost during the recovery cycle. + */ + public void testTranslogSnapshotContainsAllOpsAfterRecovery() throws IOException { + final int docs = randomIntBetween(10, 40); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + + try (Translog.Snapshot snapshot = engine.newChangesSnapshot("test", 0, docs - 1, false, true)) { + int count = 0; + Translog.Operation op; + while ((op = snapshot.next()) != null) { + assertThat(op.seqNo(), greaterThanOrEqualTo(0L)); + count++; + } + assertThat("all ops must be present in translog snapshot after recovery", count, equalTo(docs)); + } + } + } + + /** + * Concurrent indexing followed by flush, then recovery. Verifies the committed + * catalog snapshot survives and is correctly restored. + */ + public void testConcurrentIndexThenFlushAndRecover() throws Exception { + final int numThreads = randomIntBetween(2, 4); + final int docsPerThread = randomIntBetween(10, 20); + final int totalDocs = numThreads * docsPerThread; + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + + CyclicBarrier barrier = new CyclicBarrier(numThreads); + AtomicInteger failures = new AtomicInteger(0); + + Thread[] threads = new Thread[numThreads]; + for (int t = 0; t < numThreads; t++) { + final int threadId = t; + threads[t] = new Thread(() -> { + try { + barrier.await(); + for (int d = 0; d < docsPerThread; d++) { + engine.index(indexOp(threadId + "_" + d)); + } + } catch (Exception e) { + failures.incrementAndGet(); + } + }); + threads[t].start(); + } + for (Thread t : threads) { + t.join(); + } + assertThat(failures.get(), equalTo(0)); + + engine.refresh("before-flush"); + engine.flush(false, true); + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + } + + // Reopen — committed data, empty translog + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = recoverFromTranslog(engine); + assertThat("all committed — nothing to recover from translog", recovered, equalTo(0)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) totalDocs - 1)); + + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat("committed snapshot must have segments", snapshot.getSegments().size(), greaterThan(0)); + + long totalRows = snapshot.getSegments() + .stream() + .flatMap(s -> s.dfGroupedSearchableFiles().values().stream()) + .mapToLong(org.opensearch.index.engine.exec.WriterFileSet::numRows) + .sum(); + assertThat("total rows must match total docs", totalRows, equalTo((long) totalDocs)); + } + } + } + + /** + * Regression test for the CatalogSnapshotManager fix: multiple restart cycles + * with flush each time must correctly propagate the catalog snapshot. + * + * Lifecycle: index→flush→close→reopen→index→flush→close→reopen. Each reopen + * must see the accumulated data from all prior lifecycles. + */ + public void testMultipleFlushRestartCyclesAccumulateData() throws IOException { + int accumulatedDocs = 0; + + // First lifecycle + int firstBatch = randomIntBetween(5, 15); + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < firstBatch; i++) { + engine.index(indexOp(Integer.toString(accumulatedDocs + i))); + } + accumulatedDocs += firstBatch; + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + engine.refresh("flush-1"); + engine.flush(false, true); + } + + // Second lifecycle: recover, add more, flush + int secondBatch = randomIntBetween(5, 15); + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + for (int i = 0; i < secondBatch; i++) { + engine.index(indexOp(Integer.toString(accumulatedDocs + i))); + } + accumulatedDocs += secondBatch; + globalCheckpoint.set(engine.getProcessedLocalCheckpoint()); + engine.refresh("flush-2"); + engine.flush(false, true); + } + + // Third lifecycle: verify accumulated snapshot + final int expectedTotal = accumulatedDocs; + try (DataFormatAwareEngine engine = reopenEngine()) { + int recovered = recoverFromTranslog(engine); + assertThat("all committed — nothing to recover", recovered, equalTo(0)); + assertThat(engine.getProcessedLocalCheckpoint(), equalTo((long) expectedTotal - 1)); + + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + assertThat(snapshot.getSegments().size(), greaterThan(0)); + + long totalRows = snapshot.getSegments() + .stream() + .flatMap(s -> s.dfGroupedSearchableFiles().values().stream()) + .mapToLong(org.opensearch.index.engine.exec.WriterFileSet::numRows) + .sum(); + assertThat("snapshot must reflect all docs across lifecycles", totalRows, equalTo((long) expectedTotal)); + } + } + } + + /** + * After recovery, each refresh must produce segments with unique generations. + * No two segments in a snapshot should share the same generation. + */ + public void testSegmentGenerationsAreUniqueAfterRecovery() throws IOException { + final int docs = randomIntBetween(10, 30); + + try (DataFormatAwareEngine engine = createEngine()) { + engine.translogManager().recoverFromTranslog(ignore -> 0, engine.getProcessedLocalCheckpoint(), Long.MAX_VALUE); + for (int i = 0; i < docs; i++) { + engine.index(indexOp(Integer.toString(i))); + } + } + + try (DataFormatAwareEngine engine = reopenEngine()) { + recoverFromTranslog(engine); + + // Do multiple refresh cycles + int numCycles = randomIntBetween(3, 5); + for (int c = 0; c < numCycles; c++) { + engine.index(indexOp("post_" + c)); + engine.refresh("cycle-" + c); + } + + try (GatedCloseable ref = engine.acquireSnapshot()) { + CatalogSnapshot snapshot = ref.get(); + List segments = snapshot.getSegments(); + long distinctGenerations = segments.stream().map(org.opensearch.index.engine.exec.Segment::generation).distinct().count(); + assertThat("all segment generations must be unique", distinctGenerations, equalTo((long) segments.size())); + } + } + } +}