Repository: iterative/dvc

Issue:
drop external outputs
Drop current implementation of external outputs as part of #7093.




Discussion:
When dropping external outputs, can we continue to support `--outs-no-cache` with external paths, and can we allow them without the `--external` flag (which we can drop)?

_Originally posted by @dberenbaum in https://github.com/iterative/dvc/issues/7093#issuecomment-1567028789_
            
Hi, I posted this [question regarding external outputs on Discord](https://discord.com/channels/485586884165107732/485596304961962003/1111663148777340989) and am moving the discussion here as suggested.

I currently moved a larger scale data analysis from our local server to run in AWS using a Ray Cluster (https://www.ray.io/). The different steps are implemented as individual Ray jobs which are compiled into a reproducible pipeline with DVC using a `dvc.yaml`. Because the execution of jobs / stages is in AWS, and because the size of the data is too large to store locally, the inputs and outputs of each stage are external AWS S3 URIs. As part of this I run into one or the other issue (stage output always "not in cache" even right after `dvc commit -f` was run; error to run stage the first time because S3 URI prefix does not exist) indicating some bug in DVC in regards to external outputs (still need to file GitHub issues for these).

However, looking at this roadmap for the next major release version 3.0, I can see that there is a To-Do item that reads "drop external outputs". So my more immediately pressing questions are:
- Will AWS S3 URIs for stage inputs and outputs no longer be supported from DVC version 3.0 onwards?
- What is the recommendation for being able to run large reproducible data processing in the cloud (AWS)? All stage outputs should only live in AWS S3. Stage scripts stream the data directly from/to this remote storage.

An example `dvc.yaml` would look something like:
```
vars:
- s3_uri_prefix: s3://bucket/prefix
stages:
  stage_1:
    cmd: ray job submit --working-dir . -- python stage_1.py --output ${s3_uri_prefix}/stage_1
    deps:
    - stage_1.py
    outs:
    - ${s3_prefix}/stage_1:
        push: false
  stage_2:
    cmd: ray job submit --working-dir . -- python stage_2.py --input ${s3_uri_prefix}/stage_1 --output ${s3_uri_prefix}/stage_2
    deps:
    - stage_2.py
    - ${s3_uri_prefix}/stage_1
    outs:
    - ${s3_uri_prefix}/stage_2:
        push: false
```

I'd appreciate any pointers as to how DVC would still fit into the picture here.

_Originally posted by @aschuh-hf in https://github.com/iterative/dvc/issues/7093#issuecomment-1572024046_
            
@aschuh-hf I'm personally torn on this decision, but ultimately (as you mentioned) there are a lot of bugs and problems with the way we handle external outputs today. The goal is not to give up on the use case you mention above (it's more important than ever), but to build a better solution for ti.

Besides staying on DVC 2.x for now, you should be able to continue working with pipelines streaming data to/from s3 if you set `cache: false`. Given that your data is large, do you mostly use dvc to manage your pipeline, or do you use the cache and checkout older versions of your data?

To cache your s3 pipeline outputs, I hope we can use [cloud versioning](https://github.com/iterative/dvc/issues/8411) to help address this, so please also comment there with your thoughts. This could mean that DVC no longer has to do slow operations like caching an extra copy of your data, so I hope it will end up improving the workflow in the end.

_Originally posted by @dberenbaum in https://github.com/iterative/dvc/issues/7093#issuecomment-1572490969_
            
Thanks, @dberenbaum, for adding some more motivation for potentially deprecating it.

As a matter of fact, my current use case was for a data preprocessing and model evaluation pipeline. I am indeed mostly using the `dvc.yaml` in this context as a means to document the commands I run alongside the source code used to make it easier to reproduce the same results another time or re-running the same steps with different inputs. I don't for this purpose care too much about the caching of past results apart from the current run, and if I were to revert back to an earlier version I would be OK with spending the compute again to reproduce the outputs. Cloud versioning seems would fit also well in, though we haven't enabled this for the S3 bucket we are using for such experimental runs at the moment.

I can also confirm that the time spent on caching and re-computing hashes is quite a bottleneck given how slow it is for thousands of S3 objects. Given the use of a Ray cluster, the time waiting for DVC perform these operations is far greater than the actual execution of the individual stage commands.

It would mostly be sufficient for my use case here if `outs` and `deps` would merely define the dependency graph.

It would, however, also be useful if stages would still only re-run if the inputs, parameters, or command of a stage changed. If that can be done via version information provided by cloud versioning backends, that sounds good.

> [...] you should be able to continue working with pipelines streaming data to/from s3 if you set cache: false.

Thanks for reminding me of the `cache` option, I hadn't thought about that I could use it to bypass the duplication of the S3 objects (given the `s3cache` is configured to be within the same S3 bucket as the stage outputs).

> drop external outputs

I think when I read this, I mostly was concerned that S3 URIs would no longer be valid stage `deps` or `outs`. But if these are still supported that works for me.

_Originally posted by @aschuh-hf in https://github.com/iterative/dvc/issues/7093#issuecomment-1572809599_
            
@aschuch-hf Moved the discussion to a new issue to keep it organized. Sounds like it should work for your needs then if we can do it the way we are planning. One of the arguments in favor of dropping the current approach was to get feedback from users like you who need it so we can build something that works better.

> It would, however, also be useful if stages would still only re-run if the inputs, parameters, or command of a stage changed. If that can be done via version information provided by cloud versioning backends, that sounds good.

Yes, we should ensure this is still the case. The caching is only needed to recover prior versions of the pipeline, which as you say may often make more sense to recompute in the rare cases they are needed.

Patch:
diff --git a/dvc/cachemgr.py b/dvc/cachemgr.py
--- a/dvc/cachemgr.py
+++ b/dvc/cachemgr.py
@@ -32,13 +32,6 @@ def _get_odb(
 
 class CacheManager:
     CACHE_DIR = "cache"
-    CLOUD_SCHEMES = [
-        Schemes.S3,
-        Schemes.GS,
-        Schemes.SSH,
-        Schemes.HDFS,
-        Schemes.WEBHDFS,
-    ]
     FILES_DIR = "files"
 
     def __init__(self, repo):
@@ -91,16 +84,12 @@ def _init_odb(self, schemes):
             )
 
     def __getattr__(self, name):
-        if name not in self._odb and name in self.CLOUD_SCHEMES:
-            self._init_odb([name])
-
         try:
             return self._odb[name]
         except KeyError as exc:
             raise AttributeError from exc
 
     def by_scheme(self):
-        self._init_odb(self.CLOUD_SCHEMES)
         yield from self._odb.items()
 
     @property
diff --git a/dvc/commands/add.py b/dvc/commands/add.py
--- a/dvc/commands/add.py
+++ b/dvc/commands/add.py
@@ -24,8 +24,6 @@ def validate_args(self) -> None:
                 invalid_opt = "--glob option"
             elif args.no_commit:
                 invalid_opt = "--no-commit option"
-            elif args.external:
-                invalid_opt = "--external option"
         else:
             message = "{option} can't be used without --to-remote"
             if args.remote:
@@ -49,7 +47,6 @@ def run(self):
             self.repo.add(
                 self.args.targets,
                 no_commit=self.args.no_commit,
-                external=self.args.external,
                 glob=self.args.glob,
                 out=self.args.out,
                 remote=self.args.remote,
@@ -82,12 +79,6 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Don't put files/directories into cache.",
     )
-    parser.add_argument(
-        "--external",
-        action="store_true",
-        default=False,
-        help="Allow targets that are outside of the DVC repository.",
-    )
     parser.add_argument(
         "--glob",
         action="store_true",
diff --git a/dvc/config_schema.py b/dvc/config_schema.py
--- a/dvc/config_schema.py
+++ b/dvc/config_schema.py
@@ -154,13 +154,13 @@ def __call__(self, data):
         "machine": Lower,
     },
     "cache": {
-        "local": str,
-        "s3": str,
-        "gs": str,
-        "hdfs": str,
-        "webhdfs": str,
-        "ssh": str,
-        "azure": str,
+        "local": str,  # obsoleted
+        "s3": str,  # obsoleted
+        "gs": str,  # obsoleted
+        "hdfs": str,  # obsoleted
+        "webhdfs": str,  # obsoleted
+        "ssh": str,  # obsoleted
+        "azure": str,  # obsoleted
         # This is for default local cache
         "dir": str,
         **LOCAL_COMMON,
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -299,21 +299,6 @@ def __init__(self, url):
         super().__init__(f"The path '{url}' does not exist")
 
 
-class RemoteCacheRequiredError(DvcException):
-    def __init__(self, scheme, fs_path):
-        super().__init__(
-            (
-                "Current operation was unsuccessful because '{}' requires "
-                "existing cache on '{}' remote. See {} for information on how "
-                "to set up remote cache."
-            ).format(
-                fs_path,
-                scheme,
-                format_link("https://man.dvc.org/config#cache"),
-            )
-        )
-
-
 class IsADirectoryError(DvcException):  # noqa,pylint:disable=redefined-builtin
     """Raised when a file operation is requested on a directory."""
 
diff --git a/dvc/output.py b/dvc/output.py
--- a/dvc/output.py
+++ b/dvc/output.py
@@ -19,7 +19,6 @@
     ConfirmRemoveError,
     DvcException,
     MergeError,
-    RemoteCacheRequiredError,
 )
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile import check as ocheck
@@ -526,14 +525,9 @@ def use_scm_ignore(self):
     def cache(self):
         from dvc.cachemgr import LEGACY_HASH_NAMES
 
-        if self.is_in_repo:
-            odb_name = "legacy" if self.hash_name in LEGACY_HASH_NAMES else "repo"
-        else:
-            odb_name = self.protocol
-        odb = getattr(self.repo.cache, odb_name)
-        if self.use_cache and odb is None:
-            raise RemoteCacheRequiredError(self.fs.protocol, self.fs_path)
-        return odb
+        assert self.is_in_repo
+        odb_name = "legacy" if self.hash_name in LEGACY_HASH_NAMES else "repo"
+        return getattr(self.repo.cache, odb_name)
 
     @property
     def local_cache(self):
diff --git a/dvc/repo/add.py b/dvc/repo/add.py
--- a/dvc/repo/add.py
+++ b/dvc/repo/add.py
@@ -54,7 +54,6 @@ def find_targets(
 def get_or_create_stage(
     repo: "Repo",
     target: str,
-    external: bool = False,
     out: Optional[str] = None,
     to_remote: bool = False,
     force: bool = False,
@@ -79,7 +78,6 @@ def get_or_create_stage(
             fname=path,
             wdir=wdir,
             outs=[out],
-            external=external,
             force=force,
         )
         return StageInfo(stage, output_exists=False)
@@ -199,7 +197,6 @@ def add(
     repo: "Repo",
     targets: Union["StrOrBytesPath", Iterator["StrOrBytesPath"]],
     no_commit: bool = False,
-    external: bool = False,
     glob: bool = False,
     out: Optional[str] = None,
     remote: Optional[str] = None,
@@ -215,7 +212,6 @@ def add(
         target: get_or_create_stage(
             repo,
             target,
-            external=external,
             out=out,
             to_remote=to_remote,
             force=force,
diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py
--- a/dvc/stage/__init__.py
+++ b/dvc/stage/__init__.py
@@ -89,7 +89,7 @@ class RawData:
     generated_from: Optional[str] = None
 
 
-def create_stage(cls: Type[_T], repo, path, external=False, **kwargs) -> _T:
+def create_stage(cls: Type[_T], repo, path, **kwargs) -> _T:
     from dvc.dvcfile import check_dvcfile_path
 
     wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
@@ -101,8 +101,7 @@ def create_stage(cls: Type[_T], repo, path, external=False, **kwargs) -> _T:
 
     stage = loads_from(cls, repo, path, wdir, kwargs)
     fill_stage_outputs(stage, **kwargs)
-    if not external:
-        check_no_externals(stage)
+    check_no_externals(stage)
     fill_stage_dependencies(
         stage, **project(kwargs, ["deps", "erepo", "params", "fs_config"])
     )
diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py
--- a/dvc/stage/cache.py
+++ b/dvc/stage/cache.py
@@ -44,7 +44,12 @@ def _can_hash(stage):
             return False
 
     for out in stage.outs:
-        if out.protocol != "local" or not out.def_path or out.persist:
+        if (
+            out.protocol != "local"
+            or not out.def_path
+            or out.persist
+            or not out.is_in_repo
+        ):
             return False
 
     return True
@@ -112,7 +117,6 @@ def _create_stage(self, cache, wdir=None):
             cmd=cache["cmd"],
             wdir=wdir,
             outs=[out["path"] for out in cache["outs"]],
-            external=True,
         )
         StageLoader.fill_from_lock(stage, cache)
         return stage
@@ -137,7 +141,7 @@ def _uncached_outs(self, stage, cache):
         # NOTE: using copy link to make it look like a git-tracked file
         with self._cache_type_copy():
             for out in cached_stage.outs:
-                if out.def_path in outs_no_cache:
+                if out.def_path in outs_no_cache and out.is_in_repo:
                     yield out
 
     def save(self, stage):
diff --git a/dvc/stage/utils.py b/dvc/stage/utils.py
--- a/dvc/stage/utils.py
+++ b/dvc/stage/utils.py
@@ -87,28 +87,24 @@ def fill_stage_dependencies(stage, deps=None, erepo=None, params=None, fs_config
 
 
 def check_no_externals(stage):
-    from urllib.parse import urlparse
-
     from dvc.utils import format_link
 
-    # NOTE: preventing users from accidentally using external outputs. See
-    # https://github.com/iterative/dvc/issues/1545 for more details.
-
-    def _is_external(out):
-        # NOTE: in case of `remote://` notation, the user clearly knows that
-        # this is an advanced feature and so we shouldn't error-out.
-        if out.is_in_repo or urlparse(out.def_path).scheme == "remote":
+    def _is_cached_external(out):
+        if out.is_in_repo or not out.use_cache:
             return False
         return True
 
-    outs = [str(out) for out in stage.outs if _is_external(out)]
+    outs = [str(out) for out in stage.outs if _is_cached_external(out)]
     if not outs:
         return
 
     str_outs = ", ".join(outs)
-    link = format_link("https://dvc.org/doc/user-guide/managing-external-data")
+    link = format_link(
+        "https://dvc.org/doc/user-guide/pipelines/external-dependencies-and-outputs"
+    )
     raise StageExternalOutputsError(
-        f"Output(s) outside of DVC project: {str_outs}. See {link} for more info."
+        f"Cached output(s) outside of DVC project: {str_outs}. "
+        f"See {link} for more info."
     )
 
 


Test patch:
diff --git a/dvc/testing/workspace_tests.py b/dvc/testing/workspace_tests.py
--- a/dvc/testing/workspace_tests.py
+++ b/dvc/testing/workspace_tests.py
@@ -153,57 +153,6 @@ def test_import_no_download(self, tmp_dir, dvc, remote_version_aware):
         assert dvc.status() == {}
 
 
-class TestAdd:
-    @pytest.fixture
-    def hash_name(self):
-        pytest.skip()
-
-    @pytest.fixture
-    def hash_value(self):
-        pytest.skip()
-
-    @pytest.fixture
-    def dir_hash_value(self):
-        pytest.skip()
-
-    def test_add(self, tmp_dir, dvc, workspace, hash_name, hash_value):
-        from dvc.stage.exceptions import StageExternalOutputsError
-
-        workspace.gen("file", "file")
-
-        with pytest.raises(StageExternalOutputsError):
-            dvc.add(workspace.url)
-
-        dvc.add("remote://workspace/file")
-        assert (tmp_dir / "file.dvc").read_text() == (
-            "outs:\n"
-            f"- {hash_name}: {hash_value}\n"
-            "  size: 4\n"
-            "  hash: md5\n"
-            "  path: remote://workspace/file\n"
-        )
-        assert (workspace / "file").read_text() == "file"
-        assert (
-            workspace / "cache" / "files" / "md5" / hash_value[:2] / hash_value[2:]
-        ).read_text() == "file"
-
-        assert dvc.status() == {}
-
-    # pylint: disable-next=unused-argument
-    def test_add_dir(self, tmp_dir, dvc, workspace, hash_name, dir_hash_value):
-        workspace.gen({"dir": {"file": "file", "subdir": {"subfile": "subfile"}}})
-
-        dvc.add("remote://workspace/dir")
-        assert (
-            workspace
-            / "cache"
-            / "files"
-            / "md5"
-            / dir_hash_value[:2]
-            / dir_hash_value[2:]
-        ).is_file()
-
-
 def match_files(fs, entries, expected):
     entries_content = {(fs.path.normpath(d["path"]), d["isdir"]) for d in entries}
     expected_content = {(fs.path.normpath(d["path"]), d["isdir"]) for d in expected}
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -22,7 +22,6 @@
 )
 from dvc.stage import Stage
 from dvc.stage.exceptions import StageExternalOutputsError, StagePathNotFoundError
-from dvc.testing.workspace_tests import TestAdd
 from dvc.utils.fs import path_isin
 from dvc.utils.serialize import YAMLFileCorruptedError
 from dvc_data.hashfile.hash import file_md5
@@ -234,55 +233,6 @@ def test_add_filtered_files_in_dir(
         assert stage.outs[0].def_path in expected_def_paths
 
 
-class TestAddExternal(TestAdd):
-    @pytest.fixture
-    def hash_name(self):
-        return "md5"
-
-    @pytest.fixture
-    def hash_value(self):
-        return "8c7dd922ad47494fc02c388e12c00eac"
-
-    @pytest.fixture
-    def dir_hash_value(self):
-        return "b6dcab6ccd17ca0a8bf4a215a37d14cc.dir"
-
-
-def test_add_external_relpath(tmp_dir, dvc, local_cloud):
-    (fpath,) = local_cloud.gen("file", "file")
-    rel = os.path.relpath(fpath)
-
-    with pytest.raises(StageExternalOutputsError):
-        dvc.add(rel)
-
-    dvc.add(rel, external=True)
-    assert (tmp_dir / "file.dvc").read_text() == (
-        "outs:\n"
-        "- md5: 8c7dd922ad47494fc02c388e12c00eac\n"
-        "  size: 4\n"
-        "  hash: md5\n"
-        f"  path: {rel}\n"
-    )
-    assert fpath.read_text() == "file"
-    assert dvc.status() == {}
-
-
-def test_add_local_remote_file(tmp_dir, dvc):
-    """
-    Making sure that 'remote' syntax is handled properly for local outs.
-    """
-    tmp_dir.gen({"foo": "foo", "bar": "bar"})
-    tmp_dir.add_remote(url=tmp_dir.fs_path, name="myremote")
-
-    assert main(["add", "remote://myremote/foo"]) == 0
-    d = (tmp_dir / "foo.dvc").load_yaml()
-    assert d["outs"][0]["path"] == "remote://myremote/foo"
-
-    assert main(["add", (tmp_dir / "bar").fs_path]) == 0
-    d = (tmp_dir / "bar.dvc").load_yaml()
-    assert d["outs"][0]["path"] == "bar"
-
-
 def test_cmd_add(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
     ret = main(["add", "foo"])
@@ -755,15 +705,9 @@ def test_add_symlink_file(tmp_dir, dvc):
     dvc.add(os.path.join("dir", "foo"))
 
 
-@pytest.mark.parametrize("external", [True, False])
-def test_add_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
-    if external:
-        data_dir = make_tmp_dir("data")
-        data_dir.gen({"foo": "foo"})
-        target = os.fspath(data_dir)
-    else:
-        tmp_dir.gen({"data": {"foo": "foo"}})
-        target = os.path.join(".", "data")
+def test_add_symlink_dir(make_tmp_dir, tmp_dir, dvc):
+    tmp_dir.gen({"data": {"foo": "foo"}})
+    target = os.path.join(".", "data")
 
     tmp_dir.gen({"data": {"foo": "foo"}})
 
@@ -774,15 +718,9 @@ def test_add_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
         dvc.add("dir")
 
 
-@pytest.mark.parametrize("external", [True, False])
-def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
-    if external:
-        data_dir = make_tmp_dir("data")
-        data_dir.gen({"dir": {"foo": "foo"}})
-        target = os.fspath(data_dir / "dir")
-    else:
-        tmp_dir.gen({"data": {"foo": "foo"}})
-        target = os.path.join(".", "data")
+def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc):
+    tmp_dir.gen({"data": {"foo": "foo"}})
+    target = os.path.join(".", "data")
 
     (tmp_dir / "dir").symlink_to(target)
 
diff --git a/tests/func/test_gc.py b/tests/func/test_gc.py
--- a/tests/func/test_gc.py
+++ b/tests/func/test_gc.py
@@ -251,34 +251,6 @@ def test_gc_not_collect_pipeline_tracked_files(tmp_dir, dvc, run_copy):
     assert _count_files(dvc.cache.local.path) == 0
 
 
-def test_gc_external_output(tmp_dir, dvc, workspace):
-    workspace.gen({"foo": "foo", "bar": "bar"})
-
-    (foo_stage,) = dvc.add("remote://workspace/foo")
-    (bar_stage,) = dvc.add("remote://workspace/bar")
-
-    foo_hash = foo_stage.outs[0].hash_info.value
-    bar_hash = bar_stage.outs[0].hash_info.value
-
-    assert (
-        workspace / "cache" / "files" / "md5" / foo_hash[:2] / foo_hash[2:]
-    ).read_text() == "foo"
-    assert (
-        workspace / "cache" / "files" / "md5" / bar_hash[:2] / bar_hash[2:]
-    ).read_text() == "bar"
-
-    (tmp_dir / "foo.dvc").unlink()
-
-    dvc.gc(workspace=True)
-
-    assert not (
-        workspace / "cache" / "files" / "md5" / foo_hash[:2] / foo_hash[2:]
-    ).exists()
-    assert (
-        workspace / "cache" / "files" / "md5" / bar_hash[:2] / bar_hash[2:]
-    ).read_text() == "bar"
-
-
 def test_gc_all_experiments(tmp_dir, scm, dvc):
     from dvc.repo.experiments.refs import ExpRefInfo
 
diff --git a/tests/func/test_get.py b/tests/func/test_get.py
--- a/tests/func/test_get.py
+++ b/tests/func/test_get.py
@@ -129,20 +129,6 @@ def test_get_a_dvc_file(tmp_dir, erepo_dir):
         Repo.get(os.fspath(erepo_dir), "some_file.dvc")
 
 
-# https://github.com/iterative/dvc/pull/2837#discussion_r352123053
-def test_get_full_dvc_path(tmp_dir, erepo_dir, tmp_path_factory):
-    path = tmp_path_factory.mktemp("ext")
-    external_data = path / "ext_data"
-    external_data.write_text("ext_data")
-
-    with erepo_dir.chdir():
-        erepo_dir.dvc.add(os.fspath(external_data), external=True)
-        erepo_dir.scm_add("ext_data.dvc", commit="add external data")
-
-    Repo.get(os.fspath(erepo_dir), os.fspath(external_data), "ext_data_imported")
-    assert (tmp_dir / "ext_data_imported").read_text() == "ext_data"
-
-
 def test_non_cached_output(tmp_dir, erepo_dir):
     src = "non_cached_file"
     dst = src + "_imported"
diff --git a/tests/func/test_odb.py b/tests/func/test_odb.py
--- a/tests/func/test_odb.py
+++ b/tests/func/test_odb.py
@@ -81,17 +81,6 @@ def test_external_cache_dir(tmp_dir, dvc, make_tmp_dir):
     assert len(os.listdir(cache_dir)) != 0
 
 
-def test_remote_cache_references(tmp_dir, dvc):
-    with dvc.config.edit() as conf:
-        conf["remote"]["storage"] = {"url": "ssh://user@localhost:23"}
-        conf["remote"]["cache"] = {"url": "remote://storage/tmp"}
-        conf["cache"]["ssh"] = "cache"
-
-    dvc.cache = CacheManager(dvc)
-
-    assert dvc.cache.ssh.path == "/tmp/files/md5"
-
-
 def test_shared_cache_dir(tmp_dir):
     cache_dir = os.path.abspath(os.path.join(os.curdir, "cache"))
     for d in ["dir1", "dir2"]:
diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py
--- a/tests/func/test_remote.py
+++ b/tests/func/test_remote.py
@@ -9,7 +9,7 @@
 
 from dvc.cli import main
 from dvc.config import Config
-from dvc.exceptions import DownloadError, RemoteCacheRequiredError, UploadError
+from dvc.exceptions import DownloadError, UploadError
 from dvc.utils.fs import remove
 
 
@@ -257,22 +257,6 @@ def test_remote_modify_local_on_repo_config(tmp_dir, dvc):
     }
 
 
-def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
-    # https://github.com/iterative/dvc/issues/2647, is some situations
-    # (external dir dependency) cache is required to calculate dir md5
-    external_dir = tmp_path_factory.mktemp("external_dir")
-    file = external_dir / "file"
-
-    dvc.cache.local = None
-    with pytest.raises(RemoteCacheRequiredError):
-        dvc.run(
-            cmd=f"echo content > {file}",
-            outs=[os.fspath(file)],
-            name="echo",
-            external=True,
-        )
-
-
 def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
     from dvc_objects.fs import generic
 
diff --git a/tests/func/test_repo_index.py b/tests/func/test_repo_index.py
--- a/tests/func/test_repo_index.py
+++ b/tests/func/test_repo_index.py
@@ -143,17 +143,7 @@ def test_skip_graph_checks(dvc, mocker):
     assert mock_build_graph.called
 
 
-def get_index(dvc, rev):
-    if rev:
-        brancher = dvc.brancher(revs=[rev])
-        if rev != "workspace":
-            assert next(brancher) == "workspace"
-        next(brancher)
-    return Index.from_repo(dvc)
-
-
-@pytest.mark.parametrize("rev", ["workspace", "HEAD"])
-def test_used_objs(tmp_dir, scm, dvc, run_copy, rev):
+def test_used_objs(tmp_dir, scm, dvc, run_copy):
     from dvc_data.hashfile.hash_info import HashInfo
 
     dvc.scm_context.autostage = True
@@ -161,30 +151,31 @@ def test_used_objs(tmp_dir, scm, dvc, run_copy, rev):
     run_copy("foo", "bar", name="copy-foo-bar")
     scm.commit("commit")
 
-    index = get_index(dvc, rev)
-
-    expected_objs = [
-        HashInfo(
-            name="md5",
-            value="acbd18db4cc2f85cedef654fccc4a4d8",
-            obj_name="bar",
-        ),
-        HashInfo(
-            name="md5",
-            value="8c7dd922ad47494fc02c388e12c00eac",
-            obj_name="dir/subdir/file",
-        ),
-        HashInfo(
-            name="md5",
-            value="d28c9e28591aeb7e303dc6772ffa6f6b.dir",
-            obj_name="dir",
-        ),
-    ]
-
-    assert index.used_objs() == {None: set(expected_objs)}
-    assert index.used_objs("dir") == {None: set(expected_objs[1:])}
-    assert index.used_objs(".", recursive=True) == {None: set(expected_objs)}
-    assert index.used_objs("copy-foo-bar", with_deps=True) == {None: {expected_objs[0]}}
+    for _ in dvc.brancher(revs=["HEAD"]):
+        expected_objs = [
+            HashInfo(
+                name="md5",
+                value="acbd18db4cc2f85cedef654fccc4a4d8",
+                obj_name="bar",
+            ),
+            HashInfo(
+                name="md5",
+                value="8c7dd922ad47494fc02c388e12c00eac",
+                obj_name="dir/subdir/file",
+            ),
+            HashInfo(
+                name="md5",
+                value="d28c9e28591aeb7e303dc6772ffa6f6b.dir",
+                obj_name="dir",
+            ),
+        ]
+
+        assert dvc.index.used_objs() == {None: set(expected_objs)}
+        assert dvc.index.used_objs("dir") == {None: set(expected_objs[1:])}
+        assert dvc.index.used_objs(".", recursive=True) == {None: set(expected_objs)}
+        assert dvc.index.used_objs("copy-foo-bar", with_deps=True) == {
+            None: {expected_objs[0]}
+        }
 
 
 def test_view_granular_dir(tmp_dir, scm, dvc, run_copy):
diff --git a/tests/func/test_run.py b/tests/func/test_run.py
--- a/tests/func/test_run.py
+++ b/tests/func/test_run.py
@@ -534,7 +534,7 @@ def test_run_external_outputs(
         name="mystage",
         cmd="mycmd",
         deps=["remote://workspace/foo"],
-        outs=["remote://workspace/bar"],
+        outs_no_cache=["remote://workspace/bar"],
         no_exec=True,
     )
 
@@ -545,7 +545,8 @@ def test_run_external_outputs(
         "    deps:\n"
         "    - remote://workspace/foo\n"
         "    outs:\n"
-        "    - remote://workspace/bar\n"
+        "    - remote://workspace/bar:\n"
+        "        cache: false\n"
     )
 
     assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
@@ -574,9 +575,7 @@ def test_run_external_outputs(
 
     assert (local_workspace / "foo").read_text() == "foo"
     assert (local_workspace / "bar").read_text() == "bar"
-    assert (
-        local_workspace / "cache" / "files" / "md5" / bar_hash[:2] / bar_hash[2:]
-    ).read_text() == "bar"
+    assert not (local_workspace / "cache").exists()
 
 
 def test_rerun_callback(dvc):
diff --git a/tests/unit/command/test_add.py b/tests/unit/command/test_add.py
--- a/tests/unit/command/test_add.py
+++ b/tests/unit/command/test_add.py
@@ -9,7 +9,6 @@ def test_add(mocker, dvc):
         [
             "add",
             "--no-commit",
-            "--external",
             "--glob",
             "data",
         ]
@@ -25,7 +24,6 @@ def test_add(mocker, dvc):
         ["data"],
         no_commit=True,
         glob=True,
-        external=True,
         out=None,
         remote=None,
         to_remote=False,
@@ -57,7 +55,6 @@ def test_add_to_remote(mocker):
         ["s3://bucket/foo"],
         no_commit=False,
         glob=False,
-        external=False,
         out="bar",
         remote="remote",
         to_remote=True,
diff --git a/tests/unit/stage/test_stage.py b/tests/unit/stage/test_stage.py
--- a/tests/unit/stage/test_stage.py
+++ b/tests/unit/stage/test_stage.py
@@ -120,5 +120,8 @@ def test_external_outs(tmp_path_factory, dvc):
     with dvc.config.edit() as conf:
         conf["remote"]["myremote"] = {"url": os.fspath(tmp_path)}
 
-    create_stage(Stage, dvc, "path.dvc", outs=["remote://myremote/foo"])
-    create_stage(Stage, dvc, "path.dvc", outs=[os.fspath(foo)], external=True)
+    with pytest.raises(StageExternalOutputsError):
+        create_stage(Stage, dvc, "path.dvc", outs=["remote://myremote/foo"])
+
+    create_stage(Stage, dvc, "path.dvc", outs_no_cache=["remote://myremote/foo"])
+    create_stage(Stage, dvc, "path.dvc", outs_no_cache=[os.fspath(foo)])
diff --git a/tests/unit/test_info.py b/tests/unit/test_info.py
--- a/tests/unit/test_info.py
+++ b/tests/unit/test_info.py
@@ -85,14 +85,10 @@ def test_info_in_broken_git_repo(tmp_dir, dvc, scm, caplog):
 
 
 def test_caches(tmp_dir, dvc, caplog):
-    tmp_dir.add_remote(name="sshcache", url="ssh://example.com/path", default=False)
-    with tmp_dir.dvc.config.edit() as conf:
-        conf["cache"]["ssh"] = "sshcache"
-
     dvc_info = get_dvc_info()
 
     # Order of cache types is runtime dependent
-    assert re.search("Caches: (local, ssh|ssh, local)", dvc_info)
+    assert re.search("Caches: local", dvc_info)
 
 
 def test_remotes_empty(tmp_dir, dvc, caplog):