feat(sdk): Support loading pipeline from yaml

sdk/RELEASE.md

@@ -3,7 +3,7 @@
 ## Major Features and Improvements
 * Support parallelism setting in ParallelFor [\#8146](https://github.com/kubeflow/pipelines/pull/8146)
 * Support for Python v3.10 [\#8186](https://github.com/kubeflow/pipelines/pull/8186)
-* Support pipeline as a component [\#8179](https://github.com/kubeflow/pipelines/pull/8179), [\#8204](https://github.com/kubeflow/pipelines/pull/8204)
+* Support pipeline as a component [\#8179](https://github.com/kubeflow/pipelines/pull/8179), [\#8204](https://github.com/kubeflow/pipelines/pull/8204), [\#8209](https://github.com/kubeflow/pipelines/pull/8209)
 
 ## Breaking Changes
 

sdk/python/kfp/compiler/_read_write_test_config.py

@@ -50,6 +50,7 @@
             'pipeline_in_pipeline',
             'pipeline_in_pipeline_complex',
             'pipeline_with_outputs',
+            'pipeline_in_pipeline_loaded_from_yaml',
         ],
         'test_data_dir': 'sdk/python/kfp/compiler/test_data/pipelines',
         'config': {

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -1125,6 +1125,9 @@ def build_spec_by_group(
         ]
         is_parent_component_root = (group_component_spec == pipeline_spec.root)
 
+        # Track if component spec is addeded from merging pipeline spec.
+        component_spec_added = False
+
         if isinstance(subgroup, pipeline_task.PipelineTask):
 
             subgroup_task_spec = build_task_spec_for_task(
@@ -1138,30 +1141,33 @@ def build_spec_by_group(
                 task=subgroup)
             task_name_to_component_spec[subgroup.name] = subgroup_component_spec
 
-            executor_label = subgroup_component_spec.executor_label
+            if subgroup_component_spec.executor_label:
+                executor_label = utils.make_name_unique_by_adding_index(
+                    name=subgroup_component_spec.executor_label,
+                    collection=list(deployment_config.executors.keys()),
+                    delimiter='-')
+                subgroup_component_spec.executor_label = executor_label
 
-            if executor_label not in deployment_config.executors:
-                if subgroup.container_spec is not None:
-                    subgroup_container_spec = build_container_spec_for_task(
-                        task=subgroup)
-                    deployment_config.executors[
-                        executor_label].container.CopyFrom(
-                            subgroup_container_spec)
-                elif subgroup.importer_spec is not None:
-                    subgroup_importer_spec = build_importer_spec_for_task(
-                        task=subgroup)
-                    deployment_config.executors[
-                        executor_label].importer.CopyFrom(
-                            subgroup_importer_spec)
-                elif subgroup.pipeline_spec is not None:
-                    merge_deployment_spec_and_component_spec(
-                        main_pipeline_spec=pipeline_spec,
-                        main_deployment_config=deployment_config,
-                        sub_pipeline_spec=subgroup.pipeline_spec,
-                        sub_pipeline_component_name=subgroup_component_name,
-                    )
-                else:
-                    raise RuntimeError
+            if subgroup.container_spec is not None:
+                subgroup_container_spec = build_container_spec_for_task(
+                    task=subgroup)
+                deployment_config.executors[executor_label].container.CopyFrom(
+                    subgroup_container_spec)
+            elif subgroup.importer_spec is not None:
+                subgroup_importer_spec = build_importer_spec_for_task(
+                    task=subgroup)
+                deployment_config.executors[executor_label].importer.CopyFrom(
+                    subgroup_importer_spec)
+            elif subgroup.pipeline_spec is not None:
+                merge_deployment_spec_and_component_spec(
+                    main_pipeline_spec=pipeline_spec,
+                    main_deployment_config=deployment_config,
+                    sub_pipeline_spec=subgroup.pipeline_spec,
+                    sub_pipeline_component_name=subgroup_component_name,
+                )
+                component_spec_added = True
+            else:
+                raise RuntimeError
         elif isinstance(subgroup, tasks_group.ParallelFor):
 
             # "Punch the hole", adding additional inputs (other than loop
@@ -1258,8 +1264,14 @@ def build_spec_by_group(
             subgroup_task_spec.dependent_tasks.extend(
                 [utils.sanitize_task_name(dep) for dep in group_dependencies])
 
-        # Add component spec if not exists
-        if subgroup_component_name not in pipeline_spec.components:
+        # Add component spec if not already added from merging pipeline spec.
+        if not component_spec_added:
+            subgroup_component_name = utils.make_name_unique_by_adding_index(
+                name=subgroup_component_name,
+                collection=list(pipeline_spec.components.keys()),
+                delimiter='-')
+
+            subgroup_task_spec.component_ref.name = subgroup_component_name
             pipeline_spec.components[subgroup_component_name].CopyFrom(
                 subgroup_component_spec)
 

sdk/python/kfp/compiler/test_data/pipelines/pipeline_in_pipeline_complex.py

@@ -59,8 +59,8 @@ def pipeline_not_used():
 
 
 @dsl.pipeline(name='pipeline-in-pipeline-complex')
-def my_pipeline():
-    print_op1(msg='Hello')
+def my_pipeline(msg: str = 'Hello'):
+    print_op1(msg=msg)
     with dsl.ParallelFor(['Hello', 'world!']) as item:
         graph_component(msg=item)
 

sdk/python/kfp/compiler/test_data/pipelines/pipeline_in_pipeline_complex.yaml

@@ -227,9 +227,13 @@ root:
         inputs:
           parameters:
             msg:
-              runtimeValue:
-                constant: Hello
+              componentInputParameter: msg
         taskInfo:
           name: print-op1
+  inputDefinitions:
+    parameters:
+      msg:
+        defaultValue: Hello
+        parameterType: STRING
 schemaVersion: 2.1.0
 sdkVersion: kfp-2.0.0-beta.3

sdk/python/kfp/compiler/test_data/pipelines/pipeline_in_pipeline_loaded_from_yaml.py

@@ -0,0 +1,43 @@
+# Copyright 2022 The Kubeflow Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pathlib
+
+from kfp import compiler
+from kfp import components
+from kfp import dsl
+from kfp.dsl import Artifact
+from kfp.dsl import Input
+
+
+@dsl.component
+def print_op1(data: Input[Artifact]):
+    with open(data.path, 'r') as f:
+        print(f.read())
+
+
+reuse_yaml_pipeline = components.load_component_from_file(
+    pathlib.Path(__file__).parent / 'pipeline_with_outputs.yaml')
+
+
+@dsl.pipeline(name='pipeline-in-pipeline')
+def my_pipeline():
+    task = reuse_yaml_pipeline(msg='Hello')
+    print_op1(data=task.output)
+
+
+if __name__ == '__main__':
+    compiler.Compiler().compile(
+        pipeline_func=my_pipeline,
+        package_path=__file__.replace('.py', '.yaml'))