dependency: repo: try to fetch before staging (#5549)

Related to https://github.com/iterative/dvc-bench/issues/234

dvc/dependency/repo.py

@@ -78,6 +78,8 @@ class RepoDependency(LocalDependency):
 
     def download(self, to, jobs=None):
         from dvc.checkout import checkout
+        from dvc.config import NoRemoteError
+        from dvc.exceptions import NoOutputOrStageError
         from dvc.objects import save
         from dvc.objects.stage import stage
 
@@ -87,6 +89,10 @@ class RepoDependency(LocalDependency):
             if self.def_repo.get(self.PARAM_REV_LOCK) is None:
                 self.def_repo[self.PARAM_REV_LOCK] = repo.get_rev()
             path_info = PathInfo(repo.root_dir) / self.def_path
+            try:
+                repo.fetch([path_info.fspath], jobs=jobs, recursive=True)
+            except (NoOutputOrStageError, NoRemoteError):
+                pass
             obj = stage(
                 odb, path_info, repo.repo_fs, jobs=jobs, follow_subrepos=False,
             )

dvc/fs/s3.py

@@ -261,8 +261,13 @@ class S3FileSystem(BaseFileSystem):
         return path_info.path in self._list_paths(path_info)
 
     def info(self, path_info):
+        # temporary workaround, will be removed when migrating to s3fs
+        if self.isdir(path_info):
+            return {"type": "directory"}
+
         with self._get_obj(path_info) as obj:
             return {
+                "type": "file",
                 "etag": obj.e_tag.strip('"'),
                 "size": obj.content_length,
             }

dvc/objects/stage.py

@@ -81,9 +81,18 @@ def _collect_dir(path_info, fs, name, state, **kwargs):
     return dir_info
 
 
-def get_dir_hash(path_info, fs, name, state, **kwargs):
+def get_dir_hash(path_info, fs, name, odb, state, **kwargs):
     from . import Tree
 
+    value = fs.info(path_info).get(name)
+    if value:
+        hash_info = HashInfo(name, value)
+        try:
+            Tree.load(odb, hash_info)
+            return hash_info
+        except FileNotFoundError:
+            pass
+
     dir_info = _collect_dir(path_info, fs, name, state, **kwargs)
     hash_info = Tree.save_dir_info(fs.repo.odb.local, dir_info)
     hash_info.size = dir_info.size
@@ -125,7 +134,7 @@ def get_hash(path_info, fs, name, odb, **kwargs):
         return hash_info
 
     if fs.isdir(path_info):
-        hash_info = get_dir_hash(path_info, fs, name, state, **kwargs)
+        hash_info = get_dir_hash(path_info, fs, name, odb, state, **kwargs)
     else:
         hash_info = get_file_hash(path_info, fs, name)
 

tests/unit/test_external_repo.py

@@ -4,9 +4,6 @@ from unittest.mock import call
 import pytest
 
 from dvc.external_repo import external_repo
-from dvc.objects import save
-from dvc.objects.stage import stage
-from dvc.path_info import PathInfo
 from tests.unit.fs.test_repo import make_subrepo
 
 
@@ -91,36 +88,3 @@ def test_subrepo_is_constructed_properly(
             assert repo.config["remote"]["auto-generated-upstream"][
                 "url"
             ] == str(main_cache)
-
-
-def test_fetch_external_repo_jobs(tmp_dir, scm, mocker, dvc, local_remote):
-    tmp_dir.dvc_gen(
-        {
-            "dir1": {
-                "file1": "file1",
-                "file2": "file2",
-                "file3": "file3",
-                "file4": "file4",
-            },
-        },
-        commit="init",
-    )
-
-    dvc.push()
-
-    with external_repo(str(tmp_dir)) as repo:
-        spy = mocker.spy(repo.cloud, "pull")
-
-        obj = stage(
-            dvc.odb.local,
-            PathInfo(repo.root_dir) / "dir1",
-            repo.repo_fs,
-            follow_subrepos=False,
-            jobs=3,
-        )
-        save(
-            dvc.odb.local, obj, jobs=3,
-        )
-
-        run_jobs = tuple(spy.call_args_list[0])[1].get("jobs")
-        assert run_jobs == 3