schema: add strict schema validation for top-level plots (#8289)

schema: add strict schema validation for plots

dvc/annotations.py

@@ -0,0 +1,35 @@
+from dataclasses import asdict, dataclass, field, fields
+from typing import Any, ClassVar, Dict, List, Optional
+
+from funcy import compact
+
+
+@dataclass
+class Annotation:
+    PARAM_DESC: ClassVar[str] = "desc"
+    PARAM_TYPE: ClassVar[str] = "type"
+    PARAM_LABELS: ClassVar[str] = "labels"
+    PARAM_META: ClassVar[str] = "meta"
+
+    desc: Optional[str] = None
+    type: Optional[str] = None
+    labels: List[str] = field(default_factory=list)
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+    def update(self, **kwargs) -> "Annotation":
+        for attr, value in kwargs.items():
+            if value and hasattr(self, attr):
+                setattr(self, attr, value)
+        return self
+
+    def to_dict(self) -> Dict[str, str]:
+        return compact(asdict(self))
+
+
+ANNOTATION_FIELDS = [field.name for field in fields(Annotation)]
+ANNOTATION_SCHEMA = {
+    Annotation.PARAM_DESC: str,
+    Annotation.PARAM_TYPE: str,
+    Annotation.PARAM_LABELS: [str],
+    Annotation.PARAM_META: object,
+}

dvc/api/__init__.py

@@ -1,9 +1,15 @@
-from .data import (  # noqa, pylint: disable=redefined-builtin
-    get_url,
-    open,
-    read,
-)
+from dvc.fs.dvc import _DvcFileSystem as DvcFileSystem
+
+from .data import open  # pylint: disable=redefined-builtin
+from .data import get_url, read
 from .experiments import make_checkpoint
 from .params import params_show
 
-__all__ = ["get_url", "make_checkpoint", "open", "params_show", "read"]
+__all__ = [
+    "get_url",
+    "make_checkpoint",
+    "open",
+    "params_show",
+    "read",
+    "DvcFileSystem",
+]

dvc/cli/__init__.py

@@ -121,7 +121,7 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
         )
         return None
 
-    from dvc_data.build import IgnoreInCollectedDirError
+    from dvc_data.hashfile.build import IgnoreInCollectedDirError
 
     if isinstance(exc, IgnoreInCollectedDirError):
         logger.exception("")

dvc/cli/actions.py

@@ -0,0 +1,20 @@
+from argparse import _AppendAction
+
+
+class CommaSeparatedArgs(_AppendAction):  # pylint: disable=protected-access
+    def __call__(self, parser, namespace, values, option_string=None):
+        from funcy import ldistinct
+
+        items = getattr(namespace, self.dest) or []
+        items.extend(map(str.strip, values.split(",")))
+        setattr(namespace, self.dest, ldistinct(items))
+
+
+class KeyValueArgs(_AppendAction):
+    def __call__(self, parser, namespace, values, option_string=None):
+        items = getattr(namespace, self.dest) or {}
+        for value in filter(bool, values):
+            key, _, value = value.partition("=")
+            items[key.strip()] = value
+
+        setattr(namespace, self.dest, items)

dvc/commands/add.py

@@ -2,12 +2,43 @@ import argparse
 import logging
 
 from dvc.cli import completion
+from dvc.cli.actions import KeyValueArgs
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 
 logger = logging.getLogger(__name__)
 
 
+def _add_annotating_args(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--desc",
+        type=str,
+        metavar="<text>",
+        help="User description of the data.",
+    )
+    parser.add_argument(
+        "--meta",
+        metavar="key=value",
+        nargs=1,
+        action=KeyValueArgs,
+        help="Custom metadata to add to the data.",
+    )
+    parser.add_argument(
+        "--label",
+        dest="labels",
+        type=str,
+        action="append",
+        metavar="<str>",
+        help="Label for the data.",
+    )
+    parser.add_argument(
+        "--type",
+        type=str,
+        metavar="<str>",
+        help="Type of the data.",
+    )
+
+
 class CmdAdd(CmdBase):
     def run(self):
         from dvc.exceptions import (
@@ -28,6 +59,9 @@ class CmdAdd(CmdBase):
                 glob=self.args.glob,
                 desc=self.args.desc,
                 out=self.args.out,
+                type=self.args.type,
+                labels=self.args.labels,
+                meta=self.args.meta,
                 remote=self.args.remote,
                 to_remote=self.args.to_remote,
                 jobs=self.args.jobs,
@@ -109,15 +143,8 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
-    parser.add_argument(
-        "--desc",
-        type=str,
-        metavar="<text>",
-        help=(
-            "User description of the data (optional). "
-            "This doesn't affect any DVC operations."
-        ),
-    )
+
+    _add_annotating_args(parser)
     parser.add_argument(
         "targets", nargs="+", help="Input files/directories to add."
     ).complete = completion.FILE

dvc/commands/data.py

@@ -1,9 +1,12 @@
 import argparse
 import logging
-from typing import TYPE_CHECKING
+from operator import itemgetter
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Set
 
 from funcy import chunks, compact, log_durations
 
+from dvc.cli import completion
+from dvc.cli.actions import CommaSeparatedArgs
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
 from dvc.ui import ui
@@ -124,6 +127,65 @@ class CmdDataStatus(CmdBase):
         return self._show_status(status)
 
 
+class CmdDataLs(CmdBase):
+    @staticmethod
+    def _show_table(
+        d: Iterable[Dict[str, Any]],
+        filter_types: Set[str],
+        filter_labels: Set[str],
+        markdown: bool = False,
+    ) -> None:
+        from rich.style import Style
+
+        from dvc.compare import TabularData
+
+        td = TabularData(
+            columns=["Path", "Type", "Labels", "Description"], fill_value="-"
+        )
+        for entry in sorted(d, key=itemgetter("path")):
+            typ = entry.get("type", "")
+            desc = entry.get("desc", "-")
+            labels = entry.get("labels", [])
+
+            if filter_types and typ not in filter_types:
+                continue
+            if filter_labels and filter_labels.isdisjoint(labels):
+                continue
+
+            rich_label = ui.rich_text()
+            for index, label in enumerate(labels):
+                if index:
+                    rich_label.append(",")
+                style = Style(bold=label in filter_labels, color="green")
+                rich_label.append(label, style=style)
+
+            if markdown and desc:
+                desc = desc.partition("\n")[0]
+
+            path = ui.rich_text(entry["path"], style="cyan")
+            type_style = Style(bold=typ in filter_types, color="yellow")
+            typ = ui.rich_text(entry.get("type", ""), style=type_style)
+            td.append([path, typ or "-", rich_label or "-", desc])
+
+        td.render(markdown=markdown, rich_table=True)
+
+    def run(self):
+        from dvc.repo.data import ls
+
+        filter_labels = set(self.args.labels)
+        filter_types = set(self.args.type)
+        d = ls(
+            self.repo, targets=self.args.targets, recursive=self.args.recursive
+        )
+        self._show_table(
+            d,
+            filter_labels=filter_labels,
+            filter_types=filter_types,
+            markdown=self.args.markdown,
+        )
+        return 0
+
+
 def add_parser(subparsers, parent_parser):
     data_parser = subparsers.add_parser(
         "data",
@@ -181,3 +243,51 @@ def add_parser(subparsers, parent_parser):
         help="Show untracked files.",
     )
     data_status_parser.set_defaults(func=CmdDataStatus)
+
+    DATA_LS_HELP = "List data tracked by DVC with its metadata."
+    data_ls_parser = data_subparsers.add_parser(
+        "ls",
+        aliases=["list"],
+        parents=[parent_parser],
+        description=append_doc_link(DATA_LS_HELP, "data/ls"),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        help=DATA_LS_HELP,
+    )
+    data_ls_parser.add_argument(
+        "--md",
+        "--show-md",
+        dest="markdown",
+        action="store_true",
+        default=False,
+        help="Show tabulated output in the Markdown format (GFM).",
+    )
+    data_ls_parser.add_argument(
+        "--type",
+        action=CommaSeparatedArgs,
+        default=[],
+        help="Comma-separated list of type to filter.",
+    )
+    data_ls_parser.add_argument(
+        "--labels",
+        action=CommaSeparatedArgs,
+        default=[],
+        help="Comma-separated list of labels to filter.",
+    )
+    data_ls_parser.add_argument(
+        "-R",
+        "--recursive",
+        action="store_true",
+        default=False,
+        help="Recursively list from the specified directories.",
+    )
+    data_ls_parser.add_argument(
+        "targets",
+        default=None,
+        nargs="*",
+        help=(
+            "Limit command scope to these tracked files/directories, "
+            ".dvc files, or stage names."
+        ),
+    ).complete = completion.DVCFILES_AND_STAGE
+
+    data_ls_parser.set_defaults(func=CmdDataLs)

dvc/commands/imp.py

@@ -21,6 +21,9 @@ class CmdImport(CmdBase):
                 no_exec=self.args.no_exec,
                 no_download=self.args.no_download,
                 desc=self.args.desc,
+                type=self.args.type,
+                labels=self.args.labels,
+                meta=self.args.meta,
                 jobs=self.args.jobs,
             )
         except DvcException:
@@ -34,6 +37,8 @@ class CmdImport(CmdBase):
 
 
 def add_parser(subparsers, parent_parser):
+    from .add import _add_annotating_args
+
     IMPORT_HELP = (
         "Download file or directory tracked by DVC or by Git "
         "into the workspace, and track it."
@@ -84,15 +89,6 @@ def add_parser(subparsers, parent_parser):
         help="Create .dvc file including target data hash value(s)"
         " but do not actually download the file(s).",
     )
-    import_parser.add_argument(
-        "--desc",
-        type=str,
-        metavar="<text>",
-        help=(
-            "User description of the data (optional). "
-            "This doesn't affect any DVC operations."
-        ),
-    )
     import_parser.add_argument(
         "-j",
         "--jobs",
@@ -103,4 +99,6 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+
+    _add_annotating_args(import_parser)
     import_parser.set_defaults(func=CmdImport)

dvc/commands/imp_url.py

@@ -21,6 +21,9 @@ class CmdImportUrl(CmdBase):
                 remote=self.args.remote,
                 to_remote=self.args.to_remote,
                 desc=self.args.desc,
+                type=self.args.type,
+                labels=self.args.labels,
+                meta=self.args.meta,
                 jobs=self.args.jobs,
             )
         except DvcException:
@@ -35,6 +38,8 @@ class CmdImportUrl(CmdBase):
 
 
 def add_parser(subparsers, parent_parser):
+    from .add import _add_annotating_args
+
     IMPORT_HELP = (
         "Download or copy file from URL and take it under DVC control."
     )
@@ -103,13 +108,5 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
-    import_parser.add_argument(
-        "--desc",
-        type=str,
-        metavar="<text>",
-        help=(
-            "User description of the data (optional). "
-            "This doesn't affect any DVC operations."
-        ),
-    )
+    _add_annotating_args(import_parser)
     import_parser.set_defaults(func=CmdImportUrl)

dvc/commands/stage.py

@@ -76,7 +76,7 @@ class CmdStageList(CmdBase):
         # removing duplicates while maintaining order
         collected = chain.from_iterable(
             self.repo.stage.collect(
-                target=target, recursive=self.args.recursive, accept_group=True
+                target=target, recursive=self.args.recursive
             )
             for target in self.args.targets
         )

dvc/config_schema.py

