[ML] [Pipelines] Make optional input has consistent behavior in @pipeline (Azure#27904)

* raise error when unprovided pipeline required input without default value binding to optional/required input

* update test

* update

* revert

* update test

* update recording

* update 1.5 scope component code to avoid recording error

* revert test code

* update test json

* update test json

* update test json

* update test json

* update test json

Co-authored-by: Ying Chen <2601502859@qq.com>

Author: YingChen1996

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_inputs_outputs/input.py

@@ -446,8 +446,8 @@ def _normalize_self_properties(self):
                 origin_value = getattr(self, key)
                 new_value = self._simple_parse(origin_value)
                 setattr(self, key, new_value)
-        self.optional = self._simple_parse(getattr(self, "optional", "false"), _type="boolean")
-        self.optional = True if self.optional is True else None
+        if self.optional:
+            self.optional = self._simple_parse(getattr(self, "optional", "false"), _type="boolean")
 
     @classmethod
     def _get_input_by_type(cls, t: type, optional=None):

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

@@ -23,7 +23,7 @@
 from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
 from azure.core.paging import ItemPaged
 
-from test_utilities.utils import assert_job_cancel, sleep_if_live
+from test_utilities.utils import assert_job_cancel, sleep_if_live, omit_with_wildcard
 from .._util import _COMPONENT_TIMEOUT_SECOND
 from ..unittests.test_component_schema import load_component_entity_from_rest_json
 
@@ -146,7 +146,7 @@ def test_command_component(self, client: MLClient, randstr: Callable[[str], str]
                     "optional": True,
                     "type": "number",
                 },
-                "component_in_path": {"description": "A path", "type": "uri_folder"},
+                "component_in_path": {"description": "A path", "type": "uri_folder", 'optional': False},
             },
             "is_deterministic": True,
             "outputs": {"component_out_path": {"type": "uri_folder"}},
