Repository: iterative/dvc
Pull request #10287 (2024-02-07 13:41:24)

add support for tracking remote dataset
[![asciicast](https://asciinema.org/a/wU8oEI5Iibuw0k4s4wia7wopq.svg)](https://asciinema.org/a/wU8oEI5Iibuw0k4s4wia7wopq)

--- 


This adds support for virtually tracking a dataset such as dvcx dataset, dataset from remote dvc/git registries, and cloud-versioned remotes.

This PR introduces two different commands under `ds` namespace: `add` and `update`.

`dvc ds add` command adds the dataset to `dvc.yaml` and tracks the sources in `dvc.lock` file. Similarly, `dvc ds update` updates the sources in `dvc.lock` file.

#### Example Usage:

```console
dvc ds add --name dogs --url dvcx://dogs # tracking dvcx dataset
# tracking dvc/git registries
dvc ds add --name example --url dvc://git@github.com/iterative/example.git
# cloud-versioned-remotes
dvc ds add --name versioning --url s3://cloud-versioning-demo
```

To update, specify the name.
```console
dvc ds update <name>
```

`dvc ds add` freezes (in the traditional sense of `dvc import/import-url`). It keeps a "specification" of the dataset in `dvc.yaml` and also snapshots information about the dataset in the `dvc.lock` file. They are kept inside `datasets` section in both `dvc.yaml` and `dvc.lock` files.

This metadata is used in the pipelines. You can add a dependency to your stage using `ds://` scheme, followed by the name of the dataset in `dvc.yaml` file. As it is used in pipelines, the `name` of the dataset has to be unique in the repository. Different dataset of same names are not allowed. On `dvc repro`, `dvc` copies the snapshot information about the particular dataset into the `deps` field for the stage in `dvc.lock`. On successive invocation, `dvc` will compare the information from the deps field of the lock with the snapshot information in the `datasets` section and decides whether to rerun or not.

As the dataset is frozen, `dvc repro` won't rerun until the dataset is updated via `dvc update`.

Here are some examples for how `dvc.yaml` looks like:
```yaml
datasets:
- name: dogs
  url: dvcx://dogs
  type: dvcx
- name: example-get-started
  url: git@github.com:iterative/example-get-started.git
  type: dvc
  path: path
- name: dogs2
  url: dvcx://dogs@v2
  type: dvcx
- name: cloud-versioning-demo
  url: s3://cloud-versioning-demo
  type: url
```

```yaml
schema: '2.0'
datasets:
- name: dogs
  url: dvcx://dogs
  type: dvcx
  version: 3
  created_at: '2023-12-11T10:32:05.942708+00:00'
- name: example-get-started
  url: git@github.com:iterative/example-get-started.git
  type: dvc
  path: path
  rev_lock: df75c16ef61f0772d6e4bb27ba4617b06b4b5398
- name: cloud-versioning-demo
  url: s3://cloud-versioning-demo
  type: url
  meta:
    isdir: true
    size: 323919
    nfiles: 33
  files:
  - relpath: myproject/model.pt
    meta:
      size: 106433
      version_id: 5qrtnhnQ4fBzV73kqqK6pMGhTOzd_IPr
      etag: 3bc0028677ce6fb65bec8090c248b002
# truncated
```

> [!NOTE]  
> The `files` in `datasets` section differs slightly compared to cloud-versioned outputs, as this separates `relpath` and metadata in `meta`, whereas `output` has them together.

The pipeline stages keep them in `dataset` section of individual deps.

```console
dvc stage add -n train -d ds://dogs python train.py
cat dvc.yaml
```

```yaml
# truncated
stages:
  train:
    cmd: python train.py
    deps:
    - ds://dogs
```

When it is reproduced, the `dvc.lock` will look something like follows:
```yaml
# truncated
stages:
  train:
    cmd: python train.py
    deps:
    - path: ds://dogs
      dataset:
        name: dogs
        url: dvcx://dogs
        type: dvcx
        version: 3
        created_at: '2023-12-11T10:32:05.942708+00:00'
```



dberenbaum: Notes from discussion with @skshetry:
* Require a name for now
* Treat datasets as frozen -- only possible to update using `dvc ds update`
* Include version info only in `dvc.lock`, not `dvc.yaml`
* Revisit https://github.com/iterative/studio/issues/8117
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) Report
Attention: `128 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`9fbe5d9`)](https://app.codecov.io/gh/iterative/dvc/commit/9fbe5d929da6ae43af19a03f70c2958e0541a0ff?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) 90.42% compared to head [(`84a5cf5`)](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) 89.94%.
> Report is 1 commits behind head on main.

| [Files](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) | Patch % | Lines |
|---|---|---|
| [dvc/repo/datasets.py](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative#diff-ZHZjL3JlcG8vZGF0YXNldHMucHk=) | 43.87% | [87 Missing :warning: ](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) |
| [dvc/dvcfile.py](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative#diff-ZHZjL2R2Y2ZpbGUucHk=) | 20.83% | [19 Missing :warning: ](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) |
| [dvc/commands/dataset.py](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative#diff-ZHZjL2NvbW1hbmRzL2RhdGFzZXQucHk=) | 62.16% | [14 Missing :warning: ](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) |
| [dvc/dependency/dataset.py](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative#diff-ZHZjL2RlcGVuZGVuY3kvZGF0YXNldC5weQ==) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) |
| [dvc/dependency/dvcx.py](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative#diff-ZHZjL2RlcGVuZGVuY3kvZHZjeC5weQ==) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main   #10287      +/-   ##
==========================================
- Coverage   90.42%   89.94%   -0.49%     
==========================================
  Files         494      496       +2     
  Lines       37824    37984     +160     
  Branches     5470     5476       +6     
==========================================
- Hits        34204    34163      -41     
- Misses       2986     3141     +155     
- Partials      634      680      +46     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/iterative/dvc/pull/10287?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=iterative).

dberenbaum: @skshetry What's the status of this?
skshetry: This should be ready for review. The above asciinema is stale (although the only change is removal of `--type` with `dvcx://`, `dvc://` and `dvc+https://` schema). And support for pipelines with `ds://` scheme.

eg:
```console
dvc stage add -ntest -d ds://dataset_name "command"
```

I'll try to update the asciinema recording tomorrow, and add more descriptions in the PR.

skshetry: Updated the asciinema recording above.
skshetry: The `datasets` section implemented in this PR is as a list in `dvc.yaml`. `metrics`/`params`/`plots` are also a list.
But the `artifacts` section is a dictionary, which has a similar schema/format as `datasets` with `type`/`path` etc.

I find lists simpler and easier to use, with fewer indentations. But I am still undecided.
So, I'll keep an option open for changing the `datasets` schema until we document it.


dberenbaum: Nice @skshetry! It works well for me in doing QA over various different types. I mostly have UI questions.

> $ dvc ds add dvc://git@github.com:iterative/example-get-started.git@main#path=data/data.xml

Do you have thoughts on trying to incorporate the path and rev into the url somehow like this? It feels like we are now stuck needing both complex CLI options (`--rev` and `--path`) and a complex URL format (`dvc+https://github.com...`). I think we might be better either going all the way in on the URL format or else reverting back to `--type` (but then we still need `--rev`/`--path` only for dvc datasets).

Other questions:
- Should `--url` be a positional arg since it's required?
- For https and hdfs, how do we version them and what's the intended use? I tried `dvc ds add --url https://dvc.org/ --name dvc-website` and got `ERROR: unexpected error - 'HTTPSFileSystem' object has no attribute 'coalesce_version'`
- `remote://remote_name/path/to/file/or/dir (see `dvc remote`)` - is there an example of how to use this?
skshetry: > > $ dvc ds add dvc://git@github.com:iterative/example-get-started.git@main#path=data/data.xml

I am not against supporting this. But I find this url to be very complex, and I doubt if I'll remember it when I need it.

Even if we do support both, I'd prefer keeping `--rev`/`--path`. I don't think it is complex or complicated to support.

Also, parsing these scp-style path is difficult IIUC as `git@` is optional (cc @iterative/dvc, maybe you have some ideas). One idea that I have is to convert these scp-paths into ssh-paths and then parse, but I am not sure how robust this is, i.e 

```diff
- git@github.com:iterative/example-get-started.git
+ ssh://git@github.com/iterative/example-get-started.git
```



> * Should `--url` be a positional arg since it's required?

yeah, both `name` and `url` should be a positional argument.


> * For https and hdfs, how do we version them and what's the intended use? I tried `dvc ds add --url https://dvc.org/ --name dvc-website` and got `ERROR: unexpected error - 'HTTPSFileSystem' object has no attribute 'coalesce_version'`

I am not sure about the intended use case. It might be useful in Studio to show if the source changed between commits from the metadata, but I am not sure how useful that information is. We could disable it for now.

>     * `remote://remote_name/path/to/file/or/dir (see `dvc remote`)` - is there an example of how to use this?

This should just work without doing anything.

```console
dvc remote add myremote s3://bucketname/prefix
dvc ds add --url remote://myremote/path/to/file/or/dir --name name
```


dberenbaum: > I am not against supporting this. But I find this url to be very complex, and I doubt if I'll remember it when I need it.

Yup, it might be worse, especially if you think we should keep `--path` and `--rev`. Do you think we should revert to using `--type` then?

> > * For https and hdfs, how do we version them and what's the intended use? I tried `dvc ds add --url https://dvc.org/ --name dvc-website` and got `ERROR: unexpected error - 'HTTPSFileSystem' object has no attribute 'coalesce_version'`
> 
> I am not sure about the intended use case. It might be useful in Studio to show if the source changed between commits from the metadata, but I am not sure how useful that information is. We could disable it for now.

Does it include a hash in `dvc.lock`? If the dataset changes, what happens if I do `dvc repro` with the old dataset?

skshetry: > Yup, it might be worse, especially if you think we should keep --path and --rev. Do you think we should revert to using --type then?

Is `--url dvc://git@github.com:iterative/example-get-started.git` worse than `--url git@github.com:iterative/example-get-started.git --type dvc`?


> Does it include a hash in `dvc.lock`? 

It has an etag, but we don't have a hash of the content (eg: md5). To hash the content, you'd have to stream them from the remote.

> If the dataset changes, what happens if I do `dvc repro` with the old dataset?

Since the import is _frozen_ on `dvc ds add` or `dvc ds update`, we don't run any kind of checks during `dvc repro`. So it does not work with non-cloud-versioned remotes at all.

And non-cloud versioned remotes can not be frozen at all, so this will require a different solution, in which we check the sources during `dvc repro`, and update `dvc.lock` accordingly. This does not fit how the "dataset" is implemented, which requires to be frozen.
 



dberenbaum: I don't see a good reason to support non-versioned types like `https`, `hdfs`, and maybe even `remote` (I guess it depends if it's `version_aware`). It seems like external dependencies already support this scenario in a simpler way unless I'm missing something.
skshetry: I'll disable non-cloud versioned remotes/filesystems in the successive PRs. 
dberenbaum: Can you drop them from the parser help text in this PR? I was focused on that more than actually disabling them.
skshetry: > Can you drop them from the parser help text in this PR? I was focused on that more than actually disabling them.

