From 3d84df930b464216ab592d5a2ad740d90f9edeaf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20Rowlands=20=28=EB=B3=80=EA=B8=B0=ED=98=B8=29?= Date: Tue, 20 Jul 2021 19:30:01 +0900 Subject: [PATCH] objects: migrate remote push/pull to objects.transfer (#6308) * objects: migrate remote status to objects.status * data_cloud: use objects.status * objects: migrate remote._process to objects.transfer * data_cloud: use objects.transfer * objects: handle memfs staging objects in transfer * fetch/import: use unified transfer() * update get_remote and transfer exception usage * update tests * objects.transfer: use src ODB verify rules after xfer * odb: move index from remote into odb * objects.transfer: skip src status check when possible * update tests * remove dvc.remote * odb.index: migrate to diskcache * objects: fix status/transfer optimization * update index usage * state: remove dead sqlite related code * update cloud tests * drop unnecessary remove in tests * use abstract base class in odb index * status: load trees from local cache instead of remote odb when possible * handle status when dir cache is explicitly removed --- dvc/checkout.py | 6 +- dvc/data_cloud.py | 73 ++- dvc/dependency/repo.py | 11 +- dvc/exceptions.py | 15 +- dvc/fs/dvc.py | 6 +- dvc/objects/db/__init__.py | 18 + dvc/objects/db/base.py | 15 +- dvc/objects/db/index.py | 137 +++++ .../db}/slow_link_detection.py | 10 +- dvc/objects/stage.py | 5 + dvc/objects/status.py | 185 ++++++ dvc/objects/transfer.py | 218 +++++++ dvc/objects/tree.py | 4 +- dvc/output.py | 2 +- dvc/remote/__init__.py | 11 - dvc/remote/base.py | 555 ------------------ dvc/remote/index.py | 213 ------- dvc/remote/local.py | 6 - dvc/repo/add.py | 3 +- dvc/repo/fetch.py | 82 +-- dvc/repo/gc.py | 9 +- dvc/repo/imp_url.py | 4 +- dvc/repo/push.py | 10 +- dvc/repo/status.py | 18 +- dvc/stage/cache.py | 21 +- dvc/stage/imports.py | 4 +- dvc/state.py | 49 -- tests/func/{remote => objects}/__init__.py | 0 tests/func/objects/db/__init__.py | 0 tests/func/objects/db/test_index.py | 88 +++ tests/func/remote/test_index.py | 109 ---- tests/func/test_checkout.py | 4 +- tests/func/test_data_cloud.py | 81 +-- tests/func/test_external_repo.py | 2 +- tests/func/test_import.py | 16 +- tests/func/test_remote.py | 80 ++- tests/func/test_update.py | 5 +- tests/unit/objects/db/test_local.py | 14 +- tests/unit/remote/test_index.py | 21 +- tests/unit/remote/test_remote_tree.py | 32 +- tests/unit/remote/test_slow_link_detection.py | 10 +- tests/unit/test_state.py | 23 - 42 files changed, 890 insertions(+), 1285 deletions(-) create mode 100644 dvc/objects/db/index.py rename dvc/{remote => objects/db}/slow_link_detection.py (79%) create mode 100644 dvc/objects/status.py create mode 100644 dvc/objects/transfer.py delete mode 100644 dvc/remote/__init__.py delete mode 100644 dvc/remote/base.py delete mode 100644 dvc/remote/index.py delete mode 100644 dvc/remote/local.py rename tests/func/{remote => objects}/__init__.py (100%) create mode 100644 tests/func/objects/db/__init__.py create mode 100644 tests/func/objects/db/test_index.py delete mode 100644 tests/func/remote/test_index.py delete mode 100644 tests/unit/test_state.py diff --git a/dvc/checkout.py b/dvc/checkout.py index ae29a5db9..c69f58e64 100644 --- a/dvc/checkout.py +++ b/dvc/checkout.py @@ -11,11 +11,11 @@ from dvc.exceptions import ( ) from dvc.ignore import DvcIgnoreFilter from dvc.objects import check, load -from dvc.objects.errors import ObjectFormatError -from dvc.objects.stage import stage -from dvc.remote.slow_link_detection import ( # type: ignore[attr-defined] +from dvc.objects.db.slow_link_detection import ( # type: ignore[attr-defined] slow_link_guard, ) +from dvc.objects.errors import ObjectFormatError +from dvc.objects.stage import stage from dvc.types import Optional logger = logging.getLogger(__name__) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index b4a3d5a87..687237b29 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -3,9 +3,11 @@ import logging from typing import TYPE_CHECKING, Iterable, Optional +from dvc.objects.db import get_index + if TYPE_CHECKING: + from dvc.objects.db.base import ObjectDB from dvc.objects.file import HashFile - from dvc.remote.base import Remote logger = logging.getLogger(__name__) @@ -24,18 +26,18 @@ class DataCloud: def __init__(self, repo): self.repo = repo - def get_remote( + def get_remote_odb( self, name: Optional[str] = None, command: str = "", - ) -> "Remote": + ) -> "ObjectDB": from dvc.config import NoRemoteError if not name: name = self.repo.config["core"].get("remote") if name: - return self._init_remote(name) + return self._init_odb(name) if bool(self.repo.config["remote"]): error_msg = ( @@ -52,35 +54,41 @@ class DataCloud: raise NoRemoteError(error_msg) - def _init_remote(self, name): - from dvc.remote import get_remote + def _init_odb(self, name): + from dvc.fs import get_cloud_fs + from dvc.objects.db import get_odb - return get_remote(self.repo, name=name) + cls, config, path_info = get_cloud_fs(self.repo, name=name) + config["tmp_dir"] = self.repo.tmp_dir + return get_odb(cls(**config), path_info, **config) def push( self, objs: Iterable["HashFile"], jobs: Optional[int] = None, remote: Optional[str] = None, - show_checksums: bool = False, + odb: Optional["ObjectDB"] = None, ): """Push data items in a cloud-agnostic way. Args: objs: objects to push to the cloud. jobs: number of jobs that can be running simultaneously. - remote: optional remote to push to. + remote: optional name of remote to push to. By default remote from core.remote config option is used. - show_checksums: show checksums instead of file names in - information messages. + odb: optional ODB to push to. Overrides remote. """ - remote_obj = self.get_remote(remote, "push") + from dvc.objects.transfer import transfer - return remote_obj.push( + if not odb: + odb = self.get_remote_odb(remote, "push") + return transfer( self.repo.odb.local, + odb, objs, jobs=jobs, - show_checksums=show_checksums, + dest_index=get_index(odb), + cache_odb=self.repo.odb.local, ) def pull( @@ -88,25 +96,28 @@ class DataCloud: objs: Iterable["HashFile"], jobs: Optional[int] = None, remote: Optional[str] = None, - show_checksums: bool = False, + odb: Optional["ObjectDB"] = None, ): """Pull data items in a cloud-agnostic way. Args: objs: objects to pull from the cloud. jobs: number of jobs that can be running simultaneously. - remote: optional remote to pull from. + remote: optional name of remote to pull from. By default remote from core.remote config option is used. - show_checksums: show checksums instead of file names in - information messages. + odb: optional ODB to pull from. Overrides remote. """ - remote_obj = self.get_remote(remote, "pull") + from dvc.objects.transfer import transfer - return remote_obj.pull( + if not odb: + odb = self.get_remote_odb(remote, "pull") + return transfer( + odb, self.repo.odb.local, objs, jobs=jobs, - show_checksums=show_checksums, + src_index=get_index(odb), + cache_odb=self.repo.odb.local, ) def status( @@ -114,7 +125,7 @@ class DataCloud: objs: Iterable["HashFile"], jobs: Optional[int] = None, remote: Optional[str] = None, - show_checksums: bool = False, + odb: Optional["ObjectDB"] = None, log_missing: bool = True, ): """Check status of data items in a cloud-agnostic way. @@ -125,20 +136,24 @@ class DataCloud: remote: optional remote to compare cache to. By default remote from core.remote config option is used. - show_checksums: show checksums instead of file names in - information messages. + odb: optional ODB to check status from. Overrides remote. log_missing: log warning messages if file doesn't exist neither in cache, neither in cloud. """ - remote_obj = self.get_remote(remote, "status") - return remote_obj.status( + from dvc.objects.status import compare_status + + if not odb: + odb = self.get_remote_odb(remote, "status") + return compare_status( self.repo.odb.local, + odb, objs, jobs=jobs, - show_checksums=show_checksums, log_missing=log_missing, + dest_index=get_index(odb), + cache_odb=self.repo.odb.local, ) def get_url_for(self, remote, checksum): - remote_obj = self.get_remote(remote) - return str(remote_obj.odb.hash_to_path_info(checksum)) + remote_odb = self.get_remote_odb(remote) + return str(remote_odb.hash_to_path_info(checksum)) diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index d7f61a4d5..234749900 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -63,16 +63,9 @@ class RepoDependency(Dependency): def download(self, to, jobs=None): from dvc.checkout import checkout - from dvc.fs.memory import MemoryFileSystem - from dvc.objects import save - from dvc.repo.fetch import fetch_from_odb for odb, objs in self.get_used_objs().items(): - if isinstance(odb.fs, MemoryFileSystem): - for obj in objs: - save(self.repo.odb.local, obj, jobs=jobs) - else: - fetch_from_odb(self.repo, odb, objs, jobs=jobs) + self.repo.cloud.pull(objs, jobs=jobs, odb=odb) obj = self.get_obj() checkout( @@ -129,7 +122,7 @@ class RepoDependency(Dependency): recursive=True, ).items(): if odb is None: - odb = repo.cloud.get_remote().odb + odb = repo.cloud.get_remote_odb() odb.read_only = True self._check_circular_import(objs) used_objs[odb].update(objs) diff --git a/dvc/exceptions.py b/dvc/exceptions.py index a3e91d7e0..6d5db86f4 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -228,18 +228,21 @@ class GitHookAlreadyExistsError(DvcException): ) -class DownloadError(DvcException): +class FileTransferError(DvcException): + _METHOD = "transfer" + def __init__(self, amount): self.amount = amount - super().__init__(f"{amount} files failed to download") + super().__init__(f"{amount} files failed to {self._METHOD}") -class UploadError(DvcException): - def __init__(self, amount): - self.amount = amount +class DownloadError(FileTransferError): + _METHOD = "download" - super().__init__(f"{amount} files failed to upload") + +class UploadError(FileTransferError): + _METHOD = "upload" class CheckoutError(DvcException): diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py index 10e27ec5e..d6e2392b1 100644 --- a/dvc/fs/dvc.py +++ b/dvc/fs/dvc.py @@ -80,15 +80,15 @@ class DvcFileSystem(BaseFileSystem): # pylint:disable=abstract-method from dvc.config import NoRemoteError try: - remote_obj = self.repo.cloud.get_remote(remote) + remote_odb = self.repo.cloud.get_remote_odb(remote) except NoRemoteError as exc: raise FileNotFoundError from exc if out.is_dir_checksum: checksum = self._get_granular_hash(path, out).value else: checksum = out.hash_info.value - remote_info = remote_obj.odb.hash_to_path_info(checksum) - return remote_obj.fs.open( + remote_info = remote_odb.hash_to_path_info(checksum) + return remote_odb.fs.open( remote_info, mode=mode, encoding=encoding ) diff --git a/dvc/objects/db/__init__.py b/dvc/objects/db/__init__.py index 153495614..ece8e45b4 100644 --- a/dvc/objects/db/__init__.py +++ b/dvc/objects/db/__init__.py @@ -1,5 +1,10 @@ +from typing import TYPE_CHECKING + from dvc.scheme import Schemes +if TYPE_CHECKING: + from .index import ObjectDBIndexBase + def get_odb(fs, path_info, **config): from .base import ObjectDB @@ -34,6 +39,19 @@ def _get_odb(repo, settings): return get_odb(cls(**config), path_info, state=repo.state, **config) +def get_index(odb) -> "ObjectDBIndexBase": + import hashlib + + from .index import ObjectDBIndex, ObjectDBIndexNoop + + cls = ObjectDBIndex if odb.tmp_dir else ObjectDBIndexNoop + return cls( + odb.tmp_dir, + hashlib.sha256(odb.path_info.url.encode("utf-8")).hexdigest(), + odb.fs.CHECKSUM_DIR_SUFFIX, + ) + + class ODBManager: CACHE_DIR = "cache" CLOUD_SCHEMES = [ diff --git a/dvc/objects/db/base.py b/dvc/objects/db/base.py index 9bbfea0c6..21c7f2004 100644 --- a/dvc/objects/db/base.py +++ b/dvc/objects/db/base.py @@ -81,12 +81,16 @@ class ObjectDB: fs: "BaseFileSystem", hash_info: "HashInfo", move: bool = True, + verify: Optional[bool] = None, **kwargs, ): if self.read_only: raise ObjectDBPermissionError("Cannot add to read-only ODB") + + if verify is None: + verify = self.verify try: - self.check(hash_info, check_hash=self.verify) + self.check(hash_info, check_hash=verify) return except (ObjectFormatError, FileNotFoundError): pass @@ -120,8 +124,13 @@ class ObjectDB: else: raise - self.protect(cache_info) - self.state.save(cache_info, self.fs, hash_info) + try: + if verify: + self.check(hash_info, check_hash=True) + self.protect(cache_info) + self.state.save(cache_info, self.fs, hash_info) + except (ObjectFormatError, FileNotFoundError): + pass callback = kwargs.get("download_callback") if callback: diff --git a/dvc/objects/db/index.py b/dvc/objects/db/index.py new file mode 100644 index 000000000..e92812352 --- /dev/null +++ b/dvc/objects/db/index.py @@ -0,0 +1,137 @@ +import logging +import os +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Iterable, Optional, Set + +from ..errors import ObjectDBError + +if TYPE_CHECKING: + from dvc.types import StrPath + +logger = logging.getLogger(__name__) + + +class ObjectDBIndexBase(ABC): + @abstractmethod + def __init__( + self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None + ): + pass + + @abstractmethod + def __iter__(self): + pass + + @abstractmethod + def __contains__(self, hash_): + pass + + def hashes(self): + return iter(self) + + @abstractmethod + def dir_hashes(self): + pass + + @abstractmethod + def clear(self): + pass + + @abstractmethod + def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): + pass + + @abstractmethod + def intersection(self, hashes: Set[str]): + pass + + +class ObjectDBIndexNoop(ObjectDBIndexBase): + """No-op class for ODBs which are not indexed.""" + + def __init__( + self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None + ): # pylint: disable=super-init-not-called + pass + + def __iter__(self): + return iter([]) + + def __contains__(self, hash_): + return False + + def dir_hashes(self): + return [] + + def clear(self): + pass + + def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): + pass + + def intersection(self, hashes: Set[str]): + return [] + + +class ObjectDBIndex(ObjectDBIndexBase): + """Class for indexing hashes in an ODB.""" + + INDEX_SUFFIX = ".idx" + INDEX_DIR = "index" + + def __init__( + self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None + ): # pylint: disable=super-init-not-called + from diskcache import Index + + from dvc.fs.local import LocalFileSystem + from dvc.utils.fs import makedirs + + self.index_dir = os.path.join(tmp_dir, self.INDEX_DIR, name) + makedirs(self.index_dir, exist_ok=True) + self.fs = LocalFileSystem() + self.index = Index(self.index_dir) + + if not dir_suffix: + dir_suffix = self.fs.CHECKSUM_DIR_SUFFIX + self.dir_suffix = dir_suffix + + def __iter__(self): + return iter(self.index) + + def __contains__(self, hash_): + return hash_ in self.index + + def dir_hashes(self): + """Iterate over .dir hashes stored in the index.""" + yield from (hash_ for hash_, is_dir in self.index.items() if is_dir) + + def is_dir_hash(self, hash_: str): + return hash_.endswith(self.dir_suffix) + + def clear(self): + """Clear this index (to force re-indexing later).""" + from diskcache import Timeout + + try: + self.index.clear() + except Timeout as exc: + raise ObjectDBError("Failed to clear ODB index") from exc + + def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): + """Update this index, adding the specified hashes.""" + from diskcache import Timeout + + try: + with self.index.transact(): + for hash_ in dir_hashes: + self.index[hash_] = True + with self.index.transact(): + for hash_ in file_hashes: + self.index[hash_] = False + except Timeout as exc: + raise ObjectDBError("Failed to update ODB index") from exc + + def intersection(self, hashes: Set[str]): + """Iterate over values from `hashes` which exist in the index.""" + yield from hashes.intersection(self.index.keys()) diff --git a/dvc/remote/slow_link_detection.py b/dvc/objects/db/slow_link_detection.py similarity index 79% rename from dvc/remote/slow_link_detection.py rename to dvc/objects/db/slow_link_detection.py index 51208265d..1786bae6d 100644 --- a/dvc/remote/slow_link_detection.py +++ b/dvc/objects/db/slow_link_detection.py @@ -25,15 +25,15 @@ this.message = ( def slow_link_guard(f): @wraps(f) - def wrapper(remote, *args, **kwargs): + def wrapper(odb, *args, **kwargs): if this.already_displayed: - return f(remote, *args, **kwargs) + return f(odb, *args, **kwargs) - if not remote.slow_link_warning or remote.cache_types: - return f(remote, *args, **kwargs) + if not odb.slow_link_warning or odb.cache_types: + return f(odb, *args, **kwargs) start = time.time() - result = f(remote, *args, **kwargs) + result = f(odb, *args, **kwargs) delta = time.time() - start if delta >= this.timeout_seconds: diff --git a/dvc/objects/stage.py b/dvc/objects/stage.py index fdade1a12..ea0a62254 100644 --- a/dvc/objects/stage.py +++ b/dvc/objects/stage.py @@ -195,6 +195,11 @@ def get_staging(odb: Optional["ObjectDB"] = None) -> "ObjectDB": return get_odb(fs, path_info, **config) +def is_memfs_staging(odb: "ObjectDB"): + staging = get_staging() + return odb.fs == staging.fs and odb.path_info == staging.path_info + + def _load_from_state(odb, staging, path_info, fs, name): from . import load from .errors import ObjectFormatError diff --git a/dvc/objects/status.py b/dvc/objects/status.py new file mode 100644 index 000000000..ca2fc4cd3 --- /dev/null +++ b/dvc/objects/status.py @@ -0,0 +1,185 @@ +import logging +from typing import TYPE_CHECKING, Dict, Iterable, NamedTuple, Optional, Set + +from dvc.hash_info import HashInfo + +from .tree import Tree + +if TYPE_CHECKING: + from .db.base import ObjectDB + from .db.index import ObjectDBIndexBase + from .file import HashFile + +logger = logging.getLogger(__name__) + + +class StatusResult(NamedTuple): + exists: Set["HashFile"] + missing: Set["HashFile"] + + +class CompareStatusResult(NamedTuple): + ok: Set["HashFile"] + missing: Set["HashFile"] + new: Set["HashFile"] + deleted: Set["HashFile"] + + +def _indexed_dir_hashes(odb, index, hash_infos, name, cache_odb): + from . import load + + # Validate our index by verifying all indexed .dir hashes + # still exist on the remote + dir_hashes = {hash_info.value for hash_info in hash_infos} + indexed_dirs = set(index.dir_hashes()) + indexed_dir_exists = set() + if indexed_dirs: + indexed_dir_exists.update(odb.list_hashes_exists(indexed_dirs)) + missing_dirs = indexed_dirs.difference(indexed_dir_exists) + if missing_dirs: + logger.debug( + "Remote cache missing indexed .dir hashes '%s', " + "clearing remote index", + ", ".join(missing_dirs), + ) + index.clear() + + # Check if non-indexed (new) dir hashes exist on remote + dir_exists = dir_hashes.intersection(indexed_dir_exists) + dir_exists.update(odb.list_hashes_exists(dir_hashes - dir_exists)) + + # If .dir hash exists in the ODB, assume directory contents + # also exists + for dir_hash in dir_exists: + try: + tree = load(cache_odb, HashInfo(name, dir_hash)) + except FileNotFoundError: + continue + file_hashes = [entry.hash_info.value for _, entry in tree] + if dir_hash not in index: + logger.debug( + "Indexing new .dir '%s' with '%s' nested files", + dir_hash, + len(file_hashes), + ) + index.update([dir_hash], file_hashes) + yield from file_hashes + yield tree.hash_info.value + + +def _status_staging(objs: Iterable["HashFile"]) -> "StatusResult": + exists: Set["HashFile"] = set() + for obj in objs: + if isinstance(obj, Tree): + exists.update(entry for _, entry in obj) + exists.add(obj) + return StatusResult(exists, set()) + + +def status( + odb: "ObjectDB", + objs: Iterable["HashFile"], + name: Optional[str] = None, + index: Optional["ObjectDBIndexBase"] = None, + cache_odb: Optional["ObjectDB"] = None, + **kwargs, +) -> "StatusResult": + """Return status of whether or not the specified objects exist odb. + + If cache_odb is set, trees will be loaded from cache_odb instead of odb + when needed. + + Status is returned as a tuple of: + exists: objs that exist in odb + missing: objs that do not exist in ODB + """ + from .stage import is_memfs_staging + + logger.debug("Preparing to collect status from '%s'", odb.path_info) + if not name: + name = odb.fs.PARAM_CHECKSUM + + if is_memfs_staging(odb): + # assume memfs staged objects already exist + return _status_staging(objs) + + hash_objs: Dict[str, "HashFile"] = {} + hashes: Set[str] = set() + dir_infos: Set["HashInfo"] = set() + exists: Set[str] = set() + for obj in objs: + assert obj.hash_info and obj.hash_info.value + if isinstance(obj, Tree): + for _, entry in obj: + assert entry.hash_info and entry.hash_info.value + hash_objs[entry.hash_info.value] = entry + hashes.add(entry.hash_info.value) + if index: + dir_infos.add(obj.hash_info) + hashes.add(obj.hash_info.value) + hash_objs[obj.hash_info.value] = obj + + logger.debug("Collecting status from '%s'", odb.path_info) + if index and hashes: + if dir_infos: + if cache_odb is None: + cache_odb = odb + exists = hashes.intersection( + _indexed_dir_hashes(odb, index, dir_infos, name, cache_odb) + ) + hashes.difference_update(exists) + if hashes: + exists.update(index.intersection(hashes)) + hashes.difference_update(exists) + + if hashes: + exists.update( + odb.hashes_exist(hashes, name=str(odb.path_info), **kwargs) + ) + return StatusResult( + {hash_objs[hash_] for hash_ in exists}, + {hash_objs[hash_] for hash_ in (hashes - exists)}, + ) + + +def compare_status( + src: "ObjectDB", + dest: "ObjectDB", + objs: Iterable["HashFile"], + log_missing: bool = True, + check_deleted: bool = True, + src_index: Optional["ObjectDBIndexBase"] = None, + dest_index: Optional["ObjectDBIndexBase"] = None, + **kwargs, +) -> "CompareStatusResult": + """Compare status for the specified objects between two ODBs. + + Status is returned as a tuple of: + ok: hashes that exist in both src and dest + missing: hashes that do not exist in neither src nor dest + new: hashes that only exist in src + deleted: hashes that only exist in dest + """ + dest_exists, dest_missing = status(dest, objs, index=dest_index, **kwargs) + # for transfer operations we can skip src status check when all objects + # already exist in dest + if dest_missing or check_deleted: + src_exists, src_missing = status(src, objs, index=src_index, **kwargs) + else: + src_exists = dest_exists + src_missing = set() + result = CompareStatusResult( + src_exists & dest_exists, + src_missing & dest_missing, + src_exists - dest_exists, + dest_exists - src_exists, + ) + if log_missing and result.missing: + missing_desc = "\n".join( + f"name: {obj.name}, {obj.hash_info}" for obj in result.missing + ) + logger.warning( + "Some of the cache files do not exist neither locally " + f"nor on remote. Missing cache files:\n{missing_desc}" + ) + return result diff --git a/dvc/objects/transfer.py b/dvc/objects/transfer.py new file mode 100644 index 000000000..bc10d893e --- /dev/null +++ b/dvc/objects/transfer.py @@ -0,0 +1,218 @@ +import errno +import itertools +import logging +from concurrent import futures +from concurrent.futures import ThreadPoolExecutor +from functools import partial, wraps +from typing import TYPE_CHECKING, Iterable, Optional, Set + +from dvc.progress import Tqdm + +from .file import HashFile +from .tree import Tree + +if TYPE_CHECKING: + from .db.base import ObjectDB + from .db.index import ObjectDBIndexBase + +logger = logging.getLogger(__name__) + + +def _log_exceptions(func): + @wraps(func) + def wrapper(odb, obj, *args, **kwargs): + try: + func(odb, obj, *args, **kwargs) + return 0 + except Exception as exc: # pylint: disable=broad-except + # NOTE: this means we ran out of file descriptors and there is no + # reason to try to proceed, as we will hit this error anyways. + # pylint: disable=no-member + if isinstance(exc, OSError) and exc.errno == errno.EMFILE: + raise + + logger.exception( + "failed to transfer '%s' to '%s'", + obj.path_info, + odb.get(obj.hash_info).path_info, + ) + return 1 + + return wrapper + + +def _transfer(src, dest, dir_objs, file_objs, missing, jobs, **kwargs): + from . import save + from .stage import is_memfs_staging + + is_staged = is_memfs_staging(src) + func = _log_exceptions(save) + total = len(dir_objs) + len(file_objs) + if total == 0: + return 0 + with Tqdm(total=total, unit="file", desc="Transferring") as pbar: + func = pbar.wrap_fn(func) + with ThreadPoolExecutor(max_workers=jobs) as executor: + processor = partial( + _create_tasks, executor, jobs, func, src, dest, is_staged + ) + processor.save_func = func + _do_transfer( + src, + dest, + dir_objs, + file_objs, + {obj.hash_info for obj in missing}, + processor, + **kwargs, + ) + return total + + +def _create_tasks(executor, jobs, func, src, dest, is_staged, objs): + fails = 0 + obj_iter = iter(objs) + + def create_taskset(amount): + return { + executor.submit( + func, + dest, + _raw_obj(src, obj, is_staged), + move=False, + verify=src.verify, + ) + for obj in itertools.islice(obj_iter, amount) + } + + tasks = create_taskset(jobs * 5) + while tasks: + done, tasks = futures.wait(tasks, return_when=futures.FIRST_COMPLETED) + fails += sum(task.result() for task in done) + tasks.update(create_taskset(len(done))) + return fails + + +def _raw_obj(odb, obj, is_staged=False): + if is_staged: + return HashFile(obj.path_info, obj.fs, obj.hash_info) + return odb.get(obj.hash_info) + + +def _do_transfer( + src, + dest, + dir_objs, + file_objs, + missing_hashes, + processor, + src_index: Optional["ObjectDBIndexBase"] = None, + dest_index: Optional["ObjectDBIndexBase"] = None, + **kwargs, +): + from dvc.exceptions import FileTransferError + + total_fails = 0 + succeeded_dir_objs = [] + all_file_objs = set(file_objs) + + for dir_obj in dir_objs: + bound_file_objs = set() + directory_hashes = {entry.hash_info for _, entry in dir_obj} + + for file_obj in all_file_objs.copy(): + if file_obj.hash_info in directory_hashes: + bound_file_objs.add(file_obj) + all_file_objs.remove(file_obj) + + dir_fails = processor(bound_file_objs) + if dir_fails: + logger.debug( + "failed to upload full contents of '%s', " + "aborting .dir file upload", + dir_obj.name, + ) + logger.error( + "failed to upload '%s' to '%s'", + src.get(dir_obj.hash_info).path_info, + dest.get(dir_obj.hash_info).path_info, + ) + total_fails += dir_fails + 1 + elif directory_hashes.intersection(missing_hashes): + # if for some reason a file contained in this dir is + # missing both locally and in the remote, we want to + # push whatever file content we have, but should not + # push .dir file + logger.debug( + "directory '%s' contains missing files," + "skipping .dir file upload", + dir_obj.name, + ) + else: + is_dir_failed = processor.save_func( + dest, src.get(dir_obj.hash_info), move=False + ) + total_fails += is_dir_failed + if not is_dir_failed: + succeeded_dir_objs.append(dir_obj) + + # insert the rest + total_fails += processor(all_file_objs) + if total_fails: + if src_index: + src_index.clear() + raise FileTransferError(total_fails) + + # index successfully pushed dirs + if dest_index: + for dir_obj in succeeded_dir_objs: + file_hashes = {entry.hash_info.value for _, entry in dir_obj} + logger.debug( + "Indexing pushed dir '%s' with '%s' nested files", + dir_obj.hash_info, + len(file_hashes), + ) + dest_index.update([dir_obj.hash_info.value], file_hashes) + + +def transfer( + src: "ObjectDB", + dest: "ObjectDB", + objs: Iterable["HashFile"], + jobs: Optional[int] = None, + **kwargs, +) -> int: + """Transfer (copy) the specified objects from one ODB to another. + + Returns the number of successfully transferred objects + """ + from .status import compare_status + + logger.debug( + "Preparing to transfer data from '%s' to '%s'", + src.path_info, + dest.path_info, + ) + status = compare_status(src, dest, objs, check_deleted=False, **kwargs) + if not status.new: + return 0 + + files: Set["HashFile"] = set() + dirs: Set["HashFile"] = set() + for obj in status.new: + if isinstance(obj, Tree): + dirs.add(obj) + else: + files.add(obj) + if jobs is None: + jobs = dest.fs.jobs + + return _transfer( + src, + dest, + dirs, + files, + status.missing, + jobs, + **kwargs, + ) diff --git a/dvc/objects/tree.py b/dvc/objects/tree.py index 1e3d2e840..cc2c4dd53 100644 --- a/dvc/objects/tree.py +++ b/dvc/objects/tree.py @@ -1,7 +1,7 @@ import json import logging import posixpath -from typing import Optional, Tuple +from typing import Dict, Optional, Tuple from funcy import cached_property @@ -17,7 +17,7 @@ class Tree(HashFile): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - self._dict = {} + self._dict: Dict[Tuple[str], "HashFile"] = {} @cached_property def trie(self): diff --git a/dvc/output.py b/dvc/output.py index f194a4940..081df0179 100644 --- a/dvc/output.py +++ b/dvc/output.py @@ -825,7 +825,7 @@ class Output: try: objects.check(self.odb, obj) except FileNotFoundError: - self.repo.cloud.pull([obj], show_checksums=False, **kwargs) + self.repo.cloud.pull([obj], **kwargs) try: self.obj = objects.load(self.odb, self.hash_info) diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py deleted file mode 100644 index 3e695c86b..000000000 --- a/dvc/remote/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -from ..fs import get_cloud_fs -from .base import Remote -from .local import LocalRemote - - -def get_remote(repo, **kwargs): - cls, config, path_info = get_cloud_fs(repo, **kwargs) - fs = cls(**config) - if fs.scheme == "local": - return LocalRemote(fs, path_info, repo.tmp_dir, **config) - return Remote(fs, path_info, repo.tmp_dir, **config) diff --git a/dvc/remote/base.py b/dvc/remote/base.py deleted file mode 100644 index 94b70bead..000000000 --- a/dvc/remote/base.py +++ /dev/null @@ -1,555 +0,0 @@ -import errno -import hashlib -import itertools -import logging -from concurrent import futures -from concurrent.futures import ThreadPoolExecutor -from functools import partial, wraps - -from dvc.exceptions import DownloadError, UploadError -from dvc.hash_info import HashInfo -from dvc.objects.tree import Tree - -from ..progress import Tqdm -from .index import RemoteIndex - -logger = logging.getLogger(__name__) - -STATUS_OK = 1 -STATUS_MISSING = 2 -STATUS_NEW = 3 -STATUS_DELETED = 4 - -STATUS_MAP = { - # (local_exists, remote_exists) - (True, True): STATUS_OK, - (False, False): STATUS_MISSING, - (True, False): STATUS_NEW, - (False, True): STATUS_DELETED, -} - - -def _log_exceptions(func, operation): - @wraps(func) - def wrapper(from_info, to_info, *args, **kwargs): - try: - func(from_info, to_info, *args, **kwargs) - return 0 - except Exception as exc: # pylint: disable=broad-except - # NOTE: this means we ran out of file descriptors and there is no - # reason to try to proceed, as we will hit this error anyways. - # pylint: disable=no-member - if isinstance(exc, OSError) and exc.errno == errno.EMFILE: - raise - - logger.exception( - "failed to %s '%s' to '%s'", operation, from_info, to_info - ) - return 1 - - return wrapper - - -def index_locked(f): - @wraps(f) - def wrapper(obj, *args, **kwargs): - with obj.index: - return f(obj, *args, **kwargs) - - return wrapper - - -class Remote: - """Cloud remote class. - - Provides methods for indexing and garbage collecting fss which contain - DVC remotes. - """ - - INDEX_CLS = RemoteIndex - - def __init__(self, fs, path_info, tmp_dir, **config): - from dvc.objects.db import get_odb - - self.fs = fs - config["tmp_dir"] = tmp_dir - self.odb = get_odb(self.fs, path_info, **config) - - url = getattr(path_info, "fspath", path_info.url) - index_name = hashlib.sha256(url.encode("utf-8")).hexdigest() - self.index = self.INDEX_CLS( - tmp_dir, index_name, dir_suffix=self.fs.CHECKSUM_DIR_SUFFIX - ) - - def __repr__(self): - return "{class_name}: '{path_info}'".format( - class_name=type(self).__name__, - path_info=self.odb.path_info or "No path", - ) - - def __eq__(self, other): - return self.odb == other.odb - - def __hash__(self): - return hash(self.odb) - - @classmethod - def from_odb(cls, odb): - config = odb.config - return cls( - odb.fs, - odb.path_info, - config.pop("tmp_dir"), - **config, - ) - - @index_locked - def gc(self, *args, **kwargs): - removed = self.odb.gc(*args, **kwargs) - - if removed: - self.index.clear() - - return removed - - @index_locked - def status( - self, - cache, - objs, - jobs=None, - show_checksums=False, - download=False, - log_missing=True, - ): - # Return flattened dict containing all status info - dir_status, file_status, _ = self._status( - cache, - objs, - jobs=jobs, - show_checksums=show_checksums, - download=download, - log_missing=log_missing, - ) - return dict(dir_status, **file_status) - - def hashes_exist(self, hashes, **kwargs): - hashes = set(hashes) - indexed_hashes = set(self.index.intersection(hashes)) - hashes -= indexed_hashes - indexed_hashes = list(indexed_hashes) - logger.debug(f"Matched '{len(indexed_hashes)}' indexed hashes") - if not hashes: - return indexed_hashes - - return indexed_hashes + self.odb.hashes_exist(list(hashes), **kwargs) - - def _status( # pylint: disable=unused-argument - self, - cache, - objs, - jobs=None, - show_checksums=False, - download=False, - log_missing=True, - ): - """Return a tuple of (dir_status_info, file_status_info, dir_contents). - - dir_status_info contains status for .dir files, file_status_info - contains status for all other files, and dir_contents is a dict of - {dir_hash: set(file_hash, ...)} which can be used to map - a .dir file to its file contents. - """ - - logger.debug(f"Preparing to collect status from {self.odb.path_info}") - md5s = set() - dir_objs = {} - for obj in objs: - md5s.add(obj.hash_info.value) - if isinstance(obj, Tree): - dir_objs[obj.hash_info.value] = obj - for _, entry_obj in obj: - md5s.add(entry_obj.hash_info.value) - - logger.debug("Collecting information from local cache...") - local_exists = frozenset( - cache.hashes_exist(md5s, jobs=jobs, name=cache.cache_dir) - ) - - # This is a performance optimization. We can safely assume that, - # if the resources that we want to fetch are already cached, - # there's no need to check the remote storage for the existence of - # those files. - if download and local_exists == md5s: - remote_exists = local_exists - else: - logger.debug("Collecting information from remote cache...") - remote_exists = set() - if dir_objs: - remote_exists.update(self._indexed_dir_hashes(dir_objs)) - md5s.difference_update(remote_exists) - if md5s: - remote_exists.update( - self.hashes_exist( - md5s, jobs=jobs, name=str(self.odb.path_info) - ) - ) - return self._make_status( - objs, - show_checksums, - local_exists, - remote_exists, - log_missing, - ) - - def _make_status( - self, - objs, - show_checksums, - local_exists, - remote_exists, - log_missing, - ): - def make_names(obj): - if show_checksums or obj.name is None: - name = obj.hash_info.value - else: - name = obj.name - return {"name": name} - - dir_status = {} - file_status = {} - dir_contents = {} - for obj in objs: - - hash_ = obj.hash_info.value - if isinstance(obj, Tree): - dir_status[hash_] = make_names(obj) - dir_contents[hash_] = set() - for _, child_obj in obj: - child_hash = child_obj.hash_info.value - file_status[child_hash] = make_names(child_obj) - dir_contents[hash_].add(child_hash) - else: - file_status[hash_] = make_names(obj) - - self._fill_statuses(dir_status, local_exists, remote_exists) - self._fill_statuses(file_status, local_exists, remote_exists) - - if log_missing: - self._log_missing_caches(dict(dir_status, **file_status)) - - return dir_status, file_status, dir_contents - - def _indexed_dir_hashes(self, dir_objs): - # Validate our index by verifying all indexed .dir hashes - # still exist on the remote - dir_md5s = set(dir_objs.keys()) - indexed_dirs = set(self.index.dir_hashes()) - indexed_dir_exists = set() - if indexed_dirs: - indexed_dir_exists.update( - self.odb.list_hashes_exists(indexed_dirs) - ) - missing_dirs = indexed_dirs.difference(indexed_dir_exists) - if missing_dirs: - logger.debug( - "Remote cache missing indexed .dir hashes '{}', " - "clearing remote index".format(", ".join(missing_dirs)) - ) - self.index.clear() - - # Check if non-indexed (new) dir hashes exist on remote - dir_exists = dir_md5s.intersection(indexed_dir_exists) - dir_exists.update(self.odb.list_hashes_exists(dir_md5s - dir_exists)) - - # If .dir hash exists on the remote, assume directory contents - # still exists on the remote - for dir_hash in dir_exists: - file_hashes = [ - obj.hash_info.value for _, obj in dir_objs[dir_hash] - ] - if dir_hash not in self.index: - logger.debug( - "Indexing new .dir '{}' with '{}' nested files".format( - dir_hash, len(file_hashes) - ) - ) - self.index.update([dir_hash], file_hashes) - yield dir_hash - yield from file_hashes - - @staticmethod - def _fill_statuses(hash_info_dir, local_exists, remote_exists): - # Using sets because they are way faster for lookups - local = set(local_exists) - remote = set(remote_exists) - - for md5, info in hash_info_dir.items(): - status = STATUS_MAP[(md5 in local, md5 in remote)] - info["status"] = status - - def _get_plans(self, cache_obj, download, status_info, status): - cache = [] - path_infos = [] - names = [] - hashes = [] - missing = [] - for md5, info in Tqdm( - status_info.items(), desc="Analysing status", unit="file" - ): - if info["status"] == status: - cache.append(cache_obj.hash_to_path_info(md5)) - path_infos.append(self.odb.hash_to_path_info(md5)) - names.append(info["name"]) - hashes.append(md5) - elif info["status"] == STATUS_MISSING: - missing.append(md5) - - if download: - to_infos = cache - from_infos = path_infos - else: - to_infos = path_infos - from_infos = cache - - return (from_infos, to_infos, names, hashes), missing - - def _process( - self, - cache, - objs, - jobs=None, - show_checksums=False, - download=False, - ): - logger.debug( - "Preparing to {} '{}'".format( - "download data from" if download else "upload data to", - self.odb.path_info, - ) - ) - - if download: - func = _log_exceptions(self.fs.download_file, "download") - status = STATUS_DELETED - desc = "Downloading" - else: - func = _log_exceptions(self.fs.upload, "upload") - status = STATUS_NEW - desc = "Uploading" - - if jobs is None: - jobs = self.fs.jobs - - dir_status, file_status, dir_contents = self._status( - cache, - objs, - jobs=jobs, - show_checksums=show_checksums, - download=download, - ) - - dir_plans, _ = self._get_plans(cache, download, dir_status, status) - file_plans, missing_files = self._get_plans( - cache, download, file_status, status - ) - - total = len(dir_plans[0]) + len(file_plans[0]) - if total == 0: - return 0 - - with Tqdm(total=total, unit="file", desc=desc) as pbar: - func = pbar.wrap_fn(func) - self._process_plans( - download, - dir_plans, - file_plans, - dir_contents, - missing_files, - jobs, - func, - ) - - return total - - def _process_plans( - self, - download, - dir_plans, - file_plans, - dir_contents, - missing_files, - jobs, - func, - ): - with ThreadPoolExecutor(max_workers=jobs) as executor: - processor = partial(self._create_tasks, executor, jobs, func) - processor.transfer_func = func - if download: - self._download_plans(dir_plans, file_plans, processor) - else: - self._upload_plans( - dir_plans, - file_plans, - dir_contents, - missing_files, - processor, - ) - - def _create_tasks(self, executor, jobs, func, file_plans): - fails = 0 - file_plan_iterator = iter(file_plans) - - def create_taskset(amount): - return { - executor.submit(func, from_info, to_info, name) - for from_info, to_info, name, _ in itertools.islice( - file_plan_iterator, amount - ) - } - - tasks = create_taskset(jobs * 5) - while tasks: - done, tasks = futures.wait( - tasks, return_when=futures.FIRST_COMPLETED - ) - fails += sum(task.result() for task in done) - tasks.update(create_taskset(len(done))) - return fails - - def _download_plans(self, dir_plans, file_plans, processor): - plans = [*zip(*dir_plans), *zip(*file_plans)] - fails = processor(plans) - if fails: - self.index.clear() - raise DownloadError(fails) - - def _upload_plans( - self, dir_plans, file_plans, dir_contents, missing_files, processor - ): - total_fails = 0 - succeeded_dir_hashes = [] - all_file_plans = list(zip(*file_plans)) - for dir_from_info, dir_to_info, dir_name, dir_hash in zip(*dir_plans): - bound_file_plans = [] - directory_hashes = dir_contents[dir_hash] - - for file_plan in all_file_plans.copy(): - if file_plan[-1] in directory_hashes: - bound_file_plans.append(file_plan) - all_file_plans.remove(file_plan) - - dir_fails = processor(bound_file_plans) - if dir_fails: - logger.debug( - "failed to upload full contents of '{}', " - "aborting .dir file upload".format(dir_name) - ) - logger.error( - f"failed to upload '{dir_from_info}'" - f" to '{dir_to_info}'" - ) - total_fails += dir_fails + 1 - elif directory_hashes.intersection(missing_files): - # if for some reason a file contained in this dir is - # missing both locally and in the remote, we want to - # push whatever file content we have, but should not - # push .dir file - logger.debug( - "directory '%s' contains missing files," - "skipping .dir file upload", - dir_name, - ) - else: - is_dir_failed = processor.transfer_func( - dir_from_info, dir_to_info, dir_name - ) - total_fails += is_dir_failed - if not is_dir_failed: - succeeded_dir_hashes.append(dir_hash) - - # insert the rest - total_fails += processor(all_file_plans) - if total_fails: - raise UploadError(total_fails) - - # index successfully pushed dirs - for dir_hash in succeeded_dir_hashes: - file_hashes = dir_contents[dir_hash] - logger.debug( - "Indexing pushed dir '{}' with " - "'{}' nested files".format(dir_hash, len(file_hashes)) - ) - self.index.update([dir_hash], file_hashes) - - @index_locked - def push(self, cache, objs, jobs=None, show_checksums=False): - ret = self._process( - cache, - objs, - jobs=jobs, - show_checksums=show_checksums, - download=False, - ) - - if self.fs.scheme == "local": - for obj in objs: - checksum = obj.hash_info.value - cache_file = self.odb.hash_to_path_info(checksum) - if self.fs.exists(cache_file): - hash_info = HashInfo(self.fs.PARAM_CHECKSUM, checksum) - self.odb.state.save(cache_file, self.fs, hash_info) - self.odb.protect(cache_file) - - return ret - - @index_locked - def pull(self, cache, objs, jobs=None, show_checksums=False): - ret = self._process( - cache, - objs, - jobs=jobs, - show_checksums=show_checksums, - download=True, - ) - - if not self.odb.verify: - - def save_state(obj): - checksum = obj.hash_info.value - cache_file = cache.hash_to_path_info(checksum) - if cache.fs.exists(cache_file): - # We can safely save here, as existing corrupted files - # will be removed upon status, while files corrupted - # during download will not be moved from tmp_file - # (see `BaseFileSystem.download()`) - hash_info = HashInfo(cache.fs.PARAM_CHECKSUM, checksum) - cache.state.save(cache_file, cache.fs, hash_info) - cache.protect(cache_file) - - for obj in objs: - save_state(obj) - if isinstance(obj, Tree): - for _, entry_obj in obj: - save_state(entry_obj) - - return ret - - @staticmethod - def _log_missing_caches(hash_info_dict): - missing_caches = [ - (md5, info) - for md5, info in hash_info_dict.items() - if info["status"] == STATUS_MISSING - ] - if missing_caches: - missing_desc = "\n".join( - "name: {}, md5: {}".format(info["name"], md5) - for md5, info in missing_caches - ) - msg = ( - "Some of the cache files do not exist neither locally " - "nor on remote. Missing cache files:\n{}".format(missing_desc) - ) - logger.warning(msg) diff --git a/dvc/remote/index.py b/dvc/remote/index.py deleted file mode 100644 index 33a8331f4..000000000 --- a/dvc/remote/index.py +++ /dev/null @@ -1,213 +0,0 @@ -import logging -import os -import sqlite3 -import threading - -from funcy import lchunks - -from dvc.state import _connect_sqlite -from dvc.utils.fs import makedirs - -logger = logging.getLogger(__name__) - - -class RemoteIndexNoop: - """No-op class for remotes which are not indexed (i.e. local).""" - - def __init__(self, *args, **kwargs): - pass - - def __enter__(self): - pass - - def __exit__(self, typ, value, tbck): - pass - - def __iter__(self): - return iter([]) - - def __contains__(self, hash_): - return False - - @staticmethod - def hashes(): - return [] - - @staticmethod - def dir_hashes(): - return [] - - def load(self): - pass - - def dump(self): - pass - - def clear(self): - pass - - def update(self, *args): - pass - - @staticmethod - def intersection(*args): - return [] - - -class RemoteIndex: - """Class for indexing remote hashes in a sqlite3 database. - - Args: - repo: repo for this remote index. - name: name for this index. Index db will be loaded from and saved to - ``.dvc/tmp/index/{name}.idx``. - dir_suffix: suffix used for naming directory hashes - """ - - INDEX_SUFFIX = ".idx" - VERSION = 1 - INDEX_TABLE = "remote_index" - INDEX_TABLE_LAYOUT = "checksum TEXT PRIMARY KEY, dir INTEGER NOT NULL" - INDEX_DIR = "index" - - def __init__(self, tmp_dir, name, dir_suffix=".dir"): - self.path = os.path.join( - tmp_dir, self.INDEX_DIR, f"{name}{self.INDEX_SUFFIX}" - ) - - self.dir_suffix = dir_suffix - self.database = None - self.cursor = None - self.modified = False - self.lock = threading.Lock() - - def __iter__(self): - cmd = f"SELECT checksum FROM {self.INDEX_TABLE}" - for (hash_,) in self._execute(cmd): - yield hash_ - - def __enter__(self): - self.lock.acquire() - self.load() - - def __exit__(self, typ, value, tbck): - self.dump() - self.lock.release() - - def __contains__(self, hash_): - cmd = "SELECT checksum FROM {} WHERE checksum = (?)".format( - self.INDEX_TABLE - ) - self._execute(cmd, (hash_,)) - return self.cursor.fetchone() is not None - - def hashes(self): - """Iterate over hashes stored in the index.""" - return iter(self) - - def dir_hashes(self): - """Iterate over .dir hashes stored in the index.""" - cmd = f"SELECT checksum FROM {self.INDEX_TABLE} WHERE dir = 1" - for (hash_,) in self._execute(cmd): - yield hash_ - - def is_dir_hash(self, hash_): - return hash_.endswith(self.dir_suffix) - - def _execute(self, cmd, parameters=()): - return self.cursor.execute(cmd, parameters) - - def _executemany(self, cmd, seq_of_parameters): - return self.cursor.executemany(cmd, seq_of_parameters) - - def _prepare_db(self, empty=False): - if not empty: - cmd = "PRAGMA user_version;" - self._execute(cmd) - ret = self.cursor.fetchall() - assert len(ret) == 1 - assert len(ret[0]) == 1 - assert isinstance(ret[0][0], int) - version = ret[0][0] - - if version != self.VERSION: - logger.error( - "Index file version '{}' will be reformatted " - "to the current version '{}'.".format( - version, self.VERSION - ) - ) - cmd = "DROP TABLE IF EXISTS {};" - self._execute(cmd.format(self.INDEX_TABLE)) - - cmd = "CREATE TABLE IF NOT EXISTS {} ({})" - self._execute(cmd.format(self.INDEX_TABLE, self.INDEX_TABLE_LAYOUT)) - - cmd = "PRAGMA user_version = {};" - self._execute(cmd.format(self.VERSION)) - - def load(self): - """(Re)load this index database.""" - retries = 1 - while True: - assert self.database is None - assert self.cursor is None - - empty = not os.path.isfile(self.path) - makedirs(os.path.dirname(self.path), exist_ok=True) - self.database = _connect_sqlite(self.path, {"nolock": 1}) - self.cursor = self.database.cursor() - - try: - self._prepare_db(empty=empty) - return - except sqlite3.DatabaseError: - self.cursor.close() - self.database.close() - self.database = None - self.cursor = None - if retries > 0: - os.unlink(self.path) - retries -= 1 - else: - raise - - def dump(self): - """Save this index database.""" - assert self.database is not None - - self.database.commit() - self.cursor.close() - self.database.close() - self.database = None - self.cursor = None - - def clear(self): - """Clear this index (to force re-indexing later). - - Changes to the index will not committed until dump() is called. - """ - cmd = f"DELETE FROM {self.INDEX_TABLE}" - self._execute(cmd) - - def update(self, dir_hashes, file_hashes): - """Update this index, adding the specified hashes. - - Changes to the index will not committed until dump() is called. - """ - cmd = "INSERT OR IGNORE INTO {} (checksum, dir) VALUES (?, ?)".format( - self.INDEX_TABLE - ) - self._executemany(cmd, ((hash_, True) for hash_ in dir_hashes)) - self._executemany(cmd, ((hash_, False) for hash_ in file_hashes)) - - def intersection(self, hashes): - """Iterate over values from `hashes` which exist in the index.""" - # sqlite has a compile time limit of 999, see: - # https://www.sqlite.org/c3ref/c_limit_attached.html#sqlitelimitvariablenumber - for chunk in lchunks(999, hashes): - cmd = "SELECT checksum FROM {} WHERE checksum IN ({})".format( - self.INDEX_TABLE, ",".join("?" for hash_ in chunk) - ) - for (hash_,) in self._execute(cmd, chunk): - yield hash_ diff --git a/dvc/remote/local.py b/dvc/remote/local.py deleted file mode 100644 index 903580d7b..000000000 --- a/dvc/remote/local.py +++ /dev/null @@ -1,6 +0,0 @@ -from .base import Remote -from .index import RemoteIndexNoop - - -class LocalRemote(Remote): - INDEX_CLS = RemoteIndexNoop # type: ignore[assignment] diff --git a/dvc/repo/add.py b/dvc/repo/add.py index 2d07d9439..18a05282f 100644 --- a/dvc/repo/add.py +++ b/dvc/repo/add.py @@ -180,8 +180,7 @@ def add( # noqa: C901 odb = None if to_remote: - remote = repo.cloud.get_remote(kwargs.get("remote"), "add") - odb = remote.odb + odb = repo.cloud.get_remote_odb(kwargs.get("remote"), "add") with warn_link_failures() as link_failures: for stage, source in zip(progress_iter(stages), sources): diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py index a0c3ee8f2..ff9d0f5cd 100644 --- a/dvc/repo/fetch.py +++ b/dvc/repo/fetch.py @@ -1,7 +1,7 @@ import logging from dvc.config import NoRemoteError -from dvc.exceptions import DownloadError +from dvc.exceptions import DownloadError, FileTransferError from ..scheme import Schemes from . import locked @@ -16,7 +16,6 @@ def fetch( jobs=None, remote=None, all_branches=False, - show_checksums=False, with_deps=False, all_tags=False, recursive=False, @@ -36,8 +35,6 @@ def fetch( config.NoRemoteError: thrown when downloading only local files and no remote is configured """ - from dvc.fs.memory import MemoryFileSystem - if isinstance(targets, str): targets = [targets] @@ -63,36 +60,14 @@ def fetch( except DownloadError as exc: failed += exc.amount - external = set() for odb, objs in used.items(): - if odb is None: - # objs contains naive objects to be pulled from specified remote - d, f = _fetch_naive_objs( - self, - objs, - jobs=jobs, - remote=remote, - show_checksums=show_checksums, - ) - downloaded += d - failed += f - elif isinstance(odb.fs, MemoryFileSystem): - # objs contains staged import objects which should be saved - # last (after all other objects have been pulled) - external.update(objs) - else: - d, f = fetch_from_odb( - self, - odb, - objs, - jobs=jobs, - show_checksums=show_checksums, - ) - downloaded += d - failed += f - - if external: - d, f = _fetch_external(self, external, jobs=jobs) + d, f = _fetch( + self, + objs, + jobs=jobs, + remote=remote, + odb=odb, + ) downloaded += d failed += f @@ -102,8 +77,7 @@ def fetch( return downloaded -def _fetch_naive_objs(repo, objs, **kwargs): - # objs contains naive objects to be pulled from specified remote +def _fetch(repo, objs, **kwargs): downloaded = 0 failed = 0 try: @@ -111,42 +85,6 @@ def _fetch_naive_objs(repo, objs, **kwargs): except NoRemoteError: if any(obj.fs.scheme == Schemes.LOCAL for obj in objs): raise - except DownloadError as exc: + except FileTransferError as exc: failed += exc.amount return downloaded, failed - - -def fetch_from_odb(repo, odb, objs, **kwargs): - from dvc.remote.base import Remote - - downloaded = 0 - failed = 0 - remote = Remote.from_odb(odb) - try: - downloaded += remote.pull( - repo.odb.local, - objs, - **kwargs, - ) - except DownloadError as exc: - failed += exc.amount - return downloaded, failed - - -def _fetch_external(repo, objs, **kwargs): - from dvc.objects import save - from dvc.objects.errors import ObjectError - - results = [] - failed = 0 - - def callback(result): - results.append(result) - - for obj in objs: - try: - save(repo.odb.local, obj, download_callback=callback, **kwargs) - except ObjectError: - failed += 1 - - return sum(results), failed diff --git a/dvc/repo/gc.py b/dvc/repo/gc.py index f98a37448..ceb0a5631 100644 --- a/dvc/repo/gc.py +++ b/dvc/repo/gc.py @@ -49,6 +49,7 @@ def gc( from contextlib import ExitStack + from dvc.objects.db import get_index from dvc.objects.stage import get_staging from dvc.repo import Repo @@ -89,10 +90,12 @@ def gc( if not cloud: return - remote = self.cloud.get_remote(remote, "gc -c") - removed = remote.gc( + odb = self.cloud.get_remote_odb(remote, "gc -c") + removed = odb.gc( {obj for obj in used_objs if obj.fs.scheme == Schemes.LOCAL}, jobs=jobs, ) - if not removed: + if removed: + get_index(odb).clear() + else: logger.info("No unused cache to remove from remote.") diff --git a/dvc/repo/imp_url.py b/dvc/repo/imp_url.py index 6a6ce6ce4..f372a3bf6 100644 --- a/dvc/repo/imp_url.py +++ b/dvc/repo/imp_url.py @@ -74,8 +74,8 @@ def imp_url( if no_exec: stage.ignore_outs() elif to_remote: - remote = self.cloud.get_remote(remote, "import-url") - stage.outs[0].transfer(url, odb=remote.odb, jobs=jobs) + remote_odb = self.cloud.get_remote_odb(remote, "import-url") + stage.outs[0].transfer(url, odb=remote_odb, jobs=jobs) stage.save_deps() stage.md5 = stage.compute_md5() else: diff --git a/dvc/repo/push.py b/dvc/repo/push.py index 63afbcfa9..341e0a720 100644 --- a/dvc/repo/push.py +++ b/dvc/repo/push.py @@ -1,3 +1,5 @@ +from dvc.exceptions import FileTransferError, UploadError + from ..utils import glob_targets from . import locked @@ -40,6 +42,10 @@ def push( pushed = len(used_run_cache) for odb, objs in used.items(): - if odb is None: - pushed += self.cloud.push(objs, jobs, remote=remote) + if odb and odb.read_only: + continue + try: + pushed += self.cloud.push(objs, jobs, remote=remote, odb=odb) + except FileTransferError as exc: + raise UploadError(exc.amount) return pushed diff --git a/dvc/repo/status.py b/dvc/repo/status.py index 3a304cc90..db7371756 100644 --- a/dvc/repo/status.py +++ b/dvc/repo/status.py @@ -76,8 +76,6 @@ def _cloud_status( { "bar": "deleted" } """ - import dvc.remote.base as cloud - used = self.used_objs( targets, all_branches=all_branches, @@ -98,19 +96,9 @@ def _cloud_status( status_info = self.cloud.status( objs, jobs, remote=remote, log_missing=False ) - for info in status_info.values(): - name = info["name"] - status_ = info["status"] - if status_ == cloud.STATUS_OK: - continue - - prefix_map = { - cloud.STATUS_DELETED: "deleted", - cloud.STATUS_NEW: "new", - cloud.STATUS_MISSING: "missing", - } - - ret[name] = prefix_map[status_] + for status_ in ("deleted", "new", "missing"): + for obj in getattr(status_info, status_, []): + ret[obj.name] = status_ return ret diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py index 3d3b1d153..9177a7b98 100644 --- a/dvc/stage/cache.py +++ b/dvc/stage/cache.py @@ -2,6 +2,7 @@ import logging import os import tempfile from contextlib import contextmanager +from typing import Optional from funcy import cached_property, first @@ -230,23 +231,23 @@ class StageCache: return ret - def push(self, remote): - from dvc.remote.base import _log_exceptions + def push(self, remote: Optional[str]): + from dvc.objects.transfer import _log_exceptions - remote = self.repo.cloud.get_remote(remote) + odb = self.repo.cloud.get_remote_odb(remote) return self._transfer( - _log_exceptions(remote.fs.upload, "upload"), + _log_exceptions(odb.fs.upload), self.repo.odb.local, - remote.odb, + odb, ) - def pull(self, remote): - from dvc.remote.base import _log_exceptions + def pull(self, remote: Optional[str]): + from dvc.objects.transfer import _log_exceptions - remote = self.repo.cloud.get_remote(remote) + odb = self.repo.cloud.get_remote_odb(remote) return self._transfer( - _log_exceptions(remote.fs.download, "download"), - remote.odb, + _log_exceptions(odb.fs.download), + odb, self.repo.odb.local, ) diff --git a/dvc/stage/imports.py b/dvc/stage/imports.py index 8878220ae..7f7893412 100644 --- a/dvc/stage/imports.py +++ b/dvc/stage/imports.py @@ -13,8 +13,8 @@ def _update_import_on_remote(stage, remote, jobs): ) url = stage.deps[0].def_path - remote = stage.repo.cloud.get_remote(remote, "update") - stage.outs[0].transfer(url, odb=remote.odb, jobs=jobs, update=True) + odb = stage.repo.cloud.get_remote_odb(remote, "update") + stage.outs[0].transfer(url, odb=odb, jobs=jobs, update=True) def update_import(stage, rev=None, to_remote=False, remote=None, jobs=None): diff --git a/dvc/state.py b/dvc/state.py index 10ab64d89..65dca8d57 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -2,35 +2,16 @@ import logging import os -import re from abc import ABC, abstractmethod -from urllib.parse import urlencode, urlunparse -from dvc.exceptions import DvcException from dvc.fs.local import LocalFileSystem from dvc.hash_info import HashInfo from dvc.utils import relpath from dvc.utils.fs import get_inode, get_mtime_and_size, remove -SQLITE_MAX_VARIABLES_NUMBER = 999 - logger = logging.getLogger(__name__) -class StateVersionTooNewError(DvcException): - """Thrown when DVC version is older than the state database version.""" - - def __init__(self, dvc_version, expected, actual): - super().__init__( - "you are using an old version '{dvc_version}' of DVC that is " - "using state file version '{expected}', which is not compatible " - "with the state file version '{actual}', that is used in this " - "repo. Please upgrade right now!".format( - dvc_version=dvc_version, expected=expected, actual=actual - ) - ) - - class StateBase(ABC): @abstractmethod def close(self): @@ -205,33 +186,3 @@ class State(StateBase): # pylint: disable=too-many-instance-attributes with self.links as ref: for path in unused: del ref[path] - - -def _connect_sqlite(filename, options): - # Connect by URI was added in Python 3.4 and sqlite 3.7.7, - # we ignore options, which should be fine unless repo is on old NFS/CIFS - import sqlite3 - - if sqlite3.sqlite_version_info < (3, 7, 7): - return sqlite3.connect(filename) - - uri = _build_sqlite_uri(filename, options) - return sqlite3.connect(uri, uri=True) - - -def _build_sqlite_uri(filename, options): - # In the doc mentioned below we only need to replace ? -> %3f and - # # -> %23, but, if present, we also need to replace % -> %25 first - # (happens when we are on a weird FS that shows urlencoded filenames - # instead of proper ones) to not confuse sqlite. - uri_path = filename.replace("%", "%25") - - # Convert filename to uri according to https://www.sqlite.org/uri.html, 3.1 - uri_path = uri_path.replace("?", "%3f").replace("#", "%23") - if os.name == "nt": - uri_path = uri_path.replace("\\", "/") - uri_path = re.sub(r"^([a-z]:)", "/\\1", uri_path, flags=re.I) - uri_path = re.sub(r"/+", "/", uri_path) - - # Empty netloc, params and fragment - return urlunparse(("file", "", uri_path, "", urlencode(options), "")) diff --git a/tests/func/remote/__init__.py b/tests/func/objects/__init__.py similarity index 100% rename from tests/func/remote/__init__.py rename to tests/func/objects/__init__.py diff --git a/tests/func/objects/db/__init__.py b/tests/func/objects/db/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/func/objects/db/test_index.py b/tests/func/objects/db/test_index.py new file mode 100644 index 000000000..2ab5dce26 --- /dev/null +++ b/tests/func/objects/db/test_index.py @@ -0,0 +1,88 @@ +import pytest + +from dvc.exceptions import DownloadError, UploadError +from dvc.fs.local import LocalFileSystem +from dvc.objects.db import get_index +from dvc.utils.fs import remove + + +@pytest.fixture +def index(dvc, local_remote, mocker): + odb = dvc.cloud.get_remote_odb("upstream") + return get_index(odb) + + +def test_indexed_on_status(tmp_dir, dvc, index): + foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0] + bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] + baz_hash = bar.obj.trie.get(("baz",)).hash_info + dvc.push() + index.clear() + + dvc.status(cloud=True) + assert {bar.hash_info.value, baz_hash.value} == set(index.hashes()) + assert [bar.hash_info.value] == list(index.dir_hashes()) + assert foo.hash_info.value not in index.hashes() + + +def test_indexed_on_push(tmp_dir, dvc, index): + foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0] + bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] + baz_hash = bar.obj.trie.get(("baz",)).hash_info + + dvc.push() + assert {bar.hash_info.value, baz_hash.value} == set(index.hashes()) + assert [bar.hash_info.value] == list(index.dir_hashes()) + assert foo.hash_info.value not in index.hashes() + + +def test_indexed_dir_missing(tmp_dir, dvc, index): + bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] + index.update([bar.hash_info.value], []) + dvc.status(cloud=True) + assert not list(index.hashes()) + + +def test_clear_on_gc(tmp_dir, dvc, index): + (foo,) = tmp_dir.dvc_gen({"dir": {"foo": "foo content"}}) + dvc.push() + dvc.remove(foo.relpath) + + assert list(index.hashes()) + dvc.gc(workspace=True, cloud=True) + assert not list(index.hashes()) + + +def test_clear_on_download_err(tmp_dir, dvc, index, mocker): + out = tmp_dir.dvc_gen({"dir": {"foo": "foo content"}})[0].outs[0] + dvc.push() + + for _, entry in out.obj: + remove(dvc.odb.local.get(entry.hash_info).path_info) + remove(out.path_info) + + assert list(index.hashes()) + + mocker.patch( + "dvc.fs.local.LocalFileSystem.upload_fobj", side_effect=Exception + ) + with pytest.raises(DownloadError): + dvc.pull() + assert not list(index.hashes()) + + +def test_partial_upload(tmp_dir, dvc, index, mocker): + tmp_dir.dvc_gen({"foo": "foo content"}) + tmp_dir.dvc_gen({"bar": {"baz": "baz content"}}) + + original = LocalFileSystem._upload + + def unreliable_upload(self, from_file, to_info, name=None, **kwargs): + if "baz" in name: + raise Exception("stop baz") + return original(self, from_file, to_info, name, **kwargs) + + mocker.patch.object(LocalFileSystem, "upload_fobj", unreliable_upload) + with pytest.raises(UploadError): + dvc.push() + assert not list(index.hashes()) diff --git a/tests/func/remote/test_index.py b/tests/func/remote/test_index.py deleted file mode 100644 index 104930090..000000000 --- a/tests/func/remote/test_index.py +++ /dev/null @@ -1,109 +0,0 @@ -import os - -import pytest - -from dvc.exceptions import DownloadError, UploadError -from dvc.fs.local import LocalFileSystem -from dvc.remote.base import Remote -from dvc.remote.index import RemoteIndex -from dvc.remote.local import LocalRemote -from dvc.utils.fs import remove - - -@pytest.fixture(scope="function") -def remote(tmp_dir, dvc, tmp_path_factory, mocker): - url = os.fspath(tmp_path_factory.mktemp("upstream")) - dvc.config["remote"]["upstream"] = {"url": url} - dvc.config["core"]["remote"] = "upstream" - - # patch hashes_exist since the LocalRemote normally overrides - # BaseFileSystem.hashes_exist. - def hashes_exist(self, *args, **kwargs): - return Remote.hashes_exist(self, *args, **kwargs) - - mocker.patch.object(LocalRemote, "hashes_exist", hashes_exist) - - # patch index class since LocalRemote normally overrides index class - mocker.patch.object(LocalRemote, "INDEX_CLS", RemoteIndex) - - return dvc.cloud.get_remote("upstream") - - -def test_indexed_on_status(tmp_dir, dvc, tmp_path_factory, remote): - foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0] - bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] - baz_hash = bar.obj.trie.get(("baz",)).hash_info - dvc.push() - with remote.index: - remote.index.clear() - - dvc.status(cloud=True) - with remote.index: - assert {bar.hash_info.value, baz_hash.value} == set( - remote.index.hashes() - ) - assert [bar.hash_info.value] == list(remote.index.dir_hashes()) - assert foo.hash_info.value not in remote.index.hashes() - - -def test_indexed_on_push(tmp_dir, dvc, tmp_path_factory, remote): - foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0] - bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] - baz_hash = bar.obj.trie.get(("baz",)).hash_info - - dvc.push() - with remote.index: - assert {bar.hash_info.value, baz_hash.value} == set( - remote.index.hashes() - ) - assert [bar.hash_info.value] == list(remote.index.dir_hashes()) - assert foo.hash_info.value not in remote.index.hashes() - - -def test_indexed_dir_missing(tmp_dir, dvc, tmp_path_factory, remote): - bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0] - with remote.index: - remote.index.update([bar.hash_info.value], []) - dvc.status(cloud=True) - with remote.index: - assert not list(remote.index.hashes()) - - -def test_clear_on_gc(tmp_dir, dvc, tmp_path_factory, remote, mocker): - (foo,) = tmp_dir.dvc_gen({"foo": "foo content"}) - dvc.push() - dvc.remove(foo.relpath) - - mocked_clear = mocker.patch.object(remote.INDEX_CLS, "clear") - dvc.gc(workspace=True, cloud=True) - mocked_clear.assert_called_with() - - -def test_clear_on_download_err(tmp_dir, dvc, tmp_path_factory, remote, mocker): - tmp_dir.dvc_gen({"foo": "foo content"}) - dvc.push() - remove(dvc.odb.local.cache_dir) - - mocked_clear = mocker.patch.object(remote.INDEX_CLS, "clear") - mocker.patch.object(LocalFileSystem, "_download", side_effect=Exception) - with pytest.raises(DownloadError): - dvc.pull() - mocked_clear.assert_called_once_with() - - -def test_partial_upload(tmp_dir, dvc, tmp_path_factory, remote, mocker): - tmp_dir.dvc_gen({"foo": "foo content"}) - tmp_dir.dvc_gen({"bar": {"baz": "baz content"}}) - - original = LocalFileSystem._upload - - def unreliable_upload(self, from_file, to_info, name=None, **kwargs): - if "baz" in name: - raise Exception("stop baz") - return original(self, from_file, to_info, name, **kwargs) - - mocker.patch.object(LocalFileSystem, "_upload", unreliable_upload) - with pytest.raises(UploadError): - dvc.push() - with remote.index: - assert not list(remote.index.hashes()) diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index d1702b3e3..23809d9ac 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -762,8 +762,8 @@ def test_checkout_for_external_outputs(tmp_dir, dvc, workspace): file_path = workspace / "foo" dvc.add("remote://workspace/foo") - remote = dvc.cloud.get_remote("workspace") - remote.fs.remove(file_path) + odb = dvc.cloud.get_remote_odb("workspace") + odb.fs.remove(file_path) assert not file_path.exists() stats = dvc.checkout(force=True) diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py index 85cb8617e..444ff56c4 100644 --- a/tests/func/test_data_cloud.py +++ b/tests/func/test_data_cloud.py @@ -8,12 +8,6 @@ from flaky.flaky_decorator import flaky import dvc as dvc_module from dvc.external_repo import clean_repos from dvc.main import main -from dvc.remote.base import ( - STATUS_DELETED, - STATUS_MISSING, - STATUS_NEW, - STATUS_OK, -) from dvc.stage.exceptions import StageNotFound from dvc.utils.fs import move, remove from dvc.utils.serialize import dump_yaml, load_yaml @@ -49,7 +43,7 @@ def test_cloud(tmp_dir, dvc, remote): # pylint:disable=unused-argument (stage,) = tmp_dir.dvc_gen("foo", "foo") out = stage.outs[0] cache = out.cache_path - md5 = out.hash_info.value + foo_hash = out.hash_info foo_objs = out.get_used_objs().get(None, set()) (stage_dir,) = tmp_dir.dvc_gen( @@ -63,40 +57,33 @@ def test_cloud(tmp_dir, dvc, remote): # pylint:disable=unused-argument ) out_dir = stage_dir.outs[0] cache_dir = out_dir.cache_path - md5_dir = out_dir.hash_info.value dir_objs = {out_dir.obj} - entry_md5s = [entry_obj.hash_info.value for _, entry_obj in out_dir.obj] + dir_hash = out_dir.hash_info + dir_hashes = {dir_hash} | { + entry_obj.hash_info for _, entry_obj in out_dir.obj + } - def _make_dir_status(status): - expected = {md5_dir: {"name": md5_dir, "status": status}} - expected.update( - { - entry_md5: {"name": entry_md5, "status": status} - for entry_md5 in entry_md5s - } - ) - return expected + def _check_status(status, **kwargs): + for key in ("ok", "missing", "new", "deleted"): + expected = kwargs.get(key, set()) + assert expected == {obj.hash_info for obj in getattr(status, key)} # Check status - status = dvc.cloud.status(foo_objs, show_checksums=True) - expected = {md5: {"name": md5, "status": STATUS_NEW}} - assert status == expected + status = dvc.cloud.status(foo_objs) + _check_status(status, new={foo_hash}) - status_dir = dvc.cloud.status(dir_objs, show_checksums=True) - expected = _make_dir_status(STATUS_NEW) - assert status_dir == expected + status_dir = dvc.cloud.status(dir_objs) + _check_status(status_dir, new=dir_hashes) # Move cache and check status # See issue https://github.com/iterative/dvc/issues/4383 for details backup_dir = dvc.odb.local.cache_dir + ".backup" move(dvc.odb.local.cache_dir, backup_dir) - status = dvc.cloud.status(foo_objs, show_checksums=True) - expected = {md5: {"name": md5, "status": STATUS_MISSING}} - assert status == expected + status = dvc.cloud.status(foo_objs) + _check_status(status, missing={foo_hash}) - status_dir = dvc.cloud.status(dir_objs, show_checksums=True) - expected = _make_dir_status(STATUS_MISSING) - assert status_dir == expected + status_dir = dvc.cloud.status(dir_objs) + _check_status(status_dir, missing=dir_hashes) # Restore original cache: remove(dvc.odb.local.cache_dir) @@ -110,24 +97,20 @@ def test_cloud(tmp_dir, dvc, remote): # pylint:disable=unused-argument dvc.cloud.push(dir_objs) assert os.path.isfile(cache_dir) - status = dvc.cloud.status(foo_objs, show_checksums=True) - expected = {md5: {"name": md5, "status": STATUS_OK}} - assert status == expected + status = dvc.cloud.status(foo_objs) + _check_status(status, ok={foo_hash}) - status_dir = dvc.cloud.status(dir_objs, show_checksums=True) - expected = _make_dir_status(STATUS_OK) - assert status_dir == expected + status_dir = dvc.cloud.status(dir_objs) + _check_status(status_dir, ok=dir_hashes) # Remove and check status remove(dvc.odb.local.cache_dir) - status = dvc.cloud.status(foo_objs, show_checksums=True) - expected = {md5: {"name": md5, "status": STATUS_DELETED}} - assert status == expected + status = dvc.cloud.status(foo_objs) + _check_status(status, deleted={foo_hash}) - status_dir = dvc.cloud.status(dir_objs, show_checksums=True) - expected = _make_dir_status(STATUS_DELETED) - assert status_dir == expected + status_dir = dvc.cloud.status(dir_objs) + _check_status(status_dir, deleted=dir_hashes) # Pull and check status dvc.cloud.pull(foo_objs) @@ -139,13 +122,11 @@ def test_cloud(tmp_dir, dvc, remote): # pylint:disable=unused-argument dvc.cloud.pull(dir_objs) assert os.path.isfile(cache_dir) - status = dvc.cloud.status(foo_objs, show_checksums=True) - expected = {md5: {"name": md5, "status": STATUS_OK}} - assert status == expected + status = dvc.cloud.status(foo_objs) + _check_status(status, ok={foo_hash}) - status_dir = dvc.cloud.status(dir_objs, show_checksums=True) - expected = _make_dir_status(STATUS_OK) - assert status_dir == expected + status_dir = dvc.cloud.status(dir_objs) + _check_status(status_dir, ok=dir_hashes) @pytest.mark.needs_internet @@ -568,7 +549,7 @@ def test_pull_partial(tmp_dir, dvc, local_remote): def test_pull_00_prefix(tmp_dir, dvc, remote, monkeypatch): # Related: https://github.com/iterative/dvc/issues/6089 - fs_type = type(dvc.cloud.get_remote("upstream").fs) + fs_type = type(dvc.cloud.get_remote_odb("upstream").fs) monkeypatch.setattr(fs_type, "_ALWAYS_TRAVERSE", True, raising=False) monkeypatch.setattr(fs_type, "LIST_OBJECT_PAGE_SIZE", 256, raising=False) @@ -588,7 +569,7 @@ def test_pull_00_prefix(tmp_dir, dvc, remote, monkeypatch): def test_pull_no_00_prefix(tmp_dir, dvc, remote, monkeypatch): # Related: https://github.com/iterative/dvc/issues/6244 - fs_type = type(dvc.cloud.get_remote("upstream").fs) + fs_type = type(dvc.cloud.get_remote_odb("upstream").fs) monkeypatch.setattr(fs_type, "_ALWAYS_TRAVERSE", True, raising=False) monkeypatch.setattr(fs_type, "LIST_OBJECT_PAGE_SIZE", 256, raising=False) diff --git a/tests/func/test_external_repo.py b/tests/func/test_external_repo.py index 73f359268..4dbc03d42 100644 --- a/tests/func/test_external_repo.py +++ b/tests/func/test_external_repo.py @@ -51,7 +51,7 @@ def test_cache_reused(erepo_dir, mocker, local_cloud): erepo_dir.dvc_gen("file", "text", commit="add file") erepo_dir.dvc.push() - download_spy = mocker.spy(LocalFileSystem, "download_file") + download_spy = mocker.spy(LocalFileSystem, "upload_fobj") # Use URL to prevent any fishy optimizations url = f"file://{erepo_dir}" diff --git a/tests/func/test_import.py b/tests/func/test_import.py index c8c765510..a27618114 100644 --- a/tests/func/test_import.py +++ b/tests/func/test_import.py @@ -5,7 +5,6 @@ from unittest.mock import patch import pytest from funcy import first -import dvc.data_cloud as cloud from dvc.config import NoRemoteError from dvc.dvcfile import Dvcfile from dvc.exceptions import DownloadError, PathMissingError @@ -59,7 +58,9 @@ def test_import_cached_file(erepo_dir, tmp_dir, dvc, scm, monkeypatch): (tmp_dir / dst).unlink() remote_exception = NoRemoteError("dvc import") - with patch.object(cloud.DataCloud, "pull", side_effect=remote_exception): + with patch.object( + dvc.cloud, "get_remote_odb", side_effect=remote_exception + ): tmp_dir.dvc.imp(os.fspath(erepo_dir), src, dst) assert (tmp_dir / dst).is_file() @@ -305,7 +306,7 @@ def test_download_error_pulling_imported_stage(tmp_dir, dvc, erepo_dir): remove(dst_cache) with patch( - "dvc.fs.local.LocalFileSystem._download", side_effect=Exception + "dvc.fs.local.LocalFileSystem.upload_fobj", side_effect=Exception ), pytest.raises(DownloadError): dvc.pull(["foo_imported.dvc"]) @@ -528,7 +529,7 @@ def test_import_with_no_exec(tmp_dir, dvc, erepo_dir): def test_import_with_jobs(mocker, dvc, erepo_dir): - import dvc as dvc_module + import dvc.objects.transfer as otransfer with erepo_dir.chdir(): erepo_dir.dvc_gen( @@ -543,10 +544,11 @@ def test_import_with_jobs(mocker, dvc, erepo_dir): commit="init", ) - spy = mocker.spy(dvc_module.objects, "save") + spy = mocker.spy(otransfer, "transfer") dvc.imp(os.fspath(erepo_dir), "dir1", jobs=3) - run_jobs = tuple(spy.call_args_list[0])[1].get("jobs") - assert run_jobs == 3 + # the first call will be retrieving dir cache for "dir1" w/jobs None + for _args, kwargs in spy.call_args_list[1:]: + assert kwargs.get("jobs") == 3 def test_chained_import(tmp_dir, dvc, make_tmp_dir, erepo_dir, local_cloud): diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index 10dce018d..323444627 100644 --- a/tests/func/test_remote.py +++ b/tests/func/test_remote.py @@ -176,43 +176,39 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc): def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote): foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0] bar = tmp_dir.dvc_gen({"bar": "bar content"})[0].outs[0] - baz = tmp_dir.dvc_gen({"baz": {"foo": "baz content"}})[0].outs[0] + baz = tmp_dir.dvc_gen({"baz": {"foo": "foo content"}})[0].outs[0] # Faulty upload version, failing on foo - original = LocalFileSystem._upload + original = LocalFileSystem.upload_fobj + odb = dvc.cloud.get_remote_odb("upstream") - def unreliable_upload(self, from_file, to_info, name=None, **kwargs): - if "foo" in name: + def unreliable_upload(self, fobj, to_info, **kwargs): + if os.path.abspath(to_info) == os.path.abspath( + odb.get(foo.hash_info).path_info + ): raise Exception("stop foo") - return original(self, from_file, to_info, name, **kwargs) + return original(self, fobj, to_info, **kwargs) - with patch.object(LocalFileSystem, "_upload", unreliable_upload): + with patch.object(LocalFileSystem, "upload_fobj", unreliable_upload): with pytest.raises(UploadError) as upload_error_info: dvc.push() - assert upload_error_info.value.amount == 3 + assert upload_error_info.value.amount == 2 - remote = dvc.cloud.get_remote("upstream") - assert not remote.fs.exists( - remote.odb.hash_to_path_info(foo.hash_info.value) - ) - assert remote.fs.exists( - remote.odb.hash_to_path_info(bar.hash_info.value) - ) - assert not remote.fs.exists( - remote.odb.hash_to_path_info(baz.hash_info.value) - ) + assert not odb.exists(foo.hash_info) + assert odb.exists(bar.hash_info) + assert not odb.exists(baz.hash_info) # Push everything and delete local cache dvc.push() remove(dvc.odb.local.cache_dir) baz.collect_used_dir_cache() - with patch.object(LocalFileSystem, "_download", side_effect=Exception): + with patch.object(LocalFileSystem, "upload_fobj", side_effect=Exception): with pytest.raises(DownloadError) as download_error_info: dvc.pull() # error count should be len(.dir + standalone file checksums) # since files inside dir are ignored if dir cache entry is missing - assert download_error_info.value.amount == 3 + assert download_error_info.value.amount == 2 def test_raise_on_too_many_open_files( @@ -222,7 +218,7 @@ def test_raise_on_too_many_open_files( mocker.patch.object( LocalFileSystem, - "_upload", + "upload_fobj", side_effect=OSError(errno.EMFILE, "Too many open files"), ) @@ -273,16 +269,14 @@ def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote): tmp_dir.dvc_gen({"baz": "baz content"}) mocked_upload = mocker.patch.object( - LocalFileSystem, "_upload", return_value=0 + LocalFileSystem, "upload_fobj", return_value=0 ) dvc.push() # foo .dir file should be uploaded after bar - remote = dvc.cloud.get_remote("upstream") - foo_path = remote.odb.hash_to_path_info(foo.hash_info.value) - bar_path = remote.odb.hash_to_path_info( - foo.obj.trie[("bar",)].hash_info.value - ) + odb = dvc.cloud.get_remote_odb("upstream") + foo_path = odb.hash_to_path_info(foo.hash_info.value) + bar_path = odb.hash_to_path_info(foo.obj.trie[("bar",)].hash_info.value) paths = [args[1] for args, _ in mocked_upload.call_args_list] assert paths.index(foo_path) > paths.index(bar_path) @@ -419,10 +413,8 @@ def test_protect_local_remote(tmp_dir, dvc, local_remote): (stage,) = tmp_dir.dvc_gen("file", "file content") dvc.push() - remote = dvc.cloud.get_remote("upstream") - remote_cache_file = remote.odb.hash_to_path_info( - stage.outs[0].hash_info.value - ) + odb = dvc.cloud.get_remote_odb("upstream") + remote_cache_file = odb.hash_to_path_info(stage.outs[0].hash_info.value) assert os.path.exists(remote_cache_file) assert stat.S_IMODE(os.stat(remote_cache_file).st_mode) == 0o444 @@ -430,7 +422,7 @@ def test_protect_local_remote(tmp_dir, dvc, local_remote): def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote): (stage,) = tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}) - remote = dvc.cloud.get_remote("upstream") + remote_odb = dvc.cloud.get_remote_odb("upstream") odb = dvc.odb.local out = stage.outs[0] @@ -440,30 +432,24 @@ def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote): remove(odb.hash_to_path_info(file_objs[0].hash_info.value)) dvc.push() - assert not remote.fs.exists( - remote.odb.hash_to_path_info(out.hash_info.value) - ) - assert not remote.fs.exists( - remote.odb.hash_to_path_info(file_objs[0].hash_info.value) - ) - assert remote.fs.exists( - remote.odb.hash_to_path_info(file_objs[1].hash_info.value) - ) + assert not remote_odb.exists(out.hash_info) + assert not remote_odb.exists(file_objs[0].hash_info) + assert remote_odb.exists(file_objs[1].hash_info) def test_upload_exists(tmp_dir, dvc, local_remote): tmp_dir.gen("foo", "foo") - remote = dvc.cloud.get_remote("upstream") + odb = dvc.cloud.get_remote_odb("upstream") # allow uploaded files to be writable for this test, # normally they are set to read-only for DVC remotes - remote.fs.CACHE_MODE = 0o644 + odb.fs.CACHE_MODE = 0o644 from_info = PathInfo(tmp_dir / "foo") - to_info = remote.odb.path_info / "foo" - remote.fs.upload(from_info, to_info) - assert remote.fs.exists(to_info) + to_info = odb.path_info / "foo" + odb.fs.upload(from_info, to_info) + assert odb.fs.exists(to_info) tmp_dir.gen("foo", "bar") - remote.fs.upload(from_info, to_info) - with remote.fs.open(to_info) as fobj: + odb.fs.upload(from_info, to_info) + with odb.fs.open(to_info) as fobj: assert fobj.read() == "bar" diff --git a/tests/func/test_update.py b/tests/func/test_update.py index 29a1c6011..ed906ab6d 100644 --- a/tests/func/test_update.py +++ b/tests/func/test_update.py @@ -381,6 +381,9 @@ def test_update_import_url_to_remote_directory( upload_file_mock = mocker.spy(type(dvc.odb.local.fs), "upload_fobj") stage = dvc.update(stage.path, to_remote=True) + # 2 new hashes (foo2, baz2) + 1 .dir hash + assert upload_file_mock.mock.call_count == 3 + dvc.pull("data") assert (tmp_dir / "data").read_text() == { "foo": "foo", @@ -392,8 +395,6 @@ def test_update_import_url_to_remote_directory( "foo_with_different_name": "foo", }, } - # 2 new hashes (foo2, baz2) + 1 .dir hash - assert upload_file_mock.mock.call_count == 3 def test_update_import_url_to_remote_directory_changed_contents( diff --git a/tests/unit/objects/db/test_local.py b/tests/unit/objects/db/test_local.py index 14eb8330f..98d23ea71 100644 --- a/tests/unit/objects/db/test_local.py +++ b/tests/unit/objects/db/test_local.py @@ -8,7 +8,6 @@ from dvc.hash_info import HashInfo from dvc.objects.db.local import LocalObjectDB from dvc.objects.file import HashFile from dvc.path_info import PathInfo -from dvc.remote.index import RemoteIndexNoop def test_status_download_optimization(mocker, dvc): @@ -16,8 +15,9 @@ def test_status_download_optimization(mocker, dvc): And the desired files to fetch are already on the local cache, Don't check the existence of the desired files on the remote cache """ - odb = LocalObjectDB(LocalFileSystem(), PathInfo(".")) + from dvc.objects.status import compare_status + odb = LocalObjectDB(LocalFileSystem(), PathInfo(".")) objs = { HashFile( None, odb.fs, HashInfo("md5", "acbd18db4cc2f85cedef654fccc4a4d8") @@ -30,14 +30,10 @@ def test_status_download_optimization(mocker, dvc): local_exists = [obj.hash_info.value for obj in objs] mocker.patch.object(odb, "hashes_exist", return_value=local_exists) - other_remote = mocker.Mock() - other_remote.url = "other_remote" - other_remote.hashes_exist.return_value = [] - other_remote.index = RemoteIndexNoop() + src_odb = mocker.Mock() - other_remote.status(odb, objs, download=True) - - assert other_remote.hashes_exist.call_count == 0 + compare_status(src_odb, odb, objs, check_deleted=False) + assert src_odb.hashes_exist.call_count == 0 @pytest.mark.parametrize("link_name", ["hardlink", "symlink"]) diff --git a/tests/unit/remote/test_index.py b/tests/unit/remote/test_index.py index 26d60991e..0c32f2e41 100644 --- a/tests/unit/remote/test_index.py +++ b/tests/unit/remote/test_index.py @@ -3,20 +3,17 @@ import os import pytest from funcy import first -from dvc.remote.index import RemoteIndex +from dvc.objects.db.index import ObjectDBIndex -@pytest.fixture(scope="function") +@pytest.fixture def index(dvc): - idx = RemoteIndex(dvc.tmp_dir, "foo") - idx.load() - yield idx - idx.dump() - os.unlink(idx.path) + index_ = ObjectDBIndex(dvc.tmp_dir, "foo") + yield index_ def test_init(dvc, index): - assert str(index.path) == os.path.join(dvc.tmp_dir, "index", "foo.idx") + assert str(index.index_dir) == os.path.join(dvc.tmp_dir, "index", "foo") def test_is_dir_hash(dvc, index): @@ -28,10 +25,10 @@ def test_roundtrip(dvc, index): expected_dir = {"1234.dir"} expected_file = {"5678"} index.update(expected_dir, expected_file) - index.dump() - index.load() - assert set(index.dir_hashes()) == expected_dir - assert set(index.hashes()) == expected_dir | expected_file + + new_index = ObjectDBIndex(dvc.tmp_dir, "foo") + assert set(new_index.dir_hashes()) == expected_dir + assert set(new_index.hashes()) == expected_dir | expected_file def test_clear(dvc, index): diff --git a/tests/unit/remote/test_remote_tree.py b/tests/unit/remote/test_remote_tree.py index e0e7eefd3..caf7aee8e 100644 --- a/tests/unit/remote/test_remote_tree.py +++ b/tests/unit/remote/test_remote_tree.py @@ -3,8 +3,8 @@ import os import pytest from dvc.fs.s3 import S3FileSystem +from dvc.objects.db import _get_odb from dvc.path_info import PathInfo -from dvc.remote import get_remote from dvc.utils.fs import walk_files remotes = [pytest.lazy_fixture(fix) for fix in ["gs", "s3"]] @@ -29,7 +29,7 @@ FILE_WITH_CONTENTS = { def remote(request, dvc): cloud = request.param cloud.gen(FILE_WITH_CONTENTS) - return get_remote(dvc, **cloud.config) + return _get_odb(dvc, cloud.config) @pytest.mark.needs_internet @@ -47,7 +47,7 @@ def test_isdir(remote): ] for expected, path in test_cases: - assert remote.fs.isdir(remote.odb.path_info / path) == expected + assert remote.fs.isdir(remote.path_info / path) == expected @pytest.mark.needs_internet @@ -68,29 +68,29 @@ def test_exists(remote): ] for expected, path in test_cases: - assert remote.fs.exists(remote.odb.path_info / path) == expected + assert remote.fs.exists(remote.path_info / path) == expected @pytest.mark.needs_internet @pytest.mark.parametrize("remote", remotes, indirect=True) def test_walk_files(remote): files = [ - remote.odb.path_info / "data/alice", - remote.odb.path_info / "data/alpha", - remote.odb.path_info / "data/subdir-file.txt", - remote.odb.path_info / "data/subdir/1", - remote.odb.path_info / "data/subdir/2", - remote.odb.path_info / "data/subdir/3", - remote.odb.path_info / "data/subdir/empty_file", + remote.path_info / "data/alice", + remote.path_info / "data/alpha", + remote.path_info / "data/subdir-file.txt", + remote.path_info / "data/subdir/1", + remote.path_info / "data/subdir/2", + remote.path_info / "data/subdir/3", + remote.path_info / "data/subdir/empty_file", ] - assert list(remote.fs.walk_files(remote.odb.path_info / "data")) == files + assert list(remote.fs.walk_files(remote.path_info / "data")) == files @pytest.mark.parametrize("cloud", [pytest.lazy_fixture("s3")]) def test_copy_preserve_etag_across_buckets(cloud, dvc): cloud.gen(FILE_WITH_CONTENTS) - rem = get_remote(dvc, **cloud.config) + rem = _get_odb(dvc, cloud.config) s3 = rem.fs s3.fs.mkdir("another/") @@ -100,7 +100,7 @@ def test_copy_preserve_etag_across_buckets(cloud, dvc): another = S3FileSystem(**config) - from_info = rem.odb.path_info / "foo" + from_info = rem.path_info / "foo" to_info = another.PATH_CLS("s3://another/foo") rem.fs.copy(from_info, to_info) @@ -131,7 +131,7 @@ def test_isfile(remote): ] for expected, path in test_cases: - assert remote.fs.isfile(remote.odb.path_info / path) == expected + assert remote.fs.isfile(remote.path_info / path) == expected @pytest.mark.needs_internet @@ -139,7 +139,7 @@ def test_isfile(remote): def test_download_dir(remote, tmpdir): path = str(tmpdir / "data") to_info = PathInfo(path) - remote.fs.download(remote.odb.path_info / "data", to_info) + remote.fs.download(remote.path_info / "data", to_info) assert os.path.isdir(path) data_dir = tmpdir / "data" assert len(list(walk_files(path))) == 7 diff --git a/tests/unit/remote/test_slow_link_detection.py b/tests/unit/remote/test_slow_link_detection.py index 979b5b936..1bdd190b5 100644 --- a/tests/unit/remote/test_slow_link_detection.py +++ b/tests/unit/remote/test_slow_link_detection.py @@ -2,13 +2,15 @@ from unittest import mock import pytest -import dvc.remote.slow_link_detection -from dvc.remote.slow_link_detection import slow_link_guard +import dvc.objects.db.slow_link_detection +from dvc.objects.db.slow_link_detection import slow_link_guard @pytest.fixture(autouse=True) def timeout_immediately(monkeypatch): - monkeypatch.setattr(dvc.remote.slow_link_detection, "timeout_seconds", 0.0) + monkeypatch.setattr( + dvc.objects.db.slow_link_detection, "timeout_seconds", 0.0 + ) @pytest.fixture @@ -27,7 +29,7 @@ def test_show_warning_once(caplog, make_remote): slow_link_guard(lambda x: None)(remote) slow_link_guard(lambda x: None)(remote) - slow_link_detection = dvc.remote.slow_link_detection + slow_link_detection = dvc.objects.db.slow_link_detection message = slow_link_detection.message # noqa, pylint: disable=no-member assert len(caplog.records) == 1 assert caplog.records[0].message == message diff --git a/tests/unit/test_state.py b/tests/unit/test_state.py deleted file mode 100644 index 311d08e54..000000000 --- a/tests/unit/test_state.py +++ /dev/null @@ -1,23 +0,0 @@ -import pytest - -from dvc.state import _build_sqlite_uri - - -@pytest.mark.parametrize( - "path, osname, result", - [ - ("/abs/path", "posix", "file:///abs/path"), - ("C:\\abs\\path", "nt", "file:///C:/abs/path"), - ("/abs/p?ath", "posix", "file:///abs/p%3fath"), - ("C:\\abs\\p?ath", "nt", "file:///C:/abs/p%3fath"), - ("/abs/p#ath", "posix", "file:///abs/p%23ath"), - ("C:\\abs\\p#ath", "nt", "file:///C:/abs/p%23ath"), - ("/abs/path space", "posix", "file:///abs/path space"), - ("C:\\abs\\path space", "nt", "file:///C:/abs/path space"), - ("/abs/path%20encoded", "posix", "file:///abs/path%2520encoded"), - ("C:\\abs\\path%20encoded", "nt", "file:///C:/abs/path%2520encoded"), - ], -) -def test_build_uri(path, osname, result, mocker): - mocker.patch("os.name", osname) - assert _build_sqlite_uri(path, {}) == result