experiments: auto push experiments (#10323)

* add auto_push for experiment on run and on save.
add config.exp.auto_push and config.exp.git_remote

* add func tests

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* separate auto_push behaviour out of commit

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.2.0'
+    rev: 'v0.2.2'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

dvc/api/__init__.py

@@ -2,6 +2,7 @@ from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
 from .artifacts import artifacts_show
 from .data import get_url, open, read
+from .dataset import get as get_dataset
 from .experiments import exp_save, exp_show
 from .scm import all_branches, all_commits, all_tags
 from .show import metrics_show, params_show
@@ -14,6 +15,7 @@ __all__ = [
     "artifacts_show",
     "exp_save",
     "exp_show",
+    "get_dataset",
     "get_url",
     "metrics_show",
     "open",

dvc/api/dataset.py

@@ -0,0 +1,66 @@
+from typing import Literal, TypedDict, Union
+
+
+class DVCXDataset(TypedDict):
+    type: Literal["dvcx"]
+    name: str
+    version: int
+
+
+class DVCDataset(TypedDict):
+    type: Literal["dvc"]
+    url: str
+    path: str
+    sha: str
+
+
+class URLDataset(TypedDict):
+    type: Literal["url"]
+    files: list[str]
+    path: str
+
+
+def get(name: str) -> Union[DVCXDataset, DVCDataset, URLDataset]:
+    from difflib import get_close_matches
+
+    from dvc.fs import get_cloud_fs
+    from dvc.repo import Repo, datasets
+
+    repo = Repo()
+    try:
+        dataset = repo.datasets[name]
+    except datasets.DatasetNotFoundError as e:
+        add_note = getattr(e, "add_note", lambda _: None)
+        if matches := get_close_matches(name, repo.datasets):
+            add_note(f"Did you mean: {matches[0]!r}?")
+        raise
+
+    if dataset._invalidated:
+        raise ValueError(f"dataset not in sync. Sync with 'dvc ds update {name}'.")
+    if not dataset.lock:
+        raise ValueError("missing lock information")
+    if dataset.type == "dvc":
+        return DVCDataset(
+            type="dvc",
+            url=dataset.lock.url,
+            path=dataset.lock.path,
+            sha=dataset.lock.rev_lock,
+        )
+    if dataset.type == "dvcx":
+        return DVCXDataset(
+            type="dvcx", name=dataset.name_version[0], version=dataset.lock.version
+        )
+    if dataset.type == "url":
+        fs_cls, _, path = get_cloud_fs(repo.config, url=dataset.lock.url)
+        assert fs_cls
+        join_version = getattr(fs_cls, "join_version", lambda path, _: path)
+        protocol = fs_cls.protocol
+        versioned_path = join_version(path, dataset.lock.meta.version_id)
+        versioned_path = f"{protocol}://{versioned_path}"
+        files = [
+            join_version(
+                fs_cls.join(versioned_path, file.relpath), file.meta.version_id
+            )
+            for file in dataset.lock.files
+        ]
+        return URLDataset(type="url", files=files, path=versioned_path)

dvc/cli/parser.py

@@ -17,6 +17,7 @@ from dvc.commands import (
     dag,
     data,
     data_sync,
+    dataset,
     destroy,
     diff,
     du,
@@ -67,6 +68,7 @@ COMMANDS = [
     dag,
     data,
     data_sync,
+    dataset,
     destroy,
     diff,
     du,

dvc/commands/dataset.py

@@ -0,0 +1,225 @@
+from typing import TYPE_CHECKING, Optional
+
+from dvc.cli import formatter
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+if TYPE_CHECKING:
+    from rich.text import Text
+
+    from dvc.repo.datasets import Dataset, FileInfo
+
+logger = logger.getChild(__name__)
+
+
+def diff_files(old: list["FileInfo"], new: list["FileInfo"]) -> dict[str, list[str]]:
+    old_files = {d.relpath: d for d in old}
+    new_files = {d.relpath: d for d in new}
+    rest = old_files.keys() & new_files.keys()
+    return {
+        "added": list(new_files.keys() - old_files.keys()),
+        "deleted": list(old_files.keys() - new_files.keys()),
+        "modified": [p for p in rest if new_files[p] != old_files[p]],
+    }
+
+
+class CmdDatasetAdd(CmdBase):
+    @classmethod
+    def display(cls, name: str, dataset: "Dataset", action: str = "Adding"):
+        from dvc.ui import ui
+
+        assert dataset.lock
+
+        url = dataset.spec.url
+        ver: str = ""
+        if dataset.type == "dvcx":
+            ver = f"v{dataset.lock.version}"
+        if dataset.type == "dvc":
+            if dataset.lock.path:
+                url = f"{url}:/{dataset.lock.path.lstrip('/')}"
+            if rev := dataset.lock.rev:
+                ver = rev
+
+        ver_part: Optional["Text"] = None
+        if ver:
+            ver_part = ui.rich_text.assemble(" @ ", (ver, "repr.number"))
+        text = ui.rich_text.assemble("(", (url, "repr.url"), ver_part or "", ")")
+        ui.write(action, ui.rich_text(name, "cyan"), text, styled=True)
+
+    def run(self):
+        if not self.args.dvc and self.args.rev:
+            raise DvcException("--rev can't be used without --dvc")
+        if not self.args.dvc and self.args.path:
+            raise DvcException("--path can't be used without --dvc")
+
+        d = vars(self.args)
+        for key in ["dvc", "dvcx", "url"]:
+            if url := d.pop(key, None):
+                d.update({"type": key, "url": url})
+                break
+
+        existing = self.repo.datasets.get(self.args.name)
+        with self.repo.scm_context:
+            if not self.args.force and existing:
+                path = self.repo.fs.relpath(existing.manifest_path)
+                raise DvcException(
+                    f"{self.args.name} already exists in {path}, "
+                    "use the --force to overwrite"
+                )
+            dataset = self.repo.datasets.add(**d)
+            self.display(self.args.name, dataset)
+            return 0
+
+
+class CmdDatasetUpdate(CmdBase):
+    def display(self, name: str, dataset: "Dataset", new: "Dataset"):
+        from dvc.commands.checkout import log_changes
+        from dvc.ui import ui
+
+        action = "Updating"
+        if not dataset.lock:
+            return CmdDatasetAdd.display(name, new, action)
+        if dataset == new:
+            ui.write("[yellow]Nothing to update[/]", styled=True)
+            return
+
+        assert new.lock
+
+        v: Optional[tuple[str, str]] = None
+        if dataset.type == "dvcx":
+            assert new.type == "dvcx"
+            if new.lock.version < dataset.lock.version:
+                action = "Downgrading"
+
+            v = (f"v{dataset.lock.version}", f"v{new.lock.version}")
+        if dataset.type == "dvc":
+            assert new.type == "dvc"
+            v = (f"{dataset.lock.rev_lock[:9]}", f"{new.lock.rev_lock[:9]}")
+
+        if v:
+            part = ui.rich_text.assemble(
+                (v[0], "repr.number"),
+                " -> ",
+                (v[1], "repr.number"),
+            )
+        else:
+            part = ui.rich_text(dataset.spec.url, "repr.url")
+        changes = ui.rich_text.assemble("(", part, ")")
+        ui.write(action, ui.rich_text(name, "cyan"), changes, styled=True)
+        if dataset.type == "url":
+            assert new.type == "url"
+            stats = diff_files(dataset.lock.files, new.lock.files)
+            log_changes(stats)
+
+    def run(self):
+        from difflib import get_close_matches
+
+        from dvc.repo.datasets import DatasetNotFoundError
+        from dvc.ui import ui
+
+        version = None
+        if self.args.rev:
+            try:
+                version = int(self.args.rev.lstrip("v"))
+            except ValueError:
+                version = self.args.rev
+
+        d = vars(self.args) | {"version": version}
+        with self.repo.scm_context:
+            try:
+                dataset, new = self.repo.datasets.update(**d)
+            except DatasetNotFoundError:
+                logger.exception("")
+                if matches := get_close_matches(self.args.name, self.repo.datasets):
+                    ui.write(
+                        "did you mean?",
+                        ui.rich_text(matches[0], "cyan"),
+                        stderr=True,
+                        styled=True,
+                    )
+                return 1
+            self.display(self.args.name, dataset, new)
+            return 0
+
+
+def add_parser(subparsers, parent_parser):
+    ds_parser = subparsers.add_parser(
+        "dataset",
+        aliases=["ds"],
+        parents=[parent_parser],
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+    ds_subparsers = ds_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc dataset CMD --help` to display command-specific help.",
+        required=True,
+    )
+
+    dataset_add_help = "Add a dataset."
+    ds_add_parser = ds_subparsers.add_parser(
+        "add",
+        parents=[parent_parser],
+        description=append_doc_link(dataset_add_help, "dataset/add"),
+        formatter_class=formatter.RawTextHelpFormatter,
+        help=dataset_add_help,
+    )
+
+    url_exclusive_group = ds_add_parser.add_mutually_exclusive_group(required=True)
+    url_exclusive_group.add_argument(
+        "--dvcx", metavar="name", help="Name of the dvcx dataset to track"
+    )
+    url_exclusive_group.add_argument(
+        "--dvc",
+        help="Path or URL to a Git/DVC repository to track",
+        metavar="url",
+    )
+    url_exclusive_group.add_argument(
+        "--url",
+        help="""\
+URL of a cloud-versioned remote to track. Supported URLs:
+
+s3://bucket/key/path
+gs://bucket/path/to/file/or/dir
+azure://mycontainer/path
+remote://remote_name/path/to/file/or/dir (see `dvc remote`)
+""",
+    )
+    ds_add_parser.add_argument("name", help="Name of the dataset to add")
+    ds_add_parser.add_argument(
+        "--rev",
+        help="Git revision, e.g. SHA, branch, tag (only applicable with --dvc)",
+        metavar="<commit>",
+    )
+    ds_add_parser.add_argument(
+        "--path",
+        help="Path to a file or a directory within a git repository "
+        "(only applicable with --dvc)",
+    )
+    ds_add_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Overwrite existing dataset",
+    )
+    ds_add_parser.set_defaults(func=CmdDatasetAdd)
+
+    dataset_update_help = "Update a dataset."
+    ds_update_parser = ds_subparsers.add_parser(
+        "update",
+        parents=[parent_parser],
+        description=append_doc_link(dataset_update_help, "dataset/update"),
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+        help=dataset_update_help,
+    )
+    ds_update_parser.add_argument("name", help="Name of the dataset to update")
+    ds_update_parser.add_argument(
+        "--rev",
+        "--version",
+        nargs="?",
+        help="DVCX dataset version or Git revision (e.g. SHA, branch, tag)",
+        metavar="<version>",
+    )
+    ds_update_parser.set_defaults(func=CmdDatasetUpdate)

dvc/commands/studio.py

@@ -5,6 +5,7 @@ from funcy import get_in
 from dvc.cli import formatter
 from dvc.cli.utils import append_doc_link
 from dvc.commands.config import CmdConfig
+from dvc.exceptions import DvcException
 from dvc.log import logger
 
 logger = logger.getChild(__name__)
@@ -22,12 +23,21 @@ class CmdStudioLogin(CmdConfig):
         hostname = self.args.hostname or os.environ.get(DVC_STUDIO_URL) or STUDIO_URL
         scopes = self.args.scopes
 
+        studio = self.config.get("studio", {})
+        if studio.get("url", hostname) == hostname and "token" in studio:
+            raise DvcException(
+                "Token already exists. "
+                "To login with a different token, "
+                "logout using 'dvc studio logout'."
+            )
+
+        open_browser = not self.args.no_open
         try:
-            token_name, access_token = get_access_token(
+            _, access_token = get_access_token(
                 token_name=name,
                 hostname=hostname,
                 scopes=scopes,
-                use_device_code=self.args.no_open,
+                open_browser=open_browser,
                 client_name="DVC",
             )
         except StudioAuthError as e:
@@ -35,11 +45,9 @@ class CmdStudioLogin(CmdConfig):
             return 1
 
         self.save_config(hostname, access_token)
-        ui.write(
-            "Authentication has been successfully completed."
-            "The generated token will now be accessible as"
-            f" {token_name} in the user's Studio profile."
-        )
+
+        config_path = self.config.files["global"]
+        ui.write(f"Authentication complete. Saved token to {config_path}.")
         return 0
 
     def save_config(self, hostname, token):
@@ -54,12 +62,14 @@ class CmdStudioLogout(CmdConfig):
 
         with self.config.edit("global") as conf:
             if not get_in(conf, ["studio", "token"]):
-                ui.error_write("Not logged in to Studio.")
+                ui.error_write(
+                    "Not logged in to Studio. Log in with 'dvc studio login'."
+                )
                 return 1
 
             del conf["studio"]["token"]
 
-        ui.write("Logged out from Studio")
+        ui.write("Logged out from Studio (you can log back in with 'dvc studio login')")
         return 0
 
 

dvc/config_schema.py

@@ -336,6 +336,8 @@ SCHEMA = {
         "params": str,
         "plots": str,
         "live": str,
+        "auto_push": Bool,
+        "git_remote": str,
     },
     "parsing": {
         "bool": All(Lower, Choices("store_true", "boolean_optional")),

dvc/database.py

@@ -5,9 +5,9 @@ from dataclasses import dataclass
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
-from sqlalchemy import create_engine
-from sqlalchemy.engine import make_url as _make_url
-from sqlalchemy.exc import NoSuchModuleError
+from sqlalchemy import create_engine  # type: ignore[import]
+from sqlalchemy.engine import make_url as _make_url  # type: ignore[import]
+from sqlalchemy.exc import NoSuchModuleError  # type: ignore[import]
 
 from dvc import env
 from dvc.exceptions import DvcException
@@ -17,7 +17,7 @@ from dvc.utils import env2bool
 
 if TYPE_CHECKING:
     from sqlalchemy.engine import URL, Connectable, Engine
-    from sqlalchemy.sql.expression import Selectable
+    from sqlalchemy.sql.expression import Selectable  # type: ignore[import]
 
 
 logger = logger.getChild(__name__)

dvc/dependency/dataset.py

@@ -3,6 +3,7 @@ from urllib.parse import urlparse
 
 from funcy import compact, merge
 
+from dvc.exceptions import DvcException
 from dvc_data.hashfile.hash_info import HashInfo
 
 from .db import AbstractDependency
@@ -47,9 +48,15 @@ class DatasetDependency(AbstractDependency):
         )
 
     def workspace_status(self):
-        registered = self.repo.index.datasets.get(self.name, {})
+        ds = self.repo.datasets[self.name]
+        if not ds.lock:
+            return {str(self): "not in sync"}
+
         info: dict[str, Any] = self.hash_info.value if self.hash_info else {}  # type: ignore[assignment]
-        if info != registered:
+        lock = self.repo.datasets._lock_from_info(info)
+        if not lock:
+            return {str(self): "new"}
+        if lock != ds.lock:
             return {str(self): "modified"}
         return {}
 
@@ -57,7 +64,15 @@ class DatasetDependency(AbstractDependency):
         return self.workspace_status()
 
     def get_hash(self):
-        return HashInfo(self.PARAM_DATASET, self.repo.index.datasets.get(self.name, {}))
+        ds = self.repo.datasets[self.name]
+        if not ds.lock:
+            if ds._invalidated:
+                raise DvcException(
+                    "Dataset information is not in sync. "
+                    f"Run 'dvc ds update {self.name}' to sync."
+                )
+            raise DvcException("Dataset information missing from dvc.lock file")
+        return HashInfo(self.PARAM_DATASET, ds.lock.to_dict())  # type: ignore[arg-type]
 
     def save(self):
         self.hash_info = self.get_hash()

dvc/dependency/db.py

@@ -11,6 +11,7 @@ from .base import Dependency
 
 if TYPE_CHECKING:
     from dvc.output import Output
+    from dvc.repo import Repo
     from dvc.stage import Stage
 
 logger = logger.getChild(__name__)
@@ -33,7 +34,7 @@ class AbstractDependency(Dependency):
     """Dependency without workspace/fs/fs_path"""
 
     def __init__(self, stage: "Stage", info: dict[str, Any], *args, **kwargs):
-        self.repo = stage.repo
+        self.repo: "Repo" = stage.repo
         self.stage = stage
         self.fs = None
         self.fs_path = None

dvc/dvcfile.py

@@ -172,6 +172,7 @@ class SingleStageFile(FileMixin):
     from dvc.stage.loader import SingleStageLoader as LOADER  # noqa: N814
 
     datasets: ClassVar[list[dict[str, Any]]] = []
+    datasets_lock: ClassVar[list[dict[str, Any]]] = []
     metrics: ClassVar[list[str]] = []
     plots: ClassVar[Any] = {}
     params: ClassVar[list[str]] = []
@@ -240,6 +241,20 @@ class ProjectFile(FileMixin):
         if update_lock:
             self._dump_lockfile(stage, **kwargs)
 
+    def dump_dataset(self, dataset):
+        with modify_yaml(self.path, fs=self.repo.fs) as data:
+            datasets: list[dict] = data.setdefault("datasets", [])
+            loc = next(
+                (i for i, ds in enumerate(datasets) if ds["name"] == dataset["name"]),
+                None,
+            )
+            if loc is not None:
+                apply_diff(dataset, datasets[loc])
+                datasets[loc] = dataset
+            else:
+                datasets.append(dataset)
+        self.repo.scm_context.track_file(self.relpath)
+
     def _dump_lockfile(self, stage, **kwargs):
         self._lockfile.dump(stage, **kwargs)
 
@@ -308,6 +323,10 @@ class ProjectFile(FileMixin):
     def datasets(self) -> list[dict[str, Any]]:
         return self.contents.get("datasets", [])
 
+    @property
+    def datasets_lock(self) -> list[dict[str, Any]]:
+        return self.lockfile_contents.get("datasets", [])
+
     @property
     def artifacts(self) -> dict[str, Optional[dict[str, Any]]]:
         return self.contents.get("artifacts", {})
@@ -357,6 +376,24 @@ class Lockfile(FileMixin):
             self._check_gitignored()
             return {}, ""
 
+    def dump_dataset(self, dataset: dict):
+        with modify_yaml(self.path, fs=self.repo.fs) as data:
+            data.update({"schema": "2.0"})
+            if not data:
+                logger.info("Generating lock file '%s'", self.relpath)
+
+            datasets: list[dict] = data.setdefault("datasets", [])
+            loc = next(
+                (i for i, ds in enumerate(datasets) if ds["name"] == dataset["name"]),
+                None,
+            )
+            if loc is not None:
+                datasets[loc] = dataset
+            else:
+                datasets.append(dataset)
+            data.setdefault("stages", {})
+        self.repo.scm_context.track_file(self.relpath)
+
     def dump(self, stage, **kwargs):
         stage_data = serialize.to_lockfile(stage, **kwargs)
 

dvc/fs/dvc_path.py

@@ -0,0 +1,54 @@
+"""UPath implementation for DVCFileSystem.
+
+This provides a `pathlib.Path` like interface to
+work with DVCFileSystem.
+
+Examples
+--------
+
+>>> from upath import UPath
+
+>>> local = UPath("dvc://path/to/local/repo")
+>>> https = UPath("dvc+https://github.com/iterative/example-get-started", rev="main")
+>>> ssh = UPath("dvc+ssh://[email protected]:iterative/example-get-started.git")
+"""
+
+from urllib.parse import urlsplit
+
+from upath import UPath
+
+
+class DVCPath(UPath):
+    @classmethod
+    def _transform_init_args(cls, args, protocol, storage_options):
+        if not args:
+            args = ("/",)
+        elif (
+            args
+            and "url" not in storage_options
+            and protocol in {"dvc+http", "dvc+https", "dvc+ssh"}
+        ):
+            url, *rest = args
+            url = urlsplit(str(url))
+            proto = protocol.split("+")[1]
+            if proto == "ssh":
+                base_url = url.netloc + url.path
+            else:
+                base_url = url._replace(scheme=proto).geturl()
+            storage_options["url"] = base_url
+            # Assume the given path is a root url
+            args = ("/", *rest)
+        return super()._transform_init_args(args, "dvc", storage_options)
+
+    def __str__(self):
+        s = super().__str__()
+        if url := self.storage_options.get("url"):
+            return s.replace("dvc://", f"dvc+{url}", 1)
+        return s
+
+    def with_segments(self, *pathsegments):
+        obj = super().with_segments(*pathsegments)
+        # cache filesystem, as dvcfs does not cache filesystem
+        # caveat: any joinpath operation will instantiate filesystem
+        obj._fs_cached = self.fs
+        return obj

dvc/output.py

@@ -814,7 +814,7 @@ class Output:
 
         ret: dict[str, Any] = {}
         with_files = (
-            (not self.IS_DEPENDENCY or self.stage.is_import)
+            (not self.IS_DEPENDENCY or kwargs.get("datasets") or self.stage.is_import)
             and self.hash_info.isdir
             and (kwargs.get("with_files") or self.files is not None)
         )

dvc/repo/__init__.py

@@ -13,7 +13,6 @@ from dvc.exceptions import (
 )
 from dvc.ignore import DvcIgnoreFilter
 from dvc.log import logger
-from dvc.utils import as_posix
 from dvc.utils.objects import cached_property
 
 if TYPE_CHECKING:
@@ -151,6 +150,7 @@ class Repo:
         from dvc.fs import GitFileSystem, LocalFileSystem, localfs
         from dvc.lock import LockNoop, make_lock
         from dvc.repo.artifacts import Artifacts
+        from dvc.repo.datasets import Datasets
         from dvc.repo.metrics import Metrics
         from dvc.repo.params import Params
         from dvc.repo.plots import Plots
@@ -221,6 +221,7 @@ class Repo:
         self.plots: Plots = Plots(self)
         self.params: Params = Params(self)
         self.artifacts: Artifacts = Artifacts(self)
+        self.datasets: Datasets = Datasets(self)
 
         self.stage_collection_error_handler: Optional[
             Callable[[str, Exception], None]
@@ -351,16 +352,6 @@ class Repo:
         # fs.
         self._reset()
 
-    @property
-    def subrepo_relpath(self) -> str:
-        from dvc.fs import GitFileSystem
-
-        scm_root_dir = "/" if isinstance(self.fs, GitFileSystem) else self.scm.root_dir
-
-        relpath = as_posix(self.fs.relpath(self.root_dir, scm_root_dir))
-
-        return "" if relpath == "." else relpath
-
     @property
     def data_index(self) -> "DataIndex":
         from dvc_data.index import DataIndex
@@ -664,6 +655,7 @@ class Repo:
 
     def _reset(self):
         self.scm._reset()
+        self.datasets._reset()
         self.state.close()
         if "dvcfs" in self.__dict__:
             self.dvcfs.close()

dvc/repo/datasets.py

@@ -0,0 +1,423 @@
+import os
+from collections.abc import Iterator, Mapping
+from datetime import datetime
+from functools import cached_property
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional, Union, cast
+from urllib.parse import urlparse
+
+from attrs import Attribute, AttrsInstance, asdict, evolve, field, fields, frozen
+from attrs.converters import default_if_none
+
+from dvc.dvcfile import Lockfile, ProjectFile
+from dvc.exceptions import DvcException
+from dvc.log import logger
+from dvc.types import StrPath
+from dvc_data.hashfile.meta import Meta
+
+if TYPE_CHECKING:
+    from dql.dataset import DatasetRecord  # type: ignore[import]
+    from dvcx.dataset import DatasetVersion  # type: ignore[import]
+    from typing_extensions import Self
+
+    from dvc.repo import Repo
+
+
+logger = logger.getChild(__name__)
+
+
+def _get_dataset_record(name: str) -> "DatasetRecord":
+    from dvc.exceptions import DvcException
+
+    try:
+        from dvcx.catalog import get_catalog  # type: ignore[import]
+
+    except ImportError as exc:
+        raise DvcException("dvcx is not installed") from exc
+
+    catalog = get_catalog()
+    return catalog.get_remote_dataset(name)
+
+
+def _get_dataset_info(
+    name: str, record: Optional["DatasetRecord"] = None, version: Optional[int] = None
+) -> "DatasetVersion":
+    record = record or _get_dataset_record(name)
+    assert record
+    v = record.latest_version if version is None else version
+    assert v is not None
+    return record.get_version(v)
+
+
+def default_str(v) -> str:
+    return default_if_none("")(v)
+
+
+def to_datetime(d: Union[str, datetime]) -> datetime:
+    return datetime.fromisoformat(d) if isinstance(d, str) else d
+
+
+def ensure(cls):
+    def inner(v):
+        return cls.from_dict(v) if isinstance(v, dict) else v
+
+    return inner
+
+
+class SerDe:
+    def to_dict(self: AttrsInstance) -> dict[str, Any]:
+        def filter_defaults(attr: Attribute, v: Any):
+            if attr.metadata.get("exclude_falsy", False) and not v:
+                return False
+            return attr.default != v
+
+        def value_serializer(_inst, _field, v):
+            return v.isoformat() if isinstance(v, datetime) else v
+
+        return asdict(self, filter=filter_defaults, value_serializer=value_serializer)
+
+    @classmethod
+    def from_dict(cls: type["Self"], d: dict[str, Any]) -> "Self":
+        _fields = fields(cast("type[AttrsInstance]", cls))
+        kwargs = {f.name: d[f.name] for f in _fields if f.name in d}
+        return cls(**kwargs)
+
+
+@frozen(kw_only=True)
+class DatasetSpec(SerDe):
+    name: str
+    url: str
+    type: Literal["dvc", "dvcx", "url"]
+
+
+@frozen(kw_only=True)
+class DVCDatasetSpec(DatasetSpec):
+    type: Literal["dvc"]
+    path: str = field(default="", converter=default_str)
+    rev: Optional[str] = None
+
+
+@frozen(kw_only=True, order=True)
+class FileInfo(SerDe):
+    relpath: str
+    meta: Meta = field(order=False, converter=ensure(Meta))  # type: ignore[misc]
+
+
+@frozen(kw_only=True)
+class DVCDatasetLock(DVCDatasetSpec):
+    rev_lock: str
+
+
+@frozen(kw_only=True)
+class DVCXDatasetLock(DatasetSpec):
+    version: int
+    created_at: datetime = field(converter=to_datetime)
+
+
+@frozen(kw_only=True)
+class URLDatasetLock(DatasetSpec):
+    meta: Meta = field(converter=ensure(Meta))  # type: ignore[misc]
+    files: list[FileInfo] = field(
+        factory=list,
+        converter=lambda f: sorted(map(ensure(FileInfo), f)),
+        metadata={"exclude_falsy": True},
+    )
+
+
+def to_spec(lock: "Lock") -> "Spec":
+    cls = DVCDatasetSpec if lock.type == "dvc" else DatasetSpec
+    return cls(**{f.name: getattr(lock, f.name) for f in fields(cls)})
+
+
+@frozen(kw_only=True)
+class DVCDataset:
+    manifest_path: str
+    spec: DVCDatasetSpec
+    lock: Optional[DVCDatasetLock] = None
+    _invalidated: bool = field(default=False, eq=False, repr=False)
+
+    type: ClassVar[Literal["dvc"]] = "dvc"
+
+    def update(self, repo, rev: Optional[str] = None, **kwargs) -> "Self":
+        from dvc.dependency import RepoDependency
+
+        spec = self.spec
+        if rev:
+            spec = evolve(self.spec, rev=rev)
+
+        def_repo = {
+            RepoDependency.PARAM_REV: spec.rev,
+            RepoDependency.PARAM_URL: spec.url,
+        }
+        dep = RepoDependency(def_repo, None, spec.path, repo=repo)  # type: ignore[arg-type]
+        dep.save()
+        d = dep.dumpd()
+
+        repo_info = d[RepoDependency.PARAM_REPO]
+        assert isinstance(repo_info, dict)
+        rev_lock = repo_info[RepoDependency.PARAM_REV_LOCK]
+        lock = DVCDatasetLock(**spec.to_dict(), rev_lock=rev_lock)
+        return evolve(self, spec=spec, lock=lock)
+
+
+@frozen(kw_only=True)
+class DVCXDataset:
+    manifest_path: str
+    spec: "DatasetSpec"
+    lock: "Optional[DVCXDatasetLock]" = field(default=None)
+    _invalidated: bool = field(default=False, eq=False, repr=False)
+
+    type: ClassVar[Literal["dvcx"]] = "dvcx"
+
+    @property
+    def pinned(self) -> bool:
+        return self.name_version[1] is not None
+
+    @property
+    def name_version(self) -> tuple[str, Optional[int]]:
+        url = urlparse(self.spec.url)
+        path = url.netloc + url.path
+        parts = path.split("@v")
+        assert parts
+
+        name = parts[0]
+        version = int(parts[1]) if len(parts) > 1 else None
+        return name, version
+
+    def update(
+        self,
+        repo,  # noqa: ARG002
+        record: Optional["DatasetRecord"] = None,
+        version: Optional[int] = None,
+        **kwargs,
+    ) -> "Self":
+        name, _version = self.name_version
+        version = version if version is not None else _version
+        version_info = _get_dataset_info(name, record=record, version=version)
+        lock = DVCXDatasetLock(
+            **self.spec.to_dict(),
+            version=version_info.version,
+            created_at=version_info.created_at,
+        )
+        return evolve(self, lock=lock)
+
+
+@frozen(kw_only=True)
+class URLDataset:
+    manifest_path: str
+    spec: "DatasetSpec"
+    lock: "Optional[URLDatasetLock]" = None
+    _invalidated: bool = field(default=False, eq=False, repr=False)
+
+    type: ClassVar[Literal["url"]] = "url"
+
+    def update(self, repo, **kwargs):
+        from dvc.dependency import Dependency
+
+        dep = Dependency(
+            None, self.spec.url, repo=repo, fs_config={"version_aware": True}
+        )
+        dep.save()
+        d = dep.dumpd(datasets=True)
+        files = [
+            FileInfo(relpath=info["relpath"], meta=Meta.from_dict(info))
+            for info in d.get("files", [])
+        ]
+        lock = URLDatasetLock(**self.spec.to_dict(), meta=dep.meta, files=files)
+        return evolve(self, lock=lock)
+
+
+Lock = Union[DVCDatasetLock, DVCXDatasetLock, URLDatasetLock]
+Spec = Union[DatasetSpec, DVCDatasetSpec]
+Dataset = Union[DVCDataset, DVCXDataset, URLDataset]
+
+
+class DatasetNotFoundError(DvcException, KeyError):
+    def __init__(self, name, *args):
+        self.name = name
+        super().__init__("dataset not found", *args)
+
+    def __str__(self) -> str:
+        return self.msg
+
+
+class Datasets(Mapping[str, Dataset]):
+    def __init__(self, repo: "Repo") -> None:
+        self.repo: "Repo" = repo
+
+    def __repr__(self):
+        return repr(dict(self))
+
+    def __rich_repr__(self):
+        yield dict(self)
+
+    def __getitem__(self, name: str) -> Dataset:
+        try:
+            return self._datasets[name]
+        except KeyError as exc:
+            raise DatasetNotFoundError(name) from exc
+
+    def __setitem__(self, name: str, dataset: Dataset) -> None:
+        self._datasets[name] = dataset
+
+    def __contains__(self, name: object) -> bool:
+        return name in self._datasets
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._datasets)
+
+    def __len__(self) -> int:
+        return len(self._datasets)
+
+    @cached_property
+    def _spec(self) -> dict[str, tuple[str, dict[str, Any]]]:
+        return {
+            dataset["name"]: (path, dataset)
+            for path, datasets in self.repo.index._datasets.items()
+            for dataset in datasets
+        }
+
+    @cached_property
+    def _lock(self) -> dict[str, Optional[dict[str, Any]]]:
+        datasets_lock = self.repo.index._datasets_lock
+
+        def find(path, name) -> Optional[dict[str, Any]]:
+            # only look for `name` in the lock file next to the
+            # corresponding `dvc.yaml` file
+            lock = datasets_lock.get(path, [])
+            return next((dataset for dataset in lock if dataset["name"] == name), None)
+
+        return {ds["name"]: find(path, name) for name, (path, ds) in self._spec.items()}
+
+    @cached_property
+    def _datasets(self) -> dict[str, Dataset]:
+        return {
+            name: self._build_dataset(path, spec, self._lock[name])
+            for name, (path, spec) in self._spec.items()
+        }
+
+    def _reset(self) -> None:
+        self.__dict__.pop("_spec", None)
+        self.__dict__.pop("_lock", None)
+        self.__dict__.pop("_datasets", None)
+
+    @staticmethod
+    def _spec_from_info(spec: dict[str, Any]) -> Spec:
+        typ = spec.get("type")
+        if not typ:
+            raise ValueError("type should be present in spec")
+        if typ == "dvc":
+            return DVCDatasetSpec.from_dict(spec)
+        if typ in {"dvcx", "url"}:
+            return DatasetSpec.from_dict(spec)
+        raise ValueError(f"unknown dataset type: {spec.get('type', '')}")
+
+    @staticmethod
+    def _lock_from_info(lock: Optional[dict[str, Any]]) -> Optional[Lock]:
+        kl = {"dvc": DVCDatasetLock, "dvcx": DVCXDatasetLock, "url": URLDatasetLock}
+        if lock and (cls := kl.get(lock.get("type", ""))):  # type: ignore[assignment]
+            return cls.from_dict(lock)  # type: ignore[attr-defined]
+        return None
+
+    @classmethod
+    def _build_dataset(
+        cls,
+        manifest_path: str,
+        spec_data: dict[str, Any],
+        lock_data: Optional[dict[str, Any]] = None,
+    ) -> Dataset:
+        _invalidated = False
+        spec = cls._spec_from_info(spec_data)
+        lock = cls._lock_from_info(lock_data)
+        # if dvc.lock and dvc.yaml file are not in sync, we invalidate the lock.
+        if lock is not None and to_spec(lock) != spec:
+            logger.debug(
+                "invalidated lock data for %s in %s",
+                spec.name,
+                manifest_path,
+            )
+            _invalidated = True  # signal is used during `dvc repro`/`dvc status`.
+            lock = None
+
+        assert isinstance(spec, DatasetSpec)
+        if spec.type == "dvc":
+            assert lock is None or isinstance(lock, DVCDatasetLock)
+            assert isinstance(spec, DVCDatasetSpec)
+            return DVCDataset(
+                manifest_path=manifest_path,
+                spec=spec,
+                lock=lock,
+                invalidated=_invalidated,
+            )
+        if spec.type == "url":
+            assert lock is None or isinstance(lock, URLDatasetLock)
+            return URLDataset(
+                manifest_path=manifest_path,
+                spec=spec,
+                lock=lock,
+                invalidated=_invalidated,
+            )
+        if spec.type == "dvcx":
+            assert lock is None or isinstance(lock, DVCXDatasetLock)
+            return DVCXDataset(
+                manifest_path=manifest_path,
+                spec=spec,
+                lock=lock,
+                invalidated=_invalidated,
+            )
+        raise ValueError(f"unknown dataset type: {spec.type!r}")
+
+    def add(
+        self,
+        name: str,
+        url: str,
+        type: str,  # noqa: A002
+        manifest_path: StrPath = "dvc.yaml",
+        **kwargs: Any,
+    ) -> Dataset:
+        assert type in {"dvc", "dvcx", "url"}
+        kwargs.update({"name": name, "url": url, "type": type})
+        dataset = self._build_dataset(os.path.abspath(manifest_path), kwargs)
+        dataset = dataset.update(self.repo)
+
+        self.dump(dataset)
+        self[name] = dataset
+        return dataset
+
+    def update(self, name, **kwargs) -> tuple[Dataset, Dataset]:
+        dataset = self[name]
+        version = kwargs.get("version")
+
+        if dataset.type == "url" and (version or kwargs.get("rev")):
+            raise ValueError("cannot update version/revision for a url")
+        if dataset.type == "dvcx" and version is not None:
+            if not isinstance(version, int):
+                raise TypeError(
+                    f"dvcx version has to be an integer, got {type(version).__name__!r}"
+                )
+            if version < 1:
+                raise ValueError(f"dvcx version should be >=1, got {version}")
+
+        new = dataset.update(self.repo, **kwargs)
+
+        self.dump(new, old=dataset)
+        self[name] = new
+        return dataset, new
+
+    def _dump_spec(self, manifest_path: StrPath, spec: Spec) -> None:
+        spec_data = spec.to_dict()
+        assert spec_data.keys() & {"type", "name", "url"}
+        project_file = ProjectFile(self.repo, manifest_path)
+        project_file.dump_dataset(spec_data)
+
+    def _dump_lock(self, manifest_path: StrPath, lock: Lock) -> None:
+        lock_data = lock.to_dict()
+        assert lock_data.keys() & {"type", "name", "url"}
+        lockfile = Lockfile(self.repo, Path(manifest_path).with_suffix(".lock"))
+        lockfile.dump_dataset(lock_data)
+
+    def dump(self, dataset: Dataset, old: Optional[Dataset] = None) -> None:
+        if not old or old.spec != dataset.spec:
+            self._dump_spec(dataset.manifest_path, dataset.spec)
+        if dataset.lock and (not old or old.lock != dataset.lock):
+            self._dump_lock(dataset.manifest_path, dataset.lock)

dvc/repo/experiments/executor/base.py

@@ -23,7 +23,10 @@ from dvc.repo.params.show import _collect_top_level_params
 from dvc.stage.serialize import to_lockfile
 from dvc.utils import dict_sha256, env2bool, relpath
 from dvc.utils.fs import remove
-from dvc.utils.studio import env_to_config
+from dvc.utils.studio import (
+    env_to_config,
+    get_subrepo_relpath,
+)
 
 if TYPE_CHECKING:
     from queue import Queue
@@ -294,13 +297,16 @@ class BaseExecutor(ABC):
             exp_hash = cls.hash_exp(stages)
             if include_untracked:
                 dvc.scm.add(include_untracked, force=True)  # type: ignore[call-arg]
-            cls.commit(
-                dvc.scm,  # type: ignore[arg-type]
-                exp_hash,
-                exp_name=info.name,
-                force=force,
-                message=message,
-            )
+
+            with cls.auto_push(dvc):
+                cls.commit(
+                    dvc.scm,  # type: ignore[arg-type]
+                    exp_hash,
+                    exp_name=info.name,
+                    force=force,
+                    message=message,
+                )
+
             ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
             exp_ref = ExpRefInfo.from_ref(ref) if ref else None
             untracked = dvc.scm.untracked_files()
@@ -457,9 +463,6 @@ class BaseExecutor(ABC):
         from dvc.repo.checkout import checkout as dvc_checkout
         from dvc.ui import ui
 
-        auto_push = env2bool(DVC_EXP_AUTO_PUSH)
-        git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
@@ -480,9 +483,6 @@ class BaseExecutor(ABC):
             message=message,
             **kwargs,
         ) as dvc:
-            if auto_push:
-                cls._validate_remotes(dvc, git_remote)
-
             args, kwargs = cls._repro_args(dvc)
             if args:
                 targets: Optional[Union[list, str]] = args[0]
@@ -516,8 +516,6 @@ class BaseExecutor(ABC):
                     dvc,
                     info,
                     exp_hash,
-                    auto_push,
-                    git_remote,
                     repro_force,
                     message=message,
                 )
@@ -547,20 +545,18 @@ class BaseExecutor(ABC):
         dvc,
         info,
         exp_hash,
-        auto_push,
-        git_remote,
         repro_force,
         message: Optional[str] = None,
     ) -> tuple[Optional[str], Optional["ExpRefInfo"], bool]:
-        cls.commit(
-            dvc.scm,
-            exp_hash,
-            exp_name=info.name,
-            force=repro_force,
-            message=message,
-        )
-        if auto_push:
-            cls._auto_push(dvc, dvc.scm, git_remote)
+        with cls.auto_push(dvc):
+            cls.commit(
+                dvc.scm,
+                exp_hash,
+                exp_name=info.name,
+                force=repro_force,
+                message=message,
+            )
+
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         exp_ref: Optional["ExpRefInfo"] = ExpRefInfo.from_ref(ref) if ref else None
         if cls.WARN_UNTRACKED:
@@ -624,6 +620,7 @@ class BaseExecutor(ABC):
                     params=to_studio_params(dvc.params.show()),
                     dvc_studio_config=dvc_studio_config,
                     message=message,
+                    subdir=get_subrepo_relpath(dvc),
                 )
                 logger.debug("Running repro in '%s'", os.getcwd())
                 yield dvc
@@ -668,15 +665,30 @@ class BaseExecutor(ABC):
             kwargs = {}
         return args, kwargs
 
+    @classmethod
+    @contextmanager
+    def auto_push(cls, dvc: "Repo") -> Iterator[None]:
+        exp_config = dvc.config.get("exp", {})
+        auto_push = env2bool(DVC_EXP_AUTO_PUSH, exp_config.get("auto_push", False))
+        if not auto_push:
+            yield
+            return
+
+        git_remote = os.getenv(
+            DVC_EXP_GIT_REMOTE, exp_config.get("git_remote", "origin")
+        )
+        cls._validate_remotes(dvc, git_remote)
+        yield
+        cls._auto_push(dvc, git_remote)
+
     @staticmethod
     def _auto_push(
         dvc: "Repo",
-        scm: "Git",
         git_remote: Optional[str],
         push_cache=True,
         run_cache=True,
     ):
-        branch = scm.get_ref(EXEC_BRANCH, follow=False)
+        branch = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         try:
             dvc.experiments.push(
                 git_remote,
@@ -704,6 +716,7 @@ class BaseExecutor(ABC):
         message: Optional[str] = None,
     ):
         """Commit stages as an experiment and return the commit SHA."""
+
         rev = scm.get_rev()
         if not scm.is_dirty(untracked_files=False):
             logger.debug("No changes to commit")

dvc/repo/imports.py

@@ -1,145 +0,0 @@
-import os
-from functools import partial
-from tempfile import TemporaryDirectory
-from typing import TYPE_CHECKING, Union
-
-from dvc.log import logger
-
-if TYPE_CHECKING:
-    from dvc.dependency.base import Dependency
-    from dvc.repo import Repo
-    from dvc.repo.index import Index, IndexView
-    from dvc.stage import Stage
-    from dvc.types import TargetType
-    from dvc_data.hashfile.hash_info import HashInfo
-
-
-logger = logger.getChild(__name__)
-
-
-def unfetched_view(
-    index: "Index", targets: "TargetType", unpartial: bool = False, **kwargs
-) -> tuple["IndexView", "IndexView", list["Dependency"]]:
-    """Return index view of imports which have not been fetched.
-
-    Returns:
-        Tuple in the form (legacy_view, view, changed_deps) where changed_imports is a
-        list of import dependencies that cannot be fetched due to changed data source.
-    """
-    from dvc.cachemgr import LEGACY_HASH_NAMES
-
-    changed_deps: list["Dependency"] = []
-
-    def need_fetch(stage: "Stage", legacy: bool = False) -> bool:
-        if not stage.is_import or (stage.is_partial_import and not unpartial):
-            return False
-
-        out = stage.outs[0]
-        if not out.changed_cache():
-            return False
-
-        dep = stage.deps[0]
-        if dep.changed_checksum():
-            changed_deps.append(dep)
-            return False
-
-        if out.hash_name in LEGACY_HASH_NAMES and legacy:
-            return True
-        if out.hash_name not in LEGACY_HASH_NAMES and not legacy:
-            return True
-        return False
-
-    legacy_unfetched = index.targets_view(
-        targets,
-        stage_filter=partial(need_fetch, legacy=True),
-        **kwargs,
-    )
-    unfetched = index.targets_view(targets, stage_filter=need_fetch, **kwargs)
-    return legacy_unfetched, unfetched, changed_deps
-
-
-def partial_view(index: "Index", targets: "TargetType", **kwargs) -> "IndexView":
-    return index.targets_view(
-        targets,
-        stage_filter=lambda s: s.is_partial_import,
-        **kwargs,
-    )
-
-
-def unpartial_imports(index: Union["Index", "IndexView"]) -> int:
-    """Update any outs in the index which are no longer partial imports.
-
-    Returns:
-        Total number of files which were unpartialed.
-    """
-    from dvc_data.hashfile.hash_info import HashInfo
-    from dvc_data.hashfile.meta import Meta
-
-    updated = 0
-    for out in index.outs:
-        # we need to use view[key] here and since the out fields have not been
-        # updated yet (out.get_entry() would return the partial-import state)
-        workspace, key = out.index_key
-        entry = index.data[workspace][key]
-        if out.stage.is_partial_import:
-            out.hash_info = entry.hash_info or HashInfo()
-            out.meta = entry.meta or Meta()
-            out.stage.md5 = out.stage.compute_md5()
-            out.stage.dump()
-            updated += out.meta.nfiles if out.meta.nfiles is not None else 1
-    return updated
-
-
-def save_imports(
-    repo: "Repo", targets: "TargetType", unpartial: bool = False, **kwargs
-) -> set["HashInfo"]:
-    """Save (download) imports from their original source location.
-
-    Imports which are already cached will not be downloaded.
-
-    Returns:
-        Objects which were downloaded from source location.
-    """
-    from dvc.fs.callbacks import TqdmCallback
-    from dvc.stage.exceptions import DataSourceChanged
-    from dvc_data.index import md5, save
-    from dvc_data.index.checkout import apply, compare
-
-    downloaded: set["HashInfo"] = set()
-
-    legacy_unfetched, unfetched, changed = unfetched_view(
-        repo.index, targets, unpartial=unpartial, **kwargs
-    )
-    for dep in changed:
-        logger.warning(str(DataSourceChanged(f"{dep.stage} ({dep})")))
-
-    for view, cache in [
-        (legacy_unfetched, repo.cache.legacy),
-        (unfetched, repo.cache.local),
-    ]:
-        data_view = view.data["repo"]
-        if len(data_view):
-            if not cache.fs.exists(cache.path):
-                os.makedirs(cache.path)
-            with TemporaryDirectory(dir=cache.path) as tmpdir:
-                with TqdmCallback(
-                    desc="Downloading imports from source",
-                    unit="files",
-                ) as cb:
-                    diff = compare(None, data_view)
-                    apply(diff, tmpdir, cache.fs, callback=cb, storage="remote")
-                md5(data_view, name=cache.hash_name)
-                save(data_view, odb=cache, hardlink=True)
-
-            downloaded.update(
-                entry.hash_info
-                for _, entry in data_view.iteritems()
-                if entry.meta is not None
-                and not entry.meta.isdir
-                and entry.hash_info is not None
-            )
-
-    if unpartial:
-        unpartial_imports(partial_view(repo.index, targets, **kwargs))
-
-    return downloaded

dvc/repo/index.py

@@ -104,7 +104,7 @@ def collect_files(
         dirs[:] = [d for d in dirs if not is_out_or_ignored(root, d)]
 
 
-def _load_data_from_tree(index, prefix, ws, key, tree):
+def _load_data_from_tree(index, prefix, ws, key, tree, hash_name):
     from dvc_data.index import DataIndexEntry, Meta
 
     parents = set()
@@ -117,7 +117,7 @@ def _load_data_from_tree(index, prefix, ws, key, tree):
         index[(*prefix, ws, *fkey)] = DataIndexEntry(
             key=fkey,
             meta=ometa,
-            hash_info=ohi,
+            hash_info=ohi if (ohi and ohi.name == hash_name) else None,
         )
 
     for parent in parents:
@@ -151,7 +151,7 @@ def _load_data_from_outs(index, prefix, outs):
             tree = out.get_obj()
 
         if tree is not None:
-            _load_data_from_tree(index, prefix, ws, key, tree)
+            _load_data_from_tree(index, prefix, ws, key, tree, out.hash_name)
 
         entry = DataIndexEntry(
             key=key,
@@ -193,16 +193,22 @@ def _load_storage_from_import(storage_map, key, out):
         return
 
     dep = out.stage.deps[0]
-    if not out.hash_info and (
-        not dep.hash_info or dep.hash_info.name != storage_map[key].cache.odb.hash_name
-    ):
-        # partial import
+    if not out.hash_info or dep.fs.version_aware:
+        if dep.meta and dep.meta.isdir:
+            meta_token = dep.hash_info.value
+        else:
+            meta_token = tokenize(dep.meta.to_dict())
+
         fs_cache = out.repo.cache.fs_cache
         storage_map.add_cache(
             FileStorage(
                 key,
                 fs_cache.fs,
-                fs_cache.fs.join(fs_cache.path, dep.fs.protocol, tokenize(dep.fs_path)),
+                fs_cache.fs.join(
+                    fs_cache.path,
+                    dep.fs.protocol,
+                    tokenize(dep.fs_path, meta_token),
+                ),
             )
         )
 
@@ -284,6 +290,7 @@ class Index:
         params: Optional[dict[str, Any]] = None,
         artifacts: Optional[dict[str, Any]] = None,
         datasets: Optional[dict[str, list[dict[str, Any]]]] = None,
+        datasets_lock: Optional[dict[str, list[dict[str, Any]]]] = None,
     ) -> None:
         self.repo = repo
         self.stages = stages or []
@@ -292,6 +299,7 @@ class Index:
         self._params = params or {}
         self._artifacts = artifacts or {}
         self._datasets: dict[str, list[dict[str, Any]]] = datasets or {}
+        self._datasets_lock: dict[str, list[dict[str, Any]]] = datasets_lock or {}
         self._collected_targets: dict[int, list["StageInfo"]] = {}
 
     @cached_property
@@ -316,6 +324,7 @@ class Index:
         params = {}
         artifacts = {}
         datasets = {}
+        datasets_lock = {}
 
         onerror = onerror or repo.stage_collection_error_handler
         for _, idx in collect_files(repo, onerror=onerror):
@@ -325,6 +334,7 @@ class Index:
             params.update(idx._params)
             artifacts.update(idx._artifacts)
             datasets.update(idx._datasets)
+            datasets_lock.update(idx._datasets_lock)
         return cls(
             repo,
             stages=stages,
@@ -333,6 +343,7 @@ class Index:
             params=params,
             artifacts=artifacts,
             datasets=datasets,
+            datasets_lock=datasets_lock,
         )
 
     @classmethod
@@ -348,6 +359,9 @@ class Index:
             params={path: dvcfile.params} if dvcfile.params else {},
             artifacts={path: dvcfile.artifacts} if dvcfile.artifacts else {},
             datasets={path: dvcfile.datasets} if dvcfile.datasets else {},
+            datasets_lock={path: dvcfile.datasets_lock}
+            if dvcfile.datasets_lock
+            else {},
         )
 
     def update(self, stages: Iterable["Stage"]) -> "Index":
@@ -400,11 +414,6 @@ class Index:
         for stage in self.stages:
             yield from stage.outs
 
-    @cached_property
-    def datasets(self) -> dict[str, dict[str, Any]]:
-        datasets = chain.from_iterable(self._datasets.values())
-        return {dataset["name"]: dataset for dataset in datasets}
-
     @cached_property
     def out_data_keys(self) -> dict[str, set["DataIndexKey"]]:
         by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)

dvc/repo/plots/__init__.py

@@ -423,11 +423,17 @@ def _id_is_path(plot_props=None):
 
 def _adjust_sources(fs, plot_props, config_dir):
     new_plot_props = deepcopy(plot_props)
-    old_y = new_plot_props.pop("y", {})
-    new_y = {}
-    for filepath, val in old_y.items():
-        new_y[_normpath(fs.join(config_dir, filepath))] = val
-    new_plot_props["y"] = new_y
+    for axis in ["x", "y"]:
+        x_is_inferred = axis == "x" and (
+            axis not in new_plot_props or isinstance(new_plot_props[axis], str)
+        )
+        if x_is_inferred:
+            continue
+        old = new_plot_props.pop(axis, {})
+        new = {}
+        for filepath, val in old.items():
+            new[_normpath(fs.join(config_dir, filepath))] = val
+        new_plot_props[axis] = new
     return new_plot_props
 
 
@@ -442,6 +448,10 @@ def _resolve_definitions(
     config_path = os.fspath(config_path)
     config_dir = fs.dirname(config_path)
     result: dict[str, dict] = {}
+
+    plot_ids_parents = [
+        _normpath(fs.join(config_dir, plot_id)) for plot_id in definitions
+    ]
     for plot_id, plot_props in definitions.items():
         if plot_props is None:
             plot_props = {}
@@ -451,6 +461,12 @@ def _resolve_definitions(
                 unpacked = unpack_if_dir(
                     fs, data_path, props=plot_props | props, onerror=onerror
                 )
+                # use config for parent directory with most specific definition
+                unpacked["data"] = {
+                    k: v
+                    for k, v in unpacked["data"].items()
+                    if _closest_parent(fs, k, plot_ids_parents) == data_path
+                }
                 dpath.merge(result, unpacked)
         elif _matches(targets, config_path, plot_id):
             adjusted_props = _adjust_sources(fs, plot_props, config_dir)
@@ -459,6 +475,15 @@ def _resolve_definitions(
     return result
 
 
+def _closest_parent(fs, path, parents):
+    best_result = ""
+    for parent in parents:
+        common_path = fs.commonpath([path, parent])
+        if len(common_path) > len(best_result):
+            best_result = common_path
+    return best_result
+
+
 def _collect_pipeline_files(repo, targets: list[str], props, onerror=None):
     result: dict[str, dict] = {}
     top_plots = repo.index._plots
@@ -513,7 +538,7 @@ def unpack_if_dir(fs, path, props: dict[str, str], onerror: Optional[Callable] =
 
     if "data" in unpacked:
         for subpath in unpacked["data"]:
-            result["data"].update({subpath: props})
+            result["data"].update({subpath: props.copy()})
     else:
         result.update(unpacked)
 

dvc/schema.py

@@ -47,6 +47,7 @@ LOCK_FILE_STAGE_SCHEMA = {
 LOCKFILE_STAGES_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
 LOCKFILE_SCHEMA = {
     vol.Required("schema"): vol.Equal("2.0", "invalid schema version"),
+    "datasets": object,
     STAGES: LOCKFILE_STAGES_SCHEMA,
 }
 
@@ -128,7 +129,7 @@ DATASET_SCHEMA = vol.Schema(
     {vol.Required("type"): str, vol.Required("name"): str}, extra=vol.ALLOW_EXTRA
 )
 MULTI_STAGE_SCHEMA = {
-    "datasets": [DATASET_SCHEMA],
+    "datasets": object,
     PLOTS: [vol.Any(str, SINGLE_PLOT_SCHEMA)],
     STAGES: SINGLE_PIPELINE_STAGE_SCHEMA,
     VARS_KWD: VARS_SCHEMA,

dvc/testing/api_tests.py

@@ -33,9 +33,9 @@ class TestAPI:
         [
             {},
             {"url": "{path}"},
-            {"url": "{path}", "rev": "master"},
+            {"url": "{path}", "rev": "{default_branch}"},
             {"url": "file://{posixpath}"},
-            {"url": "file://{posixpath}", "rev": "master"},
+            {"url": "file://{posixpath}", "rev": "{default_branch}"},
         ],
         ids=["current", "local", "local_rev", "git", "git_rev"],
     )
@@ -61,6 +61,8 @@ class TestAPI:
 
         if url := fs_kwargs.get("url"):
             fs_kwargs["url"] = url.format(path=tmp_dir, posixpath=tmp_dir.as_posix())
+        if rev := fs_kwargs.get("rev"):
+            fs_kwargs["rev"] = rev.format(default_branch=scm.active_branch())
 
         fs = DVCFileSystem(**fs_kwargs)
 

dvc/testing/benchmarks/cli/commands/test_get.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_get(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/benchmarks/cli/commands/test_import.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_import(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/fixtures.py

@@ -221,15 +221,8 @@ def docker_compose_project_name():
 
 
 @pytest.fixture(scope="session")
-def docker_services(
-    tmp_path_factory,
-    docker_compose_command,
-    docker_compose_file,
-    docker_compose_project_name,
-    docker_setup,
-):
+def docker_services(tmp_path_factory, request):
     from filelock import FileLock
-    from pytest_docker.plugin import DockerComposeExecutor, Services
 
     if os.environ.get("CI") and os.name == "nt":
         pytest.skip("disabled for Windows on CI")
@@ -251,18 +244,12 @@ def docker_services(
         out = (err.output or b"").decode("utf-8")
         pytest.skip(f"docker-compose is not installed: {out}")
 
-    executor = DockerComposeExecutor(
-        docker_compose_command, docker_compose_file, docker_compose_project_name
-    )
-
     # making sure we don't accidentally launch docker-compose in parallel,
     # as it might result in network conflicts. Inspired by:
     # https://github.com/pytest-dev/pytest-xdist#making-session-scoped-fixtures-execute-only-once
     lockfile = tmp_path_factory.getbasetemp().parent / "docker-compose.lock"
     with FileLock(os.fspath(lockfile)):
-        executor.execute(docker_setup)
-        # note: we are not tearing down the containers here
-        return Services(executor)
+        return request.getfixturevalue("docker_services")
 
 
 @pytest.fixture

dvc/testing/path_info.py

@@ -1,6 +1,7 @@
 import os
 import pathlib
 import posixpath
+import sys
 from typing import Callable, ClassVar
 from urllib.parse import urlparse
 
@@ -27,12 +28,13 @@ class PathInfo(pathlib.PurePath, _BasePath):
     __slots__ = ()
     scheme = "local"
 
-    def __new__(cls, *args):
-        # Construct a proper subclass depending on current os
-        if cls is PathInfo:
-            cls = WindowsPathInfo if os.name == "nt" else PosixPathInfo
+    if sys.version_info < (3, 12):
 
-        return cls._from_parts(args)  # type: ignore[attr-defined]
+        def __new__(cls, *args):
+            if cls is PathInfo:
+                cls = WindowsPathInfo if os.name == "nt" else PosixPathInfo
+
+            return cls._from_parts(args)  # type: ignore[attr-defined]
 
     def as_posix(self):
         f = self._flavour  # type: ignore[attr-defined]
@@ -67,7 +69,7 @@ class PathInfo(pathlib.PurePath, _BasePath):
 
     def isin(self, other):
         if isinstance(other, (str, bytes)):
-            other = self.__class__(other)
+            other = self.__class__(other)  # type: ignore[arg-type]
         elif self.__class__ != other.__class__:
             return False
         # Use cached casefolded parts to compare paths

dvc/testing/tmp_dir.py

@@ -67,13 +67,12 @@ class TmpDir(pathlib.Path):
     def config(self):
         return {"url": self.url}
 
-    def __new__(cls, *args, **kwargs):
-        if cls is TmpDir:
-            cls = WindowsTmpDir if os.name == "nt" else PosixTmpDir
+    if sys.version_info < (3, 12):
+
+        def __new__(cls, *args, **kwargs):
+            if cls is TmpDir:
+                cls = WindowsTmpDir if os.name == "nt" else PosixTmpDir
 
-        if sys.version_info >= (3, 12):
-            return object.__new__(cls)
-        else:  # noqa: RET505
             # init parameter and `_init` method has been removed in Python 3.10.
             kw = {"init": False} if sys.version_info < (3, 10) else {}
             self = cls._from_parts(args, **kw)  # type: ignore[attr-defined]

dvc/testing/workspace_tests.py

@@ -106,6 +106,7 @@ class TestImportURLVersionAware:
         (remote_version_aware / "file").write_text("modified")
         assert dvc.status().get("file.dvc") == [
             {"changed deps": {"remote://upstream/file": "update available"}},
+            {"changed outs": {"file": "not in cache"}},
         ]
         dvc.update(str(tmp_dir / "file.dvc"))
         assert (tmp_dir / "file").read_text() == "modified"
@@ -137,6 +138,7 @@ class TestImportURLVersionAware:
         (remote_version_aware / "data_dir" / "new_file").write_text("new")
         assert dvc.status().get("data_dir.dvc") == [
             {"changed deps": {"remote://upstream/data_dir": "modified"}},
+            {"changed outs": {"data_dir": "not in cache"}},
         ]
         dvc.update(str(tmp_dir / "data_dir.dvc"))
         assert (tmp_dir / "data_dir" / "subdir" / "file").read_text() == "modified"

dvc/utils/studio.py

@@ -12,10 +12,14 @@ from dvc.env import (
     DVC_STUDIO_URL,
 )
 from dvc.log import logger
+from dvc.utils import as_posix
 
 if TYPE_CHECKING:
     from requests import Response
 
+    from dvc.repo import Repo
+
+
 logger = logger.getChild(__name__)
 
 STUDIO_URL = "https://studio.iterative.ai"
@@ -111,3 +115,13 @@ def env_to_config(env: dict[str, Any]) -> dict[str, Any]:
     if DVC_STUDIO_URL in env:
         config["url"] = env[DVC_STUDIO_URL]
     return config
+
+
+def get_subrepo_relpath(repo: "Repo") -> str:
+    from dvc.fs import GitFileSystem
+
+    scm_root_dir = "/" if isinstance(repo.fs, GitFileSystem) else repo.scm.root_dir
+
+    relpath = as_posix(repo.fs.relpath(repo.root_dir, scm_root_dir))
+
+    return "" if relpath == "." else relpath

pyproject.toml

@@ -35,10 +35,10 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=3.10,<3.12",
+    "dvc-data>=3.13,<3.15",
     "dvc-http>=2.29.0",
     "dvc-render>=1.0.1,<2",
-    "dvc-studio-client>=0.17.1,<1",
+    "dvc-studio-client>=0.20,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
     # https://github.com/iterative/dvc/issues/9654
@@ -102,8 +102,8 @@ tests = [
     "beautifulsoup4>=4.4",
     "dvc-ssh",
     "filelock",
-    "flaky",
     "pytest>=7,<9",
+    "pytest-rerunfailures",
     "pytest-cov>=4.1.0",
     "pytest-docker>=1,<4",
     "pytest-mock",
@@ -130,6 +130,14 @@ dvc = "dvc.cli:main"
 [project.entry-points."fsspec.specs"]
 dvc = "dvc.api:DVCFileSystem"
 
+[project.entry-points."universal_pathlib.implementations"]
+dvc = "dvc.fs.dvc_path:DVCPath"
+# universal_pathlib does not support fsspec url chaining yet.
+# see https://github.com/fsspec/universal_pathlib/issues/28.
+"dvc+http" = "dvc.fs.dvc_path:DVCPath"
+"dvc+https" = "dvc.fs.dvc_path:DVCPath"
+"dvc+ssh" = "dvc.fs.dvc_path:DVCPath"
+
 [project.entry-points."pyinstaller40"]
 hook-dirs = "dvc.__pyinstaller:get_hook_dirs"
 tests = "dvc.__pyinstaller:get_PyInstaller_tests"
@@ -232,6 +240,7 @@ module = [
     "ruamel.yaml.*",
     "shortuuid",
     "shtab",
+    "upath",
     "virtualenv",
     "viztracer",
     "voluptuous",

tests/conftest.py

@@ -123,7 +123,7 @@ def pytest_runtest_setup(item):
     if "CI" in os.environ and item.get_closest_marker("needs_internet") is not None:
         # remotes that need internet connection might be flaky,
         # so we rerun them in case it fails.
-        item.add_marker(pytest.mark.flaky(max_runs=5, min_passes=1))
+        item.add_marker(pytest.mark.flaky(reruns=5))
 
 
 @pytest.fixture(scope="session")

tests/func/experiments/test_remote.py

@@ -372,3 +372,40 @@ def test_push_pull_invalid_workspace(
         dvc.experiments.push(git_upstream.remote, push_cache=True)
         dvc.experiments.pull(git_upstream.remote, pull_cache=True)
         assert "failed to collect" not in caplog.text
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_run(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_save(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    dvc.experiments.save(name=exp_name, force=True)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key

tests/func/plots/test_collect.py

@@ -0,0 +1,63 @@
+import dpath
+
+from dvc.repo.plots import Plots
+
+
+def test_subdir_config_not_overwritten_by_parents(tmp_dir, scm, dvc):
+    plot_data = [
+        {"x": 1, "y": 0.1},
+        {"x": 2, "y": 0.2},
+        {"x": 3, "y": 0.3},
+    ]
+    subdir_plot_data = [
+        {"x": 1, "y": 0.2, "z": 0.1},
+        {"x": 2, "y": 0.3, "z": 0.2},
+        {"x": 3, "y": 0.4, "z": 0.3},
+    ]
+
+    (tmp_dir / "plots").mkdir()
+    (tmp_dir / "plots" / "subdir").mkdir()
+
+    (tmp_dir / "plots" / "plot.json").dump_json(plot_data)
+    (tmp_dir / "plots" / "subdir" / "plot.json").dump_json(subdir_plot_data)
+
+    plots_config = [
+        {
+            "plots/subdir/": {
+                "x": "z",
+                "y": "x",
+            }
+        },
+        {"plots": {"x": "x", "y": "y"}},
+        {
+            "subdir axis defined by filename": {
+                "x": {"plots/subdir/plot.json": "x"},
+                "y": {"plots/subdir/plot.json": "y"},
+            }
+        },
+    ]
+
+    from dvc.utils.serialize import modify_yaml
+
+    with modify_yaml("dvc.yaml") as dvcfile_content:
+        dvcfile_content["plots"] = plots_config
+
+    scm.add(
+        [
+            "plots/plot.json",
+            "plots/subdir/plot.json",
+            "dvc.yaml",
+        ]
+    )
+    scm.commit("add data sources")
+
+    plots = next(Plots(dvc).collect())
+
+    assert dpath.get(plots, "workspace/definitions/data/dvc.yaml/data") == {
+        "plots/plot.json": {"x": "x", "y": "y"},
+        "plots/subdir/plot.json": {"x": "z", "y": "x"},
+        "subdir axis defined by filename": {
+            "x": {"plots/subdir/plot.json": "x"},
+            "y": {"plots/subdir/plot.json": "y"},
+        },
+    }

tests/func/plots/test_show.py

@@ -172,6 +172,87 @@ def test_plots_show_overlap(tmp_dir, dvc, run_copy_metrics, clear_before_run):
     )
 
 
+def test_plots_show_nested_x_dict(tmp_dir, dvc, scm):
+    rel_pipeline_dir = "pipelines/data-increment"
+
+    pipeline_rel_dvclive_metrics_dir = "dvclive/plots/metrics"
+    dvc_rel_dvclive_metrics_dir = (
+        f"{rel_pipeline_dir}/{pipeline_rel_dvclive_metrics_dir}"
+    )
+
+    pipeline_dir = tmp_dir / rel_pipeline_dir
+    dvclive_metrics_dir = pipeline_dir / pipeline_rel_dvclive_metrics_dir
+    dvclive_metrics_dir.mkdir(parents=True)
+
+    def _get_plot_defn(rel_dir: str) -> dict:
+        return {
+            "template": "simple",
+            "x": {f"{rel_dir}/Max_Leaf_Nodes.tsv": "Max_Leaf_Nodes"},
+            "y": {f"{rel_dir}/Error.tsv": "Error"},
+        }
+
+    (pipeline_dir / "dvc.yaml").dump(
+        {
+            "plots": [
+                {
+                    "Error vs max_leaf_nodes": _get_plot_defn(
+                        pipeline_rel_dvclive_metrics_dir
+                    )
+                },
+            ]
+        },
+    )
+
+    dvclive_metrics_dir.gen(
+        {
+            "Error.tsv": "step\tError\n" "0\t0.11\n" "1\t0.22\n" "2\t0.44\n",
+            "Max_Leaf_Nodes.tsv": "step\tMax_Leaf_Nodes\n"
+            "0\t5\n"
+            "1\t50\n"
+            "2\t500\n",
+        }
+    )
+
+    scm.commit("add dvc.yaml and dvclive metrics")
+
+    result = dvc.plots.show()
+    assert result == {
+        "workspace": {
+            "definitions": {
+                "data": {
+                    f"{rel_pipeline_dir}/dvc.yaml": {
+                        "data": {
+                            "Error vs max_leaf_nodes": _get_plot_defn(
+                                dvc_rel_dvclive_metrics_dir
+                            )
+                        },
+                    }
+                }
+            },
+            "sources": {
+                "data": {
+                    f"{dvc_rel_dvclive_metrics_dir}/Error.tsv": {
+                        "data": [
+                            {"Error": "0.11", "step": "0"},
+                            {"Error": "0.22", "step": "1"},
+                            {"Error": "0.44", "step": "2"},
+                        ],
+                        "props": {},
+                    },
+                    f"{dvc_rel_dvclive_metrics_dir}/Max_Leaf_Nodes.tsv": {
+                        "data": [
+                            {"Max_Leaf_Nodes": "5", "step": "0"},
+                            {"Max_Leaf_Nodes": "50", "step": "1"},
+                            {"Max_Leaf_Nodes": "500", "step": "2"},
+                        ],
+                        "props": {},
+                    },
+                }
+            },
+        }
+    }
+
+
 def test_dir_plots(tmp_dir, dvc, run_copy_metrics):
     subdir = tmp_dir / "subdir"
     subdir.mkdir()

tests/func/test_data_cloud.py

@@ -3,7 +3,6 @@ import os
 import shutil
 
 import pytest
-from flaky.flaky_decorator import flaky
 
 import dvc_data
 from dvc.cli import main
@@ -207,7 +206,7 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     assert hash_spy.call_count == 10
 
 
-@flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 @pytest.mark.parametrize("erepo_type", ["git_dir", "erepo_dir"])
 def test_pull_git_imports(request, tmp_dir, dvc, scm, erepo_type):
     erepo = request.getfixturevalue(erepo_type)

tests/func/test_dataset.py

@@ -0,0 +1,414 @@
+import os
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+import pytest
+from attrs import define, evolve, field, has
+
+from dvc.dependency.base import Dependency
+from dvc.exceptions import ReproductionError
+from dvc.repo.datasets import (
+    DatasetNotFoundError,
+    DatasetSpec,
+    DVCDataset,
+    DVCDatasetLock,
+    DVCDatasetSpec,
+    DVCXDataset,
+    DVCXDatasetLock,
+    FileInfo,
+    URLDataset,
+    URLDatasetLock,
+)
+from dvc_data.hashfile.meta import Meta
+from dvc_data.index import HashInfo, Tree
+
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+
+
+@define
+class MockedDVCXVersionInfo:
+    version: int
+    created_at: datetime = field(factory=lambda: datetime.now(timezone.utc))
+
+
+def evolve_recursive(inst, **changes):
+    """Recursive attr.evolve() method, where any attr-based attributes
+    will be evolved too.
+    """
+    for key, value in changes.items():
+        v = getattr(inst, key)
+        if has(type(v)) and isinstance(value, dict):
+            value = evolve_recursive(v, **value)
+        changes[key] = value
+    return evolve(inst, **changes)
+
+
+def test_dvc(tmp_dir, scm, dvc: "Repo"):
+    datasets = dvc.datasets
+
+    tmp_dir.scm_gen("file", "file", commit="add file")
+    dataset = datasets.add("mydataset", tmp_dir.fs_path, "dvc", path="file")
+    expected = DVCDataset(
+        manifest_path=(tmp_dir / "dvc.yaml").fs_path,
+        spec=DVCDatasetSpec(
+            name="mydataset", url=tmp_dir.fs_path, type="dvc", path="file"
+        ),
+        lock=DVCDatasetLock(
+            name="mydataset",
+            url=tmp_dir.fs_path,
+            type="dvc",
+            path="file",
+            rev_lock=scm.get_rev(),
+        ),
+    )
+    assert "mydataset" in datasets
+    assert dataset == datasets["mydataset"] == expected
+    tmp_dir.scm_gen("file", "file", commit="update file")
+
+    old, new = datasets.update("mydataset")
+    assert old == dataset
+    assert old != new
+    expected = evolve_recursive(expected, lock={"rev_lock": scm.get_rev()})
+    assert new == datasets["mydataset"] == expected
+
+    # noop
+    old, new = datasets.update("mydataset")
+    assert old == new
+
+
+def test_dvcx(tmp_dir, dvc, mocker):
+    datasets = dvc.datasets
+
+    version_info = [MockedDVCXVersionInfo(1), MockedDVCXVersionInfo(2)]
+    version_info.append(version_info[1])
+    mocker.patch("dvc.repo.datasets._get_dataset_info", side_effect=version_info)
+
+    dataset = datasets.add("mydataset", "dataset", "dvcx")
+    expected = DVCXDataset(
+        manifest_path=(tmp_dir / "dvc.yaml").fs_path,
+        spec=DatasetSpec(name="mydataset", url="dataset", type="dvcx"),
+        lock=DVCXDatasetLock(
+            name="mydataset",
+            url="dataset",
+            type="dvcx",
+            version=1,
+            created_at=version_info[0].created_at,
+        ),
+    )
+    assert "mydataset" in datasets
+    assert dataset == datasets["mydataset"] == expected
+
+    old, new = datasets.update("mydataset")
+    assert old == dataset
+    assert old != new
+    expected = evolve_recursive(
+        expected, lock={"version": 2, "created_at": version_info[1].created_at}
+    )
+    assert new == datasets["mydataset"] == expected
+
+    # noop
+    old, new = datasets.update("mydataset")
+    assert old == new
+
+
+def test_url(tmp_dir, dvc, mocker):
+    datasets = dvc.datasets
+
+    tree = Tree()
+    tree.add(("foo",), Meta(version_id="1"), None)
+    tree.digest(with_meta=True)
+    tree_meta = Meta(isdir=True)
+
+    def mocked_save(d):
+        d.meta, d.obj, d.hash_info = tree_meta, tree, HashInfo("md5", "value.dir")
+
+    mocker.patch.object(Dependency, "save", mocked_save)
+
+    dataset = datasets.add("mydataset", "s3://dataset", "url")
+    expected = URLDataset(
+        manifest_path=(tmp_dir / "dvc.yaml").fs_path,
+        spec=DatasetSpec(name="mydataset", url="s3://dataset", type="url"),
+        lock=URLDatasetLock(
+            name="mydataset",
+            url="s3://dataset",
+            type="url",
+            meta=Meta(isdir=True),
+            files=[FileInfo(relpath="foo", meta=Meta(version_id="1"))],
+        ),
+    )
+    assert "mydataset" in datasets
+    assert dataset == datasets["mydataset"] == expected
+
+    tree.add(("bar",), Meta(version_id="2"), None)
+    old, new = datasets.update("mydataset")
+    assert old == dataset
+    assert old != new
+
+    assert expected.lock
+    new_files = [
+        *expected.lock.files,
+        FileInfo(relpath="bar", meta=Meta(version_id="2")),
+    ]
+
+    expected = evolve_recursive(expected, lock={"files": new_files})
+    assert new == datasets["mydataset"] == expected
+
+    # noop
+    old, new = datasets.update("mydataset")
+    assert old == new
+
+
+def test_dvc_dump(tmp_dir, dvc):
+    manifest_path = os.path.join(tmp_dir, "dvc.yaml")
+    spec = DVCDatasetSpec(
+        name="mydataset", url=tmp_dir.fs_path, type="dvc", path="path", rev="main"
+    )
+    lock = DVCDatasetLock(rev_lock="0" * 40, **spec.to_dict())
+    dataset = DVCDataset(manifest_path=manifest_path, spec=spec, lock=lock)
+
+    dvc.datasets.dump(dataset)
+
+    spec_d = {
+        "name": "mydataset",
+        "type": "dvc",
+        "url": tmp_dir.fs_path,
+        "path": "path",
+        "rev": "main",
+    }
+    assert (tmp_dir / "dvc.yaml").parse() == {"datasets": [spec_d]}
+    assert (tmp_dir / "dvc.lock").parse() == {
+        "schema": "2.0",
+        "stages": {},
+        "datasets": [{**spec_d, "rev_lock": "0" * 40}],
+    }
+
+    dvc._reset()
+    assert "_datasets" not in vars(dvc.datasets)
+    # test that we can read them back
+    assert dvc.datasets["mydataset"] == dataset
+
+
+def test_dvcx_dump(tmp_dir, dvc):
+    manifest_path = os.path.join(tmp_dir, "dvc.yaml")
+    spec = DatasetSpec(name="mydataset", url="dataset", type="dvcx")
+    dt = datetime.now(tz=timezone.utc)
+    lock = DVCXDatasetLock(version=1, created_at=dt, **spec.to_dict())
+    dataset = DVCXDataset(manifest_path=manifest_path, spec=spec, lock=lock)
+
+    dvc.datasets.dump(dataset)
+
+    spec_d = {"name": "mydataset", "type": "dvcx", "url": "dataset"}
+    assert (tmp_dir / "dvc.yaml").parse() == {"datasets": [spec_d]}
+    assert (tmp_dir / "dvc.lock").parse() == {
+        "schema": "2.0",
+        "stages": {},
+        "datasets": [{**spec_d, "version": 1, "created_at": dt.isoformat()}],
+    }
+
+    dvc._reset()
+    assert "_datasets" not in vars(dvc.datasets)
+    # test that we can read them back
+    assert dvc.datasets["mydataset"] == dataset
+
+
+def test_url_dump(tmp_dir, dvc):
+    manifest_path = os.path.join(tmp_dir, "dvc.yaml")
+    spec = DatasetSpec(name="mydataset", url="s3://dataset", type="url")
+    files = [FileInfo(relpath="foo", meta=Meta(version_id="1"))]
+    lock = URLDatasetLock(meta=Meta(isdir=True), files=files, **spec.to_dict())
+    dataset = URLDataset(manifest_path=manifest_path, spec=spec, lock=lock)
+
+    dvc.datasets.dump(dataset)
+
+    spec_d = {"name": "mydataset", "url": "s3://dataset", "type": "url"}
+    assert (tmp_dir / "dvc.yaml").parse() == {"datasets": [spec_d]}
+    assert (tmp_dir / "dvc.lock").parse() == {
+        "schema": "2.0",
+        "datasets": [
+            {
+                **spec_d,
+                "meta": {"isdir": True},
+                "files": [{"relpath": "foo", "meta": {"version_id": "1"}}],
+            }
+        ],
+        "stages": {},
+    }
+
+    dvc._reset()
+    assert "_datasets" not in vars(dvc.datasets)
+    # test that we can read them back
+    assert dvc.datasets["mydataset"] == dataset
+
+
+def test_invalidation(tmp_dir, dvc):
+    manifest_path = os.path.join(tmp_dir, "dvc.yaml")
+    spec = DatasetSpec(name="mydataset", url="url1", type="url")
+    lock = DVCXDatasetLock(
+        name="mydataset",
+        url="dataset",
+        type="dvcx",
+        version=1,
+        created_at=datetime.now(tz=timezone.utc),
+    )
+    dvc.datasets._dump_spec(manifest_path, spec)
+    dvc.datasets._dump_lock(manifest_path, lock)
+
+    assert dvc.datasets["mydataset"] == URLDataset(
+        manifest_path=manifest_path,
+        spec=spec,
+        lock=None,  # lock should be discarded
+    )
+
+
+def test_dvc_dataset_pipeline(tmp_dir, dvc, scm):
+    dvc.datasets.add("mydataset", tmp_dir.fs_path, "dvc")
+
+    stage = dvc.stage.add(cmd="echo", name="train", deps=["ds://mydataset"])
+    assert (tmp_dir / "dvc.yaml").parse() == {
+        "datasets": [{"name": "mydataset", "url": tmp_dir.fs_path, "type": "dvc"}],
+        "stages": {"train": {"cmd": "echo", "deps": ["ds://mydataset"]}},
+    }
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "new"}}]}
+    assert dvc.reproduce() == [stage]
+
+    d = (tmp_dir / "dvc.lock").parse()
+    assert d["stages"]["train"]["deps"][0] == {
+        "path": "ds://mydataset",
+        "dataset": d["datasets"][0],
+    }
+
+    assert dvc.status() == {}
+    assert dvc.reproduce() == []
+
+    tmp_dir.scm_gen("foo", "foo", commit="add foo")
+    dvc.datasets.update("mydataset")
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "modified"}}]}
+    assert dvc.reproduce() == [stage]
+
+
+def test_dvcx_dataset_pipeline(mocker, tmp_dir, dvc):
+    version_info = [MockedDVCXVersionInfo(1), MockedDVCXVersionInfo(2)]
+    mocker.patch("dvc.repo.datasets._get_dataset_info", side_effect=version_info)
+
+    dvc.datasets.add("mydataset", "dataset", "dvcx")
+
+    stage = dvc.stage.add(cmd="echo", name="train", deps=["ds://mydataset"])
+    assert (tmp_dir / "dvc.yaml").parse() == {
+        "datasets": [{"name": "mydataset", "url": "dataset", "type": "dvcx"}],
+        "stages": {"train": {"cmd": "echo", "deps": ["ds://mydataset"]}},
+    }
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "new"}}]}
+    assert dvc.reproduce() == [stage]
+
+    d = (tmp_dir / "dvc.lock").parse()
+    assert d["stages"]["train"]["deps"][0] == {
+        "path": "ds://mydataset",
+        "dataset": d["datasets"][0],
+    }
+
+    assert dvc.status() == {}
+    assert dvc.reproduce() == []
+
+    dvc.datasets.update("mydataset")
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "modified"}}]}
+    assert dvc.reproduce() == [stage]
+
+
+def test_url_dataset_pipeline(mocker, tmp_dir, dvc):
+    tree = Tree()
+    tree.add(("foo",), Meta(version_id="1"), None)
+    tree.digest(with_meta=True)
+    tree_meta = Meta(isdir=True)
+
+    def mocked_save(d):
+        d.meta, d.obj, d.hash_info = tree_meta, tree, HashInfo("md5", "value.dir")
+
+    mocker.patch.object(Dependency, "save", mocked_save)
+
+    dvc.datasets.add("mydataset", "s3://mydataset", "url")
+
+    stage = dvc.stage.add(cmd="echo", name="train", deps=["ds://mydataset"])
+    assert (tmp_dir / "dvc.yaml").parse() == {
+        "datasets": [{"name": "mydataset", "url": "s3://mydataset", "type": "url"}],
+        "stages": {"train": {"cmd": "echo", "deps": ["ds://mydataset"]}},
+    }
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "new"}}]}
+    assert dvc.reproduce() == [stage]
+
+    d = (tmp_dir / "dvc.lock").parse()
+    assert d["stages"]["train"]["deps"][0] == {
+        "path": "ds://mydataset",
+        "dataset": d["datasets"][0],
+    }
+
+    assert dvc.status() == {}
+    assert dvc.reproduce() == []
+
+    tree.add(("bar",), Meta(version_id="2"), None)
+    dvc.datasets.update("mydataset")
+
+    assert dvc.status() == {"train": [{"changed deps": {"ds://mydataset": "modified"}}]}
+    assert dvc.reproduce() == [stage]
+
+
+def test_pipeline_when_not_in_sync(tmp_dir, dvc):
+    manifest_path = os.path.join(tmp_dir, "dvc.yaml")
+    spec = DatasetSpec(name="mydataset", url="url1", type="url")
+    lock = DVCXDatasetLock(
+        name="mydataset",
+        url="dataset",
+        type="dvcx",
+        version=1,
+        created_at=datetime.now(tz=timezone.utc),
+    )
+    dvc.datasets._dump_spec(manifest_path, spec)
+    dvc.datasets._dump_lock(manifest_path, lock)
+
+    dvc.stage.add(name="train", cmd="echo", deps=["ds://mydataset"])
+    assert dvc.status() == {
+        "train": [{"changed deps": {"ds://mydataset": "not in sync"}}]
+    }
+    with pytest.raises(ReproductionError) as exc:
+        dvc.reproduce()
+    assert "not in sync" in str(exc.value.__cause__)
+
+
+def test_collect(tmp_dir, dvc):
+    manifest_path1 = os.path.join(tmp_dir, "dvc.yaml")
+    dt = datetime.now(tz=timezone.utc)
+    spec = DatasetSpec(name="mydataset1", url="url1", type="dvcx")
+    lock = DVCXDatasetLock(version=1, created_at=dt, **spec.to_dict())
+    mydataset1 = DVCXDataset(manifest_path=manifest_path1, spec=spec, lock=lock)
+    dvc.datasets.dump(mydataset1)
+
+    (tmp_dir / "sub").mkdir()
+    manifest_path2 = os.path.join(tmp_dir, "sub", "dvc.yaml")
+    spec = DVCDatasetSpec(
+        name="mydataset2", url=tmp_dir.fs_path, type="dvc", path="path"
+    )
+    lock = DVCDatasetLock(rev_lock="0" * 40, **spec.to_dict())
+    mydataset2 = DVCDataset(manifest_path=manifest_path2, spec=spec, lock=lock)
+    dvc.datasets.dump(mydataset2)
+
+    dvc._reset()
+    assert "_datasets" not in vars(dvc.datasets)
+
+    assert len(dvc.datasets) == 2
+    assert "mydataset1" in dvc.datasets
+    assert "mydataset2" in dvc.datasets
+    assert list(iter(dvc.datasets)) == ["mydataset1", "mydataset2"]
+    assert dvc.datasets["mydataset1"] == mydataset1
+    assert dvc.datasets["mydataset2"] == mydataset2
+    assert dict(dvc.datasets.items()) == {
+        "mydataset1": mydataset1,
+        "mydataset2": mydataset2,
+    }
+
+    with pytest.raises(DatasetNotFoundError, match=r"^dataset not found$"):
+        dvc.datasets["not-existing"]

tests/integration/plots/conftest.py

@@ -86,15 +86,27 @@ def repo_with_plots(tmp_dir, scm, dvc, run_copy_metrics):
 @pytest.fixture
 def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
     def make():
+        # test subdir functionality
+        linear_subdir_a_v1 = [
+            {"x": 1, "y": 0.2},
+            {"x": 2, "y": 0.3},
+            {"x": 3, "y": 0.4},
+        ]
+        # test subdir default values for x = step
+        linear_subdir_b_v1 = [
+            {"step": 1, "y": 0.2},
+            {"step": 2, "y": 0.3},
+            {"step": 3, "y": 0.4},
+        ]
         linear_train_v1 = [
             {"x": 1, "y": 0.1},
             {"x": 2, "y": 0.2},
             {"x": 3, "y": 0.3},
         ]
         linear_test_v1 = [
-            {"x": 1, "y": 0.2},
-            {"x": 2, "y": 0.3},
-            {"x": 3, "y": 0.4},
+            {"x": 1, "y": 0.3},
+            {"x": 2, "y": 0.4},
+            {"x": 3, "y": 0.5},
         ]
 
         confusion_train_v1 = [
@@ -110,6 +122,8 @@ def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
             {"actual": 1, "predicted": 0},
         ]
 
+        (tmp_dir / "linear_subdir_a_src.json").dump_json(linear_subdir_a_v1)
+        (tmp_dir / "linear_subdir_b_src.json").dump_json(linear_subdir_b_v1)
         (tmp_dir / "linear_train_src.json").dump_json(linear_train_v1)
         (tmp_dir / "linear_test_src.json").dump_json(linear_test_v1)
         (tmp_dir / "confusion_train_src.json").dump_json(confusion_train_v1)
@@ -117,6 +131,8 @@ def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
 
         scm.add(
             [
+                "linear_subdir_a_src.json",
+                "linear_subdir_b_src.json",
                 "linear_train_src.json",
                 "linear_test_src.json",
                 "confusion_train_src.json",
@@ -125,6 +141,23 @@ def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
         )
         scm.commit("add data sources")
 
+        (tmp_dir / "subdirA").mkdir()
+        (tmp_dir / "subdirB").mkdir()
+
+        run_copy_metrics(
+            "linear_subdir_a_src.json",
+            "subdirA/linear_subdir.json",
+            name="linear_subdir",
+            outs=["subdirA/linear_subdir.json"],
+            commit="linear_subdir",
+        )
+        run_copy_metrics(
+            "linear_subdir_b_src.json",
+            "subdirB/linear_subdir.json",
+            name="linear_subdir",
+            outs=["subdirB/linear_subdir.json"],
+            commit="linear_subdir",
+        )
         run_copy_metrics(
             "linear_train_src.json",
             "linear_train.json",
@@ -153,7 +186,16 @@ def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
             outs=["confusion_test.json"],
             commit="confusion_test",
         )
-        plots_config = [
+
+        subdir_a_config = {
+            "x": "x",
+            "y": "y",
+            "title": "subdir plots with x and y defined",
+        }
+
+        subdir_b_config = {"title": "subdir plots with default x and y"}
+
+        other_plots_config = [
             {
                 "linear_train_vs_test": {
                     "x": "x",
@@ -176,18 +218,35 @@ def repo_with_config_plots(tmp_dir, scm, dvc, run_copy_metrics):
         from dvc.utils.serialize import modify_yaml
 
         with modify_yaml("dvc.yaml") as dvcfile_content:
-            dvcfile_content["plots"] = plots_config
+            dvcfile_content["plots"] = [
+                {"subdirA": subdir_a_config},
+                {"subdirB": subdir_b_config},
+                *other_plots_config,
+            ]
 
         scm.add(["dvc.yaml", "dvc.lock"])
         scm.commit("commit dvc files")
+
+        # remove generate .gitignore that are considered as plot otherwise
+        (tmp_dir / "subdirA" / ".gitignore").unlink()
+        (tmp_dir / "subdirB" / ".gitignore").unlink()
+
         yield {
             "data": {
+                "subdirA/linear_subdir.json": linear_subdir_a_v1,
+                "subdirB/linear_subdir.json": linear_subdir_b_v1,
                 "linear_train.json": linear_train_v1,
                 "linear_test.json": linear_test_v1,
                 "confusion_train.json": confusion_train_v1,
                 "confusion_test.json": confusion_test_v1,
             },
-            "configs": {"dvc.yaml": plots_config},
+            "configs": {
+                "dvc.yaml": [
+                    {"subdirA/linear_subdir.json": subdir_a_config},
+                    {"subdirB/linear_subdir.json": subdir_b_config},
+                    *other_plots_config,
+                ]
+            },
         }
 
     return make

tests/integration/plots/test_plots.py

@@ -440,6 +440,8 @@ def test_repo_with_config_plots(tmp_dir, capsys, repo_with_config_plots):
     html_result = extract_vega_specs(
         html_path,
         [
+            "subdirA/linear_subdir.json",
+            "subdirB/linear_subdir.json",
             "linear_train_vs_test",
             "confusion_train_vs_test",
         ],
@@ -466,6 +468,22 @@ def test_repo_with_config_plots(tmp_dir, capsys, repo_with_config_plots):
         ]
         == ble
     )
+    assert (
+        html_result["subdirA/linear_subdir.json"]["layer"][1]["encoding"]["x"]["field"]
+        == "x"
+    )
+    assert (
+        html_result["subdirA/linear_subdir.json"]["layer"][1]["encoding"]["y"]["field"]
+        == "y"
+    )
+    assert (
+        html_result["subdirB/linear_subdir.json"]["layer"][1]["encoding"]["x"]["field"]
+        == "step"
+    )
+    assert (
+        html_result["subdirB/linear_subdir.json"]["layer"][1]["encoding"]["y"]["field"]
+        == "y"
+    )
 
 
 @pytest.mark.vscode
@@ -481,3 +499,138 @@ def test_repo_with_dvclive_plots(tmp_dir, capsys, repo_with_dvclive_plots):
         }
         assert json_result == expected_result
         assert split_json_result == expected_result
+
+
+@pytest.mark.vscode
+def test_nested_x_defn_collection(tmp_dir, dvc, scm, capsys):
+    rel_pipeline_dir = "pipelines/data-increment"
+    pipeline_rel_dvclive_metrics_dir = "dvclive/plots/metrics"
+    pipeline_rel_other_logger_dir = "other/logger"
+
+    dvc_rel_dvclive_metrics_dir = (
+        f"{rel_pipeline_dir}/{pipeline_rel_dvclive_metrics_dir}"
+    )
+    dvc_rel_other_logger_dir = f"{rel_pipeline_dir}/{pipeline_rel_other_logger_dir}"
+
+    pipeline_dir = tmp_dir / rel_pipeline_dir
+    dvclive_metrics_dir = pipeline_dir / pipeline_rel_dvclive_metrics_dir
+    dvclive_metrics_dir.mkdir(parents=True)
+    other_logger_dir = pipeline_dir / pipeline_rel_other_logger_dir
+    other_logger_dir.mkdir(parents=True)
+
+    (pipeline_dir / "dvc.yaml").dump(
+        {
+            "plots": [
+                {
+                    "Error vs max_leaf_nodes": {
+                        "template": "simple",
+                        "x": {
+                            f"{pipeline_rel_dvclive_metrics_dir}"
+                            "/Max_Leaf_Nodes.tsv": "Max_Leaf_Nodes"
+                        },
+                        "y": {f"{pipeline_rel_dvclive_metrics_dir}/Error.tsv": "Error"},
+                    }
+                },
+                {
+                    f"{pipeline_rel_other_logger_dir}/multiple_metrics.json": {
+                        "x": "x",
+                        "y": ["y1", "y2"],
+                    },
+                },
+                {
+                    f"{pipeline_rel_dvclive_metrics_dir}/Error.tsv": {"y": ["Error"]},
+                },
+                {
+                    "max leaf nodes": {
+                        "y": {
+                            f"{pipeline_rel_dvclive_metrics_dir}"
+                            "/Max_Leaf_Nodes.tsv": "Max_Leaf_Nodes"
+                        }
+                    },
+                },
+            ]
+        },
+    )
+    dvclive_metrics_dir.gen(
+        {
+            "Error.tsv": "step\tError\n" "0\t0.11\n" "1\t0.22\n" "2\t0.44\n",
+            "Max_Leaf_Nodes.tsv": "step\tMax_Leaf_Nodes\n"
+            "0\t5\n"
+            "1\t50\n"
+            "2\t500\n",
+        }
+    )
+    (other_logger_dir / "multiple_metrics.json").dump(
+        [
+            {"x": 0, "y1": 0.1, "y2": 10},
+            {"x": 1, "y1": 0.2, "y2": 22},
+        ]
+    )
+
+    scm.commit("add dvc.yaml and metrics")
+
+    _, _, split_json_result = call(capsys, subcommand="diff")
+    assert len(split_json_result.keys()) == 1
+    assert len(split_json_result["data"].keys()) == 4
+
+    separate_x_file = split_json_result["data"]["Error vs max_leaf_nodes"][0]
+
+    assert separate_x_file["anchor_definitions"]["<DVC_METRIC_DATA>"] == [
+        {"Error": "0.11", "Max_Leaf_Nodes": "5", "step": "0", "rev": "workspace"},
+        {"Error": "0.22", "Max_Leaf_Nodes": "50", "step": "1", "rev": "workspace"},
+        {"Error": "0.44", "Max_Leaf_Nodes": "500", "step": "2", "rev": "workspace"},
+    ]
+
+    same_x_file = split_json_result["data"][
+        f"{dvc_rel_other_logger_dir}/multiple_metrics.json"
+    ][0]
+    assert same_x_file["anchor_definitions"]["<DVC_METRIC_DATA>"] == [
+        {
+            "x": 0,
+            "y1": 0.1,
+            "y2": 10,
+            "dvc_inferred_y_value": 0.1,
+            "field": "y1",
+            "rev": "workspace",
+        },
+        {
+            "x": 1,
+            "y1": 0.2,
+            "y2": 22,
+            "dvc_inferred_y_value": 0.2,
+            "field": "y1",
+            "rev": "workspace",
+        },
+        {
+            "x": 0,
+            "y1": 0.1,
+            "y2": 10,
+            "dvc_inferred_y_value": 10,
+            "field": "y2",
+            "rev": "workspace",
+        },
+        {
+            "x": 1,
+            "y1": 0.2,
+            "y2": 22,
+            "dvc_inferred_y_value": 22,
+            "field": "y2",
+            "rev": "workspace",
+        },
+    ]
+
+    inferred_x_from_str = split_json_result["data"][
+        f"{dvc_rel_dvclive_metrics_dir}/Error.tsv"
+    ][0]
+    assert inferred_x_from_str["anchor_definitions"]["<DVC_METRIC_DATA>"] == [
+        {"step": 0, "Error": "0.11", "rev": "workspace"},
+        {"step": 1, "Error": "0.22", "rev": "workspace"},
+        {"step": 2, "Error": "0.44", "rev": "workspace"},
+    ]
+
+    inferred_x_from_dict = split_json_result["data"]["max leaf nodes"][0]
+    assert inferred_x_from_dict["anchor_definitions"]["<DVC_METRIC_DATA>"] == [
+        {"step": 0, "Max_Leaf_Nodes": "5", "rev": "workspace"},
+        {"step": 1, "Max_Leaf_Nodes": "50", "rev": "workspace"},
+        {"step": 2, "Max_Leaf_Nodes": "500", "rev": "workspace"},
+    ]

tests/integration/test_studio_live_experiments.py

@@ -8,8 +8,12 @@ from dvc.env import (
     DVC_STUDIO_TOKEN,
     DVC_STUDIO_URL,
 )
+from dvc.repo import Repo
+from dvc.testing.scripts import COPY_SCRIPT
+from dvc.utils.studio import get_subrepo_relpath
 
 
+@pytest.mark.studio
 @pytest.mark.parametrize("tmp", [True, False])
 @pytest.mark.parametrize("offline", [True, False])
 def test_post_to_studio(
@@ -66,10 +70,45 @@ def test_post_to_studio(
         }
 
 
+@pytest.mark.studio
 @pytest.mark.parametrize("tmp", [True, False])
-def test_post_to_studio_custom_message(
-    tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch, tmp
-):
+def test_post_to_studio_subdir(tmp_dir, scm, mocker, monkeypatch, tmp):
+    live_exp_subdir = "project_a"
+
+    tmp_dir.scm_gen(
+        {
+            live_exp_subdir: {
+                "params.yaml": "foo: 1",
+                "metrics.yaml": "foo: 1",
+                "copy.py": COPY_SCRIPT.encode("utf-8"),
+            },
+        },
+        commit="git init",
+    )
+
+    project_a_dvc = Repo.init(tmp_dir / live_exp_subdir, subdir=True)
+    with monkeypatch.context() as m:
+        m.chdir(project_a_dvc.root_dir)
+
+        exp_stage = project_a_dvc.run(
+            cmd="python copy.py params.yaml metrics.yaml",
+            metrics_no_cache=["metrics.yaml"],
+            params=["foo"],
+            name="copy-file",
+        )
+
+        scm.add(
+            [
+                ".gitignore",
+                "copy.py",
+                "dvc.lock",
+                "dvc.yaml",
+                "metrics.yaml",
+                "params.yaml",
+            ]
+        )
+        scm.commit("dvc init project_a")
+
     valid_response = mocker.MagicMock()
     valid_response.status_code = 200
     mocked_post = mocker.patch("requests.post", return_value=valid_response)
@@ -79,12 +118,16 @@ def test_post_to_studio_custom_message(
     monkeypatch.setenv(env.STUDIO_TOKEN, "STUDIO_TOKEN")
 
     baseline_sha = scm.get_rev()
-    exp_rev = first(
-        dvc.experiments.run(
-            exp_stage.addressing, params=["foo=1"], tmp_dir=tmp, message="foo"
+    with monkeypatch.context() as m:
+        m.chdir(project_a_dvc.root_dir)
+        exp_rev = first(
+            project_a_dvc.experiments.run(
+                exp_stage.addressing, params=["foo=24"], tmp_dir=tmp
+            )
         )
-    )
-    name = dvc.experiments.get_exact_name([exp_rev])[exp_rev]
+
+    name = project_a_dvc.experiments.get_exact_name([exp_rev])[exp_rev]
+    project_a_dvc.close()
     assert mocked_post.call_count == 2
 
     start_call = mocked_post.call_args_list[0]
@@ -94,7 +137,57 @@ def test_post_to_studio_custom_message(
         "repo_url": "STUDIO_REPO_URL",
         "baseline_sha": baseline_sha,
         "name": name,
-        "params": {"params.yaml": {"foo": 1}},
+        "params": {"params.yaml": {"foo": 24}},
+        "subdir": live_exp_subdir,
         "client": "dvc",
-        "message": "foo",
     }
+
+
+@pytest.mark.studio
+def test_monorepo_relpath(tmp_dir, scm):
+    from dvc.repo.destroy import destroy
+
+    tmp_dir.gen({"project_a": {}, "subdir/project_b": {}})
+
+    non_monorepo = Repo.init(tmp_dir)
+    assert get_subrepo_relpath(non_monorepo) == ""
+
+    destroy(non_monorepo)
+
+    monorepo_project_a = Repo.init(tmp_dir / "project_a", subdir=True)
+
+    assert get_subrepo_relpath(monorepo_project_a) == "project_a"
+
+    monorepo_project_b = Repo.init(tmp_dir / "subdir" / "project_b", subdir=True)
+
+    assert get_subrepo_relpath(monorepo_project_b) == "subdir/project_b"
+
+
+@pytest.mark.studio
+def test_virtual_monorepo_relpath(tmp_dir, scm):
+    from dvc.fs.git import GitFileSystem
+    from dvc.repo.destroy import destroy
+
+    tmp_dir.gen({"project_a": {}, "subdir/project_b": {}})
+    scm.commit("initial commit")
+    gfs = GitFileSystem(scm=scm, rev="master")
+
+    non_monorepo = Repo.init(tmp_dir)
+    non_monorepo.fs = gfs
+    non_monorepo.root_dir = "/"
+
+    assert get_subrepo_relpath(non_monorepo) == ""
+
+    destroy(non_monorepo)
+
+    monorepo_project_a = Repo.init(tmp_dir / "project_a", subdir=True)
+    monorepo_project_a.fs = gfs
+    monorepo_project_a.root_dir = "/project_a"
+
+    assert get_subrepo_relpath(monorepo_project_a) == "project_a"
+
+    monorepo_project_b = Repo.init(tmp_dir / "subdir" / "project_b", subdir=True)
+    monorepo_project_b.fs = gfs
+    monorepo_project_b.root_dir = "/subdir/project_b"
+
+    assert get_subrepo_relpath(monorepo_project_b) == "subdir/project_b"

tests/unit/command/test_dataset.py

@@ -0,0 +1,137 @@
+from datetime import datetime
+
+import pytest
+
+from dvc.cli import main
+
+
+@pytest.mark.parametrize(
+    "spec,lock,expected_output",
+    [
+        (
+            {"name": "ds", "url": "url", "type": "dvc", "path": "path", "rev": "main"},
+            {"rev_lock": "0" * 40},
+            "Adding ds (url:/path @ main)\n",
+        ),
+        (
+            {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
+            {"rev_lock": "0" * 40, "version": 1, "created_at": datetime.now()},
+            "Adding mydataset (dvcx://dataset @ v1)\n",
+        ),
+        (
+            {"name": "mydataset", "url": "s3://bucket/path", "type": "url"},
+            {
+                "files": [{"relpath": "foo", "meta": {"version_id": 1}}],
+                "meta": {"isdir": True},
+            },
+            "Adding mydataset (s3://bucket/path)\n",
+        ),
+    ],
+)
+def test_add(dvc, capsys, mocker, spec, lock, expected_output):
+    dataset = dvc.datasets._build_dataset("dvc.yaml", spec, spec | lock)
+
+    m = mocker.patch("dvc.repo.datasets.Datasets.add", return_value=dataset)
+
+    assert main(["dataset", "add", spec["name"], f"--{spec['type']}", spec["url"]]) == 0
+    out, err = capsys.readouterr()
+    assert out == expected_output
+    assert not err
+    m.assert_called_once()
+
+
+def test_add_already_exists(dvc, caplog, mocker):
+    spec = {"name": "ds", "url": "url", "type": "dvc"}
+    dataset = dvc.datasets._build_dataset("dvc.yaml", spec, None)
+    mocker.patch("dvc.repo.datasets.Datasets.get", return_value=dataset)
+
+    assert main(["dataset", "add", "ds", "--dvcx", "dataset"]) == 255
+    assert "ds already exists in dvc.yaml, use the --force to overwrite" in caplog.text
+
+
+@pytest.mark.parametrize("lock", ["missing", "unchanged", "updated"])
+@pytest.mark.parametrize(
+    "spec,old_lock,new_lock,expected_outputs",
+    [
+        (
+            {"name": "ds", "url": "url", "type": "dvc", "path": "path", "rev": "main"},
+            {"rev_lock": "0" * 40},
+            {"rev_lock": "1" * 40},
+            {
+                "missing": "Updating ds (url:/path @ main)\n",
+                "unchanged": "Nothing to update\n",
+                "updated": "Updating ds (000000000 -> 111111111)\n",
+            },
+        ),
+        (
+            {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
+            {"rev_lock": "0" * 40, "version": 1, "created_at": datetime.now()},
+            {"version": 2},
+            {
+                "missing": "Updating mydataset (dvcx://dataset @ v2)\n",
+                "unchanged": "Nothing to update\n",
+                "updated": "Updating mydataset (v1 -> v2)\n",
+            },
+        ),
+        (
+            {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
+            {"rev_lock": "0" * 40, "version": 2, "created_at": datetime.now()},
+            {"version": 1},
+            {
+                "missing": "Updating mydataset (dvcx://dataset @ v1)\n",
+                "unchanged": "Nothing to update\n",
+                "updated": "Downgrading mydataset (v2 -> v1)\n",
+            },
+        ),
+        (
+            {"name": "mydataset", "url": "s3://bucket/path", "type": "url"},
+            {
+                "files": [
+                    {"relpath": "bar", "meta": {"version_id": 2}},
+                    {"relpath": "baz", "meta": {"version_id": 3}},
+                    {"relpath": "foo", "meta": {"version_id": 1}},
+                ],
+                "meta": {"isdir": True},
+            },
+            {
+                "files": [
+                    {"relpath": "bar", "meta": {"version_id": 2}},  # unchanged
+                    {"relpath": "baz", "meta": {"version_id": 4}},  # modified
+                    # `foo` deleted
+                    {"relpath": "foobar", "meta": {"version_id": 5}},  # new
+                ],
+                "meta": {"isdir": True},
+            },
+            {
+                "missing": "Updating mydataset (s3://bucket/path)\n",
+                "unchanged": "Nothing to update\n",
+                "updated": (
+                    "Updating mydataset (s3://bucket/path)\n"
+                    "M\tbaz\n"
+                    "A\tfoobar\n"
+                    "D\tfoo\n"
+                ).expandtabs(8),
+            },
+        ),
+    ],
+)
+def test_update(dvc, capsys, mocker, spec, old_lock, new_lock, expected_outputs, lock):
+    if lock == "missing":
+        new_lock = spec | old_lock | new_lock
+        old_lock = None
+    elif lock == "unchanged":
+        old_lock = new_lock = spec | old_lock
+    else:
+        old_lock = spec | old_lock
+        new_lock = old_lock | new_lock
+
+    old = dvc.datasets._build_dataset("dvc.yaml", spec, old_lock)
+    new = dvc.datasets._build_dataset("dvc.yaml", spec, new_lock)
+
+    m = mocker.patch("dvc.repo.datasets.Datasets.update", return_value=(old, new))
+    assert main(["dataset", "update", spec["name"]]) == 0
+    out, err = capsys.readouterr()
+    assert out == expected_outputs[lock]
+    assert not err
+
+    m.assert_called_once()

tests/unit/command/test_help.py

@@ -29,7 +29,7 @@ def command_tuples():
 
     # the no. of commands will usually go up,
     # but if we ever remove commands and drop below, adjust the magic number accordingly
-    assert len(commands) >= 112
+    assert len(commands) >= 116
     return sorted(commands)
 
 

tests/unit/command/test_studio.py

@@ -1,4 +1,4 @@
-from dvc_studio_client.auth import AuthenticationExpired
+from dvc_studio_client.auth import AuthorizationExpiredError
 
 from dvc.cli import main
 from dvc.utils.studio import STUDIO_URL
@@ -6,9 +6,9 @@ from dvc.utils.studio import STUDIO_URL
 
 def test_studio_login_token_check_failed(mocker):
     mocker.patch(
-        "dvc_studio_client.auth.get_access_token", side_effect=AuthenticationExpired
+        "dvc_studio_client.auth.get_access_token",
+        side_effect=AuthorizationExpiredError,
     )
-
     assert main(["studio", "login"]) == 1
 
 
@@ -52,8 +52,8 @@ def test_studio_login_arguments(mocker):
         token_name="token_name",
         hostname="https://example.com",
         scopes="experiments",
-        use_device_code=True,
         client_name="DVC",
+        open_browser=False,
     )
 
 

tests/unit/repo/test_repo.py

@@ -135,51 +135,3 @@ def test_dynamic_cache_initialization(tmp_dir, scm):
     dvc.close()
 
     Repo(str(tmp_dir)).close()
-
-
-def test_monorepo_relpath(tmp_dir, scm):
-    from dvc.repo.destroy import destroy
-
-    tmp_dir.gen({"project_a": {}, "subdir/project_b": {}})
-
-    non_monorepo = Repo.init(tmp_dir)
-    assert non_monorepo.subrepo_relpath == ""
-
-    destroy(non_monorepo)
-
-    monorepo_project_a = Repo.init(tmp_dir / "project_a", subdir=True)
-
-    assert monorepo_project_a.subrepo_relpath == "project_a"
-
-    monorepo_project_b = Repo.init(tmp_dir / "subdir" / "project_b", subdir=True)
-
-    assert monorepo_project_b.subrepo_relpath == "subdir/project_b"
-
-
-def test_virtual_monorepo_relpath(tmp_dir, scm):
-    from dvc.fs.git import GitFileSystem
-    from dvc.repo.destroy import destroy
-
-    tmp_dir.gen({"project_a": {}, "subdir/project_b": {}})
-    scm.commit("initial commit")
-    gfs = GitFileSystem(scm=scm, rev="master")
-
-    non_monorepo = Repo.init(tmp_dir)
-    non_monorepo.fs = gfs
-    non_monorepo.root_dir = "/"
-
-    assert non_monorepo.subrepo_relpath == ""
-
-    destroy(non_monorepo)
-
-    monorepo_project_a = Repo.init(tmp_dir / "project_a", subdir=True)
-    monorepo_project_a.fs = gfs
-    monorepo_project_a.root_dir = "/project_a"
-
-    assert monorepo_project_a.subrepo_relpath == "project_a"
-
-    monorepo_project_b = Repo.init(tmp_dir / "subdir" / "project_b", subdir=True)
-    monorepo_project_b.fs = gfs
-    monorepo_project_b.root_dir = "/subdir/project_b"
-
-    assert monorepo_project_b.subrepo_relpath == "subdir/project_b"