I have dropped them from the help text.

Patch:
diff --git a/dvc/cli/parser.py b/dvc/cli/parser.py
index aaeca86bc..95ca31ce6 100644
--- a/dvc/cli/parser.py
+++ b/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,
diff --git a/dvc/commands/dataset.py b/dvc/commands/dataset.py
new file mode 100644
index 000000000..9cbc95c33
--- /dev/null
+++ b/dvc/commands/dataset.py
@@ -0,0 +1,205 @@
+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):
+        from urllib.parse import urlsplit
+
+        d = vars(self.args)
+        url_obj = urlsplit(self.args.url)
+        if url_obj.scheme == "dvcx":
+            d["type"] = "dvcx"
+        elif url_obj.scheme.startswith("dvc"):
+            d["type"] = "dvc"
+            protos = tuple(url_obj.scheme.split("+"))
+            if not protos or protos == ("dvc",) or protos == ("dvc", "ssh"):
+                d["url"] = url_obj.netloc + url_obj.path
+            else:
+                d["url"] = url_obj._replace(scheme=protos[1]).geturl()
+        else:
+            d["type"] = "url"
+
+        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)
+
+
+class CmdDatasetUpdate(CmdBase):
+    def display(self, name: str, dataset: "Dataset", new: "Dataset"):
+        from dvc.commands.checkout import log_changes
+        from dvc.ui import ui
+
+        if not dataset.lock:
+            return CmdDatasetAdd.display(name, new, "Updating")
+        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"
+            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("Updating", 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
+
+        with self.repo.scm_context:
+            try:
+                dataset, new = self.repo.datasets.update(**vars(self.args))
+            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)
+
+
+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,
+    )
+    ds_add_parser.add_argument(
+        "--url",
+        required=True,
+        help="""\
+Location of the data to download. 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`)
+dvcx://dataset_name
+
+To import data from dvc/git repositories, \
+add dvc:// schema to the repo url, e.g:
+dvc://git@github.com/iterative/example-get-started.git
+dvc+https://github.com/iterative/example-get-started.git""",
+    )
+    ds_add_parser.add_argument(
+        "--name", help="Name of the dataset to add", required=True
+    )
+    ds_add_parser.add_argument(
+        "--rev",
+        help="Git revision, e.g. SHA, branch, tag "
+        "(only applicable for dvc/git repository)",
+    )
+    ds_add_parser.add_argument(
+        "--path", help="Path to a file or directory within the git repository"
+    )
+    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/add"),
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+        help=dataset_update_help,
+    )
+    ds_update_parser.add_argument("name", help="Name of the dataset to update")
+    ds_update_parser.set_defaults(func=CmdDatasetUpdate)
diff --git a/dvc/database.py b/dvc/database.py
index 4289779b3..01e038e4e 100644
--- a/dvc/database.py
+++ b/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__)
diff --git a/dvc/dependency/dataset.py b/dvc/dependency/dataset.py
index 37dcea30e..e9f3128b6 100644
--- a/dvc/dependency/dataset.py
+++ b/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()
diff --git a/dvc/dependency/db.py b/dvc/dependency/db.py
index 2d663150c..6cdbd169e 100644
--- a/dvc/dependency/db.py
+++ b/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
diff --git a/dvc/dvcfile.py b/dvc/dvcfile.py
index 9837549ce..134378f3b 100644
--- a/dvc/dvcfile.py
+++ b/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)
 
