checkout: delete output if we can't create it fully (#9642)

dvc/repo/checkout.py

@@ -161,19 +161,23 @@ def checkout(  # noqa: C901
     if not force:
         _check_can_delete(diff.files_delete, new, self.root_dir, self.fs)
 
-    failed = []
+    failed = set()
+    out_paths = [out.fs_path for out in view.outs if out.use_cache and out.is_in_repo]
 
     def checkout_onerror(src_path, dest_path, _exc):
         logger.debug(
             "failed to create {%s} from {%s}", dest_path, src_path, exc_info=True
         )
-        failed.append(dest_path)
+
+        for out_path in out_paths:
+            if self.fs.path.isin_or_eq(dest_path, out_path):
+                failed.add(out_path)
 
     with Callback.as_tqdm_callback(
         unit="file",
         desc="Checkout",
     ) as cb:
-        changes = apply(
+        apply(
             diff,
             self.root_dir,
             self.fs,
@@ -184,15 +188,19 @@ def checkout(  # noqa: C901
             **kwargs,
         )
 
-    out_changes = _build_out_changes(view, changes)
+    out_changes = _build_out_changes(view, diff.changes)
 
     typ_map = {ADD: "added", DELETE: "deleted", MODIFY: "modified"}
     for key, typ in out_changes.items():
         out_path = self.fs.path.join(self.root_dir, *key)
-        self.state.save_link(out_path, self.fs)
-        stats[typ_map[typ]].append(_fspath_dir(out_path))
+
+        if out_path in failed:
+            self.fs.remove(out_path, recursive=True)
+        else:
+            self.state.save_link(out_path, self.fs)
+            stats[typ_map[typ]].append(_fspath_dir(out_path))
 
     if failed and not allow_missing:
-        raise CheckoutError(failed, stats)
+        raise CheckoutError([relpath(out_path) for out_path in failed], stats)
 
     return stats

dvc/repo/worktree.py

@@ -158,7 +158,7 @@ def push_worktree(
         ) as cb:
             cb.set_size(total)
             try:
-                stats = apply(
+                apply(
                     diff,
                     remote_obj.path,
                     remote_obj.fs,
@@ -166,7 +166,7 @@ def push_worktree(
                     latest_only=remote_obj.worktree,
                     jobs=jobs,
                 )
-                pushed += sum(len(changes) for changes in stats.values())
+                pushed += sum(len(changes) for changes in diff.changes.values())
             except VersioningNotSupported:
                 logger.exception("")
                 raise DvcException(

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.2.0,<2.3.0",
+    "dvc-data>=2.3.0,<2.4.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.9.2,<1",