diff --git a/dvc/commands/add.py b/dvc/commands/add.py
index 633cbcfc3..ca1ea6e95 100644
--- a/dvc/commands/add.py
+++ b/dvc/commands/add.py
@@ -10,17 +10,13 @@ logger = logging.getLogger(__name__)
 
 class CmdAdd(CmdBase):
     def run(self):
-        from dvc.exceptions import DvcException, RecursiveAddingWhileUsingFilename
+        from dvc.exceptions import DvcException
 
         try:
-            if len(self.args.targets) > 1 and self.args.file:
-                raise RecursiveAddingWhileUsingFilename()
-
             self.repo.add(
                 self.args.targets,
-                recursive=self.args.recursive,
                 no_commit=self.args.no_commit,
-                fname=self.args.file,
+                file=self.args.file,
                 external=self.args.external,
                 glob=self.args.glob,
                 out=self.args.out,
@@ -48,13 +44,6 @@ def add_parser(subparsers, parent_parser):
         help=ADD_HELP,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
-    parser.add_argument(
-        "-R",
-        "--recursive",
-        action="store_true",
-        default=False,
-        help="Recursively add files under directory targets.",
-    )
     parser.add_argument(
         "--no-commit",
         action="store_true",
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index 373ab83ad..2342bfe25 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -182,11 +182,6 @@ class BadMetricError(DvcException):
         )
 
 
-class RecursiveAddingWhileUsingFilename(DvcException):
-    def __init__(self):
-        super().__init__("cannot use `fname` with multiple targets or `-R|--recursive`")
-
-
 class OverlappingOutputPathsError(DvcException):
     def __init__(self, parent, overlapping_out, message):
         self.parent = parent
diff --git a/dvc/repo/add.py b/dvc/repo/add.py
index a11360bd0..60f786491 100644
--- a/dvc/repo/add.py
+++ b/dvc/repo/add.py
@@ -1,10 +1,16 @@
-import logging
 import os
 from contextlib import contextmanager
-from itertools import tee
-from typing import TYPE_CHECKING, Any, Iterator, List, NamedTuple, Optional
-
-import colorama
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    Iterator,
+    List,
+    NamedTuple,
+    Optional,
+    Tuple,
+    Union,
+)
 
 from dvc.exceptions import (
     CacheLinkError,
@@ -13,22 +19,17 @@ from dvc.exceptions import (
     OutputDuplicationError,
     OutputNotFoundError,
     OverlappingOutputPathsError,
-    RecursiveAddingWhileUsingFilename,
 )
 from dvc.repo.scm_context import scm_context
+from dvc.types import StrOrBytesPath
 from dvc.ui import ui
-from dvc.utils import LARGE_DIR_SIZE, glob_targets, resolve_output, resolve_paths
-from dvc.utils.collections import ensure_list, validate
+from dvc.utils import glob_targets, resolve_output, resolve_paths
 
 from . import locked
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.stage import Stage
-    from dvc.types import TargetType
-
-Stages = List["Stage"]
-logger = logging.getLogger(__name__)
 
 
 class StageInfo(NamedTuple):
@@ -36,6 +37,75 @@ class StageInfo(NamedTuple):
     output_exists: bool
 
 
+def find_targets(
+    targets: Union[StrOrBytesPath, Iterator[StrOrBytesPath]], glob: bool = False
+) -> List[str]:
+    if isinstance(targets, (str, bytes, os.PathLike)):
+        targets_list = [os.fsdecode(targets)]
+    else:
+        targets_list = [os.fsdecode(target) for target in targets]
+    return glob_targets(targets_list, glob=glob)
+
+
+def validate_args(targets: List[str], **kwargs: Any) -> None:
+    invalid_opt = None
+    to_remote = kwargs.get("to_remote")
+
+    if len(targets) > 1 and kwargs.get("file"):
+        raise InvalidArgumentError("cannot use '--file' with multiple targets")
+
+    if to_remote or kwargs.get("out"):
+        message = "{option} can't be used with "
+        message += "--to-remote" if to_remote else "-o"
+        if len(targets) != 1:
+            invalid_opt = "multiple targets"
+        elif kwargs.get("no_commit"):
+            invalid_opt = "--no-commit option"
+        elif kwargs.get("external"):
+            invalid_opt = "--external option"
+    else:
+        message = "{option} can't be used without --to-remote"
+        if kwargs.get("remote"):
+            invalid_opt = "--remote"
+        elif kwargs.get("jobs"):
+            invalid_opt = "--jobs"
+
+    if invalid_opt is not None:
+        raise InvalidArgumentError(message.format(option=invalid_opt))
+
+
+def get_or_create_stage(
+    repo: "Repo",
+    target: str,
+    file: Optional[str] = None,
+    transfer: bool = False,
+    external: bool = False,
+    force: bool = False,
+    out: Optional[str] = None,
+) -> StageInfo:
+    if out:
+        target = resolve_output(target, out, force=force)
+    path, wdir, out = resolve_paths(repo, target, always_local=transfer and not out)
+
+    try:
+        (out_obj,) = repo.find_outs_by_path(target, strict=False)
+        stage = out_obj.stage
+        if not stage.is_data_source:
+            raise DvcException(f"cannot update {out!r}: not a data source")
+        return StageInfo(stage, output_exists=True)
+    except OutputNotFoundError:
+        stage = repo.stage.create(
+            single_stage=True,
+            validate=False,
+            fname=file or path,
+            wdir=wdir,
+            outs=[out],
+            external=external,
+            force=force,
+        )
+        return StageInfo(stage, output_exists=False)
+
+
 OVERLAPPING_CHILD_FMT = (
     "Cannot add '{out}', because it is overlapping with other "
     "DVC tracked output: '{parent}'.\n"
@@ -51,52 +121,8 @@ OVERLAPPING_PARENT_FMT = (
 )
 
 
-def check_recursive_and_fname(args):
-    if args.recursive and args.fname:
-        raise RecursiveAddingWhileUsingFilename()
-
-
-def transform_targets(args):
-    from funcy import count_reps
-
-    counts = count_reps(ensure_list(args.targets))
-    dupes = [key for key, count in counts.items() if count > 1]
-    if dupes:
-        msg = ", ".join(f"[b]{key}[/]" for key in dupes)
-        ui.error_write(f"ignoring duplicated targets: {msg}", styled=True)
-    args.targets = list(counts)
-
-
-def check_arg_combinations(args):
-    kwargs = args.kwargs
-    invalid_opt = None
-    to_remote = args.to_remote
-    to_cache = kwargs.get("out") and not to_remote
-
-    if to_remote or to_cache:
-        message = "{option} can't be used with "
-        message += "--to-remote" if to_remote else "-o"
-        if len(args.targets) != 1:
-            invalid_opt = "multiple targets"
-        elif args.no_commit:
-            invalid_opt = "--no-commit option"
-        elif args.recursive:
-            invalid_opt = "--recursive option"
-        elif kwargs.get("external"):
-            invalid_opt = "--external option"
-    else:
-        message = "{option} can't be used without --to-remote"
-        if kwargs.get("remote"):
-            invalid_opt = "--remote"
-        elif kwargs.get("jobs"):
-            invalid_opt = "--jobs"
-
-    if invalid_opt is not None:
-        raise InvalidArgumentError(message.format(option=invalid_opt))
-
-
 @contextmanager
-def translate_graph_error(stages: Stages) -> Iterator[None]:
+def translate_graph_error(stages: List["Stage"]) -> Iterator[None]:
     try:
         yield
     except OverlappingOutputPathsError as exc:
@@ -121,18 +147,21 @@ def translate_graph_error(stages: Stages) -> Iterator[None]:
         )
 
 
-def progress_iter(stages: List[StageInfo]) -> Iterator["StageInfo"]:
+def progress_iter(stages: Dict[str, StageInfo]) -> Iterator[Tuple[str, StageInfo]]:
     total = len(stages)
     desc = "Adding..."
-    with ui.progress(stages, total=total, desc=desc, unit="file", leave=True) as pbar:
+    with ui.progress(
+        stages.items(), total=total, desc=desc, unit="file", leave=True
+    ) as pbar:
         if total == 1:
             pbar.bar_format = desc
             pbar.refresh()
 
-        for item in pbar:
+        for item, stage_info in pbar:
             if total > 1:
-                pbar.set_msg(f"{item.stage.outs[0]}")
-            yield item
+                pbar.set_msg(str(stage_info.stage.outs[0]))
+                pbar.refresh()
+            yield item, stage_info
             if total == 1:  # restore bar format for stats
                 # pylint: disable=no-member
                 pbar.bar_format = pbar.BAR_FMT_DEFAULT
@@ -159,142 +188,90 @@ def warn_link_failures() -> Iterator[List[str]]:
             ui.error_write(msg)
 
 
-VALIDATORS = (
-    check_recursive_and_fname,
-    transform_targets,
-    check_arg_combinations,
-)
-
-
-@validate(*VALIDATORS)
-@locked
-@scm_context
-def add(
-    repo: "Repo",
-    targets: "TargetType",
-    recursive: bool = False,
-    no_commit: bool = False,
-    fname: Optional[str] = None,
+def _add_transfer(
+    stage: "Stage",
+    source: str,
+    remote: Optional[str] = None,
     to_remote: bool = False,
-    **kwargs: Any,
-):
-    to_cache = bool(kwargs.get("out")) and not to_remote
-    transfer = to_remote or to_cache
-
-    glob = kwargs.get("glob", False)
-    add_targets = collect_targets(repo, targets, recursive, glob)
-    # pass one for creating stages, other one is used for iterating here
-    add_targets, sources = tee(add_targets)
-
-    # collect targets and build stages as we go
-    desc = "Collecting targets"
-    stages_it = create_stages(repo, add_targets, fname, transfer, **kwargs)
-    stages = list(ui.progress(stages_it, desc=desc, unit="file"))
-
-    stages_list = [stage for stage, _ in stages]
-    msg = "Collecting stages from the workspace"
-    with translate_graph_error(stages_list), ui.status(msg) as status:
-        # remove existing stages that are to-be replaced with these
-        # new stages for the graph checks.
-        repo.check_graph(
-            stages=stages_list, callback=lambda: status.update("Checking graph")
-        )
-
+    jobs: Optional[int] = None,
+    force: bool = False,
+) -> None:
     odb = None
     if to_remote:
-        odb = repo.cloud.get_remote_odb(kwargs.get("remote"), "add")
+        odb = stage.repo.cloud.get_remote_odb(remote, "add")
+    stage.transfer(source, odb=odb, to_remote=to_remote, jobs=jobs, force=force)
+    stage.dump()
+
+
+def _add(stage: "Stage", source: Optional[str] = None, no_commit: bool = False) -> None:
+    out = stage.outs[0]
+    path = out.fs.path.abspath(source) if source else None
+    try:
+        stage.add_outs(path, no_commit=no_commit)
+    except CacheLinkError:
+        stage.dump()
+        raise
+    stage.dump()
+
+
+@locked
+@scm_context
+def add(  # noqa: PLR0913
+    repo: "Repo",
+    targets: Union[StrOrBytesPath, Iterator[StrOrBytesPath]],
+    no_commit: bool = False,
+    file: Optional[str] = None,
+    external: bool = False,
+    glob: bool = False,
+    out: Optional[str] = None,
+    remote: Optional[str] = None,
+    to_remote: bool = False,
+    jobs: Optional[int] = None,
+    force: bool = False,
+) -> List["Stage"]:
+    transfer = to_remote or bool(out)
+    add_targets = find_targets(targets, glob=glob)
+    if not add_targets:
+        return []
+
+    validate_args(
+        add_targets,
+        no_commit=no_commit,
+        file=file,
+        external=external,
+        out=out,
+        remote=remote,
+        to_remote=to_remote,
+        jobs=jobs,
+    )
+    stages_with_targets = {
+        target: get_or_create_stage(
+            repo,
+            target,
+            file=file,
+            transfer=transfer,
+            external=external,
+            force=force,
+            out=out,
+        )
+        for target in add_targets
+    }
+
+    stages = [stage for stage, _ in stages_with_targets.values()]
+    msg = "Collecting stages from the workspace"
+    with translate_graph_error(stages), ui.status(msg) as st:
+        repo.check_graph(stages=stages, callback=lambda: st.update("Checking graph"))
+
+    if transfer:
+        assert stages_with_targets
+        (source, (stage, _)) = next(iter(stages_with_targets.items()))
+        _add_transfer(stage, source, remote, to_remote, jobs=jobs, force=force)
+        return [stage]
 
     with warn_link_failures() as link_failures:
-        for (stage, output_exists), source in zip(progress_iter(stages), sources):
-            out = stage.outs[0]
-            if to_remote or to_cache:
-                stage.transfer(source, to_remote=to_remote, odb=odb, **kwargs)
-            else:
-                try:
-                    path = out.fs.path.abspath(source) if output_exists else None
-                    stage.add_outs(path, no_commit=no_commit)
-                except CacheLinkError:
-                    link_failures.append(str(stage.relpath))
-            stage.dump()
-    return stages_list
-
-
-LARGE_DIR_RECURSIVE_ADD_WARNING = (
-    "You are adding a large directory '{target}' recursively.\n"
-    "Consider tracking it as a whole instead with "
-    "`{cyan}dvc add {target}{nc}`."
-)
-
-
-def collect_targets(
-    repo: "Repo",
-    targets: "TargetType",
-    recursive: bool = False,
-    glob: bool = False,
-) -> Iterator[str]:
-    for target in glob_targets(ensure_list(targets), glob=glob):
-        expanded_targets = _find_all_targets(repo, target, recursive=recursive)
-        for index, path in enumerate(expanded_targets):
-            if index == LARGE_DIR_SIZE:
-                msg = LARGE_DIR_RECURSIVE_ADD_WARNING.format(
-                    cyan=colorama.Fore.CYAN,
-                    nc=colorama.Style.RESET_ALL,
-                    target=target,
-                )
-                ui.error_write(msg)
-            yield path
-
-
-def _find_all_targets(
-    repo: "Repo", target: str, recursive: bool = False
-) -> Iterator[str]:
-    from dvc.dvcfile import is_dvc_file
-
-    if os.path.isdir(target) and recursive:
-        files = repo.dvcignore.find(repo.fs, target)
-        yield from (
-            path
-            for path in files
-            if not repo.is_dvc_internal(path)
-            if not is_dvc_file(path)
-            if not repo.scm.belongs_to_scm(path)
-            if not repo.scm.is_tracked(path)
-        )
-    else:
-        yield target
-
-
-def create_stages(
-    repo: "Repo",
-    targets: Iterator[str],
-    fname: Optional[str] = None,
-    transfer: bool = False,
-    external: bool = False,
-    force: bool = False,
-    **kwargs: Any,
-) -> Iterator[StageInfo]:
-    for target in targets:
-        if kwargs.get("out"):
-            target = resolve_output(target, kwargs["out"], force=force)
-        path, wdir, out = resolve_paths(
-            repo, target, always_local=transfer and not kwargs.get("out")
-        )
-
-        try:
-            (out_obj,) = repo.find_outs_by_path(target, strict=False)
-            stage = out_obj.stage
-            if not stage.is_data_source:
-                raise DvcException(f"cannot update {out!r}: not a data source")
-            output_exists = True
-        except OutputNotFoundError:
-            stage = repo.stage.create(
-                single_stage=True,
-                validate=False,
-                fname=fname or path,
-                wdir=wdir,
-                outs=[out],
-                external=external,
-                force=force,
-            )
-            output_exists = False
-        yield StageInfo(stage, output_exists)
+        for source, (stage, output_exists) in progress_iter(stages_with_targets):
+            try:
+                _add(stage, source if output_exists else None, no_commit=no_commit)
+            except CacheLinkError:
+                link_failures.append(stage.relpath)
+    return stages
diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py
index 1fe3d1a24..7acb1ac10 100644
--- a/dvc/stage/__init__.py
+++ b/dvc/stage/__init__.py
@@ -390,13 +390,14 @@ class Stage(params.StageParams):
         source: str,
         odb: Optional["ObjectDB"] = None,
         to_remote: bool = False,
-        **kwargs: Any,
+        jobs: Optional[int] = None,
+        force: bool = False,
     ) -> None:
         assert len(self.outs) == 1
         (out,) = self.outs
