diff --git a/CHANGELOG.md b/CHANGELOG.md
index d948d1bc0e27f..ab794593f6966 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - BWC version 2.2.2 ([#4383](https://github.com/opensearch-project/OpenSearch/pull/4383))
 - Support for labels on version bump PRs, skip label support for changelog verifier ([#4391](https://github.com/opensearch-project/OpenSearch/pull/4391))
 - Add a new node role 'search' which is dedicated to provide search capability ([#4689](https://github.com/opensearch-project/OpenSearch/pull/4689))
+- Introduce experimental searchable snapshot API ([#4680](https://github.com/opensearch-project/OpenSearch/pull/4680))
 
 ### Dependencies
 - Bumps `com.diffplug.spotless` from 6.9.1 to 6.10.0
diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java
new file mode 100644
index 0000000000000..96fcf0053c9ab
--- /dev/null
+++ b/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.snapshots;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+import org.hamcrest.MatcherAssert;
+import org.junit.BeforeClass;
+import org.opensearch.action.admin.cluster.node.stats.NodesStatsResponse;
+import org.opensearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
+import org.opensearch.action.admin.cluster.snapshots.restore.RestoreSnapshotRequest;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.routing.GroupShardsIterator;
+import org.opensearch.cluster.routing.ShardIterator;
+import org.opensearch.cluster.routing.ShardRouting;
+import org.opensearch.common.io.PathUtils;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.index.Index;
+import org.opensearch.monitor.fs.FsInfo;
+
+import com.carrotsearch.randomizedtesting.generators.RandomPicks;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.is;
+import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.FS;
+import static org.opensearch.common.util.CollectionUtils.iterableAsArrayList;
+
+public final class SearchableSnapshotIT extends AbstractSnapshotIntegTestCase {
+
+    @BeforeClass
+    public static void assumeFeatureFlag() {
+        assumeTrue(
+            "Searchable snapshot feature flag is enabled",
+            Boolean.parseBoolean(System.getProperty(FeatureFlags.SEARCHABLE_SNAPSHOT))
+        );
+    }
+
+    @Override
+    protected boolean addMockInternalEngine() {
+        return false;
+    }
+
+    public void testCreateSearchableSnapshot() throws Exception {
+        final Client client = client();
+        createRepository("test-repo", "fs");
+        createIndex(
+            "test-idx-1",
+            Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, "0").put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, "1").build()
+        );
+        createIndex(
+            "test-idx-2",
+            Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, "0").put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, "1").build()
+        );
+        ensureGreen();
+        indexRandomDocs("test-idx-1", 100);
+        indexRandomDocs("test-idx-2", 100);
+
+        logger.info("--> snapshot");
+        final CreateSnapshotResponse createSnapshotResponse = client.admin()
+            .cluster()
+            .prepareCreateSnapshot("test-repo", "test-snap")
+            .setWaitForCompletion(true)
+            .setIndices("test-idx-1", "test-idx-2")
+            .get();
+        MatcherAssert.assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), greaterThan(0));
+        MatcherAssert.assertThat(
+            createSnapshotResponse.getSnapshotInfo().successfulShards(),
+            equalTo(createSnapshotResponse.getSnapshotInfo().totalShards())
+        );
+
+        assertTrue(client.admin().indices().prepareDelete("test-idx-1", "test-idx-2").get().isAcknowledged());
+
+        logger.info("--> restore indices as 'remote_snapshot'");
+        client.admin()
+            .cluster()
+            .prepareRestoreSnapshot("test-repo", "test-snap")
+            .setRenamePattern("(.+)")
+            .setRenameReplacement("$1-copy")
+            .setStorageType(RestoreSnapshotRequest.StorageType.REMOTE_SNAPSHOT)
+            .setWaitForCompletion(true)
+            .execute()
+            .actionGet();
+        ensureGreen();
+
+        assertDocCount("test-idx-1-copy", 100L);
+        assertDocCount("test-idx-2-copy", 100L);
+        assertIndexDirectoryDoesNotExist("test-idx-1-copy", "test-idx-2-copy");
+    }
+
+    /**
+     * Picks a shard out of the cluster state for each given index and asserts
+     * that the 'index' directory does not exist in the node's file system.
+     * This assertion is digging a bit into the implementation details to
+     * verify that the Lucene segment files are not copied from the snapshot
+     * repository to the node's local disk for a remote snapshot index.
+     */
+    private void assertIndexDirectoryDoesNotExist(String... indexNames) {
+        final ClusterState state = client().admin().cluster().prepareState().get().getState();
+        for (String indexName : indexNames) {
+            final Index index = state.metadata().index(indexName).getIndex();
+            // Get the primary shards for the given index
+            final GroupShardsIterator<ShardIterator> shardIterators = state.getRoutingTable()
+                .activePrimaryShardsGrouped(new String[] { indexName }, false);
+            // Randomly pick one of the shards
+            final List<ShardIterator> iterators = iterableAsArrayList(shardIterators);
+            final ShardIterator shardIterator = RandomPicks.randomFrom(random(), iterators);
+            final ShardRouting shardRouting = shardIterator.nextOrNull();
+            assertNotNull(shardRouting);
+            assertTrue(shardRouting.primary());
+            assertTrue(shardRouting.assignedToNode());
+            // Get the file system stats for the assigned node
+            final String nodeId = shardRouting.currentNodeId();
+            final NodesStatsResponse nodeStats = client().admin().cluster().prepareNodesStats(nodeId).addMetric(FS.metricName()).get();
+            for (FsInfo.Path info : nodeStats.getNodes().get(0).getFs()) {
+                // Build the expected path for the index data for a "normal"
+                // index and assert it does not exist
+                final String path = info.getPath();
+                final Path file = PathUtils.get(path)
+                    .resolve("indices")
+                    .resolve(index.getUUID())
+                    .resolve(Integer.toString(shardRouting.getId()))
+                    .resolve("index");
+                MatcherAssert.assertThat("Expect file not to exist: " + file, Files.exists(file), is(false));
+            }
+        }
+    }
+}
diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java
index 1b673217a248b..3ecf5ab19c0e4 100644
--- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java
+++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java
@@ -33,6 +33,7 @@
 package org.opensearch.action.admin.cluster.snapshots.restore;
 
 import org.opensearch.LegacyESVersion;
+import org.opensearch.Version;
 import org.opensearch.action.ActionRequestValidationException;
 import org.opensearch.action.support.IndicesOptions;
 import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest;
@@ -42,6 +43,7 @@
 import org.opensearch.common.io.stream.StreamOutput;
 import org.opensearch.common.logging.DeprecationLogger;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.xcontent.ToXContentObject;
 import org.opensearch.common.xcontent.XContentBuilder;
 import org.opensearch.common.xcontent.XContentType;
@@ -68,6 +70,38 @@ public class RestoreSnapshotRequest extends ClusterManagerNodeRequest<RestoreSna
 
     private static final DeprecationLogger DEPRECATION_LOGGER = DeprecationLogger.getLogger(RestoreSnapshotRequest.class);
 
+    /**
+     * Enumeration of possible storage types
+     */
+    public enum StorageType {
+        LOCAL("local"),
+        REMOTE_SNAPSHOT("remote_snapshot");
+
+        private final String text;
+
+        StorageType(String text) {
+            this.text = text;
+        }
+
+        @Override
+        public String toString() {
+            return text;
+        }
+
+        private void toXContent(XContentBuilder builder) throws IOException {
+            builder.field("storage_type", text);
+        }
+
+        private static StorageType fromString(String string) {
+            for (StorageType type : values()) {
+                if (type.text.equals(string)) {
+                    return type;
+                }
+            }
+            throw new IllegalArgumentException("Invalid storage_type: " + string);
+        }
+    }
+
     private String snapshot;
     private String repository;
     private String[] indices = Strings.EMPTY_ARRAY;
@@ -80,6 +114,7 @@ public class RestoreSnapshotRequest extends ClusterManagerNodeRequest<RestoreSna
     private boolean includeAliases = true;
     private Settings indexSettings = EMPTY_SETTINGS;
     private String[] ignoreIndexSettings = Strings.EMPTY_ARRAY;
+    private StorageType storageType = StorageType.LOCAL;
 
     @Nullable // if any snapshot UUID will do
     private String snapshotUuid;
@@ -117,6 +152,9 @@ public RestoreSnapshotRequest(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(LegacyESVersion.V_7_10_0)) {
             snapshotUuid = in.readOptionalString();
         }
+        if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && in.getVersion().onOrAfter(Version.V_3_0_0)) {
+            storageType = in.readEnum(StorageType.class);
+        }
     }
 
     @Override
