[ML][Pipelines]support skip-no-change (Azure#30451)

* support skip-no-change

* add test

* update test

* update code logic

* fix test

* fix test

* update test recording

* update test recording

* update log info

* update pylint

* update test

* update recording

* update recording

* update recording

* update recording

* update recording

* update recording

* update recording

Author: Stephen1993

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

@@ -428,10 +428,14 @@ def _get_anonymous_hash(self) -> str:
 
         same anonymous component(same code and interface) will have same name.
         """
-        component_interface_dict = self._to_dict()
         # omit version since anonymous component's version is random guid
         # omit name since name doesn't impact component's uniqueness
-        return hash_dict(component_interface_dict, keys_to_omit=["name", "id", "version"])
+        return self._get_component_hash(keys_to_omit=["name", "id", "version"])
+
+    def _get_component_hash(self, keys_to_omit=None) -> str:
+        """Return the hash of component."""
+        component_interface_dict = self._to_dict()
+        return hash_dict(component_interface_dict, keys_to_omit=keys_to_omit)
 
     @classmethod
     def _get_resource_type(cls) -> str:
@@ -510,14 +514,15 @@ def _to_rest_object(self) -> ComponentVersion:
             component_spec=component,
             description=self.description,
             is_anonymous=self._is_anonymous,
-            properties=self.properties,
+            properties=dict(self.properties) if self.properties else {},
             tags=self.tags,
         )
         result = ComponentVersion(properties=properties)
         if self._is_anonymous:
             result.name = ANONYMOUS_COMPONENT_NAME
         else:
             result.name = self.name
+            result.properties.properties["client_component_hash"] = self._get_component_hash(keys_to_omit=["version"])
         return result
 
     def _to_dict(self) -> Dict:

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

@@ -10,11 +10,13 @@
 from inspect import Parameter, signature
 from typing import Any, Callable, Dict, Iterable, List, Optional, Union
 
+from azure.core.exceptions import ResourceNotFoundError
+
 from azure.ai.ml._restclient.v2021_10_01_dataplanepreview import (
     AzureMachineLearningWorkspaces as ServiceClient102021Dataplane,
 )
 from azure.ai.ml._restclient.v2022_10_01 import AzureMachineLearningWorkspaces as ServiceClient102022
-from azure.ai.ml._restclient.v2022_10_01.models import ListViewType
+from azure.ai.ml._restclient.v2022_10_01.models import ComponentVersion, ListViewType
 from azure.ai.ml._scope_dependent_operations import (
     OperationConfig,
     OperationsContainer,
@@ -167,6 +169,36 @@ def list(
             )
         )
 
+    @monitor_with_telemetry_mixin(logger, "ComponentVersion.Get", ActivityType.INTERNALCALL)
+    def _get_component_version(self, name: str, version: Optional[str] = DEFAULT_COMPONENT_VERSION) -> ComponentVersion:
+        """Returns ComponentVersion information about the specified component name and version.
+
+        :param name: Name of the code component.
+        :type name: str
+        :param version: Version of the component.
+        :type version: Optional[str]
+        :return: The ComponentVersion object of the specified component name and version.
+        :rtype: ~azure.ai.ml.entities.ComponentVersion
+        """
+        result = (
+            self._version_operation.get(
+                name=name,
+                version=version,
+                resource_group_name=self._resource_group_name,
+                registry_name=self._registry_name,
+                **self._init_args,
+            )
+            if self._registry_name
+            else self._version_operation.get(
+                name=name,
+                version=version,
+                resource_group_name=self._resource_group_name,
+                workspace_name=self._workspace_name,
+                **self._init_args,
+            )
+        )
+        return result
+
     @monitor_with_telemetry_mixin(logger, "Component.Get", ActivityType.PUBLICAPI)
     def get(self, name: str, version: Optional[str] = None, label: Optional[str] = None) -> Component:
         """Returns information about the specified component.
@@ -201,23 +233,7 @@ def get(self, name: str, version: Optional[str] = None, label: Optional[str] = N
         if label:
             return _resolve_label_to_asset(self, name, label)
 
-        result = (
-            self._version_operation.get(
-                name=name,
-                version=version,
-                resource_group_name=self._resource_group_name,
-                registry_name=self._registry_name,
-                **self._init_args,
-            )
-            if self._registry_name
-            else self._version_operation.get(
-                name=name,
-                version=version,
-                resource_group_name=self._resource_group_name,
-                workspace_name=self._workspace_name,
-                **self._init_args,
-            )
-        )
+        result = self._get_component_version(name, version)
         component = Component._from_rest_object(result)
         self._resolve_dependencies_for_pipeline_component_jobs(
             component,
@@ -353,6 +369,26 @@ def create_or_update(
         name, version = component._get_rest_name_version()
         rest_component_resource = component._to_rest_object()
         result = None
+        try:
+            if not component._is_anonymous and kwargs.get("skip_if_no_change"):
+                client_component_hash = rest_component_resource.properties.properties.get("client_component_hash")
+                remote_component_version = self._get_component_version(name=name)  # will raise error if not found.
+                remote_component_hash = remote_component_version.properties.properties.get("client_component_hash")
+                if client_component_hash == remote_component_hash:
+                    component.version = remote_component_version.properties.component_spec.get(
+                        "version"
+                    )  # only update the default version component instead of creating a new version component
+                    version = component.version
+                    rest_component_resource = component._to_rest_object()
+                    logger.warning(
+                        "The component is not modified compared to the default version "
+                        "and the new version component registration is skipped."
+                    )
+        except ResourceNotFoundError as e:
+            logger.info("Failed to get component version, %s", e)
+        except Exception as e:  # pylint: disable=broad-except
+            logger.error("Failed to compare client_component_hash, %s", e)
+
         try:
             if self._registry_name:
                 start_time = time.time()

sdk/ml/azure-ai-ml/tests/component/e2etests/test_component.py

@@ -1,28 +1,30 @@
 import re
+import time
 import uuid
 from itertools import tee
 from pathlib import Path
 from typing import Callable
 
 import pydash
 import pytest
+from azure.core.exceptions import HttpResponseError
+from azure.core.paging import ItemPaged
+from devtools_testutils import AzureRecordedTestCase, is_live
+from test_utilities.utils import assert_job_cancel, omit_with_wildcard, sleep_if_live
+
 from azure.ai.ml import MLClient, MpiDistribution, load_component, load_environment
 from azure.ai.ml._restclient.v2022_05_01.models import ListViewType
 from azure.ai.ml._utils._arm_id_utils import is_ARM_id_for_resource
+from azure.ai.ml.constants._assets import IPProtectionLevel
 from azure.ai.ml.constants._common import (
     ANONYMOUS_COMPONENT_NAME,
     ARM_ID_PREFIX,
     PROVIDER_RESOURCE_ID_WITH_VERSION,
     AzureMLResourceType,
 )
-from azure.ai.ml.constants._assets import IPProtectionLevel
 from azure.ai.ml.dsl._utils import _sanitize_python_variable_name
 from azure.ai.ml.entities import CommandComponent, Component, PipelineComponent
 from azure.ai.ml.entities._load_functions import load_code, load_job
-from azure.core.exceptions import HttpResponseError
-from azure.core.paging import ItemPaged
-from devtools_testutils import AzureRecordedTestCase, is_live
-from test_utilities.utils import assert_job_cancel, omit_with_wildcard, sleep_if_live
 
 from .._util import _COMPONENT_TIMEOUT_SECOND
 from ..unittests.test_component_schema import load_component_entity_from_rest_json
@@ -34,6 +36,7 @@ def create_component(
     path="./tests/test_configs/components/helloworld_component.yml",
     params_override=None,
     is_anonymous=False,
+    **kwargs,
 ):
     default_param_override = [{"name": component_name}]
     if params_override is None:
@@ -45,7 +48,7 @@ def create_component(
         source=path,
         params_override=params_override,
     )
-    return client.components.create_or_update(command_component, is_anonymous=is_anonymous)
+    return client.components.create_or_update(command_component, is_anonymous=is_anonymous, **kwargs)
 
 
 @pytest.fixture
@@ -383,7 +386,7 @@ def test_command_component_create_input_output_types(
         assert target_entity.id
         # server side will remove \n from the code now. Skip them given it's not targeted to check in this test
         # server side will return optional False for optional None input
-        omit_fields = ["id", "command", "environment", "inputs.*.optional"]
+        omit_fields = ["id", "command", "environment", "inputs.*.optional", "properties"]
         assert omit_with_wildcard(component_entity._to_dict(), *omit_fields) == omit_with_wildcard(
             target_entity._to_dict(), *omit_fields
         )
@@ -1026,3 +1029,53 @@ def test_ipp_component_create(self, ipp_registry_client: MLClient, randstr: Call
             from_rest_component.outputs["model_output_ipp"]._intellectual_property
             == command_component.outputs["model_output_ipp"]._intellectual_property
         )
+
+    def test_create_component_skip_if_no_change(self, client: MLClient, randstr):
+        component_operation = client._operation_container.all_operations[AzureMLResourceType.COMPONENT]
+        component_name = "test_skip_if_no_change"
+        try:
+            default_component = component_operation.get(name=component_name)
+        except Exception:
+            default_component = None
+        default_version = default_component.version if default_component else "1"
+        # update  default component by current local component data.
+        default_component = create_component(client, component_name=component_name, version=default_version)
+
+        # test component has no change and use skip_if_no_change parameter
+        new_version = randstr("component_version")
+        new_component = create_component(
+            client, component_name=component_name, version=new_version, skip_if_no_change=True
+        )
+        assert default_component._get_component_hash(
+            keys_to_omit=["creation_context"]
+        ) == new_component._get_component_hash(keys_to_omit=["creation_context"])
+
+        # test component has change and use skip_if_no_change parameter
+        new_version = randstr("component_version")
+        params_override = [
+            {"description": "description_{0}".format(new_version)},
+            {"display_name": "display_name_{0}".format(new_version)},
+            {"tags": {"tags": "tags_{0}".format(new_version)}},
+        ]
+        new_component = create_component(
+            client,
+            component_name=component_name,
+            version=new_version,
+            params_override=params_override,
+            skip_if_no_change=True,
+        )
+        assert default_component._get_component_hash(
+            keys_to_omit=["creation_context"]
+        ) != new_component._get_component_hash(keys_to_omit=["creation_context"])
+        assert new_component.description == "description_{0}".format(new_version)
+        assert new_component.display_name == "display_name_{0}".format(new_version)
+        assert new_component.tags == {"tags": "tags_{0}".format(new_version)}
+        assert new_component.version == new_version
+
+        # test component has no change and not use skip_if_no_change parameter
+        new_version = randstr("component_version")
+        new_component = create_component(client, component_name=component_name, version=new_version)
+        assert default_component._get_component_hash(
+            keys_to_omit=["creation_context"]
+        ) != new_component._get_component_hash(keys_to_omit=["creation_context"])
+        assert new_component.version == new_version

sdk/ml/azure-ai-ml/tests/component/e2etests/test_component_hash.py

@@ -39,6 +39,7 @@ def create_component(
 @pytest.mark.usefixtures(
     "recorded_test",
     "mock_asset_name",
+    "mock_component_hash",
 )
 @pytest.mark.pipeline_test
 class TestComponentHash(AzureRecordedTestCase):

sdk/ml/azure-ai-ml/tests/component/unittests/test_command_component_entity.py

@@ -1,28 +1,26 @@
+import enum
 import os
 import sys
+import tempfile
 from io import StringIO
-import enum
 from pathlib import Path
-import tempfile
 from unittest.mock import patch
 from zipfile import ZipFile
 
 import pydash
 import pytest
-
-from conftest import normalized_arm_id_in_object
-from test_utilities.utils import verify_entity_load_and_dump, build_temp_folder
+from test_utilities.utils import build_temp_folder, verify_entity_load_and_dump
 
 from azure.ai.ml import Input, MpiDistribution, Output, TensorFlowDistribution, command, load_component
 from azure.ai.ml._utils.utils import load_yaml
 from azure.ai.ml.constants._common import AzureMLResourceType
-from azure.ai.ml.entities import Component, CommandComponent, CommandJobLimits, JobResourceConfiguration
-from azure.ai.ml.entities._assets import Code
-from azure.ai.ml.entities._assets import Environment
+from azure.ai.ml.entities import CommandComponent, CommandJobLimits, Component, JobResourceConfiguration
+from azure.ai.ml.entities._assets import Code, Environment
 from azure.ai.ml.entities._builders import Command, Sweep
 from azure.ai.ml.entities._job.pipeline._io import PipelineInput
 from azure.ai.ml.exceptions import UnexpectedKeywordError, ValidationException
 from azure.ai.ml.sweep import Choice
+from conftest import normalized_arm_id_in_object
 
 from .._util import _COMPONENT_TIMEOUT_SECOND
 
@@ -92,7 +90,11 @@ def test_command_component_entity(self):
             environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:33",
         )
         component_dict = component._to_rest_object().as_dict()
-        omits = ["properties.component_spec.$schema", "properties.component_spec._source"]
+        omits = [
+            "properties.component_spec.$schema",
+            "properties.component_spec._source",
+            "properties.properties.client_component_hash",
+        ]
         component_dict = pydash.omit(component_dict, *omits)
 
         yaml_path = "./tests/test_configs/components/basic_component_code_arm_id.yml"
@@ -235,13 +237,15 @@ def test_command_component_instance_count(self):
             "properties.component_spec.distribution.added_property",
             "properties.component_spec.resources.properties",
             "properties.component_spec._source",
+            "properties.properties.client_component_hash",
         )
         yaml_component_dict = pydash.omit(
             yaml_component_dict,
             "properties.component_spec.$schema",
             "properties.component_spec.distribution.added_property",
             "properties.component_spec.resources.properties",
             "properties.component_spec._source",
+            "properties.properties.client_component_hash",
         )
         assert component_dict == yaml_component_dict
 

sdk/ml/azure-ai-ml/tests/component/unittests/test_data_transfer_component_entity.py

@@ -1,19 +1,16 @@
 import pydash
 import pytest
 
-from azure.ai.ml import MLClient
-from azure.ai.ml import load_component
+from azure.ai.ml import MLClient, load_component
+from azure.ai.ml.constants._component import DataCopyMode, DataTransferTaskType
 from azure.ai.ml.entities._component.datatransfer_component import (
     DataTransferCopyComponent,
-    DataTransferImportComponent,
     DataTransferExportComponent,
+    DataTransferImportComponent,
 )
-from azure.ai.ml.constants._component import DataCopyMode, DataTransferTaskType
-from .test_component_schema import (
-    load_component_entity_from_rest_json,
-    load_component_entity_from_yaml,
-)
+
 from .._util import _COMPONENT_TIMEOUT_SECOND
+from .test_component_schema import load_component_entity_from_rest_json, load_component_entity_from_yaml
 
 
 @pytest.mark.timeout(_COMPONENT_TIMEOUT_SECOND)
@@ -148,6 +145,7 @@ def test_copy_task_component_entity(self):
         omit_fields = [
             "properties.component_spec.$schema",
             "properties.component_spec._source",
+            "properties.properties.client_component_hash",
         ]
         component._validate()
         component_dict = component._to_rest_object().as_dict()

sdk/ml/azure-ai-ml/tests/component/unittests/test_parallel_component_entity.py

@@ -68,6 +68,7 @@ def test_parallel_component_entity(self):
             "properties.component_spec.$schema",
             "properties.component_spec.inputs",
             "properties.component_spec._source",
+            "properties.properties.client_component_hash",
         ]
         component_dict = component._to_rest_object().as_dict()
         component_dict = pydash.omit(component_dict, *omit_fields)

sdk/ml/azure-ai-ml/tests/component/unittests/test_spark_component_entity.py

@@ -61,6 +61,7 @@ def test_spark_component_entity(self):
         omit_fields = [
             "properties.component_spec.$schema",
             "properties.component_spec._source",
+            "properties.properties.client_component_hash",
         ]
         component_dict = component._to_rest_object().as_dict()
         component_dict = pydash.omit(component_dict, *omit_fields)