scm: use dulwich backend when fetching exps during clone/pull (#9023)

refspecs uses pygit2 backend by default, which does not support git credential
helpers. This PR forces to use dulwich backend during clone/pull operations.

dvc/commands/plots.py

@@ -150,7 +150,7 @@ class CmdPlots(CmdBase):
                 render_html(
                     renderers=renderers,
                     output_file=output_file,
-                    template_path=self._html_template_path(),
+                    html_template=self._html_template_path(),
                 )
 
                 ui.write(output_file.as_uri())

dvc/commands/update.py

@@ -31,8 +31,7 @@ class CmdUpdate(CmdBase):
 def add_parser(subparsers, parent_parser):
     UPDATE_HELP = (
         "Update data artifact imported (via dvc import or dvc import-url) "
-        "from an external DVC repository or URL, or update data from a "
-        "DVC worktree remote."
+        "from an external DVC repository or URL."
     )
     update_parser = subparsers.add_parser(
         "update",

dvc/config_schema.py

@@ -51,6 +51,8 @@ def ByUrl(mapping):
         # Windows absolute paths should really have scheme == "" (local)
         if os.name == "nt" and len(parsed.scheme) == 1 and parsed.netloc == "":
             return schemas[""](data)
+        if parsed.netloc == "":
+            return schemas[""](data)
         if parsed.scheme not in schemas:
             raise Invalid(f"Unsupported URL type {parsed.scheme}://")
 
@@ -262,6 +264,7 @@ SCHEMA = {
     # section for experimental features
     "feature": {
         Optional("machine", default=False): Bool,
+        Optional("data_index_cache", default=False): Bool,
         # enabled by default. It's of no use, kept for backward compatibility.
         Optional("parametrization", default=True): Bool,
     },

dvc/external_repo.py

@@ -253,7 +253,7 @@ def _pull(git: "Git", unshallow: bool = False):
 
     git.fetch(unshallow=unshallow)
     _merge_upstream(git)
-    fetch_all_exps(git, "origin")
+    fetch_all_exps(git, "origin", backends=["dulwich"])
 
 
 def _merge_upstream(git: "Git"):

dvc/repo/__init__.py

@@ -23,6 +23,7 @@ if TYPE_CHECKING:
     from dvc.stage import Stage
     from dvc.types import DictStrAny
     from dvc_data.hashfile.state import StateBase
+    from dvc_data.index import DataIndex
 
     from .experiments import Experiments
     from .index import Index
@@ -179,6 +180,7 @@ class Repo:
         self._fs_conf = {"repo_factory": repo_factory}
         self._fs = fs or localfs
         self._scm = scm
+        self._data_index = None
 
         if rev and not fs:
             self._scm = scm = SCM(root_dir or os.curdir)
@@ -341,6 +343,31 @@ class Repo:
         # fs.
         self._reset()
 
+    @property
+    def data_index(self) -> Optional["DataIndex"]:
+        from appdirs import user_cache_dir
+        from fsspec.utils import tokenize
+
+        from dvc_data.index import DataIndex
+
+        if not self.config["feature"].get("data_index_cache"):
+            return None
+
+        if self._data_index is None:
+            cache_dir = user_cache_dir(self.config.APPNAME, self.config.APPAUTHOR)
+            index_dir = os.path.join(
+                cache_dir,
+                "index",
+                "data",
+                # scm.root_dir and repo.root_dir don't match for subrepos
+                tokenize((self.scm.root_dir, self.root_dir)),
+            )
+            os.makedirs(index_dir, exist_ok=True)
+
+            self._data_index = DataIndex.open(os.path.join(index_dir, "db.db"))
+
+        return self._data_index
+
     def __repr__(self):
         return f"{self.__class__.__name__}: '{self.root_dir}'"
 
@@ -553,18 +580,17 @@ class Repo:
     def close(self):
         self.scm.close()
         self.state.close()
+        if self._data_index is not None:
+            self._data_index.close()
 
-    def _reset_cached_indecies(self):
+    def _reset(self):
+        self.scm._reset()  # pylint: disable=protected-access
+        self.state.close()
         self.__dict__.pop("index", None)
         self.__dict__.pop("dvcignore", None)
         self.__dict__.pop("dvcfs", None)
         self.__dict__.pop("datafs", None)
 
-    def _reset(self):
-        self.state.close()
-        self.scm._reset()  # pylint: disable=protected-access
-        self._reset_cached_indecies()
-
     def __enter__(self):
         return self
 

dvc/repo/brancher.py

@@ -1,7 +1,10 @@
+import logging
 from typing import Optional
 
 from dvc.scm import iter_revs
 
+logger = logging.getLogger(__name__)
+
 
 def brancher(  # noqa: E302
     self,
@@ -60,6 +63,7 @@ def brancher(  # noqa: E302
 
     scm = self.scm
 
+    logger.trace("switching fs to workspace")  # type: ignore[attr-defined]
     self.fs = LocalFileSystem(url=self.root_dir)
     yield "workspace"
 
@@ -82,7 +86,9 @@ def brancher(  # noqa: E302
         from dvc.fs import GitFileSystem
 
         for sha, names in found_revs.items():
-            self._reset_cached_indecies()  # pylint: disable=W0212
+            logger.trace(  # type: ignore[attr-defined]
+                "switching fs to revision %s [%s]", sha[:7], ", ".join(names)
+            )
             self.fs = GitFileSystem(scm=scm, rev=sha)
             self.root_dir = self.fs.path.join("/", *repo_root_parts)
 

dvc/repo/data.py

@@ -94,7 +94,7 @@ def _diff(
             with_missing
             and change.old
             and change.old.hash_info
-            and not old.storage_map[change.key].cache.exists(change.old.hash_info.value)
+            and not old.storage_map.cache_exists(change.old)
         ):
             # NOTE: emulating previous behaviour
             _add_change("not_in_cache", change)

dvc/repo/experiments/queue/celery.py

@@ -336,7 +336,8 @@ class LocalCeleryQueue(BaseStashQueue):
                         task_id,
                         remained_entries[entry],
                     )
-                    self.celery.backend.mark_as_failure(task_id, None)
+                    backend = self.celery.backend
+                    backend.mark_as_failure(task_id, None)  # type: ignore[attr-defined]
 
         if remained_revs:
             raise CannotKillTasksError(remained_revs)

dvc/repo/experiments/queue/remove.py

@@ -40,7 +40,7 @@ def remove_tasks(  # noqa: C901
             msg,
             queue_entry,
         ) in celery_queue._iter_queued():  # pylint: disable=protected-access
-            if queue_entry.stash_rev in stash_revs:
+            if queue_entry.stash_rev in stash_revs and msg.delivery_tag:
                 celery_queue.celery.reject(msg.delivery_tag)
     finally:
         celery_queue.stash.remove_revs(list(stash_revs.values()))
@@ -56,7 +56,8 @@ def remove_tasks(  # noqa: C901
             result: AsyncResult = AsyncResult(task_id)
             if result is not None:
                 result.forget()
-            celery_queue.celery.purge(msg.delivery_tag)
+            if msg.delivery_tag:
+                celery_queue.celery.purge(msg.delivery_tag)
     finally:
         if celery_queue.failed_stash:
             celery_queue.failed_stash.remove_revs(failed_stash_revs)

dvc/repo/experiments/utils.py

@@ -5,6 +5,7 @@ from collections import defaultdict
 from functools import wraps
 from typing import (
     TYPE_CHECKING,
+    Any,
     Callable,
     Dict,
     Generator,
@@ -279,17 +280,15 @@ def check_ref_format(scm: "Git", ref: ExpRefInfo):
         )
 
 
-def fetch_all_exps(scm: "Git", url: str, progress: Optional[Callable] = None):
+def fetch_all_exps(
+    scm: "Git", url: str, progress: Optional[Callable] = None, **kwargs: Any
+):
     refspecs = [
         f"{ref}:{ref}"
-        for ref in iter_remote_refs(scm, url, base=EXPS_NAMESPACE)
+        for ref in iter_remote_refs(scm, url, base=EXPS_NAMESPACE, **kwargs)
         if not (ref.startswith(EXEC_NAMESPACE) or ref in STASHES)
     ]
-    scm.fetch_refspecs(
-        url,
-        refspecs,
-        progress=progress,
-    )
+    scm.fetch_refspecs(url, refspecs, progress=progress, **kwargs)
 
 
 def get_random_exp_name(scm, baseline_rev):

dvc/repo/imports.py

@@ -115,7 +115,7 @@ def save_imports(
                 desc="Downloading imports from source",
                 unit="files",
             ) as cb:
-                checkout(data_view, tmpdir, cache.fs, callback=cb)
+                checkout(data_view, tmpdir, cache.fs, callback=cb, storage="data")
             md5(data_view)
             save(data_view, odb=cache, hardlink=True)
 

dvc/repo/index.py

@@ -112,6 +112,59 @@ def collect_files(
         dirs[:] = [d for d in dirs if not is_out_or_ignored(root, d)]
 
 
+def _load_data_from_outs(index, prefix, outs):
+    from dvc_data.index import DataIndexEntry
+
+    for out in outs:
+        if not out.use_cache:
+            continue
+
+        ws, key = out.index_key
+
+        entry = DataIndexEntry(
+            key=key,
+            meta=out.meta,
+            hash_info=out.hash_info,
+        )
+
+        if out.stage.is_import and not out.stage.is_repo_import:
+            dep = out.stage.deps[0]
+            entry.meta = dep.meta
+            entry.hash_info = dep.hash_info
+
+        # FIXME PyGTrie-based DataIndex doesn't remove entry.key during
+        # index.add, so we have to set the entry manually here to make
+        # index.view() work correctly.
+        index[(*prefix, ws, *key)] = entry
+
+
+def _load_storage_from_out(storage_map, key, out):
+    from dvc.config import NoRemoteError
+    from dvc_data.index import FileStorage, ObjectStorage
+
+    if out.odb:
+        storage_map.add_data(ObjectStorage(key, out.odb))
+    storage_map.add_cache(ObjectStorage(key, out.cache))
+    try:
+        remote = out.repo.cloud.get_remote(out.remote)
+        if remote.fs.version_aware:
+            storage_map.add_remote(
+                FileStorage(
+                    key=key,
+                    fs=remote.fs,
+                    path=remote.fs.path.join(remote.path, *key),
+                )
+            )
+        else:
+            storage_map.add_remote(ObjectStorage(key, remote.odb))
+    except NoRemoteError:
+        pass
+
+    if out.stage.is_import and not out.stage.is_repo_import:
+        dep = out.stage.deps[0]
+        storage_map.add_data(FileStorage(key, dep.fs, dep.fs_path))
+
+
 class Index:
     def __init__(
         self,
@@ -282,50 +335,60 @@ class Index:
         return dict(by_workspace)
 
     @cached_property
-    def data(self) -> "Dict[str, DataIndex]":
-        from collections import defaultdict
-
-        from dvc.config import NoRemoteError
-        from dvc_data.index import DataIndex, DataIndexEntry, Storage
-
-        by_workspace: dict = defaultdict(DataIndex)
-
-        by_workspace["repo"] = DataIndex()
-        by_workspace["local"] = DataIndex()
+    def data_tree(self):
+        from dvc_data.hashfile import Tree
 
+        tree = Tree()
         for out in self.outs:
             if not out.use_cache:
                 continue
 
-            workspace, key = out.index_key
-            data_index = by_workspace[workspace]
+            ws, key = out.index_key
 
-            entry = DataIndexEntry(
-                key=key,
-                meta=out.meta,
-                hash_info=out.hash_info,
-            )
+            tree.add((ws, *key), out.meta, out.hash_info)
 
-            storage = Storage(odb=out.odb, cache=out.cache)
-            try:
-                storage.remote = self.repo.cloud.get_remote_odb(out.remote)
-            except NoRemoteError:
-                pass
+        tree.digest()
 
-            if out.stage.is_import and not out.stage.is_repo_import:
-                dep = out.stage.deps[0]
+        return tree
 
-                entry.meta = dep.meta
-                entry.hash_info = dep.hash_info
+    @cached_property
+    def data(self) -> "Dict[str, DataIndex]":
+        from dvc_data.index import DataIndex
 
-                storage.odb = dep.odb
-                storage.fs = dep.fs
-                storage.path = dep.fs_path
+        prefix: "DataIndexKey"
+        loaded = False
 
-            data_index.add(entry)
-            data_index.storage_map[key] = storage
+        index = self.repo.data_index
+        if index is None:
+            index = DataIndex()
 
-        return dict(by_workspace)
+        prefix = (self.data_tree.hash_info.value,)
+        if prefix in index.ls((), detail=False):
+            loaded = True
+
+        try:
+            if not loaded:
+                _load_data_from_outs(index, prefix, self.outs)
+                index.commit()
+
+            by_workspace = {}
+            by_workspace["repo"] = index.view((*prefix, "repo"))
+            by_workspace["local"] = index.view((*prefix, "local"))
+
+            for out in self.outs:
+                if not out.use_cache:
+                    continue
+
+                ws, key = out.index_key
+                if ws not in by_workspace:
+                    by_workspace[ws] = index.view((*prefix, ws))
+
+                data_index = by_workspace[ws]
+                _load_storage_from_out(data_index.storage_map, key, out)
+
+            return by_workspace
+        finally:
+            index.close()
 
     @staticmethod
     def _hash_targets(

dvc/repo/worktree.py

@@ -119,12 +119,6 @@ def fetch_worktree(
     ):
         remote_obj = _get_remote(repo, remote_name, remote, "fetch")
         index = view.data["repo"]
-        for key in index.storage_map:
-            storage = index.storage_map[key]
-            storage.fs = remote_obj.fs
-            storage.path = remote_obj.fs.path.join(remote_obj.path, *key)
-            index.storage_map[key] = storage
-
         total = len(index)
         with Callback.as_tqdm_callback(
             unit="file",
@@ -132,7 +126,7 @@ def fetch_worktree(
             disable=total == 0,
         ) as cb:
             cb.set_size(total)
-            transferred += save(index, callback=cb, jobs=jobs)
+            transferred += save(index, callback=cb, jobs=jobs, storage="remote")
     return transferred
 
 
@@ -357,6 +351,7 @@ def _fetch_out_changes(
             update_meta=False,
             meta_only=True,
             meta_cmp_key=partial(_meta_checksum, remote.fs),
+            storage="data",
             callback=cb,
         )
         out.save()

dvc/scm.py

@@ -179,7 +179,7 @@ def clone(url: str, to_path: str, **kwargs):
         try:
             git = Git.clone(url, to_path, progress=pbar.update_git, **kwargs)
             if "shallow_branch" not in kwargs:
-                fetch_all_exps(git, url, progress=pbar.update_git)
+                fetch_all_exps(git, url, progress=pbar.update_git, backends=["dulwich"])
             return git
         except InternalCloneError as exc:
             raise CloneError("SCM error") from exc

pyproject.toml

@@ -46,14 +46,14 @@ dependencies = [
     "pygtrie>=2.3.2",
     "dpath>=2.0.2,<3",
     "shtab>=1.3.4,<2",
-    "rich>=10.13.0",
+    "rich>=12.0.0",
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
     "scmrepo==0.1.9",
-    "dvc-render==0.1.1",
+    "dvc-render>=0.1.2",
     "dvc-task==0.1.11",
     "dvclive>=1.2.2",
-    "dvc-data==0.39.3",
+    "dvc-data==0.40.1",
     "dvc-http",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.7",

tests/func/test_add.py

@@ -342,6 +342,13 @@ def test_double_add_unchanged_dir(tmp_dir, dvc):
     assert ret == 0
 
 
+@pytest.mark.skipif(os.name == "nt", reason="unsupported on Windows")
+def test_add_colon_in_filename(tmp_dir, dvc):
+    tmp_dir.gen("fo:o", "foo")
+    ret = main(["add", "fo:o"])
+    assert ret == 0
+
+
 def test_should_update_state_entry_for_file_after_add(mocker, dvc, tmp_dir):
     file_md5_counter = mocker.spy(dvc_data.hashfile.hash, "file_md5")
     tmp_dir.gen("foo", "foo")

tests/func/test_data_cloud.py

@@ -199,7 +199,8 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     hash_spy = mocker.spy(dvc_data.hashfile.hash, "file_md5")
 
     dvc.pull()
-    assert hash_spy.call_count == 0
+    # NOTE: 1 is for index.data_tree building
+    assert hash_spy.call_count == 1
 
     # Removing cache will invalidate existing state entries
     dvc.cache.local.clear()
@@ -207,7 +208,7 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     dvc.config["remote"]["upstream"]["verify"] = True
 
     dvc.pull()
-    assert hash_spy.call_count == 4
+    assert hash_spy.call_count == 6
 
 
 @flaky(max_runs=3, min_passes=1)

tests/unit/command/test_plots.py

@@ -285,7 +285,7 @@ def test_should_call_render(tmp_dir, mocker, capsys, plots_data, output):
     render_mock.assert_called_once_with(
         renderers=renderers,
         output_file=Path(tmp_dir / output / "index.html"),
-        template_path=None,
+        html_template=None,
     )