fix(sdk.v2): Fix bug on executor input with default value. (#6937)

* fix bug on executor input

* release note

* fix test on empty message

* fix part of the test

samples/test/lightweight_python_functions_v2_pipeline_test.py

@@ -42,7 +42,6 @@ def verify(run: kfp_server_api.ApiRun, mlmd_connection_config, **kwargs):
                 'artifacts': [],
                 'parameters': {
                     'message': 'message',
-                    'empty_message': '',
                 }
             },
             'name': 'preprocess',
@@ -98,8 +97,7 @@ def verify(run: kfp_server_api.ApiRun, mlmd_connection_config, **kwargs):
                         "B": 2
                     },
                     'input_list': ["a", "b", "c"],
-                    'message': 'message',
-                    'num_steps': 100,
+                    'message': 'message'
                 }
             },
             'name': 'train',
@@ -126,8 +124,8 @@ run_pipeline_func([
         pipeline_func=pipeline,
         verify_func=verify,
         mode=dsl.PipelineExecutionMode.V2_COMPATIBLE),
-    # TODO(chensun): debug and fix v2 compiler bug on optional input
-    # TestCase(pipeline_func=pipeline,
-    #          verify_func=verify,
-    #          mode=dsl.PipelineExecutionMode.V2_ENGINE),
+    TestCase(
+        pipeline_func=pipeline,
+        verify_func=verify,
+        mode=dsl.PipelineExecutionMode.V2_ENGINE),
 ])

sdk/RELEASE.md

@@ -24,7 +24,7 @@
 * Fix importer ignoring reimport setting, and switch to Protobuf.Value for import uri [\#6827](https://github.com/kubeflow/pipelines/pull/6827)
 * Fix display name support for groups [\#6832](https://github.com/kubeflow/pipelines/pull/6832)
 * Remove redundant check in set_gpu_limit [\#6866](https://github.com/kubeflow/pipelines/pull/6866)
-* Fix regression on optional inputs [\#6905](https://github.com/kubeflow/pipelines/pull/6905)
+* Fix regression on optional inputs [\#6905](https://github.com/kubeflow/pipelines/pull/6905) [\#6937](https://github.com/kubeflow/pipelines/pull/6937)
 
 ## Documentation Updates
 

sdk/python/kfp/v2/components/executor.py

@@ -72,28 +72,9 @@ class Executor():
                                            {}).get('parameterValues', None)
 
         if parameter_values is not None:
-            value = parameter_values.get(parameter_name)
-            return value
+            return parameter_values.get(parameter_name, None)
 
-        parameter = self._input.get('inputs',
-                                    {}).get('parameters',
-                                            {}).get(parameter_name, None)
-        if parameter is None:
-            return None
-
-        if parameter.get('stringValue') is not None:
-            if parameter_type == str:
-                return parameter['stringValue']
-            elif parameter_type == bool:
-                # Use `.lower()` so it can also handle 'True' and 'False' (resulted from
-                # `str(True)` and `str(False)`, respectively.
-                return json.loads(parameter['stringValue'].lower())
-            else:
-                return json.loads(parameter['stringValue'])
-        elif parameter.get('intValue') is not None:
-            return int(parameter['intValue'])
-        elif parameter.get('doubleValue') is not None:
-            return float(parameter['doubleValue'])
+        return None
 
     def _get_output_parameter_path(self, parameter_name: str):
         parameter = self._input.get('outputs',
@@ -283,7 +264,9 @@ class Executor():
             v = type_annotations.maybe_strip_optional_from_annotation(v)
 
             if self._is_parameter(v):
-                func_kwargs[k] = self._get_input_parameter_value(k, v)
+                value = self._get_input_parameter_value(k, v)
+                if value is not None:
+                    func_kwargs[k] = value
 
             if type_annotations.is_artifact_annotation(v):
                 if type_annotations.is_input_artifact(v):

sdk/python/kfp/v2/components/executor_test.py

@@ -609,10 +609,22 @@ class ExecutorTest(unittest.TestCase):
             first_message: str = 'default value',
             second_message: Optional[str] = None,
             third_message: Optional[str] = None,
+            forth_argument: str = 'abc',
+            fifth_argument: int = 100,
+            sixth_argument: float = 1.23,
+            seventh_argument: bool = True,
+            eighth_argument: list = [1, 2],
+            ninth_argument: dict = {'a': 1},
         ) -> str:
             return (f'{first_message} ({type(first_message)}), '
                     f'{second_message} ({type(second_message)}), '
-                    f'{third_message} ({type(third_message)}).')
+                    f'{third_message} ({type(third_message)}), '
+                    f'{forth_argument} ({type(forth_argument)}), '
+                    f'{fifth_argument} ({type(fifth_argument)}), '
+                    f'{sixth_argument} ({type(sixth_argument)}), '
+                    f'{seventh_argument} ({type(seventh_argument)}), '
+                    f'{eighth_argument} ({type(eighth_argument)}), '
+                    f'{ninth_argument} ({type(ninth_argument)}).')
 
         self._get_executor(test_func, executor_input).execute()
         with open(os.path.join(self._test_dir, 'output_metadata.json'),
@@ -623,7 +635,13 @@ class ExecutorTest(unittest.TestCase):
                 "parameterValues": {
                     "Output": "Hello (<class 'str'>), "
                               "World (<class 'str'>), "
-                              "None (<class 'NoneType'>)."
+                              "None (<class 'NoneType'>), "
+                              "abc (<class 'str'>), "
+                              "100 (<class 'int'>), "
+                              "1.23 (<class 'float'>), "
+                              "True (<class 'bool'>), "
+                              "[1, 2] (<class 'list'>), "
+                              "{'a': 1} (<class 'dict'>)."
                 },
             })