@@ -144,6 +182,9 @@ public void writeTo(StreamOutput out) throws IOException {
                 "restricting the snapshot UUID is forbidden in a cluster with version [" + out.getVersion() + "] nodes"
             );
         }
+        if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && out.getVersion().onOrAfter(Version.V_3_0_0)) {
+            out.writeEnum(storageType);
+        }
     }
 
     @Override
@@ -480,6 +521,22 @@ public String snapshotUuid() {
         return snapshotUuid;
     }
 
+    /**
+     * Sets the storage type for this request.
+     */
+    RestoreSnapshotRequest storageType(StorageType storageType) {
+        this.storageType = storageType;
+        return this;
+    }
+
+    /**
+     * Gets the storage type for this request. {@link StorageType#LOCAL} is the
+     * implicit default if not overridden.
+     */
+    public StorageType storageType() {
+        return storageType;
+    }
+
     /**
      * Parses restore definition
      *
@@ -537,6 +594,18 @@ public RestoreSnapshotRequest source(Map<String, Object> source) {
                 } else {
                     throw new IllegalArgumentException("malformed ignore_index_settings section, should be an array of strings");
                 }
+            } else if (name.equals("storage_type")) {
+                if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT)) {
+                    if (entry.getValue() instanceof String) {
+                        storageType(StorageType.fromString((String) entry.getValue()));
+                    } else {
+                        throw new IllegalArgumentException("malformed storage_type");
+                    }
+                } else {
+                    throw new IllegalArgumentException(
+                        "Unsupported parameter " + name + ". Feature flag is not enabled for this experimental feature"
+                    );
+                }
             } else {
                 if (IndicesOptions.isIndicesOptions(name) == false) {
                     throw new IllegalArgumentException("Unknown parameter " + name);
@@ -579,6 +648,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.value(ignoreIndexSetting);
         }
         builder.endArray();
+        if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && storageType != null) {
+            storageType.toXContent(builder);
+        }
         builder.endObject();
         return builder;
     }
@@ -605,7 +677,8 @@ public boolean equals(Object o) {
             && Objects.equals(renameReplacement, that.renameReplacement)
             && Objects.equals(indexSettings, that.indexSettings)
             && Arrays.equals(ignoreIndexSettings, that.ignoreIndexSettings)
-            && Objects.equals(snapshotUuid, that.snapshotUuid);
+            && Objects.equals(snapshotUuid, that.snapshotUuid)
+            && Objects.equals(storageType, that.storageType);
     }
 
     @Override
@@ -621,7 +694,8 @@ public int hashCode() {
             partial,
             includeAliases,
             indexSettings,
-            snapshotUuid
+            snapshotUuid,
+            storageType
         );
         result = 31 * result + Arrays.hashCode(indices);
         result = 31 * result + Arrays.hashCode(ignoreIndexSettings);
diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java
index 68397851699fb..0104637a00035 100644
--- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java
+++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java
@@ -248,4 +248,12 @@ public RestoreSnapshotRequestBuilder setIgnoreIndexSettings(List<String> ignoreI
         request.ignoreIndexSettings(ignoreIndexSettings);
         return this;
     }
+
+    /**
+     * Sets the storage type
+     */
+    public RestoreSnapshotRequestBuilder setStorageType(RestoreSnapshotRequest.StorageType storageType) {
+        request.storageType(storageType);
+        return this;
+    }
 }
diff --git a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
index 539773296ed74..728bf9d1ae90e 100644
--- a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
+++ b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java
@@ -38,6 +38,7 @@
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.io.stream.StreamOutput;
 import org.opensearch.common.io.stream.Writeable;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.xcontent.ToXContent;
 import org.opensearch.common.xcontent.ToXContentObject;
 import org.opensearch.common.xcontent.XContentBuilder;
@@ -257,12 +258,24 @@ public static class SnapshotRecoverySource extends RecoverySource {
         private final Snapshot snapshot;
         private final IndexId index;
         private final Version version;
+        private final boolean isSearchableSnapshot;
 
         public SnapshotRecoverySource(String restoreUUID, Snapshot snapshot, Version version, IndexId indexId) {
+            this(restoreUUID, snapshot, version, indexId, false);
+        }
+
+        public SnapshotRecoverySource(
+            String restoreUUID,
+            Snapshot snapshot,
+            Version version,
+            IndexId indexId,
+            boolean isSearchableSnapshot
+        ) {
             this.restoreUUID = restoreUUID;
             this.snapshot = Objects.requireNonNull(snapshot);
             this.version = Objects.requireNonNull(version);
             this.index = Objects.requireNonNull(indexId);
+            this.isSearchableSnapshot = isSearchableSnapshot;
         }
 
         SnapshotRecoverySource(StreamInput in) throws IOException {
@@ -274,6 +287,11 @@ public SnapshotRecoverySource(String restoreUUID, Snapshot snapshot, Version ver
             } else {
                 index = new IndexId(in.readString(), IndexMetadata.INDEX_UUID_NA_VALUE);
             }
+            if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && in.getVersion().onOrAfter(Version.V_3_0_0)) {
+                isSearchableSnapshot = in.readBoolean();
+            } else {
+                isSearchableSnapshot = false;
+            }
         }
 
         public String restoreUUID() {
@@ -298,6 +316,10 @@ public Version version() {
             return version;
         }
 
+        public boolean isSearchableSnapshot() {
+            return isSearchableSnapshot;
+        }
+
         @Override
         protected void writeAdditionalFields(StreamOutput out) throws IOException {
             out.writeString(restoreUUID);
@@ -308,6 +330,9 @@ protected void writeAdditionalFields(StreamOutput out) throws IOException {
             } else {
                 out.writeString(index.getName());
             }
+            if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && out.getVersion().onOrAfter(Version.V_3_0_0)) {
+                out.writeBoolean(isSearchableSnapshot);
+            }
         }
 
         @Override
