do not checkout partials imports on fetch

dvc/repo/fetch.py

@@ -127,8 +127,16 @@ def _fetch_partial_imports(repo, targets, **kwargs):
     downloaded = 0
     failed = 0
     for stage in repo.partial_imports(targets, **kwargs):
+        dep = stage.deps[0]
+        out = stage.outs[0]
         try:
-            stage.run()
+            if dep.changed_checksum():
+                raise DataSourceChanged(f"{stage} ({dep})")
+
+            logger.info("Importing '%s' -> '%s'", dep, out)
+            out.transfer(
+                dep.def_path, jobs=kwargs.get("jobs"), odb=repo.odb.local
+            )
         except DataSourceChanged as exc:
             logger.warning(f"{exc}")
             failed += 1

dvc/stage/__init__.py

@@ -554,9 +554,8 @@ class Stage(params.StageParams):
             self.remove_outs(ignore_remove=False, force=False)
 
         if (not self.frozen and self.is_import) or self.is_partial_import:
-            jobs = kwargs.get("jobs", None)
             self._sync_import(
-                dry, force, jobs, no_download, check_changed=self.frozen
+                dry, force, kwargs.get("jobs", None), no_download
             )
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
@@ -580,10 +579,8 @@ class Stage(params.StageParams):
         return run_stage(self, dry, force, **kwargs)
 
     @rwlocked(read=["deps"], write=["outs"])
-    def _sync_import(
-        self, dry, force, jobs, no_download, check_changed: bool = False
-    ):
-        sync_import(self, dry, force, jobs, no_download, check_changed)
+    def _sync_import(self, dry, force, jobs, no_download):
+        sync_import(self, dry, force, jobs, no_download)
 
     @rwlocked(read=["outs"])
     def _check_missing_outputs(self):

dvc/stage/imports.py

@@ -46,7 +46,6 @@ def sync_import(
     force=False,
     jobs=None,
     no_download=False,
-    check_changed=False,
 ):
     """Synchronize import's outs to the workspace."""
     logger.info("Importing '%s' -> '%s'", stage.deps[0], stage.outs[0])
@@ -56,13 +55,9 @@ def sync_import(
     if not force and stage.already_cached():
         stage.outs[0].checkout()
     else:
-        if check_changed:
-            old_hash_info = stage.deps[0].hash_info
         stage.save_deps()
-        if check_changed and not old_hash_info == stage.deps[0].hash_info:
-            from dvc.stage.exceptions import DataSourceChanged
-
-            raise DataSourceChanged(f"{stage} ({stage.deps[0]})")
-
         if not no_download:
-            stage.deps[0].download(stage.outs[0], jobs=jobs)
+            stage.deps[0].download(
+                stage.outs[0],
+                jobs=jobs,
+            )