support moving missing file keeping the metadata intact (#10890)

dvc/dependency/__init__.py

@@ -41,7 +41,7 @@ def _get(stage, p, info, **kwargs):
     return Dependency(stage, p, info, **kwargs)
 
 
-def loadd_from(stage, d_list):
+def loadd_from(stage, d_list) -> list[Dependency]:
     ret = []
     for d in d_list:
         p = d.pop(Output.PARAM_PATH, None)

dvc/dependency/param.py

@@ -110,6 +110,7 @@ class ParamsDependency(Dependency):
         except MissingParamsFile:
             return {}
 
+        assert self.repo
         try:
             return read_param_file(
                 self.repo.fs,

dvc/dependency/repo.py

@@ -150,6 +150,7 @@ class RepoDependency(Dependency):
         # local cache instead of the remote's cache. This avoids re-streaming of data,
         # but messes up the call to `_get_remote_config()` downstream, which will need
         # to ignore cache parameters.
+        assert self.repo
         config["cache"] = self.repo.config["cache"]
         config["cache"]["dir"] = self.repo.cache.local_cache_dir
 

dvc/output.py

@@ -42,6 +42,7 @@ from .utils import relpath
 from .utils.fs import path_isin
 
 if TYPE_CHECKING:
+    from dvc.repo import Repo
     from dvc_data.hashfile.obj import HashFile
     from dvc_data.index import DataIndexKey
 
@@ -83,7 +84,7 @@ def _get(stage, path, **kwargs):
     return Output(stage, path, **kwargs)
 
 
-def loadd_from(stage, d_list):
+def loadd_from(stage, d_list) -> list["Output"]:
     ret = []
     for d in d_list:
         p = d.pop(Output.PARAM_PATH)
@@ -329,7 +330,7 @@ class Output:
         self.annot = Annotation(
             desc=desc, type=type, labels=labels or [], meta=meta or {}
         )
-        self.repo = stage.repo if not repo and stage else repo
+        self.repo: Optional[Repo] = stage.repo if not repo and stage else repo
         meta_d = merge_file_meta_from_cloud(info or {})
         meta = Meta.from_dict(meta_d)
         # NOTE: when version_aware is not passed into get_cloud_fs, it will be
@@ -511,6 +512,7 @@ class Output:
         from dvc.cachemgr import LEGACY_HASH_NAMES
 
         assert self.is_in_repo
+        assert self.repo
         odb_name = "legacy" if self.hash_name in LEGACY_HASH_NAMES else "repo"
         return getattr(self.repo.cache, odb_name)
 
@@ -518,6 +520,7 @@ class Output:
     def local_cache(self):
         from dvc.cachemgr import LEGACY_HASH_NAMES
 
+        assert self.repo
         if self.hash_name in LEGACY_HASH_NAMES:
             return self.repo.cache.legacy
         return self.repo.cache.local
@@ -585,6 +588,7 @@ class Output:
     def index_key(self) -> tuple[str, "DataIndexKey"]:
         if self.is_in_repo:
             workspace = "repo"
+            assert self.repo
             key = self.repo.fs.relparts(self.fs_path, self.repo.root_dir)
         else:
             workspace = self.fs.protocol
@@ -640,6 +644,7 @@ class Output:
     @property
     def dvcignore(self) -> Optional["DvcIgnoreFilter"]:
         if self.fs.protocol == "local":
+            assert self.repo
             return self.repo.dvcignore
         return None
 
@@ -661,6 +666,7 @@ class Output:
         if not self.use_scm_ignore:
             return
 
+        assert self.repo
         if self.repo.scm.is_tracked(self.fspath):
             raise OutputAlreadyTrackedError(self)
 
@@ -670,6 +676,7 @@ class Output:
         if not self.use_scm_ignore:
             return
 
+        assert self.repo
         self.repo.scm_context.ignore_remove(self.fspath)
 
     def save(self) -> None:
@@ -776,6 +783,7 @@ class Output:
                         callback=cb,
                     )
             if relink:
+                assert self.repo
                 rel = self.fs.relpath(filter_info or self.fs_path)
                 with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
                     self._checkout(
@@ -957,6 +965,7 @@ class Output:
         added = not self.exists
 
         try:
+            assert self.repo
             modified = self._checkout(
                 filter_info or self.fs_path,
                 self.fs,
@@ -987,19 +996,24 @@ class Output:
         if ignore_remove:
             self.ignore_remove()
 
-    def move(self, out):
+    def move(self, out: "Output") -> None:
+        src_exists = self.exists
+        if src_exists:
+            self.fs.move(self.fs_path, out.fs_path)
+        else:
+            logger.warning("%r missing", self.fspath)
+
         if self.protocol == "local" and self.use_scm_ignore:
+            assert self.repo
             self.repo.scm_context.ignore_remove(self.fspath)
 
-        self.fs.move(self.fs_path, out.fs_path)
         self.def_path = out.def_path
         self.fs_path = out.fs_path
-        self.save()
-        self.commit()
-
-        # should already be ignored in self.save()
-        # if self.protocol == "local" and self.use_scm_ignore:
-        #     self.repo.scm_context.ignore(self.fspath)
+        try:
+            self.save()
+            self.commit()
+        except self.DoesNotExistError:
+            self.ignore()
 
     def transfer(
         self, source, odb=None, jobs=None, update=False, no_progress_bar=False
@@ -1080,6 +1094,7 @@ class Output:
             if self.remote:
                 kwargs["remote"] = self.remote
             with suppress(Exception):
+                assert self.repo
                 self.repo.cloud.pull([obj.hash_info], **kwargs)
 
         if self.obj:
@@ -1173,6 +1188,7 @@ class Output:
             return {}
 
         if self.remote:
+            assert self.repo
             remote_odb = self.repo.cloud.get_remote_odb(
                 name=self.remote, hash_name=self.hash_name
             )

dvc/repo/__init__.py

@@ -20,6 +20,7 @@ if TYPE_CHECKING:
     from dvc.fs.data import DataFileSystem
     from dvc.fs.dvc import DVCFileSystem
     from dvc.lock import LockBase
+    from dvc.output import Output
     from dvc.scm import Git, NoSCM
     from dvc.stage import Stage
     from dvc.types import DictStrAny
@@ -544,7 +545,9 @@ class Repo:
 
         return used
 
-    def find_outs_by_path(self, path, outs=None, recursive=False, strict=True):
+    def find_outs_by_path(
+        self, path, outs=None, recursive=False, strict=True
+    ) -> list["Output"]:
         # using `outs_graph` to ensure graph checks are run
         outs = outs or self.index.outs_graph
 

dvc/repo/move.py

@@ -1,10 +1,14 @@
 import os
+from typing import TYPE_CHECKING
 
 from dvc.exceptions import MoveNotDataSourceError
 from dvc.repo.scm_context import scm_context
 
 from . import locked
 
+if TYPE_CHECKING:
+    from . import Repo
+
 
 def _expand_target_path(from_path, to_path):
     if os.path.isdir(to_path):
@@ -14,7 +18,7 @@ def _expand_target_path(from_path, to_path):
 
 @locked
 @scm_context
-def move(self, from_path, to_path):
+def move(self: "Repo", from_path, to_path):
     """
     Renames an output file and modifies the stage associated
     to reflect the change on the pipeline.
@@ -70,24 +74,24 @@ def move(self, from_path, to_path):
             always_changed=stage.always_changed,
             desc=stage.desc,
         )
-
-        os.unlink(stage.path)
-        stage = new_stage
     else:
+        new_stage = stage
         to_path = os.path.relpath(to_path, stage.wdir)
 
     def with_dep_path_adjusted(dep: dependency.Dependency):
         d = dep.dumpd()
         if isinstance(dep.fs, LocalFileSystem) and not os.path.isabs(dep.def_path):
-            path = os.path.relpath(os.path.abspath(dep.def_path), stage.wdir)
-            return d | {"path": path}
+            return d | {"path": os.path.relpath(dep.fspath, new_stage.wdir)}
         return d
 
-    stage.outs = output.loadd_from(stage, [out.dumpd() | {"path": to_path}])
-    stage.deps = dependency.loadd_from(
-        stage, [with_dep_path_adjusted(dep) for dep in deps]
+    new_stage.outs = output.loadd_from(new_stage, [out.dumpd() | {"path": to_path}])
+    new_stage.deps = dependency.loadd_from(
+        new_stage, [with_dep_path_adjusted(dep) for dep in deps]
     )
-
-    out.move(stage.outs[0])
-    stage.md5 = stage.compute_md5()
-    stage.dump()
+    out.move(new_stage.outs[0])
+    new_stage.md5 = new_stage.compute_md5()
+    new_stage.dump()
+    if stage != new_stage:
+        stage.dvcfile.remove()
+        self.scm_context.track_file(stage.dvcfile.relpath)
+    return stage, new_stage

dvc/stage/__init__.py

@@ -32,7 +32,7 @@ from .utils import (
 )
 
 if TYPE_CHECKING:
-    from dvc.dependency import ParamsDependency
+    from dvc.dependency import Dependency, ParamsDependency
     from dvc.dvcfile import ProjectFile, SingleStageFile
     from dvc.output import Output
     from dvc.repo import Repo
@@ -153,8 +153,8 @@ class Stage(params.StageParams):
         self._path = path
         self.cmd = cmd
         self.wdir = wdir
-        self.outs = outs
-        self.deps = deps
+        self.outs: list[Output] = outs
+        self.deps: list[Dependency] = deps
         self.md5 = md5
         self.frozen = locked or frozen
         self.always_changed = always_changed

dvc/stage/serialize.py

@@ -178,7 +178,7 @@ def to_single_stage_lockfile(stage: "Stage", **kwargs) -> dict:
     res = OrderedDict([("cmd", stage.cmd)])
     params, deps = split_params_deps(stage)
     deps, outs = (
-        [_dumpd(item) for item in sorted(items, key=attrgetter("def_path"))]
+        [_dumpd(item) for item in sorted(items, key=attrgetter("def_path"))]  # type: ignore[call-overload]
         for items in [deps, stage.outs]
     )
     params = _serialize_params_values(params)

tests/func/test_move.py

@@ -10,8 +10,11 @@ from dvc.exceptions import MoveNotDataSourceError, OutputNotFoundError
 
 def test_move(tmp_dir, dvc, scm):
     tmp_dir.dvc_gen("foo", "bar")
+    assert (tmp_dir / "foo.dvc").exists()
     dvc.move("foo", "bar")
 
+    assert not (tmp_dir / "foo.dvc").exists()
+    assert (tmp_dir / "bar.dvc").exists()
     assert not (tmp_dir / "foo").is_file()
     assert (tmp_dir / "bar").is_file()
     # should only have the new path in the .gitignore, and only once
@@ -23,6 +26,24 @@ def test_move_non_existent_file(dvc):
         dvc.move("non_existent_file", "dst")
 
 
+def test_move_missing_file(tmp_dir, dvc, scm, caplog):
+    tmp_dir.dvc_gen("foo", "foo")
+    (tmp_dir / "foo").unlink()
+    contents = (tmp_dir / "foo.dvc").parse()
+    dvc.move("foo", "bar")
+
+    assert not (tmp_dir / "foo.dvc").exists()
+    # only the path should be changed in the dvc file
+    contents["outs"][0]["path"] = "bar"
+    assert contents == (tmp_dir / "bar.dvc").parse()
+
+    # file should not be checked out
+    assert not (tmp_dir / "foo").is_file()
+    assert not (tmp_dir / "bar").is_file()
+    # should only have the new path in the .gitignore, and only once
+    assert (tmp_dir / ".gitignore").read_text().splitlines() == ["/bar"]
+
+
 def test_move_directory(tmp_dir, dvc):
     tmp_dir.dvc_gen("data", {"foo": "foo", "bar": "bar"})
     dvc.move("data", "dst")