From 320715a5f806415e93ab07d90d553cd8e57b320f Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Fri, 5 Feb 2021 03:27:00 +0200 Subject: [PATCH] dvc: move checkout out of cache (#5415) Checkout is an operation that builds object from cache(aka object database) in a particular tree(aka filesystem). Thus it shouldn't be part of the object database itself nor of the filesystem. Pre-requisite for #5337 --- dvc/cache/base.py | 301 +-------------------------------- dvc/cache/local.py | 5 - dvc/checkout.py | 297 ++++++++++++++++++++++++++++++++ dvc/dependency/repo.py | 4 +- dvc/output/base.py | 8 +- tests/func/test_run_cache.py | 4 +- tests/unit/stage/test_cache.py | 8 +- 7 files changed, 319 insertions(+), 308 deletions(-) create mode 100644 dvc/checkout.py diff --git a/dvc/cache/base.py b/dvc/cache/base.py index 54f60c568..560cdd317 100644 --- a/dvc/cache/base.py +++ b/dvc/cache/base.py @@ -11,14 +11,8 @@ from typing import Optional from funcy import decorator from shortuuid import uuid -import dvc.prompt as prompt from dvc.dir_info import DirInfo -from dvc.exceptions import ( - CacheLinkError, - CheckoutError, - ConfirmRemoveError, - DvcException, -) +from dvc.exceptions import CacheLinkError, DvcException from dvc.progress import Tqdm from dvc.remote.slow_link_detection import ( # type: ignore[attr-defined] slow_link_guard, @@ -93,77 +87,6 @@ class CloudCache: return DirInfo.from_list(d) - def _filter_hash_info(self, hash_info, path_info, filter_info): - if not filter_info or path_info == filter_info: - return hash_info - - dir_info = self.get_dir_cache(hash_info) - hash_key = filter_info.relative_to(path_info).parts - - # Check whether it is a file that exists on the trie - hi = dir_info.trie.get(hash_key) - if hi: - return hi - - depth = len(hash_key) - filtered_dir_info = DirInfo() - try: - for key, value in dir_info.trie.items(hash_key): - filtered_dir_info.trie[key[depth:]] = value - except KeyError: - return None - - return self._get_dir_info_hash(filtered_dir_info)[0] - - def changed(self, path_info, tree, hash_info, filter_info=None): - """Checks if data has changed. - - A file is considered changed if: - - It doesn't exist on the working directory (was unlinked) - - Hash value is not computed (saving a new file) - - The hash value stored is different from the given one - - There's no file in the cache - - Args: - path_info: dict with path information. - hash: expected hash value for this data. - filter_info: an optional argument to target a specific path. - - Returns: - bool: True if data has changed, False otherwise. - """ - - path = filter_info or path_info - logger.trace( - "checking if '%s'('%s') has changed.", path_info, hash_info - ) - - if not tree.exists(path): - logger.debug("'%s' doesn't exist.", path) - return True - - hi = self._filter_hash_info(hash_info, path_info, filter_info) - if not hi: - logger.debug("hash value for '%s' is missing.", path) - return True - - if self.changed_cache(hi): - logger.debug("cache for '%s'('%s') has changed.", path, hi) - return True - - actual = tree.get_hash(path) - if hi != actual: - logger.debug( - "hash value '%s' for '%s' has changed (actual '%s').", - hi, - actual, - path, - ) - return True - - logger.trace("'%s' hasn't changed.", path) - return False - def link(self, from_info, to_info): self._link(from_info, to_info, self.cache_types) @@ -301,7 +224,7 @@ class CloudCache: ( hash_info, to_info, - ) = local_cache._get_dir_info_hash( # pylint: disable=protected-access + ) = local_cache.get_dir_info_hash( # pylint: disable=protected-access dir_info ) @@ -320,7 +243,7 @@ class CloudCache: ) return self._transfer_file(from_tree, from_info) - def _cache_is_copy(self, path_info): + def cache_is_copy(self, path_info): """Checks whether cache uses copies.""" if self.cache_type_confirmed: return self.cache_types[0] == "copy" @@ -342,7 +265,7 @@ class CloudCache: self.cache_type_confirmed = True return self.cache_types[0] == "copy" - def _get_dir_info_hash(self, dir_info): + def get_dir_info_hash(self, dir_info): import tempfile from dvc.path_info import PathInfo @@ -372,7 +295,7 @@ class CloudCache: ): return hash_info - hi, tmp_info = self._get_dir_info_hash(dir_info) + hi, tmp_info = self.get_dir_info_hash(dir_info) new_info = self.tree.hash_to_path_info(hi.value) if self.changed_cache_file(hi): self.makedirs(new_info.parent) @@ -515,220 +438,6 @@ class CloudCache: ) return self.changed_cache_file(hash_info) - def already_cached(self, path_info, tree): - current = tree.get_hash(path_info) - - if not current: - return False - - return not self.changed_cache(current) - - def safe_remove(self, path_info, tree, force=False): - if not tree.exists(path_info): - return - - if not force and not self.already_cached(path_info, tree): - msg = ( - "file '{}' is going to be removed." - " Are you sure you want to proceed?".format(str(path_info)) - ) - - if not prompt.confirm(msg): - raise ConfirmRemoveError(str(path_info)) - - tree.remove(path_info) - - def _checkout_file( - self, - path_info, - tree, - hash_info, - force, - progress_callback=None, - relink=False, - ): - """The file is changed we need to checkout a new copy""" - cache_info = self.tree.hash_to_path_info(hash_info.value) - if tree.exists(path_info): - added = False - - if not relink and self.changed(path_info, tree, hash_info): - modified = True - self.safe_remove(path_info, tree, force=force) - self.link(cache_info, path_info) - else: - modified = False - - if tree.iscopy(path_info) and self._cache_is_copy(path_info): - self.unprotect(path_info) - else: - self.safe_remove(path_info, tree, force=force) - self.link(cache_info, path_info) - else: - self.link(cache_info, path_info) - added, modified = True, False - - tree.state.save(path_info, hash_info) - if progress_callback: - progress_callback(str(path_info)) - - return added, modified - - def _checkout_dir( - self, - path_info, - tree, - hash_info, - force, - progress_callback=None, - relink=False, - filter_info=None, - ): - added, modified = False, False - # Create dir separately so that dir is created - # even if there are no files in it - if not tree.exists(path_info): - added = True - self.makedirs(path_info) - - dir_info = self.get_dir_cache(hash_info) - - logger.debug("Linking directory '%s'.", path_info) - - for entry_info, entry_hash_info in dir_info.items(path_info): - if filter_info and not entry_info.isin_or_eq(filter_info): - continue - - entry_added, entry_modified = self._checkout_file( - entry_info, - tree, - entry_hash_info, - force, - progress_callback, - relink, - ) - if entry_added or entry_modified: - modified = True - - modified = ( - self._remove_redundant_files(path_info, tree, dir_info, force) - or modified - ) - - tree.state.save(path_info, hash_info) - - # relink is not modified, assume it as nochange - return added, not added and modified and not relink - - def _remove_redundant_files(self, path_info, tree, dir_info, force): - existing_files = set(tree.walk_files(path_info)) - - needed_files = {info for info, _ in dir_info.items(path_info)} - redundant_files = existing_files - needed_files - for path in redundant_files: - self.safe_remove(path, tree, force) - - return bool(redundant_files) - - @use_state - def checkout( - self, - path_info, - tree, - hash_info, - force=False, - progress_callback=None, - relink=False, - filter_info=None, - quiet=False, - ): - if path_info.scheme not in ["local", self.tree.scheme]: - raise NotImplementedError - - failed = None - skip = False - if not hash_info: - if not quiet: - logger.warning( - "No file hash info found for '%s'. It won't be created.", - path_info, - ) - self.safe_remove(path_info, tree, force=force) - failed = path_info - - elif not relink and not self.changed( - path_info, tree, hash_info, filter_info=filter_info - ): - logger.trace("Data '%s' didn't change.", path_info) - skip = True - - elif self.changed_cache( - hash_info, path_info=path_info, filter_info=filter_info - ): - if not quiet: - logger.warning( - "Cache '%s' not found. File '%s' won't be created.", - hash_info, - path_info, - ) - self.safe_remove(path_info, tree, force=force) - failed = path_info - - if failed or skip: - if progress_callback: - progress_callback( - str(path_info), - self.get_files_number( - self.tree.path_info, hash_info, filter_info - ), - ) - if failed: - raise CheckoutError([failed]) - return - - logger.debug( - "Checking out '%s' with cache '%s'.", path_info, hash_info - ) - - return self._checkout( - path_info, - tree, - hash_info, - force, - progress_callback, - relink, - filter_info, - ) - - def _checkout( - self, - path_info, - tree, - hash_info, - force=False, - progress_callback=None, - relink=False, - filter_info=None, - ): - if not hash_info.isdir: - ret = self._checkout_file( - path_info, tree, hash_info, force, progress_callback, relink - ) - else: - ret = self._checkout_dir( - path_info, - tree, - hash_info, - force, - progress_callback, - relink, - filter_info, - ) - - tree.state.save_link(path_info) - - return ret - def get_files_number(self, path_info, hash_info, filter_info): from funcy.py3 import ilen diff --git a/dvc/cache/local.py b/dvc/cache/local.py index 911c39bb9..75855029e 100644 --- a/dvc/cache/local.py +++ b/dvc/cache/local.py @@ -81,11 +81,6 @@ class LocalCache(CloudCache): ) ] - def already_cached(self, path_info, tree): - assert path_info.scheme in ["", "local"] - - return super().already_cached(path_info, tree) - def _verify_link(self, path_info, link_type): if link_type == "hardlink" and self.tree.getsize(path_info) == 0: return diff --git a/dvc/checkout.py b/dvc/checkout.py new file mode 100644 index 000000000..a827527df --- /dev/null +++ b/dvc/checkout.py @@ -0,0 +1,297 @@ +import logging + +import dvc.prompt as prompt +from dvc.dir_info import DirInfo +from dvc.exceptions import CheckoutError, ConfirmRemoveError + +logger = logging.getLogger(__name__) + + +def _filter_hash_info(cache, hash_info, path_info, filter_info): + if not filter_info or path_info == filter_info: + return hash_info + + dir_info = cache.get_dir_cache(hash_info) + hash_key = filter_info.relative_to(path_info).parts + + # Check whether it is a file that exists on the trie + hi = dir_info.trie.get(hash_key) + if hi: + return hi + + depth = len(hash_key) + filtered_dir_info = DirInfo() + try: + for key, value in dir_info.trie.items(hash_key): + filtered_dir_info.trie[key[depth:]] = value + except KeyError: + return None + + return cache.get_dir_info_hash(filtered_dir_info)[0] + + +def _changed(path_info, tree, hash_info, cache, filter_info=None): + """Checks if data has changed. + + A file is considered changed if: + - It doesn't exist on the working directory (was unlinked) + - Hash value is not computed (saving a new file) + - The hash value stored is different from the given one + - There's no file in the cache + + Args: + path_info: dict with path information. + hash: expected hash value for this data. + filter_info: an optional argument to target a specific path. + + Returns: + bool: True if data has changed, False otherwise. + """ + + path = filter_info or path_info + logger.trace("checking if '%s'('%s') has changed.", path_info, hash_info) + + if not tree.exists(path): + logger.debug("'%s' doesn't exist.", path) + return True + + hi = _filter_hash_info(cache, hash_info, path_info, filter_info) + if not hi: + logger.debug("hash value for '%s' is missing.", path) + return True + + if cache.changed_cache(hi): + logger.debug("cache for '%s'('%s') has changed.", path, hi) + return True + + actual = tree.get_hash(path) + if hi != actual: + logger.debug( + "hash value '%s' for '%s' has changed (actual '%s').", + hi, + actual, + path, + ) + return True + + logger.trace("'%s' hasn't changed.", path) + return False + + +def _is_cached(cache, path_info, tree): + current = tree.get_hash(path_info) + + if not current: + return False + + return not cache.changed_cache(current) + + +def _remove(path_info, tree, cache, force=False): + if not tree.exists(path_info): + return + + if not force and not _is_cached(cache, path_info, tree): + msg = ( + "file '{}' is going to be removed." + " Are you sure you want to proceed?".format(str(path_info)) + ) + + if not prompt.confirm(msg): + raise ConfirmRemoveError(str(path_info)) + + tree.remove(path_info) + + +def _checkout_file( + path_info, + tree, + hash_info, + cache, + force, + progress_callback=None, + relink=False, +): + """The file is changed we need to checkout a new copy""" + cache_info = cache.tree.hash_to_path_info(hash_info.value) + if tree.exists(path_info): + added = False + + if not relink and _changed(path_info, tree, hash_info, cache): + modified = True + _remove(path_info, tree, cache, force=force) + cache.link(cache_info, path_info) + else: + modified = False + + if tree.iscopy(path_info) and cache.cache_is_copy(path_info): + cache.unprotect(path_info) + else: + _remove(path_info, tree, cache, force=force) + cache.link(cache_info, path_info) + else: + cache.link(cache_info, path_info) + added, modified = True, False + + tree.state.save(path_info, hash_info) + if progress_callback: + progress_callback(str(path_info)) + + return added, modified + + +def _remove_redundant_files(path_info, tree, dir_info, cache, force): + existing_files = set(tree.walk_files(path_info)) + + needed_files = {info for info, _ in dir_info.items(path_info)} + redundant_files = existing_files - needed_files + for path in redundant_files: + _remove(path, tree, cache, force) + + return bool(redundant_files) + + +def _checkout_dir( + path_info, + tree, + hash_info, + cache, + force, + progress_callback=None, + relink=False, + filter_info=None, +): + added, modified = False, False + # Create dir separately so that dir is created + # even if there are no files in it + if not tree.exists(path_info): + added = True + tree.makedirs(path_info) + + dir_info = cache.get_dir_cache(hash_info) + + logger.debug("Linking directory '%s'.", path_info) + + for entry_info, entry_hash_info in dir_info.items(path_info): + if filter_info and not entry_info.isin_or_eq(filter_info): + continue + + entry_added, entry_modified = _checkout_file( + entry_info, + tree, + entry_hash_info, + cache, + force, + progress_callback, + relink, + ) + if entry_added or entry_modified: + modified = True + + modified = ( + _remove_redundant_files(path_info, tree, dir_info, cache, force) + or modified + ) + + tree.state.save(path_info, hash_info) + + # relink is not modified, assume it as nochange + return added, not added and modified and not relink + + +def _checkout( + path_info, + tree, + hash_info, + cache, + force=False, + progress_callback=None, + relink=False, + filter_info=None, +): + if not hash_info.isdir: + ret = _checkout_file( + path_info, tree, hash_info, cache, force, progress_callback, relink + ) + else: + ret = _checkout_dir( + path_info, + tree, + hash_info, + cache, + force, + progress_callback, + relink, + filter_info, + ) + + tree.state.save_link(path_info) + + return ret + + +def checkout( + path_info, + tree, + hash_info, + cache, + force=False, + progress_callback=None, + relink=False, + filter_info=None, + quiet=False, +): + if path_info.scheme not in ["local", cache.tree.scheme]: + raise NotImplementedError + + failed = None + skip = False + if not hash_info: + if not quiet: + logger.warning( + "No file hash info found for '%s'. It won't be created.", + path_info, + ) + _remove(path_info, tree, cache, force=force) + failed = path_info + + elif not relink and not _changed( + path_info, tree, hash_info, cache, filter_info=filter_info + ): + logger.trace("Data '%s' didn't change.", path_info) + skip = True + + elif cache.changed_cache( + hash_info, path_info=path_info, filter_info=filter_info + ): + if not quiet: + logger.warning( + "Cache '%s' not found. File '%s' won't be created.", + hash_info, + path_info, + ) + _remove(path_info, tree, cache, force=force) + failed = path_info + + if failed or skip: + if progress_callback: + progress_callback( + str(path_info), + cache.get_files_number(path_info, hash_info, filter_info), + ) + if failed: + raise CheckoutError([failed]) + return + + logger.debug("Checking out '%s' with cache '%s'.", path_info, hash_info) + + return _checkout( + path_info, + tree, + hash_info, + cache, + force, + progress_callback, + relink, + filter_info, + ) diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index 65b8df286..dfaf3829d 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -70,6 +70,8 @@ class RepoDependency(LocalDependency): return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo} def download(self, to, jobs=None): + from dvc.checkout import checkout + cache = self.repo.cache.local with self._make_repo(cache_dir=cache.cache_dir) as repo: @@ -84,7 +86,7 @@ class RepoDependency(LocalDependency): follow_subrepos=False, ) - cache.checkout(to.path_info, to.tree, hash_info) + checkout(to.path_info, to.tree, hash_info, cache) def update(self, rev=None): if rev: diff --git a/dvc/output/base.py b/dvc/output/base.py index 881889acb..06d1ddbff 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -8,6 +8,7 @@ from voluptuous import Any import dvc.prompt as prompt from dvc.cache import NamedCache +from dvc.checkout import checkout from dvc.exceptions import ( CheckoutError, CollectCacheError, @@ -315,10 +316,12 @@ class BaseOutput: self.hash_info, filter_info=filter_info, ) - self.cache.checkout( + + checkout( self.path_info, self.tree, self.hash_info, + self.cache, relink=True, filter_info=filter_info, ) @@ -387,10 +390,11 @@ class BaseOutput: return None try: - res = self.cache.checkout( + res = checkout( self.path_info, self.tree, self.hash_info, + self.cache, force=force, progress_callback=progress_callback, relink=relink, diff --git a/tests/func/test_run_cache.py b/tests/func/test_run_cache.py index 4bb7d503b..bf84550bd 100644 --- a/tests/func/test_run_cache.py +++ b/tests/func/test_run_cache.py @@ -150,6 +150,8 @@ def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker): def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote): + from dvc.output import base + tmp_dir.gen("foo", "foo") stage = run_copy("foo", "bar", name="copy-foo-bar") @@ -157,7 +159,7 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote): mock_restore = mocker.spy(dvc.stage_cache, "restore") mock_run = mocker.patch("dvc.stage.run.cmd_run") - mock_checkout = mocker.spy(dvc.cache.local, "checkout") + mock_checkout = mocker.spy(base, "checkout") # removing any information that `dvc` could use to re-generate from (tmp_dir / "bar").unlink() diff --git a/tests/unit/stage/test_cache.py b/tests/unit/stage/test_cache.py index b6f56c75d..4e26dd70e 100644 --- a/tests/unit/stage/test_cache.py +++ b/tests/unit/stage/test_cache.py @@ -2,6 +2,8 @@ import os import pytest +from dvc.output import base + def test_stage_cache(tmp_dir, dvc, mocker): tmp_dir.gen("dep", "dep") @@ -42,7 +44,7 @@ def test_stage_cache(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(dvc.cache.local, "checkout") + checkout_spy = mocker.spy(base, "checkout") with dvc.lock, dvc.state: stage.run() @@ -95,7 +97,7 @@ def test_stage_cache_params(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(dvc.cache.local, "checkout") + checkout_spy = mocker.spy(base, "checkout") with dvc.lock, dvc.state: stage.run() @@ -149,7 +151,7 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(dvc.cache.local, "checkout") + checkout_spy = mocker.spy(base, "checkout") with dvc.lock, dvc.state: stage.run()