[ParallelRunStep] Enable mini_batch_size and retry_settings bind to pipeline input. (#33500)

* allow prs run settings binding to literal input

* v1

* revert

* revert

* revert e2e test

* reformat

---------

Co-authored-by: Xiaole Wen <[email protected]>

Author: bupt-wenxiaole

Diff for: sdk/ml/azure-ai-ml/azure/ai/ml/_schema/component/retry_settings.py

@@ -5,8 +5,9 @@
 from marshmallow import fields
 
 from azure.ai.ml._schema.core.schema import PatchedSchemaMeta
+from azure.ai.ml._schema.core.fields import DataBindingStr, UnionField
 
 
 class RetrySettingsSchema(metaclass=PatchedSchemaMeta):
-    timeout = fields.Int()
-    max_retries = fields.Int()
+    timeout = UnionField([fields.Int(), DataBindingStr])
+    max_retries = UnionField([fields.Int(), DataBindingStr])

Diff for: sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/parallel.py

@@ -33,6 +33,7 @@
 from .._job.parallel.retry_settings import RetrySettings
 from .._job.pipeline._io import NodeOutput, NodeWithGroupInputMixin
 from .._util import convert_ordered_dict_to_dict, get_rest_dict_for_node_attrs, validate_attribute_type
+from ..._utils.utils import is_data_binding_expression
 from .base_node import BaseNode
 
 module_logger = logging.getLogger(__name__)
@@ -166,7 +167,11 @@ def __init__(
 
         self._task = task
 
-        if mini_batch_size is not None and not isinstance(mini_batch_size, int):
+        if (
+            mini_batch_size is not None
+            and not isinstance(mini_batch_size, int)
+            and not is_data_binding_expression(mini_batch_size)
+        ):
             """Convert str to int."""  # pylint: disable=pointless-string-statement
             pattern = re.compile(r"^\d+([kKmMgG][bB])*$")
             if not pattern.match(mini_batch_size):

Diff for: sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/parallel/retry_settings.py

@@ -27,8 +27,8 @@ class RetrySettings(RestTranslatableMixin, DictMixin):
     def __init__(
         self,
         *,
-        timeout: Optional[int] = None,
-        max_retries: Optional[int] = None,
+        timeout: Optional[Union[int, str]] = None,
+        max_retries: Optional[Union[int, str]] = None,
         **kwargs,  # pylint: disable=unused-argument
     ):
         self.timeout = timeout