mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-09 23:10:24 +01:00
checkout: start switching to obj-based workflow (#6648)
* webhdfs: info/checksum: properly raise FileNotFoundError * checkout: start switching to obj-based workflow Related #6594
This commit is contained in:
Ruslan Kuprieiev
+106
-157
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from itertools import chain
|
||||
|
||||
from shortuuid import uuid
|
||||
|
||||
@@ -10,52 +11,19 @@ from dvc.exceptions import (
|
||||
DvcException,
|
||||
)
|
||||
from dvc.ignore import DvcIgnoreFilter
|
||||
from dvc.objects import check, load
|
||||
from dvc.objects.db.slow_link_detection import ( # type: ignore[attr-defined]
|
||||
slow_link_guard,
|
||||
)
|
||||
from dvc.objects.errors import ObjectFormatError
|
||||
from dvc.objects.diff import ROOT
|
||||
from dvc.objects.diff import diff as odiff
|
||||
from dvc.objects.stage import stage
|
||||
from dvc.objects.tree import Tree
|
||||
from dvc.types import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _changed(path_info, fs, obj, cache, state=None):
|
||||
logger.trace("checking if '%s'('%s') has changed.", path_info, obj)
|
||||
|
||||
try:
|
||||
check(cache, obj)
|
||||
except (FileNotFoundError, ObjectFormatError):
|
||||
logger.debug(
|
||||
"cache for '%s'('%s') has changed.", path_info, obj.hash_info
|
||||
)
|
||||
return True
|
||||
|
||||
actual = state.get(path_info, fs) if state else None
|
||||
if actual is None:
|
||||
try:
|
||||
actual = stage(
|
||||
cache, path_info, fs, obj.hash_info.name, dry_run=True
|
||||
)[1].hash_info
|
||||
except FileNotFoundError:
|
||||
logger.debug("'%s' doesn't exist.", path_info)
|
||||
return True
|
||||
|
||||
if obj.hash_info != actual:
|
||||
logger.debug(
|
||||
"hash value '%s' for '%s' has changed (actual '%s').",
|
||||
obj.hash_info,
|
||||
actual,
|
||||
path_info,
|
||||
)
|
||||
return True
|
||||
|
||||
logger.trace("'%s' hasn't changed.", path_info)
|
||||
return False
|
||||
|
||||
|
||||
def _remove(path_info, fs, cache, force=False):
|
||||
def _remove(path_info, fs, in_cache, force=False):
|
||||
if not fs.exists(path_info):
|
||||
return
|
||||
|
||||
@@ -63,13 +31,7 @@ def _remove(path_info, fs, cache, force=False):
|
||||
fs.remove(path_info)
|
||||
return
|
||||
|
||||
current = stage(cache, path_info, fs, fs.PARAM_CHECKSUM, dry_run=True)[
|
||||
1
|
||||
].hash_info
|
||||
try:
|
||||
obj = load(cache, current)
|
||||
check(cache, obj)
|
||||
except (FileNotFoundError, ObjectFormatError):
|
||||
if not in_cache:
|
||||
msg = (
|
||||
f"file/directory '{path_info}' is going to be removed. "
|
||||
"Are you sure you want to proceed?"
|
||||
@@ -98,7 +60,7 @@ def _verify_link(cache, path_info, link_type):
|
||||
|
||||
def _do_link(cache, from_info, to_info, link_method):
|
||||
if cache.fs.exists(to_info):
|
||||
raise DvcException(f"Link '{to_info}' already exists!")
|
||||
cache.fs.remove(to_info) # broken symlink
|
||||
|
||||
link_method(from_info, to_info)
|
||||
|
||||
@@ -130,7 +92,7 @@ def _link(cache, from_info, to_info):
|
||||
try:
|
||||
_try_links(cache, from_info, to_info, cache.cache_types)
|
||||
except FileNotFoundError as exc:
|
||||
raise CheckoutError([to_info]) from exc
|
||||
raise CheckoutError([str(to_info)]) from exc
|
||||
|
||||
|
||||
def _cache_is_copy(cache, path_info):
|
||||
@@ -160,7 +122,7 @@ def _cache_is_copy(cache, path_info):
|
||||
def _checkout_file(
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
change,
|
||||
cache,
|
||||
force,
|
||||
progress_callback=None,
|
||||
@@ -169,24 +131,24 @@ def _checkout_file(
|
||||
):
|
||||
"""The file is changed we need to checkout a new copy"""
|
||||
modified = False
|
||||
cache_info = cache.hash_to_path_info(obj.hash_info.value)
|
||||
if fs.exists(path_info):
|
||||
if not relink and _changed(path_info, fs, obj, cache, state=state):
|
||||
modified = True
|
||||
_remove(path_info, fs, cache, force=force)
|
||||
_link(cache, cache_info, path_info)
|
||||
else:
|
||||
cache_info = cache.hash_to_path_info(change.new.obj.hash_info.value)
|
||||
if change.old.obj:
|
||||
if relink:
|
||||
if fs.iscopy(path_info) and _cache_is_copy(cache, path_info):
|
||||
cache.unprotect(path_info)
|
||||
else:
|
||||
_remove(path_info, fs, cache, force=force)
|
||||
_remove(path_info, fs, change.old.in_cache, force=force)
|
||||
_link(cache, cache_info, path_info)
|
||||
else:
|
||||
modified = True
|
||||
_remove(path_info, fs, change.old.in_cache, force=force)
|
||||
_link(cache, cache_info, path_info)
|
||||
else:
|
||||
_link(cache, cache_info, path_info)
|
||||
modified = True
|
||||
|
||||
if state:
|
||||
state.save(path_info, fs, obj.hash_info)
|
||||
state.save(path_info, fs, change.new.obj.hash_info)
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(str(path_info))
|
||||
@@ -194,109 +156,83 @@ def _checkout_file(
|
||||
return modified
|
||||
|
||||
|
||||
def _remove_redundant_files(
|
||||
path_info, fs, obj, cache, force, dvcignore: Optional[DvcIgnoreFilter]
|
||||
):
|
||||
if dvcignore:
|
||||
existing_files = set(dvcignore.walk_files(fs, path_info))
|
||||
else:
|
||||
existing_files = set(fs.walk_files(path_info))
|
||||
|
||||
needed_files = {path_info.joinpath(*key) for key, _ in obj}
|
||||
redundant_files = existing_files - needed_files
|
||||
for path in redundant_files:
|
||||
_remove(path, fs, cache, force)
|
||||
|
||||
return bool(redundant_files)
|
||||
|
||||
|
||||
def _checkout_dir(
|
||||
def _diff(
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
cache,
|
||||
force,
|
||||
progress_callback=None,
|
||||
relink=False,
|
||||
dvcignore: Optional[DvcIgnoreFilter] = None,
|
||||
state=None,
|
||||
):
|
||||
modified = False
|
||||
# Create dir separately so that dir is created
|
||||
# even if there are no files in it
|
||||
if not fs.exists(path_info):
|
||||
modified = True
|
||||
fs.makedirs(path_info)
|
||||
|
||||
logger.debug("Linking directory '%s'.", path_info)
|
||||
|
||||
failed = []
|
||||
for entry_key, entry_obj in obj:
|
||||
try:
|
||||
entry_modified = _checkout_file(
|
||||
path_info.joinpath(*entry_key),
|
||||
fs,
|
||||
entry_obj,
|
||||
cache,
|
||||
force,
|
||||
progress_callback,
|
||||
relink,
|
||||
state=None,
|
||||
)
|
||||
if entry_modified:
|
||||
modified = True
|
||||
except CheckoutError as exc:
|
||||
failed.extend(exc.target_infos)
|
||||
|
||||
modified = (
|
||||
_remove_redundant_files(
|
||||
path_info, fs, obj, cache, force, dvcignore=dvcignore
|
||||
old = None
|
||||
try:
|
||||
_, old = stage(
|
||||
cache,
|
||||
path_info,
|
||||
fs,
|
||||
obj.hash_info.name if obj else cache.fs.PARAM_CHECKSUM,
|
||||
dry_run=True,
|
||||
dvcignore=dvcignore,
|
||||
)
|
||||
or modified
|
||||
)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
if failed:
|
||||
raise CheckoutError(failed)
|
||||
diff = odiff(old, obj, cache)
|
||||
|
||||
if state:
|
||||
state.save(path_info, fs, obj.hash_info)
|
||||
if relink:
|
||||
diff.modified.extend(diff.unchanged)
|
||||
|
||||
# relink is not modified, assume it as nochange
|
||||
return modified and not relink
|
||||
return diff
|
||||
|
||||
|
||||
def _checkout(
|
||||
diff,
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
cache,
|
||||
force=False,
|
||||
progress_callback=None,
|
||||
relink=False,
|
||||
dvcignore: Optional[DvcIgnoreFilter] = None,
|
||||
state=None,
|
||||
):
|
||||
if not obj.hash_info.isdir:
|
||||
ret = _checkout_file(
|
||||
path_info, fs, obj, cache, force, progress_callback, relink, state
|
||||
)
|
||||
else:
|
||||
ret = _checkout_dir(
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
cache,
|
||||
force,
|
||||
progress_callback,
|
||||
relink,
|
||||
dvcignore=dvcignore,
|
||||
state=state,
|
||||
)
|
||||
if not diff:
|
||||
return
|
||||
|
||||
if state:
|
||||
state.save_link(path_info, fs)
|
||||
for change in diff.deleted:
|
||||
entry_path = (
|
||||
path_info.joinpath(*change.old.key)
|
||||
if change.old.key != ROOT
|
||||
else path_info
|
||||
)
|
||||
_remove(entry_path, fs, change.old.in_cache, force=force)
|
||||
|
||||
return ret
|
||||
failed = []
|
||||
for change in chain(diff.added, diff.modified):
|
||||
entry_path = (
|
||||
path_info.joinpath(*change.new.key)
|
||||
if change.new.key != ROOT
|
||||
else path_info
|
||||
)
|
||||
if isinstance(change.new.obj, Tree):
|
||||
fs.makedirs(entry_path)
|
||||
continue
|
||||
|
||||
try:
|
||||
_checkout_file(
|
||||
entry_path,
|
||||
fs,
|
||||
change,
|
||||
cache,
|
||||
force,
|
||||
progress_callback,
|
||||
relink,
|
||||
state=state,
|
||||
)
|
||||
except CheckoutError as exc:
|
||||
failed.extend(exc.target_infos)
|
||||
|
||||
if failed:
|
||||
raise CheckoutError(failed)
|
||||
|
||||
|
||||
def checkout(
|
||||
@@ -311,41 +247,54 @@ def checkout(
|
||||
dvcignore: Optional[DvcIgnoreFilter] = None,
|
||||
state=None,
|
||||
):
|
||||
|
||||
if path_info.scheme not in ["local", cache.fs.scheme]:
|
||||
raise NotImplementedError
|
||||
|
||||
failed = None
|
||||
skip = False
|
||||
diff = _diff(
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
cache,
|
||||
relink=relink,
|
||||
dvcignore=dvcignore,
|
||||
)
|
||||
|
||||
failed = []
|
||||
if not obj:
|
||||
if not quiet:
|
||||
logger.warning(
|
||||
"No file hash info found for '%s'. It won't be created.",
|
||||
path_info,
|
||||
)
|
||||
_remove(path_info, fs, cache, force=force)
|
||||
failed = path_info
|
||||
|
||||
elif not relink and not _changed(path_info, fs, obj, cache, state=state):
|
||||
failed.append(str(path_info))
|
||||
elif not diff:
|
||||
logger.trace("Data '%s' didn't change.", path_info) # type: ignore
|
||||
skip = True
|
||||
|
||||
if failed or skip:
|
||||
try:
|
||||
_checkout(
|
||||
diff,
|
||||
path_info,
|
||||
fs,
|
||||
cache,
|
||||
force=force,
|
||||
progress_callback=progress_callback,
|
||||
relink=relink,
|
||||
state=state,
|
||||
)
|
||||
except CheckoutError as exc:
|
||||
failed.extend(exc.target_infos)
|
||||
|
||||
if diff and state:
|
||||
state.save_link(path_info, fs)
|
||||
if not failed:
|
||||
state.save(path_info, fs, obj.hash_info)
|
||||
|
||||
if failed or not diff:
|
||||
if progress_callback and obj:
|
||||
progress_callback(str(path_info), len(obj))
|
||||
if failed:
|
||||
raise CheckoutError([failed])
|
||||
raise CheckoutError(failed)
|
||||
return
|
||||
|
||||
logger.debug("Checking out '%s' with cache '%s'.", path_info, obj)
|
||||
|
||||
return _checkout(
|
||||
path_info,
|
||||
fs,
|
||||
obj,
|
||||
cache,
|
||||
force,
|
||||
progress_callback,
|
||||
relink,
|
||||
dvcignore=dvcignore,
|
||||
state=state,
|
||||
)
|
||||
return bool(diff) and not relink
|
||||
|
||||
+8
-2
@@ -1,3 +1,4 @@
|
||||
import errno
|
||||
import logging
|
||||
import os
|
||||
import posixpath
|
||||
@@ -131,14 +132,19 @@ class WebHDFSFileSystem(BaseFileSystem): # pylint:disable=abstract-method
|
||||
return status is not None
|
||||
|
||||
def info(self, path_info):
|
||||
st = self.hdfs_client.status(path_info.path)
|
||||
st = self.hdfs_client.status(path_info.path, strict=False)
|
||||
if not st:
|
||||
raise FileNotFoundError(
|
||||
errno.ENOENT, os.strerror(errno.ENOENT), path_info.path
|
||||
)
|
||||
return {"size": st["length"], "type": "file"}
|
||||
|
||||
def checksum(self, path_info):
|
||||
size = self.info(path_info)["size"]
|
||||
return HashInfo(
|
||||
"checksum",
|
||||
self.hdfs_client.checksum(path_info.path)["bytes"],
|
||||
size=self.hdfs_client.status(path_info.path)["length"],
|
||||
size=size,
|
||||
)
|
||||
|
||||
def copy(self, from_info, to_info, **_kwargs):
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING, List, Optional, Tuple
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .file import HashFile
|
||||
|
||||
ADD = "add"
|
||||
MODIFY = "modify"
|
||||
DELETE = "delete"
|
||||
UNCHANGED = "unchanged"
|
||||
|
||||
|
||||
@dataclass
|
||||
class TreeEntry:
|
||||
in_cache: bool
|
||||
key: Tuple[str]
|
||||
obj: Optional["HashFile"] = field(default=None)
|
||||
|
||||
def __bool__(self):
|
||||
return bool(self.obj)
|
||||
|
||||
def __eq__(self, other):
|
||||
if not isinstance(other, TreeEntry):
|
||||
return False
|
||||
|
||||
if self.key != other.key or bool(self.obj) != bool(other.obj):
|
||||
return False
|
||||
|
||||
if not self.obj:
|
||||
return False
|
||||
|
||||
return self.obj.hash_info == other.obj.hash_info
|
||||
|
||||
|
||||
@dataclass
|
||||
class Change:
|
||||
old: TreeEntry
|
||||
new: TreeEntry
|
||||
|
||||
@property
|
||||
def typ(self):
|
||||
if not self.old and not self.new:
|
||||
return UNCHANGED
|
||||
|
||||
if self.old and not self.new:
|
||||
return DELETE
|
||||
|
||||
if not self.old and self.new:
|
||||
return ADD
|
||||
|
||||
if self.old != self.new:
|
||||
return MODIFY
|
||||
|
||||
return UNCHANGED
|
||||
|
||||
def __bool__(self):
|
||||
return self.typ != UNCHANGED
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiffResult:
|
||||
added: List[Change] = field(default_factory=list, compare=True)
|
||||
modified: List[Change] = field(default_factory=list, compare=True)
|
||||
deleted: List[Change] = field(default_factory=list, compare=True)
|
||||
unchanged: List[Change] = field(default_factory=list, compare=True)
|
||||
|
||||
def __bool__(self):
|
||||
return bool(self.added or self.modified or self.deleted)
|
||||
|
||||
|
||||
ROOT = ("",)
|
||||
|
||||
|
||||
def diff(
|
||||
old: Optional["HashFile"], new: Optional["HashFile"], cache
|
||||
) -> DiffResult:
|
||||
from .tree import Tree
|
||||
|
||||
if old is None and new is None:
|
||||
return DiffResult()
|
||||
|
||||
def _get_keys(obj):
|
||||
if not obj:
|
||||
return []
|
||||
return [ROOT] + (
|
||||
[key for key, _ in obj] if isinstance(obj, Tree) else []
|
||||
)
|
||||
|
||||
old_keys = set(_get_keys(old))
|
||||
new_keys = set(_get_keys(new))
|
||||
|
||||
def _get_obj(obj, key):
|
||||
if not obj or key == ROOT:
|
||||
return obj
|
||||
|
||||
return obj.get(key)
|
||||
|
||||
def _in_cache(obj, cache):
|
||||
from . import check
|
||||
from .errors import ObjectFormatError
|
||||
|
||||
if not obj:
|
||||
return False
|
||||
|
||||
try:
|
||||
check(cache, obj)
|
||||
return True
|
||||
except (FileNotFoundError, ObjectFormatError):
|
||||
return False
|
||||
|
||||
ret = DiffResult()
|
||||
for key in old_keys | new_keys:
|
||||
old_obj = _get_obj(old, key)
|
||||
new_obj = _get_obj(new, key)
|
||||
|
||||
change = Change(
|
||||
old=TreeEntry(_in_cache(old_obj, cache), key, old_obj),
|
||||
new=TreeEntry(_in_cache(new_obj, cache), key, new_obj),
|
||||
)
|
||||
|
||||
if change.typ == ADD:
|
||||
ret.added.append(change)
|
||||
elif change.typ == MODIFY:
|
||||
ret.modified.append(change)
|
||||
elif change.typ == DELETE:
|
||||
ret.deleted.append(change)
|
||||
else:
|
||||
assert change.typ == UNCHANGED
|
||||
if not change.new.in_cache and not isinstance(
|
||||
change.new.obj, Tree
|
||||
):
|
||||
ret.modified.append(change)
|
||||
else:
|
||||
ret.unchanged.append(change)
|
||||
return ret
|
||||
@@ -496,27 +496,27 @@ def test_should_update_state_entry_for_directory_after_add(
|
||||
|
||||
ret = main(["add", "data"])
|
||||
assert ret == 0
|
||||
assert file_md5_counter.mock.call_count == 3
|
||||
assert file_md5_counter.mock.call_count == 4
|
||||
|
||||
ret = main(["status"])
|
||||
assert ret == 0
|
||||
assert file_md5_counter.mock.call_count == 3
|
||||
assert file_md5_counter.mock.call_count == 4
|
||||
|
||||
ls = "dir" if os.name == "nt" else "ls"
|
||||
ret = main(
|
||||
["run", "--single-stage", "-d", "data", "{} {}".format(ls, "data")]
|
||||
)
|
||||
assert ret == 0
|
||||
assert file_md5_counter.mock.call_count == 3
|
||||
assert file_md5_counter.mock.call_count == 4
|
||||
|
||||
os.rename("data", "data" + ".back")
|
||||
ret = main(["checkout"])
|
||||
assert ret == 0
|
||||
assert file_md5_counter.mock.call_count == 3
|
||||
assert file_md5_counter.mock.call_count == 4
|
||||
|
||||
ret = main(["status"])
|
||||
assert ret == 0
|
||||
assert file_md5_counter.mock.call_count == 3
|
||||
assert file_md5_counter.mock.call_count == 4
|
||||
|
||||
|
||||
class TestAddCommit(TestDvc):
|
||||
@@ -537,13 +537,15 @@ def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
|
||||
counter = mocker.spy(dvc_module.objects.stage, "_get_tree_obj")
|
||||
ret = main(["add", "data"])
|
||||
assert ret == 0
|
||||
assert counter.mock.call_count == 2
|
||||
|
||||
ret = main(["status"])
|
||||
assert ret == 0
|
||||
assert counter.mock.call_count == 2
|
||||
|
||||
ret = main(["status"])
|
||||
assert ret == 0
|
||||
assert counter.mock.call_count == 1
|
||||
assert counter.mock.call_count == 2
|
||||
|
||||
|
||||
class TestShouldPlaceStageInDataDirIfRepositoryBelowSymlink(TestDvc):
|
||||
|
||||
Reference in New Issue
Block a user