fetch: handle missing/modified imports

dvc/fs/dvc.py

@@ -532,6 +532,15 @@ def _prepare_credentials(self, **config) -> Dict[str, Any]:
     def fs(self) -> "_DVCFileSystem":
         return _DVCFileSystem(**self.fs_args)
 
+    @property
+    def immutable(self):
+        from dvc.scm import NoSCM
+
+        if isinstance(self.fs.repo.scm, NoSCM):
+            return False
+
+        return self.fs._repo_kwargs.get("rev") == self.fs.repo.get_rev()
+
     def getcwd(self):
         return self.fs.getcwd()
 

dvc/repo/index.py

@@ -205,7 +205,9 @@ def _load_storage_from_import(storage_map, key, out):
         return
 
     dep = out.stage.deps[0]
-    if not out.hash_info:
+    if not out.hash_info and (
+        not dep.hash_info or dep.hash_info.name != storage_map[key].cache.odb.hash_name
+    ):
         # partial import
         fs_cache = out.repo.cache.fs_cache
         storage_map.add_cache(

pyproject.toml

@@ -35,7 +35,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=3.8,<3.9",
+    "dvc-data>=3.9,<3.10",
     "dvc-http>=2.29.0",
     "dvc-render>=1.0.1,<2",
     "dvc-studio-client>=0.17.1,<1",

tests/func/test_data_cloud.py

@@ -277,6 +277,28 @@ def test_pull_partial_import(tmp_dir, dvc, local_workspace):
     assert stage.outs[0].get_hash().value == "d10b4c3ff123b26dc068d43a8bef2d23"
 
 
+def test_pull_partial_import_missing(tmp_dir, dvc, local_workspace):
+    local_workspace.gen("file", "file content")
+    dst = tmp_dir / "file"
+    dvc.imp_url("remote://workspace/file", os.fspath(dst), no_download=True)
+
+    (local_workspace / "file").unlink()
+    with pytest.raises(CheckoutError):
+        dvc.pull("file")
+    assert not dst.exists()
+
+
+def test_pull_partial_import_modified(tmp_dir, dvc, local_workspace):
+    local_workspace.gen("file", "file content")
+    dst = tmp_dir / "file"
+    dvc.imp_url("remote://workspace/file", os.fspath(dst), no_download=True)
+
+    local_workspace.gen("file", "updated file content")
+    with pytest.raises(CheckoutError):
+        dvc.pull("file")
+    assert not dst.exists()
+
+
 def test_pull_external_dvc_imports_mixed(tmp_dir, dvc, scm, erepo_dir, local_remote):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen("foo", "foo", commit="first")