serialize: fix loading of cloud metadata from .dvc/dvc.lock files (#8889)

dvc/output.py

@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type, cast
 from urllib.parse import urlparse
 
 from funcy import collecting, first, project
-from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
+from voluptuous import All, And, Any, Coerce, Length, Lower, Required, SetTo
 
 from dvc import prompt
 from dvc.exceptions import (
@@ -300,7 +300,7 @@ class Output:
     IsStageFileError: Type[DvcException] = OutputIsStageFileError
     IsIgnoredError: Type[DvcException] = OutputIsIgnoredError
 
-    def __init__(  # noqa: C901
+    def __init__(
         self,
         stage,
         path,
@@ -324,7 +324,8 @@ class Output:
             desc=desc, type=type, labels=labels or [], meta=meta or {}
         )
         self.repo = stage.repo if not repo and stage else repo
-        meta = Meta.from_dict(info or {})
+        meta_d = merge_file_meta_from_cloud(info or {})
+        meta = Meta.from_dict(meta_d)
         # NOTE: when version_aware is not passed into get_cloud_fs, it will be
         # set based on whether or not path is versioned
         fs_kwargs = {}
@@ -400,13 +401,17 @@ class Output:
             name=self.hash_name,
             value=getattr(self.meta, self.hash_name, None),
         )
+        self._compute_meta_hash_info_from_files()
+
+    def _compute_meta_hash_info_from_files(self) -> None:
         if self.files:
             tree = Tree.from_list(self.files, hash_name=self.hash_name)
             tree.digest()
             self.hash_info = tree.hash_info
             self.meta.isdir = True
             self.meta.nfiles = len(self.files)
-            self.meta.size = sum(file.get("size") for file in self.files)
+            self.meta.size = sum(f.get("size") for f in self.files)
+            self.meta.remote = first(f.get("remote") for f in self.files)
         elif self.meta.nfiles or self.hash_info and self.hash_info.isdir:
             self.meta.isdir = True
             if not self.hash_info and self.hash_name != "md5":
@@ -1282,6 +1287,8 @@ META_SCHEMA = {
     Meta.PARAM_VERSION_ID: str,
 }
 
+CLOUD_SCHEMA = All({str: {**META_SCHEMA, **CHECKSUMS_SCHEMA}}, Length(max=1))
+
 ARTIFACT_SCHEMA = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
@@ -1289,13 +1296,14 @@ ARTIFACT_SCHEMA = {
     Output.PARAM_PLOT: bool,
     Output.PARAM_PERSIST: bool,
     Output.PARAM_CHECKPOINT: bool,
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
 }
 
 DIR_FILES_SCHEMA: Dict[str, Any] = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
     Required(Tree.PARAM_RELPATH): str,
-    Output.PARAM_CLOUD: {str: {**META_SCHEMA, **CHECKSUMS_SCHEMA}},
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
 }
 
 SCHEMA = {

dvc/schema.py

@@ -4,7 +4,13 @@ from voluptuous import Any, Optional, Required, Schema
 
 from dvc import dependency, output
 from dvc.annotations import ANNOTATION_SCHEMA
-from dvc.output import CHECKSUMS_SCHEMA, DIR_FILES_SCHEMA, META_SCHEMA, Output
+from dvc.output import (
+    CHECKSUMS_SCHEMA,
+    CLOUD_SCHEMA,
+    DIR_FILES_SCHEMA,
+    META_SCHEMA,
+    Output,
+)
 from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
 from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
 from dvc.stage.params import StageParams
@@ -27,6 +33,7 @@ DATA_SCHEMA = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
     Required("path"): str,
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
     Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }
 LOCK_FILE_STAGE_SCHEMA = {

dvc/stage/loader.py

@@ -2,7 +2,7 @@ import logging
 from collections.abc import Mapping
 from copy import deepcopy
 from itertools import chain
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
 
 from funcy import get_in, lcat, once, project
 
@@ -20,6 +20,7 @@ from .utils import fill_stage_dependencies, resolve_paths
 
 if TYPE_CHECKING:
     from dvc.dvcfile import ProjectFile, SingleStageFile
+    from dvc.output import Output
 
 logger = logging.getLogger(__name__)
 
@@ -56,8 +57,10 @@ class StageLoader(Mapping):
         if not lock_data:
             return
 
+        from dvc.output import merge_file_meta_from_cloud
+
         assert isinstance(lock_data, dict)
-        items = chain(
+        items: Iterable[Tuple[str, Output]] = chain(
             ((StageParams.PARAM_DEPS, dep) for dep in stage.deps),
             ((StageParams.PARAM_OUTS, out) for out in stage.outs),
         )
@@ -74,12 +77,15 @@ class StageLoader(Mapping):
             info = get_in(checksums, [key, path], {})
             info = info.copy()
             info.pop("path", None)
-            item.meta = Meta.from_dict(info)
+
+            item.meta = Meta.from_dict(merge_file_meta_from_cloud(info))
             hash_value = getattr(item.meta, item.hash_name, None)
             item.hash_info = HashInfo(item.hash_name, hash_value)
-            if item.hash_info and item.hash_info.isdir:
-                item.meta.isdir = True
-            item.files = get_in(checksums, [key, path, item.PARAM_FILES])
+            files = get_in(checksums, [key, path, item.PARAM_FILES], None)
+            if files:
+                item.files = [merge_file_meta_from_cloud(f) for f in files]
+            # pylint: disable-next=protected-access
+            item._compute_meta_hash_info_from_files()
 
     @classmethod
     def load_stage(