[ML][Pipelines] feat: use origin name and version in request body for all anonymous components (Azure#28092)

* feat: use origin name and version in request body for all anonymous components

* refactor: minor change

* record: update recordings

Author: elliotzh

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/component.py

@@ -118,7 +118,6 @@ def __init__(
         # Store original yaml
         self._yaml_str = yaml_str
         self._other_parameter = kwargs
-        self._origin_name, self._origin_version = None, None
 
         self.successful_return_code = successful_return_code
         self.code = code
@@ -191,19 +190,9 @@ def _from_rest_object_to_init_params(cls, obj: ComponentVersionData) -> Dict:
             init_kwargs["distribution"] = DistributionConfiguration._from_rest_object(distribution)
         return init_kwargs
 
-    def _set_is_anonymous(self, is_anonymous: bool):
-        if is_anonymous:
-            self._origin_name, self._origin_version = self.name, self.version
-        super()._set_is_anonymous(is_anonymous)
-
     def _to_rest_object(self) -> ComponentVersionData:
         component = convert_ordered_dict_to_dict(self._to_dict())
 
-        if self._origin_name:
-            component["name"] = self._origin_name
-        if self._origin_version:
-            component["version"] = self._origin_version
-
         properties = ComponentVersionDetails(
             component_spec=component,
             description=self.description,

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_component/component.py

@@ -116,6 +116,13 @@ def __init__(
         self._source = (
             self._resolve_component_source_from_id(id) if id else kwargs.pop("_source", ComponentSource.CLASS)
         )
+        # use ANONYMOUS_COMPONENT_NAME instead of guid
+        is_anonymous = kwargs.pop("is_anonymous", False)
+        if not name and version is None:
+            name = ANONYMOUS_COMPONENT_NAME
+            version = "1"
+            is_anonymous = True
+
         super().__init__(
             name=name,
             version=version,
@@ -124,17 +131,13 @@ def __init__(
             tags=tags,
             properties=properties,
             creation_context=creation_context,
-            is_anonymous=kwargs.pop("is_anonymous", False),
+            is_anonymous=is_anonymous,
             base_path=kwargs.pop("base_path", None),
             source_path=kwargs.pop("source_path", None),
         )
         # store kwargs to self._other_parameter instead of pop to super class to allow component have extra
         # fields not defined in current schema.
 
-        # update component name to ANONYMOUS_COMPONENT_NAME if it is anonymous
-        if hasattr(self, "_is_anonymous"):
-            self._set_is_anonymous(self._is_anonymous)
-
         inputs = inputs if inputs else {}
         outputs = outputs if outputs else {}
 
@@ -411,25 +414,6 @@ def _from_rest_object_to_init_params(cls, obj: ComponentVersionData) -> Dict:
         # remove empty values, because some property only works for specific component, eg: distribution for command
         return {k: v for k, v in init_kwargs.items() if v is not None and v != {}}
 
-    def _set_is_anonymous(self, is_anonymous: bool):
-        """Mark this component as anonymous and overwrite component name to
-        ANONYMOUS_COMPONENT_NAME."""
-        if is_anonymous is True:
-            self._is_anonymous = True
-            self.name = ANONYMOUS_COMPONENT_NAME
-        else:
-            self._is_anonymous = False
-
-    def _update_anonymous_hash(self):
-        """For anonymous component, we use code hash + yaml hash as component
-        version so the same anonymous component(same interface and same code)
-        won't be created again.
-
-        Should be called before _to_rest_object.
-        """
-        if self._is_anonymous:
-            self.version = self._get_anonymous_hash()
-
     def _get_anonymous_hash(self) -> str:
         """Return the name of anonymous component.
 
@@ -441,6 +425,16 @@ def _get_anonymous_hash(self) -> str:
         # omit name since name doesn't impact component's uniqueness
         return hash_dict(component_interface_dict, keys_to_omit=["name", "id", "version"])
 
+    def _validate(self, raise_error=False) -> MutableValidationResult:
+        origin_name = self.name
+        # skip name validation for anonymous component as ANONYMOUS_COMPONENT_NAME will be used in component creation
+        if self._is_anonymous:
+            self.name = ANONYMOUS_COMPONENT_NAME
+        try:
+            return super()._validate(raise_error)
+        finally:
+            self.name = origin_name
+
     def _customized_validate(self) -> MutableValidationResult:
         validation_result = super(Component, self)._customized_validate()
         # If private features are enable and component has code value of type str we need to check
@@ -464,6 +458,11 @@ def _customized_validate(self) -> MutableValidationResult:
 
         return validation_result
 
+    def _get_rest_name_version(self):
+        if self._is_anonymous:
+            return ANONYMOUS_COMPONENT_NAME, self._get_anonymous_hash()
+        return self.name, self.version
+
     def _to_rest_object(self) -> ComponentVersionData:
         component = self._to_dict()
 
@@ -489,7 +488,10 @@ def _to_rest_object(self) -> ComponentVersionData:
             tags=self.tags,
         )
         result = ComponentVersionData(properties=properties)
-        result.name = self.name
+        if self._is_anonymous:
+            result.name = ANONYMOUS_COMPONENT_NAME
+        else:
+            result.name = self.name
         return result
 
     def _to_dict(self) -> Dict:

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_component_operations.py

@@ -300,15 +300,16 @@ def create_or_update(
                 **self._init_args,
             )
 
-        if not (hasattr(component, "_is_anonymous") and component._is_anonymous):
-            component._set_is_anonymous(kwargs.pop("is_anonymous", False))
+        if not component._is_anonymous:
+            component._is_anonymous = kwargs.pop("is_anonymous", False)
+
         if not skip_validation:
             self._validate(component, raise_on_failure=True)
 
         # Create all dependent resources
         self._resolve_arm_id_or_upload_dependencies(component)
 
-        component._update_anonymous_hash()
+        name, version = component._get_rest_name_version()
         rest_component_resource = component._to_rest_object()
         result = None
         try:
@@ -320,8 +321,8 @@ def create_or_update(
                     "registryName": self._registry_name,
                 }
                 poller = self._version_operation.begin_create_or_update(
-                    name=component.name,
-                    version=component.version,
+                    name=name,
+                    version=version,
                     resource_group_name=self._operation_scope.resource_group_name,
                     registry_name=self._registry_name,
                     body=rest_component_resource,
@@ -335,8 +336,8 @@ def create_or_update(
 
             else:
                 result = self._version_operation.create_or_update(
-                    name=rest_component_resource.name,
-                    version=component.version,
+                    name=name,
+                    version=version,
                     resource_group_name=self._resource_group_name,
                     workspace_name=self._workspace_name,
                     body=rest_component_resource,

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

@@ -792,10 +792,6 @@ def default_optional_pipeline():
         # TODO: optional_param_with_default should also exists
         assert len(pipeline_job.jobs["default_optional_component_1"].inputs) == 2
 
-    @pytest.mark.skipif(
-        not is_live(),
-        reason="TODO 2144070: recording is not stable for this test before the fix after we enable on-disk cache",
-    )
     def test_pipeline_with_none_parameter_has_default_optional_false(self, client: MLClient) -> None:
         default_optional_func = load_component(source=str(components_dir / "default_optional_component.yml"))
 
@@ -844,7 +840,7 @@ def pipeline_with_default_optional_parameters(
         )
         pipeline_job = client.jobs.create_or_update(pipeline, experiment_name="default_optional_pipeline")
 
-        # only the two required input exists
+        # only the two required inputs exists
         assert len(next(pipeline_job.jobs.values().__iter__()).inputs) == 2
         validate_result = pipeline._validate()
         assert validate_result.passed is True
@@ -1456,8 +1452,10 @@ def test_anon_component_in_pipeline(
         mpi_func = load_component(source=str(components_dir / "helloworld_component_mpi.yml"))
         assert mpi_func._validate().passed
 
+        invalid_component_name = "_invalid"
+
         # name of anonymous component in pipeline job should be overwritten
-        mpi_func.name = "_invalid"
+        mpi_func.name = invalid_component_name
         assert not mpi_func._validate().passed
 
         @dsl.pipeline(
@@ -1481,6 +1479,11 @@ def pipeline_distribution_components(job_in_number, job_in_path):
         created_job: PipelineJob = client.jobs.create_or_update(
             pipeline, experiment_name=experiment_name, continue_on_step_failure=True
         )
+        # Theoretically, we should keep the invalid name in request body,
+        # as component name valid in azureml-components maybe invalid in azure-ai-ml.
+        # So we leave this validation to server-side for now.
+        assert mpi_func._to_rest_object().properties.component_spec["name"] == invalid_component_name
+
         # continue_on_step_failure can't be set in create_or_update
         assert created_job.settings.continue_on_step_failure is False
         assert created_job.jobs["hello_world_component_mpi"].component.startswith(ANONYMOUS_COMPONENT_NAME)
@@ -1509,13 +1512,6 @@ def pipeline(job_in_number, job_in_other_number, job_in_path):
         job = client.jobs.create_or_update(pipeline)
         assert job.settings.force_rerun is None
 
-        # 1 interesting case: in client.jobs.create_or_update, component_func1 and/or component_func2
-        # will be renamed to azureml_anonymous in resolution, and the component name will be used as
-        # node name when we use them in the 2nd pipeline job.
-        # After we enabled in-memory cache, only component_func1 will be resolved, and name of component_func2
-        # will keep as randstr("component_name"). Here we manually rename to avoid recording change.
-        component_func2.name = "azureml_anonymous"
-
         @dsl.pipeline(
             name=randstr("pipeline_name"),
             description="The hello world pipeline job",