remote: base: don't checkout existing file (#2358)

* remote: base: don't checkout existing file

* remote: _checkout_file: split to smaller methods

* state: local: verify if file needs checkout considering link type and access rights

* tests: checkout: test avoiding unnecessary copying

* efiop review refactor

* remote: local: refactor _needs_checkout NOTE

* remote: local: remove bool cast for `protected`, wrap linking in try

* test: checkout: test relinking upon existing cache

* stage: remove obsolete unprotecing from Stage.create

* test: checkout: refactor setting link type

* test: checkout: parametrize relinking test

* remote: local: don't remove cache test file on cache_copy test

* remote: base: remove force from _needs_checkout arguments

* Update dvc/remote/local/__init__.py

Co-Authored-By: Alexander Schepanovski <suor.web@gmail.com>

* suor review refactor

* remote: base: re-add note why do we unprotect on _checkout_file

* test: checkout: tests for relinking and resetting target protection

* remove checkout behavior change

* remove unused imports

* remote: base: reuse __checkout_file in __checkout_dir

* initial dir

* checkout: test: should relink on repeated add

* remote: cleanup commented lines

* remote: base/local: add save_link flag to _checkout_file

* remote: local: add allow_copy flag to unprotect method

* Update dvc/remote/local/__init__.py

Co-Authored-By: Ruslan Kuprieiev <kupruser@gmail.com>

* remote: local: remove unnecessary removal warning on _checkout_file

* remote: local: proper cache link type detection

* test: checkout: move repeated add tests

* Update dvc/remote/local/__init__.py

Co-Authored-By: Ruslan Kuprieiev <kupruser@gmail.com>

* remote: base: remove _needs_checkout

* Update dvc/remote/base.py

Co-Authored-By: Ruslan Kuprieiev <kupruser@gmail.com>

* remote: base: remove wrappers from _checkout_file

Author: pared

dvc/remote/base.py

@@ -680,20 +680,45 @@ def safe_remove(self, path_info, force=False):
         self.remove(path_info)
 
     def _checkout_file(
-        self, path_info, checksum, force, progress_callback=None
+        self,
+        path_info,
+        checksum,
+        force,
+        progress_callback=None,
+        save_link=True,
     ):
-        cache_info = self.checksum_to_path_info(checksum)
-        if self.exists(path_info):
-            msg = "data '{}' exists. Removing before checkout."
-            logger.warning(msg.format(str(path_info)))
+        # NOTE: In case if path_info is already cached and path_info's
+        # link type matches cache link type, we would like to avoid
+        # relinking.
+        if self.changed(
+            path_info, {self.PARAM_CHECKSUM: checksum}
+        ) or not self._link_matches(path_info):
             self.safe_remove(path_info, force=force)
 
-        self.link(cache_info, path_info)
-        self.state.save_link(path_info)
-        self.state.save(path_info, checksum)
+            cache_info = self.checksum_to_path_info(checksum)
+            self.link(cache_info, path_info)
+
+            if save_link:
+                self.state.save_link(path_info)
+
+            self.state.save(path_info, checksum)
+        else:
+            # NOTE: performing (un)protection costs us +/- the same as checking
+            # if path_info is protected. Instead of implementing logic,
+            # just (un)protect according to self.protected.
+            if self.protected:
+                self.protect(path_info)
+            else:
+                # NOTE dont allow copy, because we checked before that link
+                # type matches cache, and we don't want data duplication
+                self.unprotect(path_info, allow_copy=False)
+
         if progress_callback:
             progress_callback(str(path_info))
 
+    def _link_matches(self, path_info):
+        return True
+
     def makedirs(self, path_info):
         raise NotImplementedError
 
