dvc.api for dataset (#10316)

* add support for tracking remote dataset

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://[email protected]/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 freezes
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 frozen 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 frozen 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: [email protected]: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: [email protected]: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
```

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'
```

* remodeling

* remove non-cloud-versioned remotes

* improve deserializing; handle invalidation

* dvc.api for dataset

* expose the API in the top-level

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)