Pipeline job change to support name + version in job output (Azure#28227)

* add name and version for output

* add asset_name, asset_version, merge 2022-12-01 pr

* resolve comments

* add test and modify related codes

* refine code

* fix failed test

* fix comments

* fix condition._port_name

* fix test and some comments

* revert the modification of compute

* fix comments

* fix pylint error, add tests for CommandJob, Spark, Parallel, Sweep

* delete one test, will add in another pr

* modify error message

---------

Co-authored-by: chenyangliao <[email protected]>

Author: lllilithyang

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

@@ -10,8 +10,8 @@
 from azure.ai.ml._internal._schema.component import NodeType
 from azure.ai.ml._internal.entities.component import InternalComponent
 from azure.ai.ml._internal.entities.node import InternalBaseNode
-from azure.ai.ml._restclient.v2022_10_01_preview.models import CommandJobLimits as RestCommandJobLimits
-from azure.ai.ml._restclient.v2022_10_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
+from azure.ai.ml._restclient.v2022_12_01_preview.models import CommandJobLimits as RestCommandJobLimits
+from azure.ai.ml._restclient.v2022_12_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
 from azure.ai.ml._schema import PathAwareSchema
 from azure.ai.ml._schema.core.fields import DistributionField
 from azure.ai.ml.entities import CommandJobLimits, JobResourceConfiguration

sdk/ml/azure-ai-ml/azure/ai/ml/_ml_client.py

@@ -410,7 +410,7 @@ def __init__(
         self._jobs = JobOperations(
             self._operation_scope,
             self._operation_config,
-            self._service_client_10_2022_preview,
+            self._service_client_12_2022_preview,
             self._operation_container,
             self._credential,
             _service_client_kwargs=kwargs,

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/job/input_output_entry.py

@@ -13,6 +13,7 @@
     StringTransformedEnum,
     UnionField,
     LocalPathField,
+    VersionField,
 )
 
 from azure.ai.ml._schema.core.schema import PatchedSchemaMeta, PathAwareSchema
@@ -134,6 +135,8 @@ def check_dict(self, data, **kwargs):
 
 
 class OutputSchema(PathAwareSchema):
+    name = fields.Str()
+    version = VersionField()
     mode = StringTransformedEnum(
         allowed_values=[
             InputOutputModes.MOUNT,

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/pipeline/control_flow_job.py

@@ -63,11 +63,11 @@ def resolve_inputs_outputs(self, data, **kwargs):  # pylint: disable=no-self-use
         mapping = {}
         for k, v in result.mapping.items():
             v = v if isinstance(v, list) else [v]
-            mapping[k] = [item._name for item in v]
+            mapping[k] = [item._port_name for item in v]
         result._mapping = mapping
 
         try:
-            result._condition = result._condition._name
+            result._condition = result._condition._port_name
         except AttributeError:
             result._condition = result._condition
 

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/pipeline/pipeline_job_io.py

@@ -40,7 +40,8 @@ def _serialize(self, value, attr, obj, **kwargs):
         raise ValidationError(f"Invalid output binding string '{value}' passed")
 
     def _deserialize(self, value, attr, data, **kwargs):
-        if isinstance(value, dict) and "path" in value and "mode" not in value:
+        if isinstance(value, dict) and "path" in value and "mode" not in value and \
+                'name' not in value and 'version' not in value:
             value = value["path"]
         if isinstance(value, str) and re.match(ComponentJobConstants.OUTPUT_PATTERN, value):
             return value

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

@@ -269,14 +269,14 @@ def _map_type(_meta):
                 type=_map_type(meta), description=meta.description, mode=meta.mode, is_control=meta.is_control
             )
             pipeline_output = PipelineOutput(
-                name=key,
+                port_name=key,
                 data=None,
                 meta=output_meta,
                 owner="pipeline",
                 description=self._args_description.get(key, None),
             )
-            value._owner.outputs[value._name]._data = PipelineOutput(
-                name=key,
+            value._owner.outputs[value._port_name]._data = PipelineOutput(
+                port_name=key,
                 data=value._data,
                 meta=None,
                 owner="pipeline",

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/command.py

@@ -13,10 +13,10 @@
 
 from marshmallow import INCLUDE, Schema
 
-from azure.ai.ml._restclient.v2022_10_01_preview.models import CommandJob as RestCommandJob
-from azure.ai.ml._restclient.v2022_10_01_preview.models import CommandJobLimits as RestCommandJobLimits
-from azure.ai.ml._restclient.v2022_10_01_preview.models import JobBase
-from azure.ai.ml._restclient.v2022_10_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
+from azure.ai.ml._restclient.v2022_12_01_preview.models import CommandJob as RestCommandJob
+from azure.ai.ml._restclient.v2022_12_01_preview.models import CommandJobLimits as RestCommandJobLimits
+from azure.ai.ml._restclient.v2022_12_01_preview.models import JobBase
+from azure.ai.ml._restclient.v2022_12_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
 from azure.ai.ml._schema.core.fields import NestedField, UnionField
 from azure.ai.ml._schema.job.command_job import CommandJobSchema
 from azure.ai.ml._schema.job.identity import AMLTokenIdentitySchema, ManagedIdentitySchema, UserIdentitySchema

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/do_while.py

@@ -190,16 +190,16 @@ def _validate_port(self, port, node_ports, port_type, yaml_path):
             validation_result.append_error(
                 yaml_path=yaml_path,
                 message=(
-                    f"{port_obj._name} is the {port_type} of {port_obj._owner.name}, "  # pylint: disable=protected-access
+                    f"{port_obj._port_name} is the {port_type} of {port_obj._owner.name}, "  # pylint: disable=protected-access
                     f"dowhile only accept {port_type} of the body: {self.body.name}."
                 ),
             )
-        elif port_obj is None or port_obj._name not in node_ports:  # pylint: disable=protected-access
+        elif port_obj is None or port_obj._port_name not in node_ports:  # pylint: disable=protected-access
             # Check port is exist in dowhile body.
             validation_result.append_error(
                 yaml_path=yaml_path,
                 message=(
-                    f"The {port_type} of mapping {port_obj._name if port_obj else port} does not "  # pylint: disable=protected-access
+                    f"The {port_type} of mapping {port_obj._port_name if port_obj else port} does not "  # pylint: disable=protected-access
                     f"exist in {self.body.name} {port_type}, existing {port_type}: {node_ports.keys()}"
                 ),
             )
@@ -218,7 +218,7 @@ def _validate_loop_condition(self, raise_error=True):
             )
             if validation_result.passed:
                 # Check condition is a control output.
-                condition_name = self.condition if isinstance(self.condition, str) else self.condition._name
+                condition_name = self.condition if isinstance(self.condition, str) else self.condition._port_name
                 if not self.body._outputs[condition_name].is_control:
                     validation_result.append_error(
                         yaml_path="condition",
@@ -258,7 +258,7 @@ def _validate_body_output_mapping(self, raise_error=True):
             # Validate mapping input&output should come from while body
             for output, inputs in self.mapping.items():
                 # pylint: disable=protected-access
-                output_name = output if isinstance(output, str) else output._name
+                output_name = output if isinstance(output, str) else output._port_name
                 validate_results = self._validate_port(
                     output, self.body.outputs, port_type="output", yaml_path="mapping"
                 )
@@ -271,7 +271,7 @@ def _validate_body_output_mapping(self, raise_error=True):
                         )
                         validation_result.merge_with(input_validate_results)
                         # pylint: disable=protected-access
-                        input_name = item if isinstance(item, str) else item._name
+                        input_name = item if isinstance(item, str) else item._port_name
                         input_output_mapping[input_name] = input_output_mapping.get(input_name, []) + [output_name]
                         is_primitive_type = self.body._inputs[input_name]._meta._is_primitive_type
 

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/parallel.py

@@ -13,7 +13,7 @@
 
 from marshmallow import Schema
 
-from azure.ai.ml._restclient.v2022_10_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
+from azure.ai.ml._restclient.v2022_12_01_preview.models import JobResourceConfiguration as RestJobResourceConfiguration
 from azure.ai.ml.constants._common import ARM_ID_PREFIX
 from azure.ai.ml.constants._component import NodeType
 from azure.ai.ml.entities._component.component import Component

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/spark.py

@@ -12,11 +12,11 @@
 
 from marshmallow import INCLUDE, Schema
 
-from azure.ai.ml._restclient.v2022_10_01_preview.models import IdentityConfiguration
-from azure.ai.ml._restclient.v2022_10_01_preview.models import JobBase as JobBaseData
-from azure.ai.ml._restclient.v2022_10_01_preview.models import SparkJob as RestSparkJob
-from azure.ai.ml._restclient.v2022_10_01_preview.models import SparkJobEntry as RestSparkJobEntry
-from azure.ai.ml._restclient.v2022_10_01_preview.models import (
+from azure.ai.ml._restclient.v2022_12_01_preview.models import IdentityConfiguration
+from azure.ai.ml._restclient.v2022_12_01_preview.models import JobBase as JobBaseData
+from azure.ai.ml._restclient.v2022_12_01_preview.models import SparkJob as RestSparkJob
+from azure.ai.ml._restclient.v2022_12_01_preview.models import SparkJobEntry as RestSparkJobEntry
+from azure.ai.ml._restclient.v2022_12_01_preview.models import (
     SparkResourceConfiguration as RestSparkResourceConfiguration,
 )
 from azure.ai.ml._schema.job.identity import AMLTokenIdentitySchema, ManagedIdentitySchema, UserIdentitySchema