[Pipeline] Fix condition node failed when expression is group input (Azure#30650)

* fix condition node failed when expression is group input

* update test case

* fix code style

* fix failed test case

* fix failed test case

Author: lalala123123

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_pipeline_expression.py

@@ -316,8 +316,12 @@ def _handle_pipeline_input(
                     _postfix = _update_postfix(_postfix, _name, _new_name)
                     _expression_inputs[_new_name] = ExpressionInput(_new_name, _seen_input.type, _seen_input)
             _postfix.append(_name)
+
+            param_input = pipeline_inputs
+            for group_name in _pipeline_input._group_names:
+                param_input = param_input[group_name].values
             _expression_inputs[_name] = ExpressionInput(
-                _name, pipeline_inputs[_pipeline_input._port_name].type, _pipeline_input
+                _name, param_input[_pipeline_input._port_name].type, _pipeline_input
             )
             return _postfix, _expression_inputs
 

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

@@ -8,6 +8,7 @@
 from azure.ai.ml.dsl import pipeline
 from azure.ai.ml.dsl._condition import condition
 from azure.ai.ml.dsl._do_while import do_while
+from azure.ai.ml.dsl._group_decorator import group
 from azure.ai.ml.dsl._parallel_for import parallel_for
 from azure.ai.ml.entities._builders.parallel_for import ParallelFor
 from azure.ai.ml.entities._job.pipeline._io import InputOutputBase, PipelineInput
@@ -208,6 +209,42 @@ def condition_pipeline():
             "result": {"_source": "YAML.COMPONENT", "name": "result", "type": "command"},
         }
 
+    def test_condition_with_group_input(self):
+        hello_world_component_no_paths = load_component(
+            source=r"./tests/test_configs/components/helloworld_component_no_paths.yml"
+        )
+
+        @group
+        class SubGroup:
+            num: int
+
+        @group
+        class ParentGroup:
+            input_group: SubGroup
+
+        @pipeline(
+            compute="cpu-cluster",
+        )
+        def condition_pipeline(group_input: ParentGroup):
+            node1 = hello_world_component_no_paths(component_in_number=1)
+            condition(condition=group_input.input_group.num < 100, true_block=node1)
+
+        pipeline_job = condition_pipeline(group_input=ParentGroup(input_group=SubGroup(num=10)))
+        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"]["expression_component"] == {
+            "environment_variables": {"AZURE_ML_CLI_PRIVATE_FEATURES_ENABLED": "true"},
+            "name": "expression_component",
+            "type": "command",
+            "inputs": {"num": {"job_input_type": "literal", "value": "${{parent.inputs.group_input.input_group.num}}"}},
+            "_source": "YAML.COMPONENT",
+        }
+
 
 class TestDoWhilePipelineUT(TestControlFlowPipelineUT):
     def test_invalid_do_while_pipeline(self):