push: assume version_aware push should skip existing version_id

dvc/repo/worktree.py

@@ -58,7 +58,7 @@ def outs_filter(out: "Output") -> bool:
             return False
         # If we are not enforcing push to latest version and have a version
         # for this out, we assume it still exists and can skip pushing it
-        if push and not latest_only and out.meta.version_id is not None:
+        if push and not latest_only and _has_version_id(out):
             return False
         return True
 
@@ -70,6 +70,13 @@ def outs_filter(out: "Output") -> bool:
     )
 
 
+def _has_version_id(out: "Output"):
+    if not out.is_dir_checksum:
+        return out.meta.version_id is not None
+    tree = cast("Tree", out.get_obj())
+    return all(meta.version_id is not None for _, meta, _ in tree)
+
+
 def fetch_worktree(
     repo: "Repo",
     remote: "Remote",
@@ -102,7 +109,7 @@ def push_worktree(
     jobs: Optional[int] = None,
     **kwargs,
 ) -> int:
-    from dvc_data.index import checkout
+    from dvc_data.index import DataIndex, checkout
 
     view = worktree_view(
         repo.index,
@@ -111,7 +118,16 @@ def push_worktree(
         latest_only=remote.worktree,
         **kwargs,
     )
-    new_index = view.data["repo"]
+    if remote.worktree:
+        new_index = view.data["repo"]
+    else:
+        # We assume that if we have a version_id already, the specified object
+        # version still exists on the remote.
+        # See: https://github.com/iterative/dvc/issues/8359
+        new_index = DataIndex()
+        for key, entry in view.data["repo"].iteritems():
+            if entry.meta.version_id is None:
+                new_index[key] = entry
     if remote.worktree:
         logger.debug("indexing latest worktree for '%s'", remote.path)
         old_index = _build_worktree_index(repo, remote, view)