Add test for snapshot incrementality of snapshot-backed indices (#62550)

This commit adds a test that verifies that snapshots incrementality 
is respected when a snapshot-backed index is snapshotted. This 
test mounts a snapshot as a snapshot-backed index, creates a 
new snapshot from it and then verifies that no new data blobs 
were added to the repository.

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/BaseSearchableSnapshotsIntegTestCase.java

@@ -125,14 +125,29 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends ESIntegTestCa
         return snapshotInfo.snapshotId();
     }
 
-    protected String mountSnapshot(String repositoryName, String snapshotName, String indexName, Settings indexSettings) throws Exception {
+    protected String mountSnapshot(String repositoryName, String snapshotName, String indexName, Settings restoredIndexSettings)
+        throws Exception {
         final String restoredIndexName = randomBoolean() ? indexName : randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        mountSnapshot(repositoryName, snapshotName, indexName, restoredIndexName, restoredIndexSettings);
+        return restoredIndexName;
+    }
+
+    protected void mountSnapshot(
+        String repositoryName,
+        String snapshotName,
+        String indexName,
+        String restoredIndexName,
+        Settings restoredIndexSettings
+    ) throws Exception {
         final MountSearchableSnapshotRequest mountRequest = new MountSearchableSnapshotRequest(
             restoredIndexName,
             repositoryName,
             snapshotName,
             indexName,
-            Settings.builder().put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), Boolean.FALSE.toString()).put(indexSettings).build(),
+            Settings.builder()
+                .put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), Boolean.FALSE.toString())
+                .put(restoredIndexSettings)
+                .build(),
             Strings.EMPTY_ARRAY,
             true
         );
@@ -140,7 +155,6 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends ESIntegTestCa
         final RestoreSnapshotResponse restoreResponse = client().execute(MountSearchableSnapshotAction.INSTANCE, mountRequest).get();
         assertThat(restoreResponse.getRestoreInfo().successfulShards(), equalTo(getNumShards(restoredIndexName).numPrimaries));
         assertThat(restoreResponse.getRestoreInfo().failedShards(), equalTo(0));
-        return restoredIndexName;
     }
 
     protected void createRepo(String fsRepoName) {

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -11,6 +11,7 @@ import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStatus;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
 import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
 import org.elasticsearch.action.admin.indices.shrink.ResizeType;
@@ -39,6 +40,7 @@ import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.recovery.RecoveryState;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.fs.FsRepository;
+import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
@@ -660,6 +662,52 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         }
     }
 
+    public void testSnapshotMountedIndexLeavesBlobsUntouched() throws Exception {
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final int numShards = between(1, 3);
+        createAndPopulateIndex(
+            indexName,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, numShards)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put(INDEX_SOFT_DELETES_SETTING.getKey(), true)
+        );
+        ensureGreen(indexName);
+        forceMerge();
+
+        final String repositoryName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final Path repositoryLocation = randomRepoPath();
+        createFsRepository(repositoryName, repositoryLocation);
+
+        final SnapshotId snapshotOne = createSnapshot(repositoryName, List.of(indexName));
+        assertAcked(client().admin().indices().prepareDelete(indexName));
+
+        final SnapshotStatus snapshotOneStatus = client().admin()
+            .cluster()
+            .prepareSnapshotStatus(repositoryName)
+            .setSnapshots(snapshotOne.getName())
+            .get()
+            .getSnapshots()
+            .get(0);
+        final int snapshotOneTotalFileCount = snapshotOneStatus.getStats().getTotalFileCount();
+        assertThat(snapshotOneTotalFileCount, greaterThan(0));
+
+        mountSnapshot(repositoryName, snapshotOne.getName(), indexName, indexName, Settings.EMPTY);
+        ensureGreen(indexName);
+
+        final SnapshotId snapshotTwo = createSnapshot(repositoryName, List.of(indexName));
+        final SnapshotStatus snapshotTwoStatus = client().admin()
+            .cluster()
+            .prepareSnapshotStatus(repositoryName)
+            .setSnapshots(snapshotTwo.getName())
+            .get()
+            .getSnapshots()
+            .get(0);
+        assertThat(snapshotTwoStatus.getStats().getTotalFileCount(), equalTo(snapshotOneTotalFileCount));
+        assertThat(snapshotTwoStatus.getStats().getIncrementalFileCount(), equalTo(numShards)); // one segment_N per shard
+        assertThat(snapshotTwoStatus.getStats().getProcessedFileCount(), equalTo(numShards)); // one segment_N per shard
+    }
+
     private void assertTotalHits(String indexName, TotalHits originalAllHits, TotalHits originalBarHits) throws Exception {
         final Thread[] threads = new Thread[between(1, 5)];
         final AtomicArray<TotalHits> allHits = new AtomicArray<>(threads.length);