From 6a6bc7035ab7c2525c024d536b5e4686b2fb40bf Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 23 Jan 2023 21:29:32 +0200 Subject: [PATCH] data: status: use data index (#8846) Pre-requisite for #8761 --- dvc/repo/data.py | 252 ++++++++++++++++------------------------------ dvc/repo/index.py | 39 +++++-- 2 files changed, 116 insertions(+), 175 deletions(-) diff --git a/dvc/repo/data.py b/dvc/repo/data.py index 4157b0a18..7b035eb23 100644 --- a/dvc/repo/data.py +++ b/dvc/repo/data.py @@ -1,6 +1,5 @@ import os import posixpath -from collections import defaultdict from itertools import chain from typing import ( TYPE_CHECKING, @@ -10,124 +9,98 @@ from typing import ( Iterator, List, Optional, - Tuple, TypedDict, Union, ) -from dvc.fs.git import GitFileSystem -from dvc.ui import ui - if TYPE_CHECKING: from dvc.repo import Repo from dvc.scm import Git, NoSCM - from dvc_data.hashfile.db import HashFileDB - from dvc_data.hashfile.hash_info import HashInfo - from dvc_data.hashfile.obj import HashFile + from dvc_data.index import DataIndex def posixpath_to_os_path(path: str) -> str: return path.replace(posixpath.sep, os.path.sep) -def _in_cache(obj: "HashInfo", cache: "HashFileDB") -> bool: - from dvc_objects.errors import ObjectFormatError +def _adapt_typ(typ): + from dvc_data.index.diff import ADD, DELETE, MODIFY - assert obj.value - try: - cache.check(obj.value) - return True - except (FileNotFoundError, ObjectFormatError): - return False + if typ == MODIFY: + return "modified" + + if typ == ADD: + return "added" + + if typ == DELETE: + return "deleted" + + return typ -def _shallow_diff( - root: str, - old: Optional["HashInfo"], - new: Optional["HashInfo"], - cache: "HashFileDB", -) -> Dict[str, List[str]]: - d = {} - - old_root = new_root = root - if old and old.isdir: - old_root = os.path.sep.join([root, ""]) - if new and new.isdir: - new_root = os.path.sep.join([root, ""]) - - if old and not _in_cache(old, cache): - d["not_in_cache"] = [old_root] - - if not old and not new: - return d - if not new: - return {"deleted": [old_root], **d} - if not old: - return {"added": [new_root], **d} - if old != new: - return {"modified": [new_root], **d} - return {"unchanged": [new_root], **d} - - -def _granular_diff( - root: str, - old_obj: Optional["HashFile"], - new_obj: Optional["HashFile"], - cache: "HashFileDB", -) -> Dict[str, List[str]]: - from dvc_data.hashfile.diff import ROOT - from dvc_data.hashfile.diff import diff as odiff - - def path_join(root: str, *paths: str, isdir: bool = False) -> str: - if not isdir and paths == ROOT: - return root - return os.path.sep.join([root, *paths]) - - diff_data = odiff(old_obj, new_obj, cache) - - output: Dict[str, List[str]] = defaultdict(list) - for state in ("added", "deleted", "modified", "unchanged"): - items = getattr(diff_data, state) - for item in items: # pylint: disable=not-an-iterable - entry = item.old if state == "deleted" else item.new - isdir = entry.oid.isdir if entry.oid else False - - path = path_join(root, *entry.key, isdir=isdir) - output[state].append(path) - if not item.old.in_cache and state != "added": - output["not_in_cache"].append(path) - return dict(output) - - -def _get_obj_items(root: str, obj: Optional["HashFile"]) -> List[str]: - if not obj: - return [] - - from dvc_data.hashfile.tree import Tree - - sep = os.path.sep - if isinstance(obj, Tree): - return [sep.join([root, *key]) for key, _, _ in obj] - return [root] +def _adapt_path(change): + isdir = False + if change.new and change.new.meta: + isdir = change.new.meta.isdir + elif change.old and change.old.meta: + isdir = change.old.meta.isdir + key = change.key + if isdir: + key = (*key, "") + return os.path.sep.join(key) def _diff( - root: str, - old_oid: Optional["HashInfo"], - old_obj: Optional["HashFile"], - new_oid: Optional["HashInfo"], - new_obj: Optional["HashFile"], - odb: "HashFileDB", - granular: bool = False, + old: "DataIndex", + new: "DataIndex", + *, + granular: Optional[bool] = False, + with_missing: Optional[bool] = False, ) -> Dict[str, List[str]]: - if not granular: - return _shallow_diff(root, old_oid, new_oid, odb) - if (old_oid and not old_obj) or (new_oid and not new_obj): - # we don't have enough information to give full details - unknown = _get_obj_items(root, new_obj) - shallow_diff = _shallow_diff(root, old_oid, new_oid, odb) - return {**shallow_diff, "unknown": unknown} - return _granular_diff(root, old_obj, new_obj, odb) + from dvc_data.index.diff import UNCHANGED, UNKNOWN, diff + + ret: Dict[str, List[str]] = {} + + def _add_change(typ, change): + typ = _adapt_typ(typ) + if typ not in ret: + ret[typ] = [] + + ret[typ].append(_adapt_path(change)) + + for change in diff( + old, + new, + with_unchanged=True, + shallow=not granular, + hash_only=True, + with_unknown=True, + ): + if ( + change.typ == UNCHANGED + and change.old + and change.new + and not change.old.hash_info + and not change.new.hash_info + ): + # NOTE: emulating previous behaviour + continue + + if change.typ == UNKNOWN and not change.new: + # NOTE: emulating previous behaviour + continue + + if ( + with_missing + and change.old + and not change.old.odb.exists(change.old.hash_info.value) + ): + # NOTE: emulating previous behaviour + _add_change("not_in_cache", change) + + _add_change(change.typ, change) + + return ret class GitInfo(TypedDict, total=False): @@ -169,81 +142,32 @@ def _git_info( def _diff_index_to_wtree(repo: "Repo", **kwargs: Any) -> Dict[str, List[str]]: - from dvc_data.hashfile.build import build + from .index import build_data_index - unstaged_diff = defaultdict(list) - for out in repo.index.outs: - if not out.use_cache: - continue + workspace = build_data_index( + repo.index, repo.root_dir, repo.fs, compute_hash=True + ) - try: - _, _, new = build( - out.odb, - out.fs_path, - out.fs, - out.fs.PARAM_CHECKSUM, - ignore=out.dvcignore, - dry_run=True, - ) - except FileNotFoundError: - new = None - - cache = repo.odb.repo - root = str(out) - old = out.get_obj() - - with ui.status(f"Calculating diff for {root} between index/workspace"): - d = _diff( - root, - out.hash_info, - old, - new.hash_info if new else None, - new, - cache, - **kwargs, - ) - - for state, items in d.items(): - if not items: - continue - unstaged_diff[state].extend(items) - return unstaged_diff + return _diff( + repo.index.data["repo"], + workspace, + with_missing=True, + **kwargs, + ) def _diff_head_to_index( repo: "Repo", head: str = "HEAD", **kwargs: Any ) -> Dict[str, List[str]]: - # we need to store objects from index and the HEAD to diff later - objs: Dict[str, Dict[str, Tuple[Optional["HashFile"], "HashInfo"]]] - objs = defaultdict(dict) + index = repo.index.data["repo"] - staged_diff = defaultdict(list) for rev in repo.brancher(revs=[head]): - for out in repo.index.outs: - if not out.use_cache: - continue + if rev == "workspace": + continue - root = str(out) - if isinstance(out.fs, GitFileSystem): - root = posixpath_to_os_path(root) - typ = "index" if rev == "workspace" else head - objs[root][typ] = (out.get_obj(), out.hash_info) + head_index = repo.index.data["repo"] - cache = repo.odb.repo - for root, obj_d in objs.items(): - old_obj, old_oid = obj_d.get(head, (None, None)) - new_obj, new_oid = obj_d.get("index", (None, None)) - with ui.status(f"Calculating diff for {root} between head/index"): - d = _diff( - root, old_oid, old_obj, new_oid, new_obj, cache, **kwargs - ) - - for state, items in d.items(): - if not items: - continue - staged_diff[state].extend(items) - - return staged_diff + return _diff(head_index, index, **kwargs) class Status(TypedDict): @@ -289,8 +213,6 @@ def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status: except (SCMError, NoSCMError): committed_diff = {} else: - # we don't care about not-in-cache between the head and the index. - committed_diff.pop("not_in_cache", None) unchanged &= set(committed_diff.pop("unchanged", [])) git_info = _git_info(repo.scm, untracked_files=untracked_files) diff --git a/dvc/repo/index.py b/dvc/repo/index.py index 2e1e6b99a..776798671 100644 --- a/dvc/repo/index.py +++ b/dvc/repo/index.py @@ -395,6 +395,10 @@ class IndexView: self.stages = list({stage for stage, _ in stage_infos}) self._outs_filter = outs_filter + @property + def repo(self) -> "Repo": + return self._index.repo + @property def deps(self) -> Iterator["Dependency"]: for stage in self.stages: @@ -459,9 +463,11 @@ def build_data_index( path: str, fs: "FileSystem", workspace: Optional[str] = "repo", + compute_hash: Optional[bool] = False, ) -> "DataIndex": from dvc_data.index import DataIndex, DataIndexEntry from dvc_data.index.build import build_entries, build_entry + from dvc_data.index.save import build_tree data = DataIndex() for out in index.outs: @@ -476,26 +482,39 @@ def build_data_index( out_path = fs.path.join(path, *parts) try: - entry = build_entry(out_path, fs) + out_entry = build_entry( + out_path, + fs, + compute_hash=compute_hash, + state=index.repo.state, + ) except FileNotFoundError: - entry = DataIndexEntry(path=out_path, fs=fs) + out_entry = DataIndexEntry(path=out_path, fs=fs) - entry.key = key + out_entry.key = key - if not entry.meta or not entry.meta.isdir: - data.add(entry) + if not out_entry.meta or not out_entry.meta.isdir: + data.add(out_entry) continue - entry.loaded = True - data.add(entry) - - for entry in build_entries(out_path, fs): + for entry in build_entries( + out_path, fs, compute_hash=compute_hash, state=index.repo.state + ): if not entry.key or entry.key == ("",): # NOTE: whether the root will be returned by build_entries # depends on the filesystem (e.g. local doesn't, but s3 does). - entry.key = key + continue else: entry.key = key + entry.key + data.add(entry) + if compute_hash: + tree_meta, tree = build_tree(data, key) + out_entry.meta = tree_meta + out_entry.hash_info = tree.hash_info + + out_entry.loaded = True + data.add(out_entry) + return data