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

@@ -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__)

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 = "<command>",
-    ) -> "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))

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)

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):

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
             )
 

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 = [

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:

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())

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:

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

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

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,
+    )

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):

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)

dvc/remote/__init__.py

@@ -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)

dvc/remote/base.py

@@ -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)

dvc/remote/index.py

@@ -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_

dvc/remote/local.py

@@ -1,6 +0,0 @@
-from .base import Remote
-from .index import RemoteIndexNoop
-
-
-class LocalRemote(Remote):
-    INDEX_CLS = RemoteIndexNoop  # type: ignore[assignment]

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):

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

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.")

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:

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

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
 

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,
         )
 

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):

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), ""))

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())

tests/func/remote/test_index.py

@@ -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())

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)

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)
 

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}"

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):

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"

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(

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"])

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):

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

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

tests/unit/test_state.py

@@ -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