diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py
index 77f974945..78b987276 100644
--- a/dvc/repo/fetch.py
+++ b/dvc/repo/fetch.py
@@ -33,6 +33,8 @@ def fetch(  # noqa: C901, PLR0913
         config.NoRemoteError: thrown when downloading only local files and no
             remote is configured
     """
+    from fsspec.utils import tokenize
+
     from dvc.fs.callbacks import Callback
     from dvc_data.index.fetch import fetch as ifetch
 
@@ -48,36 +50,43 @@ def fetch(  # noqa: C901, PLR0913
     except DownloadError as exc:
         failed_count += exc.amount
 
-    def _indexes():
-        for _ in self.brancher(
-            revs=revs,
-            all_branches=all_branches,
-            all_tags=all_tags,
-            all_commits=all_commits,
-        ):
-            yield self.index.targets_view(
+    indexes = []
+    index_keys = set()
+    for _ in self.brancher(
+        revs=revs,
+        all_branches=all_branches,
+        all_tags=all_tags,
+        all_commits=all_commits,
+    ):
+        saved_remote = self.config["core"].get("remote")
+        try:
+            if remote:
+                self.config["core"]["remote"] = remote
+
+            idx = self.index.targets_view(
                 targets,
                 with_deps=with_deps,
                 recursive=recursive,
-            ).data["repo"]
+            )
+            index_keys.add(idx.data_tree.hash_info.value)
+            indexes.append(idx.data["repo"])
+        finally:
+            if remote:
+                self.config["core"]["remote"] = saved_remote
 
-    saved_remote = self.config["core"].get("remote")
-    try:
-        if remote:
-            self.config["core"]["remote"] = remote
+    cache_key = ("fetch", tokenize(sorted(index_keys)))
 
-        with Callback.as_tqdm_callback(
-            desc="Fetching",
-            unit="file",
-        ) as cb:
-            fetch_transferred, fetch_failed = ifetch(
-                _indexes(),
-                jobs=jobs,
-                callback=cb,
-            )  # pylint: disable=assignment-from-no-return
-    finally:
-        if remote:
-            self.config["core"]["remote"] = saved_remote
+    with Callback.as_tqdm_callback(
+        desc="Fetching",
+        unit="file",
+    ) as cb:
+        fetch_transferred, fetch_failed = ifetch(
+            indexes,
+            jobs=jobs,
+            callback=cb,
+            cache_index=self.data_index,
+            cache_key=cache_key,
+        )  # pylint: disable=assignment-from-no-return
 
     if fetch_transferred:
         # NOTE: dropping cached index to force reloading from newly saved cache
diff --git a/dvc/repo/index.py b/dvc/repo/index.py
index fc55dc33a..a9901b2b0 100644
--- a/dvc/repo/index.py
+++ b/dvc/repo/index.py
@@ -590,6 +590,23 @@ class IndexView:
 
         return dict(ret)
 
+    @cached_property
+    def data_tree(self):
+        from dvc_data.hashfile.tree import Tree
+
+        tree = Tree()
+        for out in self.outs:
+            if not out.use_cache:
+                continue
+
+            ws, key = out.index_key
+
+            tree.add((ws, *key), out.meta, out.hash_info)
+
+        tree.digest()
+
+        return tree
+
     @cached_property
     def data(self) -> Dict[str, Union["DataIndex", "DataIndexView"]]:
         from dvc_data.index import DataIndex, view
diff --git a/pyproject.toml b/pyproject.toml
index d95b45b94..eef62ee71 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -34,7 +34,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=1.5.0,<1.6.0",
+    "dvc-data>=1.6.1,<1.7.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.9.2,<1",
diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py
index 070828e6e..644d7bc75 100644
--- a/tests/func/test_data_cloud.py
+++ b/tests/func/test_data_cloud.py
@@ -197,8 +197,8 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     hash_spy = mocker.spy(dvc_data.hashfile.hash, "file_md5")
 
     dvc.pull()
-    # NOTE: 1 is for index.data_tree building
-    assert hash_spy.call_count == 2
+    # NOTE: 2 are for index.data_tree building
+    assert hash_spy.call_count == 3
 
     # Removing cache will invalidate existing state entries
     dvc.cache.local.clear()
@@ -206,7 +206,7 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     dvc.config["remote"]["upstream"]["verify"] = True
 
     dvc.pull()
-    assert hash_spy.call_count == 8
+    assert hash_spy.call_count == 10
 
 
 @flaky(max_runs=3, min_passes=1)