move versioning into dvc-data meta

dvc/dependency/__init__.py

@@ -27,8 +27,7 @@ def _get(stage, p, info):
         params = info.pop(ParamsDependency.PARAM_PARAMS)
         return ParamsDependency(stage, p, params)
 
-    version_id = info.pop(Output.PARAM_VERSION_ID, None)
-    return Dependency(stage, p, info, version_id=version_id)
+    return Dependency(stage, p, info)
 
 
 def loadd_from(stage, d_list):

dvc/dependency/base.py

@@ -4,6 +4,7 @@
 
 from dvc.exceptions import DvcException
 from dvc.output import Output
+from dvc_data.hashfile.meta import Meta
 
 if TYPE_CHECKING:
     from dvc_data.hashfile.hash_info import HashInfo
@@ -42,7 +43,7 @@ def get_used_objs(
         from dvc_data.build import build
         from dvc_data.objects.tree import Tree, TreeError
 
-        if not self.version_id:
+        if not self.meta.version_id:
             return super().get_used_objs(**kwargs)
 
         used_obj_ids: Dict[
@@ -70,32 +71,34 @@ def get_used_objs(
         return used_obj_ids
 
     def workspace_status(self):
-        if not self.version_id:
+        if not self.meta.version_id:
             return super().workspace_status()
 
-        current = self.version_id
+        current = self.meta.version_id
         fs_path = self.fs.path.version_path(self.fs_path, None)
-        updated = self.fs.info(fs_path)["version_id"]
+        updated = self.fs.info(fs_path)[Meta.PARAM_VERSION_ID]
 
         if current != updated:
             return {str(self): "update available"}
 
         return {}
 
     def status(self):
-        if not self.version_id:
+        if not self.meta.version_id:
             return super().status()
 
         return self.workspace_status()
 
     def update(self, rev: Optional[str] = None):
-        if not self.version_id:
+        if not self.meta.version_id:
             return
 
         if rev:
-            self.version_id = rev
+            self.meta.version_id = rev
         else:
             fs_path = self.fs.path.version_path(self.fs_path, rev)
             details = self.fs.info(fs_path)
-            self.version_id = details["version_id"]
-        self.fs_path = self.fs.path.version_path(self.fs_path, self.version_id)
+            self.meta.version_id = details[Meta.PARAM_VERSION_ID]
+        self.fs_path = self.fs.path.version_path(
+            self.fs_path, self.meta.version_id
+        )

dvc/output.py

@@ -83,7 +83,6 @@ def loadd_from(stage, d_list):
         desc = d.pop(Output.PARAM_DESC, False)
         live = d.pop(Output.PARAM_LIVE, False)
         remote = d.pop(Output.PARAM_REMOTE, None)
-        version_id = d.pop(Output.PARAM_VERSION_ID, None)
         ret.append(
             _get(
                 stage,
@@ -97,7 +96,6 @@ def loadd_from(stage, d_list):
                 desc=desc,
                 live=live,
                 remote=remote,
-                version_id=version_id,
             )
         )
     return ret
@@ -259,7 +257,6 @@ class Output:
     PARAM_LIVE_SUMMARY = "summary"
     PARAM_LIVE_HTML = "html"
     PARAM_REMOTE = "remote"
-    PARAM_VERSION_ID = "version_id"
 
     METRIC_SCHEMA = Any(
         None,
@@ -289,11 +286,16 @@ def __init__(
         desc=None,
         remote=None,
         repo=None,
-        version_id: Optional[str] = None,
     ):
         self.repo = stage.repo if not repo and stage else repo
+        meta = Meta.from_dict(info)
+        # 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 = {"version_aware": True} if meta.version_id else {}
         fs_cls, fs_config, fs_path = get_cloud_fs(
-            self.repo, url=path, version_aware=version_id is not None
+            self.repo,
+            url=path,
+            **fs_kwargs,
         )
         self.fs = fs_cls(**fs_config)
 
@@ -326,7 +328,7 @@ def __init__(
         # By resolved path, which contains actual location,
         # should be absolute and don't contain remote:// refs.
         self.stage = stage
-        self.meta = Meta.from_dict(info)
+        self.meta = meta
         self.hash_info = HashInfo.from_dict(info)
         self.use_cache = False if self.IS_DEPENDENCY else cache
         self.metric = False if self.IS_DEPENDENCY else metric
@@ -342,18 +344,15 @@ def __init__(
         self.remote = remote
 
         if self.fs.version_aware:
-            self.def_path, self.version_id = self.fs.path.coalesce_version(
-                self.def_path, version_id
+            self.def_path, version_id = self.fs.path.coalesce_version(
+                self.def_path, self.meta.version_id
             )
-            self.fs_path = self.fs.path.version_path(
-                self.fs_path, self.version_id
+            self.fs_path = self.fs.path.version_path(self.fs_path, version_id)
+            self.meta.version_id = version_id
+        elif self.meta.version_id:
+            raise DvcException(
+                "Version ID unsupported for non-versioned filesystem"
             )
-        else:
-            if version_id:
-                raise DvcException(
-                    "Version ID unsupported for non-versioned filesystem"
-                )
-            self.version_id = None
 
     def _parse_path(self, fs, fs_path):
         parsed = urlparse(self.def_path)
@@ -380,9 +379,9 @@ def __repr__(self):
 
     def __str__(self):
         if self.fs.protocol != "local":
-            if self.version_id:
+            if self.meta.version_id:
                 return self.fs.path.version_path(
-                    self.def_path, self.version_id
+                    self.def_path, self.meta.version_id
                 )
             return self.def_path
 
@@ -688,8 +687,6 @@ def dumpd(self):
             path = self.def_path
 
         ret[self.PARAM_PATH] = path
-        if self.version_id:
-            ret[self.PARAM_VERSION_ID] = self.version_id
 
         if self.IS_DEPENDENCY:
             return ret
@@ -1120,10 +1117,10 @@ def is_plot(self) -> bool:
     Output.PARAM_PLOT: bool,
     Output.PARAM_PERSIST: bool,
     Output.PARAM_CHECKPOINT: bool,
-    Output.PARAM_VERSION_ID: str,
     Meta.PARAM_SIZE: int,
     Meta.PARAM_NFILES: int,
     Meta.PARAM_ISEXEC: bool,
+    Meta.PARAM_VERSION_ID: str,
 }
 
 SCHEMA = {

dvc/stage/__init__.py

@@ -265,7 +265,7 @@ def is_versioned_import(self):
         if not self.is_import:
             return False
 
-        return self.deps[0].version_id is not None
+        return self.deps[0].meta.version_id is not None
 
     @property
     def is_checkpoint(self):