cloud versioning: fix push/fetch performance

dvc/repo/fetch.py

@@ -78,6 +78,7 @@ def fetch(
                 all_tags=all_tags,
                 all_commits=all_commits,
                 targets=targets,
+                jobs=jobs,
                 with_deps=with_deps,
                 recursive=recursive,
             )
@@ -181,6 +182,7 @@ def _fetch_worktree(
     all_tags: bool = False,
     all_commits: bool = False,
     targets: Optional["TargetType"] = None,
+    jobs: Optional[int] = None,
     **kwargs,
 ) -> int:
     from dvc.repo.worktree import fetch_worktree
@@ -192,5 +194,7 @@ def _fetch_worktree(
         all_tags=all_tags,
         all_commits=all_commits,
     ):
-        downloaded += fetch_worktree(repo, remote, targets=targets, **kwargs)
+        downloaded += fetch_worktree(
+            repo, remote, targets=targets, jobs=jobs, **kwargs
+        )
     return downloaded

dvc/repo/push.py

@@ -55,6 +55,7 @@ def push(
             all_tags=all_tags,
             all_commits=all_commits,
             targets=expanded_targets,
+            jobs=jobs,
             with_deps=with_deps,
             recursive=recursive,
         )
@@ -105,6 +106,7 @@ def _push_worktree(
     all_tags: bool = False,
     all_commits: bool = False,
     targets: Optional["TargetType"] = None,
+    jobs: Optional[int] = None,
     **kwargs,
 ) -> int:
     from dvc.repo.worktree import push_worktree
@@ -114,4 +116,4 @@ def _push_worktree(
             "Multiple rev push is unsupported for cloud versioned remotes"
         )
 
-    return push_worktree(repo, remote, targets=targets, **kwargs)
+    return push_worktree(repo, remote, targets=targets, jobs=jobs, **kwargs)

dvc/repo/worktree.py

@@ -32,7 +32,6 @@ def worktree_view(
     index: "Index",
     targets: Optional["TargetType"] = None,
     push: bool = False,
-    latest_only: bool = True,
     **kwargs,
 ) -> "IndexView":
     """Return view of data that can be stored in worktree remotes.
@@ -56,10 +55,6 @@ def outs_filter(out: "Output") -> bool:
             or (push and not out.can_push)
         ):
             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:
-            return False
         return True
 
     return index.targets_view(
@@ -74,6 +69,7 @@ def fetch_worktree(
     repo: "Repo",
     remote: "Remote",
     targets: Optional["TargetType"] = None,
+    jobs: Optional[int] = None,
     **kwargs,
 ) -> int:
     from dvc_data.index import save
@@ -91,13 +87,14 @@ def fetch_worktree(
         unit="file", desc="Fetch", disable=total == 0
     ) as cb:
         cb.set_size(total)
-        return save(index, callback=cb)
+        return save(index, callback=cb, jobs=jobs)
 
 
 def push_worktree(
     repo: "Repo",
     remote: "Remote",
     targets: Optional["TargetType"] = None,
+    jobs: Optional[int] = None,
     **kwargs,
 ) -> int:
     from dvc_data.index import checkout
@@ -106,7 +103,6 @@ def push_worktree(
         repo.index,
         push=True,
         targets=targets,
-        latest_only=remote.worktree,
         **kwargs,
     )
     new_index = view.data["repo"]
@@ -139,6 +135,7 @@ def push_worktree(
             delete=remote.worktree,
             callback=cb,
             latest_only=remote.worktree,
+            jobs=jobs,
             **diff_kwargs,
         )
     if pushed: