SDK - Fix pipeline metadata serialization

Two PRs have been merged that turned out to be slightly incompatible. This PR fixes the failing tests.
Root causes:
* The pipeline parameter default values were not properly serialized when constructing the metadata object.
* The `ParameterMeta` class did not validate the default value type, so the lack of serialization has not been caught. The `ParameterMeta` was replaced by `InputSpec` which has strict type validation.
* Previously we did not have samples with complex pipeline parameter default values (e.g. lists) that could trigger the failures. Then two samples were added that had complex default values.

sdk/python/kfp/dsl/_metadata.py

@@ -14,6 +14,7 @@
 
 import warnings
 from .types import BaseType, _check_valid_type_dict, _instance_to_dict
+from ..components._data_passing import serialize_value, type_to_type_name
 from ..components._structures import ComponentSpec, InputSpec, OutputSpec
 
 
@@ -67,6 +68,8 @@ def _extract_component_metadata(func):
       arg_default = arg_default.value
     if arg in annotations:
       arg_type = _annotation_to_typemeta(annotations[arg])
+    if arg_default is not None:
+      arg_default = serialize_value(arg_default, type_name=str(arg_type) if arg_type else None) # TODO: Improve _annotation_to_typemeta or just replace the whole function with kfp.component._python_op._extract_component_interface
     inputs.append(InputSpec(name=arg, type=arg_type, default=arg_default))
   # Outputs
   outputs = []
@@ -124,7 +127,10 @@ def _extract_pipeline_metadata(func):
       if isinstance(schema_object, str):
         # In case the property value for the schema validator is a string instead of a dict.
         schema_object = json.loads(schema_object)
+      # Only validating non-serialized values
       validate(instance=arg_default, schema=schema_object)
+    if arg_default is not None:
+      arg_default = serialize_value(arg_default, type_name=str(arg_type) if arg_type else None) # TODO: Improve _annotation_to_typemeta or just replace the whole function with kfp.component._python_op._extract_component_interface
     inputs.append(InputSpec(name=arg, type=arg_type, default=arg_default))
 
   #TODO: add descriptions to the metadata

sdk/python/tests/compiler/testdata/pipelineparams.yaml

@@ -17,7 +17,7 @@ metadata:
   annotations:
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A pipeline with multiple
       pipeline params.", "inputs": [{"default": "latest", "name":
-      "tag"}, {"default": 10, "name": "sleep_ms"}], "name": "PipelineParams"}'
+      "tag"}, {"default": "10", "name": "sleep_ms"}], "name": "PipelineParams"}'
   generateName: pipelineparams-
 spec:
   entrypoint: pipelineparams

sdk/python/tests/compiler/testdata/recursive_while.yaml

@@ -3,7 +3,7 @@ kind: Workflow
 metadata:
   annotations:
     pipelines.kubeflow.org/pipeline_spec: '{"description": "shows how to use dsl.Condition.",
-      "inputs": [{"default": 12, "name": "maxVal"}],
+      "inputs": [{"default": "12", "name": "maxVal"}],
       "name": "pipeline flip coin"}'
   generateName: pipeline-flip-coin-
 spec:

sdk/python/tests/compiler/testdata/withitem_basic.yaml

@@ -2,7 +2,7 @@ apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
   annotations:
-    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": 10, "name": "my_pipe_param"}],
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "my_pipe_param"}],
       "name": "my-pipeline"}'
   generateName: my-pipeline-
 spec:

sdk/python/tests/compiler/testdata/withitem_nested.yaml

@@ -2,7 +2,7 @@ apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
   annotations:
-    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": 10, "name": "my_pipe_param"}],
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "my_pipe_param"}],
       "name": "my-pipeline"}'
   generateName: my-pipeline-
 spec:

sdk/python/tests/compiler/testdata/withparam_global.yaml

@@ -2,7 +2,7 @@ apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
   annotations:
-    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": [3, 5, 7, 9], "name":
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "[3, 5, 7, 9]", "name":
       "loopidy_doop"}], "name": "my-pipeline"}'
   generateName: my-pipeline-
 spec:

sdk/python/tests/compiler/testdata/withparam_global_dict.yaml

@@ -2,8 +2,8 @@ apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
   annotations:
-    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": [{"a": 1, "b":
-      2}, {"a": 10, "b": 20}], "name": "loopidy_doop"}], "name": "my-pipeline"}'
+    pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "[{\"a\": 1, \"b\":
+      2}, {\"a\": 10, \"b\": 20}]", "name": "loopidy_doop"}], "name": "my-pipeline"}'
   generateName: my-pipeline-
 spec:
   arguments: