Correct Runtime Artifact custom path logic/testing.

Signed-off-by: agoins <[email protected]>

Author: alyssacgoins

backend/src/v2/component/launcher_v2.go

@@ -945,6 +945,10 @@ func mergeRuntimeArtifacts(src, dst *pipelinespec.RuntimeArtifact) {
 			}
 		}
 	}
+
+	if src.CustomPath != nil && *src.CustomPath != "" {
+		dst.CustomPath = src.CustomPath
+	}
 }
 
 func getExecutorOutputFile(path string) (*pipelinespec.ExecutorOutput, error) {

backend/src/v2/metadata/converter.go

@@ -116,6 +116,10 @@ func toMLMDArtifact(runtimeArtifact *pipelinespec.RuntimeArtifact) (*pb.Artifact
 		Properties:       make(map[string]*pb.Value),
 		CustomProperties: make(map[string]*pb.Value),
 	}
+	// If runtime artifact is set with a custom path, add this path to the custom properties map.
+	if runtimeArtifact.CustomPath != nil {
+		artifact.CustomProperties["custom_path"] = stringMLMDValue(*runtimeArtifact.CustomPath)
+	}
 
 	if runtimeArtifact.Metadata != nil {
 		for k, v := range runtimeArtifact.Metadata.Fields {
@@ -177,13 +181,20 @@ func toRuntimeArtifact(artifact *pb.Artifact) (*pipelinespec.RuntimeArtifact, er
 	errorF := func(err error) (*pipelinespec.RuntimeArtifact, error) {
 		return nil, fmt.Errorf("failed to convert MLMD artifact to RuntimeArtifact: %w", err)
 	}
+	// Retrieve custom_path value from artifact.CustomProperties, if present.
+	var customPathStr string
+	if artifact.CustomProperties != nil {
+		customPath := artifact.CustomProperties["custom_path"]
+		customPathStr = customPath.GetStringValue()
+	}
 
 	rta := &pipelinespec.RuntimeArtifact{
 		Name: strconv.FormatInt(artifact.GetId(), 10),
 		Uri:  artifact.GetUri(),
 		Metadata: &structpb.Struct{
 			Fields: make(map[string]*structpb.Value),
 		},
+		CustomPath: &customPathStr,
 	}
 
 	propertiesToMetadata := func(properties map[string]*pb.Value) error {

sdk/python/kfp/dsl/types/artifact_types.py

@@ -91,7 +91,7 @@ def path(self) -> str:
 
     @path.setter
     def path(self, path: str) -> None:
-        self._set_custom_path(path)
+        self._set_path(path)
 
     def _get_path(self) -> Optional[str]:
         if self.custom_path:
@@ -112,23 +112,28 @@ def _get_path(self) -> Optional[str]:
         # uri == path for local execution
         return self.uri
 
+    def _set_path(self, path: str) -> None:
+        self.uri = convert_local_path_to_remote_path(path)
+
     @property
     def custom_path(self) -> str:
+        return self._get_custom_path()
+
+    @custom_path.getter
+    def custom_path(self):
         return self._custom_path
 
     def _get_custom_path(self) -> str:
         return self._custom_path
 
-    def _set_path(self, path: str) -> None:
-        self.uri = convert_local_path_to_remote_path(path)
-
-    def _set_custom_path(self, value: str) -> None:
-        self._custom_path = value
-
     @custom_path.setter
     def custom_path(self, value: str):
         self._custom_path = value
 
+    def set_path(self, path: str) -> None:
+        # If user specified a custom path, use it instead of the default path.
+        self._custom_path = path
+
 
 def convert_local_path_to_remote_path(path: str) -> str:
     if path.startswith(_GCS_LOCAL_MOUNT_PREFIX):