fix(sdk): avoid conflicting component names in DAG when reusing pipelines

sdk/python/kfp/compiler/compiler_test.py

@@ -909,6 +909,48 @@ def my_pipeline() -> NamedTuple('Outputs', [
             ]):
                 task = print_and_return(text='Hello')
 
+    def test_pipeline_reusing_other_pipeline_multiple_times(self):
+
+        @dsl.pipeline()
+        def reusable_pipeline():
+            print_op(message='Reused pipeline')
+
+        @dsl.pipeline()
+        def do_something_else_pipeline():
+            print_op(message='Do something else pipeline')
+
+            reusable_pipeline()
+
+        @dsl.pipeline()
+        def orchestrator_pipeline():
+            reusable_pipeline()
+
+            do_something_else_pipeline()
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_yaml = os.path.join(tmpdir, 'result.yaml')
+            compiler.Compiler().compile(
+                pipeline_func=orchestrator_pipeline, package_path=output_yaml)
+            self.assertTrue(os.path.exists(output_yaml))
+
+            with open(output_yaml, 'r') as f:
+                pipeline_spec = yaml.safe_load(f)
+                tasks = [
+                    comp.get('dag', {}).get('tasks', {})
+                    for comp in pipeline_spec['components'].values()
+                ]
+                component_refs = [[
+                    x.get('componentRef', {}).get('name')
+                    for x in task.values()
+                ]
+                                  for task in tasks]
+                all_component_refs = [
+                    item for sublist in component_refs for item in sublist
+                ]
+                counted_refs = collections.Counter(all_component_refs)
+                print(counted_refs)
+                self.assertEqual(1, max(counted_refs.values()))
+
 
 class V2NamespaceAliasTest(unittest.TestCase):
     """Test that imports of both modules and objects are aliased (e.g. all

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -1773,10 +1773,18 @@ def _merge_component_spec(
 ) -> None:
     """Merges component spec from a sub pipeline spec into the main config.
 
-    During the merge we need to ensure all component specs have unique component
+    During the merge we need to ensure all component specs have a unique component
     name, that means we might need to update the `component_ref` referenced from
     task specs in sub_pipeline_spec.
 
+    Uniqueness is determined by existing component names in main_pipeline_spec
+    and sub_pipeline_spec.
+
+    Renaming is first done in place, specifically in a LIFO order. This is to avoid
+    a rename causing a name collision with a later rename. Then, the actual merge
+    of component specs is done in a second pass. This ensures all component specs
+    are in the final state at the time of merging.
+
     Args:
         main_pipeline_spec: The main pipeline spec to merge into.
         sub_pipeline_spec: The pipeline spec of an inner pipeline whose
@@ -1800,18 +1808,22 @@ def _rename_component_refs(
             if task_spec.component_ref.name == old_component_ref:
                 task_spec.component_ref.name = new_component_ref
 
-    # Do all the renaming in place, then do the acutal merge of component specs
-    # in a second pass. This would ensure all component specs are in the final
-    # state at the time of merging.
     old_name_to_new_name = {}
-    for component_name, component_spec in sub_pipeline_spec.components.items():
+    existing_main_comp_names = list(main_pipeline_spec.components.keys())
+    for component_name, _ in sub_pipeline_spec.components.items():
         old_component_name = component_name
+        current_comp_name_collection = [
+            key for pair in old_name_to_new_name.items() for key in pair
+        ]
         new_component_name = utils.make_name_unique_by_adding_index(
             name=component_name,
-            collection=list(main_pipeline_spec.components.keys()),
+            collection=existing_main_comp_names + current_comp_name_collection,
             delimiter='-')
         old_name_to_new_name[old_component_name] = new_component_name
 
+    ordered_names = enumerate(old_name_to_new_name.items())
+    lifo_ordered_names = sorted(ordered_names, key=lambda x: x[0], reverse=True)
+    for _, (old_component_name, new_component_name) in lifo_ordered_names:
         if new_component_name != old_component_name:
             _rename_component_refs(
                 pipeline_spec=sub_pipeline_spec,