fetch: cache collected

This avoids recollecting the cache files over and over again.

For example, `dvc fetch` noop in `dvc-bench` went from ~30sec down to ~10sec
for me.

Next step is to also cache remote index, so we don't have to waste those
~10sec checking files in the remote.

Related https://github.com/iterative/dvc-data/pull/376

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

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

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",

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)