repo.collect: Add duplicates option.

Defaults to `False`.
If `True`, multiple `outs` sharing a provided `target_path` will not be filtered.

dvc/repo/collect.py

@@ -51,8 +51,8 @@ def _collect_paths(
     return target_paths
 
 
-def _filter_duplicates(
-    outs: Outputs, fs_paths: StrPaths
+def _filter_outs(
+    outs: Outputs, fs_paths: StrPaths, duplicates=False
 ) -> Tuple[Outputs, StrPaths]:
     res_outs: Outputs = []
     fs_res_paths = fs_paths
@@ -61,8 +61,9 @@ def _filter_duplicates(
         fs_path = out.repo.dvcfs.from_os_path(out.fs_path)
         if fs_path in fs_paths:
             res_outs.append(out)
-            # MUTATING THE SAME LIST!!
-            fs_res_paths.remove(fs_path)
+            if not duplicates:
+                # MUTATING THE SAME LIST!!
+                fs_res_paths.remove(fs_path)
 
     return res_outs, fs_res_paths
 
@@ -74,6 +75,7 @@ def collect(
     output_filter: FilterFn = None,
     rev: str = None,
     recursive: bool = False,
+    duplicates: bool = False,
 ) -> Tuple[Outputs, StrPaths]:
     assert targets or output_filter
 
@@ -85,4 +87,4 @@ def collect(
 
     target_paths = _collect_paths(repo, targets, recursive=recursive, rev=rev)
 
-    return _filter_duplicates(outs, target_paths)
+    return _filter_outs(outs, target_paths, duplicates=duplicates)

tests/unit/test_collect.py

@@ -8,3 +8,20 @@ def test_no_file_on_target_rev(tmp_dir, scm, dvc, caplog):
         collect(dvc, targets=["file.yaml"], rev="current_branch")
 
     assert "'file.yaml' was not found at: 'current_branch'." in caplog.text
+
+
+def test_collect_duplicates(tmp_dir, scm, dvc):
+    tmp_dir.gen("params.yaml", "foo: 1\nbar: 2")
+    tmp_dir.gen("foobar", "")
+
+    dvc.run(name="stage-1", cmd="echo stage-1", params=["foo"])
+    dvc.run(name="stage-2", cmd="echo stage-2", params=["bar"])
+
+    outs, _ = collect(dvc, deps=True, targets=["params.yaml"])
+    assert len(outs) == 1
+
+    outs, _ = collect(dvc, deps=True, targets=["params.yaml"], duplicates=True)
+    assert len(outs) == 2
+
+    outs, _ = collect(dvc, deps=True, targets=["foobar"], duplicates=True)
+    assert not outs