@@ -321,7 +346,8 @@ public void addAdditionalFields(XContentBuilder builder, ToXContent.Params param
                 .field("snapshot", snapshot.getSnapshotId().getName())
                 .field("version", version.toString())
                 .field("index", index.getName())
-                .field("restoreUUID", restoreUUID);
+                .field("restoreUUID", restoreUUID)
+                .field("isSearchableSnapshot", isSearchableSnapshot);
         }
 
         @Override
@@ -342,12 +368,13 @@ public boolean equals(Object o) {
             return restoreUUID.equals(that.restoreUUID)
                 && snapshot.equals(that.snapshot)
                 && index.equals(that.index)
-                && version.equals(that.version);
+                && version.equals(that.version)
+                && isSearchableSnapshot == that.isSearchableSnapshot;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(restoreUUID, snapshot, index, version);
+            return Objects.hash(restoreUUID, snapshot, index, version, isSearchableSnapshot);
         }
 
     }
diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java
index dc28fcaf2596f..201bd76b9aada 100644
--- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java
+++ b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java
@@ -221,9 +221,16 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
      */
     public static final Map<String, List<Setting>> FEATURE_FLAGGED_INDEX_SETTINGS = Map.of(
         FeatureFlags.REPLICATION_TYPE,
-        Collections.singletonList(IndexMetadata.INDEX_REPLICATION_TYPE_SETTING),
+        List.of(IndexMetadata.INDEX_REPLICATION_TYPE_SETTING),
         FeatureFlags.REMOTE_STORE,
-        Arrays.asList(IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING, IndexMetadata.INDEX_REMOTE_STORE_REPOSITORY_SETTING)
+        Arrays.asList(IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING, IndexMetadata.INDEX_REMOTE_STORE_REPOSITORY_SETTING),
+        FeatureFlags.SEARCHABLE_SNAPSHOT,
+        List.of(
+            IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY,
+            IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID,
+            IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME,
+            IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID
+        )
     );
 
     public static final IndexScopedSettings DEFAULT_SCOPED_SETTINGS = new IndexScopedSettings(Settings.EMPTY, BUILT_IN_INDEX_SETTINGS);