@@ -78,7 +78,9 @@ REMOTE_COMMON = {
     "url": str,
     "checksum_jobs": All(Coerce(int), Range(1)),
     "jobs": All(Coerce(int), Range(1)),
+    Optional("worktree"): Bool,
     Optional("no_traverse"): Bool,  # obsoleted
+    Optional("version_aware"): Bool,
 }
 LOCAL_COMMON = {
     "type": supported_cache_type,

dvc/data_cloud.py

@@ -3,16 +3,37 @@
 import logging
 from typing import TYPE_CHECKING, Iterable, Optional
 
-from dvc_data.db import get_index
+from funcy import cached_property
+
+from dvc_data.hashfile.db import get_index
 
 if TYPE_CHECKING:
     from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
-    from dvc_data.status import CompareStatusResult
+    from dvc_data.hashfile.status import CompareStatusResult
 
 logger = logging.getLogger(__name__)
 
 
+class Remote:
+    def __init__(self, path, fs, **config):
+        self.path = path
+        self.fs = fs
+
+        self.worktree = config.pop("worktree", False)
+        self.config = config
+
+    @cached_property
+    def odb(self):
+        from dvc_data.hashfile.db import get_odb
+
+        path = self.path
+        if self.worktree:
+            path = self.fs.path.join(path, ".dvc", "cache")
+
+        return get_odb(self.fs, path, **self.config)
+
+
 class DataCloud:
     """Class that manages dvc remotes.
 
@@ -27,18 +48,23 @@ class DataCloud:
     def __init__(self, repo):
         self.repo = repo
 
-    def get_remote_odb(
+    def get_remote(
         self,
         name: Optional[str] = None,
         command: str = "<command>",
-    ) -> "HashFileDB":
+    ) -> "Remote":
         from dvc.config import NoRemoteError
 
         if not name:
             name = self.repo.config["core"].get("remote")
 
         if name:
-            return self._init_odb(name)
+            from dvc.fs import get_cloud_fs
+
+            cls, config, fs_path = get_cloud_fs(self.repo, name=name)
+            fs = cls(**config)
+            config["tmp_dir"] = self.repo.index_db_dir
+            return Remote(fs_path, fs, **config)
 
         if bool(self.repo.config["remote"]):
             error_msg = (
@@ -55,13 +81,13 @@ class DataCloud:
 
         raise NoRemoteError(error_msg)
 
-    def _init_odb(self, name):
-        from dvc.fs import get_cloud_fs
-        from dvc_data.db import get_odb
-
-        cls, config, fs_path = get_cloud_fs(self.repo, name=name)
-        config["tmp_dir"] = self.repo.index_db_dir
-        return get_odb(cls(**config), fs_path, **config)
+    def get_remote_odb(
+        self,
+        name: Optional[str] = None,
+        command: str = "<command>",
+    ) -> "HashFileDB":
+        remote = self.get_remote(name=name, command=command)
+        return remote.odb
 
     def _log_missing(self, status: "CompareStatusResult"):
         if status.missing:
@@ -82,7 +108,7 @@ class DataCloud:
         **kwargs,
     ):
         from dvc.exceptions import FileTransferError
-        from dvc_data.transfer import TransferError, transfer
+        from dvc_data.hashfile.transfer import TransferError, transfer
 
         try:
             return transfer(src_odb, dest_odb, objs, **kwargs)
@@ -164,7 +190,7 @@ class DataCloud:
             log_missing: log warning messages if file doesn't exist
                 neither in cache, neither in cloud.
         """
-        from dvc_data.status import compare_status
+        from dvc_data.hashfile.status import compare_status
 
         if not odb:
             odb = self.get_remote_odb(remote, "status")

dvc/dependency/__init__.py

@@ -18,7 +18,7 @@ SCHEMA: Mapping[str, Any] = {
 }
 
 
-def _get(stage, p, info):
+def _get(stage, p, info, fs_config=None):
     if info and info.get(RepoDependency.PARAM_REPO):
         repo = info.pop(RepoDependency.PARAM_REPO)
         return RepoDependency(repo, stage, p, info)
@@ -27,7 +27,7 @@ def _get(stage, p, info):
         params = info.pop(ParamsDependency.PARAM_PARAMS)
         return ParamsDependency(stage, p, params)
 
-    return Dependency(stage, p, info)
+    return Dependency(stage, p, info, fs_config=fs_config)
 
 
 def loadd_from(stage, d_list):
@@ -38,10 +38,18 @@ def loadd_from(stage, d_list):
     return ret
 
 
-def loads_from(stage, s_list, erepo=None):
+def loads_from(stage, s_list, erepo=None, fs_config=None):
     assert isinstance(s_list, list)
     info = {RepoDependency.PARAM_REPO: erepo} if erepo else {}
-    return [_get(stage, s, info.copy()) for s in s_list]
+    return [
+        _get(
+            stage,
+            s,
+            info.copy(),
+            fs_config=fs_config,
+        )
+        for s in s_list
+    ]
 
 
 def _merge_params(s_list):

dvc/dependency/param.py

@@ -39,14 +39,15 @@ class ParamsDependency(Dependency):
 
     def __init__(self, stage, path, params=None, repo=None):
         self.params = list(params) if params else []
-        info = (
-            {self.PARAM_PARAMS: params} if isinstance(params, dict) else None
-        )
+        hash_info = HashInfo()
+        if isinstance(params, dict):
+            hash_info = HashInfo(self.PARAM_PARAMS, params)
         repo = repo or stage.repo
         path = path or os.path.join(repo.root_dir, self.DEFAULT_PARAMS_FILE)
-        super().__init__(stage, path, info=info, repo=repo)
+        super().__init__(stage, path, repo=repo)
+        self.hash_info = hash_info
 
-    def dumpd(self):
+    def dumpd(self, **kwargs):
         ret = super().dumpd()
         if not self.hash_info:
             ret[self.PARAM_PARAMS] = self.params or {}

dvc/dependency/repo.py

@@ -59,11 +59,11 @@ class RepoDependency(Dependency):
     def save(self):
         pass
 
-    def dumpd(self):
+    def dumpd(self, **kwargs):
         return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
 
     def download(self, to, jobs=None):
-        from dvc_data.checkout import checkout
+        from dvc_data.hashfile.checkout import checkout
 
         for odb, objs in self.get_used_objs().items():
             self.repo.cloud.pull(objs, jobs=jobs, odb=odb)
@@ -103,8 +103,8 @@ class RepoDependency(Dependency):
         from dvc.config import NoRemoteError
         from dvc.exceptions import NoOutputOrStageError, PathMissingError
         from dvc.utils import as_posix
-        from dvc_data.build import build
-        from dvc_data.objects.tree import Tree, TreeError
+        from dvc_data.hashfile.build import build
+        from dvc_data.hashfile.tree import Tree, TreeError
 
         local_odb = self.repo.odb.local
         locked = kwargs.pop("locked", True)
@@ -153,8 +153,8 @@ class RepoDependency(Dependency):
     def _check_circular_import(self, odb, obj_ids):
         from dvc.exceptions import CircularImportError
         from dvc.fs.dvc import DvcFileSystem
-        from dvc_data.db.reference import ReferenceHashFileDB
-        from dvc_data.objects.tree import Tree
+        from dvc_data.hashfile.db.reference import ReferenceHashFileDB
+        from dvc_data.hashfile.tree import Tree
 
         if not isinstance(odb, ReferenceHashFileDB):
             return

dvc/dvcfile.py

@@ -198,7 +198,7 @@ class SingleStageFile(FileMixin):
         if self.verify:
             check_dvcfile_path(self.repo, self.path)
         logger.debug("Saving information to '%s'.", relpath(self.path))
-        dump_yaml(self.path, serialize.to_single_stage_file(stage))
+        dump_yaml(self.path, serialize.to_single_stage_file(stage, **kwargs))
         self.repo.scm_context.track_file(self.relpath)
 
     def remove_stage(self, stage):  # pylint: disable=unused-argument
@@ -243,9 +243,11 @@ class PipelineFile(FileMixin):
         self._lockfile.dump(stage)
 
     @staticmethod
-    def _check_if_parametrized(stage):
+    def _check_if_parametrized(stage, action: str = "dump") -> None:
         if stage.raw_data.parametrized:
-            raise ParametrizedDumpError(f"cannot dump a parametrized {stage}")
+            raise ParametrizedDumpError(
+                f"cannot {action} a parametrized {stage}"
+            )
 
     def _dump_pipeline_file(self, stage):
         self._check_if_parametrized(stage)
@@ -291,6 +293,7 @@ class PipelineFile(FileMixin):
         self._lockfile.remove()
 
     def remove_stage(self, stage):
+        self._check_if_parametrized(stage, "remove")
         self._lockfile.remove_stage(stage)
         if not self.exists():
             return

dvc/fs/dvc.py

@@ -4,8 +4,7 @@ import os
 import posixpath
 import threading
 from contextlib import suppress
-from itertools import takewhile
-from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Tuple, Type, Union
 
 from fsspec.spec import AbstractFileSystem
 from funcy import cached_property, wrap_prop, wrap_with
@@ -29,11 +28,6 @@ def as_posix(path: str) -> str:
     return path.replace(ntpath.sep, posixpath.sep)
 
 
-def _wrap_walk(dvc_fs, *args, **kwargs):
-    for root, dnames, fnames in dvc_fs.walk(*args, **kwargs):
-        yield dvc_fs.path.join(dvc_fs.repo.root_dir, root), dnames, fnames
-
-
 # NOT the same as dvc.dvcfile.is_dvc_file()!
 def _is_dvc_file(fname):
     from dvc.dvcfile import is_valid_filename
@@ -42,20 +36,6 @@ def _is_dvc_file(fname):
     return is_valid_filename(fname) or fname == DvcIgnore.DVCIGNORE_FILE
 
 
-def _ls(fs, path):
-    dnames = []
-    fnames = []
-
-    for entry in fs.ls(path, detail=True):
-        name = fs.path.name(entry["name"])
-        if entry["type"] == "directory":
-            dnames.append(name)
-        else:
-            fnames.append(name)
-
-    return dnames, fnames
-
-
 def _merge_info(repo, fs_info, dvc_info):
     from . import utils
 
@@ -79,6 +59,10 @@ def _merge_info(repo, fs_info, dvc_info):
     return ret
 
 
+def _get_dvc_path(dvc_fs, subkey):
+    return dvc_fs.path.join(*subkey) if subkey else ""
+
+
 class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
     """DVC + git-tracked files fs.
 
@@ -91,28 +75,25 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
 
     root_marker = "/"
 
-    PARAM_REPO_URL = "repo_url"
-    PARAM_REPO_ROOT = "repo_root"
-    PARAM_REV = "rev"
-    PARAM_CACHE_DIR = "cache_dir"
-    PARAM_CACHE_TYPES = "cache_types"
-    PARAM_SUBREPOS = "subrepos"
-
     def __init__(
         self,
+        url: Optional[str] = None,
+        rev: Optional[str] = None,
         repo: Optional["Repo"] = None,
-        subrepos=False,
-        repo_factory: RepoFactory = None,
-        **kwargs,
-    ):
-        super().__init__()
-
+        subrepos: bool = False,
+        repo_factory: Optional[RepoFactory] = None,
+        **repo_kwargs: Any,
+    ) -> None:
         from pygtrie import Trie
 
+        super().__init__()
         if repo is None:
-            repo, repo_factory = self._repo_from_fs_config(
-                subrepos=subrepos, **kwargs
+            repo = self._make_repo(
+                url=url, rev=rev, subrepos=subrepos, **repo_kwargs
             )
+            assert repo
+            # pylint: disable=protected-access
+            repo_factory = repo._fs_conf["repo_factory"]
 
         if not repo_factory:
             from dvc.repo import Repo
@@ -167,64 +148,12 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
             return None
         return self.repo.url
 
-    @property
-    def config(self):
-        return {
-            self.PARAM_REPO_URL: self.repo_url,
-            self.PARAM_REPO_ROOT: self.repo.root_dir,
-            self.PARAM_REV: getattr(self.repo.fs, "rev", None),
-            self.PARAM_CACHE_DIR: os.path.abspath(self.repo.odb.local.path),
-            self.PARAM_CACHE_TYPES: self.repo.odb.local.cache_types,
-            self.PARAM_SUBREPOS: self._traverse_subrepos,
-        }
-
     @classmethod
-    def _repo_from_fs_config(
-        cls, **config
-    ) -> Tuple["Repo", Optional["RepoFactory"]]:
-        from dvc.external_repo import erepo_factory, external_repo
+    def _make_repo(cls, **kwargs) -> "Repo":
         from dvc.repo import Repo
 
-        url = config.get(cls.PARAM_REPO_URL)
-        root = config.get(cls.PARAM_REPO_ROOT)
-        assert url or root
-
-        def _open(*args, **kwargs):
-            # NOTE: if original repo was an erepo (and has a URL),
-            # we cannot use Repo.open() since it will skip erepo
-            # cache/remote setup for local URLs
-            if url is None:
-                return Repo.open(*args, **kwargs)
-            return external_repo(*args, **kwargs)
-
-        cache_dir = config.get(cls.PARAM_CACHE_DIR)
-        cache_config = (
-            {}
-            if not cache_dir
-            else {
-                "cache": {
-                    "dir": cache_dir,
-                    "type": config.get(cls.PARAM_CACHE_TYPES),
-                }
-            }
-        )
-        repo_kwargs: dict = {
-            "rev": config.get(cls.PARAM_REV),
-            "subrepos": config.get(cls.PARAM_SUBREPOS, False),
-            "uninitialized": True,
-        }
-        factory: Optional["RepoFactory"] = None
-        if url is None:
-            repo_kwargs["config"] = cache_config
-        else:
-            repo_kwargs["cache_dir"] = cache_dir
-            factory = erepo_factory(url, root, cache_config)
-
-        with _open(
-            url if url else root,
-            **repo_kwargs,
-        ) as repo:
-            return repo, factory
+        with Repo.open(uninitialized=True, **kwargs) as repo:
+            return repo
 
     def _get_repo(self, key: Key) -> "Repo":
         """Returns repo that the path falls in, using prefix.
@@ -239,24 +168,16 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
             return repo
 
         prefix_key, repo = self._subrepos_trie.longest_prefix(key)
-        prefix = self.repo.fs.path.join(
-            self.repo.root_dir,
-            *prefix_key,  # pylint: disable=not-an-iterable
-        )
-
-        path = self._from_key(key)
-        parents = (parent for parent in self.repo.fs.path.parents(path))
-        dirs = [path] + list(takewhile(lambda p: p != prefix, parents))
-        dirs.reverse()
-        self._update(dirs, starting_repo=repo)
+        dir_keys = (key[:i] for i in range(len(prefix_key) + 1, len(key) + 1))
+        self._update(dir_keys, starting_repo=repo)
         return self._subrepos_trie.get(key) or self.repo
 
     @wrap_with(threading.Lock())
-    def _update(self, dirs, starting_repo):
+    def _update(self, dir_keys, starting_repo):
         """Checks for subrepo in directories and updates them."""
         repo = starting_repo
-        for d in dirs:
-            key = self._get_key(d)
+        for key in dir_keys:
+            d = self._from_key(key)
             if self._is_dvc_repo(d):
                 repo = self.repo_factory(
                     d,
@@ -279,75 +200,64 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         repo_path = self.repo.fs.path.join(dir_path, Repo.DVC_DIR)
         return self.repo.fs.isdir(repo_path)
 
-    def _get_fs_pair(
-        self, path
-    ) -> Tuple[
-        Optional["Repo"],
-        Optional[FileSystem],
-        Optional[str],
-        Optional[DataFileSystem],
-        Optional[str],
-    ]:
+    def _get_subrepo_info(
+        self, key: Key
+    ) -> Tuple["Repo", Optional[DataFileSystem], Key]:
         """
-        Returns a pair of fss based on repo the path falls in, using prefix.
+        Returns information about the subrepo the key is part of.
         """
-        key = self._get_key_from_relative(path)
-
-        fs_path = self._from_key(key)
         repo = self._get_repo(key)
-        fs = repo.fs
+        repo_key: Key
         if repo is self.repo:
-            dvc_parts = key
-            dvc_fs = self._datafss.get(())
+            repo_key = ()
+            subkey = key
         else:
-            repo_parts = fs.path.relparts(repo.root_dir, self.repo.root_dir)
-            dvc_parts = key[len(repo_parts) :]
-            key = self._get_key(repo.root_dir)
-            dvc_fs = self._datafss.get(key)
+            repo_key = self._get_key(repo.root_dir)
+            subkey = key[len(repo_key) :]
 
-        if dvc_fs:
-            dvc_path = dvc_fs.path.join(*dvc_parts) if dvc_parts else ""
-        else:
-            dvc_path = None
+        dvc_fs = self._datafss.get(repo_key)
+        return repo, dvc_fs, subkey
 
-        return repo, fs, fs_path, dvc_fs, dvc_path
-
-    def open(
-        self, path, mode="r", encoding="utf-8", **kwargs
+    def _open(
+        self, path, mode="rb", **kwargs
     ):  # pylint: disable=arguments-renamed, arguments-differ
-        if "b" in mode:
-            encoding = None
-
-        _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
+        key = self._get_key_from_relative(path)
+        fs_path = self._from_key(key)
         try:
-            return fs.open(fs_path, mode=mode, encoding=encoding)
+            return self.repo.fs.open(fs_path, mode=mode)
         except FileNotFoundError:
+            _, dvc_fs, subkey = self._get_subrepo_info(key)
             if not dvc_fs:
                 raise
 
-        return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs)
+        dvc_path = _get_dvc_path(dvc_fs, subkey)
+        return dvc_fs.open(dvc_path, mode=mode)
 
     def isdvc(self, path, **kwargs):
-        _, _, _, dvc_fs, dvc_path = self._get_fs_pair(path)
+        key = self._get_key_from_relative(path)
+        _, dvc_fs, subkey = self._get_subrepo_info(key)
+        dvc_path = _get_dvc_path(dvc_fs, subkey)
         return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs)
 
     def ls(  # pylint: disable=arguments-differ
         self, path, detail=True, dvc_only=False, **kwargs
     ):
-        repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
-
-        dvcignore = repo.dvcignore
-        ignore_subrepos = kwargs.get("ignore_subrepos", True)
+        key = self._get_key_from_relative(path)
+        repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
         names = set()
         if dvc_fs:
             with suppress(FileNotFoundError):
+                dvc_path = _get_dvc_path(dvc_fs, subkey)
                 for entry in dvc_fs.ls(dvc_path, detail=False):
                     names.add(dvc_fs.path.name(entry))
 
-        if not dvc_only and fs:
+        ignore_subrepos = kwargs.get("ignore_subrepos", True)
+        if not dvc_only:
+            fs = self.repo.fs
+            fs_path = self._from_key(key)
             try:
-                for entry in dvcignore.ls(
+                for entry in repo.dvcignore.ls(
                     fs, fs_path, detail=False, ignore_subrepos=ignore_subrepos
                 ):
                     names.add(fs.path.name(entry))
@@ -362,8 +272,14 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         paths = []
         for name in names:
             entry_path = self.path.join(path, name)
+            entry_key = key + (name,)
             try:
-                info = self.info(entry_path, ignore_subrepos=ignore_subrepos)
+                info = self._info(
+                    entry_key,
+                    entry_path,
+                    ignore_subrepos=ignore_subrepos,
+                    check_ignored=False,
+                )
             except FileNotFoundError:
                 continue
             infos.append(info)
@@ -377,45 +293,52 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
     def get_file(  # pylint: disable=arguments-differ
         self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs
     ):
-        _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath)
-
-        if fs:
-            try:
-                fs.get_file(fs_path, lpath, callback=callback, **kwargs)
-                return
-            except FileNotFoundError:
-                if not dvc_fs:
-                    raise
+        key = self._get_key_from_relative(rpath)
+        fs_path = self._from_key(key)
+        fs = self.repo.fs
+        try:
+            fs.get_file(fs_path, lpath, callback=callback, **kwargs)
+            return
+        except FileNotFoundError:
+            _, dvc_fs, subkey = self._get_subrepo_info(key)
+            if not dvc_fs:
+                raise
+        dvc_path = _get_dvc_path(dvc_fs, subkey)
         dvc_fs.get_file(dvc_path, lpath, callback=callback, **kwargs)
 
     def info(self, path, **kwargs):
-        repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
-
-        dvcignore = repo.dvcignore
+        key = self._get_key_from_relative(path)
         ignore_subrepos = kwargs.get("ignore_subrepos", True)
+        return self._info(key, path, ignore_subrepos=ignore_subrepos)
+
+    def _info(self, key, path, ignore_subrepos=True, check_ignored=True):
+        repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
         dvc_info = None
         if dvc_fs:
             try:
-                dvc_info = dvc_fs.info(dvc_path)
+                dvc_info = dvc_fs.fs.index.info(subkey)
+                dvc_path = _get_dvc_path(dvc_fs, subkey)
+                dvc_info["name"] = dvc_path
             except FileNotFoundError:
                 pass
 
         fs_info = None
-        if fs:
-            try:
-                fs_info = fs.info(fs_path)
-                if dvcignore.is_ignored(
-                    fs, fs_path, ignore_subrepos=ignore_subrepos
-                ):
-                    fs_info = None
-            except (FileNotFoundError, NotADirectoryError):
-                if not dvc_info:
-                    raise
+        fs = self.repo.fs
+        fs_path = self._from_key(key)
+        try:
+            fs_info = fs.info(fs_path)
+            if check_ignored and repo.dvcignore.is_ignored(
+                fs, fs_path, ignore_subrepos=ignore_subrepos
+            ):
+                fs_info = None
+        except (FileNotFoundError, NotADirectoryError):
+            if not dvc_info:
+                raise
 
         # NOTE: if some parent in fs_path turns out to be a file, it means
         # that the whole repofs branch doesn't exist.
-        if fs and not fs_info and dvc_info:
+        if dvc_info and not fs_info:
             for parent in fs.path.parents(fs_path):
                 try:
                     if fs.info(parent)["type"] != "directory":
@@ -431,14 +354,6 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         info["name"] = path
         return info
 
-    def checksum(self, path):
-        _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
-
-        try:
-            return fs.checksum(fs_path)
-        except FileNotFoundError:
-            return dvc_fs.checksum(dvc_path)
-
 
 class DvcFileSystem(FileSystem):
     protocol = "local"
@@ -467,10 +382,6 @@ class DvcFileSystem(FileSystem):
     def repo_url(self):
         return self.fs.repo_url
 
-    @property
-    def config(self):
-        return self.fs.config
-
     def from_os_path(self, path):
         if os.path.isabs(path):
             path = os.path.relpath(path, self.repo.root_dir)

dvc/fs/git.py

@@ -15,6 +15,7 @@ class GitFileSystem(FileSystem):  # pylint:disable=abstract-method
     """Proxies the repo file access methods to Git objects"""
 
     protocol = "local"
+    PARAM_CHECKSUM = "md5"
 
     def __init__(
         self,

dvc/odbmgr.py

@@ -1,5 +1,5 @@
 from dvc.fs import GitFileSystem, Schemes
-from dvc_data.db import get_odb
+from dvc_data.hashfile.db import get_odb
 
 
 def _get_odb(repo, settings, fs=None):

dvc/output.py

@@ -1,10 +1,10 @@
 import logging
 import os
 from collections import defaultdict
-from typing import TYPE_CHECKING, Dict, Optional, Set, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type
 from urllib.parse import urlparse
 
-from funcy import collecting, project
+from funcy import cached_property, collecting, project
 from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
 
 from dvc import prompt
@@ -17,22 +17,25 @@ from dvc.exceptions import (
     MergeError,
     RemoteCacheRequiredError,
 )
-from dvc_data import Tree
-from dvc_data import check as ocheck
-from dvc_data import load as oload
-from dvc_data.build import build
-from dvc_data.checkout import checkout
+from dvc_data.hashfile import Tree
+from dvc_data.hashfile import check as ocheck
+from dvc_data.hashfile import load as oload
+from dvc_data.hashfile.build import build
+from dvc_data.hashfile.checkout import checkout
 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.transfer import transfer as otransfer
+from dvc_data.hashfile.transfer import transfer as otransfer
 from dvc_objects.errors import ObjectFormatError
 
+from .annotations import ANNOTATION_FIELDS, ANNOTATION_SCHEMA, Annotation
 from .fs import LocalFileSystem, RemoteMissingDepsError, Schemes, get_cloud_fs
 from .utils import relpath
 from .utils.fs import path_isin
 
 if TYPE_CHECKING:
+    from dvc_data.hashfile.obj import HashFile
+    from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
 
 logger = logging.getLogger(__name__)
@@ -80,9 +83,10 @@ def loadd_from(stage, d_list):
         plot = d.pop(Output.PARAM_PLOT, False)
         persist = d.pop(Output.PARAM_PERSIST, False)
         checkpoint = d.pop(Output.PARAM_CHECKPOINT, False)
-        desc = d.pop(Output.PARAM_DESC, False)
         live = d.pop(Output.PARAM_LIVE, False)
         remote = d.pop(Output.PARAM_REMOTE, None)
+        annot = {field: d.pop(field, None) for field in ANNOTATION_FIELDS}
+        files = d.pop(Output.PARAM_FILES, None)
         ret.append(
             _get(
                 stage,
@@ -93,9 +97,10 @@ def loadd_from(stage, d_list):
                 plot=plot,
                 persist=persist,
                 checkpoint=checkpoint,
-                desc=desc,
                 live=live,
                 remote=remote,
+                **annot,
+                files=files,
             )
         )
     return ret
@@ -187,6 +192,7 @@ def load_from_pipeline(stage, data, typ="outs"):
                 Output.PARAM_PERSIST,
                 Output.PARAM_CHECKPOINT,
                 Output.PARAM_REMOTE,
+                *ANNOTATION_FIELDS,
             ],
         )
 