@@ -712,17 +737,14 @@ def _checkout_dir(
         for entry in dir_info:
             relative_path = entry[self.PARAM_RELPATH]
             entry_checksum = entry[self.PARAM_CHECKSUM]
-            entry_cache_info = self.checksum_to_path_info(entry_checksum)
             entry_info = path_info / relative_path
-
-            entry_checksum_info = {self.PARAM_CHECKSUM: entry_checksum}
-            if self.changed(entry_info, entry_checksum_info):
-                if self.exists(entry_info):
-                    self.safe_remove(entry_info, force=force)
-                self.link(entry_cache_info, entry_info)
-                self.state.save(entry_info, entry_checksum)
-            if progress_callback:
-                progress_callback(str(entry_info))
+            self._checkout_file(
+                entry_info,
+                entry_checksum,
+                force,
+                progress_callback,
+                save_link=False,
+            )
 
         self._remove_redundant_files(path_info, dir_info, force)
 
@@ -803,7 +825,7 @@ def get_files_number(self, checksum):
         return 1
 
     @staticmethod
-    def unprotect(path_info):
+    def unprotect(path_info, allow_copy=True):
         pass
 
     def _get_unpacked_dir_names(self, checksums):

dvc/remote/local/__init__.py

@@ -78,6 +78,7 @@ def __init__(self, repo, config):
             self.cache_types = types
         else:
             self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
+        self.cache_type_confirmed = False
 
         # A clunky way to detect cache dir
         storagepath = config.get(Config.SECTION_LOCAL_STORAGEPATH, None)
@@ -188,6 +189,7 @@ def _try_links(self, from_info, to_info, link_types):
             link_method = self._get_link_method(link_types[0])
             try:
                 self._do_link(from_info, to_info, link_method)
+                self.cache_type_confirmed = True
                 return
 
             except DvcException as exc:
@@ -471,19 +473,22 @@ def _log_missing_caches(checksum_info_dict):
             logger.warning(msg)
 
     @staticmethod
-    def _unprotect_file(path):
+    def _unprotect_file(path, allow_copy=True):
         if System.is_symlink(path) or System.is_hardlink(path):
-            logger.debug("Unprotecting '{}'".format(path))
-            tmp = os.path.join(os.path.dirname(path), "." + str(uuid()))
-
-            # The operations order is important here - if some application
-            # would access the file during the process of copyfile then it
-            # would get only the part of file. So, at first, the file should be
-            # copied with the temporary name, and then original file should be
-            # replaced by new.
-            copyfile(path, tmp, name="Unprotecting '{}'".format(relpath(path)))
-            remove(path)
-            os.rename(tmp, path)
+            if allow_copy:
+                logger.debug("Unprotecting '{}'".format(path))
+                tmp = os.path.join(os.path.dirname(path), "." + str(uuid()))
+
+                # The operations order is important here - if some application
+                # would access the file during the process of copyfile then it
+                # would get only the part of file. So, at first, the file
+                # should be copied with the temporary name, and then
+                # original file should be replaced by new.
+                copyfile(
+                    path, tmp, name="Unprotecting '{}'".format(relpath(path))
+                )
+                remove(path)
+                os.rename(tmp, path)
 
         else:
             logger.debug(
@@ -493,21 +498,21 @@ def _unprotect_file(path):
 
         os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)
 
-    def _unprotect_dir(self, path):
+    def _unprotect_dir(self, path, allow_copy=True):
         for fname in walk_files(path, self.repo.dvcignore):
-            RemoteLOCAL._unprotect_file(fname)
+            self._unprotect_file(fname, allow_copy)
 
-    def unprotect(self, path_info):
+    def unprotect(self, path_info, allow_copy=True):
         path = path_info.fspath
         if not os.path.exists(path):
             raise DvcException(
                 "can't unprotect non-existing data '{}'".format(path)
             )
 
         if os.path.isdir(path):
-            self._unprotect_dir(path)
+            self._unprotect_dir(path, allow_copy)
         else:
-            RemoteLOCAL._unprotect_file(path)
+            self._unprotect_file(path, allow_copy)
 
     @staticmethod
     def protect(path_info):
@@ -581,3 +586,36 @@ def _get_unpacked_dir_names(self, checksums):
             if self.is_dir_checksum(c):
                 unpacked.add(c + self.UNPACKED_DIR_SUFFIX)
         return unpacked
+
+    def _get_cache_type(self, path_info):
+        if self.cache_type_confirmed:
+            return self.cache_types[0]
+
+        workspace_file = path_info.with_name("." + uuid())
+        test_cache_file = self.path_info / ".cache_type_test_file"
+        if not self.exists(test_cache_file):
+            with open(fspath_py35(test_cache_file), "wb") as fobj:
+                fobj.write(bytes(1))
+        try:
+            self.link(test_cache_file, workspace_file)
+        finally:
+            self.remove(workspace_file)
+            self.remove(test_cache_file)
+
+        self.cache_type_confirmed = True
+        return self.cache_types[0]
+
+    def _link_matches(self, path_info):
+        is_hardlink = System.is_hardlink(path_info)
+        is_symlink = System.is_symlink(path_info)
+        is_copy_or_reflink = not is_hardlink and not is_symlink
+
+        cache_type = self._get_cache_type(path_info)
+
+        if cache_type == "symlink":
+            return is_symlink
+
+        if cache_type == "hardlink":
+            return is_hardlink
+
+        return is_copy_or_reflink

tests/func/test_add.py

@@ -553,3 +553,81 @@ def test_readding_dir_should_not_unprotect_all(dvc_repo, repo_dir):
 
     assert not unprotect_spy.mock.called
     assert System.is_symlink(new_file)
+
+
+def test_should_not_checkout_when_adding_cached_copy(repo_dir, dvc_repo):
+    dvc_repo.cache.local.cache_types = ["copy"]
+
+    dvc_repo.add(repo_dir.FOO)
+    dvc_repo.add(repo_dir.BAR)
+
+    shutil.copy(repo_dir.BAR, repo_dir.FOO)
+
+    copy_spy = spy(shutil.copyfile)
+
+    RemoteLOCAL.CACHE_TYPE_MAP["copy"] = copy_spy
+    dvc_repo.add(repo_dir.FOO)
+
+    assert copy_spy.mock.call_count == 0
+
+
+@pytest.mark.parametrize(
+    "link,new_link,link_test_func",
+    [
+        ("hardlink", "copy", lambda path: not System.is_hardlink(path)),
+        ("symlink", "copy", lambda path: not System.is_symlink(path)),
+        ("copy", "hardlink", System.is_hardlink),
+        ("copy", "symlink", System.is_symlink),
+    ],
+)
+def test_should_relink_on_repeated_add(
+    link, new_link, link_test_func, repo_dir, dvc_repo
+):
+    dvc_repo.cache.local.cache_types = [link]
+
+    dvc_repo.add(repo_dir.FOO)
+    dvc_repo.add(repo_dir.BAR)
+
+    os.remove(repo_dir.FOO)
+    RemoteLOCAL.CACHE_TYPE_MAP[link](repo_dir.BAR, repo_dir.FOO)
+
+    dvc_repo.cache.local.cache_types = [new_link]
+
+    dvc_repo.add(repo_dir.FOO)
+
+    assert link_test_func(repo_dir.FOO)
+
+
+@pytest.mark.parametrize(
+    "link, link_func",
+    [("hardlink", System.hardlink), ("symlink", System.symlink)],
+)
+def test_should_relink_single_file_in_dir(link, link_func, dvc_repo, repo_dir):
+    dvc_repo.cache.local.cache_types = [link]
+
+    dvc_repo.add(repo_dir.DATA_DIR)
+
+    # NOTE status triggers unpacked dir creation for hardlink case
+    dvc_repo.status()
+
+    dvc_repo.unprotect(repo_dir.DATA_SUB)
+
+    link_spy = spy(link_func)
+    RemoteLOCAL.CACHE_TYPE_MAP[link] = link_spy
+    dvc_repo.add(repo_dir.DATA_DIR)
+
+    assert link_spy.mock.call_count == 1
+
+
+@pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
+def test_should_protect_on_repeated_add(link, dvc_repo, repo_dir):
+    dvc_repo.cache.local.cache_types = [link]
+    dvc_repo.cache.local.protected = True
+
+    dvc_repo.add(repo_dir.FOO)
+
+    dvc_repo.unprotect(repo_dir.FOO)
+
+    dvc_repo.add(repo_dir.FOO)
+
+    assert not os.access(repo_dir.FOO, os.W_OK)