exp run: if target is provided, only checkout target and dependencies (#5519)

* tests: test that only target and its deps are checked out on exp run

* exp run: if target is provided, only checkout target + deps

dvc/repo/experiments/executor/base.py

@@ -248,6 +248,10 @@ class BaseExecutor(ABC):
 
         with cls._repro_dvc(dvc_dir, rel_cwd) as dvc:
             args, kwargs = cls._repro_args(dvc)
+            if args:
+                targets: Optional[Union[list, str]] = args[0]
+            else:
+                targets = kwargs.get("targets")
 
             repro_force = kwargs.get("force", False)
             logger.trace(  # type: ignore[attr-defined]
@@ -265,7 +269,14 @@ class BaseExecutor(ABC):
             #   be removed/does not yet exist) so that our executor workspace
             #   is not polluted with the (persistent) out from an unrelated
             #   experiment run
-            dvc_checkout(dvc, force=True, quiet=True, allow_missing=True)
+            dvc_checkout(
+                dvc,
+                targets=targets,
+                with_deps=targets is not None,
+                force=True,
+                quiet=True,
+                allow_missing=True,
+            )
 
             checkpoint_func = partial(
                 cls.checkpoint_callback, dvc.scm, name, repro_force

tests/func/experiments/test_experiments.py

@@ -612,3 +612,24 @@ def test_remove(tmp_dir, scm, dvc, exp_stage):
     removed = dvc.experiments.remove(queue=True)
     assert removed == 1
     assert len(dvc.experiments.stash) == 0
+
+
+def test_checkout_targets_deps(tmp_dir, scm, dvc, exp_stage):
+    from dvc.utils.fs import remove
+
+    tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"}, commit="add files")
+    stage = dvc.stage.add(
+        cmd="python copy.py params.yaml metrics.yaml",
+        metrics_no_cache=["metrics.yaml"],
+        params=["foo"],
+        name="copy-file",
+        deps=["copy.py", "foo"],
+        force=True,
+    )
+    remove("foo")
+    remove("bar")
+
+    dvc.experiments.run(stage.addressing, params=["foo=2"])
+    assert (tmp_dir / "foo").exists()
+    assert (tmp_dir / "foo").read_text() == "foo"
+    assert not (tmp_dir / "bar").exists()