@@ -240,6 +246,7 @@ class Output:
     PARAM_PATH = "path"
     PARAM_CACHE = "cache"
     PARAM_CHECKPOINT = "checkpoint"
+    PARAM_FILES = "files"
     PARAM_METRIC = "metric"
     PARAM_METRIC_TYPE = "type"
     PARAM_METRIC_XPATH = "xpath"
@@ -252,7 +259,6 @@ class Output:
     PARAM_PLOT_TITLE = "title"
     PARAM_PLOT_HEADER = "header"
     PARAM_PERSIST = "persist"
-    PARAM_DESC = "desc"
     PARAM_LIVE = "live"
     PARAM_LIVE_SUMMARY = "summary"
     PARAM_LIVE_HTML = "html"
@@ -284,11 +290,29 @@ class Output:
         checkpoint=False,
         live=False,
         desc=None,
+        type=None,  # pylint: disable=redefined-builtin
+        labels=None,
+        meta=None,
         remote=None,
         repo=None,
+        fs_config=None,
+        files: List[Dict[str, Any]] = None,
     ):
+        self.annot = Annotation(
+            desc=desc, type=type, labels=labels or [], meta=meta or {}
+        )
         self.repo = stage.repo if not repo and stage else repo
-        fs_cls, fs_config, fs_path = get_cloud_fs(self.repo, url=path)
+        meta = Meta.from_dict(info or {})
+        # NOTE: when version_aware is not passed into get_cloud_fs, it will be
+        # set based on whether or not path is versioned
+        fs_kwargs = {"version_aware": True} if meta.version_id else {}
+
+        if fs_config is not None:
+            fs_kwargs.update(**fs_config)
+
+        fs_cls, fs_config, fs_path = get_cloud_fs(
+            self.repo, url=path, **fs_kwargs
+        )
         self.fs = fs_cls(**fs_config)
 
         if (
@@ -320,21 +344,38 @@ class Output:
         # By resolved path, which contains actual location,
         # should be absolute and don't contain remote:// refs.
         self.stage = stage
-        self.meta = Meta.from_dict(info)
-        self.hash_info = HashInfo.from_dict(info)
+        self.meta = meta
+        self._files = files
         self.use_cache = False if self.IS_DEPENDENCY else cache
         self.metric = False if self.IS_DEPENDENCY else metric
         self.plot = False if self.IS_DEPENDENCY else plot
         self.persist = persist
         self.checkpoint = checkpoint
         self.live = live
-        self.desc = desc
 
         self.fs_path = self._parse_path(self.fs, fs_path)
-        self.obj = None
+        self.obj: Optional["HashFile"] = None
 
         self.remote = remote
 
+        if self.fs.version_aware:
+            _, version_id = self.fs.path.coalesce_version(
+                self.def_path, self.meta.version_id
+            )
+            self.meta.version_id = version_id
+
+        if self.is_in_repo:
+            self.hash_name = "md5"
+        else:
+            self.hash_name = self.fs.PARAM_CHECKSUM
+
+        self.hash_info = HashInfo(
+            name=self.hash_name,
+            value=getattr(self.meta, self.hash_name, None),
+        )
+        if self.hash_info and self.hash_info.isdir:
+            self.meta.isdir = True
+
     def _parse_path(self, fs, fs_path):
         parsed = urlparse(self.def_path)
         if (
@@ -379,6 +420,7 @@ class Output:
         self.hash_info = HashInfo.from_dict({})
         self.meta = Meta.from_dict({})
         self.obj = None
+        self._files = None
 
     @property
     def protocol(self):
@@ -419,21 +461,27 @@ class Output:
         )
 
     def get_hash(self):
+        _, hash_info = self._get_hash_meta()
+        return hash_info
+
+    def _get_hash_meta(self):
         if self.use_cache:
             odb = self.odb
-            name = self.odb.fs.PARAM_CHECKSUM
         else:
             odb = self.repo.odb.local
-            name = self.fs.PARAM_CHECKSUM
-        _, _, obj = build(
+        _, meta, obj = build(
             odb,
             self.fs_path,
             self.fs,
-            name,
+            self.hash_name,
             ignore=self.dvcignore,
             dry_run=not self.use_cache,
         )
-        return obj.hash_info
+        return meta, obj.hash_info
+
+    def get_meta(self) -> Meta:
+        meta, _ = self._get_hash_meta()
+        return meta
 
     @property
     def is_dir_checksum(self):
@@ -452,6 +500,17 @@ class Output:
 
         return self.fs.exists(self.fs_path)
 
+    @cached_property
+    def index_key(self) -> Tuple[str, "DataIndexKey"]:
+        if self.is_in_repo:
+            workspace = "repo"
+            key = self.repo.fs.path.relparts(self.fs_path, self.repo.root_dir)
+        else:
+            workspace = self.fs.protocol
+            no_drive = self.fs.path.flavour.splitdrive(self.fs_path)[1]
+            key = self.fs.path.parts(no_drive)[1:]
+        return workspace, key
+
     def changed_checksum(self):
         return self.hash_info != self.get_hash()
 
@@ -469,6 +528,11 @@ class Output:
         except (FileNotFoundError, ObjectFormatError):
             return True
 
+    def changed_meta(self) -> bool:
+        if self.fs.version_aware and self.meta.version_id:
+            return self.meta.version_id == self.get_meta().version_id
+        return False
+
     def workspace_status(self):
         if not self.exists:
             return {str(self): "deleted"}
@@ -551,11 +615,12 @@ class Output:
                 self.repo.odb.local,
                 self.fs_path,
                 self.fs,
-                self.fs.PARAM_CHECKSUM,
+                self.hash_name,
                 ignore=self.dvcignore,
                 dry_run=True,
             )
             self.hash_info = obj.hash_info
+            self._files = None
             if not self.IS_DEPENDENCY:
                 logger.debug(
                     "Output '%s' doesn't use cache. Skipping saving.", self
@@ -568,18 +633,19 @@ class Output:
             self.odb,
             self.fs_path,
             self.fs,
-            self.odb.fs.PARAM_CHECKSUM,
+            self.hash_name,
             ignore=self.dvcignore,
         )
         self.hash_info = self.obj.hash_info
+        self._files = None
 
     def set_exec(self):
         if self.isfile() and self.meta.isexec:
             self.odb.set_exec(self.fs_path)
 
     def _checkout(self, *args, **kwargs):
-        from dvc_data.checkout import CheckoutError as _CheckoutError
-        from dvc_data.checkout import LinkError, PromptError
+        from dvc_data.hashfile.checkout import CheckoutError as _CheckoutError
+        from dvc_data.hashfile.checkout import LinkError, PromptError
 
         kwargs.setdefault("ignore", self.dvcignore)
         try:
@@ -610,7 +676,7 @@ class Output:
                     self.odb,
                     filter_info or self.fs_path,
                     self.fs,
-                    self.odb.fs.PARAM_CHECKSUM,
+                    self.hash_name,
                     ignore=self.dvcignore,
                 )
                 otransfer(
@@ -639,7 +705,7 @@ class Output:
             self.odb,
             self.fs_path,
             self.fs,
-            self.odb.fs.PARAM_CHECKSUM,
+            self.hash_name,
             ignore=self.dvcignore,
         )
         save_obj = save_obj.filter(prefix)
@@ -653,8 +719,10 @@ class Output:
         )
         return checkout_obj
 
-    def dumpd(self):
-        ret = {**self.hash_info.to_dict(), **self.meta.to_dict()}
+    def dumpd(self, **kwargs):
+        meta = self.meta.to_dict()
+        meta.pop("isdir", None)
+        ret = {**self.hash_info.to_dict(), **meta}
 
         if self.is_in_repo:
             path = self.fs.path.as_posix(
@@ -668,9 +736,7 @@ class Output:
         if self.IS_DEPENDENCY:
             return ret
 
-        if self.desc:
-            ret[self.PARAM_DESC] = self.desc
-
+        ret.update(self.annot.to_dict())
         if not self.use_cache:
             ret[self.PARAM_CACHE] = self.use_cache
 
@@ -699,6 +765,18 @@ class Output:
         if self.remote:
             ret[self.PARAM_REMOTE] = self.remote
 
+        if (
+            self.use_cache
+            and self.is_in_repo
+            and self.hash_info.isdir
+            and (kwargs.get("with_files") or self._files is not None)
+        ):
+            if self.obj:
+                obj = self.obj
+            else:
+                obj = self.get_obj()
+            ret[self.PARAM_FILES] = obj.as_list(with_meta=True)
+
         return ret
 
     def verify_metric(self):
@@ -731,14 +809,22 @@ class Output:
         ) as cb:
             self.fs.get(self.fs_path, to.fs_path, batch_size=jobs, callback=cb)
 