-        out.transfer(source, odb=odb, jobs=kwargs.get("jobs"))
+        out.transfer(source, odb=odb, jobs=jobs)
         if not to_remote:
-            out.checkout(force=kwargs.get("force"))
+            out.checkout(force=force)
             out.ignore()
 
     @rwlocked(read=["deps"], write=["outs"])
diff --git a/dvc/utils/collections.py b/dvc/utils/collections.py
index 2ef4a1410..36a9fb807 100644
--- a/dvc/utils/collections.py
+++ b/dvc/utils/collections.py
@@ -1,7 +1,5 @@
-import inspect
 from collections.abc import Mapping
-from functools import wraps
-from typing import Callable, Dict, Iterable, List, TypeVar, Union, no_type_check
+from typing import Dict, Iterable, List, Union, no_type_check
 
 
 @no_type_check
@@ -102,72 +100,6 @@ def ensure_list(item: Union[Iterable[str], str, None]) -> List[str]:
     return list(item)
 
 
-_KT = TypeVar("_KT")
-_VT = TypeVar("_VT")
-
-
-def chunk_dict(d: Dict[_KT, _VT], size: int = 1) -> List[Dict[_KT, _VT]]:
-    from funcy import chunks
-
-    return [{key: d[key] for key in chunk} for chunk in chunks(size, d)]
-
-
-class _NamespacedDict(dict):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.__dict__ = self
-
-
-def validate(*validators: Callable, post: bool = False):
-    """
-    Validate and transform arguments and results from function calls.
-
-    The validators functions are passed a dictionary of arguments, which
-    supports dot notation access too.
-
-    The key is derived from the function signature, and hence is the name of
-    the argument, whereas the value is the one passed to the function
-    (if it is not passed, default value from keyword arguments are provided).
-
-    >>> def validator(args):
-    ...    assert args["l"] >= 0 and args.b >= 0 and args.h >= 0
-
-    >>> @validate(validator)
-    ... def cuboid_area(l, b, h=1):
-    ...   return 2*(l*b + l*h + b*h)
-
-    >>> cuboid_area(5, 20)
-    250
-    >>> cuboid_area(-1, -2)
-    Traceback (most recent call last):
-      ...
-    AssertionError
-    """
-
-    def wrapped(func: Callable):
-        sig = inspect.signature(func)
-
-        @wraps(func)
-        def inner(*args, **kwargs):
-            ba = sig.bind(*args, **kwargs)
-            ba.apply_defaults()
-            ba.arguments = _NamespacedDict(ba.arguments)  # type: ignore[assignment]
-
-            if not post:
-                for validator in validators:
-                    validator(ba.arguments)
-
-            result = func(*ba.args, **ba.kwargs)
-            if post:
-                for validator in validators:
-                    result = validator(result)
-            return result
-
-        return inner
-
-    return wrapped
-
-
 def nested_contains(dictionary: Dict, phrase: str) -> bool:
     for key, val in dictionary.items():
         if key == phrase and val:
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index c72ec8bde..48b255552 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -6,10 +6,8 @@ import stat
 import textwrap
 from unittest.mock import call
 
