[Pipeline] Improve pipeline node name update logic (Azure#28044)

* Improve pipeline node name update logic

* update changelog

Author: 0mza987

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -11,6 +11,7 @@
 - Added stricter behavior for ArmStr schemas when parsing 'azureml:' prefix.
 - Improved intellisense with VS Code for fields supporting local paths and datastores.
 - Added validation for token generation with aml scope when user_identity is used in job definition aka OBO flow
+- Fixed duplicate node name error in pipeline when two node names assigned to the same node and get renamed by node.name='xx'.
 
 ### Other Changes
 - Removed dependency on API version 2021-10-01 and 2022-06-01-preview to reduce side of azure-ai-ml package.

sdk/ml/azure-ai-ml/azure/ai/ml/dsl/_pipeline_component_builder.py

@@ -366,7 +366,8 @@ def _get_name_or_component_name(node: Union[BaseNode, AutoMLJob]):
             # TODO(1979547): refactor this
             if not isinstance(v, (BaseNode, AutoMLJob, PipelineJob, ControlFlowNode)):
                 continue
-            if getattr(v, "_instance_id", None) not in valid_component_ids:
+            instance_id = getattr(v, "_instance_id", None)
+            if instance_id not in valid_component_ids:
                 continue
             name = getattr(v, "name", None) or k
             if name is not None:
@@ -380,7 +381,7 @@ def _get_name_or_component_name(node: Union[BaseNode, AutoMLJob]):
                 )
 
             # Raise error when setting a name that already exists, likely conflict with a variable name
-            if name in local_names:
+            if name in local_names and instance_id not in id_name_dict:
                 raise UserErrorException(
                     f"Duplicate node name found in pipeline: {self.name!r}, "
                     f"node name: {name!r}. Duplicate check is case-insensitive."

sdk/ml/azure-ai-ml/tests/dsl/unittests/test_dsl_pipeline.py

@@ -509,6 +509,18 @@ def pipeline_with_user_defined_nodes_2():
             "microsoftsamplescommandcomponentbasic_nopaths_test_2",
         ]
 
+        @dsl.pipeline(name="pipeline_with_user_defined_nodes_3")
+        def pipeline_with_user_defined_nodes_3():
+            node1 = component_func1()
+            node1.name = "my_node"
+            node2 = node1
+
+        pipeline = pipeline_with_user_defined_nodes_3()
+        variable_names = list(pipeline.component.jobs.keys())
+        pipeline_job_names = list(pipeline.jobs.keys())
+        assert variable_names == pipeline_job_names
+        assert variable_names == ["my_node"]
+
         @dsl.pipeline(name="pipeline_with_duplicate_user_defined_nodes_1")
         def pipeline_with_duplicate_user_defined_nodes_1():
             node1 = component_func1()