diff --git a/dvc/output.py b/dvc/output.py
index e94dd00e8..92a982cc2 100644
--- a/dvc/output.py
+++ b/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)
         )
diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py
index c10965f43..3e14069d4 100644
--- a/dvc/repo/__init__.py
+++ b/dvc/repo/__init__.py
@@ -150,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
@@ -220,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]
diff --git a/dvc/repo/datasets.py b/dvc/repo/datasets.py
new file mode 100644
index 000000000..4188d9f5a
--- /dev/null
+++ b/dvc/repo/datasets.py
@@ -0,0 +1,385 @@
+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_data.hashfile.meta import Meta
+
+if TYPE_CHECKING:
+    from dql.dataset import DatasetRecord  # type: ignore[import]
+    from typing_extensions import Self
+
+    from dvc.repo import Repo
+
+
+logger = logger.getChild(__name__)
+
+
+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 name_version(self) -> tuple[str, Optional[int]]:
+        url = urlparse(self.spec.url)
+        parts = url.netloc.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":
+        if not record:
+            try:
+                from dvcx.catalog import get_catalog  # type: ignore[import]
+
+            except ImportError as exc:
+                raise DvcException("dvcx is not installed") from exc
+
+            name, _version = self.name_version
+            version = _version or version
+            catalog = get_catalog()
+            record = catalog.get_remote_dataset(name)
+
+        assert record is not None
+        ver = version or record.latest_version
+        assert ver
+        version_info = record.get_version(ver)
+        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]]:
+            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, url: str, name: str, manifest_path: str = "dvc.yaml", **kwargs: Any
+    ) -> Dataset:
+        kwargs.update({"url": url, "name": name})
+        dataset = self._build_dataset(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]
+        new = dataset.update(self.repo, **kwargs)
+
+        self.dump(new, old=dataset)
+        self[name] = new
+        return dataset, new
+
+    def _dump_spec(self, manifest_path: str, 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: str, 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)
diff --git a/dvc/repo/index.py b/dvc/repo/index.py
index ffa624df9..2f764a786 100644
--- a/dvc/repo/index.py
+++ b/dvc/repo/index.py
@@ -290,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 []
@@ -298,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
@@ -322,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):
@@ -331,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,
@@ -339,6 +343,7 @@ class Index:
             params=params,
             artifacts=artifacts,
             datasets=datasets,
+            datasets_lock=datasets_lock,
         )
 
     @classmethod
@@ -354,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":
diff --git a/dvc/schema.py b/dvc/schema.py
index 37710ad3b..85001016f 100644
--- a/dvc/schema.py
+++ b/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,


Test patch:
diff --git a/tests/unit/command/test_help.py b/tests/unit/command/test_help.py
index 0350c11e2..488cb63f4 100644
--- a/tests/unit/command/test_help.py
+++ b/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)