-import colorama
 import pytest
 
-import dvc as dvc_module
 import dvc_data
 from dvc.cachemgr import CacheManager
 from dvc.cli import main
@@ -19,7 +17,6 @@ from dvc.exceptions import (
     DvcException,
     InvalidArgumentError,
     OverlappingOutputPathsError,
-    RecursiveAddingWhileUsingFilename,
 )
 from dvc.fs import LocalFileSystem, system
 from dvc.output import (
@@ -30,7 +27,6 @@ from dvc.output import (
 from dvc.stage import Stage
 from dvc.stage.exceptions import StageExternalOutputsError, StagePathNotFoundError
 from dvc.testing.workspace_tests import TestAdd
-from dvc.utils import LARGE_DIR_SIZE
 from dvc.utils.fs import path_isin
 from dvc.utils.serialize import YAMLFileCorruptedError
 from dvc_data.hashfile.hash import file_md5
@@ -39,7 +35,8 @@ from tests.utils import get_gitignore_content
 
 
 def test_add(tmp_dir, dvc):
-    (stage,) = tmp_dir.dvc_gen({"foo": "foo"})
+    tmp_dir.gen("foo", "foo")
+    (stage,) = dvc.add("foo")
     md5 = file_md5("foo", dvc.fs)
 
     assert stage is not None
@@ -114,32 +111,6 @@ def test_add_directory(tmp_dir, dvc):
             assert "\\" not in part
 
 
-def test_add_directory_recursive(tmp_dir, dvc):
-    tmp_dir.gen("data", {"file1": "file1", "sub": {"file2": "file2"}})
-    stages = dvc.add("data", recursive=True)
-    assert len(stages) == 2
-
-
-def test_add_cmd_directory_recursive(tmp_dir, dvc):
-    tmp_dir.gen("data", {"file1": "file1", "sub": {"file2": "file2"}})
-    assert main(["add", "--recursive", "data"]) == 0
-
-
-def test_warn_about_large_directories_recursive_add(tmp_dir, dvc, capsys):
-    warning = (
-        "You are adding a large directory 'large-dir' recursively."
-        "\nConsider tracking it as a whole instead with "
-        "`{cyan}dvc add large-dir{nc}`"
-    ).format(
-        cyan=colorama.Fore.CYAN,
-        nc=colorama.Style.RESET_ALL,
-    )
-
-    tmp_dir.gen("large-dir", {f"{i}": f"{i}" for i in range(LARGE_DIR_SIZE + 1)})
-    assert main(["add", "--recursive", "large-dir"]) == 0
-    assert warning in capsys.readouterr()[1]
-
-
 def test_add_directory_with_forward_slash(tmp_dir, dvc):
     tmp_dir.gen("directory", {"file": "file"})
     (stage,) = dvc.add("directory/")
@@ -484,12 +455,6 @@ def test_add_filename(tmp_dir, dvc):
     ret = main(["add", "foo", "bar", "--file", "error.dvc"])
     assert ret != 0
 
-    ret = main(["add", "-R", "data", "--file", "error.dvc"])
-    assert ret != 0
-
-    with pytest.raises(RecursiveAddingWhileUsingFilename):
-        dvc.add("data", recursive=True, fname="error.dvc")
-
     ret = main(["add", "data", "--file", "data_directory.dvc"])
     assert ret == 0
     assert (tmp_dir / "data_directory.dvc").exists()
@@ -523,17 +488,6 @@ def test_failed_add_cleanup(tmp_dir, scm, dvc):
     assert "/bar" not in gitignore_content
 
 
-def test_should_not_track_git_internal_files(mocker, dvc, tmp_dir):
-    stage_creator_spy = mocker.spy(dvc_module.repo.add, "create_stages")
-
-    ret = main(["add", "-R", dvc.root_dir])
-    assert ret == 0
-
-    created_stages_filenames = stage_creator_spy.mock.call_args[0][1]
-    for fname in created_stages_filenames:
-        assert ".git" not in fname
-
-
 def test_add_unprotected(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
     ret = main(["config", "cache.type", "hardlink"])
@@ -713,10 +667,10 @@ def test_add_from_data_dir(tmp_dir, scm, dvc):
 def test_add_parent_dir(tmp_dir, scm, dvc):
     tmp_dir.gen({"dir": {"file1": "file1 content"}})
     out_path = os.path.join("dir", "file1")
-    dvc.add(out_path, fname=out_path + ".dvc")
+    dvc.add(out_path, file=out_path + ".dvc")
 
     with pytest.raises(OverlappingOutputPathsError) as e:
-        dvc.add("dir", fname="dir.dvc")
+        dvc.add("dir", file="dir.dvc")
     assert str(e.value) == (
         "Cannot add 'dir', because it is overlapping with other DVC "
         "tracked output: '{out}'.\n"
@@ -943,7 +897,6 @@ def test_add_to_remote_absolute(tmp_dir, make_tmp_dir, dvc, remote):
     [
         ("multiple targets", {"targets": ["foo", "bar", "baz"]}),
         ("--no-commit", {"targets": ["foo"], "no_commit": True}),
-        ("--recursive", {"targets": ["foo"], "recursive": True}),
         ("--external", {"targets": ["foo"], "external": True}),
     ],
 )
@@ -1032,7 +985,6 @@ def test_add_to_cache_not_exists(tmp_dir, dvc, local_cloud):
     [
         ("multiple targets", {"targets": ["foo", "bar", "baz"]}),
         ("--no-commit", {"targets": ["foo"], "no_commit": True}),
-        ("--recursive", {"targets": ["foo"], "recursive": True}),
     ],
 )
 def test_add_to_cache_invalid_combinations(dvc, invalid_opt, kwargs):
@@ -1105,15 +1057,6 @@ def test_add_does_not_remove_stage_file_on_failure(tmp_dir, dvc, mocker, target)
     assert (tmp_dir / stage.path).read_text() == dvcfile_contents
 
 
-def test_add_ignore_duplicated_targets(tmp_dir, dvc, capsys):
-    tmp_dir.gen({"foo": "foo", "bar": "bar", "foobar": "foobar"})
-    stages = dvc.add(["foo", "bar", "foobar", "bar", "foo"])
-
-    _, err = capsys.readouterr()
-    assert len(stages) == 3
-    assert "ignoring duplicated targets: foo, bar" in err
-
-
 def test_add_updates_to_cloud_versioning_dir(tmp_dir, dvc):
     data_dvc = tmp_dir / "data.dvc"
     data_dvc.dump(
diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py
index d308b633f..29d5dd8d4 100644
--- a/tests/func/test_checkout.py
+++ b/tests/func/test_checkout.py
@@ -624,7 +624,7 @@ def test_checkout_with_deps(tmp_dir, dvc):
 
 def test_checkout_recursive(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
-    dvc.add("dir", recursive=True)
+    dvc.add("dir/*", glob=True)
 
     (tmp_dir / "dir" / "foo").unlink()
     (tmp_dir / "dir" / "bar").unlink()
diff --git a/tests/func/test_commit.py b/tests/func/test_commit.py
index 23d271e6e..be1589ec1 100644
--- a/tests/func/test_commit.py
+++ b/tests/func/test_commit.py
@@ -5,13 +5,14 @@ import pytest
 
 from dvc.dependency.base import DependencyDoesNotExistError
 from dvc.dvcfile import PROJECT_FILE
+from dvc.fs import localfs
 from dvc.output import OutputDoesNotExistError
 from dvc.stage.exceptions import StageCommitError
 
 
 def test_commit_recursive(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"file": "text1", "subdir": {"file2": "text2"}}})
-    stages = dvc.add("dir", recursive=True, no_commit=True)
+    stages = dvc.add(localfs.find("dir"), no_commit=True)
 
     assert len(stages) == 2
     assert dvc.status() != {}
diff --git a/tests/func/test_move.py b/tests/func/test_move.py
index 305b2e04a..3e56797d2 100644
--- a/tests/func/test_move.py
+++ b/tests/func/test_move.py
@@ -141,7 +141,7 @@ def test_should_move_to_dir_on_non_default_stage_file(tmp_dir, dvc):
     stage_file_name = "stage.dvc"
     tmp_dir.gen({"file": "file_content"})
 
-    dvc.add("file", fname=stage_file_name)
+    dvc.add("file", file=stage_file_name)
     os.mkdir("directory")
 
     dvc.move("file", "directory")
diff --git a/tests/func/test_stage_load.py b/tests/func/test_stage_load.py
index c6053dc91..62f361b5a 100644
--- a/tests/func/test_stage_load.py
+++ b/tests/func/test_stage_load.py
@@ -58,7 +58,7 @@ def test_collect(tmp_dir, scm, dvc, run_copy):
 
 def test_collect_dir_recursive(tmp_dir, dvc, run_head):
     tmp_dir.gen({"dir": {"foo": "foo"}})
-    (stage1,) = dvc.add("dir", recursive=True)
+    (stage1,) = dvc.add("dir/*", glob=True)
     with (tmp_dir / "dir").chdir():
         stage2 = run_head("foo", name="copy-foo-bar")
         stage3 = run_head("foo-1", single_stage=True)
@@ -260,7 +260,7 @@ def test_collect_granular_same_output_name_stage_name(tmp_dir, dvc, run_copy):
 
 def test_collect_granular_priority_on_collision(tmp_dir, dvc, run_copy):
     tmp_dir.gen({"dir": {"foo": "foo"}, "foo": "foo"})
-    (stage1,) = dvc.add("dir", recursive=True)
+    (stage1,) = dvc.add("dir/*", glob=True)
     stage2 = run_copy("foo", "bar", name="dir")
 
     assert dvc.stage.collect_granular("dir") == [(stage2, None)]
diff --git a/tests/func/test_status.py b/tests/func/test_status.py
index 9cd89d28e..e624a9924 100644
--- a/tests/func/test_status.py
+++ b/tests/func/test_status.py
@@ -1,6 +1,7 @@
 import os
 
 from dvc.cli import main
+from dvc.fs import localfs
 
 
 def test_quiet(tmp_dir, dvc, capsys):
@@ -96,7 +97,7 @@ def test_status_on_pipeline_stages(tmp_dir, dvc, run_copy):
 
 def test_status_recursive(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"file": "text1", "subdir": {"file2": "text2"}}})
