diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py
index 209d193fee..5bdfc9acdb 100644
--- a/dvc/repo/fetch.py
+++ b/dvc/repo/fetch.py
@@ -1,4 +1,4 @@
-from typing import List, Tuple
+from typing import TYPE_CHECKING, List, Tuple
 
 from dvc.exceptions import DownloadError
 from dvc.log import logger
@@ -7,6 +7,10 @@
 
 from . import locked
 
+if TYPE_CHECKING:
+    from dvc.output import Output
+    from dvc.stage import Stage
+
 logger = logger.getChild(__name__)
 
 
@@ -32,6 +36,7 @@ def _collect_indexes(  # noqa: PLR0913
     types=None,
     config=None,
     onerror=None,
+    push=False,
 ):
     indexes = {}
     collection_exc = None
@@ -42,6 +47,16 @@ def _collect_indexes(  # noqa: PLR0913
         core["remote"] = remote
         config["core"] = core
 
+    def stage_filter(stage: "Stage") -> bool:
+        if push and stage.is_repo_import:
+            return False
+        return True
+
+    def outs_filter(out: "Output") -> bool:
+        if push and not out.can_push:
+            return False
+        return True
+
     for rev in repo.brancher(
         revs=revs,
         all_branches=all_branches,
@@ -57,6 +72,8 @@ def _collect_indexes(  # noqa: PLR0913
                 recursive=recursive,
                 max_size=max_size,
                 types=types,
+                stage_filter=stage_filter,
+                outs_filter=outs_filter,
             )
 
             idx.data["repo"].onerror = _make_index_onerror(onerror, rev)
diff --git a/dvc/repo/push.py b/dvc/repo/push.py
index e93c52c86c..5a14181577 100644
--- a/dvc/repo/push.py
+++ b/dvc/repo/push.py
@@ -87,6 +87,7 @@ def push(  # noqa: PLR0913
         recursive=recursive,
         all_commits=all_commits,
         revs=revs,
+        push=True,
     )
 
     cache_key = (
@@ -107,6 +108,7 @@ def push(  # noqa: PLR0913
             push=True,
         )
 
+    push_transferred, push_failed = 0, 0
     try:
         with ui.progress(
             desc="Pushing",
@@ -131,6 +133,11 @@ def push(  # noqa: PLR0913
         for fs_index in data:
             fs_index.close()
 
+        if push_transferred:
+            # NOTE: dropping cached index to force reloading from newly saved
+            # metadata from version-aware remotes
+            self.drop_data_index()
+
     transferred_count += push_transferred
     failed_count += push_failed
     if failed_count:
diff --git a/dvc/testing/remote_tests.py b/dvc/testing/remote_tests.py
index a33ccefef2..4294b2bd1e 100644
--- a/dvc/testing/remote_tests.py
+++ b/dvc/testing/remote_tests.py
@@ -172,6 +172,8 @@ def test_file(self, tmp_dir, dvc, run_copy, remote_version_aware):
         run_copy("foo", "foo_copy", name="copy")
 
         dvc.push()
+        assert (remote_version_aware / "foo").read_text() == "foo"
+        assert (remote_version_aware / "foo_copy").read_text() == "foo"
         foo_dvc = (tmp_dir / "foo.dvc").read_text()
         assert "version_id" in foo_dvc
         stage = stage.reload()
@@ -181,18 +183,24 @@ def test_file(self, tmp_dir, dvc, run_copy, remote_version_aware):
 
         remove(dvc.cache.local.path)
         remove(tmp_dir / "foo")
+        remove(tmp_dir / "foo_copy")
 
         dvc.pull()
         assert (tmp_dir / "foo").read_text() == "foo"
+        assert (tmp_dir / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.push()
+        assert (remote_version_aware / "foo").read_text() == "foo"
+        assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.reproduce()
         dvc.push()
+        assert (remote_version_aware / "foo").read_text() == "foo"
+        assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
@@ -241,6 +249,12 @@ def test_dir(self, tmp_dir, dvc, run_copy, remote_version_aware):
         assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
+        dvc.cache.local.clear()
+        remove(tmp_dir / "data_dir")
+        dvc.push()
+        assert (remote_version_aware / "data_dir").exists()
+        assert (remote_version_aware / "data_dir" / "data").exists()
+
 
 class TestRemoteWorktree:
     def test_file(self, tmp_dir, dvc, remote_worktree):
diff --git a/pyproject.toml b/pyproject.toml
index 7fce5bc855..b7eda8f89c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -34,7 +34,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.20.0,<2.21.0",
+    "dvc-data>=2.21.2,<2.22.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.13.0,<1",
diff --git a/tests/func/api/test_data.py b/tests/func/api/test_data.py
index 947d5f48d8..8700cd2d70 100644
--- a/tests/func/api/test_data.py
+++ b/tests/func/api/test_data.py
@@ -41,7 +41,7 @@ def test_open_external(tmp_dir, erepo_dir, cloud):
             # NOTE: need file to be other size for Mac
             erepo_dir.dvc_gen("version", "branchver", commit="add version")
 
-    erepo_dir.dvc.push(all_branches=True)
+    assert erepo_dir.dvc.push(all_branches=True) == 2
 
     # Remove cache to force download
     remove(erepo_dir.dvc.cache.local.path)
diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py
index 190bc9ae3e..89fdb9e5c7 100644
--- a/tests/func/test_data_cloud.py
+++ b/tests/func/test_data_cloud.py
@@ -486,7 +486,7 @@ def test_pull_partial(tmp_dir, dvc, local_remote):
     clean(["foo"], dvc)
 
     stats = dvc.pull(os.path.join("foo", "bar"))
-    assert stats["fetched"] == 2
+    assert stats["fetched"] == 3
     assert (tmp_dir / "foo").read_text() == {"bar": {"baz": "baz"}}
 
 
diff --git a/tests/func/test_virtual_directory.py b/tests/func/test_virtual_directory.py
index a7f8aef53c..b609e5718b 100644
--- a/tests/func/test_virtual_directory.py
+++ b/tests/func/test_virtual_directory.py
@@ -182,7 +182,7 @@ def test_partial_checkout_and_update(M, tmp_dir, dvc, remote):
 
     assert dvc.pull("dir/subdir") == M.dict(
         added=[join("dir", "")],
-        fetched=2,
+        fetched=3,
     )
     assert (tmp_dir / "dir").read_text() == {"subdir": {"lorem": "lorem"}}