dvc.yaml: make persist/cache bool flags in outs/metrics

dvc/schema.py

@@ -2,7 +2,8 @@
 
 from dvc import dependency, output
 from dvc.output import CHECKSUMS_SCHEMA
-from dvc.stage.params import OutputParams, StageParams
+from dvc.output.base import BaseOutput
+from dvc.stage.params import StageParams
 
 STAGES = "stages"
 SINGLE_STAGE_SCHEMA = {
@@ -34,7 +35,24 @@
         Optional(StageParams.PARAM_LOCKED): bool,
         Optional(StageParams.PARAM_META): object,
         Optional(StageParams.PARAM_ALWAYS_CHANGED): bool,
-        **{Optional(p.value): [str] for p in OutputParams},
+        Optional(StageParams.PARAM_OUTS): [
+            Any(
+                str,
+                {
+                    str: {
+                        Optional(BaseOutput.PARAM_CACHE): bool,
+                        Optional(BaseOutput.PARAM_PERSIST): bool,
+                    }
+                },
+            )
+        ],
+        Optional(StageParams.PARAM_OUTS_NO_CACHE): [str],
+        Optional(StageParams.PARAM_METRICS): [
+            Any(str, {str: {Optional(BaseOutput.PARAM_CACHE): bool}})
+        ],
+        Optional(StageParams.PARAM_METRICS_NO_CACHE): [str],
+        Optional(StageParams.PARAM_OUTS_PERSIST): [str],
+        Optional(StageParams.PARAM_OUTS_PERSIST_NO_CACHE): [str],
     }
 }
 MULTI_STAGE_SCHEMA = {STAGES: SINGLE_PIPELINE_STAGE_SCHEMA}

dvc/serialize.py

@@ -24,14 +24,17 @@
 def _get_outs(stage: "PipelineStage"):
     outs_bucket = {}
     for o in sort_by_path(stage.outs):
-        bucket_key = ["metrics"] if o.metric else ["outs"]
+        key = "metrics" if o.metric else "outs"
 
-        if not o.metric and o.persist:
-            bucket_key += ["persist"]
+        extra = {}
+        if o.persist:
+            extra["persist"] = True
         if not o.use_cache:
-            bucket_key += ["no_cache"]
-        key = "_".join(bucket_key)
-        outs_bucket[key] = outs_bucket.get(key, []) + [o.def_path]
+            extra["cache"] = False
+
+        value = {o.def_path: extra} if extra else o.def_path
+
+        outs_bucket[key] = outs_bucket.get(key, []) + [value]
     return [(key, outs_bucket[key]) for key in sorted(outs_bucket.keys())]
 
 

dvc/stage/params.py

@@ -1,6 +1,3 @@
-from enum import Enum
-
-
 class StageParams:
     PARAM_MD5 = "md5"
     PARAM_CMD = "cmd"
@@ -12,11 +9,9 @@ class StageParams:
     PARAM_ALWAYS_CHANGED = "always_changed"
     PARAM_PARAMS = "params"
 
-
-class OutputParams(Enum):
-    PERSIST = "outs_persist"
-    PERSIST_NO_CACHE = "outs_persist_no_cache"
-    METRICS_NO_CACHE = "metrics_no_cache"
-    METRICS = "metrics"
-    NO_CACHE = "outs_no_cache"
-    OUTS = "outs"
+    PARAM_OUTS_PERSIST = "outs_persist"
+    PARAM_OUTS_PERSIST_NO_CACHE = "outs_persist_no_cache"
+    PARAM_METRICS_NO_CACHE = "metrics_no_cache"
+    PARAM_METRICS = "metrics"
+    PARAM_OUTS_NO_CACHE = "outs_no_cache"
+    PARAM_OUTS = "outs"

dvc/stage/utils.py

@@ -13,7 +13,6 @@
     StagePathNotFoundError,
     StagePathOutsideError,
 )
-from .params import OutputParams
 
 
 def check_stage_path(repo, path, is_wdir=False):
@@ -39,15 +38,33 @@ def check_stage_path(repo, path, is_wdir=False):
 def fill_stage_outputs(stage, **kwargs):
     assert not stage.outs
 
+    keys = [
+        "outs_persist",
+        "outs_persist_no_cache",
+        "metrics_no_cache",
+        "metrics",
+        "outs_no_cache",
+        "outs",
+    ]
+
     stage.outs = []
-    for key in (p.value for p in OutputParams):
-        stage.outs += output.loads_from(
-            stage,
-            kwargs.get(key, []),
-            use_cache="no_cache" not in key,
-            persist="persist" in key,
-            metric="metrics" in key,
-        )
+    for key in keys:
+        for entry in kwargs.get(key, []):
+            if isinstance(entry, dict):
+                assert len(entry) == 1
+                path, extra = list(entry.items())[0]
+                stage.outs += output.loadd_from(
+                    stage, [{"path": path, **extra}]
+                )
+                continue
+
+            stage.outs += output.loads_from(
+                stage,
+                [entry],
+                use_cache="no_cache" not in key,
+                persist="persist" in key,
+                metric="metrics" in key,
+            )
 
 
 def fill_stage_dependencies(stage, deps=None, erepo=None, params=None):

tests/func/test_run_multistage.py

@@ -151,7 +151,7 @@ def test_run_dump_on_multistage(tmp_dir, dvc):
             "copy-foo-foo2": {
                 "cmd": "cp foo foo2",
                 "deps": ["foo"],
-                "outs_persist": ["foo2"],
+                "outs": [{"foo2": {"persist": True}}],
                 "always_changed": True,
                 "wdir": "dir",
             },