build: fpm: don't create .build-id/* files

These are leaking into our rpm packages.

Related https://github.com/iterative/dvc-s3-repo/issues/86

dvc/cli/parser.py

@@ -2,8 +2,8 @@
 import argparse
 import logging
 import os
-import sys
 
+from dvc import __version__
 from dvc.commands import (
     add,
     cache,
@@ -128,17 +128,6 @@ class DvcParser(argparse.ArgumentParser):
         return args
 
 
-class VersionAction(argparse.Action):  # pragma: no cover
-    # pylint: disable=too-few-public-methods
-    """Shows DVC version and exits."""
-
-    def __call__(self, parser, namespace, values, option_string=None):
-        from dvc import __version__
-
-        print(__version__)
-        sys.exit(0)
-
-
 def get_parent_parser():
     """Create instances of a parser containing common arguments shared among
     all the commands.
@@ -186,13 +175,11 @@ def get_main_parser():
         help="Show this help message and exit.",
     )
 
-    # NOTE: On some python versions action='version' prints to stderr
-    # instead of stdout https://bugs.python.org/issue18920
     parser.add_argument(
         "-V",
         "--version",
-        action=VersionAction,
-        nargs=0,
+        action="version",
+        version=__version__,
         help="Show program's version.",
     )
 

dvc/commands/experiments/ls.py

@@ -18,15 +18,13 @@ class CmdExperimentsList(CmdBase):
         )
         tags = self.repo.scm.describe(exps)
         remained = {baseline for baseline, tag in tags.items() if tag is None}
-        base = "refs/heads/"
+        base = "refs/heads"
         ref_heads = self.repo.scm.describe(remained, base=base)
 
         for baseline in exps:
-            name = (
-                tags[baseline]
-                or ref_heads[baseline][len(base) :]
-                or baseline[:7]
-            )
+            name = baseline[:7]
+            if tags[baseline] or ref_heads[baseline]:
+                name = tags[baseline] or ref_heads[baseline][len(base) + 1 :]
             if not name_only:
                 print(f"{name}:")
             for exp_name in exps[baseline]:

dvc/output.py

@@ -90,6 +90,7 @@ def loadd_from(stage, d_list):
         remote = d.pop(Output.PARAM_REMOTE, None)
         annot = {field: d.pop(field, None) for field in ANNOTATION_FIELDS}
         files = d.pop(Output.PARAM_FILES, None)
+        push = d.pop(Output.PARAM_PUSH, True)
         ret.append(
             _get(
                 stage,
@@ -103,6 +104,7 @@ def loadd_from(stage, d_list):
                 remote=remote,
                 **annot,
                 files=files,
+                push=push,
             )
         )
     return ret
@@ -117,6 +119,7 @@ def loads_from(
     persist=False,
     checkpoint=False,
     remote=None,
+    push=True,
 ):
     return [
         _get(
@@ -129,6 +132,7 @@ def loads_from(
             persist=persist,
             checkpoint=checkpoint,
             remote=remote,
+            push=push,
         )
         for s in s_list
     ]
@@ -184,6 +188,7 @@ def load_from_pipeline(stage, data, typ="outs"):
                 Output.PARAM_PERSIST,
                 Output.PARAM_CHECKPOINT,
                 Output.PARAM_REMOTE,
+                Output.PARAM_PUSH,
                 *ANNOTATION_FIELDS,
             ],
         )
@@ -251,6 +256,7 @@ class Output:
     PARAM_PLOT_HEADER = "header"
     PARAM_PERSIST = "persist"
     PARAM_REMOTE = "remote"
+    PARAM_PUSH = "push"
 
     METRIC_SCHEMA = Any(
         None,
@@ -284,6 +290,7 @@ class Output:
         repo=None,
         fs_config=None,
         files: List[Dict[str, Any]] = None,
+        push: bool = True,
     ):
         self.annot = Annotation(
             desc=desc, type=type, labels=labels or [], meta=meta or {}
@@ -338,6 +345,7 @@ class Output:
         self.plot = False if self.IS_DEPENDENCY else plot
         self.persist = persist
         self.checkpoint = checkpoint
+        self.can_push = push
 
         self.fs_path = self._parse_path(self.fs, fs_path)
         self.obj: Optional["HashFile"] = None
@@ -776,6 +784,9 @@ class Output:
             if self.remote:
                 ret[self.PARAM_REMOTE] = self.remote
 
+            if not self.can_push:
+                ret[self.PARAM_PUSH] = self.can_push
+
         if (
             (not self.IS_DEPENDENCY or self.stage.is_import)
             and self.hash_info.isdir
@@ -1032,9 +1043,15 @@ class Output:
         if not self.use_cache:
             return {}
 
+        push: bool = kwargs.pop("push", False)
         if self.stage.is_repo_import:
+            if push:
+                return {}
             return self.get_used_external(**kwargs)
 
+        if push and not self.can_push:
+            return {}
+
         if not self.hash_info:
             msg = (
                 "Output '{}'({}) is missing version info. "
@@ -1180,6 +1197,34 @@ class Output:
     def is_plot(self) -> bool:
         return bool(self.plot)
 
+    def restore_fields(self, other: "Output"):
+        """Restore attributes that need to be preserved when serialized."""
+        self.annot = other.annot
+        self.remote = other.remote
+        self.can_push = other.can_push
+
+    def merge_version_meta(self, other: "Output"):
+        """Merge version meta for files which are unchanged from other."""
+        if not self.hash_info:
+            return
+        if self.hash_info.isdir:
+            return self._merge_dir_version_meta(other)
+        if self.hash_info != other.hash_info:
+            return
+        self.meta = other.meta
+
+    def _merge_dir_version_meta(self, other: "Output"):
+        from dvc_data.hashfile.tree import update_meta
+
+        if not self.obj or not other.hash_info.isdir:
+            return
+        other_obj = other.obj if other.obj is not None else other.get_obj()
+        assert isinstance(self.obj, Tree) and isinstance(other_obj, Tree)
+        updated = update_meta(self.obj, other_obj)
+        assert updated.hash_info == self.obj.hash_info
+        self.obj = updated
+        self.files = updated.as_list(with_meta=True)
+
 
 META_SCHEMA = {
     Meta.PARAM_SIZE: int,
@@ -1209,5 +1254,6 @@ SCHEMA = {
     Output.PARAM_CACHE: bool,
     Output.PARAM_METRIC: Output.METRIC_SCHEMA,
     Output.PARAM_REMOTE: str,
+    Output.PARAM_PUSH: bool,
     Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }

dvc/parsing/interpolate.py

@@ -1,3 +1,4 @@
+import os
 import re
 import typing
 from collections.abc import Iterable, Mapping
@@ -70,6 +71,16 @@ def embrace(s: str):
     return BRACE_OPEN + s + BRACE_CLOSE
 
 
+def escape_str(value):
+    if os.name == "nt":
+        from subprocess import list2cmdline
+
+        return list2cmdline([value])
+    from shlex import quote
+
+    return quote(value)
+
+
 @singledispatch
 def to_str(obj) -> str:
     return str(obj)
@@ -97,12 +108,12 @@ def _(obj: dict):
                     result += f"--no-{k} "
 
         elif isinstance(v, str):
-            result += f"--{k} '{v}' "
+            result += f"--{k} {escape_str(v)} "
 
         elif isinstance(v, Iterable):
             for n, i in enumerate(v):
                 if isinstance(i, str):
-                    i = f"'{i}'"
+                    i = escape_str(i)
                 elif isinstance(i, Iterable):
                     raise ParseError(
                         f"Cannot interpolate nested iterable in '{k}'"

dvc/render/convert.py

@@ -14,14 +14,14 @@ from dvc.render.converter.vega import VegaConverter
 
 
 def _get_converter(
-    renderer_class, props
+    renderer_class, renderer_id, props, data
 ) -> Union[VegaConverter, ImageConverter]:
     from dvc_render import ImageRenderer, VegaRenderer
 
     if renderer_class.TYPE == VegaRenderer.TYPE:
-        return VegaConverter(props)
+        return VegaConverter(renderer_id, data, props)
     if renderer_class.TYPE == ImageRenderer.TYPE:
-        return ImageConverter(props)
+        return ImageConverter(renderer_id, data, props)
 
     raise ValueError(f"Invalid renderer class {renderer_class}")
 

dvc/render/converter/__init__.py

@@ -1,9 +1,19 @@
-from typing import Dict, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 
 class Converter:
-    def __init__(self, plot_properties: Optional[Dict] = None):
-        self.plot_properties = plot_properties or {}
+    def __init__(
+        self,
+        plot_id: str,
+        data: Dict[str, Any] = None,
+        properties: Optional[Dict] = None,
+    ):
+        self.plot_id = plot_id
+        self.properties = properties or {}
+        self.data = data or {}
 
-    def convert(self, data, revision: str, filename: str, **kwargs):
+    def convert(self) -> Tuple[List[Tuple[str, str, Any]], Dict]:
+        raise NotImplementedError
+
+    def flat_datapoints(self, revision: str) -> Tuple[List[Dict], Dict]:
         raise NotImplementedError

dvc/render/converter/image.py

@@ -1,6 +1,6 @@
 import base64
 import os
-from typing import TYPE_CHECKING, Dict, List, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Tuple
 
 from dvc.render import FILENAME_FIELD, REVISION_FIELD, SRC_FIELD
 
@@ -34,25 +34,33 @@ class ImageConverter(Converter):
         base64_str = base64.b64encode(image_data).decode()
         return f"data:image;base64,{base64_str}"
 
-    def convert(
-        self, data, revision: str, filename: str, **kwargs
-    ) -> Tuple[List[Dict], Dict]:
+    def convert(self) -> Tuple[List[Tuple[str, str, Any]], Dict]:
+        datas = []
+        for filename, image_data in self.data.items():
+            datas.append((filename, "", image_data))
+        return datas, self.properties
+
+    def flat_datapoints(self, revision: str) -> Tuple[List[Dict], Dict]:
         """
         Convert the DVC Plots content to DVC Render datapoints.
         Return both generated datapoints and updated properties.
         """
-        path = self.plot_properties.get("out")
-        if path:
-            if not os.path.isdir(path):
-                os.makedirs(path, exist_ok=True)
-            src = self._write_image(
-                os.path.abspath(path), revision, filename, data
-            )
-        else:
-            src = self._encode_image(data)
-        datapoint = {
-            REVISION_FIELD: revision,
-            FILENAME_FIELD: filename,
-            SRC_FIELD: src,
-        }
-        return [datapoint], self.plot_properties
+        path = self.properties.get("out")
+        datapoints = []
+        datas, properties = self.convert()
+        for filename, _, image_data in datas:
+            if path:
+                if not os.path.isdir(path):
+                    os.makedirs(path, exist_ok=True)
+                src = self._write_image(
+                    os.path.abspath(path), revision, filename, image_data
+                )
+            else:
+                src = self._encode_image(image_data)
+            datapoint = {
+                REVISION_FIELD: revision,
+                FILENAME_FIELD: filename,
+                SRC_FIELD: src,
+            }
+            datapoints.append(datapoint)
+        return datapoints, properties

dvc/render/converter/vega.py

@@ -1,90 +1,101 @@
 from copy import deepcopy
-from functools import partial
-from typing import Dict, Iterable, List, Optional, Set, Union
+from typing import Any, Dict, Iterable, List, Tuple, Union
 
-from funcy import first, project
+from funcy import first, last
 
 from dvc.exceptions import DvcException
-from dvc.render import (
-    FILENAME_FIELD,
-    INDEX_FIELD,
-    REVISION_FIELD,
-    VERSION_FIELD,
-)
+from dvc.render import FILENAME_FIELD, INDEX_FIELD, VERSION_FIELD
 
 from . import Converter
 
 
-class FieldsNotFoundError(DvcException):
-    def __init__(self, expected_fields, found_fields):
-        expected_str = ", ".join(expected_fields)
+class FieldNotFoundError(DvcException):
+    def __init__(self, expected_field, found_fields):
         found_str = ", ".join(found_fields)
         super().__init__(
-            f"Could not find all provided fields ('{expected_str}') "
+            f"Could not find provided field ('{expected_field}') "
             f"in data fields ('{found_str}')."
         )
 
 
-class PlotDataStructureError(DvcException):
-    def __init__(self):
-        super().__init__(
-            "Plot data extraction failed. Please see "
-            "https://man.dvc.org/plots for supported data formats."
-        )
+def _lists(blob: Union[Dict, List]) -> Iterable[List]:
+    if isinstance(blob, list):
+        yield blob
+    else:
+        for _, value in blob.items():
+            if isinstance(value, dict):
+                yield from _lists(value)
+            elif isinstance(value, list):
+                yield value
 
 
-def _filter_fields(
-    datapoints: List[Dict], fields: Set, **kwargs
-) -> List[Dict]:
-    if not fields:
-        return datapoints
-    assert isinstance(fields, set)
-
-    new_data = []
-    for data_point in datapoints:
-        keys = set(data_point.keys())
-        if not fields <= keys:
-            raise FieldsNotFoundError(fields, keys)
-
-        new_data.append(project(data_point, fields))
-
-    return new_data
+def _file_field(*args):
+    for axis_def in args:
+        if axis_def is not None:
+            for file, val in axis_def.items():
+                if isinstance(val, str):
+                    yield file, val
+                elif isinstance(val, list):
+                    for field in val:
+                        yield file, field
 
 
-def _lists(dictionary: Dict):
-    for _, value in dictionary.items():
-        if isinstance(value, dict):
-            yield from _lists(value)
-        elif isinstance(value, list):
-            yield value
+def _find(
+    filename: str,
+    field: str,
+    data_series: List[Tuple[str, str, Any]],
+):
+    for data_file, data_field, data in data_series:
+        if data_file == filename and data_field == field:
+            return data_file, data_field, data
+    return None
 
 
-def _find_first_list(
-    data: Union[Dict, List], fields: Set, **kwargs
-) -> List[Dict]:
-    fields = fields or set()
-
-    if not isinstance(data, dict):
-        return data
-
-    for lst in _lists(data):
-        if (
-            all(isinstance(dp, dict) for dp in lst)
-            # if fields is empty, it will match any set
-            and set(first(lst).keys()) & fields == fields
-        ):
-            return lst
-
-    raise PlotDataStructureError()
+def _verify_field(file2datapoints: Dict[str, List], filename: str, field: str):
+    if filename in file2datapoints:
+        datapoint = first(file2datapoints[filename])
+        if field not in datapoint:
+            raise FieldNotFoundError(field, datapoint.keys())
+    return
 
 
-def _append_index(datapoints: List[Dict], **kwargs) -> List[Dict]:
-    if INDEX_FIELD in first(datapoints).keys():
-        return datapoints
+def _get_x(properties: Dict, file2datapoints: Dict[str, List[Dict]]):
+    x = properties.get("x", None)
+    if x is not None and isinstance(x, dict):
+        filename, field = first(_file_field(x))
+        _verify_field(file2datapoints, filename, field)
+        return filename, field
+    return None
 
-    for index, data_point in enumerate(datapoints):
-        data_point[INDEX_FIELD] = index
-    return datapoints
+
+def _get_ys(properties, file2datapoints: Dict[str, List[Dict]]):
+    y = properties.get("y", None)
+    if y is not None:
+        for filename, field in _file_field(y):
+            _verify_field(file2datapoints, filename, field)
+            yield filename, field
+
+
+def _is_datapoints(lst: List[Dict]):
+    """
+    check if dict keys match, datapoints with different keys mgiht lead
+    to unexpected behavior
+    """
+
+    return all(isinstance(item, dict) for item in lst) and set(
+        first(lst).keys()
+    ) == {key for keys in lst for key in keys}
+
+
+def get_datapoints(file_content: Dict):
+    result: List[Dict[str, Any]] = []
+    for lst in _lists(file_content):
+        if _is_datapoints(lst):
+            for index, datapoint in enumerate(lst):
+                if len(result) <= index:
+                    result.append({})
+                result[index].update(datapoint)
+    return result
 
 
 class VegaConverter(Converter):
@@ -95,228 +106,215 @@ class VegaConverter(Converter):
     ('x', 'y') it will attempt to fill in the blanks.
     """
 
-    def __init__(self, plot_properties: Optional[Dict] = None):
-        super().__init__(plot_properties)
+    def __init__(
+        self, plot_id: str, data: Dict = None, properties: Dict = None
+    ):
+        super().__init__(plot_id, data, properties)
+        self.plot_id = plot_id
         self.inferred_properties: Dict = {}
 
-        self.steps = []
+        # TODO we should be handling that in `convert`,
+        #      to avoid stateful `self.inferred_properties`
+        self._infer_x_y()
 
-        self._infer_x()
-        self._infer_fields()
+    def _infer_y_from_data(self):
+        if self.plot_id in self.data:
+            for lst in _lists(self.data[self.plot_id]):
+                if all(isinstance(item, dict) for item in lst):
+                    datapoint = first(lst)
+                    field = last(datapoint.keys())
+                    self.inferred_properties["y"] = {self.plot_id: field}
+                    break
 
-        self.steps.append(
-            (
-                "find_data",
-                partial(
-                    _find_first_list,
-                    fields=self.inferred_properties.get("fields", set())
-                    - {INDEX_FIELD},
-                ),
-            )
-        )
+    def _infer_x_y(self):
+        def _infer_files(from_name, to_name):
+            from_value = self.properties.get(from_name, None)
+            to_value = self.properties.get(to_name, None)
 
-        if not self.plot_properties.get("x", None):
-            self.steps.append(("append_index", partial(_append_index)))
-
-        self.steps.append(
-            (
-                "filter_fields",
-                partial(
-                    _filter_fields,
-                    fields=self.inferred_properties.get("fields", set()),
-                ),
-            )
-        )
-        self.steps.append(
-            (
-                "infer_y",
-                partial(
-                    self._infer_y,
-                ),
-            )
-        )
-
-        self.steps.append(
-            (
-                "generate_y",
-                partial(
-                    self._generate_y_values,
-                ),
-            )
-        )
-
-    def _infer_x(self):
-        if not self.plot_properties.get("x", None):
-            self.inferred_properties["x"] = INDEX_FIELD
-
-    def skip_step(self, name: str):
-        self.steps = [(_name, fn) for _name, fn in self.steps if _name != name]
-
-    def _infer_fields(self):
-        fields = self.plot_properties.get("fields", set())
-        if fields:
-            fields = {
-                *fields,
-                self.plot_properties.get("x", None),
-                self.plot_properties.get("y", None),
-                self.inferred_properties.get("x", None),
-            } - {None}
-            self.inferred_properties["fields"] = fields
-
-    def _infer_y(self, datapoints: List[Dict], **kwargs):
-        if "y" not in self.plot_properties:
-            data_fields = list(first(datapoints))
-            skip = (
-                REVISION_FIELD,
-                self.plot_properties.get("x", None)
-                or self.inferred_properties.get("x"),
-                FILENAME_FIELD,
-                VERSION_FIELD,
-            )
-            inferred_y = first(
-                f for f in reversed(data_fields) if f not in skip
-            )
-            if "y" in self.inferred_properties:
-                previous_y = self.inferred_properties["y"]
-                if previous_y != inferred_y:
-                    raise DvcException(
-                        f"Inferred y ('{inferred_y}' value does not match"
-                        f"previously matched one ('f{previous_y}')."
-                    )
+            if isinstance(to_value, str):
+                self.inferred_properties[to_name] = {}
+                if isinstance(from_value, dict):
+                    for file in from_value.keys():
+                        self.inferred_properties[to_name][file] = to_value
+                else:
+                    self.inferred_properties[to_name][self.plot_id] = to_value
             else:
-                self.inferred_properties["y"] = inferred_y
-        return datapoints
+                self.inferred_properties[to_name] = to_value
+
+        _infer_files("y", "x")
+        _infer_files("x", "y")
+
+        if self.inferred_properties.get("y", None) is None:
+            self._infer_y_from_data()
+
+    def _find_datapoints(self):
+        result = {}
+        for file, content in self.data.items():
+            result[file] = get_datapoints(content)
+
+        return result
+
+    @staticmethod
+    def infer_y_label(properties):
+        y_label = properties.get("y_label", None)
+        if y_label is None:
+            y = properties.get("y", None)
+            if isinstance(y, dict):
+                fields = {field for _, field in _file_field(y)}
+                if len(fields) == 1:
+                    y_label = first(fields)
+                else:
+                    y_label = "y"
+            elif isinstance(y, list):
+                y_label = "y"
+            elif isinstance(y, str):
+                y_label = y
+
+        return y_label
+
+    @staticmethod
+    def infer_x_label(properties):
+        x_label = properties.get("x_label", None)
+        if x_label is None:
+            x = properties.get("x", None)
+            if isinstance(x, dict):
+                fields = {field for _, field in _file_field(x)}
+                if len(fields) == 1:
+                    x_label = first(fields)
+                else:
+                    x_label = "x"
+            else:
+                x_label = INDEX_FIELD
+        return x_label
+
+    def flat_datapoints(self, revision):
+
+        file2datapoints, properties = self.convert()
+
+        props_update = {}
+
+        x = _get_x(properties, file2datapoints)
+
+        # assign "step" if no x provided
+        if not x:
+            x_file, x_field = (
+                None,
+                INDEX_FIELD,
+            )
+        else:
+            x_file, x_field = x
+        props_update["x"] = x_field
+
+        ys = list(_get_ys(properties, file2datapoints))
+
+        all_datapoints = []
+        all_y_fields = {y_field for _, y_field in ys}
+
+        # override to unified y field name if there are different y fields
+        if len(all_y_fields) > 1:
+            props_update["y"] = "dvc_inferred_y_value"
+        else:
+            props_update["y"] = first(all_y_fields)
+
+        for y_file, y_field in ys:
+            datapoints = deepcopy(file2datapoints.get(y_file, []))
+
+            if props_update.get("y", None) == "dvc_inferred_y_value":
+                _update_from_field(
+                    datapoints,
+                    field="dvc_inferred_y_value",
+                    source_field=y_field,
+                )
+
+            if x_field == INDEX_FIELD and x_file is None:
+                _update_from_index(datapoints, INDEX_FIELD)
+            else:
+                x_datapoints = file2datapoints.get(x_file, [])
+                try:
+                    _update_from_field(
+                        datapoints,
+                        field=x_field,
+                        source_datapoints=x_datapoints,
+                    )
+                except IndexError:
+                    raise DvcException(
+                        f"Cannot join '{x_field}' from '{x_file}' and "
+                        "'{y_field}' from '{y_file}'. "
+                        "They have to have same length."
+                    )
+
+            _update_all(
+                datapoints,
+                update_dict={
+                    VERSION_FIELD: {
+                        "revision": revision,
+                        FILENAME_FIELD: y_file,
+                        "field": y_field,
+                    }
+                },
+            )
+
+            all_datapoints.extend(datapoints)
+
+        if not all_datapoints:
+            return [], {}
+
+        properties = {**properties, **props_update}
+
+        return all_datapoints, properties
 
     def convert(
         self,
-        data,
-        revision: str,
-        filename: str,
-        skip: List = None,
-        **kwargs,
     ):
         """
         Convert the data. Fill necessary fields ('x', 'y') and return both
-        generated datapoints and updated properties.
+        generated datapoints and updated properties. If `x` is not provided,
+        leave it as None, fronteds should handle it.
+
+        NOTE: Studio uses this method.
+              The only thing studio FE handles is filling `x` and `y`.
+              `x/y_label` should be filled here.
+
+              Datapoints are not stripped according to config, because users
+              might be utilizing other fields in their custom plots.
         """
-        if not skip:
-            skip = []
+        datapoints = self._find_datapoints()
+        properties = {**self.properties, **self.inferred_properties}
 
-        processed = deepcopy(data)
+        properties["y_label"] = self.infer_y_label(properties)
+        properties["x_label"] = self.infer_x_label(properties)
 
-        for step_name, step in self.steps:
-            if step_name not in skip:
-                processed = step(  # type: ignore
-                    processed,
-                    revision=revision,
-                    filename=filename,
-                )
+        return datapoints, properties
 
-        return processed, {**self.plot_properties, **self.inferred_properties}
 
-    def _generate_y_values(  # noqa: C901
-        self,
-        datapoints: List[Dict],
-        revision: str,
-        filename: str,
-        **kwargs,
-    ) -> List[Dict]:
+def _update_from_field(
+    target_datapoints: List[Dict],
+    field: str,
+    source_datapoints: List[Dict] = None,
+    source_field: str = None,
+):
+    if source_datapoints is None:
+        source_datapoints = target_datapoints
+    if source_field is None:
+        source_field = field
 
-        y_values = self.plot_properties.get(
-            "y", None
-        ) or self.inferred_properties.get("y", None)
+    if len(source_datapoints) != len(target_datapoints):
+        raise IndexError(
+            "Source and target datapoints must have the same length"
+        )
 
-        assert y_values is not None
+    for index, datapoint in enumerate(target_datapoints):
+        source_datapoint = source_datapoints[index]
+        if source_field in source_datapoint:
+            datapoint[field] = source_datapoint[source_field]
 
-        result = []
-        properties_update = {}
 
-        def _add_version_info(datapoint, version_info):
-            tmp = datapoint.copy()
-            tmp[VERSION_FIELD] = version_info
-            return tmp
+def _update_from_index(datapoints: List[Dict], new_field: str):
+    for index, datapoint in enumerate(datapoints):
+        datapoint[new_field] = index
 
-        def _version_info(revision, filename=None, field=None):
-            res = {"revision": revision}
-            if filename is not None:
-                res["filename"] = filename
-            if field is not None:
-                res["field"] = field
-            return res
 
-        def _generate_y(datapoint, field):
-            tmp = datapoint.copy()
-            tmp["dvc_inferred_y_value"] = datapoint[field]
-            tmp = _add_version_info(
-                tmp, _version_info(revision, filename, field)
-            )
-            if (
-                "y_label" not in properties_update
-                and "y_label" not in self.plot_properties
-            ):
-                properties_update["y_label"] = "y"
+def _update_all(datapoints: List[Dict], update_dict: Dict):
 
-            properties_update["y"] = "dvc_inferred_y_value"
-
-            return tmp
-
-        if isinstance(y_values, str):
-            for datapoint in datapoints:
-                result.append(
-                    _add_version_info(
-                        datapoint, _version_info(revision, filename, y_values)
-                    )
-                )
-
-        if isinstance(y_values, list):
-            for datapoint in datapoints:
-                for y_val in y_values:
-                    if y_val in datapoint:
-                        result.append(_generate_y(datapoint, y_val))
-
-        if isinstance(y_values, dict):
-
-            def _to_set(values: Iterable):
-                result = set()
-                for val in values:
-                    if isinstance(val, list):
-                        for elem in val:
-                            result.add(elem)
-                    else:
-                        result.add(val)
-
-                return result
-
-            all_fields = _to_set(y_values.values())
-            if (
-                all([isinstance(field, str) for field in all_fields])
-                and len(all_fields) == 1
-            ):
-                # if we use the same field from all files,
-                # we dont have to generate it
-                y_field = all_fields.pop()
-                for datapoint in datapoints:
-                    result.append(
-                        _add_version_info(
-                            datapoint,
-                            _version_info(revision, filename, y_field),
-                        )
-                    )
-                properties_update.update({"y": y_field})
-            else:
-                for def_filename, val in y_values.items():
-                    if isinstance(val, str):
-                        fields = [val]
-                    if isinstance(val, list):
-                        fields = val
-                    for datapoint in datapoints:
-                        for field in fields:
-                            if field in datapoint and def_filename in filename:
-                                result.append(_generate_y(datapoint, field))
-
-        self.inferred_properties = {
-            **self.inferred_properties,
-            **properties_update,
-        }
-
-        return result
+    for datapoint in datapoints:
+        datapoint.update(update_dict)

dvc/render/match.py

@@ -42,7 +42,7 @@ class PlotsData:
         return dict(groups)
 
     def get_definition_data(self, target_files, rev):
-        result = []
+        result = {}
         for file in target_files:
             file_content = (
                 self.data.get(rev, {})
@@ -52,7 +52,7 @@ class PlotsData:
                 .get("data", {})
             )
             if file_content:
-                result.append((file, file_content))
+                result[file] = file_content
 
         return result
 
@@ -91,15 +91,14 @@ def match_defs_renderers(
                 renderer_cls = VegaRenderer
                 renderer_id = plot_id
 
-            converter = _get_converter(renderer_cls, props)
+            converter = _get_converter(
+                renderer_cls, inner_id, props, definitions_data
+            )
 
-            for filename, plot_data in definitions_data:
-                dps, final_props = converter.convert(
-                    revision=rev,
-                    filename=filename,
-                    data=plot_data,
-                )
-                plot_datapoints.extend(dps)
+            dps, rev_props = converter.flat_datapoints(rev)
+            if not final_props and rev_props:
+                final_props = rev_props
+            plot_datapoints.extend(dps)
 
         if "title" not in final_props:
             final_props["title"] = renderer_id

dvc/repo/__init__.py

@@ -401,6 +401,7 @@ class Repo:
         used_run_cache=None,
         revs=None,
         num=1,
+        push: bool = False,
     ):
         """Get the stages related to the given target and collect
         the `info` of its outputs.
@@ -435,6 +436,7 @@ class Repo:
                 jobs=jobs,
                 recursive=recursive,
                 with_deps=with_deps,
+                push=push,
             ).items():
                 used[odb].update(objs)
 

dvc/repo/add.py

@@ -187,7 +187,7 @@ def add(  # noqa: C901
                 stage.transfer(source, to_remote=to_remote, odb=odb, **kwargs)
             else:
                 try:
-                    stage.save()
+                    stage.save(merge_versioned=True)
                     if not no_commit:
                         stage.commit()
                 except CacheLinkError:

dvc/repo/experiments/queue/celery.py

@@ -15,7 +15,7 @@ from typing import (
 )
 
 from celery.result import AsyncResult
-from funcy import cached_property
+from funcy import cached_property, first
 from kombu.message import Message
 
 from dvc.daemon import daemonize
@@ -25,6 +25,7 @@ from dvc.ui import ui
 from ..exceptions import UnresolvedQueueExpNamesError
 from ..executor.base import EXEC_TMP_DIR, ExecutorInfo, ExecutorResult
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
+from .exceptions import CannotKillTasksError
 from .tasks import run_exp
 from .utils import fetch_running_exp_from_temp_dir
 
@@ -286,24 +287,70 @@ class LocalCeleryQueue(BaseStashQueue):
         # out of the active task list, and needs to be loaded here.
         return self._get_done_result(entry, timeout)
 
+    def _get_running_task_ids(self) -> Set[str]:
+        running_task_ids: Set[str] = set()
+        active_workers = self.worker_status()
+        for _, tasks in active_workers.items():
+            task = first(tasks)
+            if task:
+                running_task_ids.add(task["id"])
+        return running_task_ids
+
+    def _try_to_kill_tasks(
+        self, to_kill: Dict[QueueEntry, str]
+    ) -> Dict[QueueEntry, str]:
+        fail_to_kill_entries: Dict[QueueEntry, str] = {}
+        for queue_entry, rev in to_kill.items():
+            try:
+                self.proc.kill(queue_entry.stash_rev)
+                logger.debug(f"Task {rev} had been killed.")
+            except ProcessLookupError:
+                fail_to_kill_entries[queue_entry] = rev
+        return fail_to_kill_entries
+
+    def _mark_inactive_tasks_failure(self, remained_entries):
+        remained_revs: List[str] = []
+        running_ids = self._get_running_task_ids()
+        logger.debug(f"Current running tasks ids: {running_ids}.")
+        for msg, entry in self._iter_processed():
+            if entry not in remained_entries:
+                continue
+            task_id = msg.headers["id"]
+            if task_id in running_ids:
+                remained_revs.append(remained_entries[entry])
+            else:
+                result: AsyncResult = AsyncResult(task_id)
+                if not result.ready():
+                    logger.debug(
+                        f"Task id {task_id} rev {remained_entries[entry]} "
+                        "marked as failure."
+                    )
+                    self.celery.backend.mark_as_failure(task_id, None)
+
+        if remained_revs:
+            raise CannotKillTasksError(remained_revs)
+
     def kill(self, revs: Collection[str]) -> None:
-        to_kill: Set[QueueEntry] = set()
         name_dict: Dict[
             str, Optional[QueueEntry]
         ] = self.match_queue_entry_by_name(set(revs), self.iter_active())
 
-        missing_rev: List[str] = []
+        to_kill: Dict[QueueEntry, str] = {}
+        missing_revs: List[str] = []
         for rev, queue_entry in name_dict.items():
             if queue_entry is None:
-                missing_rev.append(rev)
+                missing_revs.append(rev)
             else:
-                to_kill.add(queue_entry)
+                to_kill[queue_entry] = rev
+        if missing_revs:
+            raise UnresolvedQueueExpNamesError(missing_revs)
 
-        if missing_rev:
-            raise UnresolvedQueueExpNamesError(missing_rev)
+        fail_to_kill_entries: Dict[QueueEntry, str] = self._try_to_kill_tasks(
+            to_kill
+        )
 
-        for queue_entry in to_kill:
-            self.proc.kill(queue_entry.stash_rev)
+        if fail_to_kill_entries:
+            self._mark_inactive_tasks_failure(fail_to_kill_entries)
 
     def shutdown(self, kill: bool = False):
         self.celery.control.shutdown()
@@ -359,10 +406,11 @@ class LocalCeleryQueue(BaseStashQueue):
         ) as fobj:
             ui.write(fobj.read())
 
-    def worker_status(self) -> Dict:
+    def worker_status(self) -> Dict[str, List[Dict]]:
         """Return the current active celery worker"""
         status = self.celery.control.inspect().active() or {}
         logger.debug(f"Worker status: {status}")
+
         return status
 
     def clear(self, *args, **kwargs):

dvc/repo/experiments/queue/exceptions.py

@@ -0,0 +1,12 @@
+from typing import Collection
+
+from dvc.exceptions import DvcException
+
+
+class CannotKillTasksError(DvcException):
+    def __init__(self, revs: Collection[str]):
+        rev_str = ",".join(revs)
+        super().__init__(
+            f"Task '{rev_str}' is initializing, please wait a few seconds "
+            "until the experiments start running to retry the kill operation."
+        )

dvc/repo/experiments/queue/tempdir.py

@@ -77,9 +77,8 @@ class TempDirQueue(WorkspaceQueue):
         for stash_rev in self.proc:
             infofile = self.get_infofile_path(stash_rev)
             executor_info = ExecutorInfo.load_json(infofile)
-            if (
-                not executor_info.status <= TaskStatus.SUCCESS
-                and os.path.exists(executor_info.root_dir)
+            if executor_info.status <= TaskStatus.SUCCESS and os.path.exists(
+                executor_info.root_dir
             ):
                 yield QueueEntry(
                     self.repo.root_dir,

dvc/repo/experiments/queue/utils.py

@@ -41,7 +41,7 @@ def fetch_running_exp_from_temp_dir(
         if info.git_url and fetch_refs and info.status > TaskStatus.PREPARING:
 
             def on_diverged(_ref: str, _checkpoint: bool):
-                return False
+                return True
 
             executor = TempDirExecutor.from_info(info)
             try:

dvc/repo/experiments/show.py

@@ -8,8 +8,10 @@ from typing import (
     Any,
     Callable,
     Dict,
+    Iterable,
     List,
     Optional,
+    Set,
     Tuple,
     Union,
 )
@@ -24,6 +26,7 @@ from dvc.utils import error_handler, onerror_collect, relpath
 from .refs import ExpRefInfo
 
 if TYPE_CHECKING:
+    from scmrepo.git import Git
     from scmrepo.git.objects import GitCommit
 
     from dvc.repo import Repo
@@ -134,6 +137,7 @@ def _collect_complete_experiment(
 
     checkpoint: bool = True if len(revs) > 1 else False
     prev = ""
+
     for rev in revs:
         status = ExpStatus.Running if rev in running else ExpStatus.Success
         collected_exp = collect_experiment_commit(
@@ -215,41 +219,55 @@ def _collect_branch(
     return results
 
 
-def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]):
+def get_branch_names(
+    scm: "Git", baseline_set: Iterable[str]
+) -> Dict[str, Optional[str]]:
+    names: Dict[str, Optional[str]] = {}
+    for base in [
+        f"refs/exps/{baseline[:2]}/{baseline[2:]}/"
+        for baseline in baseline_set
+    ] + ["refs/heads/", "refs/tags/"]:
+        for ref in scm.iter_refs(base=base):
+            if ref:
+                try:
+                    rev = scm.get_ref(ref)
+                    names[rev] = ref[len(base) :]
+                except KeyError:
+                    logger.debug("unreosolved ref %s", ref)
+    logger.debug("found refs for revs: %s", names)
+    return names
+
+
+def update_names(
+    repo: "Repo",
+    branch_names: Dict[str, Optional[str]],
+    result: Dict[str, Dict[str, Any]],
+):
 
     rev_set = set()
-    baseline_set = set()
     for baseline in result:
         for rev in result[baseline]:
             if rev == "baseline":
                 rev = baseline
-                baseline_set.add(rev)
             if rev != "workspace":
                 rev_set.add(rev)
 
-    names: Dict[str, Optional[str]] = {}
-    for base in ("refs/tags/", "refs/heads/"):
-        if rev_set:
-            names.update(
-                (rev, ref[len(base) :])
-                for rev, ref in repo.scm.describe(
-                    baseline_set, base=base
-                ).items()
-                if ref is not None
-            )
-            rev_set.difference_update(names.keys())
+    if rev_set:
+        rev_set.difference_update(branch_names.keys())
 
     exact_name = repo.experiments.get_exact_name(rev_set)
 
     for baseline, baseline_results in result.items():
+        name_set: Set[str] = set()
         for rev, rev_result in baseline_results.items():
             name: Optional[str] = None
             if rev == "baseline":
                 rev = baseline
                 if rev == "workspace":
                     continue
-            name = names.get(rev, None) or exact_name[rev]
-            if name:
+            name = branch_names.get(rev, None) or exact_name[rev]
+            if name and name not in name_set:
+                name_set.add(name)
                 rev_result["data"]["name"] = name
 
 
@@ -336,6 +354,40 @@ def update_new(
             to_dict[baseline][rev] = to_dict[baseline].get(rev, experiment)
 
 
+def move_properties_to_head(result: Dict[str, Dict[str, Dict[str, Any]]]):
+    for _, baseline_results in result.items():
+        checkpoint: bool = False
+        head: Dict[str, Any] = {}
+        for rev, rev_data in baseline_results.items():
+            if (
+                "data" not in rev_data
+                or rev_data["data"].get("checkpoint_tip", None) is None
+            ):
+                checkpoint = False
+                head = {}
+                continue
+
+            rev_result: Dict[str, Any] = rev_data["data"]
+            if (
+                checkpoint is True
+                and rev_result["checkpoint_tip"] == head["checkpoint_tip"]
+            ):
+                if "name" in rev_result and "name" not in head:
+                    head["name"] = rev_result["name"]
+                    del rev_result["name"]
+                if rev_result["executor"]:
+                    if not head["executor"]:
+                        head["executor"] = rev_result["executor"]
+                    rev_result["executor"] = None
+                if rev_result["status"] == ExpStatus.Running.name:
+                    head["status"] = ExpStatus.Running.name
+                    rev_result["status"] = ExpStatus.Success.name
+            else:
+                if rev_result["checkpoint_tip"] == rev:
+                    head = rev_result
+                    checkpoint = True
+
+
 def show(
     repo: "Repo",
     all_branches=False,
@@ -354,8 +406,7 @@ def show(
     if repo.scm.no_commits:
         return {}
 
-    if onerror is None:
-        onerror = _show_onerror_collect
+    onerror = onerror or _show_onerror_collect
 
     res: Dict[str, Dict] = defaultdict(OrderedDict)
 
@@ -368,6 +419,7 @@ def show(
     found_revs.update(
         iter_revs(repo.scm, revs, num, all_branches, all_tags, all_commits)
     )
+    branch_names = get_branch_names(repo.scm, found_revs)
 
     running: Dict[str, Dict] = repo.experiments.get_running_exps(
         fetch_refs=fetch_running
@@ -421,6 +473,8 @@ def show(
     update_new(res, queued_experiment)
 
     if not sha_only:
-        get_names(repo, res)
+        update_names(repo, branch_names, res)
+
+    move_properties_to_head(res)
 
     return res

dvc/repo/fetch.py

@@ -17,20 +17,6 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-def _fetch_worktree(repo, remote):
-    from dvc_data.index import save
-
-    index = repo.index.data["repo"]
-    for key, entry in index.iteritems():
-        entry.fs = remote.fs
-        entry.path = remote.fs.path.join(
-            remote.path,
-            *key,
-        )
-    save(index)
-    return len(index)
-
-
 @locked
 def fetch(
     self,
@@ -59,6 +45,7 @@ def fetch(
             remote is configured
     """
     from dvc.repo.imports import save_imports
+    from dvc.repo.worktree import fetch_worktree
     from dvc_data.hashfile.transfer import TransferResult
 
     if isinstance(targets, str):
@@ -67,7 +54,7 @@ def fetch(
     with suppress(NoRemoteError):
         _remote = self.cloud.get_remote(name=remote)
         if _remote.worktree:
-            return _fetch_worktree(self, _remote)
+            return fetch_worktree(self, _remote)
 
     failed_count = 0
 

dvc/repo/imp_url.py

@@ -97,6 +97,9 @@ def imp_url(
     else:
         stage.run(jobs=jobs, no_download=no_download)
 
+    if not no_exec and stage.deps[0].fs.version_aware:
+        stage.outs[0].can_push = False
+
     stage.frozen = frozen
 
     dvcfile.dump(stage)

dvc/repo/index.py

@@ -152,7 +152,7 @@ class Index:
 
     def targets_view(
         self,
-        targets: "TargetType",
+        targets: Optional["TargetType"],
         stage_filter: Optional[Callable[["Stage"], bool]] = None,
         outs_filter: Optional[Callable[["Output"], bool]] = None,
         **kwargs: Any,
@@ -268,6 +268,7 @@ class Index:
         force: bool = False,
         recursive: bool = False,
         jobs: int = None,
+        push: bool = False,
     ) -> "ObjectContainer":
         from collections import defaultdict
 
@@ -281,6 +282,7 @@ class Index:
                 force=force,
                 jobs=jobs,
                 filter_info=filter_info,
+                push=push,
             ).items():
                 used[odb].update(objs)
         return used

dvc/repo/plots/__init__.py

@@ -19,7 +19,7 @@ from typing import (
 
 import dpath.options
 import dpath.util
-from funcy import cached_property, first, project
+from funcy import cached_property, distinct, first, project
 
 from dvc.exceptions import DvcException
 from dvc.utils import error_handler, errored_revisions, onerror_collect
@@ -314,16 +314,18 @@ def _get_data_targets(definitions: Dict):
 
 
 def infer_data_sources(plot_id, config=None):
-    def _deduplicate(lst: List):
-        return list({elem: None for elem in lst}.keys())
-
     y = config.get("y", None)
+
     if isinstance(y, dict):
         sources = list(y.keys())
     else:
         sources = [plot_id]
 
-    return _deduplicate(source for source in sources)
+    x = config.get("x", None)
+    if isinstance(x, dict):
+        sources.append(first(x.keys()))
+
+    return distinct(source for source in sources)
 
 
 def _matches(targets, config_file, plot_id):

dvc/repo/push.py

@@ -9,48 +9,6 @@ if TYPE_CHECKING:
     from dvc_objects.db.base import ObjectDB
 
 
-def _push_worktree(repo, remote):
-    from dvc_data.index import checkout
-    from dvc_data.index.save import build_tree
-
-    index = repo.index.data["repo"]
-    checkout(index, remote.path, remote.fs)
-
-    for stage in repo.index.stages:
-        for out in stage.outs:
-            if not out.use_cache:
-                continue
-
-            if not out.is_in_repo:
-                continue
-
-            workspace, key = out.index_key
-            index = repo.index.data[workspace]
-            entry = index[key]
-            if out.isdir():
-                old_tree = out.get_obj()
-                entry.hash_info = old_tree.hash_info
-                entry.meta = out.meta
-                for subkey, entry in index.iteritems(key):
-                    if entry.meta.isdir:
-                        continue
-                    fs_path = repo.fs.path.join(repo.root_dir, *subkey)
-                    _, hash_info = old_tree.get(
-                        repo.fs.path.relparts(fs_path, out.fs_path)
-                    )
-                    entry.hash_info = hash_info
-                tree_meta, new_tree = build_tree(index, key)
-                out.obj = new_tree
-                out.hash_info = new_tree.hash_info
-                out.meta = tree_meta
-            else:
-                out.hash_info = entry.hash_info
-                out.meta = entry.meta
-        stage.dvcfile.dump(stage, with_files=True, update_pipeline=False)
-
-    return len(index)
-
-
 @locked
 def push(
     self,
@@ -68,9 +26,11 @@ def push(
     odb: Optional["ObjectDB"] = None,
     include_imports=False,
 ):
+    from dvc.repo.worktree import push_worktree
+
     _remote = self.cloud.get_remote(name=remote)
     if _remote.worktree:
-        return _push_worktree(self, _remote)
+        return push_worktree(self, _remote)
 
     used_run_cache = (
         self.stage_cache.push(remote, odb=odb) if run_cache else []
@@ -93,6 +53,7 @@ def push(
         recursive=recursive,
         used_run_cache=used_run_cache,
         revs=revs,
+        push=True,
     )
 
     pushed = len(used_run_cache)

dvc/repo/status.py

@@ -89,6 +89,7 @@ def _cloud_status(
         remote=remote,
         jobs=jobs,
         recursive=recursive,
+        push=True,
     )
 
     ret = {}

dvc/repo/worktree.py

@@ -0,0 +1,121 @@
+from typing import TYPE_CHECKING, Optional
+
+from dvc.fs.callbacks import Callback
+
+if TYPE_CHECKING:
+    from dvc.cloud import Remote
+    from dvc.index import Index, IndexView
+    from dvc.output import Output
+    from dvc.repo import Repo
+    from dvc.stage import Stage
+    from dvc.types import TargetType
+
+
+def worktree_view(
+    index: "Index",
+    targets: Optional["TargetType"] = None,
+    push: bool = False,
+    **kwargs,
+) -> "IndexView":
+    """Return view of data that can be stored in worktree remotes.
+
+    Args:
+        targets: Optional targets.
+        push: Whether the view should be restricted to pushable data only.
+
+    Additional kwargs will be passed into target collection.
+    """
+
+    def stage_filter(stage: "Stage") -> bool:
+        if push and stage.is_repo_import:
+            return False
+        return True
+
+    def outs_filter(out: "Output") -> bool:
+        if (
+            not out.is_in_repo
+            or not out.use_cache
+            or (push and not out.can_push)
+        ):
+            return False
+        return True
+
+    return index.targets_view(
+        targets,
+        stage_filter=stage_filter,
+        outs_filter=outs_filter,
+        **kwargs,
+    )
+
+
+def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
+    from dvc_data.index import save
+
+    view = worktree_view(repo.index)
+    index = view.data["repo"]
+    for key, entry in index.iteritems():
+        entry.fs = remote.fs
+        entry.path = remote.fs.path.join(
+            remote.path,
+            *key,
+        )
+    total = len(index)
+    with Callback.as_tqdm_callback(
+        unit="file", desc="Fetch", disable=total == 0
+    ) as cb:
+        cb.set_size(total)
+        return save(index, callback=cb)
+
+
+def push_worktree(repo: "Repo", remote: "Remote") -> int:
+    from dvc_data.index import checkout
+    from dvc_data.index.save import build_tree
+
+    view = worktree_view(repo.index, push=True)
+    index = view.data["repo"]
+    total = len(index)
+    with Callback.as_tqdm_callback(
+        unit="file", desc="Push", disable=total == 0
+    ) as cb:
+        cb.set_size(total)
+        pushed = checkout(
+            index,
+            remote.path,
+            remote.fs,
+            latest_only=False,
+            callback=cb,
+        )
+
+    for stage in view.stages:
+        for out in stage.outs:
+            if not out.use_cache:
+                continue
+
+            if not out.is_in_repo:
+                continue
+
+            workspace, key = out.index_key
+            index = repo.index.data[workspace]
+            entry = index[key]
+            if out.isdir():
+                old_tree = out.get_obj()
+                entry.hash_info = old_tree.hash_info
+                entry.meta = out.meta
+                for subkey, entry in index.iteritems(key):
+                    if entry.meta.isdir:
+                        continue
+                    fs_path = repo.fs.path.join(repo.root_dir, *subkey)
+                    _, hash_info = old_tree.get(
+                        repo.fs.path.relparts(fs_path, out.fs_path)
+                    )
+                    entry.hash_info = hash_info
+                tree_meta, new_tree = build_tree(index, key)
+                out.obj = new_tree
+                out.hash_info = new_tree.hash_info
+                out.meta = tree_meta
+            else:
+                out.hash_info = entry.hash_info
+                out.meta = entry.meta
+        stage.dvcfile.dump(stage, with_files=True, update_pipeline=False)
+
+    return pushed

dvc/schema.py

@@ -51,6 +51,7 @@ OUT_PSTAGE_DETAILED_SCHEMA = {
         Output.PARAM_PERSIST: bool,
         Output.PARAM_CHECKPOINT: bool,
         Output.PARAM_REMOTE: str,
+        Output.PARAM_PUSH: bool,
     }
 }
 
@@ -103,7 +104,7 @@ def either_or(primary, fallback, fallback_includes=None):
 
 
 PLOT_DEFINITION = {
-    Output.PARAM_PLOT_X: str,
+    Output.PARAM_PLOT_X: Any(str, {str: str}),
     Output.PARAM_PLOT_Y: Any(str, [str], {str: Any(str, [str])}),
     Output.PARAM_PLOT_X_LABEL: str,
     Output.PARAM_PLOT_Y_LABEL: str,

dvc/stage/__init__.py

@@ -113,10 +113,11 @@ def restore_fields(stage):
     stage.meta = old.meta
     stage.desc = old.desc
 
-    old_fields = {out.def_path: (out.annot, out.remote) for out in old.outs}
+    old_outs = {out.def_path: out for out in old.outs}
     for out in stage.outs:
-        if out_fields := old_fields.get(out.def_path, None):
-            out.annot, out.remote = out_fields
+        old_out = old_outs.get(out.def_path, None)
+        if old_out is not None:
+            out.restore_fields(old_out)
 
 
 class Stage(params.StageParams):
@@ -462,10 +463,12 @@ class Stage(params.StageParams):
         logger.debug("Computed %s md5: '%s'", self, m)
         return m
 
-    def save(self, allow_missing=False):
+    def save(self, allow_missing: bool = False, merge_versioned: bool = False):
         self.save_deps(allow_missing=allow_missing)
 
-        self.save_outs(allow_missing=allow_missing)
+        self.save_outs(
+            allow_missing=allow_missing, merge_versioned=merge_versioned
+        )
         self.md5 = self.compute_md5()
 
         self.repo.stage_cache.save(self)
@@ -480,15 +483,40 @@ class Stage(params.StageParams):
                 if not allow_missing:
                     raise
 
-    def save_outs(self, allow_missing=False):
+    def save_outs(
+        self, allow_missing: bool = False, merge_versioned: bool = False
+    ):
         from dvc.output import OutputDoesNotExistError
 
+        from .exceptions import StageFileDoesNotExistError, StageNotFound
+
+        if merge_versioned:
+            try:
+                old = self.reload()
+                old_outs = {out.def_path: out for out in old.outs}
+                merge_versioned = any(
+                    (
+                        out.files is not None
+                        or (
+                            out.meta is not None
+                            and out.meta.version_id is not None
+                        )
+                    )
+                    for out in old_outs.values()
+                )
+            except (StageFileDoesNotExistError, StageNotFound):
+                merge_versioned = False
+
         for out in self.outs:
             try:
                 out.save()
             except OutputDoesNotExistError:
                 if not (allow_missing or out.checkpoint):
                     raise
+            if merge_versioned:
+                old_out = old_outs.get(out.def_path)
+                if old_out is not None:
+                    out.merge_version_meta(old_out)
 
     def ignore_outs(self):
         for out in self.outs:

dvc/stage/serialize.py

@@ -29,6 +29,7 @@ PARAM_PERSIST = Output.PARAM_PERSIST
 PARAM_CHECKPOINT = Output.PARAM_CHECKPOINT
 PARAM_DESC = Annotation.PARAM_DESC
 PARAM_REMOTE = Output.PARAM_REMOTE
+PARAM_PUSH = Output.PARAM_PUSH
 
 DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
 
@@ -54,6 +55,8 @@ def _get_flags(out):
         yield from out.plot.items()
     if out.remote:
         yield PARAM_REMOTE, out.remote
+    if not out.can_push:
+        yield PARAM_PUSH, False
 
 
 def _serialize_out(out):

dvc/testing/api_tests.py

@@ -13,14 +13,22 @@ class TestAPI:
         assert api.get_url("foo") == expected_url
 
     def test_open(self, tmp_dir, dvc, remote):
-        tmp_dir.dvc_gen("foo", "foo-text")
+        tmp_dir.dvc_gen(
+            {
+                "foo": "foo-text",
+                "dir": {"bar": "bar-text"},
+            }
+        )
         dvc.push()
 
         # Remove cache to force download
         remove(dvc.odb.local.path)
 
-        with api.open("foo") as fd:
-            assert fd.read() == "foo-text"
+        with api.open("foo") as fobj:
+            assert fobj.read() == "foo-text"
+
+        with api.open("dir/bar") as fobj:
+            assert fobj.read() == "bar-text"
 
     @pytest.mark.parametrize(
         "clear_cache", [True, False], ids=["cache", "no_cache"]

dvc/ui/__init__.py

@@ -270,7 +270,9 @@ class Console:
 
         path = Path(file).resolve()
         url = (
-            relpath(path) if "Microsoft" in uname().release else path.as_uri()
+            relpath(path)
+            if "microsoft" in uname().release.lower()
+            else path.as_uri()
         )
 
         opened = webbrowser.open(url)

dvc/utils/hydra.py

@@ -54,8 +54,12 @@ def compose_and_dump(
     with initialize_config_dir(config_dir, version_base=None):
         cfg = compose(config_name=config_name, overrides=overrides)
 
-    dumper = DUMPERS[Path(output_file).suffix.lower()]
-    dumper(output_file, OmegaConf.to_object(cfg))
+    suffix = Path(output_file).suffix.lower()
+    if suffix not in [".yml", ".yaml"]:
+        dumper = DUMPERS[suffix]
+        dumper(output_file, OmegaConf.to_object(cfg))
+    else:
+        Path(output_file).write_text(OmegaConf.to_yaml(cfg), encoding="utf-8")
 
 
 def apply_overrides(path: "StrPath", overrides: List[str]) -> None:

dvc/utils/plots.py

@@ -1,2 +1,6 @@
-def get_plot_id(plot_id: str, config_file_path: str = ""):
-    return f"{config_file_path}::{plot_id}" if config_file_path else plot_id
+def get_plot_id(config_plot_id: str, config_file_path: str = ""):
+    return (
+        f"{config_file_path}::{config_plot_id}"
+        if config_file_path
+        else config_plot_id
+    )

pyproject.toml

@@ -51,10 +51,10 @@ dependencies = [
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
     "scmrepo==0.1.3",
-    "dvc-render==0.0.12",
-    "dvc-task==0.1.5",
+    "dvc-render==0.0.14",
+    "dvc-task==0.1.6",
     "dvclive>=1.0",
-    "dvc-data==0.25.3",
+    "dvc-data==0.28.0",
     "dvc-http==2.27.2",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.6",
@@ -70,7 +70,7 @@ classifiers = [
 ]
 
 [project.optional-dependencies]
-azure = ["dvc-azure==2.20.4"]
+azure = ["dvc-azure==2.20.5"]
 gdrive = ["dvc-gdrive==2.19.1"]
 gs = ["dvc-gs==2.20.0"]
 hdfs = ["dvc-hdfs==2.19.0; python_version < '3.11'"]

scripts/fpm/build.py

@@ -44,6 +44,10 @@ if args.pkg == "osxpkg":
         ]
     )
 else:
+    if args.pkg == "rpm":
+        # https://github.com/jordansissel/fpm/issues/1503
+        flags.extend(["--rpm-rpmbuild-define", '"_build_id_links none"'])
+
     bash_dir = build / "etc" / "bash_completion.d"
     dirs = ["usr", "etc"]
     flags.extend(["--depends", "git >= 1.7.0"])  # needed for gitpython

tests/conftest.py

@@ -1,3 +1,4 @@
+import gc
 import json
 import os
 import sys
@@ -259,3 +260,19 @@ def mock_hydra_conf(mocker):
     # NOTE: using sentinel here so that any imports from `hydra.conf`
     # return a mock.
     sys.modules["hydra.conf"] = mocker.sentinel
+
+
+@pytest.fixture(autouse=True)
+def gc_collect_on_dvc_close_on_win_311(mocker):
+    if sys.version_info < (3, 11) and os.name != "nt":
+        return
+
+    from dvc.repo import Repo
+
+    close = Repo.close
+
+    def wrapped(repo):
+        close(repo)
+        gc.collect()
+
+    mocker.patch("dvc.repo.Repo.close", wrapped)

tests/func/experiments/test_experiments.py

@@ -417,7 +417,7 @@ def test_packed_args_exists(tmp_dir, scm, dvc, exp_stage, caplog):
         assert "Temporary DVC file" in caplog.text
 
 
-def test_list(tmp_dir, scm, dvc, exp_stage):
+def _prepare_experiments(tmp_dir, scm, dvc, exp_stage):
     baseline_a = scm.get_rev()
     results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
     exp_a = first(results)
@@ -433,6 +433,18 @@ def test_list(tmp_dir, scm, dvc, exp_stage):
     exp_c = first(results)
     ref_info_c = first(exp_refs_by_rev(scm, exp_c))
 
+    return baseline_a, baseline_c, ref_info_a, ref_info_b, ref_info_c
+
+
+def test_list(tmp_dir, scm, dvc, exp_stage):
+    (
+        baseline_a,
+        baseline_c,
+        ref_info_a,
+        ref_info_b,
+        ref_info_c,
+    ) = _prepare_experiments(tmp_dir, scm, dvc, exp_stage)
+
     assert dvc.experiments.ls() == {baseline_c: [ref_info_c.name]}
 
     exp_list = dvc.experiments.ls(rev=ref_info_a.baseline_sha)
@@ -447,6 +459,25 @@ def test_list(tmp_dir, scm, dvc, exp_stage):
     }
 
 
+def test_list_cli(tmp_dir, scm, dvc, capsys, exp_stage):
+    from dvc.cli import main
+
+    baseline_a, _, ref_info_a, ref_info_b, ref_info_c = _prepare_experiments(
+        tmp_dir, scm, dvc, exp_stage
+    )
+
+    # Make sure that we prioritize the current branch name
+    scm.checkout("branch", True)
+
+    capsys.readouterr()
+    assert main(["exp", "list", "-A"]) == 0
+    cap = capsys.readouterr()
+    assert set(cap.out.split()) == set(
+        ["branch:", baseline_a[:7] + ":"]
+        + [ref_info_a.name, ref_info_b.name, ref_info_c.name]
+    )
+
+
 def test_subdir(tmp_dir, scm, dvc, workspace):
     subdir = tmp_dir / "dir"
     subdir.gen("copy.py", COPY_SCRIPT)

tests/func/parsing/test_interpolated_entry.py

@@ -6,6 +6,7 @@ import pytest
 from dvc.dependency import _merge_params
 from dvc.parsing import DEFAULT_PARAMS_FILE, DataResolver
 from dvc.parsing.context import recurse_not_a_node
+from dvc.parsing.interpolate import escape_str
 
 from . import (
     CONTEXT_DATA,
@@ -272,14 +273,17 @@ def test_cmd_dict(tmp_dir, dvc, bool_config, list_config):
         if list_config:
             conf["parsing"]["list"] = list_config
 
+    string = "spaced string"
+    mixed_quote_string = "quote\"'d"
     data = {
         "dict": {
             "foo": "foo",
             "bar": 2,
-            "string": "spaced string",
+            "string": string,
+            "mixed_quote_string": mixed_quote_string,
             "bool": True,
             "bool-false": False,
-            "list": [1, 2, "foo"],
+            "list": [1, 2, "foo", mixed_quote_string],
             "nested": {"foo": "foo"},
         }
     }
@@ -296,9 +300,10 @@ def test_cmd_dict(tmp_dir, dvc, bool_config, list_config):
         bool_resolved = " --bool --no-bool-false"
 
     if list_config is None or list_config == "nargs":
-        list_resolved = " --list 1 2 'foo'"
+        list_resolved = f" --list 1 2 foo {escape_str(mixed_quote_string)}"
     else:
-        list_resolved = " --list 1 --list 2 --list 'foo'"
+        list_resolved = " --list 1 --list 2 --list foo"
+        list_resolved += f" --list {escape_str(mixed_quote_string)}"
 
     assert_stage_equal(
         resolver.resolve(),
@@ -306,11 +311,12 @@ def test_cmd_dict(tmp_dir, dvc, bool_config, list_config):
             "stages": {
                 "stage1": {
                     "cmd": "python script.py"
-                    " --foo 'foo' --bar 2"
-                    " --string 'spaced string'"
+                    " --foo foo --bar 2"
+                    f" --string {escape_str(string)}"
+                    f" --mixed_quote_string {escape_str(mixed_quote_string)}"
                     f"{bool_resolved}"
                     f"{list_resolved}"
-                    " --nested.foo 'foo'"
+                    " --nested.foo foo"
                 }
             }
         },

tests/func/plots/test_show.py

@@ -336,16 +336,15 @@ def test_collect_non_existing_dir(tmp_dir, dvc, run_copy_metrics):
 @pytest.mark.parametrize(
     "plot_config,expected_datafiles",
     [
-        # TODO - enable providing data files for x
-        # (
-        #     {
-        #         "comparison": {
-        #             "x": {"data1.json": "a"},
-        #             "y": {"sub/dir/data2.json": "b"},
-        #         }
-        #     },
-        #     ["data1.json", os.path.join("sub", "dir", "data2.json")],
-        # ),
+        (
+            {
+                "comparison": {
+                    "x": {"data1.json": "a"},
+                    "y": {"sub/dir/data2.json": "b"},
+                }
+            },
+            ["data1.json", os.path.join("sub", "dir", "data2.json")],
+        ),
         (
             {"data1.json": {"x": "c", "y": "a", "title": "File as key test"}},
             ["data1.json"],

tests/func/test_import.py

@@ -304,7 +304,7 @@ def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
 
     remove("dir_imported")
-    remove(dvc.odb.local.path)
+    dvc.odb.local.clear()
 
     dvc.pull(["dir_imported.dvc"])
 
@@ -320,7 +320,7 @@ def test_pull_wildcard_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.imp(os.fspath(erepo_dir), "dir123", "dir_imported123")
 
     remove("dir_imported123")
-    remove(dvc.odb.local.path)
+    dvc.odb.local.clear()
 
     dvc.pull(["dir_imported*.dvc"], glob=True)
 
@@ -546,9 +546,8 @@ def test_pull_imported_stage_from_subrepos(
     dvc.imp(os.fspath(erepo_dir), path, out="out")
 
     # clean everything
-    remove(dvc.odb.local.path)
+    dvc.odb.local.clear()
     remove("out")
-    makedirs(dvc.odb.local.path)
 
     stats = dvc.pull(["out.dvc"])
 
@@ -633,7 +632,7 @@ def test_chained_import(tmp_dir, dvc, make_tmp_dir, erepo_dir, local_cloud):
     assert (dst / "foo").read_text() == "foo"
     assert (dst / "bar").read_text() == "bar"
 
-    remove(dvc.odb.local.path)
+    dvc.odb.local.clear()
     remove("dir_imported")
 
     # pulled objects should come from the original upstream repo's remote,

tests/func/test_used_objs.py

@@ -51,3 +51,16 @@ def test_from_gitfs_when_pwd_not_in_root(
 
     with wdir.chdir():
         assert dvc.used_objs([target], revs=[scm.get_rev()])
+
+
+def test_used_objs_push(tmp_dir, scm, dvc):
+    stage = tmp_dir.dvc_gen("foo", "foo")[0]
+    hash_info = stage.outs[0].hash_info
+
+    stage.outs[0].can_push = True
+    assert stage.get_used_objs(push=False) == {None: {hash_info}}
+    assert stage.get_used_objs(push=True) == {None: {hash_info}}
+
+    stage.outs[0].can_push = False
+    assert stage.get_used_objs(push=False) == {None: {hash_info}}
+    assert stage.get_used_objs(push=True) == {}

tests/func/utils/test_hydra.py

@@ -174,12 +174,25 @@ def test_compose_and_dump(tmp_dir, suffix, overrides, expected):
     from dvc.utils.hydra import compose_and_dump
 
     config_name = "config"
-    config_dir = hydra_setup(tmp_dir, "conf", "config")
     output_file = tmp_dir / f"params.{suffix}"
+    config_dir = hydra_setup(tmp_dir, "conf", "config")
     compose_and_dump(output_file, config_dir, config_name, overrides)
     assert output_file.parse() == expected
 
 
+@pytest.mark.skipif(sys.version_info >= (3, 11), reason="unsupported on 3.11")
+def test_compose_and_dump_yaml_handles_string(tmp_dir):
+    """Regression test for 8583"""
+    from dvc.utils.hydra import compose_and_dump
+
+    config = tmp_dir / "conf" / "config.yaml"
+    config.parent.mkdir()
+    config.write_text("foo: 'no'\n")
+    output_file = tmp_dir / "params.yaml"
+    compose_and_dump(output_file, str(config.parent), "config", [])
+    assert output_file.read_text() == "foo: 'no'\n"
+
+
 @pytest.mark.parametrize(
     "overrides, expected",
     [

tests/integration/plots/test_plots.py

@@ -456,9 +456,7 @@ def test_repo_with_config_plots(tmp_dir, capsys, repo_with_config_plots):
             "dvc.yaml::confusion_train_vs_test",
         ],
     )
-    assert html_result["dvc.yaml::linear_train_vs_test"]["data"][
-        "values"
-    ] == _update_datapoints(
+    ble = _update_datapoints(
         plots["data"]["linear_train.json"],
         {
             REVISION_FIELD: "linear_train.json",
@@ -470,4 +468,7 @@ def test_repo_with_config_plots(tmp_dir, capsys, repo_with_config_plots):
         },
     )
 
+    assert (
+        html_result["dvc.yaml::linear_train_vs_test"]["data"]["values"] == ble
+    )
     # TODO check json results once vscode is able to handle flexible plots

tests/unit/command/test_plots.py

@@ -178,7 +178,7 @@ def test_plots_diff_open(tmp_dir, dvc, mocker, capsys, plots_data, auto_open):
 def test_plots_diff_open_WSL(tmp_dir, dvc, mocker, plots_data):
     mocked_open = mocker.patch("webbrowser.open", return_value=True)
     mocked_uname_result = mocker.MagicMock()
-    mocked_uname_result.release = "Microsoft"
+    mocked_uname_result.release = "microsoft"
     mocker.patch("platform.uname", return_value=mocked_uname_result)
 
     cli_args = parse_args(

tests/unit/render/test_image_converter.py

@@ -3,10 +3,9 @@ from dvc.render.converter.image import ImageConverter
 
 
 def test_image_converter_no_out():
-    data = b"content"
-    converter = ImageConverter()
-
-    datapoints, _ = converter.convert(data, "r", "image.png")
+    data = {"image.png": b"content"}
+    converter = ImageConverter("image.png", data)
+    datapoints, _ = converter.flat_datapoints("r")
 
     assert datapoints[0] == {
         REVISION_FIELD: "r",
@@ -16,10 +15,10 @@ def test_image_converter_no_out():
 
 
 def test_image_converter_with_out(tmp_dir):
-    data = b"content"
-    converter = ImageConverter({"out": tmp_dir / "foo"})
+    data = {"image.png": b"content"}
+    converter = ImageConverter("image.png", data, {"out": tmp_dir / "foo"})
 
-    datapoints, _ = converter.convert(data, "r", "image.png")
+    datapoints, _ = converter.flat_datapoints("r")
 
     assert datapoints[0] == {
         REVISION_FIELD: "r",
@@ -32,10 +31,10 @@ def test_image_converter_with_out(tmp_dir):
 
 def test_image_converter_with_slash_in_revision(tmp_dir):
     """Regression test for #7934"""
-    data = b"content"
-    converter = ImageConverter({"out": tmp_dir / "foo"})
+    data = {"image.png": b"content"}
+    converter = ImageConverter("image.png", data, {"out": tmp_dir / "foo"})
 
-    datapoints, _ = converter.convert(data, "feature/r", "image.png")
+    datapoints, _ = converter.flat_datapoints("feature/r")
 
     assert datapoints[0] == {
         REVISION_FIELD: "feature/r",

tests/unit/render/test_match.py

@@ -111,6 +111,8 @@ def test_match_renderers(mocker):
         "title": "config_file_1::plot_id_1",
         "x": "x",
         "y": "y",
+        "x_label": "x",
+        "y_label": "y",
     }
 
 

tests/unit/render/test_vega_converter.py

@@ -3,38 +3,7 @@ from collections import OrderedDict
 import pytest
 
 from dvc.render import VERSION_FIELD
-from dvc.render.converter.vega import (
-    FieldsNotFoundError,
-    PlotDataStructureError,
-    VegaConverter,
-    _filter_fields,
-    _find_first_list,
-    _lists,
-)
-
-
-def test_find_first_list_in_dict():
-    m1 = [{"accuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
-    m2 = [{"x": 1}, {"x": 2}]
-    dmetric = OrderedDict([("t1", m1), ("t2", m2)])
-
-    assert _find_first_list(dmetric, fields=set()) == m1
-    assert _find_first_list(dmetric, fields={"x"}) == m2
-
-    with pytest.raises(PlotDataStructureError):
-        _find_first_list(dmetric, fields={"foo"})
-
-
-def test_filter_fields():
-    m = [{"accuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
-
-    assert _filter_fields(m, fields=set()) == m
-
-    expected = [{"accuracy": 1}, {"accuracy": 3}]
-    assert _filter_fields(m, fields={"accuracy"}) == expected
-
-    with pytest.raises(FieldsNotFoundError):
-        _filter_fields(m, fields={"bad_field"})
+from dvc.render.converter.vega import VegaConverter, _lists
 
 
 @pytest.mark.parametrize(
@@ -58,7 +27,7 @@ def test_finding_lists(dictionary, expected_result):
     "input_data,properties,expected_datapoints,expected_properties",
     [
         pytest.param(
-            {"metric": [{"v": 1}, {"v": 2}]},
+            {"f": {"metric": [{"v": 1}, {"v": 2}]}},
             {},
             [
                 {
@@ -80,41 +49,11 @@ def test_finding_lists(dictionary, expected_result):
                     },
                 },
             ],
-            {"x": "step", "y": "v"},
+            {"x": "step", "y": "v", "x_label": "step", "y_label": "v"},
             id="default_x_y",
         ),
         pytest.param(
-            {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]},
-            {"fields": {"v"}},
-            [
-                {
-                    "v": 1,
-                    "step": 0,
-                    VERSION_FIELD: {
-                        "revision": "r",
-                        "filename": "f",
-                        "field": "v",
-                    },
-                },
-                {
-                    "v": 2,
-                    "step": 1,
-                    VERSION_FIELD: {
-                        "revision": "r",
-                        "filename": "f",
-                        "field": "v",
-                    },
-                },
-            ],
-            {
-                "x": "step",
-                "y": "v",
-                "fields": {"v", "step"},
-            },
-            id="filter_fields",
-        ),
-        pytest.param(
-            {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]},
+            {"f": {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]}},
             {"x": "v", "y": "v2"},
             [
                 {
@@ -136,52 +75,24 @@ def test_finding_lists(dictionary, expected_result):
                     },
                 },
             ],
-            {"x": "v", "y": "v2"},
+            {"x": "v", "y": "v2", "x_label": "v", "y_label": "v2"},
             id="choose_x_y",
         ),
         pytest.param(
             {
-                "metric": [
-                    {"v": 1, "v2": 0.1, "v3": 0.01, "v4": 0.001},
-                    {"v": 2, "v2": 0.2, "v3": 0.02, "v4": 0.002},
-                ]
-            },
-            {"x": "v3", "y": "v4", "fields": {"v"}},
-            [
-                {
-                    "v": 1,
-                    "v3": 0.01,
-                    "v4": 0.001,
-                    VERSION_FIELD: {
-                        "revision": "r",
-                        "filename": "f",
-                        "field": "v4",
+                "f": {
+                    "some": "noise",
+                    "very": {
+                        "nested": {
+                            "metric": [
+                                {"v": 1, "v2": 0.1},
+                                {"v": 2, "v2": 0.2},
+                            ]
+                        }
                     },
-                },
-                {
-                    "v": 2,
-                    "v3": 0.02,
-                    "v4": 0.002,
-                    VERSION_FIELD: {
-                        "revision": "r",
-                        "filename": "f",
-                        "field": "v4",
-                    },
-                },
-            ],
-            {"x": "v3", "y": "v4", "fields": {"v", "v3", "v4"}},
-            id="append_x_y_to_fields",
-        ),
-        pytest.param(
-            {
-                "some": "noise",
-                "very": {
-                    "nested": {
-                        "metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]
-                    }
-                },
+                }
             },
-            {"x": "v", "y": "v2"},
+            {"x": "v", "y": "v2", "x_label": "x", "y_label": "y"},
             [
                 {
                     "v": 1,
@@ -202,61 +113,128 @@ def test_finding_lists(dictionary, expected_result):
                     },
                 },
             ],
-            {"x": "v", "y": "v2"},
+            {"x": "v", "y": "v2", "x_label": "x", "y_label": "y"},
             id="find_in_nested_structure",
         ),
         pytest.param(
-            {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]},
+            {"f": {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]}},
             {"y": {"f": ["v", "v2"]}},
             [
                 {
-                    "v": 1,
-                    "v2": 0.1,
                     VERSION_FIELD: {
                         "revision": "r",
                         "filename": "f",
                         "field": "v",
                     },
                     "dvc_inferred_y_value": 1,
-                    "step": 0,
-                },
-                {
                     "v": 1,
                     "v2": 0.1,
-                    VERSION_FIELD: {
-                        "revision": "r",
-                        "filename": "f",
-                        "field": "v2",
-                    },
-                    "dvc_inferred_y_value": 0.1,
                     "step": 0,
                 },
                 {
-                    "v": 2,
-                    "v2": 0.2,
                     VERSION_FIELD: {
                         "revision": "r",
                         "filename": "f",
                         "field": "v",
                     },
                     "dvc_inferred_y_value": 2,
+                    "v": 2,
+                    "v2": 0.2,
                     "step": 1,
                 },
                 {
-                    "v": 2,
-                    "v2": 0.2,
                     VERSION_FIELD: {
                         "revision": "r",
                         "filename": "f",
                         "field": "v2",
                     },
+                    "dvc_inferred_y_value": 0.1,
+                    "v2": 0.1,
+                    "v": 1,
+                    "step": 0,
+                },
+                {
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                    "v": 2,
+                    "v2": 0.2,
                     "dvc_inferred_y_value": 0.2,
                     "step": 1,
                 },
             ],
-            {"x": "step", "y": "dvc_inferred_y_value", "y_label": "y"},
+            {
+                "x": "step",
+                "y": "dvc_inferred_y_value",
+                "y_label": "y",
+                "x_label": "step",
+            },
             id="y_def_list",
         ),
+        pytest.param(
+            {
+                "f": {
+                    "metric": [{"v": 1}, {"v": 2}],
+                    "other_metric": [{"z": 3}, {"z": 4}],
+                }
+            },
+            {"y": {"f": ["v", "z"]}},
+            [
+                {
+                    "dvc_inferred_y_value": 1,
+                    "z": 3,
+                    "v": 1,
+                    "step": 0,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 2,
+                    "z": 4,
+                    "step": 1,
+                    "v": 2,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 3,
+                    "v": 1,
+                    "z": 3,
+                    "step": 0,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "z",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 4,
+                    "v": 2,
+                    "z": 4,
+                    "step": 1,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "z",
+                    },
+                },
+            ],
+            {
+                "x": "step",
+                "y": "dvc_inferred_y_value",
+                "y_label": "y",
+                "x_label": "step",
+            },
+            id="multi_source_json",
+        ),
     ],
 )
 def test_convert(
@@ -265,32 +243,26 @@ def test_convert(
     expected_datapoints,
     expected_properties,
 ):
-    converter = VegaConverter(properties)
-    datapoints, resolved_properties = converter.convert(
-        data=input_data, revision="r", filename="f"
-    )
+    converter = VegaConverter("f", input_data, properties)
+    datapoints, resolved_properties = converter.flat_datapoints("r")
 
     assert datapoints == expected_datapoints
     assert resolved_properties == expected_properties
 
 
-def test_convert_skip_step():
-    converter = VegaConverter()
-    converter.skip_step("append_index")
-    converter.skip_step("generate_y")
-
-    datapoints, resolved_properties = converter.convert(
-        data={"a": "b", "metric": [{"v": 1}, {"v": 2}]},
-        revision="r",
-        filename="f",
-    )
-
-    assert datapoints == [
-        {
-            "v": 1,
-        },
-        {
-            "v": 2,
-        },
-    ]
-    assert resolved_properties == {"x": "step", "y": "v"}
+@pytest.mark.parametrize(
+    "properties,label",
+    [
+        ({"x": {"actual.csv": "actual"}}, "actual"),
+        (
+            {"x": {"train_actual.csv": "actual", "val_actual.csv": "actual"}},
+            "actual",
+        ),
+        (
+            {"x": {"actual.csv": "actual", "predicted.csv": "predicted"}},
+            "x",
+        ),
+    ],
+)
+def test_infer_x_label(properties, label):
+    assert VegaConverter.infer_x_label(properties) == label

tests/unit/repo/experiments/queue/test_celery.py

@@ -2,11 +2,12 @@ import time
 
 import pytest
 from celery import shared_task
-from flaky.flaky_decorator import flaky
+from celery.result import AsyncResult
 
 from dvc.exceptions import DvcException
 from dvc.repo.experiments.exceptions import UnresolvedExpNamesError
 from dvc.repo.experiments.queue.base import QueueDoneResult
+from dvc.repo.experiments.queue.exceptions import CannotKillTasksError
 
 
 def test_shutdown_no_tasks(test_queue, mocker):
@@ -51,16 +52,14 @@ def test_shutdown_with_kill(test_queue, mocker):
     shutdown_spy.assert_called_once()
 
 
-# pytest-celery worker thread may finish the task before we check for PENDING
-@flaky(max_runs=3, min_passes=1)
 def test_post_run_after_kill(test_queue):
 
     from celery import chain
 
     sig_bar = test_queue.proc.run_signature(
-        ["python3", "-c", "import time; time.sleep(5)"], name="bar"
+        ["python3", "-c", "import time; time.sleep(10)"], name="bar"
     )
-    result_bar = sig_bar.freeze()
+    sig_bar.freeze()
     sig_foo = _foo.s()
     result_foo = sig_foo.freeze()
     run_chain = chain(sig_bar, sig_foo)
@@ -69,25 +68,28 @@ def test_post_run_after_kill(test_queue):
     timeout = time.time() + 10
 
     while True:
-        if result_bar.status == "STARTED" or result_bar.ready():
+        try:
+            test_queue.proc.kill("bar")
+            assert result_foo.status == "PENDING"
             break
+        except ProcessLookupError:
+            time.sleep(0.1)
         if time.time() > timeout:
-            raise AssertionError()
-
-    assert result_foo.status == "PENDING"
-    test_queue.proc.kill("bar")
+            raise TimeoutError()
 
     assert result_foo.get(timeout=10) == "foo"
 
 
 def test_celery_queue_kill(test_queue, mocker):
 
-    mock_entry = mocker.Mock(stash_rev=_foo.name)
+    mock_entry_foo = mocker.Mock(stash_rev="foo")
+    mock_entry_bar = mocker.Mock(stash_rev="bar")
+    mock_entry_foobar = mocker.Mock(stash_rev="foobar")
 
     mocker.patch.object(
         test_queue,
         "iter_active",
-        return_value={mock_entry},
+        return_value={mock_entry_foo, mock_entry_bar, mock_entry_foobar},
     )
     mocker.patch.object(
         test_queue,
@@ -100,12 +102,54 @@ def test_celery_queue_kill(test_queue, mocker):
     mocker.patch.object(
         test_queue,
         "match_queue_entry_by_name",
-        return_value={"bar": mock_entry},
+        return_value={
+            "bar": mock_entry_bar,
+            "foo": mock_entry_foo,
+            "foobar": mock_entry_foobar,
+        },
+    )
+    mocker.patch.object(
+        test_queue,
+        "_get_running_task_ids",
+        return_value={"foo", "foobar"},
+    )
+    mocker.patch.object(
+        test_queue,
+        "_iter_processed",
+        return_value=[
+            (mocker.Mock(headers={"id": "foo"}), mock_entry_foo),
+            (mocker.Mock(headers={"id": "bar"}), mock_entry_bar),
+            (mocker.Mock(headers={"id": "foobar"}), mock_entry_foobar),
+        ],
+    )
+    mocker.patch.object(
+        AsyncResult,
+        "ready",
+        return_value=False,
+    )
+    mark_mocker = mocker.patch.object(
+        test_queue.celery.backend,
+        "mark_as_failure",
     )
 
-    spy = mocker.patch.object(test_queue.proc, "kill")
-    test_queue.kill("bar")
-    assert spy.called_once_with(mock_entry.stash_rev)
+    def kill_function(rev):
+        if rev == "foo":
+            return True
+        raise ProcessLookupError
+
+    kill_mock = mocker.patch.object(
+        test_queue.proc,
+        "kill",
+        side_effect=mocker.MagicMock(side_effect=kill_function),
+    )
+    with pytest.raises(
+        CannotKillTasksError, match="Task 'foobar' is initializing,"
+    ):
+        test_queue.kill(["bar", "foo", "foobar"])
+    assert kill_mock.called_once_with(mock_entry_foo.stash_rev)
+    assert kill_mock.called_once_with(mock_entry_bar.stash_rev)
+    assert kill_mock.called_once_with(mock_entry_foobar.stash_rev)
+    mark_mocker.assert_called_once_with("bar", None)
 
 
 @pytest.mark.parametrize("status", ["FAILURE", "SUCCESS"])

tests/unit/repo/experiments/test_show.py

@@ -1,7 +1,13 @@
+import pytest
 from funcy import first
 from scmrepo.git import Git
 
-from dvc.repo.experiments.show import get_names
+from dvc.repo.experiments.show import (
+    ExpStatus,
+    get_branch_names,
+    move_properties_to_head,
+    update_names,
+)
 
 
 def test_get_show_branch(tmp_dir, scm: "Git", dvc, exp_stage):
@@ -21,7 +27,192 @@ def test_get_show_branch(tmp_dir, scm: "Git", dvc, exp_stage):
         baseline: {"baseline": {"data": {}}, exp_a: {"data": {}}},
     }
 
-    get_names(dvc, result)
+    branch_names = get_branch_names(scm, ["workspace", exp_a, baseline])
+    assert branch_names == {exp_a: new_branch, baseline: base_branch}
+
+    update_names(dvc, branch_names, result)
     assert result[exp_a]["baseline"]["data"] == {"name": new_branch}
     assert result[baseline]["baseline"]["data"] == {"name": base_branch}
     assert result[baseline][exp_a]["data"] == {"name": new_branch}
+
+
+@pytest.mark.parametrize(
+    "res_in,res_out",
+    [
+        (
+            {
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "3": {
+                    "data": {
+                        "checkpoint_tip": "3",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "4": {
+                    "data": {
+                        "checkpoint_tip": "3",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "b",
+                    }
+                },
+            },
+            {
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "3": {
+                    "data": {
+                        "checkpoint_tip": "3",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "b",
+                    }
+                },
+                "4": {
+                    "data": {
+                        "checkpoint_tip": "3",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+            },
+        ),
+        (
+            {
+                "0": {"error": {"msg": "something"}},
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "3": {"data": {"name": "b"}},
+            },
+            {
+                "0": {"error": {"msg": "something"}},
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "3": {"data": {"name": "b"}},
+            },
+        ),
+        (
+            {
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "3": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "b",
+                    }
+                },
+                "4": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                    }
+                },
+            },
+            {
+                "1": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": "local",
+                        "status": ExpStatus.Running.name,
+                        "name": "a",
+                    }
+                },
+                "2": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+                "3": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                        "name": "b",
+                    }
+                },
+                "4": {
+                    "data": {
+                        "checkpoint_tip": "1",
+                        "executor": None,
+                        "status": ExpStatus.Success.name,
+                    }
+                },
+            },
+        ),
+    ],
+)
+def test_move_properties_to_head(res_in, res_out):
+    move_properties_to_head({"baseline": res_in})
+    assert res_in == res_out