scm: use dulwich backend when fetching exps during clone/pull (#9023)

refspecs uses pygit2 backend by default, which does not support git credential
helpers. This PR forces to use dulwich backend during clone/pull operations.

dvc/cachemgr.py

@@ -14,7 +14,7 @@ def _get_odb(repo, settings, fs=None):
     return get_odb(fs, fs_path, state=repo.state, **config)
 
 
-class ODBManager:
+class CacheManager:
     CACHE_DIR = "cache"
     CLOUD_SCHEMES = [
         Schemes.S3,

dvc/cli/__init__.py

@@ -153,7 +153,7 @@ def main(argv=None):  # noqa: C901
     from dvc._debug import debugtools
     from dvc.config import ConfigError
     from dvc.exceptions import DvcException, NotDvcRepoError
-    from dvc.logger import disable_other_loggers, set_loggers_level
+    from dvc.logger import set_loggers_level
 
     # NOTE: stderr/stdout may be closed if we are running from dvc.daemon.
     # On Linux we directly call cli.main after double forking and closing
@@ -166,7 +166,6 @@ def main(argv=None):  # noqa: C901
         logging.disable(logging.INFO)
 
     args = None
-    disable_other_loggers()
 
     outerLogLevel = logger.level
     try:

dvc/commands/get.py

@@ -5,7 +5,6 @@ from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
-from dvc.scm import CloneError
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +31,7 @@ class CmdGet(CmdBaseNoRepo):
 
     def _get_file_from_repo(self):
         from dvc.repo import Repo
+        from dvc.scm import CloneError
 
         try:
             Repo.get(

dvc/commands/imp.py

@@ -5,13 +5,14 @@ from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
-from dvc.scm import CloneError
 
 logger = logging.getLogger(__name__)
 
 
 class CmdImport(CmdBase):
     def run(self):
+        from dvc.scm import CloneError
+
         try:
             self.repo.imp(
                 self.args.url,

dvc/commands/plots.py

@@ -150,7 +150,7 @@ class CmdPlots(CmdBase):
                 render_html(
                     renderers=renderers,
                     output_file=output_file,
-                    template_path=self._html_template_path(),
+                    html_template=self._html_template_path(),
                 )
 
                 ui.write(output_file.as_uri())

dvc/commands/update.py

@@ -31,8 +31,7 @@ class CmdUpdate(CmdBase):
 def add_parser(subparsers, parent_parser):
     UPDATE_HELP = (
         "Update data artifact imported (via dvc import or dvc import-url) "
-        "from an external DVC repository or URL, or update data from a "
-        "DVC worktree remote."
+        "from an external DVC repository or URL."
     )
     update_parser = subparsers.add_parser(
         "update",

dvc/config_schema.py

@@ -51,6 +51,8 @@ def ByUrl(mapping):
         # Windows absolute paths should really have scheme == "" (local)
         if os.name == "nt" and len(parsed.scheme) == 1 and parsed.netloc == "":
             return schemas[""](data)
+        if parsed.netloc == "":
+            return schemas[""](data)
         if parsed.scheme not in schemas:
             raise Invalid(f"Unsupported URL type {parsed.scheme}://")
 
@@ -262,6 +264,7 @@ SCHEMA = {
     # section for experimental features
     "feature": {
         Optional("machine", default=False): Bool,
+        Optional("data_index_cache", default=False): Bool,
         # enabled by default. It's of no use, kept for backward compatibility.
         Optional("parametrization", default=True): Bool,
     },

dvc/data_cloud.py

@@ -147,12 +147,12 @@ class DataCloud:
         """
         odb = odb or self.get_remote_odb(remote, "push")
         return self.transfer(
-            self.repo.odb.local,
+            self.repo.cache.local,
             odb,
             objs,
             jobs=jobs,
             dest_index=get_index(odb),
-            cache_odb=self.repo.odb.local,
+            cache_odb=self.repo.cache.local,
             validate_status=self._log_missing,
         )
 
@@ -175,11 +175,11 @@ class DataCloud:
         odb = odb or self.get_remote_odb(remote, "pull")
         return self.transfer(
             odb,
-            self.repo.odb.local,
+            self.repo.cache.local,
             objs,
             jobs=jobs,
             src_index=get_index(odb),
-            cache_odb=self.repo.odb.local,
+            cache_odb=self.repo.cache.local,
             verify=odb.verify,
             validate_status=self._log_missing,
         )
@@ -206,12 +206,12 @@ class DataCloud:
         if not odb:
             odb = self.get_remote_odb(remote, "status")
         return compare_status(
-            self.repo.odb.local,
+            self.repo.cache.local,
             odb,
             objs,
             jobs=jobs,
             dest_index=get_index(odb),
-            cache_odb=self.repo.odb.local,
+            cache_odb=self.repo.cache.local,
         )
 
     def get_url_for(self, remote, checksum):

dvc/dependency/repo.py

@@ -80,7 +80,7 @@ class RepoDependency(Dependency):
             to.fs_path,
             to.fs,
             obj,
-            self.repo.odb.local,
+            self.repo.cache.local,
             ignore=None,
             state=self.repo.state,
             prompt=confirm,
@@ -111,7 +111,7 @@ class RepoDependency(Dependency):
         from dvc_data.hashfile.build import build
         from dvc_data.hashfile.tree import Tree, TreeError
 
-        local_odb = self.repo.odb.local
+        local_odb = self.repo.cache.local
         locked = kwargs.pop("locked", True)
         with self._make_repo(locked=locked, cache_dir=local_odb.path) as repo:
             used_obj_ids = defaultdict(set)

dvc/external_repo.py

@@ -149,7 +149,7 @@ def _get_remote_config(url):
             name = "auto-generated-upstream"
             return {
                 "core": {"remote": name},
-                "remote": {name: {"url": repo.odb.local.path}},
+                "remote": {name: {"url": repo.cache.local.path}},
             }
 
         # Use original remote to make sure that we are using correct url,
@@ -253,7 +253,7 @@ def _pull(git: "Git", unshallow: bool = False):
 
     git.fetch(unshallow=unshallow)
     _merge_upstream(git)
-    fetch_all_exps(git, "origin")
+    fetch_all_exps(git, "origin", backends=["dulwich"])
 
 
 def _merge_upstream(git: "Git"):

dvc/info.py

@@ -41,14 +41,14 @@ def get_dvc_info():
             # can't auto-create it, as it might cause issues if the user
             # later decides to enable shared cache mode with
             # `dvc config cache.shared group`.
-            if os.path.exists(repo.odb.local.path):
+            if os.path.exists(repo.cache.local.path):
                 info.append(f"Cache types: {_get_linktype_support_info(repo)}")
-                fs_type = get_fs_type(repo.odb.local.path)
+                fs_type = get_fs_type(repo.cache.local.path)
                 info.append(f"Cache directory: {fs_type}")
             else:
                 info.append("Cache types: " + error_link("no-dvc-cache"))
 
-            info.append(f"Caches: {_get_caches(repo.odb)}")
+            info.append(f"Caches: {_get_caches(repo.cache)}")
             info.append(f"Remotes: {_get_remotes(repo.config)}")
 
             root_directory = repo.root_dir
@@ -84,7 +84,7 @@ def _get_remotes(config):
 
 
 def _get_linktype_support_info(repo):
-    odb = repo.odb.local
+    odb = repo.cache.local
 
     links = generic.test_links(
         ["reflink", "hardlink", "symlink"],

dvc/logger.py

@@ -1,7 +1,9 @@
 """Manages logging configuration for DVC repo."""
 
+import logging
 import logging.config
 import logging.handlers
+import sys
 
 import colorama
 
@@ -66,12 +68,11 @@ class LoggingException(Exception):
         super().__init__(msg)
 
 
-def excludeFilter(level):
-    class ExcludeLevelFilter(logging.Filter):
-        def filter(self, record):  # noqa: A003
-            return record.levelno < level
+def exclude_filter(level: int):
+    def filter_fn(record: "logging.LogRecord") -> bool:
+        return record.levelno < level
 
-    return ExcludeLevelFilter
+    return filter_fn
 
 
 class ColorFormatter(logging.Formatter):
@@ -86,7 +87,8 @@ class ColorFormatter(logging.Formatter):
     exception cause, just the message and the traceback.
     """
 
-    color_code = {
+    reset = colorama.Fore.RESET
+    color_codes = {
         "TRACE": colorama.Fore.GREEN,
         "DEBUG": colorama.Fore.BLUE,
         "WARNING": colorama.Fore.YELLOW,
@@ -94,47 +96,46 @@ class ColorFormatter(logging.Formatter):
         "CRITICAL": colorama.Fore.RED,
     }
 
-    def format(self, record):  # noqa: A003
+    def __init__(self, log_colors: bool = True) -> None:
+        super().__init__()
+        self.log_colors = log_colors
+
+    def format(self, record) -> str:  # noqa: A003, C901
         record.message = record.getMessage()
         msg = self.formatMessage(record)
 
-        if record.levelname == "INFO":
+        if record.levelno == logging.INFO:
             return msg
 
-        if record.exc_info:
-            if getattr(record, "tb_only", False):
-                cause = ""
-            else:
-                cause = ": ".join(_iter_causes(record.exc_info[1]))
+        ei = record.exc_info
+        if ei:
+            cause = ""
+            if not getattr(record, "tb_only", False):
+                cause = ": ".join(_iter_causes(ei[1]))
+            sep = " - " if msg and cause else ""
+            msg = msg + sep + cause
 
-            msg = "{message}{separator}{cause}".format(
-                message=msg or "",
-                separator=" - " if msg and cause else "",
-                cause=cause,
-            )
+        asctime = ""
+        if _is_verbose():
+            asctime = self.formatTime(record, self.datefmt)
+            if ei and not record.exc_text:
+                record.exc_text = self.formatException(ei)
+            if record.exc_text:
+                if msg[-1:] != "\n":
+                    msg = msg + "\n"
+                msg = msg + record.exc_text + "\n"
+            if record.stack_info:
+                if msg[-1:] != "\n":
+                    msg = msg + "\n"
+                msg = msg + self.formatStack(record.stack_info) + "\n"
 
-            if _is_verbose():
-                msg += _stack_trace(record.exc_info)
-
-        return "{asctime}{color}{levelname}{nc}: {msg}".format(
-            asctime=self.formatTime(record, self.datefmt),
-            color=self.color_code[record.levelname],
-            nc=colorama.Fore.RESET,
-            levelname=record.levelname,
-            msg=msg,
-        )
-
-    def formatTime(self, record, datefmt=None):
-        # only show if current level is set to DEBUG
-        # also, skip INFO as it is used for UI
-        if not _is_verbose() or record.levelno == logging.INFO:
-            return ""
-
-        return "{green}{date}{nc} ".format(
-            green=colorama.Fore.GREEN,
-            date=super().formatTime(record, datefmt),
-            nc=colorama.Fore.RESET,
-        )
+        level = record.levelname
+        if self.log_colors:
+            color = self.color_codes[level]
+            if asctime:
+                asctime = color + asctime + self.reset
+            level = color + level + self.reset
+        return asctime + (" " if asctime else "") + level + ": " + msg
 
 
 class LoggerHandler(logging.StreamHandler):
@@ -179,31 +180,44 @@ def _iter_causes(exc):
         exc = exc.__cause__
 
 
-def _stack_trace(exc_info):
-    import traceback
-
-    return "\n{red}{line}{nc}\n{trace}{red}{line}{nc}".format(
-        red=colorama.Fore.RED,
-        line="-" * 60,
-        trace="".join(traceback.format_exception(*exc_info)),
-        nc=colorama.Fore.RESET,
-    )
-
-
-def disable_other_loggers():
-    logging.captureWarnings(True)
-    loggerDict = logging.root.manager.loggerDict  # pylint: disable=no-member
-    for logger_name, logger in loggerDict.items():
-        if logger_name != "dvc" and not logger_name.startswith("dvc."):
-            logger.disabled = True  # type: ignore[union-attr]
-
-
 def set_loggers_level(level: int = logging.INFO) -> None:
     for name in ["dvc", "dvc_objects", "dvc_data"]:
         logging.getLogger(name).setLevel(level)
 
 
-def setup(level: int = logging.INFO) -> None:
+def setup(level: int = logging.INFO, log_colors: bool = True) -> None:
+    colorama.init()
+
+    formatter = ColorFormatter(log_colors=log_colors and sys.stdout.isatty())
+
+    console_info = LoggerHandler(sys.stdout)
+    console_info.setLevel(logging.INFO)
+    console_info.setFormatter(formatter)
+    console_info.addFilter(exclude_filter(logging.WARNING))
+
+    console_debug = LoggerHandler(sys.stdout)
+    console_debug.setLevel(logging.DEBUG)
+    console_debug.setFormatter(formatter)
+    console_debug.addFilter(exclude_filter(logging.INFO))
+
+    addLoggingLevel("TRACE", logging.DEBUG - 5)
+
+    console_trace = LoggerHandler(sys.stdout)
+    console_trace.setLevel(logging.TRACE)  # type: ignore[attr-defined]
+    console_trace.setFormatter(formatter)
+    console_trace.addFilter(exclude_filter(logging.DEBUG))
+
+    err_formatter = ColorFormatter(log_colors=log_colors and sys.stderr.isatty())
+    console_errors = LoggerHandler(sys.stderr)
+    console_errors.setLevel(logging.WARNING)
+    console_errors.setFormatter(err_formatter)
+
+    for name in ["dvc", "dvc_objects", "dvc_data"]:
+        logger = logging.getLogger(name)
+        logger.setLevel(level)
+        for handler in [console_info, console_debug, console_trace, console_errors]:
+            logger.addHandler(handler)
+
     if level >= logging.DEBUG:
         # Unclosed session errors for asyncio/aiohttp are only available
         # on the tracing mode for extensive debug purposes. They are really
@@ -214,80 +228,3 @@ def setup(level: int = logging.INFO) -> None:
         # action.
         logging.getLogger("asyncio").setLevel(logging.CRITICAL)
         logging.getLogger("aiohttp").setLevel(logging.CRITICAL)
-
-    addLoggingLevel("TRACE", logging.DEBUG - 5)
-    logging.config.dictConfig(
-        {
-            "version": 1,
-            "filters": {
-                "exclude_errors": {"()": excludeFilter(logging.WARNING)},
-                "exclude_info": {"()": excludeFilter(logging.INFO)},
-                "exclude_debug": {"()": excludeFilter(logging.DEBUG)},
-            },
-            "formatters": {"color": {"()": ColorFormatter}},
-            "handlers": {
-                "console_info": {
-                    "class": "dvc.logger.LoggerHandler",
-                    "level": "INFO",
-                    "formatter": "color",
-                    "stream": "ext://sys.stdout",
-                    "filters": ["exclude_errors"],
-                },
-                "console_debug": {
-                    "class": "dvc.logger.LoggerHandler",
-                    "level": "DEBUG",
-                    "formatter": "color",
-                    "stream": "ext://sys.stdout",
-                    "filters": ["exclude_info"],
-                },
-                "console_trace": {
-                    "class": "dvc.logger.LoggerHandler",
-                    "level": "TRACE",
-                    "formatter": "color",
-                    "stream": "ext://sys.stdout",
-                    "filters": ["exclude_debug"],
-                },
-                "console_errors": {
-                    "class": "dvc.logger.LoggerHandler",
-                    "level": "WARNING",
-                    "formatter": "color",
-                    "stream": "ext://sys.stderr",
-                },
-            },
-            "loggers": {
-                "dvc": {
-                    "level": level,
-                    "handlers": [
-                        "console_info",
-                        "console_debug",
-                        "console_trace",
-                        "console_errors",
-                    ],
-                },
-                "dvc_objects": {
-                    "level": level,
-                    "handlers": [
-                        "console_info",
-                        "console_debug",
-                        "console_trace",
-                        "console_errors",
-                    ],
-                },
-                "dvc_data": {
-                    "level": level,
-                    "handlers": [
-                        "console_info",
-                        "console_debug",
-                        "console_trace",
-                        "console_errors",
-                    ],
-                },
-            },
-            "disable_existing_loggers": False,
-        }
-    )
-
-    # NOTE: colorama init needs to go after logging config to avoid potential
-    # conflicts with libs that modify logging (like tensorflow/abseil)
-    # https://github.com/iterative/dvc/issues/8387
-    colorama.init()

dvc/output.py

@@ -1,6 +1,8 @@
 import logging
 import os
+import posixpath
 from collections import defaultdict
+from operator import itemgetter
 from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type, cast
 from urllib.parse import urlparse
 
@@ -27,7 +29,6 @@ from dvc_data.hashfile.hash_info import HashInfo
 from dvc_data.hashfile.istextfile import istextfile
 from dvc_data.hashfile.meta import Meta
 from dvc_data.hashfile.transfer import transfer as otransfer
-from dvc_data.index import DataIndexEntry
 from dvc_objects.errors import ObjectFormatError
 
 from .annotations import ANNOTATION_FIELDS, ANNOTATION_SCHEMA, Annotation
@@ -37,6 +38,7 @@ from .utils import relpath
 from .utils.fs import path_isin
 
 if TYPE_CHECKING:
+    from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.obj import HashFile
     from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
@@ -229,6 +231,17 @@ def merge_file_meta_from_cloud(entry: Dict) -> Dict:
     return entry
 
 
+def _serialize_tree_obj_to_files(obj: "Tree") -> List[Dict[str, Any]]:
+    key = obj.PARAM_RELPATH
+    return sorted(
+        (
+            {key: posixpath.sep.join(parts), **hi.to_dict(), **meta.to_dict()}
+            for parts, meta, hi in obj
+        ),
+        key=itemgetter(key),
+    )
+
+
 class OutputDoesNotExistError(DvcException):
     def __init__(self, path):
         msg = f"output '{path}' does not exist"
@@ -381,6 +394,7 @@ class Output:
         self.fs_path = self._parse_path(self.fs, fs_path)
         self.obj: Optional["HashFile"] = None
 
+        self.odb: Optional["HashFileDB"] = None
         self.remote = remote
 
         if self.fs.version_aware:
@@ -402,8 +416,13 @@ class Output:
 
     def _compute_meta_hash_info_from_files(self) -> None:
         if self.files:
+            from dvc_data.hashfile.db import HashFileDB
+
             tree = Tree.from_list(self.files, hash_name=self.hash_name)
-            tree.digest()
+            tree.digest(with_meta=True)
+            self.odb = HashFileDB(tree.fs, tree.path + ".odb")
+            self.odb.add(tree.path, tree.fs, tree.hash_info.value)
+
             self.hash_info = tree.hash_info
             self.meta.isdir = True
             self.meta.nfiles = len(self.files)
@@ -487,16 +506,18 @@ class Output:
         return self.use_cache or self.stage.is_repo_import
 
     @property
-    def odb(self):
+    def cache(self):
         odb_name = "repo" if self.is_in_repo else self.protocol
-        odb = getattr(self.repo.odb, odb_name)
+        odb = getattr(self.repo.cache, odb_name)
         if self.use_cache and odb is None:
             raise RemoteCacheRequiredError(self.fs.protocol, self.fs_path)
         return odb
 
     @property
     def cache_path(self):
-        return self.odb.fs.unstrip_protocol(self.odb.oid_to_path(self.hash_info.value))
+        return self.cache.fs.unstrip_protocol(
+            self.cache.oid_to_path(self.hash_info.value)
+        )
 
     def get_hash(self):
         _, hash_info = self._get_hash_meta()
@@ -504,9 +525,9 @@ class Output:
 
     def _get_hash_meta(self):
         if self.use_cache:
-            odb = self.odb
+            odb = self.cache
         else:
-            odb = self.repo.odb.local
+            odb = self.repo.cache.local
         _, meta, obj = build(
             odb,
             self.fs_path,
@@ -548,26 +569,6 @@ class Output:
             key = self.fs.path.parts(no_drive)[1:]
         return workspace, key
 
-    def get_entry(self) -> "DataIndexEntry":
-        if self.files and not self.obj:
-            self.obj = self.get_obj()
-
-        entry = DataIndexEntry(
-            meta=self.meta,
-            obj=self.obj,
-            hash_info=self.hash_info,
-        )
-        if self.stage.is_import and not self.stage.is_repo_import:
-            dep = self.stage.deps[0]
-            entry.fs = dep.fs
-            entry.path = dep.fs_path
-            entry.meta = dep.meta
-            if not entry.obj:
-                if not dep.obj and dep.files:
-                    dep.obj = dep.get_obj()
-                entry.obj = dep.obj
-        return entry
-
     def changed_checksum(self):
         return self.hash_info != self.get_hash()
 
@@ -580,7 +581,7 @@ class Output:
             return True
 
         try:
-            ocheck(self.odb, obj)
+            ocheck(self.cache, obj)
             return False
         except (FileNotFoundError, ObjectFormatError):
             return True
@@ -669,7 +670,7 @@ class Output:
 
         if self.use_cache:
             _, self.meta, self.obj = build(
-                self.odb,
+                self.cache,
                 self.fs_path,
                 self.fs,
                 self.hash_name,
@@ -677,7 +678,7 @@ class Output:
             )
         else:
             _, self.meta, self.obj = build(
-                self.repo.odb.local,
+                self.repo.cache.local,
                 self.fs_path,
                 self.fs,
                 self.hash_name,
@@ -692,7 +693,7 @@ class Output:
 
     def set_exec(self) -> None:
         if self.isfile() and self.meta.isexec:
-            self.odb.set_exec(self.fs_path)
+            self.cache.set_exec(self.fs_path)
 
     def _checkout(self, *args, **kwargs) -> Optional[bool]:
         from dvc_data.hashfile.checkout import CheckoutError as _CheckoutError
@@ -722,7 +723,7 @@ class Output:
                 obj = self._commit_granular_dir(filter_info)
             else:
                 staging, _, obj = build(
-                    self.odb,
+                    self.cache,
                     filter_info or self.fs_path,
                     self.fs,
                     self.hash_name,
@@ -730,7 +731,7 @@ class Output:
                 )
                 otransfer(
                     staging,
-                    self.odb,
+                    self.cache,
                     {obj.hash_info},
                     shallow=False,
                     hardlink=True,
@@ -739,7 +740,7 @@ class Output:
                 filter_info or self.fs_path,
                 self.fs,
                 obj,
-                self.odb,
+                self.cache,
                 relink=True,
                 state=self.repo.state,
                 prompt=prompt.confirm,
@@ -749,7 +750,7 @@ class Output:
     def _commit_granular_dir(self, filter_info) -> Optional["HashFile"]:
         prefix = self.fs.path.parts(self.fs.path.relpath(filter_info, self.fs_path))
         staging, _, save_obj = build(
-            self.odb,
+            self.cache,
             self.fs_path,
             self.fs,
             self.hash_name,
@@ -757,10 +758,10 @@ class Output:
         )
         save_obj = cast(Tree, save_obj)
         save_obj = cast(Tree, save_obj.filter(prefix))
-        checkout_obj = save_obj.get_obj(self.odb, prefix)
+        checkout_obj = save_obj.get_obj(self.cache, prefix)
         otransfer(
             staging,
-            self.odb,
+            self.cache,
             {save_obj.hash_info} | {oid for _, _, oid in save_obj},
             shallow=True,
             hardlink=True,
@@ -823,7 +824,8 @@ class Output:
             if obj:
                 obj = cast("Tree", obj)
                 ret[self.PARAM_FILES] = [
-                    split_file_meta_from_cloud(f) for f in obj.as_list(with_meta=True)
+                    split_file_meta_from_cloud(f)
+                    for f in _serialize_tree_obj_to_files(obj)
                 ]
         return ret
 
@@ -858,7 +860,7 @@ class Output:
             obj = tree
         elif self.hash_info:
             try:
-                obj = oload(self.odb, self.hash_info)
+                obj = oload(self.cache, self.hash_info)
             except (FileNotFoundError, ObjectFormatError):
                 return None
         else:
@@ -869,7 +871,7 @@ class Output:
         if filter_info and filter_info != self.fs_path:
             prefix = fs_path.relparts(filter_info, self.fs_path)
             obj = cast(Tree, obj)
-            obj = obj.get_obj(self.odb, prefix)
+            obj = obj.get_obj(self.cache, prefix)
 
         return obj
 
@@ -905,7 +907,7 @@ class Output:
                 filter_info or self.fs_path,
                 self.fs,
                 obj,
-                self.odb,
+                self.cache,
                 force=force,
                 progress_callback=progress_callback,
                 relink=relink,
@@ -946,7 +948,7 @@ class Output:
         self, source, odb=None, jobs=None, update=False, no_progress_bar=False
     ):
         if odb is None:
-            odb = self.odb
+            odb = self.cache
 
         cls, config, from_info = get_cloud_fs(self.repo, url=source)
         from_fs = cls(**config)
@@ -998,15 +1000,15 @@ class Output:
 
     def unprotect(self):
         if self.exists:
-            self.odb.unprotect(self.fs_path)
+            self.cache.unprotect(self.fs_path)
 
     def get_dir_cache(self, **kwargs):
         if not self.is_dir_checksum:
             raise DvcException("cannot get dir cache for file checksum")
 
-        obj = self.odb.get(self.hash_info.value)
+        obj = self.cache.get(self.hash_info.value)
         try:
-            ocheck(self.odb, obj)
+            ocheck(self.cache, obj)
         except FileNotFoundError:
             if self.remote:
                 kwargs["remote"] = self.remote
@@ -1016,7 +1018,7 @@ class Output:
             return self.obj
 
         try:
-            self.obj = oload(self.odb, self.hash_info)
+            self.obj = oload(self.cache, self.hash_info)
         except (FileNotFoundError, ObjectFormatError):
             self.obj = None
 
@@ -1035,7 +1037,7 @@ class Output:
             logger.debug("failed to pull cache for '%s'", self)
 
         try:
-            ocheck(self.odb, self.odb.get(self.hash_info.value))
+            ocheck(self.cache, self.cache.get(self.hash_info.value))
         except FileNotFoundError:
             msg = (
                 "Missing cache for directory '{}'. "
@@ -1099,7 +1101,7 @@ class Output:
         else:
             obj = self.get_obj(filter_info=kwargs.get("filter_info"))
             if not obj:
-                obj = self.odb.get(self.hash_info.value)
+                obj = self.cache.get(self.hash_info.value)
 
         if not obj:
             return {}
@@ -1182,7 +1184,7 @@ class Output:
 
         try:
             merged = merge(
-                self.odb,
+                self.cache,
                 ancestor_info,
                 self.hash_info,
                 other.hash_info,
@@ -1191,12 +1193,12 @@ class Output:
         except TreeMergeError as exc:
             raise MergeError(str(exc)) from exc
 
-        self.odb.add(merged.path, merged.fs, merged.oid)
+        self.cache.add(merged.path, merged.fs, merged.oid)
 
         self.hash_info = merged.hash_info
         self.files = None
         self.meta = Meta(
-            size=du(self.odb, merged),
+            size=du(self.cache, merged),
             nfiles=len(merged),
         )
 

dvc/repo/__init__.py

@@ -23,6 +23,7 @@ if TYPE_CHECKING:
     from dvc.stage import Stage
     from dvc.types import DictStrAny
     from dvc_data.hashfile.state import StateBase
+    from dvc_data.index import DataIndex
 
     from .experiments import Experiments
     from .index import Index
@@ -162,11 +163,11 @@ class Repo:
         repo_factory: Optional[Callable] = None,
         scm: Optional[Union["Git", "NoSCM"]] = None,
     ):
+        from dvc.cachemgr import CacheManager
         from dvc.config import Config
         from dvc.data_cloud import DataCloud
         from dvc.fs import GitFileSystem, LocalFileSystem, localfs
         from dvc.lock import LockNoop, make_lock
-        from dvc.odbmgr import ODBManager
         from dvc.repo.metrics import Metrics
         from dvc.repo.params import Params
         from dvc.repo.plots import Plots
@@ -179,6 +180,7 @@ class Repo:
         self._fs_conf = {"repo_factory": repo_factory}
         self._fs = fs or localfs
         self._scm = scm
+        self._data_index = None
 
         if rev and not fs:
             self._scm = scm = SCM(root_dir or os.curdir)
@@ -205,12 +207,12 @@ class Repo:
         self.stage: "StageLoad" = StageLoad(self)
 
         self.lock: "LockBase"
-        self.odb: ODBManager
+        self.cache: CacheManager
         self.state: "StateBase"
         if isinstance(self.fs, GitFileSystem) or not self.dvc_dir:
             self.lock = LockNoop()
             self.state = StateNoop()
-            self.odb = ODBManager(self)
+            self.cache = CacheManager(self)
             self.tmp_dir = None
         else:
             self.fs.makedirs(cast(str, self.tmp_dir), exist_ok=True)
@@ -228,7 +230,7 @@ class Repo:
                 self.lock = LockNoop()
                 self.state = StateNoop()
 
-            self.odb = ODBManager(self)
+            self.cache = CacheManager(self)
 
             self.stage_cache = StageCache(self)
 
@@ -341,6 +343,31 @@ class Repo:
         # fs.
         self._reset()
 
+    @property
+    def data_index(self) -> Optional["DataIndex"]:
+        from appdirs import user_cache_dir
+        from fsspec.utils import tokenize
+
+        from dvc_data.index import DataIndex
+
+        if not self.config["feature"].get("data_index_cache"):
+            return None
+
+        if self._data_index is None:
+            cache_dir = user_cache_dir(self.config.APPNAME, self.config.APPAUTHOR)
+            index_dir = os.path.join(
+                cache_dir,
+                "index",
+                "data",
+                # scm.root_dir and repo.root_dir don't match for subrepos
+                tokenize((self.scm.root_dir, self.root_dir)),
+            )
+            os.makedirs(index_dir, exist_ok=True)
+
+            self._data_index = DataIndex.open(os.path.join(index_dir, "db.db"))
+
+        return self._data_index
+
     def __repr__(self):
         return f"{self.__class__.__name__}: '{self.root_dir}'"
 
@@ -385,14 +412,14 @@ class Repo:
         return init(root_dir=root_dir, no_scm=no_scm, force=force, subdir=subdir)
 
     def unprotect(self, target):
-        return self.odb.repo.unprotect(target)
+        return self.cache.repo.unprotect(target)
 
     def _ignore(self):
         flist = [self.config.files["local"]]
         if tmp_dir := self.tmp_dir:
             flist.append(tmp_dir)
-        if path_isin(self.odb.repo.path, self.root_dir):
-            flist.append(self.odb.repo.path)
+        if path_isin(self.cache.repo.path, self.root_dir):
+            flist.append(self.cache.repo.path)
 
         for file in flist:
             self.scm_context.ignore(file)
@@ -553,18 +580,17 @@ class Repo:
     def close(self):
         self.scm.close()
         self.state.close()
+        if self._data_index is not None:
+            self._data_index.close()
 
-    def _reset_cached_indecies(self):
+    def _reset(self):
+        self.scm._reset()  # pylint: disable=protected-access
+        self.state.close()
         self.__dict__.pop("index", None)
         self.__dict__.pop("dvcignore", None)
         self.__dict__.pop("dvcfs", None)
         self.__dict__.pop("datafs", None)
 
-    def _reset(self):
-        self.state.close()
-        self.scm._reset()  # pylint: disable=protected-access
-        self._reset_cached_indecies()
-
     def __enter__(self):
         return self
 

dvc/repo/brancher.py

@@ -1,7 +1,10 @@
+import logging
 from typing import Optional
 
 from dvc.scm import iter_revs
 
+logger = logging.getLogger(__name__)
+
 
 def brancher(  # noqa: E302
     self,
@@ -60,6 +63,7 @@ def brancher(  # noqa: E302
 
     scm = self.scm
 
+    logger.trace("switching fs to workspace")  # type: ignore[attr-defined]
     self.fs = LocalFileSystem(url=self.root_dir)
     yield "workspace"
 
@@ -82,7 +86,9 @@ def brancher(  # noqa: E302
         from dvc.fs import GitFileSystem
 
         for sha, names in found_revs.items():
-            self._reset_cached_indecies()  # pylint: disable=W0212
+            logger.trace(  # type: ignore[attr-defined]
+                "switching fs to revision %s [%s]", sha[:7], ", ".join(names)
+            )
             self.fs = GitFileSystem(scm=scm, rev=sha)
             self.root_dir = self.fs.path.join("/", *repo_root_parts)
 

dvc/repo/data.py

@@ -94,7 +94,7 @@ def _diff(
             with_missing
             and change.old
             and change.old.hash_info
-            and not old.odb_map[change.key].exists(change.old.hash_info.value)
+            and not old.storage_map.cache_exists(change.old)
         ):
             # NOTE: emulating previous behaviour
             _add_change("not_in_cache", change)

dvc/repo/diff.py

@@ -162,9 +162,9 @@ def _output_paths(  # noqa: C901
         obj: Optional["HashFile"]
         if on_working_fs:
             _, _, obj = build(
-                repo.odb.repo,
+                repo.cache.repo,
                 output.fs_path,
-                repo.odb.repo.fs,
+                repo.cache.repo.fs,
                 "md5",
                 dry_run=True,
                 ignore=output.dvcignore,
@@ -219,7 +219,7 @@ def _filter_missing(dvcfs: "DVCFileSystem", paths: Set[str]) -> Iterator[str]:
             if (
                 entry
                 and info["type"] == "directory"
-                and not dvcfs.repo.odb.local.exists(entry.hash_info.value)
+                and not dvcfs.repo.cache.local.exists(entry.hash_info.value)
             ):
                 yield path
         except FileNotFoundError:

dvc/repo/experiments/executor/base.py

@@ -32,6 +32,7 @@ from dvc.repo.experiments.refs import (
     EXEC_CHECKPOINT,
     ExpRefInfo,
 )
+from dvc.repo.experiments.utils import to_studio_params
 from dvc.stage.serialize import to_lockfile
 from dvc.ui import ui
 from dvc.utils import dict_sha256, env2bool, relpath
@@ -334,21 +335,17 @@ class BaseExecutor(ABC):
 
         # fetch experiments
         try:
-            dest_scm.fetch_refspecs(
-                self.git_url,
-                [f"{ref}:{ref}" for ref in refs],
-                on_diverged=on_diverged_ref,
-                force=force,
-                **kwargs,
-            )
+            refspecs = [f"{ref}:{ref}" for ref in refs]
             # update last run checkpoint (if it exists)
             if has_checkpoint:
-                dest_scm.fetch_refspecs(
-                    self.git_url,
-                    [f"{EXEC_CHECKPOINT}:{EXEC_CHECKPOINT}"],
-                    force=True,
-                    **kwargs,
-                )
+                refspecs.append(f"{EXEC_CHECKPOINT}:{EXEC_CHECKPOINT}")
+            dest_scm.fetch_refspecs(
+                self.git_url,
+                refspecs,
+                on_diverged=on_diverged_ref,
+                force=force or has_checkpoint,
+                **kwargs,
+            )
         except SCMError:
             pass
 
@@ -545,6 +542,8 @@ class BaseExecutor(ABC):
         log_errors: bool = True,
         **kwargs,
     ):
+        from dvc_studio_client.post_live_metrics import post_live_metrics
+
         from dvc.repo import Repo
         from dvc.stage.monitor import CheckpointKilledError
 
@@ -561,6 +560,13 @@ class BaseExecutor(ABC):
             os.chdir(dvc.root_dir)
 
         try:
+            post_live_metrics(
+                "start",
+                info.baseline_rev,
+                info.name,
+                "dvc",
+                params=to_studio_params(dvc.params.show()),
+            )
             logger.debug("Running repro in '%s'", os.getcwd())
             yield dvc
             info.status = TaskStatus.SUCCESS
@@ -578,6 +584,14 @@ class BaseExecutor(ABC):
             info.status = TaskStatus.FAILED
             raise
         finally:
+            post_live_metrics(
+                "done",
+                info.baseline_rev,
+                info.name,
+                "dvc",
+                experiment_rev=dvc.experiments.scm.get_ref(EXEC_BRANCH),
+            )
+
             if infofile is not None:
                 info.dump_json(infofile)
             dvc.close()
@@ -702,13 +716,10 @@ class BaseExecutor(ABC):
 
     @staticmethod
     def _set_log_level(level):
-        from dvc.logger import disable_other_loggers
-
         # When executor.reproduce is run in a multiprocessing child process,
         # dvc.cli.main will not be called for that child process so we need to
         # setup logging ourselves
         dvc_logger = logging.getLogger("dvc")
-        disable_other_loggers()
         if level is not None:
             dvc_logger.setLevel(level)
 

dvc/repo/experiments/executor/local.py

@@ -153,7 +153,7 @@ class TempDirExecutor(BaseLocalExecutor):
         self, repo: "Repo", rev: str, run_cache: bool = True  # noqa: ARG002
     ):
         """Initialize DVC cache."""
-        self._config(repo.odb.repo.path)
+        self._config(repo.cache.repo.path)
 
     def cleanup(self, infofile: str):
         super().cleanup(infofile)

dvc/repo/experiments/executor/ssh.py

@@ -217,12 +217,12 @@ class SSHExecutor(BaseExecutor):
 
     @contextmanager
     def get_odb(self):
-        from dvc.odbmgr import ODBManager, get_odb
+        from dvc.cachemgr import CacheManager, get_odb
 
         cache_path = posixpath.join(
             self._repo_abspath,
             self.dvc_dir,
-            ODBManager.CACHE_DIR,
+            CacheManager.CACHE_DIR,
         )
 
         with self.sshfs() as fs:

dvc/repo/experiments/queue/celery.py

@@ -336,7 +336,8 @@ class LocalCeleryQueue(BaseStashQueue):
                         task_id,
                         remained_entries[entry],
                     )
-                    self.celery.backend.mark_as_failure(task_id, None)
+                    backend = self.celery.backend
+                    backend.mark_as_failure(task_id, None)  # type: ignore[attr-defined]
 
         if remained_revs:
             raise CannotKillTasksError(remained_revs)

dvc/repo/experiments/queue/remove.py

@@ -40,7 +40,7 @@ def remove_tasks(  # noqa: C901
             msg,
             queue_entry,
         ) in celery_queue._iter_queued():  # pylint: disable=protected-access
-            if queue_entry.stash_rev in stash_revs:
+            if queue_entry.stash_rev in stash_revs and msg.delivery_tag:
                 celery_queue.celery.reject(msg.delivery_tag)
     finally:
         celery_queue.stash.remove_revs(list(stash_revs.values()))
@@ -56,7 +56,8 @@ def remove_tasks(  # noqa: C901
             result: AsyncResult = AsyncResult(task_id)
             if result is not None:
                 result.forget()
-            celery_queue.celery.purge(msg.delivery_tag)
+            if msg.delivery_tag:
+                celery_queue.celery.purge(msg.delivery_tag)
     finally:
         if celery_queue.failed_stash:
             celery_queue.failed_stash.remove_revs(failed_stash_revs)

dvc/repo/experiments/queue/workspace.py

@@ -13,7 +13,7 @@ from dvc.repo.experiments.exceptions import ExpQueueEmptyError
 from dvc.repo.experiments.executor.base import ExecutorInfo, TaskStatus
 from dvc.repo.experiments.executor.local import WorkspaceExecutor
 from dvc.repo.experiments.refs import EXEC_BRANCH, WORKSPACE_STASH
-from dvc.repo.experiments.utils import get_exp_rwlock, to_studio_params
+from dvc.repo.experiments.utils import get_exp_rwlock
 from dvc.rwlock import RWLOCK_FILE, RWLOCK_LOCK, SCHEMA
 from dvc.utils import relpath
 from dvc.utils.fs import remove
@@ -94,8 +94,6 @@ class WorkspaceQueue(BaseStashQueue):
     def _reproduce_entry(
         self, entry: QueueEntry, executor: "BaseExecutor"
     ) -> Dict[str, Dict[str, str]]:
-        from dvc_studio_client.post_live_metrics import post_live_metrics
-
         from dvc.stage.monitor import CheckpointKilledError
 
         results: Dict[str, Dict[str, str]] = defaultdict(dict)
@@ -103,13 +101,6 @@ class WorkspaceQueue(BaseStashQueue):
         infofile = self.get_infofile_path(exec_name)
         try:
             rev = entry.stash_rev
-            post_live_metrics(
-                "start",
-                executor.info.baseline_rev,
-                executor.info.name,
-                "dvc",
-                params=to_studio_params(self.repo.params.show()),
-            )
             exec_result = executor.reproduce(
                 info=executor.info,
                 rev=rev,
@@ -131,13 +122,6 @@ class WorkspaceQueue(BaseStashQueue):
         except Exception as exc:  # noqa: BLE001
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
-            post_live_metrics(
-                "done",
-                executor.info.baseline_rev,
-                executor.info.name,
-                "dvc",
-                experiment_rev=first(results.get(rev, {})),
-            )
             executor.cleanup(infofile)
         return results
 

dvc/repo/experiments/run.py

@@ -61,20 +61,19 @@ def run(  # noqa: C901
         )
 
     if hydra_sweep:
-        if kwargs.get("name"):
-            raise InvalidArgumentError(
-                "Sweep overrides can't be used alongside `--name`"
-            )
         from dvc.utils.hydra import get_hydra_sweeps
 
         sweeps = get_hydra_sweeps(path_overrides)
+        name_prefix = kwargs.get("name")
     else:
         sweeps = [path_overrides]
 
     if not kwargs.get("checkpoint_resume", None):
         kwargs["reset"] = True
 
-    for sweep_overrides in sweeps:
+    for idx, sweep_overrides in enumerate(sweeps):
+        if hydra_sweep and name_prefix is not None:
+            kwargs["name"] = f"{name_prefix}-{idx+1}"
         queue_entry = repo.experiments.queue_one(
             repo.experiments.celery_queue,
             targets=targets,

dvc/repo/experiments/utils.py

@@ -5,6 +5,7 @@ from collections import defaultdict
 from functools import wraps
 from typing import (
     TYPE_CHECKING,
+    Any,
     Callable,
     Dict,
     Generator,
@@ -279,17 +280,15 @@ def check_ref_format(scm: "Git", ref: ExpRefInfo):
         )
 
 
-def fetch_all_exps(scm: "Git", url: str, progress: Optional[Callable] = None):
+def fetch_all_exps(
+    scm: "Git", url: str, progress: Optional[Callable] = None, **kwargs: Any
+):
     refspecs = [
         f"{ref}:{ref}"
-        for ref in iter_remote_refs(scm, url, base=EXPS_NAMESPACE)
+        for ref in iter_remote_refs(scm, url, base=EXPS_NAMESPACE, **kwargs)
         if not (ref.startswith(EXEC_NAMESPACE) or ref in STASHES)
     ]
-    scm.fetch_refspecs(
-        url,
-        refspecs,
-        progress=progress,
-    )
+    scm.fetch_refspecs(url, refspecs, progress=progress, **kwargs)
 
 
 def get_random_exp_name(scm, baseline_rev):
@@ -327,7 +326,9 @@ def to_studio_params(dvc_params):
         "params.yaml": {"foo": 1}
     }
     """
-    result = {}
+    result: Dict = {}
+    if not dvc_params:
+        return result
     for rev_data in dvc_params.values():
         for file_name, file_data in rev_data["data"].items():
             result[file_name] = file_data["data"]

dvc/repo/gc.py

@@ -85,7 +85,7 @@ def gc(
             ).values():
                 used_obj_ids.update(obj_ids)
 
-    for scheme, odb in self.odb.by_scheme():
+    for scheme, odb in self.cache.by_scheme():
         if not odb:
             continue
 

dvc/repo/imports.py

@@ -77,6 +77,7 @@ def unpartial_imports(index: Union["Index", "IndexView"]) -> int:
             else:
                 out.hash_info = cast("HashInfo", entry.hash_info)
                 out.meta = entry.meta
+            out.stage.md5 = out.stage.compute_md5()
             out.stage.dump()
             updated += out.meta.nfiles if out.meta.nfiles is not None else 1
     return updated
@@ -106,7 +107,7 @@ def save_imports(
 
     data_view = unfetched.data["repo"]
     if len(data_view):
-        cache = repo.odb.local
+        cache = repo.cache.local
         if not cache.fs.exists(cache.path):
             os.makedirs(cache.path)
         with TemporaryDirectory(dir=cache.path) as tmpdir:
@@ -114,7 +115,7 @@ def save_imports(
                 desc="Downloading imports from source",
                 unit="files",
             ) as cb:
-                checkout(data_view, tmpdir, cache.fs, callback=cb)
+                checkout(data_view, tmpdir, cache.fs, callback=cb, storage="data")
             md5(data_view)
             save(data_view, odb=cache, hardlink=True)
 

dvc/repo/index.py

@@ -112,6 +112,59 @@ def collect_files(
         dirs[:] = [d for d in dirs if not is_out_or_ignored(root, d)]
 
 
+def _load_data_from_outs(index, prefix, outs):
+    from dvc_data.index import DataIndexEntry
+
+    for out in outs:
+        if not out.use_cache:
+            continue
+
+        ws, key = out.index_key
+
+        entry = DataIndexEntry(
+            key=key,
+            meta=out.meta,
+            hash_info=out.hash_info,
+        )
+
+        if out.stage.is_import and not out.stage.is_repo_import:
+            dep = out.stage.deps[0]
+            entry.meta = dep.meta
+            entry.hash_info = dep.hash_info
+
+        # FIXME PyGTrie-based DataIndex doesn't remove entry.key during
+        # index.add, so we have to set the entry manually here to make
+        # index.view() work correctly.
+        index[(*prefix, ws, *key)] = entry
+
+
+def _load_storage_from_out(storage_map, key, out):
+    from dvc.config import NoRemoteError
+    from dvc_data.index import FileStorage, ObjectStorage
+
+    if out.odb:
+        storage_map.add_data(ObjectStorage(key, out.odb))
+    storage_map.add_cache(ObjectStorage(key, out.cache))
+    try:
+        remote = out.repo.cloud.get_remote(out.remote)
+        if remote.fs.version_aware:
+            storage_map.add_remote(
+                FileStorage(
+                    key=key,
+                    fs=remote.fs,
+                    path=remote.fs.path.join(remote.path, *key),
+                )
+            )
+        else:
+            storage_map.add_remote(ObjectStorage(key, remote.odb))
+    except NoRemoteError:
+        pass
+
+    if out.stage.is_import and not out.stage.is_repo_import:
+        dep = out.stage.deps[0]
+        storage_map.add_data(FileStorage(key, dep.fs, dep.fs_path))
+
+
 class Index:
     def __init__(
         self,
@@ -282,35 +335,60 @@ class Index:
         return dict(by_workspace)
 
     @cached_property
-    def data(self) -> "Dict[str, DataIndex]":
-        from collections import defaultdict
-
-        from dvc.config import NoRemoteError
-        from dvc_data.index import DataIndex
-
-        by_workspace: dict = defaultdict(DataIndex)
-
-        by_workspace["repo"] = DataIndex()
-        by_workspace["local"] = DataIndex()
+    def data_tree(self):
+        from dvc_data.hashfile import Tree
 
+        tree = Tree()
         for out in self.outs:
             if not out.use_cache:
                 continue
 
-            workspace, key = out.index_key
-            data_index = by_workspace[workspace]
+            ws, key = out.index_key
 
-            entry = out.get_entry()
-            entry.key = key
-            data_index.add(entry)
+            tree.add((ws, *key), out.meta, out.hash_info)
 
-            data_index.odb_map[key] = out.odb
-            try:
-                data_index.remote_map[key] = self.repo.cloud.get_remote_odb(out.remote)
-            except NoRemoteError:
-                pass
+        tree.digest()
 
-        return dict(by_workspace)
+        return tree
+
+    @cached_property
+    def data(self) -> "Dict[str, DataIndex]":
+        from dvc_data.index import DataIndex
+
+        prefix: "DataIndexKey"
+        loaded = False
+
+        index = self.repo.data_index
+        if index is None:
+            index = DataIndex()
+
+        prefix = (self.data_tree.hash_info.value,)
+        if prefix in index.ls((), detail=False):
+            loaded = True
+
+        try:
+            if not loaded:
+                _load_data_from_outs(index, prefix, self.outs)
+                index.commit()
+
+            by_workspace = {}
+            by_workspace["repo"] = index.view((*prefix, "repo"))
+            by_workspace["local"] = index.view((*prefix, "local"))
+
+            for out in self.outs:
+                if not out.use_cache:
+                    continue
+
+                ws, key = out.index_key
+                if ws not in by_workspace:
+                    by_workspace[ws] = index.view((*prefix, ws))
+
+                data_index = by_workspace[ws]
+                _load_storage_from_out(data_index.storage_map, key, out)
+
+            return by_workspace
+        finally:
+            index.close()
 
     @staticmethod
     def _hash_targets(
@@ -459,7 +537,7 @@ class IndexView:
             workspace, key = out.index_key
             if filter_info and out.fs.path.isin(filter_info, out.fs_path):
                 key = key + out.fs.path.relparts(filter_info, out.fs_path)
-            if out.meta.isdir:
+            if out.meta.isdir or out.stage.is_import and out.stage.deps[0].meta.isdir:
                 prefixes[workspace].recursive.add(key)
             prefixes[workspace].explicit.update(key[:i] for i in range(len(key), 0, -1))
         return prefixes
@@ -523,7 +601,7 @@ def build_data_index(
                 state=index.repo.state,
             )
         except FileNotFoundError:
-            out_entry = DataIndexEntry(path=out_path, fs=fs)
+            out_entry = DataIndexEntry()
 
         out_entry.key = key
         data.add(out_entry)

dvc/repo/worktree.py

@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Set, Tuple, Uni
 
 from funcy import first
 
+from dvc.exceptions import DvcException
 from dvc.fs.callbacks import Callback
 from dvc.stage.exceptions import StageUpdateError
 
@@ -118,12 +119,6 @@ def fetch_worktree(
     ):
         remote_obj = _get_remote(repo, remote_name, remote, "fetch")
         index = view.data["repo"]
-        for key, entry in index.iteritems():
-            entry.fs = remote_obj.fs
-            entry.path = remote_obj.fs.path.join(
-                remote_obj.path,
-                *key,
-            )
         total = len(index)
         with Callback.as_tqdm_callback(
             unit="file",
@@ -131,7 +126,7 @@ def fetch_worktree(
             disable=total == 0,
         ) as cb:
             cb.set_size(total)
-            transferred += save(index, callback=cb, jobs=jobs)
+            transferred += save(index, callback=cb, jobs=jobs, storage="remote")
     return transferred
 
 
@@ -144,6 +139,7 @@ def push_worktree(
 ) -> int:
     from dvc.repo.index import build_data_index
     from dvc_data.index import checkout
+    from dvc_data.index.checkout import VersioningNotSupported
 
     pushed = 0
     stages: Set["Stage"] = set()
@@ -176,17 +172,23 @@ def push_worktree(
             disable=total == 0,
         ) as cb:
             cb.set_size(total)
-            pushed += checkout(
-                new_index,
-                remote_obj.path,
-                remote_obj.fs,
-                old=old_index,
-                delete=remote_obj.worktree,
-                callback=cb,
-                latest_only=remote_obj.worktree,
-                jobs=jobs,
-                **diff_kwargs,
-            )
+            try:
+                pushed += checkout(
+                    new_index,
+                    remote_obj.path,
+                    remote_obj.fs,
+                    old=old_index,
+                    delete=remote_obj.worktree,
+                    callback=cb,
+                    latest_only=remote_obj.worktree,
+                    jobs=jobs,
+                    **diff_kwargs,
+                )
+            except VersioningNotSupported:
+                logger.exception("")
+                raise DvcException(
+                    f"remote {remote_obj.name!r} does not support versioning"
+                ) from None
 
         for out in view.outs:
             workspace, _key = out.index_key
@@ -349,6 +351,7 @@ def _fetch_out_changes(
             update_meta=False,
             meta_only=True,
             meta_cmp_key=partial(_meta_checksum, remote.fs),
+            storage="data",
             callback=cb,
         )
         out.save()
@@ -368,6 +371,13 @@ def _get_diff_indexes(
         old.add(entry)
     for _, entry in remote_index.iteritems(key):
         new.add(entry)
+
+    for prefix, storage in local_index.storage_map.items():
+        old.storage_map[prefix] = storage
+
+    for prefix, storage in remote_index.storage_map.items():
+        new.storage_map[prefix] = storage
+
     return old, new
 
 

dvc/scm.py

@@ -179,7 +179,7 @@ def clone(url: str, to_path: str, **kwargs):
         try:
             git = Git.clone(url, to_path, progress=pbar.update_git, **kwargs)
             if "shallow_branch" not in kwargs:
-                fetch_all_exps(git, url, progress=pbar.update_git)
+                fetch_all_exps(git, url, progress=pbar.update_git, backends=["dulwich"])
             return git
         except InternalCloneError as exc:
             raise CloneError("SCM error") from exc

dvc/stage/cache.py

@@ -61,7 +61,7 @@ def _get_stage_hash(stage):
 class StageCache:
     def __init__(self, repo):
         self.repo = repo
-        self.cache_dir = os.path.join(self.repo.odb.local.path, "runs")
+        self.cache_dir = os.path.join(self.repo.cache.local.path, "runs")
 
     def _get_cache_dir(self, key):
         return os.path.join(self.cache_dir, key[:2], key)
@@ -120,12 +120,12 @@ class StageCache:
 
     @contextmanager
     def _cache_type_copy(self):
-        cache_types = self.repo.odb.local.cache_types
-        self.repo.odb.local.cache_types = ["copy"]
+        cache_types = self.repo.cache.local.cache_types
+        self.repo.cache.local.cache_types = ["copy"]
         try:
             yield
         finally:
-            self.repo.odb.local.cache_types = cache_types
+            self.repo.cache.local.cache_types = cache_types
 
     def _uncached_outs(self, stage, cache):
         # NOTE: using temporary stage to avoid accidentally modifying original
@@ -167,13 +167,13 @@ class StageCache:
         COMPILED_LOCK_FILE_STAGE_SCHEMA(cache)
 
         path = self._get_cache_path(cache_key, cache_value)
-        parent = self.repo.odb.local.fs.path.parent(path)
-        self.repo.odb.local.makedirs(parent)
+        parent = self.repo.cache.local.fs.path.parent(path)
+        self.repo.cache.local.makedirs(parent)
         tmp = tempfile.NamedTemporaryFile(delete=False, dir=parent).name
         assert os.path.exists(parent)
         assert os.path.isdir(parent)
         dump_yaml(tmp, cache)
-        self.repo.odb.local.move(tmp, path)
+        self.repo.cache.local.move(tmp, path)
 
     def restore(self, stage, run_cache=True, pull=False, dry=False):
         from .serialize import to_single_stage_lockfile
@@ -259,11 +259,11 @@ class StageCache:
 
     def push(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
         dest_odb = odb or self.repo.cloud.get_remote_odb(remote, "push --run-cache")
-        return self.transfer(self.repo.odb.local, dest_odb)
+        return self.transfer(self.repo.cache.local, dest_odb)
 
     def pull(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
         odb = odb or self.repo.cloud.get_remote_odb(remote, "fetch --run-cache")
-        return self.transfer(odb, self.repo.odb.local)
+        return self.transfer(odb, self.repo.cache.local)
 
     def get_used_objs(self, used_run_cache, *args, **kwargs):
         """Return used cache for the specified run-cached stages."""

dvc/stage/serialize.py

@@ -154,7 +154,7 @@ def to_pipeline_file(stage: "PipelineStage"):
 
 
 def to_single_stage_lockfile(stage: "Stage", **kwargs) -> dict:
-    from dvc.output import split_file_meta_from_cloud
+    from dvc.output import _serialize_tree_obj_to_files, split_file_meta_from_cloud
 
     assert stage.cmd
 
@@ -165,7 +165,8 @@ def to_single_stage_lockfile(stage: "Stage", **kwargs) -> dict:
             if obj:
                 obj = cast("Tree", obj)
                 ret[item.PARAM_FILES] = [
-                    split_file_meta_from_cloud(f) for f in obj.as_list(with_meta=True)
+                    split_file_meta_from_cloud(f)
+                    for f in _serialize_tree_obj_to_files(obj)
                 ]
         else:
             meta_d = item.meta.to_dict()

dvc/testing/api_tests.py

@@ -24,7 +24,7 @@ class TestAPI:
         dvc.push()
 
         # Remove cache to force download
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
 
         with api.open("foo") as fobj:
             assert fobj.read() == "foo-text"
@@ -62,7 +62,7 @@ class TestAPI:
         dvc.push()
 
         if clear_cache:
-            remove(dvc.odb.repo.path)
+            remove(dvc.cache.repo.path)
 
         if url := fs_kwargs.get("url"):
             fs_kwargs["url"] = url.format(path=tmp_dir, posixpath=tmp_dir.as_posix())

dvc/testing/fixtures.py

@@ -183,7 +183,7 @@ def local_remote(make_remote):
 @pytest.fixture
 def make_workspace(tmp_dir, dvc, make_cloud):
     def _make_workspace(name, typ="local"):
-        from dvc.odbmgr import ODBManager
+        from dvc.cachemgr import CacheManager
 
         cloud = make_cloud(typ)  # pylint: disable=W0621
 
@@ -197,7 +197,7 @@ def make_workspace(tmp_dir, dvc, make_cloud):
             with dvc.config.edit() as conf:
                 conf["cache"][scheme] = f"{name}-cache"
 
-            dvc.odb = ODBManager(dvc)
+            dvc.cache = CacheManager(dvc)
 
         return cloud
 

dvc/testing/remote_tests.py

@@ -10,6 +10,12 @@ from dvc_data.hashfile.tree import Tree
 # pylint: disable=unused-argument
 
 
+def _check_status(status, **kwargs):
+    for key in ("ok", "missing", "new", "deleted"):
+        expected = kwargs.get(key, set())
+        assert expected == set(getattr(status, key))
+
+
 class TestRemote:
     def test(self, tmp_dir, dvc, remote):  # pylint: disable=W0613
         (stage,) = tmp_dir.dvc_gen("foo", "foo")
@@ -33,11 +39,6 @@ class TestRemote:
         dir_hash = out_dir.hash_info
         dir_hashes = {dir_hash} | {oid for _, _, oid in out_dir.obj}
 
-        def _check_status(status, **kwargs):
-            for key in ("ok", "missing", "new", "deleted"):
-                expected = kwargs.get(key, set())
-                assert expected == set(getattr(status, key))
-
         # Check status
         status = dvc.cloud.status(foo_hashes)
         _check_status(status, new={foo_hash})
@@ -47,8 +48,8 @@ class TestRemote:
 
         # Move cache and check status
         # See issue https://github.com/iterative/dvc/issues/4383 for details
-        backup_dir = dvc.odb.local.path + ".backup"
-        shutil.move(dvc.odb.local.path, backup_dir)
+        backup_dir = dvc.cache.local.path + ".backup"
+        shutil.move(dvc.cache.local.path, backup_dir)
         status = dvc.cloud.status(foo_hashes)
         _check_status(status, missing={foo_hash})
 
@@ -56,8 +57,8 @@ class TestRemote:
         _check_status(status_dir, missing=dir_hashes)
 
         # Restore original cache:
-        remove(dvc.odb.local.path)
-        shutil.move(backup_dir, dvc.odb.local.path)
+        remove(dvc.cache.local.path)
+        shutil.move(backup_dir, dvc.cache.local.path)
 
         # Push and check status
         dvc.cloud.push(foo_hashes)
@@ -74,7 +75,7 @@ class TestRemote:
         _check_status(status_dir, ok=dir_hashes)
 
         # Remove and check status
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
 
         status = dvc.cloud.status(foo_hashes)
         _check_status(status, deleted={foo_hash})
@@ -114,6 +115,58 @@ class TestRemote:
         dvc.pull(run_cache=True)
         assert list(stage_cache_dir.rglob("*")) == expected
 
+    @pytest.mark.xfail(raises=NotImplementedError, strict=False)
+    def test_pull_00_prefix(self, tmp_dir, dvc, remote, monkeypatch):
+        # Related: https://github.com/iterative/dvc/issues/6089
+
+        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)
+
+        # foo's md5 checksum is 00411460f7c92d2124a67ea0f4cb5f85
+        # bar's md5 checksum is 0000000018e6137ac2caab16074784a6
+        foo_out = tmp_dir.dvc_gen("foo", "363")[0].outs[0]
+        bar_out = tmp_dir.dvc_gen("bar", "jk8ssl")[0].outs[0]
+        expected_hashes = {foo_out.hash_info, bar_out.hash_info}
+
+        dvc.push()
+        status = dvc.cloud.status(expected_hashes)
+        _check_status(status, ok=expected_hashes)
+
+        dvc.cache.local.clear()
+        remove(tmp_dir / "foo")
+        remove(tmp_dir / "bar")
+
+        stats = dvc.pull()
+        assert stats["fetched"] == 2
+        assert set(stats["added"]) == {"foo", "bar"}
+
+    @pytest.mark.xfail(raises=NotImplementedError, strict=False)
+    def test_pull_no_00_prefix(self, tmp_dir, dvc, remote, monkeypatch):
+        # Related: https://github.com/iterative/dvc/issues/6244
+
+        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)
+
+        # foo's md5 checksum is 14ffd92a6cbf5f2f657067df0d5881a6
+        # bar's md5 checksum is 64020400f00960c0ef04052547b134b3
+        foo_out = tmp_dir.dvc_gen("foo", "dvc")[0].outs[0]
+        bar_out = tmp_dir.dvc_gen("bar", "cml")[0].outs[0]
+        expected_hashes = {foo_out.hash_info, bar_out.hash_info}
+
+        dvc.push()
+        status = dvc.cloud.status(expected_hashes)
+        _check_status(status, ok=expected_hashes)
+
+        dvc.cache.local.clear()
+        remove(tmp_dir / "foo")
+        remove(tmp_dir / "bar")
+
+        stats = dvc.pull()
+        assert stats["fetched"] == 2
+        assert set(stats["added"]) == {"foo", "bar"}
+
 
 class TestRemoteVersionAware:
     def test_file(self, tmp_dir, dvc, remote_version_aware):  # pylint: disable=W0613
@@ -125,7 +178,7 @@ class TestRemoteVersionAware:
         out = stage.outs[0]
         assert out.meta.version_id
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
 
         dvc.pull()
@@ -152,7 +205,7 @@ class TestRemoteVersionAware:
             assert file["version_id"]
             assert file["remote"] == "upstream"
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "data_dir")
 
         dvc.pull()
@@ -172,7 +225,7 @@ class TestRemoteWorktree:
         out = stage.outs[0]
         assert out.meta.version_id
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
 
         dvc.pull()
@@ -199,7 +252,7 @@ class TestRemoteWorktree:
             assert file["version_id"]
             assert file["remote"] == "upstream"
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "data_dir")
 
         dvc.pull()
@@ -232,7 +285,7 @@ class TestRemoteWorktree:
         tmp_dir.scm_add([tmp_dir / "data_dir.dvc"], commit="v2")
         assert not (remote_worktree / "data_dir" / "data").exists()
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "data_dir")
         # pulling the original pushed version should still succeed
         scm.checkout(v1)
@@ -282,7 +335,7 @@ class TestRemoteWorktree:
         assert (tmp_dir / "data_dir" / "data").read_text() == "modified"
         assert (tmp_dir / "data_dir" / "new_data").read_text() == "new data"
 
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
         remove(tmp_dir / "data_dir")
         dvc.pull()

dvc/testing/workspace_tests.py

@@ -27,7 +27,7 @@ class TestImport:
         pytest.skip()
 
     def test_import_dir(self, tmp_dir, dvc, workspace, stage_md5, dir_md5):
-        from dvc.odbmgr import ODBManager
+        from dvc.cachemgr import CacheManager
 
         workspace.gen({"dir": {"file": "file", "subdir": {"subfile": "subfile"}}})
 
@@ -35,7 +35,7 @@ class TestImport:
         # to import dirs
         with dvc.config.edit() as conf:
             del conf["cache"]
-        dvc.odb = ODBManager(dvc)
+        dvc.cache = CacheManager(dvc)
 
         assert not (tmp_dir / "dir").exists()  # sanity check
         dvc.imp_url("remote://workspace/dir")
@@ -93,7 +93,7 @@ class TestImportURLVersionAware:
         assert (tmp_dir / "file").read_text() == "file"
         assert dvc.status() == {}
 
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
         remove(tmp_dir / "file")
         dvc.pull()
         assert (tmp_dir / "file").read_text() == "file"
@@ -106,7 +106,7 @@ class TestImportURLVersionAware:
         assert (tmp_dir / "file").read_text() == "modified"
         assert dvc.status() == {}
 
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
         remove(tmp_dir / "file")
         dvc.pull()
         assert (tmp_dir / "file").read_text() == "modified"
@@ -119,7 +119,7 @@ class TestImportURLVersionAware:
         assert (tmp_dir / "data_dir" / "subdir" / "file").read_text() == "file"
         assert dvc.status() == {}
 
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
         remove(tmp_dir / "data_dir")
         dvc.pull()
         assert (tmp_dir / "data_dir" / "subdir" / "file").read_text() == "file"
@@ -134,12 +134,22 @@ class TestImportURLVersionAware:
         assert (tmp_dir / "data_dir" / "new_file").read_text() == "new"
         assert dvc.status() == {}
 
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
         remove(tmp_dir / "data_dir")
         dvc.pull()
         assert (tmp_dir / "data_dir" / "subdir" / "file").read_text() == "modified"
         assert (tmp_dir / "data_dir" / "new_file").read_text() == "new"
 
+    def test_import_no_download(self, tmp_dir, dvc, remote_version_aware):
+        remote_version_aware.gen({"data_dir": {"subdir": {"file": "file"}}})
+        dvc.imp_url("remote://upstream/data_dir", version_aware=True, no_download=True)
+        stage = first(dvc.index.stages)
+        assert not stage.outs[0].can_push
+
+        dvc.pull()
+        assert (tmp_dir / "data_dir" / "subdir" / "file").read_text() == "file"
+        assert dvc.status() == {}
+
 
 class TestAdd:
     @pytest.fixture

pyproject.toml

@@ -46,14 +46,14 @@ dependencies = [
     "pygtrie>=2.3.2",
     "dpath>=2.0.2,<3",
     "shtab>=1.3.4,<2",
-    "rich>=10.13.0",
+    "rich>=12.0.0",
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
-    "scmrepo==0.1.7",
-    "dvc-render==0.1.0",
+    "scmrepo==0.1.9",
+    "dvc-render>=0.1.2",
     "dvc-task==0.1.11",
     "dvclive>=1.2.2",
-    "dvc-data==0.37.5",
+    "dvc-data==0.40.1",
     "dvc-http",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.7",
@@ -70,7 +70,7 @@ classifiers = [
 ]
 
 [project.optional-dependencies]
-azure = ["dvc-azure==2.21.0"]
+azure = ["dvc-azure==2.21.1"]
 gdrive = ["dvc-gdrive==2.19.1"]
 gs = ["dvc-gs==2.21.1"]
 hdfs = ["dvc-hdfs==2.19.0"]
@@ -90,7 +90,7 @@ tests = [
     # Test requirements
     "pytest==7.2.1",
     "pytest-cov==4.0.0",
-    "pytest-xdist==3.1.0",
+    "pytest-xdist==3.2.0",
     "pytest-mock==3.10.0",
     "pytest-lazy-fixture==0.6.3",
     # https://github.com/docker/docker-py/issues/2902

tests/dir_helpers.py

@@ -49,8 +49,6 @@ import pathlib
 
 import pytest
 
-from dvc.logger import disable_other_loggers
-
 __all__ = [
     "run_copy",
     "run_head",
@@ -60,9 +58,6 @@ __all__ = [
     "git_downstream",
 ]
 
-# see https://github.com/iterative/dvc/issues/3167
-disable_other_loggers()
-
 
 @pytest.fixture
 def run_copy(tmp_dir, copy_script, dvc):

tests/func/api/test_data.py

@@ -44,7 +44,7 @@ def test_open_external(tmp_dir, erepo_dir, cloud):
     erepo_dir.dvc.push(all_branches=True)
 
     # Remove cache to force download
-    remove(erepo_dir.dvc.odb.local.path)
+    remove(erepo_dir.dvc.cache.local.path)
 
     # Using file url to force clone to tmp repo
     repo_url = f"file://{erepo_dir.as_posix()}"
@@ -59,7 +59,7 @@ def test_open_granular(tmp_dir, dvc, remote):
     dvc.push()
 
     # Remove cache to force download
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     with api.open("dir/foo") as fd:
         assert fd.read() == "foo-text"
@@ -69,7 +69,7 @@ def test_missing(tmp_dir, dvc, remote):
     tmp_dir.dvc_gen("foo", "foo")
 
     # Remove cache to make foo missing
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     api.read("foo")
 
@@ -129,7 +129,7 @@ def test_api_missing_local_cache_exists_on_remote(
     dvc.push()
 
     # Remove cache to make foo missing
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
     remove(first(files))
 
     repo_url = f"file://{tmp_dir.as_posix()}" if as_external else None
@@ -202,7 +202,7 @@ def test_open_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
     erepo_dir.add_remote(config=local_cloud.config, default=True)
     erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create file")
     erepo_dir.dvc.push(remote="other")
-    remove(erepo_dir.dvc.odb.local.path)
+    remove(erepo_dir.dvc.cache.local.path)
 
     with api.open(
         os.path.join("dir", "foo"),

tests/func/data/db/test_index.py

@@ -59,7 +59,7 @@ def test_clear_on_download_err(tmp_dir, dvc, index, mocker):
     dvc.push()
 
     for _, _, hi in out.obj:
-        remove(dvc.odb.local.get(hi.value).path)
+        remove(dvc.cache.local.get(hi.value).path)
     remove(out.fs_path)
 
     assert list(index.hashes())

tests/func/experiments/test_remote.py

@@ -303,12 +303,12 @@ def test_push_pull_cache(
         with open(path, encoding="utf-8") as f:
             assert f.read() == str(x)
 
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     dvc.experiments.pull(remote, [ref_info.name], pull_cache=True)
     for x in range(2, checkpoint_stage.iterations + 1):
         hash_ = digest(str(x))
-        path = os.path.join(dvc.odb.local.path, hash_[:2], hash_[2:])
+        path = os.path.join(dvc.cache.local.path, hash_[:2], hash_[2:])
         assert os.path.exists(path)
         with open(path, encoding="utf-8") as f:
             assert f.read() == str(x)

tests/func/experiments/test_set_params.py

@@ -131,9 +131,12 @@ def test_hydra_sweep_requires_queue(params_repo, dvc):
         dvc.experiments.run(params=["db=mysql,postgresql"])
 
 
-def test_hydra_sweep_cant_use_name(tmp_dir, params_repo, dvc):
-    with pytest.raises(
-        InvalidArgumentError,
-        match="Sweep overrides can't be used alongside `--name`",
-    ):
-        dvc.experiments.run(params=["db=mysql,postgresql"], queue=True, name="foo")
+def test_hydra_sweep_prefix_name(tmp_dir, params_repo, dvc):
+    prefix = "foo"
+    db_values = ["mysql", "postgresql"]
+    param = "+db=" + ",".join(db_values)
+    dvc.experiments.run(params=[param], queue=True, name=prefix)
+    expected_names = [f"{prefix}-{i+1}" for i, _ in enumerate(db_values)]
+    exp_names = [entry.name for entry in dvc.experiments.celery_queue.iter_queued()]
+    for name, expected in zip(exp_names, expected_names):
+        assert name == expected

tests/func/experiments/test_utils.py

@@ -1,4 +1,3 @@
-from dvc_studio_client import env
 from funcy import first
 
 
@@ -28,38 +27,3 @@ def test_generate_random_exp_name(tmp_dir, dvc, scm, exp_stage, mocker):
     # Can use same name because of different baseline_rev
     ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
     assert dvc.experiments.get_exact_name([ref])[ref] == "0-0"
-
-
-def test_post_to_studio(tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch):
-    valid_response = mocker.MagicMock()
-    valid_response.status_code = 200
-    mocked_post = mocker.patch("requests.post", return_value=valid_response)
-
-    monkeypatch.setenv(env.STUDIO_ENDPOINT, "https://0.0.0.0")
-    monkeypatch.setenv(env.STUDIO_REPO_URL, "STUDIO_REPO_URL")
-    monkeypatch.setenv(env.STUDIO_TOKEN, "STUDIO_TOKEN")
-
-    baseline_sha = scm.get_rev()
-    exp_rev = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
-    name = dvc.experiments.get_exact_name([exp_rev])[exp_rev]
-    assert mocked_post.call_count == 2
-
-    start_call, done_call = mocked_post.call_args_list
-
-    assert start_call.kwargs["json"] == {
-        "type": "start",
-        "repo_url": "STUDIO_REPO_URL",
-        "baseline_sha": baseline_sha,
-        "name": name,
-        "params": {"params.yaml": {"foo": 1}},
-        "client": "dvc",
-    }
-
-    assert done_call.kwargs["json"] == {
-        "type": "done",
-        "repo_url": "STUDIO_REPO_URL",
-        "baseline_sha": baseline_sha,
-        "name": name,
-        "client": "dvc",
-        "experiment_rev": exp_rev,
-    }

tests/func/metrics/test_show.py

@@ -259,7 +259,7 @@ def test_metrics_show_overlap(tmp_dir, dvc, run_copy_metrics, clear_before_run):
     # so as it works even for optimized cases
     if clear_before_run:
         remove(data_dir)
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
 
     dvc._reset()
 

tests/func/plots/test_show.py

@@ -161,7 +161,7 @@ def test_plots_show_overlap(tmp_dir, dvc, run_copy_metrics, clear_before_run):
     # so as it works even for optimized cases
     if clear_before_run:
         remove(data_dir)
-        remove(dvc.odb.local.path)
+        remove(dvc.cache.local.path)
 
     dvc._reset()
 

tests/func/test_add.py

@@ -12,6 +12,7 @@ import pytest
 import dvc as dvc_module
 import dvc_data
 from dvc.annotations import Annotation
+from dvc.cachemgr import CacheManager
 from dvc.cli import main
 from dvc.config import ConfigError
 from dvc.dvcfile import DVC_FILE_SUFFIX
@@ -23,7 +24,6 @@ from dvc.exceptions import (
     RecursiveAddingWhileUsingFilename,
 )
 from dvc.fs import LocalFileSystem, system
-from dvc.odbmgr import ODBManager
 from dvc.output import (
     OutputAlreadyTrackedError,
     OutputDoesNotExistError,
@@ -110,7 +110,7 @@ def test_add_directory(tmp_dir, dvc):
 
     hash_info = stage.outs[0].hash_info
 
-    obj = load(dvc.odb.local, hash_info)
+    obj = load(dvc.cache.local, hash_info)
     for key, _, _ in obj:
         for part in key:
             assert "\\" not in part
@@ -342,6 +342,13 @@ def test_double_add_unchanged_dir(tmp_dir, dvc):
     assert ret == 0
 
 
+@pytest.mark.skipif(os.name == "nt", reason="unsupported on Windows")
+def test_add_colon_in_filename(tmp_dir, dvc):
+    tmp_dir.gen("fo:o", "foo")
+    ret = main(["add", "fo:o"])
+    assert ret == 0
+
+
 def test_should_update_state_entry_for_file_after_add(mocker, dvc, tmp_dir):
     file_md5_counter = mocker.spy(dvc_data.hashfile.hash, "file_md5")
     tmp_dir.gen("foo", "foo")
@@ -407,12 +414,12 @@ def test_add_commit(tmp_dir, dvc):
     ret = main(["add", "foo", "--no-commit"])
     assert ret == 0
     assert os.path.isfile("foo")
-    assert not os.path.exists(dvc.odb.local.path)
+    assert not os.path.exists(dvc.cache.local.path)
 
     ret = main(["commit", "foo.dvc"])
     assert ret == 0
     assert os.path.isfile("foo")
-    assert dvc.odb.local.exists("acbd18db4cc2f85cedef654fccc4a4d8")
+    assert dvc.cache.local.exists("acbd18db4cc2f85cedef654fccc4a4d8")
 
 
 def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
@@ -583,7 +590,7 @@ def test_windows_should_add_when_cache_on_different_drive(
     tmp_dir, dvc, temporary_windows_drive
 ):
     dvc.config["cache"]["dir"] = temporary_windows_drive
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
     (stage,) = tmp_dir.dvc_gen({"file": "file"})
     cache_path = stage.outs[0].cache_path
@@ -596,12 +603,12 @@ def test_windows_should_add_when_cache_on_different_drive(
 def test_readding_dir_should_not_unprotect_all(tmp_dir, dvc, mocker):
     tmp_dir.gen("dir/data", "data")
 
-    dvc.odb.local.cache_types = ["symlink"]
+    dvc.cache.local.cache_types = ["symlink"]
 
     dvc.add("dir")
     tmp_dir.gen("dir/new_file", "new_file_content")
 
-    unprotect_spy = mocker.spy(dvc.odb.local, "unprotect")
+    unprotect_spy = mocker.spy(dvc.cache.local, "unprotect")
     dvc.add("dir")
 
     assert not unprotect_spy.mock.called
@@ -609,13 +616,13 @@ def test_readding_dir_should_not_unprotect_all(tmp_dir, dvc, mocker):
 
 
 def test_should_not_checkout_when_adding_cached_copy(tmp_dir, dvc, mocker):
-    dvc.odb.local.cache_types = ["copy"]
+    dvc.cache.local.cache_types = ["copy"]
 
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     shutil.copy("bar", "foo")
 
-    copy_spy = mocker.spy(dvc.odb.local.fs, "copy")
+    copy_spy = mocker.spy(dvc.cache.local.fs, "copy")
 
     dvc.add("foo")
 
@@ -637,11 +644,11 @@ def test_should_relink_on_repeated_add(link, new_link, link_test_func, tmp_dir,
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     os.remove("foo")
-    getattr(dvc.odb.local.fs, link)(
+    getattr(dvc.cache.local.fs, link)(
         (tmp_dir / "bar").fs_path, (tmp_dir / "foo").fs_path
     )
 
-    dvc.odb.local.cache_types = [new_link]
+    dvc.cache.local.cache_types = [new_link]
 
     dvc.add("foo")
 
@@ -650,7 +657,7 @@ def test_should_relink_on_repeated_add(link, new_link, link_test_func, tmp_dir,
 
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
 def test_should_protect_on_repeated_add(link, tmp_dir, dvc):
-    dvc.odb.local.cache_types = [link]
+    dvc.cache.local.cache_types = [link]
 
     tmp_dir.dvc_gen({"foo": "foo"})
 
@@ -709,7 +716,7 @@ def test_not_raises_on_re_add(tmp_dir, dvc):
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
 def test_add_empty_files(tmp_dir, dvc, link):
     file = "foo"
-    dvc.odb.local.cache_types = [link]
+    dvc.cache.local.cache_types = [link]
     stages = tmp_dir.dvc_gen(file, "")
 
     assert (tmp_dir / file).exists()
@@ -718,7 +725,7 @@ def test_add_empty_files(tmp_dir, dvc, link):
 
 
 def test_add_optimization_for_hardlink_on_empty_files(tmp_dir, dvc, mocker):
-    dvc.odb.local.cache_types = ["hardlink"]
+    dvc.cache.local.cache_types = ["hardlink"]
     tmp_dir.gen({"foo": "", "bar": "", "lorem": "lorem", "ipsum": "ipsum"})
     m = mocker.spy(LocalFileSystem, "is_hardlink")
     stages = dvc.add(["foo", "bar", "lorem", "ipsum"])

tests/func/test_checkout.py

@@ -105,9 +105,9 @@ def test_checkout_corrupted_cache_dir(tmp_dir, dvc):
 
     # NOTE: modifying cache file for one of the files inside the directory
     # to check if dvc will detect that the cache is corrupted.
-    obj = load(dvc.odb.local, out.hash_info)
+    obj = load(dvc.cache.local, out.hash_info)
     _, _, entry_oid = list(obj)[0]
-    cache = dvc.odb.local.oid_to_path(entry_oid.value)
+    cache = dvc.cache.local.oid_to_path(entry_oid.value)
 
     os.chmod(cache, 0o644)
     with open(cache, "w+", encoding="utf-8") as fobj:
@@ -161,7 +161,7 @@ class TestCheckoutCleanWorkingDir:
 
 def test_checkout_selective_remove(tmp_dir, dvc):
     # Use copy to test for changes in the inodes
-    dvc.odb.local.cache_types = ["copy"]
+    dvc.cache.local.cache_types = ["copy"]
     tmp_dir.dvc_gen({"data": {"foo": "foo", "bar": "bar"}})
 
     foo_inode = system.inode(os.path.join("data", "foo"))
@@ -357,7 +357,7 @@ def test_checkout_moved_cache_dir_with_symlinks(tmp_dir, dvc):
     assert system.is_symlink(os.path.join("data", "file"))
     old_data_link = os.path.realpath(os.path.join("data", "file"))
 
-    old_cache_dir = dvc.odb.local.path
+    old_cache_dir = dvc.cache.local.path
     new_cache_dir = old_cache_dir + "_new"
     os.rename(old_cache_dir, new_cache_dir)
 
@@ -395,7 +395,7 @@ def test_checkout_no_checksum(tmp_dir, dvc):
     [("hardlink", system.is_hardlink), ("symlink", system.is_symlink)],
 )
 def test_checkout_relink(tmp_dir, dvc, link, link_test_func):
-    dvc.odb.local.cache_types = [link]
+    dvc.cache.local.cache_types = [link]
 
     tmp_dir.dvc_gen({"dir": {"data": "text"}})
     dvc.unprotect("dir/data")
@@ -408,7 +408,7 @@ def test_checkout_relink(tmp_dir, dvc, link, link_test_func):
 
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
 def test_checkout_relink_protected(tmp_dir, dvc, link):
-    dvc.odb.local.cache_types = [link]
+    dvc.cache.local.cache_types = [link]
 
     tmp_dir.dvc_gen("foo", "foo")
     dvc.unprotect("foo")
@@ -595,7 +595,7 @@ def test_checkout_with_relink_existing(tmp_dir, dvc, link):
     (tmp_dir / "foo").unlink()
 
     tmp_dir.dvc_gen("bar", "bar")
-    dvc.odb.local.cache_types = [link]
+    dvc.cache.local.cache_types = [link]
 
     stats = dvc.checkout(relink=True)
     assert stats == {**empty_checkout, "added": ["foo"]}

tests/func/test_data_cloud.py

@@ -46,7 +46,7 @@ def test_cloud_cli(tmp_dir, dvc, remote, mocker):
         _kwargs["jobs"] == jobs for (_args, _kwargs) in oids_exist.call_args_list
     )
 
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
     oids_exist.reset_mock()
 
     assert main(["fetch", *args]) == 0
@@ -86,7 +86,7 @@ def test_cloud_cli(tmp_dir, dvc, remote, mocker):
     assert main(["gc", "-cw", "-f", *args]) == 0
     assert _list_oids_traverse.called
     assert all(_kwargs["jobs"] == 2 for (_args, _kwargs) in oids_exist.call_args_list)
-    shutil.move(dvc.odb.local.path, dvc.odb.local.path + ".back")
+    shutil.move(dvc.cache.local.path, dvc.cache.local.path + ".back")
 
     assert main(["fetch", *args]) == 0
 
@@ -155,7 +155,7 @@ def test_missing_cache(tmp_dir, dvc, local_remote, caplog):
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     # purge cache
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     header = (
         "Some of the cache files do not exist "
@@ -194,20 +194,21 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     # remove artifacts and cache to trigger fetching
     remove("file")
     remove("dir")
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     hash_spy = mocker.spy(dvc_data.hashfile.hash, "file_md5")
 
     dvc.pull()
-    assert hash_spy.call_count == 0
+    # NOTE: 1 is for index.data_tree building
+    assert hash_spy.call_count == 1
 
     # Removing cache will invalidate existing state entries
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     dvc.config["remote"]["upstream"]["verify"] = True
 
     dvc.pull()
-    assert hash_spy.call_count == 4
+    assert hash_spy.call_count == 6
 
 
 @flaky(max_runs=3, min_passes=1)
@@ -224,9 +225,9 @@ def test_pull_git_imports(tmp_dir, dvc, scm, erepo):
 
     assert dvc.pull()["fetched"] == 0
 
-    for item in ["foo", "new_dir", dvc.odb.local.path]:
+    for item in ["foo", "new_dir", dvc.cache.local.path]:
         remove(item)
-    os.makedirs(dvc.odb.local.path, exist_ok=True)
+    os.makedirs(dvc.cache.local.path, exist_ok=True)
     clean_repos()
 
     assert dvc.pull(force=True)["fetched"] == 3
@@ -295,7 +296,7 @@ def test_pull_external_dvc_imports_mixed(tmp_dir, dvc, scm, erepo_dir, local_rem
 
 def clean(outs, dvc=None):
     if dvc:
-        dvc.odb.local.clear()
+        dvc.cache.local.clear()
     for path in outs:
         remove(path)
     if dvc:
@@ -471,11 +472,11 @@ def test_push_pull_fetch_pipeline_stages(tmp_dir, dvc, run_copy, local_remote):
 
     dvc.pull("copy-foo-bar")
     assert (tmp_dir / "bar").exists()
-    assert len(recurse_list_dir(dvc.odb.local.path)) == 2
+    assert len(recurse_list_dir(dvc.cache.local.path)) == 2
     clean(["bar"], dvc)
 
     dvc.fetch("copy-foo-bar")
-    assert len(recurse_list_dir(dvc.odb.local.path)) == 2
+    assert len(recurse_list_dir(dvc.cache.local.path)) == 2
 
 
 def test_pull_partial(tmp_dir, dvc, local_remote):
@@ -521,7 +522,7 @@ def test_output_remote(tmp_dir, dvc, make_remote):
 
     dvc.pull()
 
-    assert set(dvc.odb.local.all()) == {
+    assert set(dvc.cache.local.all()) == {
         "37b51d194a7513e45b56f6524f2d51f2",
         "acbd18db4cc2f85cedef654fccc4a4d8",
         "f97c5d29941bfb1b2fdab0874906ab82",
@@ -554,7 +555,7 @@ def test_target_remote(tmp_dir, dvc, make_remote):
 
     dvc.pull(remote="myremote")
 
-    assert set(dvc.odb.local.all()) == {
+    assert set(dvc.cache.local.all()) == {
         "acbd18db4cc2f85cedef654fccc4a4d8",
         "f97c5d29941bfb1b2fdab0874906ab82",
         "6b18131dc289fd37006705affe961ef8.dir",

tests/func/test_data_status.py

@@ -234,7 +234,7 @@ def test_untracked_newly_added_files(M, tmp_dir, dvc, scm):
 def test_missing_cache_workspace_exists(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
     tmp_dir.dvc_gen("foobar", "foobar")
-    remove(dvc.odb.repo.path)
+    remove(dvc.cache.repo.path)
 
     assert dvc.data_status(untracked_files="all") == {
         **EMPTY_STATUS,
@@ -260,7 +260,7 @@ def test_missing_cache_workspace_exists(M, tmp_dir, dvc, scm):
 def test_missing_cache_missing_workspace(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
     tmp_dir.dvc_gen("foobar", "foobar")
-    for path in [dvc.odb.repo.path, "dir", "foobar"]:
+    for path in [dvc.cache.repo.path, "dir", "foobar"]:
         remove(path)
 
     assert dvc.data_status(untracked_files="all") == {
@@ -285,7 +285,7 @@ def test_missing_cache_missing_workspace(M, tmp_dir, dvc, scm):
 def test_git_committed_missing_cache_workspace_exists(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}, commit="add dir")
     tmp_dir.dvc_gen("foobar", "foobar", commit="add foobar")
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     assert dvc.data_status(untracked_files="all") == {
         **EMPTY_STATUS,
@@ -308,7 +308,7 @@ def test_git_committed_missing_cache_workspace_exists(M, tmp_dir, dvc, scm):
 def test_git_committed_missing_cache_missing_workspace(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}, commit="add dir")
     tmp_dir.dvc_gen("foobar", "foobar", commit="add foobar")
-    for path in [dvc.odb.repo.path, "dir", "foobar"]:
+    for path in [dvc.cache.repo.path, "dir", "foobar"]:
         remove(path)
 
     assert dvc.data_status(untracked_files="all") == {
@@ -330,7 +330,7 @@ def test_partial_missing_cache(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
 
     # remove "foo" from cache
-    odb = dvc.odb.repo
+    odb = dvc.cache.repo
     odb.fs.rm(odb.oid_to_path("acbd18db4cc2f85cedef654fccc4a4d8"))
 
     assert dvc.data_status() == {
@@ -354,7 +354,7 @@ def test_missing_dir_object_from_head(M, tmp_dir, dvc, scm):
     (stage,) = tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}, commit="add dir")
     remove("dir")
     tmp_dir.dvc_gen({"dir": {"foobar": "foobar"}})
-    odb = dvc.odb.repo
+    odb = dvc.cache.repo
     odb.fs.rm(odb.oid_to_path(stage.outs[0].hash_info.value))
 
     assert dvc.data_status() == {
@@ -376,7 +376,7 @@ def test_missing_dir_object_from_index(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}, commit="add dir")
     remove("dir")
     (stage,) = tmp_dir.dvc_gen({"dir": {"foobar": "foobar"}})
-    odb = dvc.odb.repo
+    odb = dvc.cache.repo
     odb.fs.rm(odb.oid_to_path(stage.outs[0].hash_info.value))
 
     assert dvc.data_status() == {

tests/func/test_diff.py

@@ -250,10 +250,10 @@ def test_diff_no_cache(tmp_dir, scm, dvc):
     tmp_dir.dvc_gen({"dir": {"file": "modified file content"}}, commit="second")
     scm.tag("v2")
 
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     # invalidate_dir_info to force cache loading
-    dvc.odb.local._dir_info = {}
+    dvc.cache.local._dir_info = {}
 
     diff = dvc.diff("v1", "v2")
     assert diff["added"] == []

tests/func/test_external_repo.py

@@ -115,7 +115,7 @@ def test_relative_remote(erepo_dir, tmp_dir):
     erepo_dir.dvc.push()
 
     (erepo_dir / "file").unlink()
-    remove(erepo_dir.dvc.odb.local.path)
+    remove(erepo_dir.dvc.cache.local.path)
 
     url = os.fspath(erepo_dir)
 
@@ -195,12 +195,12 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir):
         assert (tmp_dir / "out").read_text() == expected_files
 
         # clear cache to test saving to cache
-        cache_dir = tmp_dir / repo.odb.local.path
+        cache_dir = tmp_dir / repo.cache.local.path
         remove(cache_dir)
         os.makedirs(cache_dir)
 
         staging, _, obj = build(
-            repo.odb.local,
+            repo.cache.local,
             "dir",
             repo.dvcfs,
             "md5",
@@ -208,7 +208,7 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir):
         )
         transfer(
             staging,
-            repo.odb.local,
+            repo.cache.local,
             {obj.hash_info},
             shallow=False,
             hardlink=True,

tests/func/test_gc.py

@@ -20,7 +20,7 @@ def good_and_bad_cache(tmp_dir, dvc):
         {"sub": {"data_sub": "data_sub", "data": "data", "тест": "проверка"}},
     )
     raw_dir_hash = stage.outs[0].hash_info.as_raw().value
-    odb = dvc.odb.local
+    odb = dvc.cache.local
 
     bad_cache = {raw_dir_hash}
     for i in ["123", "234", "345"]:
@@ -33,14 +33,14 @@ def good_and_bad_cache(tmp_dir, dvc):
 
 def test_gc_api(dvc, good_and_bad_cache):
     dvc.gc(workspace=True)
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     good_cache, bad_cache = good_and_bad_cache
     assert set(odb.oids_exist([*good_cache, *bad_cache])) == good_cache
 
 
 def test_gc_cli(dvc, good_and_bad_cache):
     assert main(["gc", "-wf"]) == 0
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     good_cache, bad_cache = good_and_bad_cache
     assert set(odb.oids_exist([*good_cache, *bad_cache])) == good_cache
 
@@ -59,7 +59,7 @@ def test_gc_branches_tags(tmp_dir, dvc, scm):
     dvc.remove("file.dvc")
     tmp_dir.dvc_gen("file", "master", commit="trash")
 
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     assert len(list(odb.all())) == 4
 
     dvc.gc(all_tags=True, all_branches=True)
@@ -76,12 +76,12 @@ def test_gc_multiple_dvc_repos(tmp_dir, scm, dvc, erepo_dir):
     tmp_dir.dvc_gen("only_in_first", "only in main repo")
     tmp_dir.dvc_gen("in_both", "in both repos")
 
-    erepo_dir.dvc.odb.local.path = dvc.odb.local.path
+    erepo_dir.dvc.cache.local.path = dvc.cache.local.path
     with erepo_dir.chdir():
         erepo_dir.dvc_gen("in_both", "in both repos")
         erepo_dir.dvc_gen("only_in_second", "only in additional repo")
 
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     assert len(list(odb.all())) == 3
 
     dvc.gc(repos=[erepo_dir], workspace=True)
@@ -97,11 +97,11 @@ def test_all_commits(tmp_dir, scm, dvc):
     tmp_dir.dvc_gen("testfile", "modified", commit="modified")
     tmp_dir.dvc_gen("testfile", "workspace")
 
-    n = _count_files(dvc.odb.local.path)
+    n = _count_files(dvc.cache.local.path)
     dvc.gc(all_commits=True)
 
     # Only one uncommitted file should go away
-    assert _count_files(dvc.odb.local.path) == n - 1
+    assert _count_files(dvc.cache.local.path) == n - 1
 
 
 def test_gc_no_dir_cache(tmp_dir, dvc):
@@ -113,9 +113,9 @@ def test_gc_no_dir_cache(tmp_dir, dvc):
     with pytest.raises(CollectCacheError):
         dvc.gc(workspace=True)
 
-    assert _count_files(dvc.odb.local.path) == 4
+    assert _count_files(dvc.cache.local.path) == 4
     dvc.gc(force=True, workspace=True)
-    assert _count_files(dvc.odb.local.path) == 2
+    assert _count_files(dvc.cache.local.path) == 2
 
 
 def _count_files(path):
@@ -240,14 +240,14 @@ def test_gc_not_collect_pipeline_tracked_files(tmp_dir, dvc, run_copy):
 
     run_copy("foo", "foo2", name="copy")
     shutil.rmtree(dvc.stage_cache.cache_dir)
-    assert _count_files(dvc.odb.local.path) == 1
+    assert _count_files(dvc.cache.local.path) == 1
     dvc.gc(workspace=True, force=True)
-    assert _count_files(dvc.odb.local.path) == 1
+    assert _count_files(dvc.cache.local.path) == 1
 
     # remove pipeline file and lockfile and check
     load_file(dvc, PROJECT_FILE).remove(force=True)
     dvc.gc(workspace=True, force=True)
-    assert _count_files(dvc.odb.local.path) == 0
+    assert _count_files(dvc.cache.local.path) == 0
 
 
 def test_gc_external_output(tmp_dir, dvc, workspace):
@@ -328,13 +328,15 @@ def test_date(tmp_dir, scm, dvc):
 
     dvc.gc(commit_date=datestamp)
 
-    assert _count_files(dvc.odb.local.path) == 1
-    assert dvc.odb.local.exists("9ae73c65f418e6f79ceb4f0e4a4b98d5")  # "modified"
+    assert _count_files(dvc.cache.local.path) == 1
+    assert dvc.cache.local.exists("9ae73c65f418e6f79ceb4f0e4a4b98d5")  # "modified"
 
     tmp_dir.dvc_gen("testfile", "modified, again", commit="modify")
 
     datestamp = (now.date() - datetime.timedelta(days=1)).isoformat()
     dvc.gc(commit_date=datestamp)
-    assert _count_files(dvc.odb.local.path) == 2
-    assert dvc.odb.local.exists("9ae73c65f418e6f79ceb4f0e4a4b98d5")
-    assert dvc.odb.local.exists("3bcf3b1be3e794a97a5a6b93a005784c")  # "modified, again"
+    assert _count_files(dvc.cache.local.path) == 2
+    assert dvc.cache.local.exists("9ae73c65f418e6f79ceb4f0e4a4b98d5")
+    assert dvc.cache.local.exists(
+        "3bcf3b1be3e794a97a5a6b93a005784c"
+    )  # "modified, again"

tests/func/test_get.py

@@ -3,9 +3,9 @@ import os
 
 import pytest
 
+from dvc.cachemgr import CacheManager
 from dvc.cli import main
 from dvc.fs import system
-from dvc.odbmgr import ODBManager
 from dvc.repo import Repo
 from dvc.repo.get import GetDVCFileError
 from dvc.testing.tmp_dir import make_subrepo
@@ -76,7 +76,7 @@ def test_cache_type_is_properly_overridden(tmp_dir, erepo_dir):
     with erepo_dir.chdir():
         with erepo_dir.dvc.config.edit() as conf:
             conf["cache"]["type"] = "symlink"
-        erepo_dir.dvc.odb = ODBManager(erepo_dir.dvc)
+        erepo_dir.dvc.cache = CacheManager(erepo_dir.dvc)
         erepo_dir.scm_add(
             [erepo_dir.dvc.config.files["repo"]], "set cache type to symlinks"
         )

tests/func/test_ignore.py

@@ -477,7 +477,7 @@ def test_pull_ignore(tmp_dir, dvc, local_cloud):
     foo_path.unlink()
     assert not foo_path.exists()
 
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
     dvc.pull()
 
     assert foo_path.exists()

tests/func/test_import.py

@@ -5,11 +5,11 @@ import pytest
 from funcy import first
 
 from dvc.annotations import Annotation
+from dvc.cachemgr import CacheManager
 from dvc.config import NoRemoteError
 from dvc.dvcfile import load_file
 from dvc.exceptions import DownloadError, PathMissingError
 from dvc.fs import system
-from dvc.odbmgr import ODBManager
 from dvc.scm import Git
 from dvc.stage.exceptions import StagePathNotFoundError
 from dvc.testing.tmp_dir import make_subrepo
@@ -272,7 +272,7 @@ def test_cache_type_is_properly_overridden(tmp_dir, scm, dvc, erepo_dir):
     with erepo_dir.chdir():
         with erepo_dir.dvc.config.edit() as conf:
             conf["cache"]["type"] = "symlink"
-        erepo_dir.dvc.odb = ODBManager(erepo_dir.dvc)
+        erepo_dir.dvc.cache = CacheManager(erepo_dir.dvc)
         erepo_dir.scm_add(
             [erepo_dir.dvc.config.files["repo"]],
             "set source repo cache type to symlink",
@@ -294,7 +294,7 @@ def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
 
     remove("dir_imported")
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     dvc.pull(["dir_imported.dvc"])
 
@@ -308,7 +308,7 @@ def test_pull_wildcard_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.imp(os.fspath(erepo_dir), "dir123", "dir_imported123")
 
     remove("dir_imported123")
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     dvc.pull(["dir_imported*.dvc"], glob=True)
 
@@ -536,7 +536,7 @@ def test_pull_imported_stage_from_subrepos(tmp_dir, dvc, erepo_dir, is_dvc, file
     dvc.imp(os.fspath(erepo_dir), path, out="out")
 
     # clean everything
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
     remove("out")
 
     stats = dvc.pull(["out.dvc"])
@@ -606,7 +606,7 @@ def test_chained_import(tmp_dir, dvc, make_tmp_dir, erepo_dir, local_cloud):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}, commit="init")
     erepo_dir.dvc.push()
-    remove(erepo_dir.dvc.odb.local.path)
+    remove(erepo_dir.dvc.cache.local.path)
     remove(os.fspath(erepo_dir / "dir"))
 
     erepo2 = make_tmp_dir("erepo2", scm=True, dvc=True)
@@ -614,7 +614,7 @@ def test_chained_import(tmp_dir, dvc, make_tmp_dir, erepo_dir, local_cloud):
         erepo2.dvc.imp(os.fspath(erepo_dir), "dir")
         erepo2.scm.add("dir.dvc")
         erepo2.scm.commit("import")
-    remove(erepo2.dvc.odb.local.path)
+    remove(erepo2.dvc.cache.local.path)
     remove(os.fspath(erepo2 / "dir"))
 
     dvc.imp(os.fspath(erepo2), "dir", "dir_imported")
@@ -622,14 +622,14 @@ def test_chained_import(tmp_dir, dvc, make_tmp_dir, erepo_dir, local_cloud):
     assert (dst / "foo").read_text() == "foo"
     assert (dst / "bar").read_text() == "bar"
 
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
     remove("dir_imported")
 
     # pulled objects should come from the original upstream repo's remote,
     # no cache or remote should be needed from the intermediate repo
     dvc.pull(["dir_imported.dvc"])
-    assert not os.path.exists(erepo_dir.dvc.odb.local.path)
-    assert not os.path.exists(erepo2.dvc.odb.local.path)
+    assert not os.path.exists(erepo_dir.dvc.cache.local.path)
+    assert not os.path.exists(erepo2.dvc.cache.local.path)
     assert (dst / "foo").read_text() == "foo"
     assert (dst / "bar").read_text() == "bar"
 

tests/func/test_odb.py

@@ -4,8 +4,8 @@ import stat
 import configobj
 import pytest
 
+from dvc.cachemgr import CacheManager
 from dvc.cli import main
-from dvc.odbmgr import ODBManager
 from dvc.utils import relpath
 from dvc_data.hashfile.hash_info import HashInfo
 from dvc_objects.errors import ObjectFormatError
@@ -15,12 +15,12 @@ def test_cache(tmp_dir, dvc):
     cache1_md5 = "123"
     cache2_md5 = "234"
     cache1 = os.path.join(
-        dvc.odb.local.path,
+        dvc.cache.local.path,
         cache1_md5[0:2],
         cache1_md5[2:],
     )
     cache2 = os.path.join(
-        dvc.odb.local.path,
+        dvc.cache.local.path,
         cache2_md5[0:2],
         cache2_md5[2:],
     )
@@ -29,7 +29,7 @@ def test_cache(tmp_dir, dvc):
     assert os.path.exists(cache1)
     assert os.path.exists(cache2)
 
-    odb = ODBManager(dvc)
+    odb = CacheManager(dvc)
 
     md5_list = list(odb.local.all())
     assert len(md5_list) == 2
@@ -46,16 +46,16 @@ def test_cache_load_bad_dir_cache(tmp_dir, dvc):
     from dvc_data.hashfile import load
 
     dir_hash = "123.dir"
-    fname = os.fspath(dvc.odb.local.oid_to_path(dir_hash))
+    fname = os.fspath(dvc.cache.local.oid_to_path(dir_hash))
     tmp_dir.gen({fname: "<clearly>not,json"})
     with pytest.raises(ObjectFormatError):
-        load(dvc.odb.local, HashInfo("md5", dir_hash))
+        load(dvc.cache.local, HashInfo("md5", dir_hash))
 
     dir_hash = "234.dir"
-    fname = os.fspath(dvc.odb.local.oid_to_path(dir_hash))
+    fname = os.fspath(dvc.cache.local.oid_to_path(dir_hash))
     tmp_dir.gen({fname: '{"a": "b"}'})
     with pytest.raises(ObjectFormatError):
-        load(dvc.odb.local, HashInfo("md5", dir_hash))
+        load(dvc.cache.local, HashInfo("md5", dir_hash))
 
 
 def test_external_cache_dir(tmp_dir, dvc, make_tmp_dir):
@@ -63,8 +63,8 @@ def test_external_cache_dir(tmp_dir, dvc, make_tmp_dir):
 
     with dvc.config.edit() as conf:
         conf["cache"]["dir"] = cache_dir.fs_path
-    assert not os.path.exists(dvc.odb.local.path)
-    dvc.odb = ODBManager(dvc)
+    assert not os.path.exists(dvc.cache.local.path)
+    dvc.cache = CacheManager(dvc)
 
     tmp_dir.dvc_gen({"foo": "foo"})
 
@@ -87,9 +87,9 @@ def test_remote_cache_references(tmp_dir, dvc):
         conf["remote"]["cache"] = {"url": "remote://storage/tmp"}
         conf["cache"]["ssh"] = "cache"
 
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
-    assert dvc.odb.ssh.path == "/tmp"
+    assert dvc.cache.ssh.path == "/tmp"
 
 
 def test_shared_cache_dir(tmp_dir):
@@ -126,7 +126,7 @@ def test_shared_cache_dir(tmp_dir):
 def test_cache_link_type(tmp_dir, scm, dvc):
     with dvc.config.edit() as conf:
         conf["cache"]["type"] = "reflink,copy"
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
     stages = tmp_dir.dvc_gen({"foo": "foo"})
     assert len(stages) == 1
@@ -154,7 +154,7 @@ def test_cmd_cache_relative_path(tmp_dir, scm, dvc, make_tmp_dir):
     assert ret == 0
 
     dvc.config.load()
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
     # NOTE: we are in the repo's root and config is in .dvc/, so
     # dir path written to config should be just one level above.
@@ -170,7 +170,7 @@ def test_cmd_cache_relative_path(tmp_dir, scm, dvc, make_tmp_dir):
 
 
 def test_default_cache_type(dvc):
-    assert dvc.odb.local.cache_types == ["reflink", "copy"]
+    assert dvc.cache.local.cache_types == ["reflink", "copy"]
 
 
 @pytest.mark.skipif(os.name == "nt", reason="Not supported for Windows.")
@@ -181,8 +181,8 @@ def test_shared_cache(tmp_dir, dvc, group):
     if group:
         with dvc.config.edit() as conf:
             conf["cache"].update({"shared": "group"})
-    dvc.odb = ODBManager(dvc)
-    cache_dir = dvc.odb.local.path
+    dvc.cache = CacheManager(dvc)
+    cache_dir = dvc.cache.local.path
 
     assert not os.path.exists(cache_dir)
 

tests/func/test_remote.py

@@ -139,16 +139,16 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
     tree = Tree.from_list([{"relpath": "1", "md5": "1"}, {"relpath": "2", "md5": "2"}])
     tree.digest()
     with patch("dvc_data.hashfile.build._build_tree", return_value=(None, tree)):
-        _, _, obj = build(dvc.odb.local, path, dvc.odb.local.fs, "md5")
+        _, _, obj = build(dvc.cache.local, path, dvc.cache.local.fs, "md5")
         hash1 = obj.hash_info
 
     # remove the raw dir obj to force building the tree on the next build call
-    dvc.odb.local.fs.remove(dvc.odb.local.oid_to_path(hash1.as_raw().value))
+    dvc.cache.local.fs.remove(dvc.cache.local.oid_to_path(hash1.as_raw().value))
 
     tree = Tree.from_list([{"md5": "1", "relpath": "1"}, {"md5": "2", "relpath": "2"}])
     tree.digest()
     with patch("dvc_data.hashfile.build._build_tree", return_value=(None, tree)):
-        _, _, obj = build(dvc.odb.local, path, dvc.odb.local.fs, "md5")
+        _, _, obj = build(dvc.cache.local, path, dvc.cache.local.fs, "md5")
         hash2 = obj.hash_info
 
     assert hash1 == hash2
@@ -201,7 +201,7 @@ def test_partial_push_n_pull(  # noqa: C901
 
     # Push everything and delete local cache
     dvc.push()
-    dvc.odb.local.clear()
+    dvc.cache.local.clear()
 
     baz._collect_used_dir_cache()
 
@@ -263,7 +263,7 @@ def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
     external_dir = tmp_path_factory.mktemp("external_dir")
     file = external_dir / "file"
 
-    dvc.odb.local = None
+    dvc.cache.local = None
     with pytest.raises(RemoteCacheRequiredError):
         dvc.run(
             cmd=f"echo content > {file}",
@@ -426,7 +426,7 @@ def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote):
     (stage,) = tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
     remote_odb = dvc.cloud.get_remote_odb("upstream")
 
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     out = stage.outs[0]
     file_objs = [entry_obj for _, _, entry_obj in out.obj]
 

tests/func/test_remove.py

@@ -53,10 +53,10 @@ def test_remove_non_existent_file(tmp_dir, dvc):
 
 def test_remove_broken_symlink(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
-    dvc.odb.local.cache_types = ["symlink"]
+    dvc.cache.local.cache_types = ["symlink"]
 
     (stage,) = dvc.add("foo")
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
     assert system.is_symlink("foo")
 
     with pytest.raises(ObjectDBError):

tests/func/test_repo.py

@@ -1,11 +1,11 @@
+from dvc.cachemgr import CacheManager
 from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
 from dvc.fs import system
-from dvc.odbmgr import ODBManager
 
 
 def test_destroy(tmp_dir, dvc, run_copy):
     dvc.config["cache"]["type"] = ["symlink"]
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
     tmp_dir.dvc_gen("file", "text")
     tmp_dir.dvc_gen({"dir": {"file": "lorem", "subdir/file": "ipsum"}})

tests/func/test_repro.py

@@ -837,11 +837,11 @@ def test_repro_no_commit(tmp_dir, dvc, copy_script, run_stage):
         cmd="python copy.py foo file1",
         name="run1",
     )
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
     ret = main(["repro", stage.addressing, "--no-commit"])
     assert ret == 0
     # run-cache should be skipped if `-no-commit`.
-    assert not os.path.isdir(dvc.odb.local.path)
+    assert not os.path.isdir(dvc.cache.local.path)
 
 
 class TestReproAlreadyCached:

tests/func/test_run_single_stage.py

@@ -682,12 +682,12 @@ def test_run_commit(dvc):
     )
     assert ret == 0
     assert os.path.isfile(fname)
-    assert not os.path.exists(dvc.odb.local.path)
+    assert not os.path.exists(dvc.cache.local.path)
 
     ret = main(["commit", fname + ".dvc"])
     assert ret == 0
     assert os.path.isfile(fname)
-    assert len(list(dvc.odb.local.all())) == 1
+    assert len(list(dvc.cache.local.all())) == 1
 
 
 @pytest.mark.parametrize(
@@ -789,7 +789,7 @@ def test_should_not_checkout_upon_corrupted_local_hardlink_cache(
     mocker, tmp_dir, dvc, copy_script
 ):
     tmp_dir.gen("foo", "foo")
-    dvc.odb.local.cache_types = ["hardlink"]
+    dvc.cache.local.cache_types = ["hardlink"]
 
     stage = dvc.run(
         deps=["foo"],

tests/func/test_unprotect.py

@@ -4,7 +4,7 @@ import os
 def test_unprotect(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
 
-    dvc.odb.local.cache_types = ["hardlink"]
+    dvc.cache.local.cache_types = ["hardlink"]
     dvc.add("foo")
     cache = os.path.join(".dvc", "cache", "ac", "bd18db4cc2f85cedef654fccc4a4d8")
     assert not os.access("foo", os.W_OK)

tests/integration/conftest.py

@@ -0,0 +1,2 @@
+# pylint disable=unused-argument
+from tests.unit.repo.experiments.conftest import exp_stage, test_queue  # noqa: F401

tests/integration/test_studio_live_experiments.py

@@ -0,0 +1,41 @@
+import pytest
+from dvc_studio_client import env
+from funcy import first
+
+
+@pytest.mark.parametrize("tmp", [True, False])
+def test_post_to_studio(tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch, tmp):
+    valid_response = mocker.MagicMock()
+    valid_response.status_code = 200
+    mocked_post = mocker.patch("requests.post", return_value=valid_response)
+
+    monkeypatch.setenv(env.STUDIO_ENDPOINT, "https://0.0.0.0")
+    monkeypatch.setenv(env.STUDIO_REPO_URL, "STUDIO_REPO_URL")
+    monkeypatch.setenv(env.STUDIO_TOKEN, "STUDIO_TOKEN")
+
+    baseline_sha = scm.get_rev()
+    exp_rev = first(
+        dvc.experiments.run(exp_stage.addressing, params=["foo=1"], tmp_dir=tmp)
+    )
+    name = dvc.experiments.get_exact_name([exp_rev])[exp_rev]
+    assert mocked_post.call_count == 2
+
+    start_call, done_call = mocked_post.call_args_list
+
+    assert start_call.kwargs["json"] == {
+        "type": "start",
+        "repo_url": "STUDIO_REPO_URL",
+        "baseline_sha": baseline_sha,
+        "name": name,
+        "params": {"params.yaml": {"foo": 1}},
+        "client": "dvc",
+    }
+
+    assert done_call.kwargs["json"] == {
+        "type": "done",
+        "repo_url": "STUDIO_REPO_URL",
+        "baseline_sha": baseline_sha,
+        "name": name,
+        "client": "dvc",
+        "experiment_rev": exp_rev,
+    }

tests/unit/command/test_plots.py

@@ -285,7 +285,7 @@ def test_should_call_render(tmp_dir, mocker, capsys, plots_data, output):
     render_mock.assert_called_once_with(
         renderers=renderers,
         output_file=Path(tmp_dir / output / "index.html"),
-        template_path=None,
+        html_template=None,
     )
 
 

tests/unit/data/db/test_local.py

@@ -32,7 +32,7 @@ def test_status_download_optimization(mocker, dvc):
 
 @pytest.mark.parametrize("link_name", ["hardlink", "symlink"])
 def test_is_protected(tmp_dir, dvc, link_name):
-    odb = dvc.odb.local
+    odb = dvc.cache.local
     fs = odb.fs
     link_method = getattr(fs, link_name)
 
@@ -67,7 +67,7 @@ def test_protect_ignore_errors(tmp_dir, dvc, mocker, err):
     tmp_dir.gen("foo", "foo")
 
     mock_chmod = mocker.patch("os.chmod", side_effect=OSError(err, "something"))
-    dvc.odb.local.protect("foo")
+    dvc.cache.local.protect("foo")
     assert mock_chmod.called
 
 
@@ -76,7 +76,7 @@ def test_set_exec_ignore_errors(tmp_dir, dvc, mocker, err):
     tmp_dir.gen("foo", "foo")
 
     mock_chmod = mocker.patch("os.chmod", side_effect=OSError(err, "something"))
-    dvc.odb.local.set_exec("foo")
+    dvc.cache.local.set_exec("foo")
     assert mock_chmod.called
 
 
@@ -88,7 +88,7 @@ def test_staging_file(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
     fs = LocalFileSystem()
 
-    local_odb = dvc.odb.local
+    local_odb = dvc.cache.local
     staging_odb, _, obj = build(local_odb, (tmp_dir / "foo").fs_path, fs, "md5")
 
     assert not local_odb.exists(obj.hash_info.value)
@@ -113,7 +113,7 @@ def test_staging_dir(tmp_dir, dvc):
 
     tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
     fs = LocalFileSystem()
-    local_odb = dvc.odb.local
+    local_odb = dvc.cache.local
 
     staging_odb, _, obj = build(local_odb, (tmp_dir / "dir").fs_path, fs, "md5")
 

tests/unit/fs/test_data.py

@@ -58,7 +58,7 @@ def test_open_dirty_hash(tmp_dir, dvc):
 def test_open_no_remote(tmp_dir, dvc):
     tmp_dir.dvc_gen("file", "file")
     (tmp_dir / "file").unlink()
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     fs = DataFileSystem(index=dvc.index.data["repo"])
     with pytest.raises(FileNotFoundError):
@@ -238,11 +238,11 @@ def test_get_hash_granular(tmp_dir, dvc):
     fs = DataFileSystem(index=dvc.index.data["repo"])
     subdir = "dir/subdir"
     assert fs.info(subdir).get("md5") is None
-    _, _, obj = build(dvc.odb.local, subdir, fs, "md5", dry_run=True)
+    _, _, obj = build(dvc.cache.local, subdir, fs, "md5", dry_run=True)
     assert obj.hash_info == HashInfo("md5", "af314506f1622d107e0ed3f14ec1a3b5.dir")
     data = posixpath.join(subdir, "data")
     assert fs.info(data)["md5"] == "8d777f385d3dfec8815d20f7496026dc"
-    _, _, obj = build(dvc.odb.local, data, fs, "md5", dry_run=True)
+    _, _, obj = build(dvc.cache.local, data, fs, "md5", dry_run=True)
     assert obj.hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc")
 
 
@@ -253,7 +253,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
     fs = DataFileSystem(index=dvc.index.data["repo"])
     expected = "8c7dd922ad47494fc02c388e12c00eac"
     assert fs.info("file").get("md5") == expected
-    _, _, obj = build(dvc.odb.local, "file", fs, "md5", dry_run=True)
+    _, _, obj = build(dvc.cache.local, "file", fs, "md5", dry_run=True)
     assert obj.hash_info == HashInfo("md5", expected)
 
 
@@ -264,5 +264,5 @@ def test_get_hash_dirty_dir(tmp_dir, dvc):
     fs = DataFileSystem(index=dvc.index.data["repo"])
     expected = "5ea40360f5b4ec688df672a4db9c17d1.dir"
     assert fs.info("dir").get("md5") == expected
-    _, _, obj = build(dvc.odb.local, "dir", fs, "md5", dry_run=True)
+    _, _, obj = build(dvc.cache.local, "dir", fs, "md5", dry_run=True)
     assert obj.hash_info == HashInfo("md5", expected)

tests/unit/fs/test_dvc.py

@@ -493,7 +493,7 @@ def test_get_hash_cached_file(tmp_dir, dvc, mocker):
     fs = DVCFileSystem(repo=dvc)
     expected = "acbd18db4cc2f85cedef654fccc4a4d8"
     assert fs.info("foo").get("md5") is None
-    _, _, obj = build(dvc.odb.local, "foo", fs, "md5")
+    _, _, obj = build(dvc.cache.local, "foo", fs, "md5")
     assert obj.hash_info == HashInfo("md5", expected)
     (tmp_dir / "foo").unlink()
     assert fs.info("foo")["md5"] == expected
@@ -504,12 +504,12 @@ def test_get_hash_cached_dir(tmp_dir, dvc, mocker):
     fs = DVCFileSystem(repo=dvc)
     expected = "8761c4e9acad696bee718615e23e22db.dir"
     assert fs.info("dir").get("md5") is None
-    _, _, obj = build(dvc.odb.local, "dir", fs, "md5")
+    _, _, obj = build(dvc.cache.local, "dir", fs, "md5")
     assert obj.hash_info == HashInfo("md5", "8761c4e9acad696bee718615e23e22db.dir")
 
     shutil.rmtree(tmp_dir / "dir")
     assert fs.info("dir")["md5"] == expected
-    _, _, obj = build(dvc.odb.local, "dir", fs, "md5")
+    _, _, obj = build(dvc.cache.local, "dir", fs, "md5")
     assert obj.hash_info == HashInfo("md5", "8761c4e9acad696bee718615e23e22db.dir")
 
 
@@ -518,10 +518,10 @@ def test_get_hash_cached_granular(tmp_dir, dvc, mocker):
     fs = DVCFileSystem(repo=dvc)
     subdir = "dir/subdir"
     assert fs.info(subdir).get("md5") is None
-    _, _, obj = build(dvc.odb.local, subdir, fs, "md5")
+    _, _, obj = build(dvc.cache.local, subdir, fs, "md5")
     assert obj.hash_info == HashInfo("md5", "af314506f1622d107e0ed3f14ec1a3b5.dir")
     assert fs.info(posixpath.join(subdir, "data")).get("md5") is None
-    _, _, obj = build(dvc.odb.local, posixpath.join(subdir, "data"), fs, "md5")
+    _, _, obj = build(dvc.cache.local, posixpath.join(subdir, "data"), fs, "md5")
     assert obj.hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc")
     (tmp_dir / "dir" / "subdir" / "data").unlink()
     assert (
@@ -543,7 +543,7 @@ def test_get_hash_mixed_dir(tmp_dir, scm, dvc):
     tmp_dir.scm.commit("add dir")
 
     fs = DVCFileSystem(repo=dvc)
-    _, _, obj = build(dvc.odb.local, "dir", fs, "md5")
+    _, _, obj = build(dvc.cache.local, "dir", fs, "md5")
     assert obj.hash_info == HashInfo("md5", "e1d9e8eae5374860ae025ec84cfd85c7.dir")
 
 
@@ -561,7 +561,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
     # hash_file(file) should return workspace hash, not DVC cached hash
     fs = DVCFileSystem(repo=dvc)
     assert fs.info("file").get("md5") is None
-    staging, _, obj = build(dvc.odb.local, "file", fs, "md5")
+    staging, _, obj = build(dvc.cache.local, "file", fs, "md5")
     assert obj.hash_info == something_hash_info
     check(staging, obj)
 
@@ -573,7 +573,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
 
     # tmp_dir/file can be built even though it is missing in workspace since
     # repofs will use the DVC cached hash (and refer to the local cache object)
-    _, _, obj = build(dvc.odb.local, "file", fs, "md5")
+    _, _, obj = build(dvc.cache.local, "file", fs, "md5")
     assert obj.hash_info == file_hash_info
 
 
@@ -582,7 +582,7 @@ def test_get_hash_dirty_dir(tmp_dir, dvc):
     (tmp_dir / "dir" / "baz").write_text("baz")
 
     fs = DVCFileSystem(repo=dvc)
-    _, meta, obj = build(dvc.odb.local, "dir", fs, "md5")
+    _, meta, obj = build(dvc.cache.local, "dir", fs, "md5")
     assert obj.hash_info == HashInfo("md5", "ba75a2162ca9c29acecb7957105a0bc2.dir")
     assert meta.nfiles == 3
 

tests/unit/output/test_output.py

@@ -113,7 +113,7 @@ def test_remote_missing_dependency_on_dir_pull(tmp_dir, scm, dvc, mocker):
         conf["core"] = {"remote": "s3"}
 
     remove("dir")
-    remove(dvc.odb.local.path)
+    remove(dvc.cache.local.path)
 
     mocker.patch(
         "dvc.data_cloud.DataCloud.get_remote_odb",

tests/unit/stage/test_cache.py

@@ -146,16 +146,16 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker):
 def test_shared_stage_cache(tmp_dir, dvc, run_copy):
     import stat
 
-    from dvc.odbmgr import ODBManager
+    from dvc.cachemgr import CacheManager
 
     tmp_dir.gen("foo", "foo")
 
     with dvc.config.edit() as config:
         config["cache"]["shared"] = "group"
 
-    dvc.odb = ODBManager(dvc)
+    dvc.cache = CacheManager(dvc)
 
-    assert not os.path.exists(dvc.odb.local.path)
+    assert not os.path.exists(dvc.cache.local.path)
 
     run_copy("foo", "bar", name="copy-foo-bar")
 
@@ -184,7 +184,7 @@ def test_shared_stage_cache(tmp_dir, dvc, run_copy):
         dir_mode = 0o2775
         file_mode = 0o664
 
-    assert _mode(dvc.odb.local.path) == dir_mode
+    assert _mode(dvc.cache.local.path) == dir_mode
     assert _mode(dvc.stage_cache.cache_dir) == dir_mode
     assert _mode(parent_cache_dir) == dir_mode
     assert _mode(cache_dir) == dir_mode

tests/unit/test_external_repo.py

@@ -55,7 +55,7 @@ def test_subrepo_is_constructed_properly(
 
     subrepo = tmp_dir / "subrepo"
     make_subrepo(subrepo, scm)
-    local_cache = subrepo.dvc.odb.local.path
+    local_cache = subrepo.dvc.cache.local.path
 
     tmp_dir.scm_gen("bar", "bar", commit="add bar")
     subrepo.dvc_gen("foo", "foo", commit="add foo")
@@ -72,16 +72,16 @@ def test_subrepo_is_constructed_properly(
 
         assert repo.url == str(tmp_dir)
         assert repo.config["cache"]["dir"] == str(cache_dir)
-        assert repo.odb.local.path == str(cache_dir)
-        assert subrepo.odb.local.path == str(cache_dir)
+        assert repo.cache.local.path == str(cache_dir)
+        assert subrepo.cache.local.path == str(cache_dir)
 
         assert repo.config["cache"]["type"] == ["symlink"]
-        assert repo.odb.local.cache_types == ["symlink"]
-        assert subrepo.odb.local.cache_types == ["symlink"]
+        assert repo.cache.local.cache_types == ["symlink"]
+        assert subrepo.cache.local.cache_types == ["symlink"]
 
         assert subrepo.config["remote"]["auto-generated-upstream"]["url"] == local_cache
         if root_is_dvc:
-            main_cache = tmp_dir.dvc.odb.local.path
+            main_cache = tmp_dir.dvc.cache.local.path
             assert repo.config["remote"]["auto-generated-upstream"]["url"] == str(
                 main_cache
             )

tests/unit/test_info.py

@@ -45,7 +45,7 @@ def find_supported_remotes(string):
 def test_info_in_repo(scm_init, tmp_dir):
     tmp_dir.init(scm=scm_init, dvc=True)
     # Create `.dvc/cache`, that is needed to check supported link types.
-    os.mkdir(tmp_dir.dvc.odb.local.path)
+    os.mkdir(tmp_dir.dvc.cache.local.path)
 
     dvc_info = get_dvc_info()
 
@@ -113,7 +113,7 @@ def test_remotes(tmp_dir, dvc, caplog):
 
 
 def test_fs_info_in_repo(tmp_dir, dvc, caplog):
-    os.mkdir(dvc.odb.local.path)
+    os.mkdir(dvc.cache.local.path)
     dvc_info = get_dvc_info()
 
     assert re.search(r"Cache directory: .* on .*", dvc_info)

tests/unit/test_logger.py

@@ -33,7 +33,7 @@ class TestColorFormatter:
         with caplog.at_level(logging.DEBUG, logger="dvc"):
             logger.debug("message")
 
-            expected = "{green}{datetime}{nc} {blue}DEBUG{nc}: message".format(
+            expected = "{blue}{datetime}{nc} {blue}DEBUG{nc}: message".format(
                 **colors, datetime=dt
             )
 
@@ -103,12 +103,9 @@ class TestColorFormatter:
                 logger.exception("")
 
             expected = (
-                "{green}{datetime}{nc} "
+                "{red}{datetime}{nc} "
                 "{red}ERROR{nc}: description\n"
-                "{red}{line}{nc}\n"
-                "{stack_trace}"
-                "{red}{line}{nc}".format(
-                    line="-" * 60,
+                "{stack_trace}".format(
                     stack_trace=stack_trace,
                     **colors,
                     datetime=dt,
@@ -126,12 +123,9 @@ class TestColorFormatter:
                 logger.debug("", exc_info=True)
 
             expected = (
-                "{green}{datetime}{nc} "
+                "{blue}{datetime}{nc} "
                 "{blue}DEBUG{nc}: description\n"
-                "{red}{line}{nc}\n"
-                "{stack_trace}"
-                "{red}{line}{nc}".format(
-                    line="-" * 60,
+                "{stack_trace}".format(
                     stack_trace=stack_trace,
                     datetime=dt,
                     **colors,
@@ -149,12 +143,9 @@ class TestColorFormatter:
                 logger.exception("something", extra={"tb_only": True})
 
             expected = (
-                "{green}{datetime}{nc} "
+                "{red}{datetime}{nc} "
                 "{red}ERROR{nc}: something\n"
-                "{red}{line}{nc}\n"
-                "{stack_trace}"
-                "{red}{line}{nc}".format(
-                    line="-" * 60,
+                "{stack_trace}".format(
                     stack_trace=stack_trace,
                     **colors,
                     datetime=dt,
@@ -175,12 +166,9 @@ class TestColorFormatter:
                     logger.exception("message")
 
             expected = (
-                "{green}{datetime}{nc} "
+                "{red}{datetime}{nc} "
                 "{red}ERROR{nc}: message - second: first\n"
-                "{red}{line}{nc}\n"
-                "{stack_trace}"
-                "{red}{line}{nc}".format(
-                    line="-" * 60,
+                "{stack_trace}".format(
                     stack_trace=stack_trace,
                     **colors,
                     datetime=dt,