support xxx model (Azure#28225)

Author: D-W-

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

@@ -38,6 +38,11 @@ class ParallelFor(LoopNode, NodeIOMixin):
         AssetTypes.URI_FILE: AssetTypes.MLTABLE,
         AssetTypes.URI_FOLDER: AssetTypes.MLTABLE,
         AssetTypes.MLTABLE: AssetTypes.MLTABLE,
+        AssetTypes.MLFLOW_MODEL: AssetTypes.MLTABLE,
+        AssetTypes.TRITON_MODEL: AssetTypes.MLTABLE,
+        AssetTypes.CUSTOM_MODEL: AssetTypes.MLTABLE,
+        # legacy path support
+        "path": AssetTypes.MLTABLE,
         ComponentParameterTypes.NUMBER: ComponentParameterTypes.STRING,
         ComponentParameterTypes.STRING: ComponentParameterTypes.STRING,
         ComponentParameterTypes.BOOLEAN: ComponentParameterTypes.STRING,
@@ -129,6 +134,8 @@ def _convert_output_meta(self, outputs):
             if output.type in self.OUT_TYPE_MAPPING:
                 new_type = self.OUT_TYPE_MAPPING[output.type]
             else:
+                # when loop body introduces some new output type, this will be raised as a reminder to support is in
+                # parallel for
                 raise UserErrorException(
                     "Referencing output with type {} is not supported in parallel_for node.".format(output.type)
                 )

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

@@ -259,6 +259,10 @@ def my_pipeline():
             ({"type": "uri_file"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
             ({"type": "uri_folder"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
             ({"type": "mltable"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
+            ({"type": "mlflow_model"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
+            ({"type": "triton_model"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
+            ({"type": "custom_model"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
+            ({"type": "path"}, {'job_output_type': 'mltable'}, {'type': 'mltable'}, True),
             ({"type": "number"}, {}, {'type': 'string'}, False),
             ({"type": "string", "is_control": True}, {}, {'type': 'string', "is_control": True}, False),
             ({"type": "boolean", "is_control": True}, {}, {'type': 'string', "is_control": True}, False),
@@ -298,30 +302,3 @@ def my_pipeline():
         pipeline_component = my_job.component
         rest_component = pipeline_component._to_rest_object().as_dict()
         assert rest_component["properties"]["component_spec"]["outputs"] == {'output': component_out_dict}
-
-    @pytest.mark.parametrize(
-        "out_type", ["mlflow_model", "triton_model", "custom_model"]
-    )
-    def test_parallel_for_output_unsupported_case(self, out_type):
-        basic_component = load_component(
-            source="./tests/test_configs/components/helloworld_component.yml",
-            params_override=[
-                {"outputs.component_out_path": {"type": out_type}}
-            ]
-        )
-
-        @pipeline
-        def my_pipeline():
-            body = basic_component(component_in_path=Input(path="test_path1"))
-
-            parallel_for(
-                body=body,
-                items={
-                    "iter1": {"component_in_number": 1},
-                    "iter2": {"component_in_number": 2}
-                }
-            )
-
-        with pytest.raises(UserErrorException) as e:
-            my_pipeline()
-        assert f"Referencing output with type {out_type} is not supported" in str(e.value)