From 6a7b55fdf7d09fe45b079bdda1022effbb7001f7 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 31 Jul 2022 19:22:57 +0300 Subject: [PATCH] datafs: use dvc-data's datafs --- dvc/fs/data.py | 181 ++----------------------------------- dvc/fs/dvc.py | 23 ++--- dvc/repo/__init__.py | 11 ++- dvc/repo/diff.py | 2 +- dvc/repo/get.py | 4 +- dvc/repo/index.py | 40 +++++--- dvc/repo/ls.py | 2 +- setup.cfg | 2 +- tests/unit/fs/test_data.py | 53 ++++++----- tests/unit/fs/test_dvc.py | 1 + 10 files changed, 87 insertions(+), 232 deletions(-) diff --git a/dvc/fs/data.py b/dvc/fs/data.py index 9d4ab8710..c3612aed2 100644 --- a/dvc/fs/data.py +++ b/dvc/fs/data.py @@ -1,183 +1,16 @@ import logging +import os import threading -import typing -from fsspec import AbstractFileSystem from funcy import cached_property, wrap_prop +from dvc.utils import as_posix +from dvc_data.fs import DataFileSystem as _DataFileSystem from dvc_objects.fs.base import FileSystem -from dvc_objects.fs.callbacks import DEFAULT_CALLBACK - -if typing.TYPE_CHECKING: - from dvc.types import AnyPath logger = logging.getLogger(__name__) -class _DataFileSystem(AbstractFileSystem): # pylint:disable=abstract-method - """DVC repo fs. - - Args: - repo: DVC repo. - """ - - root_marker = "/" - - def __init__(self, workspace=None, **kwargs): - super().__init__(**kwargs) - self.repo = kwargs["repo"] - self.workspace = workspace or "repo" - - @cached_property - def path(self): - from . import Path - - def _getcwd(): - return self.root_marker - - return Path(self.sep, getcwd=_getcwd) - - @property - def config(self): - raise NotImplementedError - - def _get_key(self, path): - if self.workspace != "repo": - from . import get_cloud_fs - - cls, kwargs, fs_path = get_cloud_fs(None, url=path) - fs = cls(**kwargs) - return (self.workspace, *fs.path.parts(fs_path)) - - path = self.path.abspath(path) - if path == self.root_marker: - return (self.workspace,) - - key = self.path.relparts(path, self.root_marker) - if key == (".") or key == (""): - key = () - - return (self.workspace, *key) - - def _get_fs_path(self, path: "AnyPath", remote=None): - from dvc.config import NoRemoteError - - info = self.info(path) - if info["type"] == "directory": - raise IsADirectoryError - - value = info.get("md5") - if not value: - raise FileNotFoundError - - cache_path = self.repo.odb.local.oid_to_path(value) - - if self.repo.odb.local.fs.exists(cache_path): - return self.repo.odb.local.fs, cache_path - - try: - remote_odb = self.repo.cloud.get_remote_odb(remote) - except NoRemoteError as exc: - raise FileNotFoundError from exc - remote_fs_path = remote_odb.oid_to_path(value) - return remote_odb.fs, remote_fs_path - - def open( # type: ignore - self, path: str, mode="r", encoding=None, **kwargs - ): # pylint: disable=arguments-renamed, arguments-differ - fs, fspath = self._get_fs_path(path, **kwargs) - return fs.open(fspath, mode=mode, encoding=encoding) - - def ls(self, path, detail=True, **kwargs): - info = self.info(path) - if info["type"] != "directory": - return [info] if detail else [path] - - root_key = self._get_key(path) - try: - entries = [ - self.sep.join((path, name)) if path else name - for name in self.repo.index.tree.ls(prefix=root_key) - ] - except KeyError as exc: - raise FileNotFoundError from exc - - if not detail: - return entries - - return [self.info(epath) for epath in entries] - - def isdvc(self, path, recursive=False, strict=True): - try: - info = self.info(path) - except FileNotFoundError: - return False - - recurse = recursive or not strict - return bool(info.get("outs") if recurse else info.get("isout")) - - def info(self, path, **kwargs): - from dvc_data.hashfile.meta import Meta - - key = self._get_key(path) - - try: - outs = list(self.repo.index.tree.iteritems(key)) # noqa: B301 - except KeyError as exc: - raise FileNotFoundError from exc - - ret = { - "type": "file", - "size": 0, - "isexec": False, - "isdvc": False, - "outs": outs, - "name": path, - } - - if len(outs) > 1 and outs[0][0] != key: - shortest = self.repo.index.tree.shortest_prefix(key) - if shortest: - assert shortest[1][1].isdir - if len(shortest[0]) <= len(key): - ret["isdvc"] = True - - ret["type"] = "directory" - return ret - - item_key, (meta, hash_info) = outs[0] - - meta = meta or Meta() - - if key != item_key: - assert item_key[: len(key)] == key - ret["type"] = "directory" - return ret - - ret["size"] = meta.size - ret["isexec"] = meta.isexec - ret[hash_info.name] = hash_info.value - ret["isdvc"] = True - ret["isout"] = True - ret["meta"] = meta - if hash_info and hash_info.isdir: - ret["type"] = "directory" - return ret - - def get_file( # pylint: disable=arguments-differ - self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs - ): - fs, path = self._get_fs_path(rpath) - fs.get_file(path, lpath, callback=callback, **kwargs) - - def checksum(self, path): - info = self.info(path) - md5 = info.get("md5") - if md5: - return md5 - raise NotImplementedError - - class DataFileSystem(FileSystem): protocol = "local" @@ -194,6 +27,8 @@ class DataFileSystem(FileSystem): def isdvc(self, path, **kwargs): return self.fs.isdvc(path, **kwargs) - @property - def repo(self): - return self.fs.repo + def from_os_path(self, path): + if os.path.isabs(path): + path = os.path.splitdrive(path)[1] + + return as_posix(path) diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py index 989a2b734..715447cdf 100644 --- a/dvc/fs/dvc.py +++ b/dvc/fs/dvc.py @@ -135,7 +135,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method """Keep a datafs instance of each repo.""" if hasattr(repo, "dvc_dir"): - self._datafss[key] = DataFileSystem(repo=repo) + self._datafss[key] = DataFileSystem(index=repo.index.data["repo"]) def _get_key(self, path): parts = self.repo.fs.path.relparts(path, self.repo.root_dir) @@ -252,7 +252,9 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method scm=self.repo.scm, repo_factory=self.repo_factory, ) - self._datafss[key] = DataFileSystem(repo=repo) + self._datafss[key] = DataFileSystem( + index=repo.index.data["repo"] + ) self._subrepos_trie[key] = repo def _is_dvc_repo(self, dir_path): @@ -268,6 +270,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method def _get_fs_pair( self, path ) -> Tuple[ + Optional["Repo"], Optional[FileSystem], Optional[str], Optional[DataFileSystem], @@ -299,7 +302,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method else: dvc_path = None - return fs, fs_path, dvc_fs, dvc_path + return repo, fs, fs_path, dvc_fs, dvc_path def open( self, path, mode="r", encoding="utf-8", **kwargs @@ -307,7 +310,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method if "b" in mode: encoding = None - fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) + _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) try: return fs.open(fs_path, mode=mode, encoding=encoding) except FileNotFoundError: @@ -317,15 +320,14 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs) def isdvc(self, path, **kwargs): - _, _, dvc_fs, dvc_path = self._get_fs_pair(path) + _, _, _, dvc_fs, dvc_path = self._get_fs_pair(path) return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs) def ls( # pylint: disable=arguments-differ self, path, detail=True, dvc_only=False, **kwargs ): - fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) + repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) - repo = dvc_fs.repo if dvc_fs else self.repo dvcignore = repo.dvcignore ignore_subrepos = kwargs.get("ignore_subrepos", True) @@ -378,7 +380,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method def get_file( # pylint: disable=arguments-differ self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs ): - fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath) + _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath) if fs: try: @@ -390,9 +392,8 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method dvc_fs.get_file(dvc_path, lpath, callback=callback, **kwargs) def info(self, path, **kwargs): - fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) + repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) - repo = dvc_fs.repo if dvc_fs else self.repo dvcignore = repo.dvcignore ignore_subrepos = kwargs.get("ignore_subrepos", True) @@ -434,7 +435,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method return info def checksum(self, path): - fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) + _, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path) try: return fs.checksum(fs_path) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 304d8f8c3..2fb8cf035 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -480,7 +480,7 @@ class Repo: def datafs(self): from dvc.fs.data import DataFileSystem - return DataFileSystem(repo=self) + return DataFileSystem(index=self.index.data["repo"]) @cached_property def dvcfs(self): @@ -501,18 +501,23 @@ class Repo: from dvc.fs.dvc import DvcFileSystem if os.path.isabs(path): - fs = DataFileSystem(repo=self, workspace="local") + fs = DataFileSystem(index=self.index.data["local"]) fs_path = path else: fs = DvcFileSystem(repo=self, subrepos=True) fs_path = fs.from_os_path(path) try: + if remote: + remote_odb = self.cloud.get_remote_odb(name=remote) + oid = fs.info(fs_path)["dvc_info"]["md5"] + fs = remote_odb.fs + fs_path = remote_odb.oid_to_path(oid) + with fs.open( fs_path, mode=mode, encoding=encoding, - remote=remote, ) as fobj: yield fobj except FileNotFoundError as exc: diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py index 540934b18..d2e67821b 100644 --- a/dvc/repo/diff.py +++ b/dvc/repo/diff.py @@ -190,7 +190,7 @@ def _filter_missing(dvcfs, paths): if ( dvc_info and info["type"] == "directory" - and not dvc_info["meta"].obj + and not dvc_info["obj"] ): yield path except FileNotFoundError: diff --git a/dvc/repo/get.py b/dvc/repo/get.py index 5f4029a2c..0f473c019 100644 --- a/dvc/repo/get.py +++ b/dvc/repo/get.py @@ -54,8 +54,8 @@ def get(url, path, out=None, rev=None, jobs=None): if os.path.isabs(path): from dvc.fs.data import DataFileSystem - fs = DataFileSystem(repo=repo, workspace="local") - fs_path = path + fs = DataFileSystem(index=repo.index.data["local"]) + fs_path = fs.from_os_path(path) else: fs = repo.dvcfs fs_path = fs.from_os_path(path) diff --git a/dvc/repo/index.py b/dvc/repo/index.py index 7d31b6bb3..9e3fa1a05 100644 --- a/dvc/repo/index.py +++ b/dvc/repo/index.py @@ -25,8 +25,8 @@ if TYPE_CHECKING: from dvc.repo.stage import StageLoad from dvc.stage import Stage from dvc.types import StrPath, TargetType - from dvc_data import Tree from dvc_data.hashfile.hash_info import HashInfo + from dvc_data.index import DataIndex from dvc_objects.db import ObjectDB @@ -168,33 +168,47 @@ class Index: return build_outs_graph(self.graph, self.outs_trie) @cached_property - def tree(self) -> "Tree": - from dvc.config import NoRemoteError - from dvc_data import Tree + def data(self) -> "Dict[str, DataIndex]": + from collections import defaultdict - tree = Tree() + from dvc.config import NoRemoteError + from dvc_data.index import DataIndex, DataIndexEntry + + by_workspace: dict = defaultdict(DataIndex) + + by_workspace["repo"] = DataIndex() + by_workspace["local"] = DataIndex() for out in self.outs: if not out.use_cache: continue if out.is_in_repo: - fs_key = "repo" + workspace = "repo" key = self.repo.fs.path.relparts( out.fs_path, self.repo.root_dir ) else: - fs_key = out.fs.protocol - key = out.fs.path.parts(out.fs_path) + workspace = out.fs.protocol + no_drive = out.fs.path.flavour.splitdrive(out.fs_path)[1] + key = out.fs.path.parts(no_drive)[1:] - out.meta.odb = out.odb try: - out.meta.remote = self.repo.cloud.get_remote_odb(out.remote) + remote = self.repo.cloud.get_remote_odb(out.remote) except NoRemoteError: - out.meta.remote = None - tree.add((fs_key,) + key, out.meta, out.hash_info) + remote = None - return tree + data_index = by_workspace[workspace] + + data_index[key] = DataIndexEntry( + meta=out.meta, + obj=out.obj, + hash_info=out.hash_info, + odb=out.odb, + remote=remote, + ) + + return dict(by_workspace) def used_objs( self, diff --git a/dvc/repo/ls.py b/dvc/repo/ls.py index 728c3e590..ae55c5401 100644 --- a/dvc/repo/ls.py +++ b/dvc/repo/ls.py @@ -85,7 +85,7 @@ def _ls( ret = {} for name, info in infos.items(): dvc_info = info.get("dvc_info", {}) - if dvc_info.get("outs") or not dvc_only: + if dvc_info.get("entries") or not dvc_only: ret[name] = { "isout": dvc_info.get("isout", False), "isdir": info["type"] == "directory", diff --git a/setup.cfg b/setup.cfg index 2889a151c..bf55af6ef 100644 --- a/setup.cfg +++ b/setup.cfg @@ -69,7 +69,7 @@ install_requires = dvc-render==0.0.9 dvc-task==0.1.0 dvclive>=0.10.0 - dvc-data==0.1.6 + dvc-data==0.1.8 [options.extras_require] all = diff --git a/tests/unit/fs/test_data.py b/tests/unit/fs/test_data.py index 77f2ad309..e6dc99b7e 100644 --- a/tests/unit/fs/test_data.py +++ b/tests/unit/fs/test_data.py @@ -4,7 +4,6 @@ import shutil import pytest import dvc_data -from dvc.config import NoRemoteError from dvc.fs.data import DataFileSystem from dvc.utils.fs import remove from dvc_data.build import build @@ -14,15 +13,15 @@ from dvc_data.hashfile.hash_info import HashInfo @pytest.mark.parametrize( "path, key", [ - ("", ("repo",)), - (".", ("repo",)), - ("/", ("repo",)), - ("foo", ("repo", "foo")), - ("dir/foo", ("repo", "dir", "foo")), + ("", ()), + (".", ()), + ("/", ()), + ("foo", ("foo",)), + ("dir/foo", ("dir", "foo")), ], ) def test_get_key(tmp_dir, dvc, path, key): - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.fs._get_key(path) == key @@ -31,7 +30,7 @@ def test_exists(tmp_dir, dvc): dvc.add("foo") (tmp_dir / "foo").unlink() - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.exists("foo") @@ -40,7 +39,7 @@ def test_open(tmp_dir, dvc): dvc.add("foo") (tmp_dir / "foo").unlink() - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) with fs.open("foo", "r") as fobj: assert fobj.read() == "foo" @@ -49,7 +48,7 @@ def test_open_dirty_hash(tmp_dir, dvc): tmp_dir.dvc_gen("file", "file") (tmp_dir / "file").write_text("something") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) with fs.open("file", "r") as fobj: # NOTE: Unlike DvcFileSystem, DataFileSystem should not # be affected by a dirty workspace. @@ -61,18 +60,17 @@ def test_open_no_remote(tmp_dir, dvc): (tmp_dir / "file").unlink() remove(dvc.odb.local.cache_dir) - fs = DataFileSystem(repo=dvc) - with pytest.raises(FileNotFoundError) as exc_info: + fs = DataFileSystem(index=dvc.index.data["repo"]) + with pytest.raises(FileNotFoundError): with fs.open("file", "r"): pass - assert isinstance(exc_info.value.__cause__, NoRemoteError) def test_open_dirty_no_hash(tmp_dir, dvc): tmp_dir.gen("file", "file") (tmp_dir / "file.dvc").write_text("outs:\n- path: file\n") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) # NOTE: Unlike DvcFileSystem, DataFileSystem should not # be affected by a dirty workspace. with pytest.raises(FileNotFoundError): @@ -95,7 +93,7 @@ def test_open_in_history(tmp_dir, scm, dvc): if rev == "workspace": continue - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) with fs.open("foo", "r") as fobj: assert fobj.read() == "foo" @@ -103,7 +101,7 @@ def test_open_in_history(tmp_dir, scm, dvc): def test_isdir_isfile(tmp_dir, dvc): tmp_dir.gen({"datafile": "data", "datadir": {"foo": "foo", "bar": "bar"}}) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert not fs.isdir("datadir") assert not fs.isfile("datadir") assert not fs.isdir("datafile") @@ -113,6 +111,7 @@ def test_isdir_isfile(tmp_dir, dvc): shutil.rmtree(tmp_dir / "datadir") (tmp_dir / "datafile").unlink() + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.isdir("datadir") assert not fs.isfile("datadir") assert not fs.isdir("datafile") @@ -124,7 +123,7 @@ def test_isdir_mixed(tmp_dir, dvc): dvc.add(str(tmp_dir / "dir" / "foo")) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.isdir("dir") assert not fs.isfile("dir") @@ -142,7 +141,7 @@ def test_walk(tmp_dir, dvc): ) dvc.add("dir", recursive=True) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) expected = [ "dir/subdir1", @@ -176,7 +175,7 @@ def test_walk_dir(tmp_dir, dvc): ) dvc.add("dir") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) expected = [ "dir/subdir1", @@ -198,7 +197,7 @@ def test_walk_dir(tmp_dir, dvc): def test_walk_missing(tmp_dir, dvc): - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) for _ in fs.walk("dir"): pass @@ -206,7 +205,7 @@ def test_walk_missing(tmp_dir, dvc): def test_walk_not_a_dir(tmp_dir, dvc): tmp_dir.dvc_gen("foo", "foo") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) for _ in fs.walk("foo"): pass @@ -215,14 +214,14 @@ def test_walk_not_a_dir(tmp_dir, dvc): def test_isdvc(tmp_dir, dvc): tmp_dir.gen({"foo": "foo", "bar": "bar"}) dvc.add("foo") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.isdvc("foo") assert not fs.isdvc("bar") def test_get_hash_file(tmp_dir, dvc): tmp_dir.dvc_gen({"foo": "foo"}) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) assert fs.info("foo")["md5"] == "acbd18db4cc2f85cedef654fccc4a4d8" @@ -230,7 +229,7 @@ def test_get_hash_dir(tmp_dir, dvc, mocker): tmp_dir.dvc_gen( {"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}} ) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) hash_file_spy = mocker.spy(dvc_data.hashfile.hash, "hash_file") assert fs.info("dir")["md5"] == "8761c4e9acad696bee718615e23e22db.dir" assert not hash_file_spy.called @@ -240,7 +239,7 @@ def test_get_hash_granular(tmp_dir, dvc): tmp_dir.dvc_gen( {"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}} ) - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) subdir = "dir/subdir" assert fs.info(subdir).get("md5") is None _, _, obj = build(dvc.odb.local, subdir, fs, "md5", dry_run=True) @@ -257,7 +256,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc): tmp_dir.dvc_gen("file", "file") (tmp_dir / "file").write_text("something") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) expected = "8c7dd922ad47494fc02c388e12c00eac" assert fs.info("file").get("md5") == expected _, _, obj = build(dvc.odb.local, "file", fs, "md5", dry_run=True) @@ -268,7 +267,7 @@ def test_get_hash_dirty_dir(tmp_dir, dvc): tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}}) (tmp_dir / "dir" / "baz").write_text("baz") - fs = DataFileSystem(repo=dvc) + fs = DataFileSystem(index=dvc.index.data["repo"]) expected = "5ea40360f5b4ec688df672a4db9c17d1.dir" assert fs.info("dir").get("md5") == expected _, _, obj = build(dvc.odb.local, "dir", fs, "md5", dry_run=True) diff --git a/tests/unit/fs/test_dvc.py b/tests/unit/fs/test_dvc.py index 60fd96969..17c8c9d58 100644 --- a/tests/unit/fs/test_dvc.py +++ b/tests/unit/fs/test_dvc.py @@ -107,6 +107,7 @@ def test_isdir_isfile(tmp_dir, dvc): (tmp_dir / "datafile").unlink() (tmp_dir / "subdir" / "baz").unlink() + fs = DvcFileSystem(repo=dvc) assert fs.isdir("datadir") assert not fs.isfile("datadir") assert fs.isdvc("datadir")