-    def get_obj(self, filter_info=None, **kwargs):
+    def get_obj(
+        self, filter_info: Optional[str] = None, **kwargs
+    ) -> Optional["HashFile"]:
+        obj: Optional["HashFile"] = None
         if self.obj:
             obj = self.obj
         elif self.hash_info:
-            try:
-                obj = oload(self.odb, self.hash_info)
-            except (FileNotFoundError, ObjectFormatError):
-                return None
+            if self._files:
+                tree = Tree.from_list(self._files, hash_name=self.hash_name)
+                tree.digest()
+                obj = tree
+            else:
+                try:
+                    obj = oload(self.odb, self.hash_info)
+                except (FileNotFoundError, ObjectFormatError):
+                    return None
         else:
             return None
 
@@ -858,6 +944,7 @@ class Output:
         )
 
         self.hash_info = obj.hash_info
+        self._files = None
         return obj
 
     def get_files_number(self, filter_info=None):
@@ -928,6 +1015,7 @@ class Output:
 
         obj = self.get_obj()
         if filter_info and filter_info != self.fs_path:
+            assert obj
             prefix = self.fs.path.parts(
                 self.fs.path.relpath(filter_info, self.fs_path)
             )
@@ -1020,7 +1108,7 @@ class Output:
         other = out.dumpd()
 
         ignored = [
-            self.fs.PARAM_CHECKSUM,
+            self.hash_name,
             Meta.PARAM_SIZE,
             Meta.PARAM_NFILES,
         ]
@@ -1040,8 +1128,8 @@ class Output:
             )
 
     def merge(self, ancestor, other):
-        from dvc_data.objects.tree import MergeError as TreeMergeError
-        from dvc_data.objects.tree import du, merge
+        from dvc_data.hashfile.tree import MergeError as TreeMergeError
+        from dvc_data.hashfile.tree import du, merge
 
         assert other
 
@@ -1064,6 +1152,7 @@ class Output:
         self.odb.add(merged.path, merged.fs, merged.oid)
 
         self.hash_info = merged.hash_info
+        self._files = None
         self.meta = Meta(
             size=du(self.odb, merged),
             nfiles=len(merged),
@@ -1086,21 +1175,33 @@ class Output:
         return bool(self.plot) or bool(self.live)
 
 
+META_SCHEMA = {
+    Meta.PARAM_SIZE: int,
+    Meta.PARAM_NFILES: int,
+    Meta.PARAM_ISEXEC: bool,
+    Meta.PARAM_VERSION_ID: str,
+}
+
 ARTIFACT_SCHEMA = {
     **CHECKSUMS_SCHEMA,
+    **META_SCHEMA,
     Required(Output.PARAM_PATH): str,
     Output.PARAM_PLOT: bool,
     Output.PARAM_PERSIST: bool,
     Output.PARAM_CHECKPOINT: bool,
-    Meta.PARAM_SIZE: int,
-    Meta.PARAM_NFILES: int,
-    Meta.PARAM_ISEXEC: bool,
+}
+
+DIR_FILES_SCHEMA: Dict[str, Any] = {
+    **CHECKSUMS_SCHEMA,
+    **META_SCHEMA,
+    Required(Tree.PARAM_RELPATH): str,
 }
 
 SCHEMA = {
     **ARTIFACT_SCHEMA,
+    **ANNOTATION_SCHEMA,
     Output.PARAM_CACHE: bool,
     Output.PARAM_METRIC: Output.METRIC_SCHEMA,
-    Output.PARAM_DESC: str,
     Output.PARAM_REMOTE: str,
+    Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }

dvc/parsing/context.py

@@ -159,7 +159,7 @@ class Value(Node):
 PRIMITIVES = (int, float, str, bytes, bool)
 
 
-class Container(Node, ABC):
+class Container(Node, ABC):  # noqa: B024
     meta: Meta
     data: Union[list, dict]
     _key_transform = staticmethod(identity)

dvc/repo/add.py

@@ -264,6 +264,7 @@ def create_stages(
             outs=[out],
             external=external,
         )
-        if kwargs.get("desc"):
-            stage.outs[0].desc = kwargs["desc"]
+
+        out_obj = stage.outs[0]
+        out_obj.annot.update(**kwargs)
         yield stage

dvc/repo/data.py

@@ -1,13 +1,16 @@
 import os
 import posixpath
 from collections import defaultdict
+from itertools import chain
 from typing import (
     TYPE_CHECKING,
     Any,
     Dict,
     Iterable,
+    Iterator,
     List,
     Optional,
+    Tuple,
     TypedDict,
     cast,
 )
@@ -74,8 +77,8 @@ def _granular_diff(
     new_obj: Optional["HashFile"],
     cache: "HashFileDB",
 ) -> Dict[str, List[str]]:
-    from dvc_data.diff import ROOT
-    from dvc_data.diff import diff as odiff
+    from dvc_data.hashfile.diff import ROOT
+    from dvc_data.hashfile.diff import diff as odiff
 
     def path_join(root: str, *paths: str, isdir: bool = False) -> str:
         if not isdir and paths == ROOT:
@@ -102,7 +105,7 @@ def _get_obj_items(root: str, obj: Optional["HashFile"]) -> List[str]:
     if not obj:
         return []
 
-    from dvc_data.objects.tree import Tree
+    from dvc_data.hashfile.tree import Tree
 
     sep = os.path.sep
     if isinstance(obj, Tree):
@@ -153,6 +156,8 @@ def _git_info(scm: "Base", untracked_files: str = "all") -> GitInfo:
         empty_repo = False
 
     staged, unstaged, untracked = scm.status(untracked_files=untracked_files)
+    if os.name == "nt":
+        untracked = [posixpath_to_os_path(path) for path in untracked]
     # NOTE: order is important here.
     return GitInfo(
         staged=staged,
@@ -164,7 +169,7 @@ def _git_info(scm: "Base", untracked_files: str = "all") -> GitInfo:
 
 
 def _diff_index_to_wtree(repo: "Repo", **kwargs: Any) -> Dict[str, List[str]]:
-    from dvc_data.build import build
+    from dvc_data.hashfile.build import build
 
     unstaged_diff = defaultdict(list)
     for out in repo.index.outs:
@@ -210,7 +215,9 @@ def _diff_head_to_index(
     repo: "Repo", head: str = "HEAD", **kwargs: Any
 ) -> Dict[str, List[str]]:
     # we need to store objects from index and the HEAD to diff later
-    objs: Dict[str, Dict[str, "HashFile"]] = defaultdict(dict)
+    objs: Dict[str, Dict[str, Tuple["HashFile", "HashInfo"]]]
+    objs = defaultdict(dict)
+
     staged_diff = defaultdict(list)
     for rev in repo.brancher(revs=[head]):
         for out in repo.index.outs:
@@ -250,7 +257,9 @@ class Status(TypedDict):
     git: GitInfo
 
 
-def _transform_git_paths_to_dvc(repo: "Repo", files: Iterable[str]):
+def _transform_git_paths_to_dvc(
+    repo: "Repo", files: Iterable[str]
+) -> List[str]:
     """Transform files rel. to Git root to DVC root, and drop outside files."""
     rel = repo.fs.path.relpath(repo.root_dir, repo.scm.root_dir).rstrip("/")
 
@@ -298,3 +307,18 @@ def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status:
         unchanged=list(unchanged),
         git=git_info,
     )
+
+
+def ls(
+    repo: "Repo",
+    targets: List[Optional[str]] = None,
+    recursive: bool = False,
+) -> Iterator[Dict[str, Any]]:
+    targets = targets or [None]
+    pairs = chain.from_iterable(
+        repo.stage.collect_granular(target, recursive=recursive)
+        for target in targets
+    )
+    for stage, filter_info in pairs:
+        for out in stage.filter_outs(filter_info):
+            yield {"path": str(out), **out.annot.to_dict()}

dvc/repo/diff.py

@@ -115,7 +115,7 @@ def _paths_checksums(repo, targets):
 
 def _output_paths(repo, targets):
     from dvc.fs import LocalFileSystem
-    from dvc_data.build import build
+    from dvc_data.hashfile.build import build
 
     on_working_fs = isinstance(repo.fs, LocalFileSystem)
 

dvc/repo/experiments/executor/manager/base.py

@@ -21,7 +21,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-class BaseExecutorManager(ABC, Mapping):
+class BaseExecutorManager(ABC, Mapping):  # noqa: B024
     """Manages executors for a collection of experiments to be run."""
 
     EXECUTOR_CLS: Type = BaseExecutor

dvc/repo/experiments/run.py

@@ -2,6 +2,7 @@ import logging
 from typing import Dict, Iterable, Optional
 
 from dvc.dependency.param import ParamsDependency
+from dvc.exceptions import InvalidArgumentError
 from dvc.repo import locked
 from dvc.ui import ui
 from dvc.utils.cli_parse import to_path_overrides
@@ -27,6 +28,8 @@ def run(
     Returns a dict mapping new experiment SHAs to the results
     of `repro` for that experiment.
     """
+    from dvc.utils.hydra import to_hydra_overrides
+
     if run_all:
         entries = list(repo.experiments.celery_queue.iter_queued())
         return repo.experiments.reproduce_celery(entries, jobs=jobs)
@@ -42,19 +45,42 @@ def run(
         # Force `_update_params` even if `--set-param` was not used
         path_overrides[hydra_output_file] = []
 
-    if queue:
-        if not kwargs.get("checkpoint_resume", None):
-            kwargs["reset"] = True
+    hydra_sweep = any(
+        x.is_sweep_override()
+        for param_file in path_overrides
+        for x in to_hydra_overrides(path_overrides[param_file])
+    )
+
+    if hydra_sweep and not queue:
+        raise InvalidArgumentError(
+            "Sweep overrides can't be used without `--queue`"
+        )
+
+    if not queue:
+        return repo.experiments.reproduce_one(
+            targets=targets, params=path_overrides, tmp_dir=tmp_dir, **kwargs
+        )
+
+    if hydra_sweep:
+        from dvc.utils.hydra import get_hydra_sweeps
+
+        sweeps = get_hydra_sweeps(path_overrides)
+    else:
+        sweeps = [path_overrides]
+
+    if not kwargs.get("checkpoint_resume", None):
+        kwargs["reset"] = True
+
+    for sweep_overrides in sweeps:
         queue_entry = repo.experiments.queue_one(
             repo.experiments.celery_queue,
             targets=targets,
-            params=path_overrides,
+            params=sweep_overrides,
             **kwargs,
         )
+        if sweep_overrides:
+            ui.write(f"Queueing with overrides '{sweep_overrides}'.")
         name = queue_entry.name or queue_entry.stash_rev[:7]
         ui.write(f"Queued experiment '{name}' for future execution.")
-        return {}
 
-    return repo.experiments.reproduce_one(
-        targets=targets, params=path_overrides, tmp_dir=tmp_dir, **kwargs
-    )
+    return {}

dvc/repo/fetch.py

@@ -1,6 +1,8 @@
 import logging
+from contextlib import suppress
 from typing import TYPE_CHECKING, Optional
 
+from dvc.config import NoRemoteError
 from dvc.exceptions import DownloadError, FileTransferError
 from dvc.fs import Schemes
 
@@ -12,6 +14,36 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _fetch_worktree(repo, remote):
+    from dvc_data.index import save
+
+    index = repo.index.data["repo"]
+    for key, entry in index.iteritems():
+        entry.fs = remote.fs
+        entry.path = remote.fs.path.join(
+            remote.path,
+            *key,
+        )
+    save(index)
+
+    for stage in repo.index.stages:
+        for out in stage.outs:
+            if not out.use_cache:
+                continue
+
+            if not out.is_in_repo:
+                continue
+
+            workspace, key = out.index_key
+            entry = repo.index.data[workspace][key]
+            out.hash_info = entry.hash_info
+            out.meta = entry.meta
+
+        stage.dvcfile.dump(stage)
+
+    return len(index)
+
+
 @locked
 def fetch(
     self,
@@ -42,6 +74,11 @@ def fetch(
     if isinstance(targets, str):
         targets = [targets]
 
+    with suppress(NoRemoteError):
+        _remote = self.cloud.get_remote(name=remote)
+        if _remote.worktree:
+            return _fetch_worktree(self, _remote)
+
     used = self.used_objs(
         targets,
         all_branches=all_branches,

dvc/repo/gc.py

@@ -60,8 +60,8 @@ def gc(
     from contextlib import ExitStack
 
     from dvc.repo import Repo
-    from dvc_data.db import get_index
-    from dvc_data.gc import gc as ogc
+    from dvc_data.hashfile.db import get_index
+    from dvc_data.hashfile.gc import gc as ogc
 
     if not repos:
         repos = []

dvc/repo/imp_url.py

@@ -22,7 +22,11 @@ def imp_url(
     remote=None,
     to_remote=False,
     desc=None,
+    type=None,  # pylint: disable=redefined-builtin
+    labels=None,
+    meta=None,
     jobs=None,
+    fs_config=None,
 ):
     from dvc.dvcfile import Dvcfile
     from dvc.stage import Stage, create_stage, restore_fields
@@ -58,12 +62,12 @@ def imp_url(
         deps=[url],
         outs=[out],
         erepo=erepo,
+        fs_config=fs_config,
     )
     restore_fields(stage)
 
-    if desc:
-        stage.outs[0].desc = desc
-
+    out_obj = stage.outs[0]
+    out_obj.annot.update(desc=desc, type=type, labels=labels, meta=meta)
     dvcfile = Dvcfile(self, stage.path)
     dvcfile.remove()
 

dvc/repo/index.py

@@ -183,15 +183,7 @@ class Index:
             if not out.use_cache:
                 continue
 
-            if out.is_in_repo:
-                workspace = "repo"
-                key = self.repo.fs.path.relparts(
-                    out.fs_path, self.repo.root_dir
-                )
-            else:
-                workspace = out.fs.protocol
-                no_drive = out.fs.path.flavour.splitdrive(out.fs_path)[1]
-                key = out.fs.path.parts(no_drive)[1:]
+            workspace, key = out.index_key
 
             try:
                 remote = self.repo.cloud.get_remote_odb(out.remote)
@@ -205,6 +197,7 @@ class Index:
                 obj=out.obj,
                 hash_info=out.hash_info,
                 odb=out.odb,
+                cache=out.odb,
                 remote=remote,
             )
 

dvc/repo/push.py

@@ -9,6 +9,52 @@ if TYPE_CHECKING:
     from dvc_objects.db.base import ObjectDB
 
 
+def _push_worktree(repo, remote):
+    from dvc_data.hashfile.tree import tree_from_index
+    from dvc_data.index import build, checkout, collect
+
+    index = repo.index.data["repo"]
+    checkout(index, remote.path, remote.fs, force=True)
+    build(index, remote.path, remote.fs)
+    if any(out.isdir() for out in repo.index.outs):
+        collect(index, remote.path, remote.fs)
+
+    for stage in repo.index.stages:
+        for out in stage.outs:
+            if not out.use_cache:
+                continue
+
+            if not out.is_in_repo:
+                continue
+
+            workspace, key = out.index_key
+            index = repo.index.data[workspace]
+            entry = index[key]
+            if out.isdir():
+                old_tree = out.get_obj()
+                entry.hash_info = old_tree.hash_info
+                entry.meta = out.meta
+                for subkey, entry in index.iteritems(key):
+                    if entry.meta.isdir:
+                        continue
+                    fs_path = repo.fs.path.join(repo.root_dir, *subkey)
+                    _, hash_info = old_tree.get(
+                        repo.fs.path.relparts(fs_path, out.fs_path)
+                    )
+                    entry.hash_info = hash_info
+                tree_meta, new_tree = tree_from_index(index, key)
+                new_tree.digest()
+                out.obj = new_tree
+                out.hash_info = new_tree.hash_info
+                out.meta = tree_meta
+            else:
+                out.hash_info = entry.hash_info
+                out.meta = entry.meta
+        stage.dvcfile.dump(stage, with_files=True)
+
+    return len(index)
+
+
 @locked
 def push(
     self,
@@ -26,6 +72,10 @@ def push(
     odb: Optional["ObjectDB"] = None,
     include_imports=False,
 ):
+    _remote = self.cloud.get_remote(name=remote)
+    if _remote.worktree:
+        return _push_worktree(self, _remote)
+
     used_run_cache = (
         self.stage_cache.push(remote, odb=odb) if run_cache else []
     )

dvc/repo/remove.py

@@ -19,7 +19,7 @@ logger = logging.getLogger(__name__)
 @locked
 def remove(self: "Repo", target: str, outs: bool = False):
     try:
-        stages = self.stage.from_target(target)
+        stages = self.stage.from_target(target, accept_group=False)
     except (StageNotFound, StageFileDoesNotExistError) as e:
         # If the user specified a tracked file as a target instead of a stage,
         # e.g. `data.csv` instead of `data.csv.dvc`,

dvc/repo/reproduce.py

@@ -102,7 +102,6 @@ def reproduce(
     from .graph import get_pipeline, get_pipelines
 
     glob = kwargs.pop("glob", False)
-    accept_group = not glob
 
     if isinstance(targets, str):
         targets = [targets]
@@ -137,7 +136,6 @@ def reproduce(
                 self.stage.collect(
                     target,
                     recursive=recursive,
-                    accept_group=accept_group,
                     glob=glob,
                 )
             )

dvc/repo/stage.py

@@ -82,7 +82,6 @@ def _collect_specific_target(
     target: str,
     with_deps: bool,
     recursive: bool,
-    accept_group: bool,
 ) -> Tuple[StageIter, "OptStr", "OptStr"]:
     from dvc.dvcfile import is_valid_filename
 
@@ -98,13 +97,11 @@ def _collect_specific_target(
         msg = "Checking if stage '%s' is in '%s'"
         logger.debug(msg, target, PIPELINE_FILE)
         if not (recursive and loader.fs.isdir(target)):
-            stages = _maybe_collect_from_dvc_yaml(
-                loader, target, with_deps, accept_group=accept_group
-            )
+            stages = _maybe_collect_from_dvc_yaml(loader, target, with_deps)
             if stages:
                 return stages, file, name
     elif not with_deps and is_valid_filename(file):
-        stages = loader.load_all(file, name, accept_group=accept_group)
+        stages = loader.load_all(file, name)
         return stages, file, name
     return [], file, name
 
@@ -209,7 +206,7 @@ class StageLoad:
         return stage
 
     def from_target(
-        self, target: str, accept_group: bool = False, glob: bool = False
+        self, target: str, accept_group: bool = True, glob: bool = False
     ) -> StageList:
         """
         Returns a list of stage from the provided target.
@@ -250,15 +247,12 @@ class StageLoad:
         self,
         stages: "StageLoader",
         name: str = None,
-        accept_group: bool = False,
+        accept_group: bool = True,
         glob: bool = False,
     ) -> Iterable[str]:
 
-        assert not (accept_group and glob)
-
         if not name:
             return stages.keys()
-
         if accept_group and stages.is_foreach_generated(name):
             return self._get_group_keys(stages, name)
         if glob:
@@ -269,7 +263,7 @@ class StageLoad:
         self,
         path: str = None,
         name: str = None,
-        accept_group: bool = False,
+        accept_group: bool = True,
         glob: bool = False,
     ) -> StageList:
         """Load a list of stages from a file.
@@ -338,7 +332,6 @@ class StageLoad:
         with_deps: bool = False,
         recursive: bool = False,
         graph: "DiGraph" = None,
-        accept_group: bool = False,
         glob: bool = False,
     ) -> StageIter:
         """Collect list of stages from the provided target.
