diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java
index 8b219a8e45d..dc2784d7f67 100644
--- a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java
+++ b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java
@@ -42,6 +42,10 @@ import org.junit.Before;
 import java.nio.file.Path;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.is;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 
@@ -258,4 +262,22 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request("ds");
         expectThrows(ResourceNotFoundException.class, () -> client.admin().indices().getDataStreams(getRequest).actionGet());
     }
+
+    public void testDataStreamNotIncludedInLimitedSnapshot() throws ExecutionException, InterruptedException {
+        final String snapshotName = "test-snap";
+        CreateSnapshotResponse createSnapshotResponse = client.admin().cluster()
+                .prepareCreateSnapshot(REPO, snapshotName)
+                .setWaitForCompletion(true)
+                .setIndices("does-not-exist-*")
+                .setIncludeGlobalState(true)
+                .get();
+        assertThat(createSnapshotResponse.getSnapshotInfo().state(), is(SnapshotState.SUCCESS));
+
+        assertThat(client().admin().indices()
+                .deleteDataStream(new DeleteDataStreamAction.Request("*")).get().isAcknowledged(), is(true));
+
+        final RestoreSnapshotResponse restoreSnapshotResponse =
+                client().admin().cluster().prepareRestoreSnapshot(REPO, snapshotName).get();
+        assertThat(restoreSnapshotResponse.getRestoreInfo().indices(), empty());
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java b/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
index cd2002c6d81..9a55a515ad4 100644
--- a/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
+++ b/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
@@ -216,9 +216,12 @@ public class RestoreService implements ClusterStateApplier {
                     dataStreams = new HashMap<>();
                 } else {
                     globalMetadata = repository.getSnapshotGlobalMetadata(snapshotId);
-                    dataStreams = globalMetadata.dataStreams();
-                    if (request.includeGlobalState() == false) {
-                        dataStreams.keySet().retainAll(requestedDataStreams);
+                    final Map<String, DataStream> dataStreamsInSnapshot = globalMetadata.dataStreams();
+                    dataStreams = new HashMap<>(requestedDataStreams.size());
+                    for (String requestedDataStream : requestedDataStreams) {
+                        final DataStream dataStreamInSnapshot = dataStreamsInSnapshot.get(requestedDataStream);
+                        assert dataStreamInSnapshot != null : "DataStream [" + requestedDataStream + "] not found in snapshot";
+                        dataStreams.put(requestedDataStream, dataStreamInSnapshot);
                     }
                 }
                 requestIndices.removeAll(dataStreams.keySet());
diff --git a/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java b/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
index bfae3986b7b..147b13b0df9 100644
--- a/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
+++ b/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
@@ -210,13 +210,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                 indices = Arrays.asList(indexNameExpressionResolver.concreteIndexNames(currentState,
                     request.indicesOptions(), true, request.indices()));
 
-                Map<String, DataStream> allDataStreams = currentState.metadata().dataStreams();
-                List<String> dataStreams;
-                if (request.includeGlobalState()) {
-                    dataStreams = new ArrayList<>(allDataStreams.keySet());
-                } else {
-                    dataStreams = indexNameExpressionResolver.dataStreamNames(currentState, request.indicesOptions(), request.indices());
-                }
+                final List<String> dataStreams =
+                        indexNameExpressionResolver.dataStreamNames(currentState, request.indicesOptions(), request.indices());
 
                 logger.trace("[{}][{}] creating snapshot for indices [{}]", repositoryName, snapshotName, indices);
                 newSnapshot = new SnapshotsInProgress.Entry(
@@ -507,9 +502,10 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
     }
 
     private static Metadata metadataForSnapshot(SnapshotsInProgress.Entry snapshot, Metadata metadata) {
+        final Metadata.Builder builder;
         if (snapshot.includeGlobalState() == false) {
             // Remove global state from the cluster state
-            Metadata.Builder builder = Metadata.builder();
+            builder = Metadata.builder();
             for (IndexId index : snapshot.indices()) {
                 final IndexMetadata indexMetadata = metadata.index(index.getName());
                 if (indexMetadata == null) {
@@ -518,21 +514,21 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                     builder.put(indexMetadata, false);
                 }
             }
-
-            Map<String, DataStream> dataStreams = new HashMap<>();
-            for (String dataStreamName : snapshot.dataStreams()) {
-                DataStream dataStream = metadata.dataStreams().get(dataStreamName);
-                if (dataStream == null) {
-                    assert snapshot.partial() : "Data stream [" + dataStreamName +
-                        "] was deleted during a snapshot but snapshot was not partial.";
-                } else {
-                    dataStreams.put(dataStreamName, dataStream);
-                }
-            }
-            builder.dataStreams(dataStreams);
-            metadata = builder.build();
+        } else {
+            builder = Metadata.builder(metadata);
         }
-        return metadata;
+        // Only keep those data streams in the metadata that were actually requested by the initial snapshot create operation
+        Map<String, DataStream> dataStreams = new HashMap<>();
+        for (String dataStreamName : snapshot.dataStreams()) {
+            DataStream dataStream = metadata.dataStreams().get(dataStreamName);
+            if (dataStream == null) {
+                assert snapshot.partial() : "Data stream [" + dataStreamName +
+                        "] was deleted during a snapshot but snapshot was not partial.";
+            } else {
+                dataStreams.put(dataStreamName, dataStream);
+            }
+        };
+        return builder.dataStreams(dataStreams).build();
     }
 
     /**