diff --git a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java
index fa39dc9ac5aa0..7297479776da9 100644
--- a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java
+++ b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java
@@ -29,6 +29,14 @@ public class FeatureFlags {
      */
     public static final String REMOTE_STORE = "opensearch.experimental.feature.remote_store.enabled";
 
+    /**
+     * Gates the functionality of a new parameter to the snapshot restore API
+     * that allows for creation of a new index type that searches a snapshot
+     * directly in a remote repository without restoring all index data to disk
+     * ahead of time.
+     */
+    public static final String SEARCHABLE_SNAPSHOT = "opensearch.experimental.feature.searchable_snapshot.enabled";
+
     /**
      * Used to test feature flags whose values are expected to be booleans.
      * This method returns true if the value is "true" (case-insensitive),
diff --git a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java
index c43f539243d7a..f5bd86854a181 100644
--- a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java
+++ b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java
@@ -159,7 +159,8 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) {
                 nodeEnv.availableShardPaths(request.shardId)
             );
             if (shardStateMetadata != null) {
-                if (indicesService.getShardOrNull(shardId) == null) {
+                if (indicesService.getShardOrNull(shardId) == null
+                    && shardStateMetadata.indexDataLocation == ShardStateMetadata.IndexDataLocation.LOCAL) {
                     final String customDataPath;
                     if (request.getCustomDataPath() != null) {
                         customDataPath = request.getCustomDataPath();
diff --git a/server/src/main/java/org/opensearch/index/IndexModule.java b/server/src/main/java/org/opensearch/index/IndexModule.java
index e52a2ba39ed52..2a462f6165184 100644
--- a/server/src/main/java/org/opensearch/index/IndexModule.java
+++ b/server/src/main/java/org/opensearch/index/IndexModule.java
@@ -70,12 +70,14 @@
 import org.opensearch.index.shard.SearchOperationListener;
 import org.opensearch.index.similarity.SimilarityService;
 import org.opensearch.index.store.FsDirectoryFactory;
+import org.opensearch.index.store.RemoteSnapshotDirectoryFactory;
 import org.opensearch.indices.IndicesQueryCache;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
 import org.opensearch.indices.mapper.MapperRegistry;
 import org.opensearch.indices.recovery.RecoveryState;
 import org.opensearch.plugins.IndexStorePlugin;
+import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
 import org.opensearch.threadpool.ThreadPool;
@@ -94,6 +96,7 @@
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 /**
  * IndexModule represents the central extension point for index level custom implementations like:
@@ -390,15 +393,6 @@ IndexEventListener freeze() { // pkg private for testing
         }
     }
 
-    public static boolean isBuiltinType(String storeType) {
-        for (Type type : Type.values()) {
-            if (type.match(storeType)) {
-                return true;
-            }
-        }
-        return false;
-    }
-
     /**
      * Type of file system
      *
@@ -409,7 +403,8 @@ public enum Type {
         NIOFS("niofs"),
         MMAPFS("mmapfs"),
         SIMPLEFS("simplefs"),
-        FS("fs");
+        FS("fs"),
+        REMOTE_SNAPSHOT("remote_snapshot");
 
         private final String settingsKey;
         private final boolean deprecated;
@@ -426,7 +421,7 @@ public enum Type {
         private static final Map<String, Type> TYPES;
 
         static {
-            final Map<String, Type> types = new HashMap<>(4);
+            final Map<String, Type> types = new HashMap<>(values().length);
             for (final Type type : values()) {
                 types.put(type.settingsKey, type);
             }
@@ -441,6 +436,10 @@ public boolean isDeprecated() {
             return deprecated;
         }
 
+        static boolean containsSettingsKey(String key) {
+            return TYPES.containsKey(key);
+        }
+
         public static Type fromSettingsKey(final String key) {
             final Type type = TYPES.get(key);
             if (type == null) {
@@ -459,6 +458,13 @@ public boolean match(String setting) {
             return getSettingsKey().equals(setting);
         }
 
+        /**
+         * Convenience method to check whether the given IndexSettings contains
+         * an {@link #INDEX_STORE_TYPE_SETTING} set to the value of this type.
+         */
+        public boolean match(IndexSettings settings) {
+            return match(INDEX_STORE_TYPE_SETTING.get(settings.getSettings()));
+        }
     }
 
     public static Type defaultStoreType(final boolean allowMmap) {
@@ -562,7 +568,7 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory(
         if (storeType.isEmpty() || Type.FS.getSettingsKey().equals(storeType)) {
             type = defaultStoreType(allowMmap);
         } else {
-            if (isBuiltinType(storeType)) {
+            if (Type.containsSettingsKey(storeType)) {
                 type = Type.fromSettingsKey(storeType);
             } else {
                 type = null;
@@ -572,7 +578,7 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory(
             throw new IllegalArgumentException("store type [" + storeType + "] is not allowed because mmap is disabled");
         }
         final IndexStorePlugin.DirectoryFactory factory;
-        if (storeType.isEmpty() || isBuiltinType(storeType)) {
+        if (storeType.isEmpty()) {
             factory = DEFAULT_DIRECTORY_FACTORY;
         } else {
             factory = indexStoreFactories.get(storeType);
@@ -641,4 +647,26 @@ private void ensureNotFrozen() {
         }
     }
 
+    public static Map<String, IndexStorePlugin.DirectoryFactory> createBuiltInDirectoryFactories(
+        Supplier<RepositoriesService> repositoriesService
+    ) {
+        final Map<String, IndexStorePlugin.DirectoryFactory> factories = new HashMap<>();
+        for (Type type : Type.values()) {
+            switch (type) {
+                case HYBRIDFS:
+                case NIOFS:
+                case FS:
+                case MMAPFS:
+                case SIMPLEFS:
+                    factories.put(type.getSettingsKey(), DEFAULT_DIRECTORY_FACTORY);
+                    break;
+                case REMOTE_SNAPSHOT:
+                    factories.put(type.getSettingsKey(), new RemoteSnapshotDirectoryFactory(repositoriesService));
+                    break;
+                default:
+                    throw new IllegalStateException("No directory factory mapping for built-in type " + type);
+            }
+        }
+        return factories;
+    }
 }
diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java
index 74b104392133f..e9d64d39aef35 100644
--- a/server/src/main/java/org/opensearch/index/IndexSettings.java
+++ b/server/src/main/java/org/opensearch/index/IndexSettings.java
@@ -551,6 +551,30 @@ public final class IndexSettings {
         Property.Dynamic
     );
 
+    public static final Setting<String> SEARCHABLE_SNAPSHOT_REPOSITORY = Setting.simpleString(
+        "index.searchable_snapshot.repository",
+        Property.IndexScope,
+        Property.InternalIndex
+    );
+
+    public static final Setting<String> SEARCHABLE_SNAPSHOT_ID_UUID = Setting.simpleString(
+        "index.searchable_snapshot.snapshot_id.uuid",
+        Property.IndexScope,
+        Property.InternalIndex
+    );
+
+    public static final Setting<String> SEARCHABLE_SNAPSHOT_ID_NAME = Setting.simpleString(
+        "index.searchable_snapshot.snapshot_id.name",
+        Property.IndexScope,
+        Property.InternalIndex
+    );
+
+    public static final Setting<String> SEARCHABLE_SNAPSHOT_INDEX_ID = Setting.simpleString(
+        "index.searchable_snapshot.index.id",
+        Property.IndexScope,
+        Property.InternalIndex
+    );
+
     private final Index index;
     private final Version version;
     private final Logger logger;
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 8e91d071a6b9a..d9037574070ad 100644
--- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java
+++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -678,7 +678,7 @@ public void onFailure(Exception e) {
             this.shardRouting = newRouting;
 
             assert this.shardRouting.primary() == false || this.shardRouting.started() == false || // note that we use started and not
-                                                                                                   // active to avoid relocating shards
+            // active to avoid relocating shards
                 this.indexShardOperationPermits.isBlocked() || // if permits are blocked, we are still transitioning
                 this.replicationTracker.isPrimaryMode() : "a started primary with non-pending operation term must be in primary mode "
                     + this.shardRouting;
@@ -1938,7 +1938,12 @@ public void openEngineAndRecoverFromTranslog() throws IOException {
                 translogRecoveryStats::incrementRecoveredOperations
             );
         };
-        loadGlobalCheckpointToReplicationTracker();
+
+        // Do not load the global checkpoint if this is a remote snapshot index
+        if (IndexModule.Type.REMOTE_SNAPSHOT.match(indexSettings) == false) {
+            loadGlobalCheckpointToReplicationTracker();
+        }
+
         innerOpenEngineAndTranslog(replicationTracker);
         getEngine().recoverFromTranslog(translogRecoveryRunner, Long.MAX_VALUE);
     }
@@ -3050,13 +3055,18 @@ public void startRecovery(
                 }
                 break;
             case SNAPSHOT:
-                final String repo = ((SnapshotRecoverySource) recoveryState.getRecoverySource()).snapshot().getRepository();
-                executeRecovery(
-                    "from snapshot",
-                    recoveryState,
-                    recoveryListener,
-                    l -> restoreFromRepository(repositoriesService.repository(repo), l)
-                );
+                final SnapshotRecoverySource recoverySource = (SnapshotRecoverySource) recoveryState.getRecoverySource();
+                if (recoverySource.isSearchableSnapshot()) {
+                    executeRecovery("from snapshot (remote)", recoveryState, recoveryListener, this::recoverFromStore);
+                } else {
+                    final String repo = recoverySource.snapshot().getRepository();
+                    executeRecovery(
+                        "from snapshot",
+                        recoveryState,
+                        recoveryListener,
+                        l -> restoreFromRepository(repositoriesService.repository(repo), l)
+                    );
+                }
                 break;
             case LOCAL_SHARDS:
                 final IndexMetadata indexMetadata = indexSettings().getIndexMetadata();
@@ -3217,10 +3227,15 @@ private static void persistMetadata(
                 writeReason = "routing changed from " + currentRouting + " to " + newRouting;
             }
             logger.trace("{} writing shard state, reason [{}]", shardId, writeReason);
+
+            final ShardStateMetadata.IndexDataLocation indexDataLocation = IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.exists(
+                indexSettings.getSettings()
+            ) ? ShardStateMetadata.IndexDataLocation.REMOTE : ShardStateMetadata.IndexDataLocation.LOCAL;
             final ShardStateMetadata newShardStateMetadata = new ShardStateMetadata(
                 newRouting.primary(),
                 indexSettings.getUUID(),
-                newRouting.allocationId()
+                newRouting.allocationId(),
+                indexDataLocation
             );
             ShardStateMetadata.FORMAT.writeAndCleanup(newShardStateMetadata, shardPath.getShardStatePath());
         } else {
diff --git a/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java b/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java
index ccc620fc8cf64..c7e380f842fa0 100644
--- a/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java
+++ b/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java
@@ -484,7 +484,8 @@ private void newAllocationId(ShardPath shardPath, Terminal terminal) throws IOEx
         final ShardStateMetadata newShardStateMetadata = new ShardStateMetadata(
             shardStateMetadata.primary,
             shardStateMetadata.indexUUID,
-            newAllocationId
+            newAllocationId,
+            ShardStateMetadata.IndexDataLocation.LOCAL
         );
 
         ShardStateMetadata.FORMAT.writeAndCleanup(newShardStateMetadata, shardStatePath);
diff --git a/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java b/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java
index 9cd9149cda913..9e334bc6ffd54 100644
--- a/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java
+++ b/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java
@@ -56,17 +56,39 @@ public final class ShardStateMetadata {
     private static final String PRIMARY_KEY = "primary";
     private static final String INDEX_UUID_KEY = "index_uuid";
     private static final String ALLOCATION_ID_KEY = "allocation_id";
+    private static final String INDEX_DATA_LOCATION_KEY = "index_data_location";
+
+    /**
+     * Enumeration of types of data locations for an index
+     */
+    public enum IndexDataLocation {
+        /**
+         * Indicates index data is on the local disk
+         */
+        LOCAL,
+        /**
+         * Indicates index data is remote, such as for a searchable snapshot
+         * index
+         */
+        REMOTE
+    }
 
     public final String indexUUID;
     public final boolean primary;
     @Nullable
     public final AllocationId allocationId; // can be null if we read from legacy format (see fromXContent and MultiDataPathUpgrader)
+    public final IndexDataLocation indexDataLocation;
 
     public ShardStateMetadata(boolean primary, String indexUUID, AllocationId allocationId) {
+        this(primary, indexUUID, allocationId, IndexDataLocation.LOCAL);
+    }
+
+    public ShardStateMetadata(boolean primary, String indexUUID, AllocationId allocationId, IndexDataLocation indexDataLocation) {
         assert indexUUID != null;
         this.primary = primary;
         this.indexUUID = indexUUID;
         this.allocationId = allocationId;
+        this.indexDataLocation = Objects.requireNonNull(indexDataLocation);
     }
 
     @Override
@@ -89,6 +111,9 @@ public boolean equals(Object o) {
         if (Objects.equals(allocationId, that.allocationId) == false) {
             return false;
         }
+        if (Objects.equals(indexDataLocation, that.indexDataLocation) == false) {
+            return false;
+        }
 
         return true;
     }
@@ -98,17 +123,16 @@ public int hashCode() {
         int result = indexUUID.hashCode();
         result = 31 * result + (allocationId != null ? allocationId.hashCode() : 0);
         result = 31 * result + (primary ? 1 : 0);
+        result = 31 * result + indexDataLocation.hashCode();
         return result;
     }
 
     @Override
     public String toString() {
-        return "primary [" + primary + "], allocation [" + allocationId + "]";
+        return "primary [" + primary + "], allocation [" + allocationId + "], index data location [" + indexDataLocation + "]";
     }
 
-    public static final MetadataStateFormat<ShardStateMetadata> FORMAT = new MetadataStateFormat<ShardStateMetadata>(
-        SHARD_STATE_FILE_PREFIX
-    ) {
+    public static final MetadataStateFormat<ShardStateMetadata> FORMAT = new MetadataStateFormat<>(SHARD_STATE_FILE_PREFIX) {
 
         @Override
         protected XContentBuilder newXContentBuilder(XContentType type, OutputStream stream) throws IOException {
@@ -124,6 +148,11 @@ public void toXContent(XContentBuilder builder, ShardStateMetadata shardStateMet
             if (shardStateMetadata.allocationId != null) {
                 builder.field(ALLOCATION_ID_KEY, shardStateMetadata.allocationId);
             }
+            // Omit the index data location field if it is LOCAL (the implicit default)
+            // to maintain compatibility for local indices
+            if (shardStateMetadata.indexDataLocation != IndexDataLocation.LOCAL) {
+                builder.field(INDEX_DATA_LOCATION_KEY, shardStateMetadata.indexDataLocation);
+            }
         }
 
         @Override
@@ -136,6 +165,7 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException
             String currentFieldName = null;
             String indexUUID = IndexMetadata.INDEX_UUID_NA_VALUE;
             AllocationId allocationId = null;
+            IndexDataLocation indexDataLocation = IndexDataLocation.LOCAL;
             while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                 if (token == XContentParser.Token.FIELD_NAME) {
                     currentFieldName = parser.currentName();
@@ -144,6 +174,13 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException
                         primary = parser.booleanValue();
                     } else if (INDEX_UUID_KEY.equals(currentFieldName)) {
                         indexUUID = parser.text();
+                    } else if (INDEX_DATA_LOCATION_KEY.equals(currentFieldName)) {
+                        final String stringValue = parser.text();
+                        try {
+                            indexDataLocation = IndexDataLocation.valueOf(stringValue);
+                        } catch (IllegalArgumentException e) {
+                            throw new CorruptStateException("unexpected value for data location [" + stringValue + "]");
+                        }
                     } else {
                         throw new CorruptStateException("unexpected field in shard state [" + currentFieldName + "]");
                     }
@@ -160,7 +197,7 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException
             if (primary == null) {
                 throw new CorruptStateException("missing value for [primary] in shard state");
             }
-            return new ShardStateMetadata(primary, indexUUID, allocationId);
+            return new ShardStateMetadata(primary, indexUUID, allocationId, indexDataLocation);
         }
     };
 }
diff --git a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
index 06916c4cc87fe..6ca5036808818 100644
--- a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
+++ b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java
@@ -104,9 +104,6 @@ final class StoreRecovery {
      */
     void recoverFromStore(final IndexShard indexShard, ActionListener<Boolean> listener) {
         if (canRecover(indexShard)) {
-            RecoverySource.Type recoveryType = indexShard.recoveryState().getRecoverySource().getType();
-            assert recoveryType == RecoverySource.Type.EMPTY_STORE || recoveryType == RecoverySource.Type.EXISTING_STORE
-                : "expected store recovery type but was: " + recoveryType;
             ActionListener.completeWith(recoveryListener(indexShard, listener), () -> {
                 logger.debug("starting recovery from store ...");
                 internalRecoverFromStore(indexShard);
diff --git a/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java b/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java
new file mode 100644
index 0000000000000..0757d88a4099a
--- /dev/null
+++ b/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java
@@ -0,0 +1,169 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.index.store;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.Lock;
+import org.apache.lucene.store.NoLockFactory;
+import org.apache.lucene.util.SetOnce;
+import org.opensearch.common.blobstore.BlobContainer;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.common.lucene.store.ByteArrayIndexInput;
+import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
+import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.snapshots.SnapshotId;
+
+/**
+ * Trivial in-memory implementation of a Directory that reads from a snapshot
+ * in a repository. This is functional but is only temporary to demonstrate
+ * functional searchable snapshot functionality. The proper implementation will
+ * be implemented per https://github.com/opensearch-project/OpenSearch/issues/3114.
+ *
+ * @opensearch.internal
+ */
+public final class InMemoryRemoteSnapshotDirectory extends Directory {
+
+    private final BlobStoreRepository blobStoreRepository;
+    private final SnapshotId snapshotId;
+    private final BlobPath blobPath;
+    private final SetOnce<BlobContainer> blobContainer = new SetOnce<>();
+    private final SetOnce<Map<String, BlobStoreIndexShardSnapshot.FileInfo>> fileInfoMap = new SetOnce<>();
+
+    public InMemoryRemoteSnapshotDirectory(BlobStoreRepository blobStoreRepository, BlobPath blobPath, SnapshotId snapshotId) {
+        this.blobStoreRepository = blobStoreRepository;
+        this.snapshotId = snapshotId;
+        this.blobPath = blobPath;
+    }
+
+    @Override
+    public String[] listAll() throws IOException {
+        return fileInfoMap().keySet().toArray(new String[0]);
+    }
+
+    @Override
+    public void deleteFile(String name) throws IOException {}
+
+    @Override
+    public IndexOutput createOutput(String name, IOContext context) {
+        return NoopIndexOutput.INSTANCE;
+    }
+
+    @Override
+    public IndexInput openInput(String name, IOContext context) throws IOException {
+        final BlobStoreIndexShardSnapshot.FileInfo fileInfo = fileInfoMap().get(name);
+
+        // Virtual files are contained entirely in the metadata hash field
+        if (fileInfo.name().startsWith("v__")) {
+            return new ByteArrayIndexInput(name, fileInfo.metadata().hash().bytes);
+        }
+
+        try (InputStream is = blobContainer().readBlob(fileInfo.name())) {
+            return new ByteArrayIndexInput(name, is.readAllBytes());
+        }
+    }
+
+    @Override
+    public void close() throws IOException {}
+
+    @Override
+    public long fileLength(String name) throws IOException {
+        initializeIfNecessary();
+        return fileInfoMap.get().get(name).length();
+    }
+
+    @Override
+    public Set<String> getPendingDeletions() throws IOException {
+        return Collections.emptySet();
+    }
+
+    @Override
+    public IndexOutput createTempOutput(String prefix, String suffix, IOContext context) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void sync(Collection<String> names) throws IOException {}
+
+    @Override
+    public void syncMetaData() {}
+
+    @Override
+    public void rename(String source, String dest) throws IOException {}
+
+    @Override
+    public Lock obtainLock(String name) throws IOException {
+        return NoLockFactory.INSTANCE.obtainLock(null, null);
+    }
+
+    static class NoopIndexOutput extends IndexOutput {
+
+        final static NoopIndexOutput INSTANCE = new NoopIndexOutput();
+
+        NoopIndexOutput() {
+            super("noop", "noop");
+        }
+
+        @Override
+        public void close() throws IOException {
+
+        }
+
+        @Override
+        public long getFilePointer() {
+            return 0;
+        }
+
+        @Override
+        public long getChecksum() throws IOException {
+            return 0;
+        }
+
+        @Override
+        public void writeByte(byte b) throws IOException {
+
+        }
+
+        @Override
+        public void writeBytes(byte[] b, int offset, int length) throws IOException {
+
+        }
+    }
+
+    private BlobContainer blobContainer() {
+        initializeIfNecessary();
+        return blobContainer.get();
+    }
+
+    private Map<String, BlobStoreIndexShardSnapshot.FileInfo> fileInfoMap() {
+        initializeIfNecessary();
+        return fileInfoMap.get();
+    }
+
+    /**
+     * Bit of a hack to lazily initialize the blob store to avoid running afoul
+     * of the assertion in {@code BlobStoreRepository#assertSnapshotOrGenericThread}.
+     */
+    private void initializeIfNecessary() {
+        if (blobContainer.get() == null || fileInfoMap.get() == null) {
+            blobContainer.set(blobStoreRepository.blobStore().blobContainer(blobPath));
+            final BlobStoreIndexShardSnapshot snapshot = blobStoreRepository.loadShardSnapshot(blobContainer.get(), snapshotId);
+            fileInfoMap.set(
+                snapshot.indexFiles().stream().collect(Collectors.toMap(BlobStoreIndexShardSnapshot.FileInfo::physicalName, f -> f))
+            );
+        }
+    }
+}
diff --git a/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java b/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java
new file mode 100644
index 0000000000000..bf7806b836b65
--- /dev/null
+++ b/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.index.store;
+
+import java.io.IOException;
+import java.util.function.Supplier;
+
+import org.apache.lucene.store.Directory;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.index.IndexSettings;
+import org.opensearch.index.shard.ShardPath;
+import org.opensearch.plugins.IndexStorePlugin;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.repositories.Repository;
+import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.snapshots.SnapshotId;
+
+/**
+ * Factory for a Directory implementation that can read directly from index
+ * data stored remotely in a repository.
+ *
+ * @opensearch.internal
+ */
+public final class RemoteSnapshotDirectoryFactory implements IndexStorePlugin.DirectoryFactory {
+    private final Supplier<RepositoriesService> repositoriesService;
+
+    public RemoteSnapshotDirectoryFactory(Supplier<RepositoriesService> repositoriesService) {
+        this.repositoriesService = repositoriesService;
+    }
+
+    @Override
+    public Directory newDirectory(IndexSettings indexSettings, ShardPath shardPath) throws IOException {
+        final String repositoryName = IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.get(indexSettings.getSettings());
+        final Repository repository = repositoriesService.get().repository(repositoryName);
+        assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
+        final BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository;
+        final BlobPath blobPath = new BlobPath().add("indices")
+            .add(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.get(indexSettings.getSettings()))
+            .add(Integer.toString(shardPath.getShardId().getId()));
+        final SnapshotId snapshotId = new SnapshotId(
+            IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME.get(indexSettings.getSettings()),
+            IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID.get(indexSettings.getSettings())
+        );
+        return new InMemoryRemoteSnapshotDirectory(blobStoreRepository, blobPath, snapshotId);
+    }
+}
diff --git a/server/src/main/java/org/opensearch/indices/IndicesService.java b/server/src/main/java/org/opensearch/indices/IndicesService.java
index 6808803ee0988..8c3172e911686 100644
--- a/server/src/main/java/org/opensearch/indices/IndicesService.java
+++ b/server/src/main/java/org/opensearch/indices/IndicesService.java
@@ -111,6 +111,7 @@
 import org.opensearch.index.engine.InternalEngineFactory;
 import org.opensearch.index.engine.NRTReplicationEngineFactory;
 import org.opensearch.index.engine.NoOpEngine;
+import org.opensearch.index.engine.ReadOnlyEngine;
 import org.opensearch.index.fielddata.IndexFieldDataCache;
 import org.opensearch.index.flush.FlushStats;
 import org.opensearch.index.get.GetStats;
@@ -132,6 +133,7 @@
 import org.opensearch.index.shard.IndexingOperationListener;
 import org.opensearch.index.shard.IndexingStats;
 import org.opensearch.index.shard.ShardId;
+import org.opensearch.index.translog.TranslogStats;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.cluster.IndicesClusterStateService;
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
@@ -338,13 +340,6 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
         this.metaStateService = metaStateService;
         this.engineFactoryProviders = engineFactoryProviders;
 
-        // do not allow any plugin-provided index store type to conflict with a built-in type
-        for (final String indexStoreType : directoryFactories.keySet()) {
-            if (IndexModule.isBuiltinType(indexStoreType)) {
-                throw new IllegalStateException("registered index store type [" + indexStoreType + "] conflicts with a built-in type");
-            }
-        }
-
         this.directoryFactories = directoryFactories;
         this.recoveryStateFactories = recoveryStateFactories;
         // doClose() is called when shutting down a node, yet there might still be ongoing requests
@@ -772,6 +767,9 @@ private EngineFactory getEngineFactory(final IndexSettings idxSettings) {
             if (idxSettings.isSegRepEnabled()) {
                 return new NRTReplicationEngineFactory();
             }
+            if (IndexModule.Type.REMOTE_SNAPSHOT.match(idxSettings)) {
+                return config -> new ReadOnlyEngine(config, new SeqNoStats(0, 0, 0), new TranslogStats(), true, Function.identity(), false);
+            }
             return new InternalEngineFactory();
         } else if (engineFactories.size() == 1) {
             assert engineFactories.get(0).isPresent();
diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
index 0454751f58925..fd77d9a7ba0b9 100644
--- a/server/src/main/java/org/opensearch/node/Node.java
+++ b/server/src/main/java/org/opensearch/node/Node.java
@@ -37,6 +37,7 @@
 import org.apache.lucene.util.Constants;
 import org.apache.lucene.util.SetOnce;
 import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance;
+import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexingPressureService;
 import org.opensearch.tasks.TaskResourceTrackingService;
 import org.opensearch.threadpool.RunnableTaskExecutionListener;
@@ -213,6 +214,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -604,11 +606,23 @@ protected Node(
                 .map(plugin -> (Function<IndexSettings, Optional<EngineFactory>>) plugin::getEngineFactory)
                 .collect(Collectors.toList());
 
-            final Map<String, IndexStorePlugin.DirectoryFactory> indexStoreFactories = pluginsService.filterPlugins(IndexStorePlugin.class)
+            final Map<String, IndexStorePlugin.DirectoryFactory> builtInDirectoryFactories = IndexModule.createBuiltInDirectoryFactories(
+                repositoriesServiceReference::get
+            );
+            final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories = new HashMap<>();
+            pluginsService.filterPlugins(IndexStorePlugin.class)
                 .stream()
                 .map(IndexStorePlugin::getDirectoryFactories)
                 .flatMap(m -> m.entrySet().stream())
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
+                .forEach((k, v) -> {
+                    // do not allow any plugin-provided index store type to conflict with a built-in type
+                    if (builtInDirectoryFactories.containsKey(k)) {
+                        throw new IllegalStateException("registered index store type [" + k + "] conflicts with a built-in type");
+                    }
+                    directoryFactories.put(k, v);
+                });
+            directoryFactories.putAll(builtInDirectoryFactories);
 
             final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories = pluginsService.filterPlugins(
                 IndexStorePlugin.class
@@ -653,7 +667,7 @@ protected Node(
                 client,
                 metaStateService,
                 engineFactoryProviders,
-                indexStoreFactories,
+                Map.copyOf(directoryFactories),
                 searchModule.getValuesSourceRegistry(),
                 recoveryStateFactories,
                 remoteDirectoryFactory
diff --git a/server/src/main/java/org/opensearch/snapshots/RestoreService.java b/server/src/main/java/org/opensearch/snapshots/RestoreService.java
index 417498467622a..643f176a82221 100644
--- a/server/src/main/java/org/opensearch/snapshots/RestoreService.java
+++ b/server/src/main/java/org/opensearch/snapshots/RestoreService.java
@@ -84,7 +84,9 @@
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.Index;
+import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.ShardId;
@@ -431,20 +433,30 @@ public ClusterState execute(ClusterState currentState) {
                                 .getMaxNodeVersion()
                                 .minimumIndexCompatibilityVersion();
                             for (Map.Entry<String, String> indexEntry : indices.entrySet()) {
+                                String renamedIndexName = indexEntry.getKey();
                                 String index = indexEntry.getValue();
                                 boolean partial = checkPartial(index);
-                                SnapshotRecoverySource recoverySource = new SnapshotRecoverySource(
+
+                                IndexMetadata snapshotIndexMetadata = updateIndexSettings(
+                                    metadata.index(index),
+                                    request.indexSettings(),
+                                    request.ignoreIndexSettings()
+                                );
+                                final boolean isSearchableSnapshot = FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT)
+                                    && IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey().equals(request.storageType().toString());
+                                if (isSearchableSnapshot) {
+                                    snapshotIndexMetadata = addSnapshotToIndexSettings(
+                                        snapshotIndexMetadata,
+                                        snapshot,
+                                        repositoryData.resolveIndexId(index)
+                                    );
+                                }
+                                final SnapshotRecoverySource recoverySource = new SnapshotRecoverySource(
                                     restoreUUID,
                                     snapshot,
                                     snapshotInfo.version(),
-                                    repositoryData.resolveIndexId(index)
-                                );
-                                String renamedIndexName = indexEntry.getKey();
-                                IndexMetadata snapshotIndexMetadata = metadata.index(index);
-                                snapshotIndexMetadata = updateIndexSettings(
-                                    snapshotIndexMetadata,
-                                    request.indexSettings(),
-                                    request.ignoreIndexSettings()
+                                    repositoryData.resolveIndexId(index),
+                                    isSearchableSnapshot
                                 );
                                 try {
                                     snapshotIndexMetadata = metadataIndexUpgradeService.upgradeIndexMetadata(
@@ -1217,4 +1229,16 @@ public void applyClusterState(ClusterChangedEvent event) {
             logger.warn("Failed to update restore state ", t);
         }
     }
+
+    private static IndexMetadata addSnapshotToIndexSettings(IndexMetadata metadata, Snapshot snapshot, IndexId indexId) {
+        final Settings newSettings = Settings.builder()
+            .put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey())
+            .put(IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.getKey(), snapshot.getRepository())
+            .put(IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID.getKey(), snapshot.getSnapshotId().getUUID())
+            .put(IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME.getKey(), snapshot.getSnapshotId().getName())
+            .put(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.getKey(), indexId.getId())
+            .put(metadata.getSettings())
+            .build();
+        return IndexMetadata.builder(metadata).settings(newSettings).build();
+    }
 }
diff --git a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java
index c9404226f001c..d9e40c102febe 100644
--- a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java
+++ b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java
@@ -326,7 +326,8 @@ public void testShardStateMetaHashCodeEquals() {
         ShardStateMetadata meta = new ShardStateMetadata(
             randomBoolean(),
             randomRealisticUnicodeOfCodepointLengthBetween(1, 10),
-            allocationId
+            allocationId,
+            randomFrom(ShardStateMetadata.IndexDataLocation.values())
         );
 
         assertEquals(meta, new ShardStateMetadata(meta.primary, meta.indexUUID, meta.allocationId));
@@ -338,7 +339,12 @@ public void testShardStateMetaHashCodeEquals() {
         Set<Integer> hashCodes = new HashSet<>();
         for (int i = 0; i < 30; i++) { // just a sanity check that we impl hashcode
             allocationId = randomBoolean() ? null : randomAllocationId();
-            meta = new ShardStateMetadata(randomBoolean(), randomRealisticUnicodeOfCodepointLengthBetween(1, 10), allocationId);
+            meta = new ShardStateMetadata(
+                randomBoolean(),
+                randomRealisticUnicodeOfCodepointLengthBetween(1, 10),
+                allocationId,
+                randomFrom(ShardStateMetadata.IndexDataLocation.values())
+            );
             hashCodes.add(meta.hashCode());
         }
         assertTrue("more than one unique hashcode expected but got: " + hashCodes.size(), hashCodes.size() > 1);
diff --git a/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java b/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java
index beda468b45fb0..25ec7c7987855 100644
--- a/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java
+++ b/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java
@@ -35,6 +35,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
+import org.opensearch.gateway.WriteStateException;
 import org.opensearch.index.Index;
 import org.opensearch.test.OpenSearchTestCase;
 
@@ -50,7 +51,7 @@ public void testLoadShardPath() throws IOException {
             ShardId shardId = new ShardId("foo", "0xDEADBEEF", 0);
             Path[] paths = env.availableShardPaths(shardId);
             Path path = randomFrom(paths);
-            ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, "0xDEADBEEF", AllocationId.newInitializing()), path);
+            writeShardStateMetadata("0xDEADBEEF", path);
             ShardPath shardPath = ShardPath.loadShardPath(logger, env, shardId, "");
             assertEquals(path, shardPath.getDataPath());
             assertEquals("0xDEADBEEF", shardPath.getShardId().getIndex().getUUID());
@@ -66,7 +67,7 @@ public void testFailLoadShardPathOnMultiState() throws IOException {
             ShardId shardId = new ShardId("foo", indexUUID, 0);
             Path[] paths = env.availableShardPaths(shardId);
             assumeTrue("This test tests multi data.path but we only got one", paths.length > 1);
-            ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing()), paths);
+            writeShardStateMetadata(indexUUID, paths);
             Exception e = expectThrows(IllegalStateException.class, () -> ShardPath.loadShardPath(logger, env, shardId, ""));
             assertThat(e.getMessage(), containsString("more than one shard state found"));
         }
@@ -77,7 +78,7 @@ public void testFailLoadShardPathIndexUUIDMissmatch() throws IOException {
             ShardId shardId = new ShardId("foo", "foobar", 0);
             Path[] paths = env.availableShardPaths(shardId);
             Path path = randomFrom(paths);
-            ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, "0xDEADBEEF", AllocationId.newInitializing()), path);
+            writeShardStateMetadata("0xDEADBEEF", path);
             Exception e = expectThrows(IllegalStateException.class, () -> ShardPath.loadShardPath(logger, env, shardId, ""));
             assertThat(e.getMessage(), containsString("expected: foobar on shard path"));
         }
@@ -121,7 +122,7 @@ public void testGetRootPaths() throws IOException {
             ShardId shardId = new ShardId("foo", indexUUID, 0);
             Path[] paths = env.availableShardPaths(shardId);
             Path path = randomFrom(paths);
-            ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing()), path);
+            writeShardStateMetadata(indexUUID, path);
             ShardPath shardPath = ShardPath.loadShardPath(logger, env, shardId, customDataPath);
             boolean found = false;
             for (Path p : env.nodeDataPaths()) {
@@ -148,4 +149,10 @@ public void testGetRootPaths() throws IOException {
         }
     }
 
+    private static void writeShardStateMetadata(String indexUUID, Path... paths) throws WriteStateException {
+        ShardStateMetadata.FORMAT.writeAndCleanup(
+            new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing(), ShardStateMetadata.IndexDataLocation.LOCAL),
+            paths
+        );
+    }
 }
diff --git a/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java b/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java
index cf8d6677b4227..ce40de0e9aa71 100644
--- a/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java
+++ b/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java
@@ -57,6 +57,8 @@
 import java.util.Arrays;
 import java.util.Locale;
 
+import static org.opensearch.test.store.MockFSDirectoryFactory.FILE_SYSTEM_BASED_STORE_TYPES;
+
 public class FsDirectoryFactoryTests extends OpenSearchTestCase {
 
     public void testPreload() throws IOException {
@@ -170,7 +172,7 @@ public void testStoreDirectory() throws IOException {
         // default
         doTestStoreDirectory(tempDir, null, IndexModule.Type.FS);
         // explicit directory impls
-        for (IndexModule.Type type : IndexModule.Type.values()) {
+        for (IndexModule.Type type : FILE_SYSTEM_BASED_STORE_TYPES) {
             doTestStoreDirectory(tempDir, type.name().toLowerCase(Locale.ROOT), type);
         }
     }
diff --git a/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java b/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java
index 47952af1cd06c..e38b62c419334 100644
--- a/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java
+++ b/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java
@@ -63,10 +63,15 @@
 import java.io.PrintStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Random;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 public class MockFSDirectoryFactory implements IndexStorePlugin.DirectoryFactory {
+    public static final List<IndexModule.Type> FILE_SYSTEM_BASED_STORE_TYPES = Arrays.stream(IndexModule.Type.values())
+        .filter(t -> (t == IndexModule.Type.REMOTE_SNAPSHOT) == false)
+        .collect(Collectors.toUnmodifiableList());
 
     public static final Setting<Double> RANDOM_IO_EXCEPTION_RATE_ON_OPEN_SETTING = Setting.doubleSetting(
         "index.store.mock.random.io_exception_rate_on_open",
@@ -168,7 +173,7 @@ private Directory randomDirectoryService(Random random, IndexSettings indexSetti
                     .put(indexSettings.getIndexMetadata().getSettings())
                     .put(
                         IndexModule.INDEX_STORE_TYPE_SETTING.getKey(),
-                        RandomPicks.randomFrom(random, IndexModule.Type.values()).getSettingsKey()
+                        RandomPicks.randomFrom(random, FILE_SYSTEM_BASED_STORE_TYPES).getSettingsKey()
                     )
             )
             .build();