@@ -172,9 +172,11 @@ def test_parallel_component(self, client: MLClient, randstr: Callable[[str], str
             "error_threshold": 10,
             "input_data": "${{inputs.score_input}}",
             "inputs": {
-                "label": {"description": "Other reference data for batch scoring, " "e.g. labels.", "type": "uri_file"},
-                "score_input": {"description": "The data to be split and scored in " "parallel.", "type": "mltable"},
-                "score_model": {"description": "The model for batch score.", "type": "custom_model"},
+                "label": {"description": "Other reference data for batch scoring, " "e.g. labels.",
+                          "type": "uri_file", 'optional': False},
+                "score_input": {"description": "The data to be split and scored in " "parallel.", "type": "mltable",
+                                'optional': False},
+                "score_model": {"description": "The model for batch score.", "type": "custom_model", 'optional': False},
             },
             "is_deterministic": True,
             "max_concurrency_per_instance": 12,
@@ -243,7 +245,7 @@ def test_spark_component(self, client: MLClient, randstr: Callable[[], str]) ->
              'description': 'Aml Spark dataset test module',
              'display_name': 'Aml Spark dataset test module',
              'entry': {'file': 'kmeans_example.py'},
-             'inputs': {'file_input': {'type': 'uri_file'}},
+             'inputs': {'file_input': {'type': 'uri_file', 'optional': False}},
              'is_deterministic': True,
              'outputs': {'output': {'type': 'uri_folder'}},
              'type': 'spark',
@@ -283,10 +285,10 @@ def test_command_component_create_input_output_types(
         component_entity._creation_context = None
         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
-        omit_fields = ["id", "command", "environment"]
-        assert pydash.omit(component_entity._to_dict(), *omit_fields) == pydash.omit(
-            target_entity._to_dict(), *omit_fields
-        )
+        # server side will return optional False for optional None input
+        omit_fields = ["id", "command", "environment", "inputs.*.optional"]
+        assert omit_with_wildcard(component_entity._to_dict(), *omit_fields) == \
+               omit_with_wildcard(target_entity._to_dict(), *omit_fields)
 
     def test_command_component_with_code(self, client: MLClient, randstr: Callable[[str], str]) -> None:
         component_name = randstr("component_name")
@@ -340,6 +342,7 @@ def test_component_update(self, client: MLClient, randstr: Callable[[str], str])
             "creation_context",
             "resources",
             "id",
+            "inputs.component_in_path.optional"  # backend will return component inputs as optional:False
         )
         expected_dict = pydash.omit(
             dict(target_entity._to_dict()),
@@ -746,15 +749,15 @@ def test_simple_pipeline_component_create(self, client: MLClient, randstr: Calla
             "display_name": "Hello World Pipeline Component",
             "is_deterministic": False,
             "inputs": {
-                "component_in_path": {"type": "uri_folder", "description": "A path"},
+                "component_in_path": {"type": "uri_folder", "description": "A path", "optional": False},
                 "component_in_number": {
                     "type": "number",
                     "optional": True,
                     "default": "10.99",
                     "description": "A number",
                 },
                 # The azureml: prefix has been resolve and removed by service
-                "node_compute": {"type": "string", "default": "cpu-cluster"},
+                "node_compute": {"type": "string", "default": "cpu-cluster", "optional": False},
             },
             "type": "pipeline",
         }
@@ -784,7 +787,8 @@ def test_helloworld_nested_pipeline_component(self, client: MLClient, randstr: C
             "display_name": "Hello World Pipeline Component",
             "is_deterministic": False,
             "inputs": {
-                "component_in_path": {"type": "uri_folder", "description": "A path for pipeline component"},
+                "component_in_path": {"type": "uri_folder", "description": "A path for pipeline component",
+                                      "optional": False},
                 "component_in_number": {
                     "type": "number",
                     "optional": True,

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

@@ -2346,3 +2346,214 @@ def outer_pipeline():
 
         validate_result = pipeline_job._validate()
         assert validate_result.passed
+
+    def test_dsl_pipeline_with_unprovided_pipeline_optional_input(self, client: MLClient) -> None:
+        component_func = load_component(source=str(components_dir / "default_optional_component.yml"))
+
+        # optional pipeline input binding to optional node input
+        @dsl.pipeline()
+        def pipeline_func(optional_input: Input(optional=True, type="uri_file")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_input=optional_input,
+            )
+
+        pipeline_job = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}
+
+        # optional pipeline parameter binding to optional node parameter
+        @dsl.pipeline()
+        def pipeline_func(optional_param: Input(optional=True, type="string"),
+                          optional_param_duplicate: Input(optional=True, type="string")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_param=optional_param,
+                optional_param_with_default=optional_param_duplicate,
+            )
+
+        pipeline_job = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}
+
+    def test_dsl_pipeline_with_unprovided_pipeline_required_input(self, client: MLClient) -> None:
+        component_func = load_component(source=str(components_dir / "default_optional_component.yml"))
+
+        # required pipeline input binding to optional node input
+        @dsl.pipeline()
+        def pipeline_func(required_input: Input(optional=False, type="uri_file")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_input=required_input,
+            )
+
+        pipeline_job = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {
+            'inputs.required_input': "Required input 'required_input' for pipeline "
+                                     "'pipeline_func' not provided."
+        }
+
+        # required pipeline parameter binding to optional node parameter
+        @dsl.pipeline()
+        def pipeline_func(required_param: Input(optional=False, type="string"),
+                          required_param_duplicate: Input(optional=False, type="string")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_param=required_param,
+                optional_param_with_default=required_param_duplicate,
+            )
+
+        pipeline_job = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {
+            'inputs.required_param': "Required input 'required_param' for pipeline "
+                                     "'pipeline_func' not provided.",
+            'inputs.required_param_duplicate': "Required input 'required_param_duplicate' for pipeline "
+                                               "'pipeline_func' not provided."
+        }
+
+        # required pipeline parameter with default value binding to optional node parameter
+        @dsl.pipeline()
+        def pipeline_func(required_param: Input(optional=False, type="string", default="pipeline_required_param")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param=required_param,
+                optional_param=required_param,
+                optional_param_with_default=required_param,
+            )
+
+        pipeline_job = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}
+
+    def test_dsl_pipeline_with_pipeline_component_unprovided_pipeline_optional_input(self, client: MLClient) -> None:
+        component_func = load_component(source=str(components_dir / "default_optional_component.yml"))
+
+        # optional pipeline input binding to optional node input
+        @dsl.pipeline()
+        def subgraph_pipeline(optional_input: Input(optional=True, type="uri_file")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_input=optional_input,
+            )
+
+        @dsl.pipeline()
+        def root_pipeline():
+            subgraph_node = subgraph_pipeline(
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}
+
+        # optional pipeline parameter binding to optional node parameter
+        @dsl.pipeline()
+        def subgraph_pipeline(optional_parameter: Input(optional=True, type="string"),
+                              optional_parameter_duplicate: Input(optional=True, type="string")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_param=optional_parameter,
+                optional_param_with_default=optional_parameter_duplicate,
+            )
+
+        @dsl.pipeline()
+        def root_pipeline():
+            subgraph_node = subgraph_pipeline(
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}
+
+    def test_dsl_pipeline_with_pipeline_component_unprovided_pipeline_required_input(self, client: MLClient) -> None:
+        component_func = load_component(source=str(components_dir / "default_optional_component.yml"))
+
+        # required pipeline input binding to optional node input
+        @dsl.pipeline()
+        def subgraph_pipeline(required_input: Input(optional=False, type="uri_file")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_input=required_input
+            )
+
+        @dsl.pipeline()
+        def root_pipeline():
+            subgraph_node = subgraph_pipeline(
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {
+            'jobs.subgraph_node.inputs.required_input': "Required input 'required_input' for component 'subgraph_node'"
+                                                        " not provided."
+        }
+
+        @dsl.pipeline()
+        def root_pipeline(required_input: Input(optional=False, type="uri_file")):
+            subgraph_node = subgraph_pipeline(
+                required_input=required_input
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {
+            'inputs.required_input': "Required input 'required_input' for pipeline 'root_pipeline' not provided."
+        }
+
+        # required pipeline parameter binding to optional node parameter
+        @dsl.pipeline()
+        def subgraph_pipeline(required_parameter: Input(optional=False, type="string")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_param=required_parameter
+            )
+
+        @dsl.pipeline()
+        def root_pipeline():
+            subgraph_node = subgraph_pipeline(
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {
+            'jobs.subgraph_node.inputs.required_parameter': "Required input 'required_parameter' for component "
+                                                            "'subgraph_node' not provided."
+        }
+
+        # required pipeline parameter with default value binding to optional node parameter
+        @dsl.pipeline()
+        def subgraph_pipeline(required_parameter: Input(optional=False, type="string", default="subgraph_pipeline")):
+            component_func(
+                required_input=Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv"),
+                required_param="def",
+                optional_param=required_parameter
+            )
+
+        @dsl.pipeline()
+        def root_pipeline():
+            subgraph_node = subgraph_pipeline(
+            )
+
+        pipeline_job = root_pipeline()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        validate_result = pipeline_job._validate()
+        assert validate_result.error_messages == {}

sdk/ml/azure-ai-ml/tests/internal/unittests/test_component.py

@@ -76,7 +76,7 @@ def test_load_from_registered_internal_command_component_rest_obj(self):
             "display_name": "0.0.1",
             "is_deterministic": True,
             "successful_return_code": "Zero",
-            "inputs": {"train_data": {"type": "path"}},
+            "inputs": {"train_data": {"type": "path", "optional": False}},
             "outputs": {"output_dir": {"type": "path", "datastore_mode": "Upload"}},
             "command": "sh ls.sh {inputs.input_dir} {inputs.file_name} {outputs.output_dir}",
             "environment": {"name": "AzureML-Minimal", "version": "45", "os": "Linux"},
@@ -91,7 +91,7 @@ def test_load_from_registered_internal_command_component_rest_obj(self):
             "display_name": "0.0.1",
             "is_deterministic": True,
             "successful_return_code": "Zero",
-            "inputs": {"train_data": {"type": "path"}},  # optional will be drop if False
+            "inputs": {"train_data": {"type": "path", "optional": False}},  # optional will be drop if False
             "outputs": {"output_dir": {"type": "path", "datastore_mode": "Upload"}},
             "command": "sh ls.sh {inputs.input_dir} {inputs.file_name} {outputs.output_dir}",
             "environment": {"name": "AzureML-Minimal", "version": "45", "os": "Linux"},
@@ -136,10 +136,12 @@ def test_load_from_registered_internal_scope_component_rest_obj(self):
             "inputs": {
                 "TextData": {
                     "type": "AnyFile",
+                    'optional': False,
                     "description": "relative path on ADLS storage",
                 },
                 "ExtractionClause": {
                     "type": "string",
+                    'optional': False,
                     "description": 'the extraction clause,something like "column1:string, column2:int"',
                 },
             },
@@ -161,12 +163,12 @@ def test_load_from_registered_internal_scope_component_rest_obj(self):
             "inputs": {
                 "TextData": {
                     "type": "AnyFile",
-                    # "optional": False,  # expected. optional will be dropped if it's False
+                    "optional": False,
                     "description": "relative path on ADLS storage",
                 },
                 "ExtractionClause": {
                     "type": "string",
-                    # "optional": False,  # expected. optional will be dropped if it's False
+                    "optional": False,
                     "description": 'the extraction clause,something like "column1:string, column2:int"',
                 },
             },
@@ -201,9 +203,6 @@ def test_component_serialization(self, yaml_path):
                     input_port["enum"] = list(map(lambda x: str(x), input_port["enum"]))
                 if "default" in input_port:
                     input_port["default"] = str(input_port["default"])
-            # optional will be dropped if it's False
-            if "optional" in input_port and input_port["optional"] is False:
-                del input_port["optional"]
 
         # code will be dumped as absolute path
         if "code" in expected_dict:

sdk/ml/azure-ai-ml/tests/internal/unittests/test_pipeline_job.py

@@ -408,10 +408,6 @@ def test_load_pipeline_job_with_internal_components_as_node(self):
         with open(yaml_path, encoding="utf-8") as yaml_file:
             yaml_dict = yaml.safe_load(yaml_file)
 
-        # handle some known difference in yaml_dict
-        for _input in yaml_dict["inputs"].values():
-            if "optional" in _input and _input["optional"] is False:
-                del _input["optional"]
         yaml_dict["code"] = parse_local_path(yaml_dict["code"], scope_internal_func.base_path)
 
         command_func = load_component("./tests/test_configs/components/helloworld_component.yml")