@@ -347,15 +340,14 @@ class StageLoad:
             target: if not provided, all of the stages in the graph are
                 returned.
                 Target can be:
-                - a stage name in the `dvc.yaml` file.
+                - a foreach group name or a stage name in the `dvc.yaml` file.
+                - a generated stage name from a foreach group.
                 - a path to `dvc.yaml` or `.dvc` file.
                 - in case of a stage to a dvc.yaml file in a different
                   directory than current working directory, it can be a path
                   to dvc.yaml file, followed by a colon `:`, followed by stage
                   name (eg: `../dvc.yaml:build`).
                 - in case of `recursive`, it can be a path to a directory.
-                - in case of `accept_group`, it can be a group name of
-                    `foreach` generated stage.
                 - in case of `glob`, it can be a wildcard pattern to match
                   stages. Example: `build*` for stages in `dvc.yaml` file, or
                   `../dvc.yaml:build*` for stages in dvc.yaml in a different
@@ -367,8 +359,6 @@ class StageLoad:
             recursive: if true and if `target` is a directory, all of the
                 stages inside that directory is returned.
             graph: graph to use. Defaults to `repo.graph`.
-            accept_group: if true, all of the `foreach` generated stages of
-                the specified target is returned.
             glob: Use `target` as a pattern to match stages in a file.
         """
         if not target:
@@ -380,7 +370,7 @@ class StageLoad:
             path = self.fs.path.abspath(target)
             return collect_inside_path(path, graph or self.graph)
 
-        stages = self.from_target(target, accept_group=accept_group, glob=glob)
+        stages = self.from_target(target, glob=glob)
         if not with_deps:
             return stages
 
@@ -392,14 +382,14 @@ class StageLoad:
         with_deps: bool = False,
         recursive: bool = False,
         graph: "DiGraph" = None,
-        accept_group: bool = False,
     ) -> List[StageInfo]:
         """Collects a list of (stage, filter_info) from the given target.
 
         Priority is in the order of following in case of ambiguity:
         - .dvc file or .yaml file
         - dir if recursive and directory exists
-        - stage_name
+        - foreach_group_name or stage_name
+        - generated stage name from a foreach group
         - output file
 
         Args:
@@ -418,7 +408,7 @@ class StageLoad:
         target = as_posix(target)
 
         stages, file, _ = _collect_specific_target(
-            self, target, with_deps, recursive, accept_group
+            self, target, with_deps, recursive
         )
         if not stages:
             if not (recursive and self.fs.isdir(target)):
@@ -440,7 +430,6 @@ class StageLoad:
                     with_deps,
                     recursive,
                     graph,
-                    accept_group=accept_group,
                 )
             except StageFileDoesNotExistError as exc:
                 # collect() might try to use `target` as a stage name

dvc/schema.py

@@ -3,11 +3,11 @@ from collections.abc import Mapping
 from voluptuous import Any, Optional, Required, Schema
 
 from dvc import dependency, output
-from dvc.output import CHECKSUMS_SCHEMA, Output
+from dvc.annotations import ANNOTATION_SCHEMA
+from dvc.output import CHECKSUMS_SCHEMA, DIR_FILES_SCHEMA, META_SCHEMA, Output
 from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
 from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
 from dvc.stage.params import StageParams
