[Parallel Run Step] Support run settings binding to pipeline literal input. (Azure#29355)

* allow prs run settings binding to literal input

* add test case

* e2e test cases

* fix lint

* fix black

---------

Co-authored-by: Xiaole Wen <xiwe@microsoft.com>

Author: bupt-wenxiaole

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

@@ -12,8 +12,6 @@
 from typing import Dict, List, Optional, Union
 
 from marshmallow import Schema
-
-from azure.ai.ml._restclient.v2023_02_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
@@ -166,7 +164,7 @@ def __init__(
         self.environment_variables = {} if environment_variables is None else environment_variables
 
         if isinstance(self.component, ParallelComponent):
-            self.resources = self.resources or self.component.resources
+            self.resources = self.resources or copy.deepcopy(self.component.resources)
             self.input_data = self.input_data or self.component.input_data
             self.max_concurrency_per_instance = (
                 self.max_concurrency_per_instance or self.component.max_concurrency_per_instance
@@ -175,7 +173,7 @@ def __init__(
                 self.mini_batch_error_threshold or self.component.mini_batch_error_threshold
             )
             self.mini_batch_size = self.mini_batch_size or self.component.mini_batch_size
-            self.partition_keys = self.partition_keys or self.component.partition_keys
+            self.partition_keys = self.partition_keys or copy.deepcopy(self.component.partition_keys)
 
             if not self.task:
                 self.task = self.component.task
@@ -266,9 +264,9 @@ def _attr_type_map(cls) -> dict:
             "resources": (dict, JobResourceConfiguration),
             "task": (dict, ParallelTask),
             "logging_level": str,
-            "max_concurrency_per_instance": int,
-            "error_threshold": int,
-            "mini_batch_error_threshold": int,
+            "max_concurrency_per_instance": (str, int),
+            "error_threshold": (str, int),
+            "mini_batch_error_threshold": (str, int),
             "environment_variables": dict,
         }
 
@@ -357,8 +355,7 @@ def _from_rest_object_to_init_params(cls, obj: dict) -> Dict:
                 obj["task"].environment = task_env[len(ARM_ID_PREFIX) :]
 
         if "resources" in obj and obj["resources"]:
-            resources = RestJobResourceConfiguration.from_dict(obj["resources"])
-            obj["resources"] = JobResourceConfiguration._from_rest_object(resources)
+            obj["resources"] = JobResourceConfiguration._from_dict(obj["resources"])
 
         if "partition_keys" in obj and obj["partition_keys"]:
             obj["partition_keys"] = json.dumps(obj["partition_keys"])

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/job_resource_configuration.py

@@ -146,6 +146,12 @@ def _to_rest_object(self) -> RestJobResourceConfiguration:
             shm_size=self.shm_size,
         )
 
+    @classmethod
+    def _from_dict(cls, dct: dict):
+        """Convert a dict to an Input object."""
+        obj = cls(**dict(dct.items()))
+        return obj
+
     @classmethod
     def _from_rest_object(cls, obj: Optional[RestJobResourceConfiguration]) -> Optional["JobResourceConfiguration"]:
         if obj is None:

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

@@ -64,6 +64,81 @@ def check_name_and_version(output, output_name, output_version):
     assert output.version == output_version
 
 
+def build_pipeline_with_parallel_run_function(data, literal_input=None):
+    # command job with dict distribution
+    environment = "AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:33"
+    inputs = {
+        "job_data_path": Input(
+            type=AssetTypes.MLTABLE,
+            path="./tests/test_configs/dataset/mnist-data",
+            mode=InputOutputModes.EVAL_MOUNT,
+        ),
+        "job_data_path_optional": Input(
+            type=AssetTypes.MLTABLE,
+            mode=InputOutputModes.EVAL_MOUNT,
+            optional=True,
+        ),
+    }
+    input_data = "${{inputs.job_data_path}}"
+    outputs = {"job_output_path": Output(type=AssetTypes.URI_FOLDER, mode="rw_mount")}
+    expected_resources = {"instance_count": 2}
+
+    task = RunFunction(
+        code="./tests/test_configs/dsl_pipeline/parallel_component_with_file_input/src/",
+        entry_script="score.py",
+        program_arguments="--job_output_path ${{outputs.job_output_path}}",
+        environment=environment,
+    )
+    logging_level = "DEBUG"
+    max_concurrency_per_instance = 1
+    error_threshold = 1
+    mini_batch_error_threshold = 1
+    mini_batch_size = "5"
+
+    # Parallel from parallel_run_function()
+    parallel_function = parallel_run_function(
+        display_name="my-evaluate-job",
+        inputs=inputs,
+        outputs=outputs,
+        mini_batch_size=mini_batch_size,
+        task=task,
+        logging_level=logging_level,
+        max_concurrency_per_instance=max_concurrency_per_instance,
+        error_threshold=error_threshold,
+        mini_batch_error_threshold=mini_batch_error_threshold,
+        resources=expected_resources,
+        input_data=input_data,
+    )
+    if literal_input is None:
+
+        @dsl.pipeline(experiment_name="test_pipeline_with_parallel_function", default_compute="cpu-cluster")
+        def parallel_in_pipeline(job_data_path):
+            node1 = parallel_function(job_data_path=job_data_path)
+            # TODO 2104247: node1.task will be kept as a local path when submitting the pipeline job.
+            node1.task = None
+            return {
+                "pipeline_output": node1.outputs.job_output_path,
+            }
+
+        return parallel_in_pipeline(data)
+    else:
+
+        @dsl.pipeline(experiment_name="test_pipeline_with_parallel_function", default_compute="cpu-cluster")
+        def parallel_in_pipeline(job_data_path, literal_input):
+            node1 = parallel_function(job_data_path=job_data_path)
+            # TODO 2104247: node1.task will be kept as a local path when submitting the pipeline job.
+            node1.task = None
+            node1.resources.instance_count = literal_input
+            node1.max_concurrency_per_instance = literal_input
+            node1.error_threshold = literal_input
+            node1.mini_batch_error_threshold = literal_input
+            return {
+                "pipeline_output": node1.outputs.job_output_path,
+            }
+
+        return parallel_in_pipeline(data, literal_input)
+
+
 @pytest.mark.usefixtures(
     "enable_environment_id_arm_expansion",
     "enable_pipeline_private_preview_features",
@@ -1568,6 +1643,44 @@ def parallel_in_pipeline(job_data_path, score_model):
         assert_job_input_output_types(pipeline_job)
         assert pipeline_job.settings.default_compute == "cpu-cluster"
 
+    def test_parallel_components_with_tabular_input_bind_to_literal_input(self, client: MLClient) -> None:
+        components_dir = tests_root_dir / "test_configs/dsl_pipeline/parallel_component_with_tabular_input"
+
+        batch_inference = load_component(source=str(components_dir / "tabular_input_e2e.yml"))
+
+        # Construct pipeline
+        @dsl.pipeline(default_compute="cpu-cluster")
+        def parallel_in_pipeline(job_data_path, score_model, literal_input):
+            batch_inference_node = batch_inference(job_data_path=job_data_path, score_model=score_model)
+            batch_inference_node.mini_batch_size = 5
+            batch_inference_node.max_concurrency_per_instance = literal_input
+            batch_inference_node.error_threshold = literal_input
+            batch_inference_node.mini_batch_error_threshold = literal_input
+
+        pipeline = parallel_in_pipeline(
+            job_data_path=Input(
+                type=AssetTypes.MLTABLE,
+                path="./tests/test_configs/dataset/neural-iris-mltable",
+                mode=InputOutputModes.DIRECT,
+            ),
+            score_model=Input(
+                path="./tests/test_configs/model", type=AssetTypes.URI_FOLDER, mode=InputOutputModes.DOWNLOAD
+            ),
+            literal_input=2,
+        )
+        # submit pipeline job
+        pipeline_job = assert_job_cancel(pipeline, client, experiment_name="parallel_in_pipeline")
+
+        # check required fields in job dict
+        job_dict = pipeline_job._to_dict()
+        expected_keys = ["status", "properties", "creation_context"]
+        for k in expected_keys:
+            assert k in job_dict.keys(), f"failed to get {k} in {job_dict}"
+
+        # original job did not change
+        assert_job_input_output_types(pipeline_job)
+        assert pipeline_job.settings.default_compute == "cpu-cluster"
+
     def test_parallel_components_with_file_input(self, client: MLClient) -> None:
         components_dir = tests_root_dir / "test_configs/dsl_pipeline/parallel_component_with_file_input"
 
@@ -1599,67 +1712,12 @@ def parallel_in_pipeline(job_data_path):
         assert pipeline_job.settings.default_compute == "cpu-cluster"
 
     def test_parallel_run_function(self, client: MLClient):
-        # command job with dict distribution
-        environment = "AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:33"
-        inputs = {
-            "job_data_path": Input(
-                type=AssetTypes.MLTABLE,
-                path="./tests/test_configs/dataset/mnist-data",
-                mode=InputOutputModes.EVAL_MOUNT,
-            ),
-            "job_data_path_optional": Input(
-                type=AssetTypes.MLTABLE,
-                mode=InputOutputModes.EVAL_MOUNT,
-                optional=True,
-            ),
-        }
-        input_data = "${{inputs.job_data_path}}"
-        outputs = {"job_output_path": Output(type=AssetTypes.URI_FOLDER, mode="rw_mount")}
-        expected_resources = {"instance_count": 2}
-
-        task = RunFunction(
-            code="./tests/test_configs/dsl_pipeline/parallel_component_with_file_input/src/",
-            entry_script="score.py",
-            program_arguments="--job_output_path ${{outputs.job_output_path}}",
-            environment=environment,
-        )
-        logging_level = "DEBUG"
-        max_concurrency_per_instance = 1
-        error_threshold = 1
-        mini_batch_error_threshold = 1
-        mini_batch_size = "5"
-
-        # Parallel from parallel_run_function()
-        parallel_function = parallel_run_function(
-            display_name="my-evaluate-job",
-            inputs=inputs,
-            outputs=outputs,
-            mini_batch_size=mini_batch_size,
-            task=task,
-            logging_level=logging_level,
-            max_concurrency_per_instance=max_concurrency_per_instance,
-            error_threshold=error_threshold,
-            mini_batch_error_threshold=mini_batch_error_threshold,
-            resources=expected_resources,
-            input_data=input_data,
-        )
-
         data = Input(
             type=AssetTypes.MLTABLE,
             path="./tests/test_configs/dataset/mnist-data",
             mode=InputOutputModes.EVAL_MOUNT,
         )
-
-        @dsl.pipeline(experiment_name="test_pipeline_with_parallel_function", default_compute="cpu-cluster")
-        def parallel_in_pipeline(job_data_path):
-            node1 = parallel_function(job_data_path=job_data_path)
-            # TODO 2104247: node1.task will be kept as a local path when submitting the pipeline job.
-            node1.task = None
-            return {
-                "pipeline_output": node1.outputs.job_output_path,
-            }
-
-        pipeline = parallel_in_pipeline(data)
+        pipeline = build_pipeline_with_parallel_run_function(data)
 
         pipeline_job = client.create_or_update(pipeline)  # submit pipeline job
 
@@ -1712,6 +1770,65 @@ def parallel_in_pipeline(job_data_path):
         assert_job_input_output_types(pipeline_job)
         assert pipeline_job.settings.default_compute == "cpu-cluster"
 
+    def test_parallel_run_function_run_settings_bind_to_literal_input(self, client: MLClient):
+        data = Input(
+            type=AssetTypes.MLTABLE,
+            path="./tests/test_configs/dataset/mnist-data",
+            mode=InputOutputModes.EVAL_MOUNT,
+        )
+        pipeline = build_pipeline_with_parallel_run_function(data, 2)
+
+        pipeline_job = client.create_or_update(pipeline)  # submit pipeline job
+
+        actual_job = omit_with_wildcard(pipeline_job._to_rest_object().properties.as_dict(), *common_omit_fields)
+        expected_job = {
+            "tags": {},
+            "is_archived": False,
+            "job_type": "Pipeline",
+            "inputs": {
+                "job_data_path": {"job_input_type": "mltable", "mode": "EvalMount"},
+                "literal_input": {"job_input_type": "literal", "value": "2"},
+            },
+            "jobs": {
+                "node1": {
+                    "input_data": "${{inputs.job_data_path}}",
+                    "display_name": "my-evaluate-job",
+                    "inputs": {
+                        "job_data_path": {
+                            "job_input_type": "literal",
+                            "value": "${{parent.inputs.job_data_path}}",
+                        }
+                    },
+                    "name": "node1",
+                    "mini_batch_size": 5,
+                    "logging_level": "DEBUG",
+                    "max_concurrency_per_instance": "${{parent.inputs.literal_input}}",
+                    "error_threshold": "${{parent.inputs.literal_input}}",
+                    "mini_batch_error_threshold": "${{parent.inputs.literal_input}}",
+                    "outputs": {"job_output_path": {"type": "literal", "value": "${{parent.outputs.pipeline_output}}"}},
+                    "resources": {"instance_count": "${{parent.inputs.literal_input}}"},
+                    "type": "parallel",
+                },
+            },
+            "outputs": {
+                "pipeline_output": {
+                    "mode": "ReadWriteMount",
+                    "job_output_type": "uri_folder",
+                }
+            },
+            "settings": {"default_compute": "cpu-cluster"},
+        }
+        assert expected_job == actual_job
+        # check required fields in job dict
+        job_dict = pipeline_job._to_dict()
+        expected_keys = ["status", "properties", "creation_context"]
+        for k in expected_keys:
+            assert k in job_dict.keys(), f"failed to get {k} in {job_dict}"
+
+        # original job did not change
+        assert_job_input_output_types(pipeline_job)
+        assert pipeline_job.settings.default_compute == "cpu-cluster"
+
     def test_parallel_job(self, randstr: Callable[[str], str], client: MLClient):
         environment = "AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:33"
         inputs = {

sdk/ml/azure-ai-ml/tests/pipeline_job/e2etests/test_pipeline_job.py

@@ -541,6 +541,58 @@ def test_pipeline_job_with_parallel_job(
         # assert on the number of converted jobs to make sure we didn't drop the parallel job
         assert len(created_job.jobs.items()) == 1
 
+    @pytest.mark.parametrize(
+        "pipeline_job_path",
+        [
+            "file_component_literal_input_e2e.yml",
+        ],
+    )
+    def test_pipeline_job_with_parallel_component_job_bind_to_literal_input(
+        self, client: MLClient, randstr: Callable[[str], str], pipeline_job_path: str
+    ) -> None:
+        base_file_name = "./tests/test_configs/pipeline_jobs/helloworld_pipeline_job_defaults_with_parallel_job_"
+        params_override = [{"name": randstr("name")}]
+        pipeline_job = load_job(
+            source=base_file_name + pipeline_job_path,
+            params_override=params_override,
+        )
+        created_job = client.jobs.create_or_update(pipeline_job)
+
+        for job in created_job.jobs.values():
+            # The parallel job must be translated to component job in the pipeline job.
+            assert isinstance(job, Parallel)
+
+        # assert on the number of converted jobs to make sure we didn't drop the parallel job
+        assert len(created_job.jobs.items()) == 1
+
+    @pytest.mark.skip(
+        reason="The task for fixing this is tracked by "
+        "https://msdata.visualstudio.com/Vienna/_workitems/edit/2298433"
+    )
+    @pytest.mark.parametrize(
+        "pipeline_job_path",
+        [
+            "file_literal_input_e2e.yml",
+        ],
+    )
+    def test_pipeline_job_with_inline_parallel_job_bind_to_literal_input(
+        self, client: MLClient, randstr: Callable[[str], str], pipeline_job_path: str
+    ) -> None:
+        base_file_name = "./tests/test_configs/pipeline_jobs/helloworld_pipeline_job_defaults_with_parallel_job_"
+        params_override = [{"name": randstr("name")}]
+        pipeline_job = load_job(
+            source=base_file_name + pipeline_job_path,
+            params_override=params_override,
+        )
+        created_job = client.jobs.create_or_update(pipeline_job)
+
+        for job in created_job.jobs.values():
+            # The parallel job must be translated to component job in the pipeline job.
+            assert isinstance(job, Parallel)
+
+        # assert on the number of converted jobs to make sure we didn't drop the parallel job
+        assert len(created_job.jobs.items()) == 1
+
     def test_pipeline_job_with_multiple_parallel_job(self, client: MLClient, randstr: Callable[[str], str]) -> None:
         params_override = [{"name": randstr("name")}]
         pipeline_job = load_job(