[ML][Pipelines] Fix if else block intersection check (Azure#29274)

* fix intersection

* fix tests

Author: D-W-

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

@@ -1,7 +1,7 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
-from marshmallow import fields, post_dump
+from marshmallow import fields, post_dump, ValidationError
 
 from azure.ai.ml._schema import StringTransformedEnum
 from azure.ai.ml._schema.core.fields import DataBindingStr, NodeBindingStr, UnionField
@@ -24,4 +24,25 @@ def simplify_blocks(self, data, **kwargs):  # pylint: disable=unused-argument, n
         for block in block_keys:
             if isinstance(data.get(block), list) and len(data.get(block)) == 1:
                 data[block] = data.get(block)[0]
+
+        # validate blocks intersection
+        def _normalize_blocks(key):
+            blocks = data.get(key, [])
+            if blocks:
+                if not isinstance(blocks, list):
+                    blocks = [blocks]
+            else:
+                blocks = []
+            return blocks
+
+        true_block = _normalize_blocks("true_block")
+        false_block = _normalize_blocks("false_block")
+
+        if not true_block and not false_block:
+            raise ValidationError("True block and false block cannot be empty at the same time.")
+
+        intersection = set(true_block).intersection(set(false_block))
+        if intersection:
+            raise ValidationError(f"True block and false block cannot contain same nodes: {intersection}")
+
         return data

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

@@ -121,29 +121,4 @@ def _validate_params(self, raise_error=True) -> MutableValidationResult:
                         message=f"'{name}' of dsl.condition has invalid binding expression: {block}, {error_tail}",
                     )
 
-        def _get_intersection(lst1, lst2):
-            if not lst1:
-                lst1 = []
-            if not lst2:
-                lst2 = []
-            return list(set(lst1) & set(lst2))
-
-        intersection = _get_intersection(self.true_block, self.false_block)
-
-        if not self.true_block and not self.false_block:
-            validation_result.append_error(
-                yaml_path="true_block",
-                message="'true_block' and 'false_block' of dsl.condition node cannot both be empty.",
-            )
-        elif self.true_block is self.false_block:
-            validation_result.append_error(
-                yaml_path="true_block",
-                message="'true_block' and 'false_block' of dsl.condition node cannot be the same object.",
-            )
-        elif intersection:
-            validation_result.append_error(
-                yaml_path="true_block",
-                message="'true_block' and 'false_block' of dsl.condition has intersection.",
-            )
-
         return validation_result.try_raise(self._get_validation_error_target(), raise_error=raise_error)

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

@@ -356,6 +356,43 @@ def condition_pipeline():
             "type": "if_else",
         }
 
