fetch: introduce --max-size (#9696)

Needed for iterative/studio#4782

.dvc/.gitignore

@@ -0,0 +1,3 @@
+/config.local
+/tmp
+/cache

.github/workflows/build.yaml

@@ -1,4 +1,4 @@
-name: Build packages
+name: Build package
 on:
   push:
     branches:
@@ -7,39 +7,35 @@ on:
     types: [released, prereleased]
   workflow_dispatch:
 
-permissions:  # added using https://github.com/step-security/secure-workflows
+permissions:
   contents: read
+  id-token: write
 
 jobs:
   pip:
-    runs-on: ubuntu-20.04
-    permissions:
-      id-token: write
+    runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
       with:
         fetch-depth: 0
 
-    - name: Set up Python 3.8
-      uses: actions/setup-python@v4
+    - uses: actions/setup-python@v4
       with:
         python-version: 3.8
 
-    - name: Install
-      run: pip install --upgrade pip wheel
-
-    - name: Force version for Test PyPI uploads
+    - run: python -m pip install --upgrade pip setuptools_scm build twine
+    - name: Force version for Test PyPI upload
       if: ${{ !startsWith(github.ref, 'refs/tags') }}
-      run: |
-        pip install setuptools_scm
-        echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
+      run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
 
-    - name: Build packages
-      run: ./scripts/build_package.sh
+    - run: |
+        echo 'PKG = "pip"'>dvc/_build.py
+        python -m build
       env:
         SETUPTOOLS_SCM_PRETEND_VERSION: ${{ env.version }}
+    - run: twine check --strict dist/*
 
-    - name: Publish packages to PyPI
+    - name: Publish to PyPI
       if: github.event_name == 'release'
       uses: pypa/gh-action-pypi-publish@release/v1
 
@@ -47,5 +43,5 @@ jobs:
       if: ${{ github.event_name == 'release' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }}
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
-        repository_url: https://test.pypi.org/legacy/
-        skip_existing: true
+        repository-url: https://test.pypi.org/legacy/
+        skip-existing: true

dvc/__pyinstaller/__init__.py

@@ -0,0 +1,5 @@
+import os
+
+
+def get_hook_dirs():
+    return [os.path.dirname(__file__)]

dvc/__pyinstaller/hook-asyncssh.py

@@ -1 +1,2 @@
+# ruff: noqa: N999
 hiddenimports = ["win32timezone"]

dvc/__pyinstaller/hook-celery.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 # pylint: disable=import-error
 from PyInstaller.utils.hooks import collect_submodules, is_module_or_submodule
 

dvc/__pyinstaller/hook-dvc.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 # needed for `dvc doctor` to show dep versions

dvc/__pyinstaller/hook-dvc.utils.flatten.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 datas = copy_metadata("flatten-dict")

dvc/__pyinstaller/hook-dvc_task.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 # pylint: disable=import-error
 from PyInstaller.utils.hooks import collect_submodules
 

dvc/__pyinstaller/hook-fsspec.py

@@ -1 +1,2 @@
+# ruff: noqa: N999
 hiddenimports = ["fsspec.implementations.memory"]

dvc/__pyinstaller/hook-pydrive2.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 datas = copy_metadata("pydrive2")

dvc/_debug.py

@@ -198,33 +198,84 @@ def debugtools(args: Optional["Namespace"] = None, **kwargs):
 def add_debugging_flags(parser):
     from argparse import SUPPRESS
 
-    parser.add_argument("--cprofile", action="store_true", default=False, help=SUPPRESS)
-    parser.add_argument("--yappi", action="store_true", default=False, help=SUPPRESS)
+    # For detailed info see:
+    # https://github.com/iterative/dvc/wiki/Debugging,-Profiling-and-Benchmarking-DVC
+    args, _ = parser.parse_known_args()
+    verbose = args.verbose
+
+    def debug_help(msg):
+        if verbose:
+            return msg
+        return SUPPRESS
+
+    parser = parser.add_argument_group("debug options")
+
+    parser.add_argument(
+        "--cprofile",
+        action="store_true",
+        default=False,
+        help=debug_help("Generate cprofile data for tools like snakeviz / tuna"),
+    )
+    parser.add_argument(
+        "--cprofile-dump", help=debug_help("Location to dump cprofile file")
+    )
+    parser.add_argument(
+        "--yappi",
+        action="store_true",
+        default=False,
+        help=debug_help(
+            "Generate a callgrind file for use with tools like "
+            "kcachegrind / qcachegrind"
+        ),
+    )
     parser.add_argument(
         "--yappi-separate-threads",
         action="store_true",
         default=False,
-        help=SUPPRESS,
+        help=debug_help("Generate one callgrind file per thread"),
     )
     parser.add_argument(
-        "--viztracer", action="store_true", default=False, help=SUPPRESS
-    )
-    parser.add_argument("--viztracer-depth", type=int, help=SUPPRESS)
-    parser.add_argument(
-        "--viztracer-async", action="store_true", default=False, help=SUPPRESS
-    )
-    parser.add_argument("--cprofile-dump", help=SUPPRESS)
-    parser.add_argument("--pdb", action="store_true", default=False, help=SUPPRESS)
-    parser.add_argument(
-        "--instrument", action="store_true", default=False, help=SUPPRESS
+        "--viztracer",
+        action="store_true",
+        default=False,
+        help=debug_help("Generate a viztracer file for use with vizviewer"),
     )
     parser.add_argument(
-        "--instrument-open", action="store_true", default=False, help=SUPPRESS
+        "--viztracer-depth",
+        type=int,
+        help=debug_help("Set viztracer maximum stack depth"),
+    )
+    parser.add_argument(
+        "--viztracer-async",
+        action="store_true",
+        default=False,
+        help=debug_help("Treat async tasks as threads"),
+    )
+    parser.add_argument(
+        "--pdb",
+        action="store_true",
+        default=False,
+        help=debug_help("Drop into the pdb/ipdb debugger on any exception"),
+    )
+    parser.add_argument(
+        "--instrument",
+        action="store_true",
+        default=False,
+        help=debug_help("Use pyinstrument CLI profiler"),
+    )
+    parser.add_argument(
+        "--instrument-open",
+        action="store_true",
+        default=False,
+        help=debug_help("Use pyinstrument web profiler"),
     )
     parser.add_argument(
         "--show-stack",
         "--ss",
         action="store_true",
         default=False,
-        help=SUPPRESS,
+        help=debug_help(
+            r"Use Ctrl+T on macOS or Ctrl+\ on Linux to print the stack "
+            "frame currently executing. Unavailable on Windows."
+        ),
     )

dvc/cli/parser.py

@@ -137,7 +137,6 @@ def get_parent_parser():
     from dvc._debug import add_debugging_flags
 
     parent_parser = argparse.ArgumentParser(add_help=False)
-    add_debugging_flags(parent_parser)
     log_level_group = parent_parser.add_mutually_exclusive_group()
     log_level_group.add_argument(
         "-q", "--quiet", action="count", default=0, help="Be quiet."
@@ -145,6 +144,7 @@ def get_parent_parser():
     log_level_group.add_argument(
         "-v", "--verbose", action="count", default=0, help="Be verbose."
     )
+    add_debugging_flags(parent_parser)
 
     return parent_parser
 

dvc/commands/data_sync.py

@@ -91,6 +91,7 @@ class CmdDataFetch(CmdDataBase):
                 with_deps=self.args.with_deps,
                 recursive=self.args.recursive,
                 run_cache=self.args.run_cache,
+                max_size=self.args.max_size,
             )
             self.log_summary({"fetched": processed_files_count})
         except DvcException:
@@ -322,6 +323,11 @@ def add_parser(subparsers, _parent_parser):
         default=False,
         help="Fetch run history for all stages.",
     )
+    fetch_parser.add_argument(
+        "--max-size",
+        type=int,
+        help="Fetch data files/directories that are each below specified size (bytes).",
+    )
     fetch_parser.set_defaults(func=CmdDataFetch)
 
     # Status

dvc/dagascii.py

@@ -209,6 +209,30 @@ def draw(vertices, edges):
     Args:
         vertices (list): list of graph vertices.
         edges (list): list of graph edges.
+
+    Returns:
+        str: ASCII representation
+
+    Example:
+        >>> from dvc.dagascii import draw
+        >>> vertices = [1, 2, 3, 4]
+        >>> edges = [(1, 2), (2, 3), (2, 4), (1, 4)]
+        >>> print(draw(vertices, edges))
+        +---+     +---+
+        | 3 |     | 4 |
+        +---+    *+---+
+          *    **   *
+          *  **     *
+          * *       *
+        +---+       *
+        | 2 |      *
+        +---+     *
+             *    *
+              *  *
+               **
+             +---+
+             | 1 |
+             +---+
     """
     # pylint: disable=too-many-locals
     # NOTE: coordinates might me negative, so we need to shift

dvc/output.py

@@ -20,6 +20,7 @@ from dvc.exceptions import (
     DvcException,
     MergeError,
 )
+from dvc.utils import format_link
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile import check as ocheck
 from dvc_data.hashfile import load as oload
@@ -678,6 +679,14 @@ class Output:
     # pylint: enable=no-member
 
     def save(self) -> None:
+        if self.use_cache and not self.is_in_repo:
+            raise DvcException(
+                f"Saving cached external output {str(self)} is not supported "
+                "since DVC 3.0. See "
+                f"{format_link('https://dvc.org/doc/user-guide/upgrade')} "
+                "for more info."
+            )
+
         if not self.exists:
             raise self.DoesNotExistError(self)
 
@@ -1270,7 +1279,7 @@ class Output:
             raise FileNotFoundError(  # noqa: B904
                 errno.ENOENT,
                 os.strerror(errno.ENOENT),
-                self.fs.path.relpath(path),
+                path,
             )
 
         new = tree.from_trie(trie)
@@ -1393,8 +1402,7 @@ class Output:
         otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
 
         if relink:
-            rel = self.fs.path.relpath(path)
-            with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as callback:
+            with CheckoutCallback(desc="Checking out {path}", unit="files") as callback:
                 self._checkout(
                     path,
                     self.fs,

dvc/repo/experiments/executor/base.py

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.repo.experiments.stash import ExpStashEntry
     from dvc.scm import Git
-    from dvc.stage import PipelineStage
+    from dvc.stage import PipelineStage, Stage
 
 logger = logging.getLogger(__name__)
 
@@ -255,6 +255,15 @@ class BaseExecutor(ABC):
             **kwargs,
         )
 
+    @classmethod
+    def _get_stage_files(cls, stages: List["Stage"]) -> List[str]:
+        from dvc.stage.utils import _get_stage_files
+
+        ret: List[str] = []
+        for stage in stages:
+            ret.extend(_get_stage_files(stage))
+        return ret
+
     @classmethod
     def _get_top_level_paths(cls, repo: "Repo") -> List["str"]:
         return list(
@@ -459,25 +468,16 @@ class BaseExecutor(ABC):
             should force overwrite any existing duplicates.
         """
         from dvc.repo.checkout import checkout as dvc_checkout
-        from dvc.repo.reproduce import reproduce as dvc_reproduce
-        from dvc.stage import PipelineStage
         from dvc.ui import ui
 
         auto_push = env2bool(DVC_EXP_AUTO_PUSH)
         git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
 
-        unchanged = []
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
             cls._set_log_level(log_level)
 
-        def filter_pipeline(stages):
-            unchanged.extend(
-                [stage for stage in stages if isinstance(stage, PipelineStage)]
-            )
-
         exp_hash: Optional[str] = None
         exp_ref: Optional["ExpRefInfo"] = None
         repro_force: bool = False
@@ -519,12 +519,10 @@ class BaseExecutor(ABC):
                     recursive=kwargs.get("recursive", False),
                 )
 
-            stages = dvc_reproduce(
-                dvc,
-                *args,
-                on_unchanged=filter_pipeline,
-                **kwargs,
-            )
+            stages = dvc.reproduce(*args, **kwargs)
+            if paths := cls._get_stage_files(stages):
+                logger.debug("Staging stage-related files: %s", paths)
+                dvc.scm_context.add(paths)
             if paths := cls._get_top_level_paths(dvc):
                 logger.debug("Staging top-level files: %s", paths)
                 dvc.scm_context.add(paths)

dvc/repo/fetch.py

@@ -1,6 +1,8 @@
 import logging
+from typing import Dict, List, Tuple
 
 from dvc.exceptions import DownloadError
+from dvc_data.index import DataIndex, FileStorage
 
 from . import locked
 
@@ -20,6 +22,7 @@ def fetch(  # noqa: C901, PLR0913
     all_commits=False,
     run_cache=False,
     revs=None,
+    max_size=None,
 ) -> int:
     """Download data items from a cloud and imported repositories
 
@@ -68,6 +71,7 @@ def fetch(  # noqa: C901, PLR0913
                 targets,
                 with_deps=with_deps,
                 recursive=recursive,
+                max_size=max_size,
             )
             index_keys.add(idx.data_tree.hash_info.value)
             indexes.append(idx.data["repo"])
@@ -84,6 +88,7 @@ def fetch(  # noqa: C901, PLR0913
         data = collect(
             indexes, cache_index=self.data_index, cache_key=cache_key, callback=cb
         )
+    failed_count += _log_unversioned(data)
 
     with Callback.as_tqdm_callback(
         desc="Fetching",
@@ -109,3 +114,31 @@ def fetch(  # noqa: C901, PLR0913
         raise DownloadError(failed_count)
 
     return transferred_count
+
+
+def _log_unversioned(data: Dict[Tuple[str, str], "DataIndex"]) -> int:
+    unversioned: List[str] = []
+    for by_fs, fs_index in data.items():
+        remote = fs_index.storage_map[()].remote
+        if not isinstance(remote, FileStorage) or not remote.fs.version_aware:
+            continue
+
+        fs = remote.fs
+        index = DataIndex()
+        index.storage_map = fs_index.storage_map
+        for key, entry in fs_index.iteritems():
+            if entry.meta and not entry.meta.isdir and entry.meta.version_id is None:
+                unversioned.append(fs.unstrip_protocol(fs.path.join(remote.path, *key)))
+            else:
+                index[key] = entry
+        fs_index.close()
+        data[by_fs] = index
+    if unversioned:
+        logger.warning(
+            (
+                "Some files are missing cloud version information and will not be "
+                "fetched from the remote:\n%s"
+            ),
+            "\n".join(unversioned),
+        )
+    return len(unversioned)

dvc/repo/graph.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Iterator, List, Set
+from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Set, TypeVar
 
 from dvc.fs import localfs
 from dvc.utils.fs import path_isin
@@ -8,6 +8,8 @@ if TYPE_CHECKING:
 
     from dvc.stage import Stage
 
+T = TypeVar("T")
+
 
 def check_acyclic(graph: "DiGraph") -> None:
     import networkx as nx
@@ -39,6 +41,30 @@ def get_pipelines(graph: "DiGraph"):
     return [graph.subgraph(c).copy() for c in nx.weakly_connected_components(graph)]
 
 
+def get_subgraph_of_nodes(
+    graph: "DiGraph", sources: Optional[List[Any]] = None, downstream: bool = False
+) -> "DiGraph":
+    from networkx import dfs_postorder_nodes, reverse_view
+
+    if not sources:
+        return graph
+
+    g = reverse_view(graph) if downstream else graph
+    nodes = []
+    for source in sources:
+        nodes.extend(dfs_postorder_nodes(g, source))
+    return graph.subgraph(nodes)
+
+
+def get_steps(
+    graph: "DiGraph", sources: Optional[List[T]] = None, downstream: bool = False
+) -> List[T]:
+    from networkx import dfs_postorder_nodes
+
+    sub = get_subgraph_of_nodes(graph, sources, downstream=downstream)
+    return list(dfs_postorder_nodes(sub))
+
+
 def collect_pipeline(stage: "Stage", graph: "DiGraph") -> Iterator["Stage"]:
     import networkx as nx
 

dvc/repo/index.py

@@ -492,6 +492,7 @@ class Index:
         targets: Optional["TargetType"],
         stage_filter: Optional[Callable[["Stage"], bool]] = None,
         outs_filter: Optional[Callable[["Output"], bool]] = None,
+        max_size: Optional[int] = None,
         **kwargs: Any,
     ) -> "IndexView":
         """Return read-only view of index for the specified targets.
@@ -514,7 +515,17 @@ class Index:
             for stage_info in self.collect_targets(targets, **kwargs)
             if not stage_filter or stage_filter(stage_info.stage)
         ]
-        return IndexView(self, stage_infos, outs_filter=outs_filter)
+
+        def _outs_filter(out):
+            if max_size and out.meta and out.meta.size and out.meta.size >= max_size:
+                return False
+
+            if outs_filter:
+                return outs_filter(out)
+
+            return True
+
+        return IndexView(self, stage_infos, outs_filter=_outs_filter)
 
 
 class _DataPrefixes(NamedTuple):

dvc/repo/pull.py

@@ -1,5 +1,6 @@
 import logging
 
+from dvc.exceptions import CheckoutError
 from dvc.repo import locked
 from dvc.utils import glob_targets
 
@@ -38,13 +39,17 @@ def pull(  # noqa: PLR0913
         recursive=recursive,
         run_cache=run_cache,
     )
-    stats = self.checkout(
-        targets=expanded_targets,
-        with_deps=with_deps,
-        force=force,
-        recursive=recursive,
-        allow_missing=allow_missing,
-    )
+    try:
+        stats = self.checkout(
+            targets=expanded_targets,
+            with_deps=with_deps,
+            force=force,
+            recursive=recursive,
+            allow_missing=allow_missing,
+        )
+    except CheckoutError as exc:
+        exc.stats["fetched"] = processed_files_count
+        raise
 
     stats["fetched"] = processed_files_count
     return stats

dvc/repo/reproduce.py

@@ -1,13 +1,18 @@
 import logging
-from typing import TYPE_CHECKING, Iterator, List
+from typing import TYPE_CHECKING, Iterable, List, Optional, Union, cast
+
+from funcy import ldistinct
 
 from dvc.exceptions import ReproductionError
 from dvc.repo.scm_context import scm_context
 from dvc.stage.cache import RunCacheNotSupported
+from dvc.utils.collections import ensure_list
 
 from . import locked
 
 if TYPE_CHECKING:
+    from networkx import DiGraph
+
     from dvc.stage import Stage
 
     from . import Repo
@@ -15,116 +20,47 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-def _reproduce_stage(stage: "Stage", **kwargs) -> List["Stage"]:
+def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
     if stage.frozen and not stage.is_import:
         logger.warning(
             "%s is frozen. Its dependencies are not going to be reproduced.",
             stage,
         )
 
-    stage = stage.reproduce(**kwargs)
-    if not stage:
-        return []
-
-    if not kwargs.get("dry", False):
+    ret = stage.reproduce(**kwargs)
+    if ret and not kwargs.get("dry", False):
         stage.dump(update_pipeline=False)
-        _track_stage(stage)
-
-    return [stage]
+    return ret
 
 
-def _get_stage_files(stage: "Stage") -> Iterator[str]:
-    yield stage.dvcfile.relpath
-    for dep in stage.deps:
-        if (
-            not dep.use_scm_ignore
-            and dep.is_in_repo
-            and not stage.repo.dvcfs.isdvc(stage.repo.dvcfs.from_os_path(str(dep)))
-        ):
-            yield dep.fs_path
-    for out in stage.outs:
-        if not out.use_scm_ignore and out.is_in_repo:
-            yield out.fs_path
+def collect_stages(
+    repo: "Repo",
+    targets: Iterable[str],
+    recursive: bool = False,
+    glob: bool = False,
+) -> List["Stage"]:
+    stages: List["Stage"] = []
+    for target in targets:
+        stages.extend(repo.stage.collect(target, recursive=recursive, glob=glob))
+    return ldistinct(stages)
 
 
-def _track_stage(stage: "Stage") -> None:
-    from dvc.utils import relpath
-
-    context = stage.repo.scm_context
-    for path in _get_stage_files(stage):
-        context.track_file(relpath(path))
-    return context.track_changed_files()
+def _remove_frozen_stages(graph: "DiGraph") -> "DiGraph":
+    g = cast("DiGraph", graph.copy())
+    for stage in graph:
+        if stage.frozen:
+            # NOTE: disconnect frozen stage from its dependencies
+            g.remove_edges_from(graph.out_edges(stage))
+    return g
 
 
-@locked
-@scm_context
-def reproduce(  # noqa: C901, PLR0912
-    self: "Repo",
-    targets=None,
-    recursive=False,
-    pipeline=False,
-    all_pipelines=False,
-    **kwargs,
-):
-    from .graph import get_pipeline, get_pipelines
-
-    glob = kwargs.pop("glob", False)
-
-    if isinstance(targets, str):
-        targets = [targets]
-
-    if not all_pipelines and not targets:
-        from dvc.dvcfile import PROJECT_FILE
-
-        targets = [PROJECT_FILE]
-
-    interactive = kwargs.get("interactive", False)
-    if not interactive:
-        kwargs["interactive"] = self.config["core"].get("interactive", False)
-
-    stages = set()
-    if pipeline or all_pipelines:
-        pipelines = get_pipelines(self.index.graph)
-        if all_pipelines:
-            used_pipelines = pipelines
-        else:
-            used_pipelines = []
-            for target in targets:
-                stage = self.stage.get_target(target)
-                used_pipelines.append(get_pipeline(pipelines, stage))
-
-        for pline in used_pipelines:
-            for stage in pline:
-                if pline.in_degree(stage) == 0:
-                    stages.add(stage)
-    else:
-        for target in targets:
-            stages.update(
-                self.stage.collect(
-                    target,
-                    recursive=recursive,
-                    glob=glob,
-                )
-            )
-
-    if kwargs.get("pull", False) and kwargs.get("run_cache", True):
-        logger.debug("Pulling run cache")
-        try:
-            self.stage_cache.pull(None)
-        except RunCacheNotSupported as e:
-            logger.warning("Failed to pull run cache: %s", e)
-
-    return _reproduce_stages(self.index.graph, list(stages), **kwargs)
-
-
-def _reproduce_stages(  # noqa: C901
-    graph,
-    stages,
-    downstream=False,
-    single_item=False,
-    on_unchanged=None,
-    **kwargs,
-):
+def plan_repro(
+    graph: "DiGraph",
+    stages: Optional[List["Stage"]] = None,
+    pipeline: bool = False,
+    downstream: bool = False,
+    all_pipelines: bool = False,
+) -> List["Stage"]:
     r"""Derive the evaluation of the given node for the given graph.
 
     When you _reproduce a stage_, you want to _evaluate the descendants_
@@ -160,80 +96,88 @@ def _reproduce_stages(  # noqa: C901
 
     The derived evaluation of _downstream_ B would be: [B, D, E]
     """
-    steps = _get_steps(graph, stages, downstream, single_item)
+    from .graph import get_pipeline, get_pipelines, get_steps
 
-    force_downstream = kwargs.pop("force_downstream", False)
-    result: List["Stage"] = []
-    unchanged: List["Stage"] = []
-    # `ret` is used to add a cosmetic newline.
-    ret: List["Stage"] = []
+    if pipeline or all_pipelines:
+        pipelines = get_pipelines(graph)
+        if stages and pipeline:
+            pipelines = [get_pipeline(pipelines, stage) for stage in stages]
 
-    for stage in steps:
-        if ret:
-            logger.info("")
+        leaves: List["Stage"] = []
+        for pline in pipelines:
+            leaves.extend(node for node in pline if pline.in_degree(node) == 0)
+        stages = ldistinct(leaves)
 
+    active = _remove_frozen_stages(graph)
+    return get_steps(active, stages, downstream=downstream)
+
+
+@locked
+@scm_context
+def reproduce(  # noqa: C901
+    self: "Repo",
+    targets: Union[Iterable[str], str, None] = None,
+    recursive: bool = False,
+    pipeline: bool = False,
+    all_pipelines: bool = False,
+    downstream: bool = False,
+    single_item: bool = False,
+    glob: bool = False,
+    **kwargs,
+):
+    from dvc.dvcfile import PROJECT_FILE
+
+    if not kwargs.get("interactive", False):
+        kwargs["interactive"] = self.config["core"].get("interactive", False)
+
+    stages: List["Stage"] = []
+    if not all_pipelines:
+        targets_list = ensure_list(targets or PROJECT_FILE)
+        stages = collect_stages(self, targets_list, recursive=recursive, glob=glob)
+
+    if kwargs.get("pull", False) and kwargs.get("run_cache", True):
+        logger.debug("Pulling run cache")
         try:
-            kwargs["upstream"] = result + unchanged
-            ret = _reproduce_stage(stage, **kwargs)
+            self.stage_cache.pull(None)
+        except RunCacheNotSupported as e:
+            logger.warning("Failed to pull run cache: %s", e)
 
-            if len(ret) == 0:
-                unchanged.extend([stage])
-            elif force_downstream:
-                # NOTE: we are walking our pipeline from the top to the
-                # bottom. If one stage is changed, it will be reproduced,
-                # which tells us that we should force reproducing all of
-                # the other stages down below, even if their direct
-                # dependencies didn't change.
-                kwargs["force"] = True
+    steps = stages
+    if pipeline or all_pipelines or not single_item:
+        graph = self.index.graph
+        steps = plan_repro(
+            graph,
+            stages,
+            pipeline=pipeline,
+            downstream=downstream,
+            all_pipelines=all_pipelines,
+        )
+    return _reproduce_stages(steps, **kwargs)
 
-            if ret:
-                result.extend(ret)
+
+def _reproduce_stages(
+    stages: List["Stage"],
+    force_downstream: bool = False,
+    **kwargs,
+) -> List["Stage"]:
+    result: List["Stage"] = []
+    for i, stage in enumerate(stages):
+        try:
+            ret = _reproduce_stage(stage, upstream=stages[:i], **kwargs)
         except Exception as exc:  # noqa: BLE001
             raise ReproductionError(stage.addressing) from exc
 
-    if on_unchanged is not None:
-        on_unchanged(unchanged)
+        if not ret:
+            continue
+
+        result.append(ret)
+        if force_downstream:
+            # NOTE: we are walking our pipeline from the top to the
+            # bottom. If one stage is changed, it will be reproduced,
+            # which tells us that we should force reproducing all of
+            # the other stages down below, even if their direct
+            # dependencies didn't change.
+            kwargs["force"] = True
+        if i < len(stages) - 1:
+            logger.info("")  # add a newline
     return result
-
-
-def _get_steps(graph, stages, downstream, single_item):
-    import networkx as nx
-
-    active = graph.copy()
-    if not single_item:
-        # NOTE: frozen stages don't matter for single_item
-        for stage in graph:
-            if stage.frozen:
-                # NOTE: disconnect frozen stage from its dependencies
-                active.remove_edges_from(graph.out_edges(stage))
-
-    all_pipelines: List["Stage"] = []
-    for stage in stages:
-        if downstream:
-            # NOTE (py3 only):
-            # Python's `deepcopy` defaults to pickle/unpickle the object.
-            # Stages are complex objects (with references to `repo`,
-            # `outs`, and `deps`) that cause struggles when you try
-            # to serialize them. We need to create a copy of the graph
-            # itself, and then reverse it, instead of using
-            # graph.reverse() directly because it calls `deepcopy`
-            # underneath -- unless copy=False is specified.
-            nodes = nx.dfs_postorder_nodes(active.reverse(copy=False), stage)
-            all_pipelines += reversed(list(nodes))
-        else:
-            all_pipelines += nx.dfs_postorder_nodes(active, stage)
-
-    steps = []
-    for stage in all_pipelines:
-        if stage not in steps:
-            # NOTE: order of steps still matters for single_item
-            if single_item and stage not in stages:
-                continue
-
-            steps.append(stage)
-
-    return steps
-
-
-def _repro_callback(experiments_callback, unchanged, stages):
-    experiments_callback(unchanged, stages)

dvc/stage/cache.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
 from funcy import first
 
 from dvc import fs
-from dvc.exceptions import DvcException
+from dvc.exceptions import CollectCacheError, DvcException
 from dvc.utils import dict_sha256, relpath
 
 if TYPE_CHECKING:
@@ -179,7 +179,7 @@ class StageCache:
         dump_yaml(tmp, cache)
         self.repo.cache.legacy.move(tmp, path)
 
-    def restore(self, stage, run_cache=True, pull=False, dry=False):
+    def restore(self, stage, run_cache=True, pull=False, dry=False):  # noqa: C901
         from .serialize import to_single_stage_lockfile
 
         if not _can_hash(stage):
@@ -203,8 +203,11 @@ class StageCache:
         cached_stage = self._create_stage(cache, wdir=stage.wdir)
 
         if pull and not dry:
-            for objs in cached_stage.get_used_objs().values():
-                self.repo.cloud.pull(objs)
+            try:
+                for objs in cached_stage.get_used_objs().values():
+                    self.repo.cloud.pull(objs)
+            except CollectCacheError as exc:
+                raise RunCacheNotFoundError(stage) from exc
 
         if not cached_stage.outs_cached():
             raise RunCacheNotFoundError(stage)

dvc/stage/utils.py

@@ -280,3 +280,27 @@ def validate_kwargs(
         kwargs.pop("name", None)
 
     return kwargs
+
+
+def _get_stage_files(stage: "Stage") -> List[str]:
+    from dvc.dvcfile import ProjectFile
+    from dvc.utils import relpath
+
+    ret: List[str] = []
+    file = stage.dvcfile
+    ret.append(file.relpath)
+    if isinstance(file, ProjectFile):
+        ret.append(file._lockfile.relpath)  # pylint: disable=protected-access
+
+    for dep in stage.deps:
+        if (
+            not dep.use_scm_ignore
+            and dep.is_in_repo
+            and not stage.repo.dvcfs.isdvc(stage.repo.dvcfs.from_os_path(str(dep)))
+        ):
+            ret.append(relpath(dep.fs_path))
+
+    for out in stage.outs:
+        if not out.use_scm_ignore and out.is_in_repo:
+            ret.append(relpath(out.fs_path))
+    return ret

pyproject.toml

@@ -34,13 +34,14 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.3.0,<2.4.0",
+    "dvc-data>=2.3.2,<2.4.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.9.2,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
-    "flufl.lock>=5",
+    # https://github.com/iterative/dvc/issues/9654
+    "flufl.lock>=5,<8",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
     "hydra-core>=1.1",
@@ -82,6 +83,7 @@ lint = [
     "types-tabulate",
     "types-toml",
     "types-tqdm",
+    "types-pyinstaller",
 ]
 oss = ["dvc-oss==2.19"]
 s3 = ["dvc-s3==2.23.0"]
@@ -128,6 +130,10 @@ dvc = "dvc.api:DVCFileSystem"
 [project.entry-points."pytest11"]
 dvc-testing = "dvc.testing.plugin"
 
+[project.entry-points."pyinstaller40"]
+hook-dirs = "dvc.__pyinstaller:get_hook_dirs"
+tests = "dvc.__pyinstaller:get_PyInstaller_tests"
+
 [tool.setuptools]
 license-files = ["LICENSE"]
 
@@ -292,11 +298,10 @@ ignore-names = ["M", "SCM"]
 "dvc/commands/**" = ["N806"]
 "dvc/testing/**" = ["ARG002"]
 "dvc/testing/benchmarks/**" = ["ARG001"]
-"scripts/**" = ["T201", "INP001", "TRY002"]
 
 [tool.ruff.pylint]
 max-args = 10
 
 [tool.bandit]
-exclude_dirs = ["tests", "scripts", "dvc/repo/experiments/utils.py"]
+exclude_dirs = ["tests", "dvc/repo/experiments/utils.py"]
 skips = ["B101"]

scripts/build-requirements.txt

@@ -1,3 +0,0 @@
-# NOTE: using strict version due to
-# https://github.com/iterative/dvc/issues/7949
-PyInstaller==5.9.0

scripts/build_package.sh

@@ -1,15 +0,0 @@
-#!/bin/bash
-
-set -e
-set -x
-
-if [ ! -d "dvc" ]; then
-  echo "Please run this script from repository root" >&2
-  exit 1
-fi
-
-echo 'PKG = "pip"' >dvc/_build.py
-
-python -m pip install build twine
-python -m build
-twine check dist/*

scripts/pyinstaller/.gitignore

@@ -1,4 +0,0 @@
-/build
-/dist
-/dvc.spec
-/dvc

scripts/pyinstaller/build.py

@@ -1,47 +0,0 @@
-import os
-import pathlib
-from subprocess import STDOUT, check_call, check_output
-
-path = pathlib.Path(__file__).parent.absolute()
-hooks = path / "hooks"
-dvc = path.parent.parent / "dvc"
-entry = dvc / "__main__.py"
-
-check_call(
-    [
-        "pyinstaller",
-        "--additional-hooks-dir",
-        os.fspath(hooks),
-        "--name",
-        "dvc",
-        "-y",
-        os.fspath(entry),
-    ],
-    cwd=path,
-    stderr=STDOUT,
-)
-
-out = check_output(
-    [
-        path / "dist" / "dvc" / "dvc",
-        "doctor",
-    ],
-    stderr=STDOUT,
-).decode()
-
-remotes = [
-    "s3",
-    "oss",
-    "gdrive",
-    "gs",
-    "hdfs",
-    "http",
-    "webhdfs",
-    "azure",
-    "ssh",
-    "webdav",
-]
-
-print(out)
-for remote in remotes:
-    assert f"\t{remote}" in out, f"Missing support for {remote}"

scripts/pyinstaller/hooks/hook-dvc.system.py

@@ -1,4 +0,0 @@
-import os
-
-if os.name == "nt":
-    hiddenimports = ["win32file"]

scripts/pyinstaller/hooks/hook-dvc.tree.gs.py

@@ -1,9 +0,0 @@
-from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
-
-datas = copy_metadata("google-cloud-storage")
-
-# NOTE: including "requests" metadata because of google-resumable-media
-# bug. See [1] for more info.
-#
-# [1] https://github.com/googleapis/google-resumable-media-python/issues/59
-datas += copy_metadata("requests")

scripts/pyinstaller/hooks/hook-google_compute_engine.logger.py

@@ -1,4 +0,0 @@
-from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
-
-datas = copy_metadata("google-compute-engine")
-hiddenimports = ["google_cloud_engine"]

tests/func/repro/test_repro.py

@@ -1282,3 +1282,17 @@ def test_repro_rm_recursive(tmp_dir, dvc):
     dvc.reproduce()
     assert (tmp_dir / "dir").exists()
     assert not (tmp_dir / "dir" / "foo").exists()
+
+
+def test_repro_single_item_with_multiple_targets(tmp_dir, dvc, copy_script):
+    stage1 = dvc.stage.add(cmd="echo foo > foo", outs=["foo"], name="gen-foo")
+    with dvc.lock:
+        stage1.run()
+
+    stage2 = dvc.stage.add(
+        cmd="python copy.py foo bar", deps=["foo"], outs=["bar"], name="copy-foo-bar"
+    )
+    assert dvc.reproduce(["copy-foo-bar", "gen-foo"], single_item=True) == [
+        stage2,
+        stage1,
+    ]

tests/unit/command/test_data_sync.py

@@ -18,6 +18,8 @@ def test_fetch(mocker, dvc):
             "--with-deps",
             "--recursive",
             "--run-cache",
+            "--max-size",
+            "10",
         ]
     )
     assert cli_args.func == CmdDataFetch
@@ -37,6 +39,7 @@ def test_fetch(mocker, dvc):
         with_deps=True,
         recursive=True,
         run_cache=True,
+        max_size=10,
     )
 
 

tests/unit/repo/test_graph.py

@@ -0,0 +1,69 @@
+import pytest
+from networkx import DiGraph
+from networkx.utils import graphs_equal
+
+from dvc.repo.graph import get_steps, get_subgraph_of_nodes
+
+
+@pytest.mark.parametrize(
+    "nodes,downstream,expected_edges",
+    [
+        ([], False, {1: [2, 3], 2: [4, 5], 3: [6, 7], 8: [9]}),
+        ([1], False, {1: [2, 3], 2: [4, 5], 3: [6, 7]}),
+        ([2], False, {2: [4, 5]}),
+        ([3], False, {3: [6, 7]}),
+        ([8], False, [(8, 9)]),
+        ([2, 3, 8], False, {2: [4, 5], 3: [6, 7], 8: [9]}),
+        ([4], False, {4: []}),
+        ([], True, {1: [2, 3], 2: [4, 5], 3: [6, 7], 8: [9]}),
+        ([1], True, {1: []}),
+        ([9], True, [(8, 9)]),
+        ([2], True, [(1, 2)]),
+        ([6], True, [(1, 3), (3, 6)]),
+        ([2, 3, 8], True, {1: [2, 3], 8: []}),
+        ([4, 7], True, {1: [2, 3], 2: [4], 3: [7]}),
+    ],
+)
+def test_subgraph_of_nodes(nodes, downstream, expected_edges):
+    r"""
+             1
+           /   \
+          2     3      8
+         / \   / \     |
+        4   5 6   7    9
+    """
+    graph = DiGraph({1: [2, 3], 2: [4, 5], 3: [6, 7], 8: [9]})
+    subgraph = get_subgraph_of_nodes(graph, nodes, downstream=downstream)
+    expected = DiGraph(expected_edges)
+    assert graphs_equal(expected, subgraph)
+
+
+@pytest.mark.parametrize(
+    "nodes,downstream,expected_steps",
+    [
+        ([], False, [4, 5, 2, 6, 7, 3, 1, 9, 8]),
+        ([1], False, [4, 5, 2, 6, 7, 3, 1]),
+        ([2], False, [4, 5, 2]),
+        ([3], False, [6, 7, 3]),
+        ([8], False, [9, 8]),
+        ([2, 3, 8], False, [4, 5, 2, 6, 7, 3, 9, 8]),
+        ([4], False, [4]),
+        ([], True, [4, 5, 2, 6, 7, 3, 1, 9, 8]),
+        ([1], True, [1]),
+        ([9], True, [9, 8]),
+        ([2], True, [2, 1]),
+        ([6], True, [6, 3, 1]),
+        ([2, 3, 8], True, [8, 2, 3, 1]),
+        ([4, 7], True, [4, 2, 7, 3, 1]),
+    ],
+)
+def test_steps(nodes, downstream, expected_steps):
+    r"""
+             1
+           /   \
+          2     3      8
+         / \   / \     |
+        4   5 6   7    9
+    """
+    graph = DiGraph({1: [2, 3], 2: [4, 5], 3: [6, 7], 8: [9]})
+    assert get_steps(graph, nodes, downstream=downstream) == expected_steps

tests/unit/repo/test_reproduce.py

@@ -1,6 +1,8 @@
-import os
+from itertools import chain
 
-from dvc.repo.reproduce import _get_stage_files
+from networkx import DiGraph
+
+from dvc.repo.reproduce import plan_repro
 
 
 def test_number_reproduces(tmp_dir, dvc, mocker):
@@ -21,40 +23,48 @@ def test_number_reproduces(tmp_dir, dvc, mocker):
     assert reproduce_stage_mock.call_count == 5
 
 
-def test_get_stage_files(tmp_dir, dvc):
-    tmp_dir.dvc_gen("dvc-dep", "dvc-dep")
-    tmp_dir.gen("other-dep", "other-dep")
+def test_repro_plan(mocker):
+    r"""
+             n1
+           /   \
+          n2    m3    n8
+         / \   / \     |
+        n4 n5 n6 n7    n9
+    """
 
-    stage = dvc.stage.add(
-        name="stage",
-        cmd="foo",
-        deps=["dvc-dep", "other-dep"],
-        outs=["dvc-out"],
-        outs_no_cache=["other-out"],
-    )
-    result = set(_get_stage_files(stage))
-    assert result == {
-        stage.dvcfile.relpath,
-        str(tmp_dir / "other-dep"),
-        str(tmp_dir / "other-out"),
-    }
+    # note: downstream steps may not be stable, use AnyOf in such cases
+    class AnyOf:
+        def __init__(self, *items):
+            self.items = items
 
+        def __eq__(self, other: object) -> bool:
+            return any(item == other for item in self.items)
 
-def test_get_stage_files_wdir(tmp_dir, dvc):
-    tmp_dir.gen({"dir": {"dvc-dep": "dvc-dep", "other-dep": "other-dep"}})
-    dvc.add(os.path.join("dir", "dvc-dep"))
+    n = mocker.sentinel
+    n1, n2, n3, n4, n5, n6, n7, n8, n9 = (getattr(n, f"n{i}") for i in range(1, 10))
+    edges = {n1: [n2, n3], n2: [n4, n5], n3: [n6, n7], n8: [n9]}
+    for node in chain.from_iterable([n, *v] for n, v in edges.items()):
+        node.frozen = False
 
-    stage = dvc.stage.add(
-        name="stage",
-        cmd="foo",
-        wdir="dir",
-        deps=["dvc-dep", "other-dep"],
-        outs=["dvc-out"],
-        outs_no_cache=["other-out"],
-    )
-    result = set(_get_stage_files(stage))
-    assert result == {
-        stage.dvcfile.relpath,
-        str(tmp_dir / "dir" / "other-dep"),
-        str(tmp_dir / "dir" / "other-out"),
-    }
+    g = DiGraph(edges)
+    assert plan_repro(g) == [n4, n5, n2, n6, n7, n3, n1, n9, n8]
+    assert plan_repro(g, [n1]) == [n4, n5, n2, n6, n7, n3, n1]
+    assert plan_repro(g, [n4], downstream=True) == [n4, n2, n1]
+    assert plan_repro(g, [n8], True) == plan_repro(g, [n9], True) == [n9, n8]
+    assert plan_repro(g, [n2, n8], True) == [n4, n5, n2, n6, n7, n3, n1, n9, n8]
+    assert plan_repro(g, [n2, n3], downstream=True) == [
+        AnyOf(n2, n3),
+        AnyOf(n2, n3),
+        n1,
+    ]
+
+    n2.frozen = True
+    assert plan_repro(g) == [n2, n6, n7, n3, n1, n9, n8, n4, n5]
+    assert plan_repro(g, [n1]) == [n2, n6, n7, n3, n1]
+    assert plan_repro(g, [n4], downstream=True) == [n4]
+    assert plan_repro(g, [n2, n8], pipeline=True) == [n2, n6, n7, n3, n1, n9, n8]
+    assert plan_repro(g, [n2, n3], downstream=True) == [
+        AnyOf(n2, n3),
+        AnyOf(n2, n3),
+        n1,
+    ]

tests/unit/stage/test_utils.py

@@ -1,7 +1,7 @@
 import os
 
 from dvc.fs import localfs
-from dvc.stage.utils import resolve_paths
+from dvc.stage.utils import _get_stage_files, resolve_paths
 
 
 def test_resolve_paths():
@@ -19,3 +19,40 @@ def test_resolve_paths():
     path, wdir = resolve_paths(fs=localfs, path=file_path, wdir="../../some-dir")
     assert path == os.path.abspath(file_path)
     assert wdir == os.path.abspath("some-dir")
+
+
+def test_get_stage_files(tmp_dir, dvc):
+    tmp_dir.dvc_gen("dvc-dep", "dvc-dep")
+    tmp_dir.gen("other-dep", "other-dep")
+    stage = dvc.stage.create(
+        name="stage",
+        cmd="foo",
+        deps=["dvc-dep", "other-dep"],
+        outs=["dvc-out"],
+        outs_no_cache=["other-out"],
+    )
+    assert _get_stage_files(stage) == [
+        "dvc.yaml",
+        "dvc.lock",
+        "other-dep",
+        "other-out",
+    ]
+
+
+def test_get_stage_files_wdir(tmp_dir, dvc):
+    tmp_dir.gen({"dir": {"dvc-dep": "dvc-dep", "other-dep": "other-dep"}})
+    dvc.add(os.path.join("dir", "dvc-dep"))
+    stage = dvc.stage.create(
+        name="stage",
+        cmd="foo",
+        wdir="dir",
+        deps=["dvc-dep", "other-dep"],
+        outs=["dvc-out"],
+        outs_no_cache=["other-out"],
+    )
+    assert _get_stage_files(stage) == [
+        "dvc.yaml",
+        "dvc.lock",
+        os.path.join("dir", "other-dep"),
+        os.path.join("dir", "other-out"),
+    ]