fix bug: op_to_template resolve the raw arguments by mapping to the a…

…rgument_inputs but the argument_inputs lost the type information

sdk/python/kfp/compiler/compiler.py

@@ -223,9 +223,12 @@ def _process_args(self, raw_args, argument_inputs):
       matches += _match_serialized_pipelineparam(str(processed_args[i]))
       unsanitized_argument_inputs = {}
       for x in list(set(matches)):
-        sanitized_str = str(dsl.PipelineParam(K8sHelper.sanitize_k8s_name(x[1]), K8sHelper.sanitize_k8s_name(x[0]), x[2]))
-        unsanitized_argument_inputs[sanitized_str] = str(dsl.PipelineParam(x[1], x[0], x[2]))
-
+        if len(x) == 3 or (len(x) == 4 and x[3] == ''):
+          sanitized_str = str(dsl.PipelineParam(K8sHelper.sanitize_k8s_name(x[1]), K8sHelper.sanitize_k8s_name(x[0]), x[2]))
+          unsanitized_argument_inputs[sanitized_str] = str(dsl.PipelineParam(x[1], x[0], x[2]))
+        elif len(x) == 4:
+          sanitized_str = str(dsl.PipelineParam(K8sHelper.sanitize_k8s_name(x[1]), K8sHelper.sanitize_k8s_name(x[0]), x[2], TypeMeta.from_dict_or_str(x[3])))
+          unsanitized_argument_inputs[sanitized_str] = str(dsl.PipelineParam(x[1], x[0], x[2], TypeMeta.from_dict_or_str(x[3])))
       if argument_inputs:
         for param in argument_inputs:
           if str(param) in unsanitized_argument_inputs:
@@ -257,7 +260,6 @@ def _build_conventional_artifact(name, path):
           }
         },
       }
-
     processed_arguments = self._process_args(op.arguments, op.argument_inputs)
     processed_command = self._process_args(op.command, op.argument_inputs)
 

sdk/python/kfp/dsl/_metadata.py

@@ -48,6 +48,9 @@ def to_dict_or_str(self):
   @staticmethod
   def from_dict_or_str(json):
     type_meta = TypeMeta()
+    if isinstance(json, str) and '{' in json:
+      import ast
+      json = ast.literal_eval(json)
     if isinstance(json, dict):
       if not _check_valid_type_dict(json):
         raise ValueError(json + ' is not a valid type string')

sdk/python/kfp/dsl/_pipeline_param.py

@@ -48,7 +48,13 @@ def _extract_pipelineparams(payloads: str or list[str]):
   matches = []
   for payload in payloads:
     matches += _match_serialized_pipelineparam(payload)
-  return [PipelineParam(x[1], x[0], x[2]) for x in list(set(matches))]
+  pipeline_params = []
+  for x in list(set(matches)):
+    if len(x) == 3 or (len(x) == 4 and x[3] == ''):
+      pipeline_params.append(PipelineParam(x[1], x[0], x[2]))
+    elif len(x) == 4:
+      pipeline_params.append(PipelineParam(x[1], x[0], x[2], TypeMeta.from_dict_or_str(x[3])))
+  return pipeline_params
 
 class PipelineParam(object):
   """Representing a future value that is passed between pipeline components.