+    @pytest.mark.skipif(condition=not is_live(), reason="TODO(2177353): check why recorded tests failure.")
+    def test_if_else_multiple_blocks_subgraph(self, client: MLClient):
+        hello_world_component_no_paths = load_component(
+            source="./tests/test_configs/components/helloworld_component_no_paths.yml"
+        )
+        basic_component = load_component(
+            source="./tests/test_configs/components/component_with_conditional_output/spec.yaml"
+        )
+
+        @pipeline()
+        def subgraph():
+            hello_world_component_no_paths(component_in_number=2)
+
+        @pipeline(
+            compute="cpu-cluster",
+        )
+        def condition_pipeline():
+            result = basic_component()
+
+            node1 = hello_world_component_no_paths(component_in_number=1)
+
+            node2 = subgraph()
+
+            condition(condition=result.outputs.output, true_block=[node1, node2])
+
+        pipeline_job = condition_pipeline()
+        with include_private_preview_nodes_in_pipeline():
+            pipeline_job = assert_job_cancel(pipeline_job, client)
+
+        dsl_pipeline_job_dict = omit_with_wildcard(pipeline_job._to_rest_object().as_dict(), *omit_fields)
+        assert dsl_pipeline_job_dict["properties"]["jobs"]["conditionnode"] == {
+            "_source": "DSL",
+            "condition": "${{parent.jobs.result.outputs.output}}",
+            "true_block": ["${{parent.jobs.node1}}", "${{parent.jobs.node2}}"],
+            "type": "if_else",
+        }
+
 
 @pytest.mark.skipif(
     condition=is_live(),

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

@@ -1,11 +1,14 @@
 import pytest
+from marshmallow import ValidationError
 
 from azure.ai.ml import Input, load_component
 from azure.ai.ml.constants._component import ComponentSource
 from azure.ai.ml.dsl import pipeline
 from azure.ai.ml.dsl._condition import condition
 from azure.ai.ml.dsl._parallel_for import parallel_for
+from azure.ai.ml.entities._job.pipeline._io import InputOutputBase
 from azure.ai.ml.exceptions import ValidationException
+from test_utilities.utils import omit_with_wildcard
 
 from .._util import _DSL_TIMEOUT_SECOND
 
@@ -67,10 +70,139 @@ def condition_pipeline():
             # true block and false block has intersection
             condition(condition=result.outputs.output, false_block=[node1, node2], true_block=[node1])
 
-        with pytest.raises(ValidationException) as e:
+        with pytest.raises(ValidationError) as e:
             pipeline_job = condition_pipeline()
             pipeline_job._validate(raise_error=True)
-        assert "'true_block' and 'false_block' of dsl.condition has intersection" in str(e.value)
+        assert "True block and false block cannot contain same nodes: {'${{parent.jobs.node1}}'" in str(e.value)
+
+        @pipeline(compute="cpu-cluster")
+        def condition_pipeline():
+            result = basic_component()
+            node1 = hello_world_component_no_paths()
+            node2 = hello_world_component_no_paths()
+            # true block and false block has intersection
+            condition(condition=result.outputs.output, false_block=[node1], true_block=[node2])
+
+        # no error raise
+        pipeline_job = condition_pipeline()
+        pipeline_job._validate(raise_error=True)
+
+    def test_create_dsl_condition_illegal_cases(self):
+        basic_component = load_component(
+            source="./tests/test_configs/components/component_with_conditional_output/spec.yaml"
+        )
+        basic_node = basic_component()
+
+        with pytest.raises(ValidationException) as e:
+            node = condition(condition=1, true_block=basic_node)
+            node._validate(raise_error=True)
+
+        assert f"must be an instance of {str}, {bool} or {InputOutputBase}" in str(e.value)
+
+        with pytest.raises(ValidationException) as e:
+            node = condition(condition=basic_node.outputs.output3, true_block=basic_node)
+            node._validate(raise_error=True)
+
+        assert "must have 'is_control' field with value 'True'" in str(e.value)
+
+        with pytest.raises(ValidationError) as e:
+            node = condition(condition="${{parent.jobs.xxx.outputs.output}}")
+            node._validate(raise_error=True)
+
+        assert "True block and false block cannot be empty at the same time." in str(e.value)
+
+        with pytest.raises(ValidationError) as e:
+            node = condition(
+                condition="${{parent.jobs.xxx.outputs.output}}", true_block=basic_node, false_block=basic_node
+            )
+            node._validate(raise_error=True)
+
+        assert "True block and false block cannot contain same nodes" in str(e.value)
+
+        with pytest.raises(ValidationException) as e:
+            node = condition(condition="xxx", true_block=basic_node)
+            node._validate(raise_error=True)
+
+        assert "condition has invalid binding expression: xxx" in str(e.value)
+
+    def test_condition_node(self):
+        hello_world_component_no_paths = load_component(
+            source="./tests/test_configs/components/helloworld_component_no_paths.yml"
+        )
+        node1 = hello_world_component_no_paths()
+        node1.name = "node1"
+        node2 = hello_world_component_no_paths()
+        node2.name = "node2"
+        control_node = condition(
+            condition="${{parent.jobs.condition_predicate.outputs.output}}", false_block=node1, true_block=node2
+        )
+        assert control_node._to_rest_object() == {
+            "_source": "DSL",
+            "condition": "${{parent.jobs.condition_predicate.outputs.output}}",
+            "false_block": "${{parent.jobs.node1}}",
+            "true_block": "${{parent.jobs.node2}}",
+            "type": "if_else",
+        }
+
+        # test boolean type condition
+        control_node = condition(condition=True, false_block=node1, true_block=node2)
+        assert control_node._to_rest_object() == {
+            "_source": "DSL",
+            "condition": True,
+            "false_block": "${{parent.jobs.node1}}",
+            "true_block": "${{parent.jobs.node2}}",
+            "type": "if_else",
+        }
+
+    def test_condition_pipeline(self):
+        basic_component = load_component(
+            source="./tests/test_configs/components/component_with_conditional_output/spec.yaml"
+        )
+
+        hello_world_component_no_paths = load_component(
+            source="./tests/test_configs/components/helloworld_component_no_paths.yml"
+        )
+
+        @pipeline(
+            name="test_mldesigner_component_with_conditional_output",
+            compute="cpu-cluster",
+        )
+        def condition_pipeline():
+            result = basic_component()
+            node1 = hello_world_component_no_paths(component_in_number=1)
+            node2 = hello_world_component_no_paths(component_in_number=2)
+            condition(condition=result.outputs.output, false_block=node1, true_block=node2)
+
+        pipeline_job = condition_pipeline()
+        omit_fields = [
+            "name",
+            "properties.display_name",
+            "properties.jobs.*.componentId",
+            "properties.settings",
+        ]
+        dsl_pipeline_job_dict = omit_with_wildcard(pipeline_job._to_rest_object().as_dict(), *omit_fields)
+        assert dsl_pipeline_job_dict["properties"]["jobs"] == {
+            "conditionnode": {
+                "_source": "DSL",
+                "condition": "${{parent.jobs.result.outputs.output}}",
+                "false_block": "${{parent.jobs.node1}}",
+                "true_block": "${{parent.jobs.node2}}",
+                "type": "if_else",
+            },
+            "node1": {
+                "_source": "YAML.COMPONENT",
+                "inputs": {"component_in_number": {"job_input_type": "literal", "value": "1"}},
+                "name": "node1",
+                "type": "command",
+            },
+            "node2": {
+                "_source": "YAML.COMPONENT",
+                "inputs": {"component_in_number": {"job_input_type": "literal", "value": "2"}},
+                "name": "node2",
+                "type": "command",
+            },
+            "result": {"_source": "YAML.COMPONENT", "name": "result", "type": "command"},
+        }
 
 
 class TestDoWhilePipelineUT(TestControlFlowPipelineUT):

sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_controlflow_pipeline_job.py

@@ -26,24 +26,24 @@ class TestIfElseUI(TestControlFlowPipelineJobUT):
         [
             # None true & None false
             (
-                ValidationException,
+                ValidationError,
                 "./tests/test_configs/pipeline_jobs/invalid/if_else/none_true_none_false.yml",
-                "of dsl.condition node cannot both be empty",
-                '"path": "jobs.conditionnode.true_block",',
+                "True block and false block cannot be empty at the same time.",
+                "",
             ),
             # None true & empty false
             (
-                ValidationException,
+                ValidationError,
                 "./tests/test_configs/pipeline_jobs/invalid/if_else/none_true_empty_false.yml",
-                "of dsl.condition node cannot both be empty",
-                '"path": "jobs.conditionnode.true_block",',
+                "True block and false block cannot be empty at the same time.",
+                "",
             ),
             # true & false intersection
             (
-                ValidationException,
+                ValidationError,
                 "./tests/test_configs/pipeline_jobs/invalid/if_else/true_false_intersection.yml",
-                "of dsl.condition has intersection",
-                '"path": "jobs.conditionnode.true_block",',
+                "True block and false block cannot contain same nodes:",
+                "",
             ),
             # invalid binding
             (