-from dvc_data.hashfile.meta import Meta
 
 STAGES = "stages"
 SINGLE_STAGE_SCHEMA = {
@@ -25,10 +25,9 @@ SINGLE_STAGE_SCHEMA = {
 
 DATA_SCHEMA = {
     **CHECKSUMS_SCHEMA,
+    **META_SCHEMA,
     Required("path"): str,
-    Meta.PARAM_SIZE: int,
-    Meta.PARAM_NFILES: int,
-    Meta.PARAM_ISEXEC: bool,
+    Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }
 LOCK_FILE_STAGE_SCHEMA = {
     Required(StageParams.PARAM_CMD): Any(str, list),
@@ -47,16 +46,15 @@ LOCKFILE_V2_SCHEMA = {
 
 OUT_PSTAGE_DETAILED_SCHEMA = {
     str: {
+        **ANNOTATION_SCHEMA,  # type: ignore
         Output.PARAM_CACHE: bool,
         Output.PARAM_PERSIST: bool,
         Output.PARAM_CHECKPOINT: bool,
-        Output.PARAM_DESC: str,
         Output.PARAM_REMOTE: str,
     }
 }
 
 PLOTS = "plots"
-PLOTS_SCHEMA = dict
 PLOT_PROPS = {
     Output.PARAM_PLOT_TEMPLATE: str,
     Output.PARAM_PLOT_X: str,
@@ -112,6 +110,15 @@ def either_or(primary, fallback, fallback_includes=None):
     return validator
 
 
+PLOT_DEFINITION = {
+    Output.PARAM_PLOT_X: str,
+    Output.PARAM_PLOT_Y: Any(str, [str], {str: Any(str, [str])}),
+    Output.PARAM_PLOT_X_LABEL: str,
+    Output.PARAM_PLOT_Y_LABEL: str,
+    Output.PARAM_PLOT_TITLE: str,
+    Output.PARAM_PLOT_TEMPLATE: str,
+}
+PLOTS_SCHEMA = {str: Any(PLOT_DEFINITION, None)}
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,

dvc/stage/__init__.py

@@ -88,7 +88,7 @@ def create_stage(cls, repo, path, external=False, **kwargs):
     if not external:
         check_no_externals(stage)
     fill_stage_dependencies(
-        stage, **project(kwargs, ["deps", "erepo", "params"])
+        stage, **project(kwargs, ["deps", "erepo", "params", "fs_config"])
     )
     check_circular_dependency(stage)
     check_duplicated_arguments(stage)
@@ -110,16 +110,13 @@ def restore_fields(stage):
     # will be used to restore comments later
     # noqa, pylint: disable=protected-access
     stage._stage_text = old._stage_text
-
     stage.meta = old.meta
     stage.desc = old.desc
 
-    old_fields = {out.def_path: (out.desc, out.remote) for out in old.outs}
-
+    old_fields = {out.def_path: (out.annot, out.remote) for out in old.outs}
     for out in stage.outs:
-        out_fields = old_fields.get(out.def_path, None)
-        if out_fields:
-            out.desc, out.remote = out_fields
+        if out_fields := old_fields.get(out.def_path, None):
+            out.annot, out.remote = out_fields
 
 
 class Stage(params.StageParams):
@@ -464,8 +461,8 @@ class Stage(params.StageParams):
     def reload(self):
         return self.dvcfile.stage
 
-    def dumpd(self):
-        return get_dump(self)
+    def dumpd(self, **kwargs):
+        return get_dump(self, **kwargs)
 
     def compute_md5(self):
         # `dvc add`ed files don't need stage md5

dvc/stage/cache.py

@@ -9,7 +9,7 @@ from funcy import cached_property, first
 from dvc import fs
 from dvc.exceptions import DvcException
 from dvc.utils import dict_sha256, relpath
-from dvc_data.transfer import _log_exceptions
+from dvc_data.hashfile.transfer import _log_exceptions
 
 if TYPE_CHECKING:
     from dvc_objects.db import ObjectDB

dvc/stage/loader.py

@@ -69,7 +69,8 @@ class StageLoader(Mapping):
             info = info.copy()
             info.pop("path", None)
             item.meta = Meta.from_dict(info)
-            item.hash_info = HashInfo.from_dict(info)
+            hash_value = getattr(item.meta, item.hash_name, None)
+            item.hash_info = HashInfo(item.hash_name, hash_value)
 
     @classmethod
     def load_stage(cls, dvcfile, name, stage_data, lock_data=None):

dvc/stage/serialize.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, no_type_check
 from funcy import post_processing
 
 from dvc.dependency import ParamsDependency
-from dvc.output import Output
+from dvc.output import Annotation, Output
 from dvc.utils.collections import apply_diff
 from dvc.utils.serialize import parse_yaml_for_update
 
@@ -27,7 +27,7 @@ PARAM_METRIC = Output.PARAM_METRIC
 PARAM_PLOT = Output.PARAM_PLOT
 PARAM_PERSIST = Output.PARAM_PERSIST
 PARAM_CHECKPOINT = Output.PARAM_CHECKPOINT
-PARAM_DESC = Output.PARAM_DESC
+PARAM_DESC = Annotation.PARAM_DESC
 PARAM_REMOTE = Output.PARAM_REMOTE
 
 DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
@@ -38,8 +38,9 @@ sort_by_path = partial(sorted, key=attrgetter("def_path"))
 
 @post_processing(OrderedDict)
 def _get_flags(out):
-    if out.desc:
-        yield PARAM_DESC, out.desc
+    annot = out.annot.to_dict()
+    yield from annot.items()
+
     if not out.use_cache:
         yield PARAM_CACHE, False
     if out.checkpoint:
@@ -155,10 +156,12 @@ def to_single_stage_lockfile(stage: "Stage") -> dict:
     assert stage.cmd
 
     def _dumpd(item):
+        meta_d = item.meta.to_dict()
+        meta_d.pop("isdir", None)
         ret = [
             (item.PARAM_PATH, item.def_path),
             *item.hash_info.to_dict().items(),
-            *item.meta.to_dict().items(),
+            *meta_d.items(),
         ]
 
         return OrderedDict(ret)
@@ -185,8 +188,8 @@ def to_lockfile(stage: "PipelineStage") -> dict:
     return {stage.name: to_single_stage_lockfile(stage)}
 
 
-def to_single_stage_file(stage: "Stage"):
-    state = stage.dumpd()
+def to_single_stage_file(stage: "Stage", **kwargs):
+    state = stage.dumpd(**kwargs)
 
     # When we load a stage we parse yaml with a fast parser, which strips
     # off all the comments and formatting. To retain those on update we do

dvc/stage/utils.py

@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Union
 
 from funcy import concat, first, lsplit, rpartial, without
 
+from dvc.annotations import ANNOTATION_FIELDS
 from dvc.exceptions import InvalidArgumentError
 from dvc_data.hashfile.meta import Meta
 
@@ -107,12 +108,16 @@ def _load_live_output(
     return outs
 
 
-def fill_stage_dependencies(stage, deps=None, erepo=None, params=None):
+def fill_stage_dependencies(
+    stage, deps=None, erepo=None, params=None, fs_config=None
+):
     from dvc.dependency import loads_from, loads_params
 
     assert not stage.deps
     stage.deps = []
-    stage.deps += loads_from(stage, deps or [], erepo=erepo)
+    stage.deps += loads_from(
+        stage, deps or [], erepo=erepo, fs_config=fs_config
+    )
     stage.deps += loads_params(stage, params or [])
 
 
@@ -194,9 +199,9 @@ def compute_md5(stage):
     return dict_md5(
         d,
         exclude=[
+            *ANNOTATION_FIELDS,
             stage.PARAM_LOCKED,  # backward compatibility
             stage.PARAM_FROZEN,
-            Output.PARAM_DESC,
             Output.PARAM_METRIC,
             Output.PARAM_PERSIST,
             Output.PARAM_CHECKPOINT,
@@ -221,7 +226,7 @@ def resolve_paths(fs, path, wdir=None):
     return path, wdir
 
 
-def get_dump(stage):
+def get_dump(stage, **kwargs):
     return {
         key: value
         for key, value in {
@@ -230,8 +235,8 @@ def get_dump(stage):
             stage.PARAM_CMD: stage.cmd,
             stage.PARAM_WDIR: resolve_wdir(stage.wdir, stage.path),
             stage.PARAM_FROZEN: stage.frozen,
-            stage.PARAM_DEPS: [d.dumpd() for d in stage.deps],
-            stage.PARAM_OUTS: [o.dumpd() for o in stage.outs],
+            stage.PARAM_DEPS: [d.dumpd(**kwargs) for d in stage.deps],
+            stage.PARAM_OUTS: [o.dumpd(**kwargs) for o in stage.outs],
             stage.PARAM_ALWAYS_CHANGED: stage.always_changed,
             stage.PARAM_META: stage.meta,
         }.items()

dvc/utils/hydra.py

@@ -1,11 +1,13 @@
+from collections import defaultdict
 from pathlib import Path
 from typing import TYPE_CHECKING, List
 
 from hydra import compose, initialize_config_dir
 from hydra._internal.config_loader_impl import ConfigLoaderImpl
+from hydra._internal.core_plugins.basic_sweeper import BasicSweeper
 from hydra.core.override_parser.overrides_parser import OverridesParser
+from hydra.core.override_parser.types import ValueType
 from hydra.errors import ConfigCompositionException, OverrideParseException
-from hydra.types import RunMode
 from omegaconf import OmegaConf
 
 from dvc.exceptions import InvalidArgumentError
@@ -59,11 +61,7 @@ def apply_overrides(path: "StrPath", overrides: List[str]) -> None:
     modify_data = MODIFIERS[suffix]
     with modify_data(path) as original_data:
         try:
-            parser = OverridesParser.create()
-            parsed = parser.parse_overrides(overrides=overrides)
-            ConfigLoaderImpl.validate_sweep_overrides_legal(
-                parsed, run_mode=RunMode.RUN, from_shell=True
-            )
+            parsed = to_hydra_overrides(overrides)
 
             new_data = OmegaConf.create(
                 to_omegaconf(original_data),
@@ -78,3 +76,34 @@ def apply_overrides(path: "StrPath", overrides: List[str]) -> None:
 
         merge_dicts(original_data, new_data)
         remove_missing_keys(original_data, new_data)
+
+
+def to_hydra_overrides(path_overrides):
+    parser = OverridesParser.create()
+    return parser.parse_overrides(overrides=path_overrides)
+
+
+def get_hydra_sweeps(path_overrides):
+    merged_overrides = []
+    for path, overrides in path_overrides.items():
+        merged_overrides.extend(
+            [f"{path}___{override}" for override in overrides]
+        )
+
+    hydra_overrides = to_hydra_overrides(merged_overrides)
+    for hydra_override in hydra_overrides:
+        if hydra_override.value_type == ValueType.GLOB_CHOICE_SWEEP:
+            raise InvalidArgumentError(
+                f"Glob override '{hydra_override.input_line}' "
+                "is not supported."
+            )
+
+    splits = BasicSweeper.split_arguments(hydra_overrides, None)[0]
+    sweeps = []
+    for split in splits:
+        sweep_overrides = defaultdict(list)
+        for merged_override in split:
+            path, override = merged_override.split("___")
+            sweep_overrides[path].append(override)
+        sweeps.append(dict(sweep_overrides))
+    return sweeps

setup.cfg

@@ -62,11 +62,11 @@ install_requires =
     rich>=10.13.0
     pyparsing>=2.4.7
     typing-extensions>=3.7.4
-    scmrepo==0.0.25
-    dvc-render==0.0.9
+    scmrepo==0.1.1
+    dvc-render==0.0.10
     dvc-task==0.1.2
     dvclive>=0.10.0
-    dvc-data==0.4.0
+    dvc-data==0.8.0
     dvc-http==2.19.1
     hydra-core>=1.1.0
 
@@ -86,12 +86,13 @@ dev =
     %(tests)s
     dvc-data[cli]
 # Extra dependencies for remote integrations
-azure = dvc-azure==2.19.0
+azure = dvc-azure==2.20.2
 gdrive = dvc-gdrive==2.19.0
 gs = dvc-gs==2.19.1
-hdfs = dvc-hdfs==2.19.0; python_version < '3.11'
+hdfs =
+    dvc-hdfs==2.19.0; python_version < '3.11'
 oss = dvc-oss==2.19.0
-s3 = dvc-s3==2.19.0
+s3 = dvc-s3==2.20.0
 ssh = dvc-ssh==2.19.0
 # gssapi should not be included in all_remotes, because it doesn't have wheels
 # for Linux, so it will fail to compile if user doesn't have all the
@@ -106,7 +107,7 @@ tests =
     %(terraform)s
     dvc-ssh==2.19.0
     # Test requirements
-    pytest==7.1.2
+    pytest==7.1.3
     pytest-cov==3.0.0
     pytest-xdist==2.5.0
     pytest-mock==3.8.2
@@ -115,11 +116,10 @@ tests =
     # https://github.com/docker/docker-py/issues/2902
     pytest-docker==0.11.0; python_version < '3.10' or sys_platform != 'win32'
     flaky==3.7.0
-    mock==4.0.3
     pytest-timeout==2.1.0
     filelock==3.8.0
     # pylint requirements
-    pylint==2.15.0
+    pylint==2.15.2
     # we use this to suppress pytest-related false positives in our tests.
     pylint-pytest==1.1.2
     # we use this to suppress some messages in tests, eg: foo/bar naming,
@@ -145,6 +145,8 @@ exclude =
 [options.entry_points]
 console_scripts =
     dvc = dvc.cli:main
+fsspec.specs =
+    dvc = dvc.api:DvcFileSystem
 
 [flake8]
 ignore=

tests/func/data/db/test_index.py

@@ -3,7 +3,7 @@ import pytest
 from dvc.exceptions import DownloadError, UploadError
 from dvc.fs import LocalFileSystem
 from dvc.utils.fs import remove
-from dvc_data.db import get_index
+from dvc_data.hashfile.db import get_index
 
 
 @pytest.fixture

tests/func/experiments/test_set_params.py

@@ -2,6 +2,8 @@ import sys
 
 import pytest
 
+from dvc.exceptions import InvalidArgumentError
+
 from ..utils.test_hydra import hydra_setup
 
 
@@ -85,3 +87,57 @@ def test_hydra_compose_and_dump(
             "goo": {"bag": 3.0},
             "lorem": False,
         }
+
+
+@pytest.mark.parametrize(
+    "hydra_enabled,overrides,expected",
+    [
+        (
+            pytest.param(
+                True,
+                marks=pytest.mark.xfail(
+                    sys.version_info >= (3, 11), reason="unsupported on 3.11"
+                ),
+            ),
+            ["db=mysql,postgresql"],
+            [
+                {"params.yaml": ["db=mysql"]},
+                {"params.yaml": ["db=postgresql"]},
+            ],
+        ),
+        (
+            False,
+            ["foo=bar,baz"],
+            [{"params.yaml": ["foo=bar"]}, {"params.yaml": ["foo=baz"]}],
+        ),
+    ],
+)
+def test_hydra_sweep(
+    tmp_dir, params_repo, dvc, mocker, hydra_enabled, overrides, expected
+):
+    patched = mocker.patch.object(dvc.experiments, "queue_one")
+
+    if hydra_enabled:
+        hydra_setup(
+            tmp_dir,
+            config_dir="conf",
+            config_name="config",
+        )
+        with dvc.config.edit() as conf:
+            conf["hydra"]["enabled"] = True
+
+    dvc.experiments.run(params=overrides, queue=True)
+
+    assert patched.call_count == len(expected)
+    for e in expected:
+        patched.assert_any_call(
+            mocker.ANY,
+            params=e,
+            reset=True,
+            targets=None,
+        )
+
+
+def test_hydra_sweep_requires_queue(params_repo, dvc):
+    with pytest.raises(InvalidArgumentError):
+        dvc.experiments.run(params=["db=mysql,postgresql"])

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.cli import main
 from dvc.dvcfile import DVC_FILE_SUFFIX
 from dvc.exceptions import (
@@ -103,7 +104,7 @@ def test_add_unsupported_file(dvc):
 
 
 def test_add_directory(tmp_dir, dvc):
-    from dvc_data import load
+    from dvc_data.hashfile import load
 
     (stage,) = tmp_dir.dvc_gen({"dir": {"file": "file"}})
 
@@ -457,7 +458,7 @@ class TestAddCommit(TestDvc):
 
 def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     tmp_dir.gen({"data/data": "foo"})
-    counter = mocker.spy(dvc_data.build, "_build_tree")
+    counter = mocker.spy(dvc_data.hashfile.build, "_build_tree")
     ret = main(["add", "data"])
     assert ret == 0
     assert counter.mock.call_count == 3
@@ -881,7 +882,7 @@ def test_add_with_cache_link_error(tmp_dir, dvc, mocker, capsys):
     tmp_dir.gen("foo", "foo")
 
     mocker.patch(
-        "dvc_data.checkout.test_links",
+        "dvc_data.hashfile.checkout.test_links",
         return_value=[],
     )
     dvc.add("foo")
@@ -903,6 +904,10 @@ def test_add_preserve_fields(tmp_dir, dvc):
         outs:
         - path: foo # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
           remote: testremote
         meta: some metadata
     """
@@ -916,6 +921,10 @@ def test_add_preserve_fields(tmp_dir, dvc):
         outs:
         - path: foo # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
           remote: testremote
           md5: acbd18db4cc2f85cedef654fccc4a4d8
           size: 3
@@ -1161,3 +1170,23 @@ def test_add_ignore_duplicated_targets(tmp_dir, dvc, capsys):
     _, err = capsys.readouterr()
     assert len(stages) == 3
     assert "ignoring duplicated targets: foo, bar" in err
+
+
+def test_add_with_annotations(M, tmp_dir, dvc):
+    tmp_dir.gen("foo", "foo")
+
+    annot = {
+        "desc": "foo desc",
+        "labels": ["l1", "l2"],
+        "type": "t1",
+        "meta": {"key": "value"},
+    }
+    (stage,) = dvc.add("foo", **annot)
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+    # try to selectively update/overwrite some annotations
+    annot = {**annot, "type": "t2"}
+    (stage,) = dvc.add("foo", type="t2")
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])

tests/func/test_checkout.py

@@ -90,7 +90,7 @@ class TestCheckoutCorruptedCacheFile(TestRepro):
 
 class TestCheckoutCorruptedCacheDir(TestDvc):
     def test(self):
-        from dvc_data import load
+        from dvc_data.hashfile import load
 
         # NOTE: using 'copy' so that cache and link don't have same inode
         ret = main(["config", "cache.type", "copy"])

tests/func/test_commit.py

@@ -47,6 +47,13 @@ def test_commit_preserve_fields(tmp_dir, dvc):
         outs:
         - path: foo # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
+          meta:
+            key1: value1
+            key2: value2
           remote: testremote
         meta: some metadata
     """
@@ -61,6 +68,13 @@ def test_commit_preserve_fields(tmp_dir, dvc):
         outs:
         - path: foo # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
+          meta:
+            key1: value1
+            key2: value2
           remote: testremote
           md5: acbd18db4cc2f85cedef654fccc4a4d8
           size: 3

tests/func/test_data_cloud.py

@@ -13,8 +13,8 @@ from dvc.testing.test_remote import (  # noqa, pylint: disable=unused-import
     TestRemote,
 )
 from dvc.utils.fs import remove
-from dvc_data.db.local import LocalHashFileDB
 from dvc_data.hashfile.db import HashFileDB
+from dvc_data.hashfile.db.local import LocalHashFileDB
 
 
 def test_cloud_cli(tmp_dir, dvc, remote, mocker):

tests/func/test_data_ls.py

@@ -0,0 +1,31 @@
+from operator import itemgetter
+
+from dvc.repo.data import ls
+
+
+def test_data_ls(tmp_dir, dvc):
+    assert not list(ls(dvc))
+
+    tmp_dir.dvc_gen("bar", "bar")
+    tmp_dir.gen("foo", "foo")
+    dvc.add(
+        "foo",
+        meta={"key": "value"},
+        labels=["l1", "l2"],
+        type="t1",
+        desc="foo",
+    )
+
+    foo_entry = {
+        "path": "foo",
+        "desc": "foo",
+        "type": "t1",
+        "labels": ["l1", "l2"],
+        "meta": {"key": "value"},
+    }
+    assert sorted(ls(dvc), key=itemgetter("path")) == [
+        {"path": "bar"},
+        foo_entry,
+    ]
+    assert list(ls(dvc, targets=["foo"])) == [foo_entry]
+    assert list(ls(dvc, targets=["foo"], recursive=True)) == [foo_entry]

tests/func/test_data_status.py

@@ -4,7 +4,7 @@ from os.path import join
 import pytest
 
 from dvc.repo import Repo
-from dvc.repo.data import _transform_git_paths_to_dvc
+from dvc.repo.data import _transform_git_paths_to_dvc, posixpath_to_os_path
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils.fs import remove
 
@@ -31,7 +31,7 @@ def test_git_to_dvc_path_wdir_transformation(tmp_dir, scm, path):
         subdir.gen(struct)
         _, _, untracked = scm.status(untracked_files="all")
         # make order independent of the platforms for easier test assertions
-        untracked = sorted(untracked, reverse=True)
+        untracked = sorted(map(posixpath_to_os_path, untracked), reverse=True)
         assert _transform_git_paths_to_dvc(dvc, untracked) == [
             "file",
             join("dir", "foo"),

tests/func/test_dvcfile.py

@@ -3,6 +3,7 @@ import textwrap
 
 import pytest
 
+from dvc.annotations import Annotation
 from dvc.dvcfile import (
     PIPELINE_FILE,
     PIPELINE_LOCK,
@@ -335,7 +336,7 @@ def test_dvcfile_dump_preserves_desc(tmp_dir, dvc, run_copy):
     (tmp_dir / dvcfile.path).dump(data)
 
     assert stage.desc == stage_desc
-    stage.outs[0].desc = out_desc
+    stage.outs[0].annot.desc = out_desc
     dvcfile.dump(stage)
     loaded = dvcfile._load()[0]
     assert loaded == data
@@ -382,3 +383,31 @@ def test_dvcfile_try_dumping_parametrized_stage(tmp_dir, dvc, data, name):
         dvcfile.dump(stage)
 
     assert str(exc.value) == f"cannot dump a parametrized stage: '{name}'"
+
+
+def test_dvcfile_load_dump_stage_with_desc_meta(tmp_dir, dvc):
+    data = {
+        "stages": {
+            "stage1": {
+                "cmd": "cp foo bar",
+                "desc": "stage desc",
+                "meta": {"key1": "value1", "key2": "value2"},
+                "deps": ["foo"],
+                "outs": [
+                    {"bar": {"desc": "bar desc", "meta": {"key": "value"}}}
+                ],
+            }
+        }
+    }
+    (tmp_dir / "dvc.yaml").dump(data)
+
+    stage = dvc.stage.load_one(name="stage1")
+    assert stage.meta == {"key1": "value1", "key2": "value2"}
+    assert stage.desc == "stage desc"
+    assert stage.outs[0].annot == Annotation(
+        desc="bar desc", meta={"key": "value"}
+    )
+
+    # sanity check
+    stage.dump()
+    assert (tmp_dir / "dvc.yaml").parse() == data

tests/func/test_external_repo.py

@@ -7,8 +7,8 @@ from dvc.external_repo import CLONES, external_repo
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils import relpath
 from dvc.utils.fs import makedirs, remove
-from dvc_data.build import build
-from dvc_data.transfer import transfer
+from dvc_data.hashfile.build import build
+from dvc_data.hashfile.transfer import transfer
 
 
 def test_external_repo(erepo_dir, mocker):

tests/func/test_gc.py

@@ -11,7 +11,7 @@ from dvc.exceptions import CollectCacheError
 from dvc.fs import LocalFileSystem
 from dvc.repo import Repo as DvcRepo
 from dvc.utils.fs import remove
-from dvc_data.db.local import LocalHashFileDB
+from dvc_data.hashfile.db.local import LocalHashFileDB
 from tests.basic_env import TestDir, TestDvcGit
 
 

tests/func/test_ignore.py

@@ -10,7 +10,7 @@ from dvc.pathspec_math import PatternInfo, merge_patterns
 from dvc.repo import Repo
 from dvc.testing.tmp_dir import TmpDir
 from dvc.types import List
-from dvc_data.build import IgnoreInCollectedDirError
+from dvc_data.hashfile.build import IgnoreInCollectedDirError
 from dvc_data.hashfile.utils import get_mtime_and_size
 
 

tests/func/test_import.py

@@ -6,6 +6,7 @@ import pytest
 from funcy import first
 from scmrepo.git import Git
 
+from dvc.annotations import Annotation
 from dvc.config import NoRemoteError
 from dvc.dvcfile import Dvcfile
 from dvc.exceptions import DownloadError, PathMissingError
@@ -527,7 +528,7 @@ def test_import_with_no_exec(tmp_dir, dvc, erepo_dir):
 
 
 def test_import_with_jobs(mocker, dvc, erepo_dir):
-    import dvc_data.transfer as otransfer
+    import dvc_data.hashfile.transfer as otransfer
 
     with erepo_dir.chdir():
         erepo_dir.dvc_gen(
@@ -627,3 +628,26 @@ def test_parameterized_repo(tmp_dir, dvc, scm, erepo_dir, paths):
         "url": os.fspath(erepo_dir),
         "rev_lock": erepo_dir.scm.get_rev(),
     }
+
+
+def test_import_with_annotations(M, tmp_dir, scm, dvc, erepo_dir):
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("foo", "foo content", commit="create foo")
+
+    annot = {
+        "desc": "foo desc",
+        "labels": ["l1", "l2"],
+        "type": "t1",
+        "meta": {"key": "value"},
+    }
+    stage = dvc.imp(os.fspath(erepo_dir), "foo", "foo", no_exec=True, **annot)
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+    # try to selectively update/overwrite some annotations
+    annot = {**annot, "type": "t2"}
+    stage = dvc.imp(
+        os.fspath(erepo_dir), "foo", "foo", no_exec=True, type="t2"
+    )
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])

tests/func/test_import_url.py

@@ -5,8 +5,9 @@ from uuid import uuid4
 
 import pytest
 
+from dvc.annotations import Annotation
 from dvc.cli import main
-from dvc.dependency.base import DependencyDoesNotExistError
+from dvc.dependency.base import Dependency, DependencyDoesNotExistError
 from dvc.exceptions import InvalidArgumentError
 from dvc.stage import Stage
 from dvc.testing.test_workspace import TestImport as _TestImport
@@ -132,7 +133,7 @@ class TestImport(_TestImport):
         return False
 
 
-def test_import_url_preserve_meta(tmp_dir, dvc):
+def test_import_url_preserve_fields(tmp_dir, dvc):
     text = textwrap.dedent(
         """\
         # top comment
@@ -142,6 +143,12 @@ def test_import_url_preserve_meta(tmp_dir, dvc):
         outs:
         - path: bar # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
+          meta:
+            key: value
         meta: some metadata
     """
     )
@@ -160,6 +167,12 @@ def test_import_url_preserve_meta(tmp_dir, dvc):
         outs:
         - path: bar # out comment
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
+          meta:
+            key: value
           md5: acbd18db4cc2f85cedef654fccc4a4d8
           size: 3
         meta: some metadata
@@ -281,11 +294,57 @@ def test_import_url_no_download(tmp_dir, dvc, local_workspace):
     assert not dst.exists()
 
     out = stage.outs[0]
-    assert out.hash_info.value is None
-    assert out.hash_info.name is None
+    assert not out.hash_info
     assert out.meta.size is None
 
     status = dvc.status()
     assert status["file.dvc"] == [
         {"changed outs": {"file": "deleted"}},
     ]
+
+
+def test_imp_url_with_annotations(M, tmp_dir, dvc, local_workspace):
+    local_workspace.gen("foo", "foo")
+    annot = {
+        "desc": "foo desc",
+        "labels": ["l1", "l2"],
+        "type": "t1",
+        "meta": {"key": "value"},
+    }
+    stage = dvc.imp_url(
+        "remote://workspace/foo",
+        os.fspath(tmp_dir / "foo"),
+        no_exec=True,
+        **annot,
+    )
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+    # try to selectively update/overwrite some annotations
+    annot = {**annot, "type": "t2"}
+    stage = dvc.imp_url(
+        "remote://workspace/foo",
+        os.fspath(tmp_dir / "foo"),
+        no_exec=True,
+        type="t2",
+    )
+    assert stage.outs[0].annot == Annotation(**annot)
+    assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+
+def test_import_url_fs_config(tmp_dir, dvc, workspace, mocker):
+    import dvc.fs as dvc_fs
+
+    workspace.gen("foo", "foo")
+
+    url = "remote://workspace/foo"
+    get_fs_config = mocker.spy(dvc_fs, "get_fs_config")
+    dep_init = mocker.spy(Dependency, "__init__")
+    dvc.imp_url(url, fs_config={"jobs": 42})
+
+    dep_init_kwargs = dep_init.call_args[1]
+    assert dep_init_kwargs.get("fs_config") == {"jobs": 42}
+
+    assert get_fs_config.call_args_list[0][1] == {"url": "foo"}
+    assert get_fs_config.call_args_list[1][1] == {"url": url, "jobs": 42}
+    assert get_fs_config.call_args_list[2][1] == {"name": "workspace"}

tests/func/test_odb.py

@@ -43,7 +43,7 @@ def test_cache(tmp_dir, dvc):
 
 
 def test_cache_load_bad_dir_cache(tmp_dir, dvc):
-    from dvc_data import load
+    from dvc_data.hashfile import load
 
     dir_hash = "123.dir"
     fname = os.fspath(dvc.odb.local.oid_to_path(dir_hash))

tests/func/test_remote.py

@@ -139,8 +139,8 @@ def test_upper_case_remote(tmp_dir, dvc, local_cloud):
 
 
 def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
-    from dvc_data.build import build
-    from dvc_data.objects.tree import Tree
+    from dvc_data.hashfile.build import build
+    from dvc_data.hashfile.tree import Tree
 
     tmp_dir.gen({"data": {"1": "1 content", "2": "2 content"}})
 
@@ -150,7 +150,9 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
         [{"relpath": "1", "md5": "1"}, {"relpath": "2", "md5": "2"}]
     )
     tree.digest()
-    with patch("dvc_data.build._build_tree", return_value=(None, tree)):
+    with patch(
+        "dvc_data.hashfile.build._build_tree", return_value=(None, tree)
+    ):
         _, _, obj = build(dvc.odb.local, path, dvc.odb.local.fs, "md5")
         hash1 = obj.hash_info
 
@@ -161,7 +163,9 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
         [{"md5": "1", "relpath": "1"}, {"md5": "2", "relpath": "2"}]
     )
     tree.digest()
-    with patch("dvc_data.build._build_tree", return_value=(None, tree)):
+    with patch(
+        "dvc_data.hashfile.build._build_tree", return_value=(None, tree)
+    ):
         _, _, obj = build(dvc.odb.local, path, dvc.odb.local.fs, "md5")
         hash2 = obj.hash_info
 

tests/func/test_run_single_stage.py

@@ -972,6 +972,12 @@ def test_run_force_preserves_comments_and_meta(tmp_dir, dvc, run_copy):
       # comment preserved
       - path: bar
         desc: out desc
+        type: mytype
+        labels:
+        - label1
+        - label2
+        meta:
+          key: value
       meta:
         name: copy-foo-bar
     """
@@ -996,6 +1002,12 @@ def test_run_force_preserves_comments_and_meta(tmp_dir, dvc, run_copy):
         # comment preserved
         - path: bar
           desc: out desc
+          type: mytype
+          labels:
+          - label1
+          - label2
+          meta:
+            key: value
           md5: acbd18db4cc2f85cedef654fccc4a4d8
           size: 3
         meta:

tests/func/test_stage.py

@@ -3,6 +3,7 @@ import tempfile
 
 import pytest
 
+from dvc.annotations import Annotation
 from dvc.cli import main
 from dvc.dvcfile import SingleStageFile
 from dvc.fs import LocalFileSystem
@@ -10,6 +11,8 @@ from dvc.output import Output
 from dvc.repo import Repo, lock_repo
 from dvc.stage import PipelineStage, Stage
 from dvc.stage.run import run_stage
+from dvc.stage.utils import compute_md5
+from dvc.utils import dict_md5
 from dvc.utils.serialize import dump_yaml, load_yaml
 from dvc.utils.strictyaml import YAMLValidationError
 from tests.basic_env import TestDvc
@@ -164,40 +167,68 @@ def test_md5_ignores_comments(tmp_dir, dvc):
     assert not new_stage.changed_stage()
 
 
-def test_meta_is_preserved(tmp_dir, dvc):
-    (stage,) = tmp_dir.dvc_gen("foo", "foo content")
-
-    # Add meta to DVC-file
-    data = (tmp_dir / stage.path).parse()
-    data["meta"] = {"custom_key": 42}
-    (tmp_dir / stage.path).dump(data)
-
-    # Loading and dumping to test that it works and meta is retained
-    dvcfile = SingleStageFile(dvc, stage.path)
-    new_stage = dvcfile.stage
-    dvcfile.dump(new_stage)
-
-    new_data = (tmp_dir / stage.path).parse()
-    assert new_data["meta"] == data["meta"]
+def test_md5_ignores_annotations(tmp_dir, dvc):
+    data = {
+        "desc": "stage desc",
+        "meta": {"key1": "value1", "key2": "value2"},
+        "outs": [
+            {
+                "md5": "d3b07384d113edec49eaa6238ad5ff00",
+                "size": 4,
+                "path": "foo",
+                "desc": "foo desc",
+                "type": "mytype",
+                "labels": ["get-started", "dataset-registry"],
+                "meta": {"key1": "value1"},
+            }
+        ],
+    }
+    (tmp_dir / "foo.dvc").dump(data)
+    stage = dvc.stage.load_one("foo.dvc")
+    assert compute_md5(stage) == "1822617147b53ae6f9eb4b3c87c0b6f3"
+    assert (
+        dict_md5(
+            {
+                "outs": [
+                    {"md5": "d3b07384d113edec49eaa6238ad5ff00", "path": "foo"}
+                ]
+            }
+        )
+        == "1822617147b53ae6f9eb4b3c87c0b6f3"
+    )
 
 
-def test_desc_is_preserved(tmp_dir, dvc):
-    (stage,) = tmp_dir.dvc_gen("foo", "foo content")
+def test_meta_desc_is_preserved(tmp_dir, dvc):
+    data = {
+        "desc": "stage desc",
+        "meta": {"key1": "value1", "key2": "value2"},
+        "outs": [
+            {
+                "md5": "d3b07384d113edec49eaa6238ad5ff00",
+                "size": 4,
+                "path": "foo",
+                "desc": "foo desc",
+                "type": "mytype",
+                "labels": ["get-started", "dataset-registry"],
+                "meta": {"key": "value"},
+            }
+        ],
+    }
+    (tmp_dir / "foo.dvc").dump(data)
+    stage = dvc.stage.load_one("foo.dvc")
 
-    data = (tmp_dir / stage.path).parse()
-    stage_desc = "test stage description"
-    out_desc = "test out description"
-    data["desc"] = stage_desc
-    data["outs"][0]["desc"] = out_desc
-    (tmp_dir / stage.path).dump(data)
+    assert stage.meta == {"key1": "value1", "key2": "value2"}
+    assert stage.desc == "stage desc"
+    assert stage.outs[0].annot == Annotation(
+        desc="foo desc",
+        type="mytype",
+        labels=["get-started", "dataset-registry"],
+        meta={"key": "value"},
+    )
 
-    dvcfile = SingleStageFile(dvc, stage.path)
-    new_stage = dvcfile.stage
-    dvcfile.dump(new_stage)
-
-    new_data = (tmp_dir / stage.path).parse()
-    assert new_data["desc"] == stage_desc
-    assert new_data["outs"][0]["desc"] == out_desc
+    # sanity check
+    stage.dump()
+    assert (tmp_dir / "foo.dvc").parse() == data
 
 
 def test_parent_repo_collect_stages(tmp_dir, scm, dvc):

tests/func/test_stage_load.py

@@ -116,23 +116,19 @@ def stages(tmp_dir, run_copy):
 
 
 def test_collect_not_a_group_stage_with_group_flag(tmp_dir, dvc, stages):
-    assert set(dvc.stage.collect("copy-bar-foobar", accept_group=True)) == {
+    assert set(dvc.stage.collect("copy-bar-foobar")) == {
         stages["copy-bar-foobar"]
     }
-    assert set(
-        dvc.stage.collect("copy-bar-foobar", accept_group=True, with_deps=True)
-    ) == {
+    assert set(dvc.stage.collect("copy-bar-foobar", with_deps=True)) == {
         stages["copy-bar-foobar"],
         stages["copy-foo-bar"],
         stages["foo-generate"],
     }
+    assert set(dvc.stage.collect_granular("copy-bar-foobar")) == {
+        (stages["copy-bar-foobar"], None)
+    }
     assert set(
-        dvc.stage.collect_granular("copy-bar-foobar", accept_group=True)
-    ) == {(stages["copy-bar-foobar"], None)}
-    assert set(
-        dvc.stage.collect_granular(
-            "copy-bar-foobar", accept_group=True, with_deps=True
-        )
+        dvc.stage.collect_granular("copy-bar-foobar", with_deps=True)
     ) == {
         (stages["copy-bar-foobar"], None),
         (stages["copy-foo-bar"], None),
@@ -153,11 +149,8 @@ def test_collect_generated(tmp_dir, dvc):
     assert len(all_stages) == 5
 
     assert set(dvc.stage.collect()) == all_stages
-    assert set(dvc.stage.collect("build", accept_group=True)) == all_stages
-    assert (
-        set(dvc.stage.collect("build", accept_group=True, with_deps=True))
-        == all_stages
-    )
+    assert set(dvc.stage.collect("build")) == all_stages
+    assert set(dvc.stage.collect("build", with_deps=True)) == all_stages
     assert set(dvc.stage.collect("build*", glob=True)) == all_stages
     assert (
         set(dvc.stage.collect("build*", glob=True, with_deps=True))
@@ -165,17 +158,9 @@ def test_collect_generated(tmp_dir, dvc):
     )
 
     stages_info = {(stage, None) for stage in all_stages}
+    assert set(dvc.stage.collect_granular("build")) == stages_info
     assert (
-        set(dvc.stage.collect_granular("build", accept_group=True))
-        == stages_info
-    )
-    assert (
-        set(
-            dvc.stage.collect_granular(
-                "build", accept_group=True, with_deps=True
-            )
-        )
-        == stages_info
+        set(dvc.stage.collect_granular("build", with_deps=True)) == stages_info
     )
 
 

tests/func/utils/test_hydra.py

@@ -178,3 +178,54 @@ def test_compose_and_dump(tmp_dir, suffix, overrides, expected):
     output_file = tmp_dir / f"params.{suffix}"
     compose_and_dump(output_file, config_dir, config_name, overrides)
     assert output_file.parse() == expected
+
+
+@pytest.mark.parametrize(
+    "overrides, expected",
+    [
+        (
+            {"params.yaml": ["foo=1,2", "bar=3,4"]},
+            [
+                {"params.yaml": ["foo=1", "bar=3"]},
+                {"params.yaml": ["foo=1", "bar=4"]},
+                {"params.yaml": ["foo=2", "bar=3"]},
+                {"params.yaml": ["foo=2", "bar=4"]},
+            ],
+        ),
+        (
+            {"params.yaml": ["foo=choice(1,2)"]},
+            [{"params.yaml": ["foo=1"]}, {"params.yaml": ["foo=2"]}],
+        ),
+        (
+            {"params.yaml": ["foo=range(1, 3)"]},
+            [{"params.yaml": ["foo=1"]}, {"params.yaml": ["foo=2"]}],
+        ),
+        (
+            {"params.yaml": ["foo=1,2"], "others.yaml": ["bar=3"]},
+            [
+                {"params.yaml": ["foo=1"], "others.yaml": ["bar=3"]},
+                {"params.yaml": ["foo=2"], "others.yaml": ["bar=3"]},
+            ],
+        ),
+        (
+            {"params.yaml": ["foo=1,2"], "others.yaml": ["bar=3,4"]},
+            [
+                {"params.yaml": ["foo=1"], "others.yaml": ["bar=3"]},
+                {"params.yaml": ["foo=1"], "others.yaml": ["bar=4"]},
+                {"params.yaml": ["foo=2"], "others.yaml": ["bar=3"]},
+                {"params.yaml": ["foo=2"], "others.yaml": ["bar=4"]},
+            ],
+        ),
+    ],
+)
+def test_hydra_sweeps(overrides, expected):
+    from dvc.utils.hydra import get_hydra_sweeps
+
+    assert get_hydra_sweeps(overrides) == expected
+
+
+def test_invalid_sweep():
+    from dvc.utils.hydra import get_hydra_sweeps
+
+    with pytest.raises(InvalidArgumentError):
+        get_hydra_sweeps({"params.yaml": ["foo=glob(*)"]})

tests/unit/cli/test_main.py

@@ -4,7 +4,7 @@ import pytest
 from funcy import raiser
 
 from dvc.cli import main
-from dvc_data.build import IgnoreInCollectedDirError
+from dvc_data.hashfile.build import IgnoreInCollectedDirError
 from dvc_data.hashfile.cache import DiskError
 from dvc_objects.fs.base import FileSystem, RemoteMissingDepsError
 

tests/unit/command/test_add.py

@@ -37,6 +37,9 @@ def test_add(mocker, dvc):
         remote=None,
         to_remote=False,
         desc="stage description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=None,
     )
 
@@ -71,6 +74,9 @@ def test_add_to_remote(mocker):
         remote="remote",
         to_remote=True,
         desc=None,
+        type=None,
+        labels=None,
+        meta=None,
         jobs=None,
     )
 

tests/unit/command/test_annotations_flags.py

@@ -0,0 +1,47 @@
+import pytest
+
+from dvc.cli import parse_args
+from tests.utils.asserts import called_once_with_subset
+
+
+@pytest.mark.parametrize(
+    "func, args",
+    [
+        ("add", ("add", "model.pkl")),
+        ("imp", ("import", "https://github.com/org/repo", "model.pkl")),
+        ("imp_url", ("import-url", "https://dvc.org/data.xml", "data.xml")),
+    ],
+    ids=["add", "import", "import-url"],
+)
+def test_add_annotations(mocker, func, args):
+    mocker.patch("dvc.repo.Repo")
+    cli_args = parse_args(
+        [
+            *args,
+            "--label",
+            "example-get-started",
+            "--label",
+            "model-registry",
+            "--type",
+            "model",
+            "--desc",
+            "description",
+            "--meta",
+            "key1=value1",
+            "--meta",
+            "key2=value2",
+        ]
+    )
+
+    _, *posargs = args
+    cmd = cli_args.func(cli_args)
+
+    assert cmd.run() == 0
+    assert called_once_with_subset(
+        getattr(cmd.repo, func),
+        posargs,
+        desc="description",
+        type="model",
+        labels=["example-get-started", "model-registry"],
+        meta={"key1": "value1", "key2": "value2"},
+    )

tests/unit/command/test_compat_flag.py

@@ -15,6 +15,7 @@ def _id_gen(val) -> str:
     "args, key",
     [
         (["dag", "--show-md"], "markdown"),
+        (["data", "ls", "--show-md"], "markdown"),
         (["diff", "--show-json"], "json"),
         (["diff", "--show-md"], "markdown"),
         (["experiments", "diff", "--show-json"], "json"),

tests/unit/command/test_data_ls.py

@@ -0,0 +1,96 @@
+import pytest
+
+from dvc.cli import parse_args
+from dvc.commands.data import CmdDataLs
+
+
+def test_cli(mocker):
+    mocker.patch("dvc.repo.Repo")
+    ls = mocker.patch("dvc.repo.data.ls", return_value={})
+    show_table_spy = mocker.spy(CmdDataLs, "_show_table")
+
+    cli_args = parse_args(
+        [
+            "data",
+            "ls",
+            "--labels",
+            "label1,label2",
+            "--labels",
+            "label3",
+            "--type",
+            "type1,type2",
+            "--type",
+            "type3",
+            "target",
+            "--recursive",
+            "--md",
+        ]
+    )
+
+    assert cli_args.func == CmdDataLs
+    cmd = cli_args.func(cli_args)
+    assert cmd.run() == 0
+
+    ls.assert_called_once_with(
+        cmd.repo,
+        targets=["target"],
+        recursive=True,
+    )
+    show_table_spy.assert_called_once_with(
+        {},
+        filter_labels={"label1", "label2", "label3"},
+        filter_types={"type1", "type2", "type3"},
+        markdown=True,
+    )
+
+
+EXAMPLE_DATA = [
+    {
+        "path": "model.pkl",
+        "desc": "desc",
+        "type": "model",
+        "labels": ["get-started", "dataset-registry"],
+        "meta": {"key": "value", "key1": "value1"},
+    },
+    {
+        "path": "model.pkl",
+        "desc": "desc",
+        "type": "model",
+        "labels": ["get-started", "example"],
+    },
+    {"path": "unlabeled.txt"},
+]
+
+
+@pytest.mark.parametrize(
+    "markdown, expected_output",
+    [
+        (
+            True,
+            """\
+| Path      | Type   | Labels                       | Description   |
+|-----------|--------|------------------------------|---------------|
+| model.pkl | model  | get-started,dataset-registry | desc          |
+| model.pkl | model  | get-started,example          | desc          |\n""",
+        ),
+        (
+            False,
+            """\
+ Path       Type   Labels                        Description
+ model.pkl  model  get-started,dataset-registry  desc
+ model.pkl  model  get-started,example           desc""",
+        ),
+    ],
+    ids=["markdown", "default"],
+)
+def test_ls(capsys, markdown, expected_output):
+    CmdDataLs._show_table(
+        EXAMPLE_DATA,
+        filter_types={"model"},
+        filter_labels={"get-started", "dataset-registry"},
+        markdown=markdown,
+    )
+    out, err = capsys.readouterr()
+    assert not err
+    out = "\n".join(line.rstrip() for line in out.splitlines())
+    assert out == expected_output

tests/unit/command/test_imp.py

@@ -36,6 +36,9 @@ def test_import(mocker):
         no_exec=False,
         no_download=False,
         desc="description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=3,
     )
 
@@ -72,6 +75,9 @@ def test_import_no_exec(mocker):
         no_exec=True,
         no_download=False,
         desc="description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=None,
     )
 
@@ -108,5 +114,8 @@ def test_import_no_download(mocker):
         no_exec=False,
         no_download=True,
         desc="description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=None,
     )

tests/unit/command/test_imp_url.py

@@ -37,6 +37,9 @@ def test_import_url(mocker):
         remote=None,
         to_remote=False,
         desc="description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=4,
     )
 
@@ -92,6 +95,9 @@ def test_import_url_no_exec_download_flags(mocker, flag, expected):
         to_remote=False,
         desc="description",
         jobs=None,
+        type=None,
+        labels=None,
+        meta=None,
         **expected
     )
 
@@ -125,6 +131,9 @@ def test_import_url_to_remote(mocker):
         remote="remote",
         to_remote=True,
         desc="description",
+        type=None,
+        labels=None,
+        meta=None,
         jobs=None,
     )
 

tests/unit/data/db/test_local.py

@@ -4,7 +4,7 @@ import os
 import pytest
 
 from dvc.fs import LocalFileSystem
-from dvc_data.db.local import LocalHashFileDB
+from dvc_data.hashfile.db.local import LocalHashFileDB
 from dvc_data.hashfile.hash_info import HashInfo
 
 
@@ -13,7 +13,7 @@ def test_status_download_optimization(mocker, dvc):
     And the desired files to fetch are already on the local cache,
     Don't check the existence of the desired files on the remote cache
     """
-    from dvc_data.status import compare_status
+    from dvc_data.hashfile.status import compare_status
 
     odb = LocalHashFileDB(LocalFileSystem(), os.getcwd())
     obj_ids = {
@@ -85,9 +85,9 @@ def test_set_exec_ignore_errors(tmp_dir, dvc, mocker, err):
 
 
 def test_staging_file(tmp_dir, dvc):
-    from dvc_data import check
-    from dvc_data.build import build
-    from dvc_data.transfer import transfer
+    from dvc_data.hashfile import check
+    from dvc_data.hashfile.build import build
+    from dvc_data.hashfile.transfer import transfer
 
     tmp_dir.gen("foo", "foo")
     fs = LocalFileSystem()
@@ -113,9 +113,9 @@ def test_staging_file(tmp_dir, dvc):
 
 
 def test_staging_dir(tmp_dir, dvc):
-    from dvc_data import check
-    from dvc_data.build import build
-    from dvc_data.transfer import transfer
+    from dvc_data.hashfile import check
+    from dvc_data.hashfile.build import build
+    from dvc_data.hashfile.transfer import transfer
 
     tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
     fs = LocalFileSystem()

tests/unit/fs/test_data.py

@@ -6,7 +6,7 @@ import pytest
 import dvc_data
 from dvc.fs.data import DataFileSystem
 from dvc.utils.fs import remove
-from dvc_data.build import build
+from dvc_data.hashfile.build import build
 from dvc_data.hashfile.hash_info import HashInfo
 
 

tests/unit/fs/test_dvc.py

@@ -7,7 +7,7 @@ import pytest
 
 from dvc.fs.dvc import DvcFileSystem
 from dvc.testing.tmp_dir import make_subrepo
-from dvc_data.build import build
+from dvc_data.hashfile.build import build
 from dvc_data.hashfile.hash_info import HashInfo
 
 
@@ -566,7 +566,7 @@ def test_get_hash_mixed_dir(tmp_dir, scm, dvc):
 
 
 def test_get_hash_dirty_file(tmp_dir, dvc):
-    from dvc_data import check
+    from dvc_data.hashfile import check
     from dvc_data.hashfile.hash import hash_file
 
     tmp_dir.dvc_gen("file", "file")

tests/unit/output/test_annotations.py

@@ -0,0 +1,29 @@
+import pytest
+
+from dvc.annotations import Annotation
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"desc": "desc", "type": "type", "labels": ["label1", "label2"]},
+        {"desc": "desc", "type": "type", "meta": {"key": "value"}},
+    ],
+)
+def test_annotation_to_dict(kwargs):
+    annot = Annotation(**kwargs)
+    assert annot.to_dict() == kwargs
+
+
+def test_annotation_update():
+    annot = Annotation(desc="desc", labels=["label1", "label2"])
+    annot.update(
+        labels=["label"], type="type", new="new", meta={"key": "value"}
+    )
+
+    assert vars(annot) == {
+        "desc": "desc",
+        "type": "type",
+        "labels": ["label"],
+        "meta": {"key": "value"},
+    }

tests/unit/output/test_output.py

@@ -90,6 +90,8 @@ def test_get_used_objs(exists, expected_message, mocker, caplog):
     mocker.patch.object(
         stage.repo.dvcignore, "check_ignore", return_value=_no_match("path")
     )
+    stage.repo.fs.version_aware = False
+    stage.repo.fs.PARAM_CHECKSUM = "md5"
 
     output = Output(stage, "path")
 

tests/unit/remote/test_index.py

@@ -3,7 +3,7 @@ import os
 import pytest
 from funcy import first
 
-from dvc_data.db.index import ObjectDBIndex
+from dvc_data.hashfile.db.index import ObjectDBIndex
 
 
 @pytest.fixture

tests/unit/remote/test_remote.py

@@ -5,7 +5,7 @@ from dvc_gs import GSFileSystem
 from dvc_s3 import S3FileSystem
 
 from dvc.fs import get_cloud_fs
-from dvc_data.db import get_index
+from dvc_data.hashfile.db import get_index
 
 
 def test_remote_with_hash_jobs(dvc):

tests/unit/repo/experiments/conftest.py

@@ -24,6 +24,10 @@ def test_queue(tmp_dir, dvc, scm, mocker) -> LocalCeleryQueue:
     """
     celery_queue = dvc.experiments.celery_queue
     mocker.patch.object(celery_queue, "spawn_worker")
-    with _thread_worker(celery_queue.celery, concurrency=1) as worker:
+    with _thread_worker(
+        celery_queue.celery,
+        concurrency=1,
+        ping_task_timeout=30,
+    ) as worker:
         mocker.patch.object(celery_queue, "worker", return_value=worker)
         yield celery_queue

tests/utils/asserts.py

@@ -11,7 +11,7 @@ def called_once_with_subset(m: Mock, *args: Any, **kwargs: Any) -> bool:
     m.assert_called_once()
     m_args, m_kwargs = m.call_args
 
-    expected_args = m_args + (ANY,) * (len(m_args) - len(args))
+    expected_args = m_args + (ANY,) * (len(m_args) - len(args) - 1)
     expected_kwargs = {k: kwargs.get(k, ANY) for k in m_kwargs}
     m.assert_called_with(*expected_args, **expected_kwargs)