repo: cache index for cloud versioning

This allows us to reuse cloud versioning index that we've built in other
operations like `data status`.

Note, this only works with `dvc config feature.data_index_cache true`
right now.

Also note that index is not populated for regular remotes, it will come
in a separate PR as there are some http-related issues to figure out.

Related #8962

dvc/data_cloud.py

@@ -18,10 +18,11 @@ logger = logging.getLogger(__name__)
 
 
 class Remote:
-    def __init__(self, name: str, path: str, fs: "FileSystem", **config):
+    def __init__(self, name: str, path: str, fs: "FileSystem", *, index=None, **config):
         self.path = path
         self.fs = fs
         self.name = name
+        self.index = index
 
         self.worktree: bool = config.pop("worktree", False)
         self.config = config
@@ -75,7 +76,11 @@ class DataCloud:
 
             fs = cls(**config)
             config["tmp_dir"] = self.repo.index_db_dir
-            return Remote(name, fs_path, fs, **config)
+            if self.repo.data_index is not None:
+                index = self.repo.data_index.view(("remotes", name))
+            else:
+                index = None
+            return Remote(name, fs_path, fs, index=index, **config)
 
         if bool(self.repo.config["remote"]):
             error_msg = (

dvc/repo/index.py

@@ -153,10 +153,11 @@ def _load_storage_from_out(storage_map, key, out):
                     key=key,
                     fs=remote.fs,
                     path=remote.fs.path.join(remote.path, *key),
+                    index=remote.index,
                 )
             )
         else:
-            storage_map.add_remote(ObjectStorage(key, remote.odb))
+            storage_map.add_remote(ObjectStorage(key, remote.odb, index=remote.index))
     except NoRemoteError:
         pass
 

dvc/repo/worktree.py

@@ -190,6 +190,11 @@ def push_worktree(
                     f"remote {remote_obj.name!r} does not support versioning"
                 ) from None
 
+        if remote_obj.index is not None:
+            for key, entry in new_index.iteritems():
+                remote_obj.index[key] = entry
+            remote_obj.index.commit()
+
         for out in view.outs:
             workspace, _key = out.index_key
             _merge_push_meta(out, repo.index.data[workspace], remote_obj.name)

pyproject.toml

@@ -53,7 +53,7 @@ dependencies = [
     "dvc-render>=0.1.2",
     "dvc-task==0.1.11",
     "dvclive>=1.2.2",
-    "dvc-data==0.40.1",
+    "dvc-data==0.40.3",
     "dvc-http",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.7",