-    stages = dvc.add("dir", recursive=True, no_commit=True)
+    stages = dvc.add(localfs.find("dir"), no_commit=True)
 
     assert len(stages) == 2
 
diff --git a/tests/unit/command/test_add.py b/tests/unit/command/test_add.py
index d58bccbf4..507d58443 100644
--- a/tests/unit/command/test_add.py
+++ b/tests/unit/command/test_add.py
@@ -8,7 +8,6 @@ def test_add(mocker, dvc):
     cli_args = parse_args(
         [
             "add",
-            "--recursive",
             "--no-commit",
             "--external",
             "--glob",
@@ -26,10 +25,9 @@ def test_add(mocker, dvc):
 
     m.assert_called_once_with(
         ["data"],
-        recursive=True,
         no_commit=True,
         glob=True,
-        fname="file",
+        file="file",
         external=True,
         out=None,
         remote=None,
@@ -60,10 +58,9 @@ def test_add_to_remote(mocker):
 
     m.assert_called_once_with(
         ["s3://bucket/foo"],
-        recursive=False,
         no_commit=False,
         glob=False,
-        fname=None,
+        file=None,
         external=False,
         out="bar",
         remote="remote",
diff --git a/tests/unit/fs/test_data.py b/tests/unit/fs/test_data.py
index 180091b89..b6ae9431a 100644
--- a/tests/unit/fs/test_data.py
+++ b/tests/unit/fs/test_data.py
@@ -4,6 +4,7 @@ import shutil
 import pytest
 
 import dvc_data
+from dvc.fs import localfs
 from dvc.fs.data import DataFileSystem
 from dvc.utils.fs import remove
 from dvc_data.hashfile.build import build
@@ -137,7 +138,7 @@ def test_walk(tmp_dir, dvc):
         }
     )
 
-    dvc.add("dir", recursive=True)
+    dvc.add(localfs.find("dir"))
     fs = DataFileSystem(index=dvc.index.data["repo"])
 
     expected = [
diff --git a/tests/unit/fs/test_dvc.py b/tests/unit/fs/test_dvc.py
index dbe68b805..f3519b375 100644
--- a/tests/unit/fs/test_dvc.py
+++ b/tests/unit/fs/test_dvc.py
@@ -5,6 +5,7 @@ from unittest import mock
 
 import pytest
 
+from dvc.fs import localfs
 from dvc.fs.dvc import DVCFileSystem
 from dvc.testing.tmp_dir import make_subrepo
 from dvc_data.hashfile.build import build
@@ -185,7 +186,7 @@ def test_walk(tmp_dir, dvc, dvcfiles, extra_expected):
             }
         }
     )
