Repository: iterative/dvc
Pull request: remote: base: don't checkout existing file
* [x] Have you followed the guidelines in our
      [Contributing document](https://dvc.org/doc/user-guide/contributing)?

* [x] Does your PR affect documented changes or does it add new functionality
      that should be documented? If yes, have you created a PR for
      [dvc.org](https://github.com/iterative/dvc.org) documenting it or at
      least opened an issue for it? If so, please add a link to it.

-----
Fixes #2016 

Problem statement:
Misleading warning (Removing before checkout)
During a manual update of a file, I've faced the following message:
>WARNING: data 'foo' exists. Removing before checkout.

It puzzled me, I thought it would overwrite my file, since checkout restores the file based on current DVC file.
I don't know why the file was replaced by the same file. It took a long time (huge file on a HDD).

@efiop provided me a script that reproduces the message:
```
#!/bin/bash     
                
set -e          
set -x          
                
rm -rf myrepo   
mkdir myrepo    
cd myrepo       
                
git init        
dvc init        
                
echo foo > foo  
dvc add foo     
                
echo bar > bar  
dvc add bar     
                
echo bar > foo  
dvc add foo     
cat foo         
dvc status
```

I think a better message and avoiding re-copying the file would be great improvements.

Hints from issue discussion:
Discord context: https://discordapp.com/channels/485586884165107732/485596304961962003/579053609271164929
Also when running `dvc run -o metrics.txt "echo 0.9643 > metrics.txt"` twice or if metrics.txt is already in cache.
Need to make sure this also works properly for directories.

Commit message: test: checkout: move repeated add tests

Patch to review:
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 619b4799d2..1f782057fe 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -680,20 +680,33 @@ 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)))
+        if self._needs_checkout(path_info, checksum):
             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:
+            if self.protected:
+                self.protect(path_info)
+            else:
+                self.unprotect(path_info, False)
+
         if progress_callback:
             progress_callback(str(path_info))
 
+    def _needs_checkout(self, path_info, checksum):
+        return True
+
     def makedirs(self, path_info):
         raise NotImplementedError
 
@@ -712,17 +725,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)
 
@@ -800,7 +810,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):
diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py
index 168c1f9c09..ba0822ff68 100644
--- a/dvc/remote/local/__init__.py
+++ b/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,11 +498,11 @@ 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)
+            RemoteLOCAL._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(
@@ -505,9 +510,9 @@ def unprotect(self, path_info):
             )
 
         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,52 @@ def _get_unpacked_dir_names(self, checksums):
             if self.is_dir_checksum(c):
                 unpacked.add(c + self.UNPACKED_DIR_SUFFIX)
         return unpacked
+
+    def _needs_checkout(self, path_info, checksum):
+        # NOTE: In case if path_info is already cached, and cache type is
+        # 'copy' we would like to avoid relinking.
+        # `symlink/hardlink/reflink` is cheap, but if we need `copy` links,
+        # then to save on copying we should only relink if we already have
+        # `hardlink` or a `symlink`, we don't care about reflinks, because
+        # they are indistinguishable from copy anyway.
+
+        if not self.changed(
+            path_info, {self.PARAM_CHECKSUM: checksum}
+        ) and self._is_same_link_as_cache(path_info):
+            msg = "File '{}' didn't change"
+            logger.debug(msg.format(str(path_info)))
+            return False
+        return True
+
+    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 _is_same_link_as_cache(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
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 6e7516c498..c11e0a3b6a 100644
--- a/tests/func/test_add.py
+++ b/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)


Code review comments:
diff_hunk: @@ -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)
+            RemoteLOCAL._unprotect_file(fname, allow_copy)
path: dvc/remote/local/__init__.py
text: @user1:
Sorry for picking up these, but while we are at it, we might as well...
```suggestion
            self._unprotect_file(fname, allow_copy)
```

@author:
done

diff_hunk: @@ -581,3 +586,52 @@ def _get_unpacked_dir_names(self, checksums):
             if self.is_dir_checksum(c):
                 unpacked.add(c + self.UNPACKED_DIR_SUFFIX)
         return unpacked
+
+    def _needs_checkout(self, path_info, checksum):
+        # NOTE: In case if path_info is already cached, and cache type is
+        # 'copy' we would like to avoid relinking.
+        # `symlink/hardlink/reflink` is cheap, but if we need `copy` links,
+        # then to save on copying we should only relink if we already have
+        # `hardlink` or a `symlink`, we don't care about reflinks, because
+        # they are indistinguishable from copy anyway.
+
+        if not self.changed(
+            path_info, {self.PARAM_CHECKSUM: checksum}
+        ) and self._is_same_link_as_cache(path_info):
+            msg = "File '{}' didn't change"
+            logger.debug(msg.format(str(path_info)))
path: dvc/remote/local/__init__.py
text: @user1:
I don't think we need this message, as self.changed already prints it.

```suggestion
```

@user1:
And in that case, having this wrapper function seems a bit excessive :slightly_smiling_face: You could totally replace 
```
       if self._needs_checkout(path_info, checksum):
```
in _checkout_file with
```
       if not self._link_type_matches(path_info) or self.changed(path_info, checksum):
           ...
```
and have it more readable, since it describes it in one statement. Also, note that I've put link type detection first, since it should be faster than `self.changed()`.

@author:
done

@author:
Done, still I left `changed` check first, as it includes checking if `path_info` exists. We can either duplicate `exists` check or leave `changed` first.