-    dvc.add(str(tmp_dir / "dir"), recursive=True)
+    dvc.add(localfs.find("dir"))
     tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
     fs = DVCFileSystem(repo=dvc)
 
diff --git a/tests/unit/utils/test_collections.py b/tests/unit/utils/test_collections.py
index dab4d5269..6f26abce9 100644
--- a/tests/unit/utils/test_collections.py
+++ b/tests/unit/utils/test_collections.py
@@ -4,11 +4,9 @@ import pytest
 
 from dvc.utils.collections import (
     apply_diff,
-    chunk_dict,
     merge_dicts,
     remove_missing_keys,
     to_omegaconf,
-    validate,
 )
 from dvc.utils.serialize import dumps_yaml
 
@@ -61,88 +59,6 @@ def test_apply_diff_seq():
     assert dest["l"] is inner, "Updates inner lists"
 
 
-def test_chunk_dict():
-    assert chunk_dict({}) == []
-
-    d = {"a": 1, "b": 2, "c": 3}
-    assert chunk_dict(d) == [{"a": 1}, {"b": 2}, {"c": 3}]
-    assert chunk_dict(d, 2) == [{"a": 1, "b": 2}, {"c": 3}]
-    assert chunk_dict(d, 3) == [d]
-    assert chunk_dict(d, 4) == [d]
-
-
-def _test_func(x, y, *args, j=3, k=5, **kwargs):
-    pass
-
-
-def test_pre_validate_decorator_required_args(mocker):
-    mock = mocker.MagicMock()
-
-    func_mock = mocker.create_autospec(_test_func)
-    func = validate(mock)(func_mock)
-
-    func("x", "y")
-
-    func_mock.assert_called_once_with("x", "y", j=3, k=5)
-    mock.assert_called_once()
-
-    (args,) = mock.call_args[0]
-    assert args.x == "x"
-    assert args.y == "y"
-    assert args.args == ()
-    assert args.j == 3
-    assert args.k == 5
-    assert args.kwargs == {}
-
-
-def test_pre_validate_decorator_kwargs_args(mocker):
-    mock = mocker.MagicMock()
-    func_mock = mocker.create_autospec(_test_func)
-    func = validate(mock)(func_mock)
-
-    func("x", "y", "part", "of", "args", j=1, k=10, m=100, n=1000)
-
-    func_mock.assert_called_once_with(
-        "x", "y", "part", "of", "args", j=1, k=10, m=100, n=1000
-    )
-    mock.assert_called_once()
-    (args,) = mock.call_args[0]
-    assert args.x == "x"
-    assert args.y == "y"
-    assert args.args == ("part", "of", "args")
-    assert args.j == 1
-    assert args.k == 10
-    assert args.kwargs == {"m": 100, "n": 1000}
-
-
-def test_pre_validate_update_args(mocker):
-    def test_validator(args):
-        args.w += 50
-        del args.x
-        args.y = 100
-
-    def test_func(w=1, x=5, y=10, z=15):
-        pass
-
-    mock = mocker.create_autospec(test_func)
-    func = validate(test_validator)(mock)
-
-    func(100, 100)
-    mock.assert_called_once_with(w=150, y=100, z=15)
-
-
-def test_post_validate_decorator(mocker):
-    def none_filter(result):
-        return list(filter(None, result))
-
-    test_func = mocker.MagicMock(return_value=[1, None, 2])
-    func = validate(none_filter, post=True)(test_func)
-
-    result = func()
-    test_func.assert_called_once()
-    assert result == [1, 2]
-
-
 def is_serializable(d):
     json.dumps(d)
     dumps_yaml(d)