two version of R2R are here HEAD master

13 files changed, 4199 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/__init__.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/__init__.py
new file mode 100644
index 00000000..fdf8caba
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/__init__.py
@@ -0,0 +1,5 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_attr_dict.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_attr_dict.py
new file mode 100644
index 00000000..cf8d92be
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_attr_dict.py
@@ -0,0 +1,161 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+
+import logging
+from abc import ABC
+from typing import Any, Dict, Generic, List, Optional, TypeVar
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+
+class _AttrDict(Generic[K, V], Dict, ABC):
+    """This class is used for accessing values with instance.some_key. It supports the following scenarios:
+
+    1. Setting arbitrary attribute, eg: obj.resource_layout.node_count = 2
+      1.1 Setting same nested filed twice will return same object, eg:
+              obj.resource_layout.node_count = 2
+              obj.resource_layout.process_count_per_node = 2
+              obj.resource_layout will be {"node_count": 2, "process_count_per_node": 2}
+      1.2 Only public attribute is supported, eg: obj._resource_layout._node_count = 2 will raise AttributeError
+      1.3 All set attribute can be recorded, eg:
+              obj.target = "aml"
+              obj.resource_layout.process_count_per_node = 2
+              obj.get_attr() will return {"target": "aml", "resource_layout": {"process_count_per_node": 2}}
+    2. Getting arbitrary attribute, getting non-exist attribute will return an empty dict.
+    3. Calling arbitrary methods is not allowed, eg: obj.resource_layout() should raise AttributeError
+    """
+
+    def __init__(self, allowed_keys: Optional[Dict] = None, **kwargs: Any):
+        """Initialize a attribute dictionary.
+
+        :param allowed_keys: A dictionary of keys that allowed to set as arbitrary attributes. None means all keys can
+            be set as arbitrary attributes.
+
+        :type dict
+        :param kwargs: A dictionary of additional configuration parameters.
+        :type kwargs: dict
+        """
+        super(_AttrDict, self).__init__(**kwargs)
+        if allowed_keys is None:
+            # None allowed_keys means no restriction on keys can be set for _AttrDict
+            self._allowed_keys = {}
+            self._key_restriction = False
+        else:
+            # Otherwise use allowed_keys to restrict keys can be set for _AttrDict
+            self._allowed_keys = dict(allowed_keys)
+            self._key_restriction = True
+        self._logger = logging.getLogger("attr_dict")
+
+    def _initializing(self) -> bool:
+        # use this to indicate ongoing init process, sub class need to make sure this return True during init process.
+        return False
+
+    def _get_attrs(self) -> dict:
+        """Get all arbitrary attributes which has been set, empty values are excluded.
+
+        :return: A dict which contains all arbitrary attributes set by user.
+        :rtype: dict
+        """
+
+        # TODO: check this
+        def remove_empty_values(data: Dict) -> Dict:
+            if not isinstance(data, dict):
+                return data
+            # skip empty dicts as default value of _AttrDict is empty dict
+            return {k: remove_empty_values(v) for k, v in data.items() if v or not isinstance(v, dict)}
+
+        return remove_empty_values(self)
+
+    def _is_arbitrary_attr(self, attr_name: str) -> bool:
+        """Checks if a given attribute name should be treat as arbitrary attribute.
+
+        Attributes inside _AttrDict can be non-arbitrary attribute or arbitrary attribute.
+        Non-arbitrary attributes are normal attributes like other object which stores in self.__dict__.
+        Arbitrary attributes are attributes stored in the dictionary it self, what makes it special it it's value
+        can be an instance of _AttrDict
+        Take `obj = _AttrDict(allowed_keys={"resource_layout": {"node_count": None}})` as an example.
+        `obj.some_key` is accessing non-arbitrary attribute.
+        `obj.resource_layout` is accessing arbitrary attribute, user can use `obj.resource_layout.node_count = 1` to
+        assign value to it.
+
+        :param attr_name: Attribute name
+        :type attr_name: str
+        :return: If the given attribute name should be treated as arbitrary attribute.
+        :rtype: bool
+        """
+        # Internal attribute won't be set as arbitrary attribute.
+        if attr_name.startswith("_"):
+            return False
+        # All attributes set in __init__ won't be set as arbitrary attribute
+        if self._initializing():
+            return False
+        # If there's key restriction, only keys in it can be set as arbitrary attribute.
+        if self._key_restriction and attr_name not in self._allowed_keys:
+            return False
+        # Attributes already in attribute dict will not be set as arbitrary attribute.
+        try:
+            self.__getattribute__(attr_name)
+        except AttributeError:
+            return True
+        return False
+
+    def __getattr__(self, key: Any) -> Any:
+        if not self._is_arbitrary_attr(key):
+            return super().__getattribute__(key)
+        self._logger.debug("getting %s", key)
+        try:
+            return super().__getitem__(key)
+        except KeyError:
+            allowed_keys = self._allowed_keys.get(key, None) if self._key_restriction else None
+            result: Any = _AttrDict(allowed_keys=allowed_keys)
+            self.__setattr__(key, result)
+            return result
+
+    def __setattr__(self, key: Any, value: V) -> None:
+        if not self._is_arbitrary_attr(key):
+            super().__setattr__(key, value)
+        else:
+            self._logger.debug("setting %s to %s", key, value)
+            super().__setitem__(key, value)
+
+    def __setitem__(self, key: Any, value: V) -> None:
+        self.__setattr__(key, value)
+
+    def __getitem__(self, item: V) -> Any:
+        # support attr_dict[item] since dumping it in marshmallow requires this.
+        return self.__getattr__(item)
+
+    def __dir__(self) -> List:
+        # For Jupyter Notebook auto-completion
+        return list(super().__dir__()) + list(self.keys())
+
+
+def has_attr_safe(obj: Any, attr: Any) -> bool:
+    if isinstance(obj, _AttrDict):
+        has_attr = not obj._is_arbitrary_attr(attr)
+    elif isinstance(obj, dict):
+        return attr in obj
+    else:
+        has_attr = hasattr(obj, attr)
+    return has_attr
+
+
+def try_get_non_arbitrary_attr(obj: Any, attr: str) -> Optional[Any]:
+    """Try to get non-arbitrary attribute for potential attribute dict.
+
+    Will not create target attribute if it is an arbitrary attribute in _AttrDict.
+
+    :param obj: The obj
+    :type obj: Any
+    :param attr: The attribute name
+    :type attr: str
+    :return: obj.attr
+    :rtype: Any
+    """
+    if has_attr_safe(obj, attr):
+        return obj[attr] if isinstance(obj, dict) else getattr(obj, attr)
+    return None
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_component_translatable.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_component_translatable.py
new file mode 100644
index 00000000..22be939d
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_component_translatable.py
@@ -0,0 +1,412 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+# pylint: disable=protected-access, redefined-builtin
+# disable redefined-builtin to use input as argument name
+import re
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+
+from pydash import get
+
+from azure.ai.ml._utils.utils import is_data_binding_expression
+from azure.ai.ml.constants._common import AssetTypes
+from azure.ai.ml.constants._component import ComponentJobConstants
+from azure.ai.ml.entities._inputs_outputs import Input, Output
+from azure.ai.ml.entities._job.pipeline._io import PipelineInput, PipelineOutput
+from azure.ai.ml.entities._job.sweep.search_space import Choice, Randint, SweepDistribution
+from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, JobException
+
+# avoid circular import error
+if TYPE_CHECKING:
+    from azure.ai.ml.entities._builders import BaseNode
+    from azure.ai.ml.entities._component.component import Component
+
+
+class ComponentTranslatableMixin:
+    _PYTHON_SDK_TYPE_MAPPING = {
+        float: "number",
+        int: "integer",
+        bool: "boolean",
+        str: "string",
+    }
+
+    @classmethod
+    def _find_source_from_parent_inputs(cls, input: str, pipeline_job_inputs: dict) -> Tuple[str, Optional[str]]:
+        """Find source type and mode of input/output from parent input.
+
+        :param input: The input name
+        :type input: str
+        :param pipeline_job_inputs: The pipeline job inputs
+        :type pipeline_job_inputs: dict
+        :return: A 2-tuple of the type and the mode
+        :rtype: Tuple[str, Optional[str]]
+        """
+        _input_name = input.split(".")[2][:-2]
+        if _input_name not in pipeline_job_inputs.keys():
+            msg = "Failed to find top level definition for input binding {}."
+            raise JobException(
+                message=msg.format(input),
+                no_personal_data_message=msg.format("[input]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        input_data = pipeline_job_inputs[_input_name]
+        input_type = type(input_data)
+        if input_type in cls._PYTHON_SDK_TYPE_MAPPING:
+            return cls._PYTHON_SDK_TYPE_MAPPING[input_type], None
+        return getattr(input_data, "type", AssetTypes.URI_FOLDER), getattr(input_data, "mode", None)
+
+    @classmethod
+    def _find_source_from_parent_outputs(cls, input: str, pipeline_job_outputs: dict) -> Tuple[str, Optional[str]]:
+        """Find source type and mode of input/output from parent output.
+
+        :param input: The input name
+        :type input: str
+        :param pipeline_job_outputs: The pipeline job outputs
+        :type pipeline_job_outputs: dict
+        :return: A 2-tuple of the type and the mode
+        :rtype: Tuple[str, Optional[str]]
+        """
+        _output_name = input.split(".")[2][:-2]
+        if _output_name not in pipeline_job_outputs.keys():
+            msg = "Failed to find top level definition for output binding {}."
+            raise JobException(
+                message=msg.format(input),
+                no_personal_data_message=msg.format("[input]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        output_data = pipeline_job_outputs[_output_name]
+        output_type = type(output_data)
+        if output_type in cls._PYTHON_SDK_TYPE_MAPPING:
+            return cls._PYTHON_SDK_TYPE_MAPPING[output_type], None
+        if isinstance(output_data, dict):
+            if "type" in output_data:
+                output_data_type = output_data["type"]
+            else:
+                output_data_type = AssetTypes.URI_FOLDER
+            if "mode" in output_data:
+                output_data_mode = output_data["mode"]
+            else:
+                output_data_mode = None
+            return output_data_type, output_data_mode
+        return getattr(output_data, "type", AssetTypes.URI_FOLDER), getattr(output_data, "mode", None)
+
+    @classmethod
+    def _find_source_from_other_jobs(
+        cls, input: str, jobs_dict: dict, pipeline_job_dict: dict
+    ) -> Tuple[str, Optional[str]]:
+        """Find source type and mode of input/output from other job.
+
+        :param input: The input name
+        :type input: str
+        :param jobs_dict: The job dict
+        :type jobs_dict:
+        :param pipeline_job_dict: The pipeline job dict
+        :type pipeline_job_dict: dict
+        :return: A 2-tuple of the type and the mode
+        :rtype: Tuple[str, Optional[str]]
+        """
+        from azure.ai.ml.entities import CommandJob
+        from azure.ai.ml.entities._builders import BaseNode
+        from azure.ai.ml.entities._job.automl.automl_job import AutoMLJob
+        from azure.ai.ml.parallel import ParallelJob
+
+        _input_regex = r"\${{parent.jobs.([^.]+).([^.]+).([^.]+)}}"
+        m = re.match(_input_regex, input)
+        if m is None:
+            msg = "Failed to find top level definition for job binding {}."
+            raise JobException(
+                message=msg.format(input),
+                no_personal_data_message=msg.format("[input]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        _input_job_name, _io_type, _name = m.groups()
+        _input_job = jobs_dict[_input_job_name]
+
+        # we only support input of one job is from output of another output, but input mode should be decoupled with
+        # output mode, so we always return None source_mode
+        source_mode = None
+        if isinstance(_input_job, BaseNode):
+            # If source is base node, get type from io builder
+            _source = _input_job[_io_type][_name]
+            try:
+                source_type = _source.type
+                # Todo: get component type for registered component, and no need following codes
+                # source_type is None means _input_job's component is registered component which results in its
+                # input/output type is None.
+                if source_type is None:
+                    if _source._data is None:
+                        # return default type if _input_job's output data is None
+                        source_type = AssetTypes.URI_FOLDER
+                    elif isinstance(_source._data, Output):
+                        # if _input_job data is a Output object and we return its type.
+                        source_type = _source._data.type
+                    else:
+                        #  otherwise _input_job's input/output is bound to pipeline input/output, we continue
+                        #  infer the type according to _source._data. Will return corresponding pipeline
+                        #  input/output type because we didn't get the component.
+                        source_type, _ = cls._find_source_input_output_type(_source._data, pipeline_job_dict)
+                return source_type, source_mode
+            except AttributeError as e:
+                msg = "Failed to get referenced component type {}."
+                raise JobException(
+                    message=msg.format(_input_regex),
+                    no_personal_data_message=msg.format("[_input_regex]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                ) from e
+        if isinstance(_input_job, (CommandJob, ParallelJob)):
+            # If source has not parsed to Command yet, infer type
+            _source = get(_input_job, f"{_io_type}.{_name}")
+            if isinstance(_source, str):
+                source_type, _ = cls._find_source_input_output_type(_source, pipeline_job_dict)
+                return source_type, source_mode
+            return getattr(_source, "type", AssetTypes.URI_FOLDER), source_mode
+        if isinstance(_input_job, AutoMLJob):
+            # If source is AutoMLJob, only outputs is supported
+            if _io_type != "outputs":
+                msg = f"Only binding to AutoMLJob output is supported, currently got {_io_type}"
+                raise JobException(
+                    message=msg,
+                    no_personal_data_message=msg,
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                )
+            # AutoMLJob's output type can only be MLTABLE
+            return AssetTypes.MLTABLE, source_mode
+        msg = f"Unknown referenced source job type: {type(_input_job)}."
+        raise JobException(
+            message=msg,
+            no_personal_data_message=msg,
+            target=ErrorTarget.PIPELINE,
+            error_category=ErrorCategory.USER_ERROR,
+        )
+
+    @classmethod
+    def _find_source_input_output_type(cls, input: str, pipeline_job_dict: dict) -> Tuple[str, Optional[str]]:
+        """Find source type and mode of input/output.
+
+        :param input: The input binding
+        :type input: str
+        :param pipeline_job_dict: The pipeline job dict
+        :type pipeline_job_dict: dict
+        :return: A 2-tuple of the type and the mode
+        :rtype: Tuple[str, Optional[str]]
+        """
+        pipeline_job_inputs = pipeline_job_dict.get("inputs", {})
+        pipeline_job_outputs = pipeline_job_dict.get("outputs", {})
+        jobs_dict = pipeline_job_dict.get("jobs", {})
+        if is_data_binding_expression(input, ["parent", "inputs"]):
+            return cls._find_source_from_parent_inputs(input, pipeline_job_inputs)
+        if is_data_binding_expression(input, ["parent", "outputs"]):
+            return cls._find_source_from_parent_outputs(input, pipeline_job_outputs)
+        if is_data_binding_expression(input, ["parent", "jobs"]):
+            try:
+                return cls._find_source_from_other_jobs(input, jobs_dict, pipeline_job_dict)
+            except JobException as e:
+                raise e
+            except Exception as e:
+                msg = "Failed to find referenced source for input binding {}"
+                raise JobException(
+                    message=msg.format(input),
+                    no_personal_data_message=msg.format("[input]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.SYSTEM_ERROR,
+                ) from e
+        else:
+            msg = "Job input in a pipeline can bind only to a job output or a pipeline input"
+            raise JobException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+
+    @classmethod
+    def _to_input(
+        cls,  # pylint: disable=unused-argument
+        input: Union[Input, str, bool, int, float],
+        pipeline_job_dict: Optional[dict] = None,
+        **kwargs: Any,
+    ) -> Input:
+        """Convert a single job input value to component input.
+
+        :param input: The input
+        :type input: Union[Input, str, bool, int, float]
+        :param pipeline_job_dict: The pipeline job dict
+        :type pipeline_job_dict: Optional[dict]
+        :return: The Component Input
+        :rtype: Input
+        """
+        pipeline_job_dict = pipeline_job_dict or {}
+        input_variable: Dict = {}
+
+        if isinstance(input, str) and bool(re.search(ComponentJobConstants.INPUT_PATTERN, input)):
+            # handle input bindings
+            input_variable["type"], input_variable["mode"] = cls._find_source_input_output_type(
+                input, pipeline_job_dict
+            )
+
+        elif isinstance(input, Input):
+            input_variable = input._to_dict()
+        elif isinstance(input, SweepDistribution):
+            if isinstance(input, Choice):
+                if input.values is not None:
+                    input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[type(input.values[0])]
+            elif isinstance(input, Randint):
+                input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[int]
+            else:
+                input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[float]
+
+            input_variable["optional"] = False
+        elif type(input) in cls._PYTHON_SDK_TYPE_MAPPING:
+            input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[type(input)]
+            input_variable["default"] = input
+        elif isinstance(input, PipelineInput):
+            # Infer input type from input data
+            input_variable = input._to_input()._to_dict()
+        else:
+            msg = "'{}' is not supported as component input, supported types are '{}'.".format(
+                type(input), cls._PYTHON_SDK_TYPE_MAPPING.keys()
+            )
+            raise JobException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        return Input(**input_variable)
+
+    @classmethod
+    def _to_input_builder_function(cls, input: Union[Dict, SweepDistribution, Input, str, bool, int, float]) -> Input:
+        input_variable = {}
+
+        if isinstance(input, Input):
+            input_variable = input._to_dict()
+        elif isinstance(input, SweepDistribution):
+            if isinstance(input, Choice):
+                if input.values is not None:
+                    input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[type(input.values[0])]
+            elif isinstance(input, Randint):
+                input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[int]
+            else:
+                input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[float]
+
+            input_variable["optional"] = False
+        else:
+            input_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[type(input)]
+            input_variable["default"] = input
+        return Input(**input_variable)
+
+    @classmethod
+    def _to_output(
+        cls,  # pylint: disable=unused-argument
+        output: Optional[Union[Output, Dict, str, bool, int, float]],
+        pipeline_job_dict: Optional[dict] = None,
+        **kwargs: Any,
+    ) -> Output:
+        """Translate output value to Output and infer component output type
+        from linked pipeline output, its original type or default type.
+
+        :param output: The output
+        :type output: Union[Output, str, bool, int, float]
+        :param pipeline_job_dict: The pipeline job dict
+        :type pipeline_job_dict: Optional[dict]
+        :return: The output object
+        :rtype: Output
+        """
+        pipeline_job_dict = pipeline_job_dict or {}
+        output_type = None
+        if not pipeline_job_dict or output is None:
+            try:
+                output_type = output.type  # type: ignore
+            except AttributeError:
+                # default to url_folder if failed to get type
+                output_type = AssetTypes.URI_FOLDER
+            output_variable = {"type": output_type}
+            return Output(**output_variable)
+        output_variable = {}
+
+        if isinstance(output, str) and bool(re.search(ComponentJobConstants.OUTPUT_PATTERN, output)):
+            # handle output bindings
+            output_variable["type"], output_variable["mode"] = cls._find_source_input_output_type(
+                output, pipeline_job_dict
+            )
+
+        elif isinstance(output, Output):
+            output_variable = output._to_dict()
+
+        elif isinstance(output, PipelineOutput):
+            output_variable = output._to_output()._to_dict()
+
+        elif type(output) in cls._PYTHON_SDK_TYPE_MAPPING:
+            output_variable["type"] = cls._PYTHON_SDK_TYPE_MAPPING[type(output)]
+            output_variable["default"] = output
+        else:
+            msg = "'{}' is not supported as component output, supported types are '{}'.".format(
+                type(output), cls._PYTHON_SDK_TYPE_MAPPING.keys()
+            )
+            raise JobException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        return Output(**output_variable)
+
+    def _to_inputs(self, inputs: Optional[Dict], **kwargs: Any) -> Dict:
+        """Translate inputs to Inputs.
+
+        :param inputs: mapping from input name to input object.
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        :return: mapping from input name to translated component input.
+        :rtype: Dict[str, Input]
+        """
+        pipeline_job_dict = kwargs.get("pipeline_job_dict", {})
+        translated_component_inputs = {}
+        if inputs is not None:
+            for io_name, io_value in inputs.items():
+                translated_component_inputs[io_name] = self._to_input(io_value, pipeline_job_dict)
+        return translated_component_inputs
+
+    def _to_outputs(self, outputs: Optional[Dict], **kwargs: Any) -> Dict:
+        """Translate outputs to Outputs.
+
+        :param outputs: mapping from output name to output object.
+        :type outputs: Dict[str, Output]
+        :return: mapping from output name to translated component output.
+        :rtype: Dict[str, Output]
+        """
+        # Translate outputs to Outputs.
+        pipeline_job_dict = kwargs.get("pipeline_job_dict", {})
+        translated_component_outputs = {}
+        if outputs is not None:
+            for output_name, output_value in outputs.items():
+                translated_component_outputs[output_name] = self._to_output(output_value, pipeline_job_dict)
+        return translated_component_outputs
+
+    def _to_component(self, context: Optional[Dict] = None, **kwargs: Any) -> Union["Component", str]:
+        """Translate to Component.
+
+        :param context: The context
+        :type context: Optional[context]
+        :return: Translated Component.
+        :rtype: Component
+        """
+        # Note: Source of translated component should be same with Job
+        # And should be set after called _to_component/_to_node as job has no _source now.
+        raise NotImplementedError()
+
+    def _to_node(self, context: Optional[Dict] = None, **kwargs: Any) -> "BaseNode":
+        """Translate to pipeline node.
+
+        :param context: The context
+        :type context: Optional[context]
+        :return: Translated node.
+        :rtype: BaseNode
+        """
+        # Note: Source of translated component should be same with Job
+        # And should be set after called _to_component/_to_node as job has no _source now.
+        raise NotImplementedError()
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/__init__.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/__init__.py
new file mode 100644
index 00000000..3ccde947
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/__init__.py
@@ -0,0 +1,21 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+"""Classes in this package converts input & output set by user to pipeline job input & output."""
+
+from .attr_dict import OutputsAttrDict, _GroupAttrDict
+from .base import InputOutputBase, NodeInput, NodeOutput, PipelineInput, PipelineOutput
+from .mixin import AutoMLNodeIOMixin, NodeWithGroupInputMixin, PipelineJobIOMixin
+
+__all__ = [
+    "PipelineOutput",
+    "PipelineInput",
+    "NodeOutput",
+    "NodeInput",
+    "InputOutputBase",
+    "OutputsAttrDict",
+    "_GroupAttrDict",
+    "NodeWithGroupInputMixin",
+    "AutoMLNodeIOMixin",
+    "PipelineJobIOMixin",
+]
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/attr_dict.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/attr_dict.py
new file mode 100644
index 00000000..0ae08bcd
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/attr_dict.py
@@ -0,0 +1,170 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+from enum import Enum
+from typing import Any, Dict, List, Optional, Union
+
+from azure.ai.ml.entities._assets import Data
+from azure.ai.ml.entities._inputs_outputs import GroupInput, Input, Output
+from azure.ai.ml.entities._job.pipeline._attr_dict import K
+from azure.ai.ml.entities._job.pipeline._io.base import NodeInput, NodeOutput, PipelineInput
+from azure.ai.ml.exceptions import (
+    ErrorCategory,
+    ErrorTarget,
+    UnexpectedAttributeError,
+    UnexpectedKeywordError,
+    ValidationException,
+)
+
+
+class InputsAttrDict(dict):
+    def __init__(self, inputs: dict, **kwargs: Any):
+        self._validate_inputs(inputs)
+        super(InputsAttrDict, self).__init__(**inputs, **kwargs)
+
+    @classmethod
+    def _validate_inputs(cls, inputs: Any) -> None:
+        msg = "Pipeline/component input should be a \
+        azure.ai.ml.entities._job.pipeline._io.NodeInput with owner, got {}."
+        for val in inputs.values():
+            if isinstance(val, NodeInput) and val._owner is not None:  # pylint: disable=protected-access
+                continue
+            if isinstance(val, _GroupAttrDict):
+                continue
+            raise ValidationException(
+                message=msg.format(val),
+                no_personal_data_message=msg.format("[val]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+
+    def __setattr__(
+        self,
+        key: str,
+        value: Union[int, bool, float, str, NodeOutput, PipelineInput, Input],
+    ) -> None:
+        # Extract enum value.
+        value = value.value if isinstance(value, Enum) else value
+        original_input = self.__getattr__(key)  # Note that an exception will be raised if the keyword is invalid.
+        if isinstance(original_input, _GroupAttrDict) or isinstance(value, _GroupAttrDict):
+            # Set the value directly if is parameter group.
+            self._set_group_with_type_check(key, GroupInput.custom_class_value_to_attr_dict(value))
+            return
+        original_input._data = original_input._build_data(value)
+
+    def _set_group_with_type_check(self, key: Any, value: Any) -> None:
+        msg = "{!r} is expected to be a parameter group, but got {}."
+        if not isinstance(value, _GroupAttrDict):
+            raise ValidationException(
+                message=msg.format(key, type(value)),
+                no_personal_data_message=msg.format("[key]", "[value_type]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        self.__setitem__(key, GroupInput.custom_class_value_to_attr_dict(value))
+
+    def __getattr__(self, item: Any) -> NodeInput:
+        res: NodeInput = self.__getitem__(item)
+        return res
+
+
+class _GroupAttrDict(InputsAttrDict):
+    """This class is used for accessing values with instance.some_key."""
+
+    @classmethod
+    def _validate_inputs(cls, inputs: Any) -> None:
+        msg = "Pipeline/component input should be a azure.ai.ml.entities._job.pipeline._io.NodeInput, got {}."
+        for val in inputs.values():
+            if isinstance(val, NodeInput) and val._owner is not None:  # pylint: disable=protected-access
+                continue
+            if isinstance(val, _GroupAttrDict):
+                continue
+            # Allow PipelineInput as Group may appear at top level pipeline input.
+            if isinstance(val, PipelineInput):
+                continue
+            raise ValidationException(
+                message=msg.format(val),
+                no_personal_data_message=msg.format("[val]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+
+    def __getattr__(self, name: K) -> Any:
+        if name not in self:
+            raise UnexpectedAttributeError(keyword=name, keywords=list(self))
+        return super().__getitem__(name)
+
+    def __getitem__(self, item: K) -> Any:
+        # We raise this exception instead of KeyError
+        if item not in self:
+            raise UnexpectedKeywordError(func_name="ParameterGroup", keyword=item, keywords=list(self))
+        return super().__getitem__(item)
+
+    # For Jupyter Notebook auto-completion
+    def __dir__(self) -> List:
+        return list(super().__dir__()) + list(self.keys())
+
+    def flatten(self, group_parameter_name: Optional[str]) -> Dict:
+        # Return the flattened result of self
+
+        group_parameter_name = group_parameter_name if group_parameter_name else ""
+        flattened_parameters = {}
+        msg = "'%s' in parameter group should be a azure.ai.ml.entities._job._io.NodeInput, got '%s'."
+        for k, v in self.items():
+            flattened_name = ".".join([group_parameter_name, k])
+            if isinstance(v, _GroupAttrDict):
+                flattened_parameters.update(v.flatten(flattened_name))
+            elif isinstance(v, NodeInput):
+                flattened_parameters[flattened_name] = v._to_job_input()  # pylint: disable=protected-access
+            else:
+                raise ValidationException(
+                    message=msg % (flattened_name, type(v)),
+                    no_personal_data_message=msg % ("name", "type"),
+                    target=ErrorTarget.PIPELINE,
+                )
+        return flattened_parameters
+
+    def insert_group_name_for_items(self, group_name: Any) -> None:
+        # Insert one group name for all items.
+        for v in self.values():
+            if isinstance(v, _GroupAttrDict):
+                v.insert_group_name_for_items(group_name)
+            elif isinstance(v, PipelineInput):
+                # Insert group names for pipeline input
+                v._group_names = [group_name] + v._group_names  # pylint: disable=protected-access
+
+
+class OutputsAttrDict(dict):
+    def __init__(self, outputs: dict, **kwargs: Any):
+        for val in outputs.values():
+            if not isinstance(val, NodeOutput) or val._owner is None:
+                msg = "Pipeline/component output should be a azure.ai.ml.dsl.Output with owner, got {}."
+                raise ValidationException(
+                    message=msg.format(val),
+                    no_personal_data_message=msg.format("[val]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                )
+        super(OutputsAttrDict, self).__init__(**outputs, **kwargs)
+
+    def __getattr__(self, item: Any) -> NodeOutput:
+        return self.__getitem__(item)
+
+    def __getitem__(self, item: Any) -> NodeOutput:
+        if item not in self:
+            # We raise this exception instead of KeyError as OutputsAttrDict doesn't support add new item after
+            # __init__.
+            raise UnexpectedAttributeError(keyword=item, keywords=list(self))
+        res: NodeOutput = super().__getitem__(item)
+        return res
+
+    def __setattr__(self, key: str, value: Union[Data, Output]) -> None:
+        if isinstance(value, Output):
+            mode = value.mode
+            value = Output(type=value.type, path=value.path, mode=mode, name=value.name, version=value.version)
+        original_output = self.__getattr__(key)  # Note that an exception will be raised if the keyword is invalid.
+        original_output._data = original_output._build_data(value)
+
+    def __setitem__(self, key: str, value: Output) -> None:
+        return self.__setattr__(key, value)
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/base.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/base.py
new file mode 100644
index 00000000..b17972ae
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/base.py
@@ -0,0 +1,848 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+
+import copy
+import re
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, TypeVar, Union, cast, overload
+
+from azure.ai.ml._utils.utils import is_data_binding_expression
+from azure.ai.ml.constants import AssetTypes
+from azure.ai.ml.constants._component import IOConstants
+from azure.ai.ml.entities._assets._artifacts.data import Data
+from azure.ai.ml.entities._assets._artifacts.model import Model
+from azure.ai.ml.entities._inputs_outputs import Input, Output
+from azure.ai.ml.entities._job.pipeline._pipeline_expression import PipelineExpressionMixin
+from azure.ai.ml.entities._util import resolve_pipeline_parameter
+from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, UserErrorException, ValidationException
+
+# avoid circular import error
+if TYPE_CHECKING:
+    from azure.ai.ml.entities import PipelineJob
+    from azure.ai.ml.entities._builders import BaseNode
+
+T = TypeVar("T")
+
+
+def _build_data_binding(data: Union[str, "PipelineInput", "Output"]) -> Union[str, Output]:
+    """Build input builders to data bindings.
+
+    :param data: The data to build a data binding from
+    :type data: Union[str, PipelineInput, Output]
+    :return: A data binding string if data isn't a str, otherwise data
+    :rtype: str
+    """
+    result: Union[str, Output] = ""
+
+    if isinstance(data, (InputOutputBase)):
+        # Build data binding when data is PipelineInput, Output
+        result = data._data_binding()
+    else:
+        # Otherwise just return the data
+        result = data
+    return result
+
+
+def _resolve_builders_2_data_bindings(
+    data: Union[list, dict, str, "PipelineInput", "Output"]
+) -> Union[dict, list, str, Output]:
+    """Traverse data and build input builders inside it to data bindings.
+
+    :param data: The bindings to resolve
+    :type data: Union[list, dict, str, "PipelineInput", "Output"]
+    :return:
+       * A dict if data was a dict
+       * A list if data was a list
+       * A str otherwise
+    :rtype: Union[list, dict, str]
+    """
+    if isinstance(data, dict):
+        for key, val in data.items():
+            if isinstance(val, (dict, list)):
+                data[key] = _resolve_builders_2_data_bindings(val)
+            else:
+                data[key] = _build_data_binding(val)
+        return data
+    if isinstance(data, list):
+        resolved_data = []
+        for val in data:
+            resolved_data.append(_resolve_builders_2_data_bindings(val))
+        return resolved_data
+    return _build_data_binding(data)
+
+
+def _data_to_input(data: Union[Data, Model]) -> Input:
+    """Convert a Data object to an Input object.
+
+    :param data: The data to convert
+    :type data: Data
+    :return: The Input object
+    :rtype: Input
+    """
+    if data.id:
+        return Input(type=data.type, path=data.id)
+    return Input(type=data.type, path=f"{data.name}:{data.version}")
+
+
+class InputOutputBase(ABC):
+    # TODO: refine this code, always use _data to store builder level settings and use _meta to store definition
+    # TODO: when _data missing, return value from _meta
+
+    def __init__(
+        self,
+        meta: Optional[Union[Input, Output]],
+        data: Optional[Union[int, bool, float, str, Input, Output, "PipelineInput"]],
+        default_data: Optional[Union[int, bool, float, str, Input, Output]] = None,
+        **kwargs: Any,
+    ):
+        """Base class of input & output.
+
+        :param meta: Metadata of this input/output, eg: type, min, max, etc.
+        :type meta: Union[Input, Output]
+        :param data: Actual value of input/output, None means un-configured data.
+        :type data: Union[None, int, bool, float, str,
+                          azure.ai.ml.Input,
+                          azure.ai.ml.Output]
+        :param default_data: default value of input/output, None means un-configured data.
+        :type default_data: Union[None, int, bool, float, str,
+                          azure.ai.ml.Input,
+                          azure.ai.ml.Output]
+        """
+        self._meta = meta
+        self._original_data = data
+        self._data: Any = self._build_data(data)
+        self._default_data = default_data
+        self._type: str = meta.type if meta is not None else kwargs.pop("type", None)
+        self._mode = self._get_mode(original_data=data, data=self._data, kwargs=kwargs)
+        self._description = (
+            self._data.description
+            if self._data is not None and hasattr(self._data, "description") and self._data.description
+            else kwargs.pop("description", None)
+        )
+        # TODO: remove this
+        self._attribute_map: Dict = {}
+        self._name: Optional[str] = ""
+        self._version: Optional[str] = ""
+        super(InputOutputBase, self).__init__(**kwargs)
+
+    @abstractmethod
+    def _build_data(self, data: T) -> Union[T, str, Input, "InputOutputBase"]:
+        """Validate if data matches type and translate it to Input/Output acceptable type.
+
+        :param data: The data
+        :type data: T
+        :return: The built data
+        :rtype: Union[T, str, Input, InputOutputBase]
+        """
+
+    @abstractmethod
+    def _build_default_data(self) -> None:
+        """Build default data when data not configured."""
+
+    @property
+    def type(self) -> str:
+        """Type of input/output.
+
+        :return: The type
+        :rtype: str
+        """
+        return self._type
+
+    @type.setter
+    def type(self, type: Any) -> None:  # pylint: disable=redefined-builtin
+        # For un-configured input/output, we build a default data entry for them.
+        self._build_default_data()
+        self._type = type
+        if isinstance(self._data, (Input, Output)):
+            self._data.type = type
+        elif self._data is not None and not isinstance(
+            self._data, (int, float, str)
+        ):  # when type of self._data is InputOutputBase or its child class
+            self._data._type = type
+
+    @property
+    def mode(self) -> Optional[str]:
+        return self._mode
+
+    @mode.setter
+    def mode(self, mode: Optional[str]) -> None:
+        # For un-configured input/output, we build a default data entry for them.
+        self._build_default_data()
+        self._mode = mode
+        if isinstance(self._data, (Input, Output)):
+            self._data.mode = mode
+        elif self._data is not None and not isinstance(self._data, (int, float, str)):
+            self._data._mode = mode
+
+    @property
+    def description(self) -> Any:
+        return self._description
+
+    @description.setter
+    def description(self, description: str) -> None:
+        # For un-configured input/output, we build a default data entry for them.
+        self._build_default_data()
+        self._description = description
+        if isinstance(self._data, (Input, Output)):
+            self._data.description = description
+        elif self._data is not None and not isinstance(self._data, (int, float, str)):
+            self._data._description = description
+
+    @property
+    def path(self) -> Optional[str]:
+        # This property is introduced for static intellisense.
+        if hasattr(self._data, "path"):
+            if self._data is not None and not isinstance(self._data, (int, float, str)):
+                res: Optional[str] = self._data.path
+                return res
+        msg = f"{type(self._data)} does not have path."
+        raise ValidationException(
+            message=msg,
+            no_personal_data_message=msg,
+            target=ErrorTarget.PIPELINE,
+            error_category=ErrorCategory.USER_ERROR,
+        )
+
+    @path.setter
+    def path(self, path: str) -> None:
+        # For un-configured input/output, we build a default data entry for them.
+        self._build_default_data()
+        if hasattr(self._data, "path"):
+            if self._data is not None and not isinstance(self._data, (int, float, str)):
+                self._data.path = path
+        else:
+            msg = f"{type(self._data)} does not support setting path."
+            raise ValidationException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+
+    def _data_binding(self) -> str:
+        """Return data binding string representation for this input/output.
+
+        :return: The data binding string
+        :rtype: str
+        """
+        raise NotImplementedError()
+
+    # Why did we have this function? It prevents the DictMixin from being applied.
+    # Unclear if we explicitly do NOT want the mapping protocol to be applied to this, or it this was just
+    # confirmation that it didn't at the time.
+    def keys(self) -> None:
+        # This property is introduced to raise catchable Exception in marshmallow mapping validation trial.
+        raise TypeError(f"'{type(self).__name__}' object is not a mapping")
+
+    def __str__(self) -> str:
+        try:
+            return self._data_binding()
+        except AttributeError:
+            return super(InputOutputBase, self).__str__()
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    @classmethod
+    def _get_mode(
+        cls,
+        original_data: Optional[Union[int, bool, float, str, Input, Output, "PipelineInput"]],
+        data: Optional[Union[int, bool, float, str, Input, Output]],
+        kwargs: dict,
+    ) -> Optional[str]:
+        """Get mode of this input/output builder.
+
+        :param original_data: Original value of input/output.
+        :type original_data: Union[None, int, bool, float, str
+                          azure.ai.ml.Input,
+                          azure.ai.ml.Output,
+                          azure.ai.ml.entities._job.pipeline._io.PipelineInput]
+        :param data: Built input/output data.
+        :type data: Union[None, int, bool, float, str
+                          azure.ai.ml.Input,
+                          azure.ai.ml.Output]
+        :param kwargs: The kwargs
+        :type kwargs: Dict
+        :return: The mode
+        :rtype: Optional[str]
+        """
+        # pipeline level inputs won't pass mode to bound node level inputs
+        if isinstance(original_data, PipelineInput):
+            return None
+        return data.mode if data is not None and hasattr(data, "mode") else kwargs.pop("mode", None)
+
+    @property
+    def _is_primitive_type(self) -> bool:
+        return self.type in IOConstants.PRIMITIVE_STR_2_TYPE
+
+
+class NodeInput(InputOutputBase):
+    """Define one input of a Component."""
+
+    def __init__(
+        self,
+        port_name: str,
+        meta: Optional[Input],
+        *,
+        data: Optional[Union[int, bool, float, str, Output, "PipelineInput", Input]] = None,
+        # TODO: Bug Item number: 2883405
+        owner: Optional[Union["BaseComponent", "PipelineJob"]] = None,  # type: ignore
+        **kwargs: Any,
+    ):
+        """Initialize an input of a component.
+
+        :param name: The name of the input.
+        :type name: str
+        :param meta: Metadata of this input, eg: type, min, max, etc.
+        :type meta: Input
+        :param data: The input data. Valid types include int, bool, float, str,
+            Output of another component or pipeline input and Input.
+            Note that the output of another component or pipeline input associated should be reachable in the scope
+            of current pipeline. Input is introduced to support case like
+            TODO: new examples
+            component.inputs.xxx = Input(path="arm_id")
+        :type data: Union[int, bool, float, str
+                          azure.ai.ml.Output,
+                          azure.ai.ml.Input]
+        :param owner: The owner component of the input, used to calculate binding.
+        :type owner: Union[azure.ai.ml.entities.BaseNode, azure.ai.ml.entities.PipelineJob]
+        :param kwargs: A dictionary of additional configuration parameters.
+        :type kwargs: dict
+        """
+        # TODO: validate data matches type in meta
+        # TODO: validate supported data
+        self._port_name = port_name
+        self._owner = owner
+        super().__init__(meta=meta, data=data, **kwargs)
+
+    def _build_default_data(self) -> None:
+        """Build default data when input not configured."""
+        if self._data is None:
+            self._data = Input()
+
+    def _build_data(self, data: T) -> Union[T, str, Input, InputOutputBase]:
+        """Build input data according to assigned input
+
+        eg: node.inputs.key = data
+
+        :param data: The data
+        :type data: T
+        :return: The built data
+        :rtype: Union[T, str, Input, "PipelineInput", "NodeOutput"]
+        """
+        _data: Union[T, str, NodeOutput] = resolve_pipeline_parameter(data)
+        if _data is None:
+            return _data
+        # Unidiomatic typecheck: Checks that data is _exactly_ this type, and not potentially a subtype
+        if type(_data) is NodeInput:  # pylint: disable=unidiomatic-typecheck
+            msg = "Can not bind input to another component's input."
+            raise ValidationException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        if isinstance(_data, (PipelineInput, NodeOutput)):
+            # If value is input or output, it's a data binding, we require it have a owner so we can convert it to
+            # a data binding, eg: ${{inputs.xxx}}
+            if isinstance(_data, NodeOutput) and _data._owner is None:
+                msg = "Setting input binding {} to output without owner is not allowed."
+                raise ValidationException(
+                    message=msg.format(_data),
+                    no_personal_data_message=msg.format("[_data]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                )
+            return _data
+        # for data binding case, set is_singular=False for case like "${{parent.inputs.job_in_folder}}/sample1.csv"
+        if isinstance(_data, Input) or is_data_binding_expression(_data, is_singular=False):
+            return _data
+        if isinstance(_data, (Data, Model)):
+            return _data_to_input(_data)
+        # self._meta.type could be None when sub pipeline has no annotation
+        if isinstance(self._meta, Input) and self._meta.type and not self._meta._is_primitive_type:
+            if isinstance(_data, str):
+                return Input(type=self._meta.type, path=_data)
+            msg = "only path input is supported now but get {}: {}."
+            raise UserErrorException(
+                message=msg.format(type(_data), _data),
+                no_personal_data_message=msg.format(type(_data), "[_data]"),
+            )
+        return _data
+
+    def _to_job_input(self) -> Optional[Union[Input, str, Output]]:
+        """convert the input to Input, this logic will change if backend contract changes."""
+        result: Optional[Union[Input, str, Output]] = None
+
+        if self._data is None:
+            # None data means this input is not configured.
+            result = None
+        elif isinstance(self._data, (PipelineInput, NodeOutput)):
+            # Build data binding when data is PipelineInput, Output
+            result = Input(path=self._data._data_binding(), mode=self.mode)
+        elif is_data_binding_expression(self._data):
+            result = Input(path=self._data, mode=self.mode)
+        else:
+            data_binding = _build_data_binding(self._data)
+            if is_data_binding_expression(self._data):
+                result = Input(path=data_binding, mode=self.mode)
+            else:
+                result = data_binding
+            # TODO: validate is self._data is supported
+
+        return result
+
+    def _data_binding(self) -> str:
+        msg = "Input binding {} can only come from a pipeline, currently got {}"
+        # call type(self._owner) to avoid circular import
+        raise ValidationException(
+            message=msg.format(self._port_name, type(self._owner)),
+            target=ErrorTarget.PIPELINE,
+            no_personal_data_message=msg.format("[port_name]", "[owner]"),
+            error_category=ErrorCategory.USER_ERROR,
+        )
+
+    def _copy(self, owner: Any) -> "NodeInput":
+        return NodeInput(
+            port_name=self._port_name,
+            data=self._data,
+            owner=owner,
+            meta=cast(Input, self._meta),
+        )
+
+    def _deepcopy(self) -> "NodeInput":
+        return NodeInput(
+            port_name=self._port_name,
+            data=copy.copy(self._data),
+            owner=self._owner,
+            meta=cast(Input, self._meta),
+        )
+
+    def _get_data_owner(self) -> Optional["BaseNode"]:
+        """Gets the data owner of the node
+
+        Note: This only works for @pipeline, not for YAML pipeline.
+
+        Note: Inner step will be returned as the owner when node's input is from sub pipeline's output.
+            @pipeline
+            def sub_pipeline():
+                inner_node = component_func()
+                return inner_node.outputs
+
+            @pipeline
+            def root_pipeline():
+                pipeline_node = sub_pipeline()
+                node = copy_files_component_func(input_dir=pipeline_node.outputs.output_dir)
+                owner = node.inputs.input_dir._get_data_owner()
+                assert owner == pipeline_node.nodes[0]
+
+        :return: The node if Input is from another node's output. Returns None for literal value.
+        :rtype: Optional[BaseNode]
+        """
+        from azure.ai.ml.entities import Pipeline
+        from azure.ai.ml.entities._builders import BaseNode
+
+        def _resolve_data_owner(data: Any) -> Optional["BaseNode"]:
+            if isinstance(data, BaseNode) and not isinstance(data, Pipeline):
+                return data
+            while isinstance(data, PipelineInput):
+                # for pipeline input, it's original value(can be literal value or another node's output)
+                # is stored in _original_data
+                return _resolve_data_owner(data._original_data)
+            if isinstance(data, NodeOutput):
+                if isinstance(data._owner, Pipeline):
+                    # for input from subgraph's output, trace back to inner node
+                    return _resolve_data_owner(data._binding_output)
+                # for input from another node's output, return the node
+                return _resolve_data_owner(data._owner)
+            return None
+
+        return _resolve_data_owner(self._data)
+
+
+class NodeOutput(InputOutputBase, PipelineExpressionMixin):
+    """Define one output of a Component."""
+
+    def __init__(
+        self,
+        port_name: str,
+        meta: Optional[Union[Input, Output]],
+        *,
+        data: Optional[Union[Output, str]] = None,
+        # TODO: Bug Item number: 2883405
+        owner: Optional[Union["BaseComponent", "PipelineJob"]] = None,  # type: ignore
+        binding_output: Optional["NodeOutput"] = None,
+        **kwargs: Any,
+    ):
+        """Initialize an Output of a component.
+
+        :param port_name: The port_name of the output.
+        :type port_name: str
+        :param name: The name used to register NodeOutput/PipelineOutput data.
+        :type name: str
+        :param version: The version used to register NodeOutput/PipelineOutput data.
+        :ype version: str
+        :param data: The output data. Valid types include str, Output
+        :type data: Union[str
+                          azure.ai.ml.entities.Output]
+        :param mode: The mode of the output.
+        :type mode: str
+        :param owner: The owner component of the output, used to calculate binding.
+        :type owner: Union[azure.ai.ml.entities.BaseNode, azure.ai.ml.entities.PipelineJob]
+        :param binding_output: The node output bound to pipeline output, only available for pipeline.
+        :type binding_output: azure.ai.ml.entities.NodeOutput
+        :param kwargs: A dictionary of additional configuration parameters.
+        :type kwargs: dict
+        :raises ~azure.ai.ml.exceptions.ValidationException: Raised if object cannot be successfully validated.
+            Details will be provided in the error message.
+        """
+        # Allow inline output binding with string, eg: "component_out_path_1": "${{parents.outputs.job_out_data_1}}"
+        if data is not None and not isinstance(data, (Output, str)):
+            msg = "Got unexpected type for output: {}."
+            raise ValidationException(
+                message=msg.format(data),
+                target=ErrorTarget.PIPELINE,
+                no_personal_data_message=msg.format("[data]"),
+            )
+        super().__init__(meta=meta, data=data, **kwargs)
+        self._port_name = port_name
+        self._owner = owner
+        self._name: Optional[str] = self._data.name if isinstance(self._data, Output) else None
+        self._version: Optional[str] = self._data.version if isinstance(self._data, Output) else None
+
+        self._assert_name_and_version()
+
+        # store original node output to be able to trace back to inner node from a pipeline output builder.
+        self._binding_output = binding_output
+
+    @property
+    def port_name(self) -> str:
+        """The output port name, eg: node.outputs.port_name.
+
+        :return: The port name
+        :rtype: str
+        """
+        return self._port_name
+
+    @property
+    def name(self) -> Optional[str]:
+        """Used in registering output data.
+
+        :return: The output name
+        :rtype: str
+        """
+        return self._name
+
+    @name.setter
+    def name(self, name: str) -> None:
+        """Assigns the name to NodeOutput/PipelineOutput and builds data according to the name.
+
+        :param name: The new name
+        :type name: str
+        """
+        self._build_default_data()
+        self._name = name
+        if isinstance(self._data, Output):
+            self._data.name = name
+        elif isinstance(self._data, InputOutputBase):
+            self._data._name = name
+        else:
+            raise UserErrorException(
+                f"We support self._data of Input, Output, InputOutputBase, NodeOutput and NodeInput,"
+                f"but got type: {type(self._data)}."
+            )
+
+    @property
+    def version(self) -> Optional[str]:
+        """Used in registering output data.
+
+        :return: The output data
+        :rtype: str
+        """
+        return self._version
+
+    @version.setter
+    def version(self, version: str) -> None:
+        """Assigns the version to NodeOutput/PipelineOutput and builds data according to the version.
+
+        :param version: The new version
+        :type version: str
+        """
+        self._build_default_data()
+        self._version = version
+        if isinstance(self._data, Output):
+            self._data.version = version
+        elif isinstance(self._data, InputOutputBase):
+            self._data._version = version
+        else:
+            raise UserErrorException(
+                f"We support self._data of Input, Output, InputOutputBase, NodeOutput and NodeInput,"
+                f"but got type: {type(self._data)}."
+            )
+
+    @property
+    def path(self) -> Any:
+        # For node output path,
+        if self._data is not None and hasattr(self._data, "path"):
+            return self._data.path
+        return None
+
+    @path.setter
+    def path(self, path: Optional[str]) -> None:
+        # For un-configured output, we build a default data entry for them.
+        self._build_default_data()
+        if self._data is not None and hasattr(self._data, "path"):
+            self._data.path = path
+        else:
+            # YAML job will have string output binding and do not support setting path for it.
+            msg = f"{type(self._data)} does not support setting path."
+            raise ValidationException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+
+    def _assert_name_and_version(self) -> None:
+        if self.name and not (re.match("^[A-Za-z0-9_-]*$", self.name) and len(self.name) <= 255):
+            raise UserErrorException(
+                f"The output name {self.name} can only contain alphanumeric characters, dashes and underscores, "
+                f"with a limit of 255 characters."
+            )
+        if self.version and not self.name:
+            raise UserErrorException("Output name is required when output version is specified.")
+
+    def _build_default_data(self) -> None:
+        """Build default data when output not configured."""
+        if self._data is None:
+            # _meta will be None when node._component is not a Component object
+            # so we just leave the type inference work to backend
+            self._data = Output(type=None)  # type: ignore[call-overload]
+
+    def _build_data(self, data: T) -> Any:
+        """Build output data according to assigned input, eg: node.outputs.key = data
+
+        :param data: The data
+        :type data: T
+        :return: `data`
+        :rtype: T
+        """
+        if data is None:
+            return data
+        if not isinstance(data, (Output, str)):
+            msg = f"{self.__class__.__name__} only allow set {Output.__name__} object, {type(data)} is not supported."
+            raise ValidationException(
+                message=msg,
+                target=ErrorTarget.PIPELINE,
+                no_personal_data_message=msg,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        res: T = cast(T, data)
+        return res
+
+    def _to_job_output(self) -> Optional[Output]:
+        """Convert the output to Output, this logic will change if backend contract changes."""
+        if self._data is None:
+            # None data means this output is not configured.
+            result = None
+        elif isinstance(self._data, str):
+            result = Output(
+                type=AssetTypes.URI_FOLDER, path=self._data, mode=self.mode, name=self.name, version=self.version
+            )
+        elif isinstance(self._data, Output):
+            result = self._data
+        elif isinstance(self._data, PipelineOutput):
+            result = Output(
+                type=AssetTypes.URI_FOLDER,
+                path=self._data._data_binding(),
+                mode=self.mode,
+                name=self._data.name,
+                version=self._data.version,
+                description=self.description,
+            )
+        else:
+            msg = "Got unexpected type for output: {}."
+            raise ValidationException(
+                message=msg.format(self._data),
+                target=ErrorTarget.PIPELINE,
+                no_personal_data_message=msg.format("[data]"),
+            )
+        return result
+
+    def _data_binding(self) -> str:
+        if self._owner is not None:
+            return f"${{{{parent.jobs.{self._owner.name}.outputs.{self._port_name}}}}}"
+
+        return ""
+
+    def _copy(self, owner: Any) -> "NodeOutput":
+        return NodeOutput(
+            port_name=self._port_name,
+            data=cast(Output, self._data),
+            owner=owner,
+            meta=self._meta,
+        )
+
+    def _deepcopy(self) -> "NodeOutput":
+        return NodeOutput(
+            port_name=self._port_name,
+            data=cast(Output, copy.copy(self._data)),
+            owner=self._owner,
+            meta=self._meta,
+            binding_output=self._binding_output,
+        )
+
+
+class PipelineInput(NodeInput, PipelineExpressionMixin):
+    """Define one input of a Pipeline."""
+
+    def __init__(self, name: str, meta: Optional[Input], group_names: Optional[List[str]] = None, **kwargs: Any):
+        """Initialize a PipelineInput.
+
+        :param name: The name of the input.
+        :type name: str
+        :param meta: Metadata of this input, eg: type, min, max, etc.
+        :type meta: Input
+        :param group_names: The input parameter's group names.
+        :type group_names: List[str]
+        """
+        super(PipelineInput, self).__init__(port_name=name, meta=meta, **kwargs)
+        self._group_names = group_names if group_names else []
+
+    def result(self) -> Any:
+        """Return original value of pipeline input.
+
+        :return: The original value of pipeline input
+        :rtype: Any
+
+        Example:
+
+        .. code-block:: python
+
+           @pipeline
+           def pipeline_func(param1):
+             # node1's param1 will get actual value of param1 instead of a input binding.
+             node1 = component_func(param1=param1.result())
+        """
+
+        # use this to break self loop
+        original_data_cache: Set = set()
+        original_data = self._original_data
+        while isinstance(original_data, PipelineInput) and original_data not in original_data_cache:
+            original_data_cache.add(original_data)
+            original_data = original_data._original_data
+        return original_data
+
+    def __str__(self) -> str:
+        return self._data_binding()
+
+    @overload
+    def _build_data(self, data: Union[Model, Data]) -> Input: ...
+
+    @overload
+    def _build_data(self, data: T) -> Any: ...
+
+    def _build_data(self, data: Union[Model, Data, T]) -> Any:
+        """Build data according to input type.
+
+        :param data: The data
+        :type data: Union[Model, Data, T]
+        :return:
+            * Input if data is a Model or Data
+            * data otherwise
+        :rtype: Union[Input, T]
+        """
+        if data is None:
+            return data
+        # Unidiomatic typecheck: Checks that data is _exactly_ this type, and not potentially a subtype
+        if type(data) is NodeInput:  # pylint: disable=unidiomatic-typecheck
+            msg = "Can not bind input to another component's input."
+            raise ValidationException(message=msg, no_personal_data_message=msg, target=ErrorTarget.PIPELINE)
+        if isinstance(data, (PipelineInput, NodeOutput)):
+            # If value is input or output, it's a data binding, owner is required to convert it to
+            # a data binding, eg: ${{parent.inputs.xxx}}
+            if isinstance(data, NodeOutput) and data._owner is None:
+                msg = "Setting input binding {} to output without owner is not allowed."
+                raise ValidationException(
+                    message=msg.format(data),
+                    no_personal_data_message=msg.format("[data]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                )
+            return data
+        if isinstance(data, (Data, Model)):
+            # If value is Data, we convert it to an corresponding Input
+            return _data_to_input(data)
+        return data
+
+    def _data_binding(self) -> str:
+        full_name = "%s.%s" % (".".join(self._group_names), self._port_name) if self._group_names else self._port_name
+        return f"${{{{parent.inputs.{full_name}}}}}"
+
+    def _to_input(self) -> Optional[Union[Input, Output]]:
+        """Convert pipeline input to component input for pipeline component.
+
+        :return: The component input
+        :rtype: Input
+        """
+        if self._data is None:
+            # None data means this input is not configured.
+            return self._meta
+        data_type = self._data.type if isinstance(self._data, Input) else None
+        # If type is asset type, return data type without default.
+        # Else infer type from data and set it as default.
+        if data_type and data_type.lower() in AssetTypes.__dict__.values():
+            if not isinstance(self._data, (int, float, str)):
+                result = Input(type=data_type, mode=self._data.mode)
+        elif type(self._data) in IOConstants.PRIMITIVE_TYPE_2_STR:
+            result = Input(
+                type=IOConstants.PRIMITIVE_TYPE_2_STR[type(self._data)],
+                default=self._data,
+            )
+        else:
+            msg = f"Unsupported Input type {type(self._data)} detected when translate job to component."
+            raise ValidationException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        return result  # pylint: disable=possibly-used-before-assignment
+
+
+class PipelineOutput(NodeOutput):
+    """Define one output of a Pipeline."""
+
+    def _to_job_output(self) -> Optional[Output]:
+        result: Optional[Output] = None
+        if isinstance(self._data, Output):
+            # For pipeline output with type Output, always pass to backend.
+            result = self._data
+        elif self._data is None and self._meta and self._meta.type:
+            # For un-configured pipeline output with meta, we need to return Output with accurate type,
+            # so it won't default to uri_folder.
+            result = Output(type=self._meta.type, mode=self._meta.mode, description=self._meta.description)
+        else:
+            result = super(PipelineOutput, self)._to_job_output()
+        # Copy meta type to avoid built output's None type default to uri_folder.
+        if self.type and result is not None and not result.type:
+            result.type = self.type
+        return result
+
+    def _data_binding(self) -> str:
+        return f"${{{{parent.outputs.{self._port_name}}}}}"
+
+    def _to_output(self) -> Optional[Output]:
+        """Convert pipeline output to component output for pipeline component."""
+        if self._data is None:
+            # None data means this input is not configured.
+            return None
+        if isinstance(self._meta, Output):
+            return self._meta
+        # Assign type directly as we didn't have primitive output type for now.
+        if not isinstance(self._data, (int, float, str)):
+            return Output(type=self._data.type, mode=self._data.mode)
+        return Output()
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/mixin.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/mixin.py
new file mode 100644
index 00000000..6c3d9357
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_io/mixin.py
@@ -0,0 +1,623 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+import copy
+from typing import Any, Dict, List, Optional, Tuple, Type, Union
+
+from azure.ai.ml._restclient.v2023_04_01_preview.models import JobInput as RestJobInput
+from azure.ai.ml._restclient.v2023_04_01_preview.models import JobOutput as RestJobOutput
+from azure.ai.ml.constants._component import ComponentJobConstants
+from azure.ai.ml.entities._inputs_outputs import GroupInput, Input, Output
+from azure.ai.ml.entities._util import copy_output_setting
+from azure.ai.ml.exceptions import ErrorTarget, ValidationErrorType, ValidationException
+
+from ..._input_output_helpers import (
+    from_rest_data_outputs,
+    from_rest_inputs_to_dataset_literal,
+    to_rest_data_outputs,
+    to_rest_dataset_literal_inputs,
+)
+from .._pipeline_job_helpers import from_dict_to_rest_io, process_sdk_component_job_io
+from .attr_dict import InputsAttrDict, OutputsAttrDict, _GroupAttrDict
+from .base import NodeInput, NodeOutput, PipelineInput, PipelineOutput
+
+
+class NodeIOMixin:
+    """Provides ability to wrap node inputs/outputs and build data bindings
+    dynamically."""
+
+    @classmethod
+    def _get_supported_inputs_types(cls) -> Optional[Any]:
+        return None
+
+    @classmethod
+    def _get_supported_outputs_types(cls) -> Optional[Any]:
+        return None
+
+    @classmethod
+    def _validate_io(cls, value: Any, allowed_types: Optional[tuple], *, key: Optional[str] = None) -> None:
+        if allowed_types is None:
+            return
+
+        if value is None or isinstance(value, allowed_types):
+            pass
+        else:
+            msg = "Expecting {} for input/output {}, got {} instead."
+            raise ValidationException(
+                message=msg.format(allowed_types, key, type(value)),
+                no_personal_data_message=msg.format(allowed_types, "[key]", type(value)),
+                target=ErrorTarget.PIPELINE,
+                error_type=ValidationErrorType.INVALID_VALUE,
+            )
+
+    def _build_input(
+        self,
+        name: str,
+        meta: Optional[Input],
+        data: Optional[Union[dict, int, bool, float, str, Output, "PipelineInput", Input]],
+    ) -> NodeInput:
+        # output mode of last node should not affect input mode of next node
+        if isinstance(data, NodeOutput):
+            # Decoupled input and output
+            # value = copy.deepcopy(value)
+            data = data._deepcopy()  # pylint: disable=protected-access
+            data.mode = None
+        elif isinstance(data, dict):
+            # Use type comparison instead of is_instance to skip _GroupAttrDict
+            # when loading from yaml io will be a dict,
+            # like {'job_data_path': '${{parent.inputs.pipeline_job_data_path}}'}
+            # parse dict to allowed type
+            data = Input(**data)
+
+            # parameter group can be of custom type, so we don't check it here
+            if meta is not None and not isinstance(meta, GroupInput):
+                self._validate_io(data, self._get_supported_inputs_types(), key=name)
+        return NodeInput(port_name=name, meta=meta, data=data, owner=self)
+
+    def _build_output(self, name: str, meta: Optional[Output], data: Optional[Union[Output, str]]) -> NodeOutput:
+        if isinstance(data, dict):
+            data = Output(**data)
+
+        self._validate_io(data, self._get_supported_outputs_types(), key=name)
+        # For un-configured outputs, settings it to None, so we won't pass extra fields(eg: default mode)
+        return NodeOutput(port_name=name, meta=meta, data=data, owner=self)
+
+    # pylint: disable=unused-argument
+    def _get_default_input_val(self, val: Any):  # type: ignore
+        # use None value as data placeholder for unfilled inputs.
+        # server side will fill the default value
+        return None
+
+    def _build_inputs_dict(
+        self,
+        inputs: Dict[str, Union[Input, str, bool, int, float]],
+        *,
+        input_definition_dict: Optional[dict] = None,
+    ) -> InputsAttrDict:
+        """Build an input attribute dict so user can get/set inputs by
+        accessing attribute, eg: node1.inputs.xxx.
+
+        :param inputs: Provided kwargs when parameterizing component func.
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        :keyword input_definition_dict: Static input definition dict. If not provided, will build inputs without meta.
+        :paramtype input_definition_dict: dict
+        :return: Built dynamic input attribute dict.
+        :rtype: InputsAttrDict
+        """
+        if input_definition_dict is not None:
+            # TODO: validate inputs.keys() in input_definitions.keys()
+            input_dict = {}
+            for key, val in input_definition_dict.items():
+                if key in inputs.keys():
+                    # If input is set through component functions' kwargs, create an input object with real value.
+                    data = inputs[key]
+                else:
+                    data = self._get_default_input_val(val)  # pylint: disable=assignment-from-none
+
+                val = self._build_input(name=key, meta=val, data=data)
+                input_dict[key] = val
+        else:
+            input_dict = {key: self._build_input(name=key, meta=None, data=val) for key, val in inputs.items()}
+        return InputsAttrDict(input_dict)
+
+    def _build_outputs_dict(
+        self, outputs: Dict, *, output_definition_dict: Optional[dict] = None, none_data: bool = False
+    ) -> OutputsAttrDict:
+        """Build an output attribute dict so user can get/set outputs by
+        accessing attribute, eg: node1.outputs.xxx.
+
+        :param outputs: Provided kwargs when parameterizing component func.
+        :type outputs: Dict[str, Output]
+        :keyword output_definition_dict: Static output definition dict.
+        :paramtype output_definition_dict: Dict
+        :keyword none_data: If True, will set output data to None.
+        :paramtype none_data: bool
+        :return: Built dynamic output attribute dict.
+        :rtype: OutputsAttrDict
+        """
+        if output_definition_dict is not None:
+            # TODO: check if we need another way to mark a un-configured output instead of just set None.
+            # Create None as data placeholder for all outputs.
+            output_dict = {}
+            for key, val in output_definition_dict.items():
+                if key in outputs.keys():
+                    # If output has given value, create an output object with real value.
+                    val = self._build_output(name=key, meta=val, data=outputs[key])
+                else:
+                    val = self._build_output(name=key, meta=val, data=None)
+                output_dict[key] = val
+        else:
+            output_dict = {}
+            for key, val in outputs.items():
+                output_val = self._build_output(name=key, meta=None, data=val if not none_data else None)
+                output_dict[key] = output_val
+        return OutputsAttrDict(output_dict)
+
+    def _build_inputs(self) -> Dict:
+        """Build inputs of this component to a dict dict which maps output to
+        actual value.
+
+        The built input dict will have same input format as other jobs, eg:
+        {
+           "input_data": Input(path="path/to/input/data", mode="Mount"),
+           "input_value": 10,
+           "learning_rate": "${{jobs.step1.inputs.learning_rate}}"
+        }
+
+        :return: The input dict
+        :rtype: Dict[str, Union[Input, str, bool, int, float]]
+        """
+        inputs = {}
+        # pylint: disable=redefined-builtin
+        for name, input in self.inputs.items():  # type: ignore
+            if isinstance(input, _GroupAttrDict):
+                # Flatten group inputs into inputs dict
+                inputs.update(input.flatten(group_parameter_name=name))
+                continue
+            inputs[name] = input._to_job_input()  # pylint: disable=protected-access
+        return inputs
+
+    def _build_outputs(self) -> Dict[str, Output]:
+        """Build outputs of this component to a dict which maps output to
+        actual value.
+
+        The built output dict will have same output format as other jobs, eg:
+        {
+            "eval_output": "${{jobs.eval.outputs.eval_output}}"
+        }
+
+        :return: The output dict
+        :rtype: Dict[str, Output]
+        """
+        outputs = {}
+        for name, output in self.outputs.items():  # type: ignore
+            if isinstance(output, NodeOutput):
+                output = output._to_job_output()  # pylint: disable=protected-access
+            outputs[name] = output
+        # Remove non-configured output
+        return {k: v for k, v in outputs.items() if v is not None}
+
+    def _to_rest_inputs(self) -> Dict[str, Dict]:
+        """Translate input builders to rest input dicts.
+
+        The built dictionary's format aligns with component job's input yaml, eg:
+        {
+           "input_data": {"data": {"path": "path/to/input/data"},  "mode"="Mount"},
+           "input_value": 10,
+           "learning_rate": "${{jobs.step1.inputs.learning_rate}}"
+        }
+
+        :return: The REST inputs
+        :rtype: Dict[str, Dict]
+        """
+        built_inputs = self._build_inputs()
+        return self._input_entity_to_rest_inputs(input_entity=built_inputs)
+
+    @classmethod
+    def _input_entity_to_rest_inputs(cls, input_entity: Dict[str, Input]) -> Dict[str, Dict]:
+        # Convert io entity to rest io objects
+        input_bindings, dataset_literal_inputs = process_sdk_component_job_io(
+            input_entity, [ComponentJobConstants.INPUT_PATTERN]
+        )
+
+        # parse input_bindings to InputLiteral(value=str(binding))
+        rest_inputs = {**input_bindings, **dataset_literal_inputs}
+        # Note: The function will only be called from BaseNode,
+        # and job_type is used to enable dot in pipeline job input keys,
+        # so pass job_type as None directly here.
+        rest_inputs = to_rest_dataset_literal_inputs(rest_inputs, job_type=None)
+
+        # convert rest io to dict
+        rest_dataset_literal_inputs = {}
+        for name, val in rest_inputs.items():
+            rest_dataset_literal_inputs[name] = val.as_dict()
+            if hasattr(val, "mode") and val.mode:
+                rest_dataset_literal_inputs[name].update({"mode": val.mode.value})
+        return rest_dataset_literal_inputs
+
+    def _to_rest_outputs(self) -> Dict[str, Dict]:
+        """Translate output builders to rest output dicts.
+
+        The built dictionary's format aligns with component job's output yaml, eg:
+        {"eval_output": "${{jobs.eval.outputs.eval_output}}"}
+
+        :return: The REST outputs
+        :rtype: Dict[str, Dict]
+        """
+        built_outputs = self._build_outputs()
+
+        # Convert io entity to rest io objects
+        output_bindings, data_outputs = process_sdk_component_job_io(
+            built_outputs, [ComponentJobConstants.OUTPUT_PATTERN]
+        )
+        rest_data_outputs = to_rest_data_outputs(data_outputs)
+
+        # convert rest io to dict
+        # parse output_bindings to {"value": binding, "type": "literal"} since there's no mode for it
+        rest_output_bindings = {}
+        for key, binding in output_bindings.items():
+            rest_output_bindings[key] = {"value": binding["value"], "type": "literal"}
+            if "mode" in binding:
+                rest_output_bindings[key].update({"mode": binding["mode"].value})
+            if "name" in binding:
+                rest_output_bindings[key].update({"name": binding["name"]})
+            if "version" in binding:
+                rest_output_bindings[key].update({"version": binding["version"]})
+
+        def _rename_name_and_version(output_dict: Dict) -> Dict:
+            # NodeOutput can only be registered with name and version, therefore we rename here
+            if "asset_name" in output_dict.keys():
+                output_dict["name"] = output_dict.pop("asset_name")
+            if "asset_version" in output_dict.keys():
+                output_dict["version"] = output_dict.pop("asset_version")
+            return output_dict
+
+        rest_data_outputs = {name: _rename_name_and_version(val.as_dict()) for name, val in rest_data_outputs.items()}
+        self._update_output_types(rest_data_outputs)
+        rest_data_outputs.update(rest_output_bindings)
+        return rest_data_outputs
+
+    @classmethod
+    def _from_rest_inputs(cls, inputs: Dict) -> Dict[str, Union[Input, str, bool, int, float]]:
+        """Load inputs from rest inputs.
+
+        :param inputs: The REST inputs
+        :type inputs: Dict[str, Union[str, dict]]
+        :return: Input dict
+        :rtype: Dict[str, Union[Input, str, bool, int, float]]
+        """
+
+        # JObject -> RestJobInput/RestJobOutput
+        input_bindings, rest_inputs = from_dict_to_rest_io(inputs, RestJobInput, [ComponentJobConstants.INPUT_PATTERN])
+
+        # RestJobInput/RestJobOutput -> Input/Output
+        dataset_literal_inputs = from_rest_inputs_to_dataset_literal(rest_inputs)
+
+        return {**dataset_literal_inputs, **input_bindings}
+
+    @classmethod
+    def _from_rest_outputs(cls, outputs: Dict[str, Union[str, dict]]) -> Dict:
+        """Load outputs from rest outputs.
+
+        :param outputs: The REST outputs
+        :type outputs: Dict[str, Union[str, dict]]
+        :return: Output dict
+        :rtype: Dict[str, Output]
+        """
+
+        # JObject -> RestJobInput/RestJobOutput
+        output_bindings, rest_outputs = from_dict_to_rest_io(
+            outputs, RestJobOutput, [ComponentJobConstants.OUTPUT_PATTERN]
+        )
+
+        # RestJobInput/RestJobOutput -> Input/Output
+        data_outputs = from_rest_data_outputs(rest_outputs)
+
+        return {**data_outputs, **output_bindings}
+
+    def _update_output_types(self, rest_data_outputs: dict) -> None:
+        """Update output types in rest_data_outputs according to meta level output.
+
+        :param rest_data_outputs: The REST data outputs
+        :type rest_data_outputs: Dict
+        """
+
+        for name, rest_output in rest_data_outputs.items():
+            original_output = self.outputs[name]  # type: ignore
+            # for configured output with meta, "correct" the output type to file to avoid the uri_folder default value
+            if original_output and original_output.type:
+                if original_output.type in ["AnyFile", "uri_file"]:
+                    rest_output["job_output_type"] = "uri_file"
+
+
+def flatten_dict(
+    dct: Optional[Dict],
+    _type: Union[Type["_GroupAttrDict"], Type[GroupInput]],
+    *,
+    allow_dict_fields: Optional[List[str]] = None,
+) -> Dict:
+    """Flatten inputs/input_definitions dict for inputs dict build.
+
+    :param dct: The dictionary to flatten
+    :type dct: Dict
+    :param _type: Either _GroupAttrDict or GroupInput (both have the method `flatten`)
+    :type _type: Union[Type["_GroupAttrDict"], Type[GroupInput]]
+    :keyword allow_dict_fields: A list of keys for dictionary values that will be included in flattened output
+    :paramtype allow_dict_fields: Optional[List[str]]
+    :return: The flattened dict
+    :rtype: Dict
+    """
+    _result = {}
+    if dct is not None:
+        for key, val in dct.items():
+            # to support passing dict value as parameter group
+            if allow_dict_fields and key in allow_dict_fields and isinstance(val, dict):
+                # for child dict, all values are allowed to be dict
+                for flattened_key, flattened_val in flatten_dict(
+                    val, _type, allow_dict_fields=list(val.keys())
+                ).items():
+                    _result[key + "." + flattened_key] = flattened_val
+                continue
+            val = GroupInput.custom_class_value_to_attr_dict(val)
+            if isinstance(val, _type):
+                _result.update(val.flatten(group_parameter_name=key))
+                continue
+            _result[key] = val
+    return _result
+
+
+class NodeWithGroupInputMixin(NodeIOMixin):
+    """This class provide build_inputs_dict for a node to use ParameterGroup as an input."""
+
+    @classmethod
+    def _validate_group_input_type(
+        cls,
+        input_definition_dict: dict,
+        inputs: Dict[str, Union[Input, str, bool, int, float]],
+    ) -> None:
+        """Raise error when group input receive a value not group type.
+
+        :param input_definition_dict: The input definition dict
+        :type input_definition_dict: dict
+        :param inputs: The inputs
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        """
+        # Note: We put and extra validation here instead of doing it in pipeline._validate()
+        # due to group input will be discarded silently if assign it to a non-group parameter.
+        group_msg = "'%s' is defined as a parameter group but got input '%s' with type '%s'."
+        non_group_msg = "'%s' is defined as a parameter but got a parameter group as input."
+        for key, val in inputs.items():
+            definition = input_definition_dict.get(key)
+            val = GroupInput.custom_class_value_to_attr_dict(val)
+            if val is None:
+                continue
+            # 1. inputs.group = 'a string'
+            if isinstance(definition, GroupInput) and not isinstance(val, (_GroupAttrDict, dict)):
+                raise ValidationException(
+                    message=group_msg % (key, val, type(val)),
+                    no_personal_data_message=group_msg % ("[key]", "[val]", "[type(val)]"),
+                    target=ErrorTarget.PIPELINE,
+                    type=ValidationErrorType.INVALID_VALUE,
+                )
+            # 2. inputs.str_param = group
+            if not isinstance(definition, GroupInput) and isinstance(val, _GroupAttrDict):
+                raise ValidationException(
+                    message=non_group_msg % key,
+                    no_personal_data_message=non_group_msg % "[key]",
+                    target=ErrorTarget.PIPELINE,
+                    type=ValidationErrorType.INVALID_VALUE,
+                )
+
+    @classmethod
+    def _flatten_inputs_and_definition(
+        cls,
+        inputs: Dict[str, Union[Input, str, bool, int, float]],
+        input_definition_dict: dict,
+    ) -> Tuple[Dict, Dict]:
+        """
+        Flatten all GroupInput(definition) and GroupAttrDict recursively and build input dict.
+        For example:
+        input_definition_dict = {
+            "group1": GroupInput(
+                values={
+                    "param1": GroupInput(
+                         values={
+                             "param1_1": Input(type="str"),
+                         }
+                    ),
+                    "param2": Input(type="int"),
+                }
+            ),
+            "group2": GroupInput(
+                values={
+                    "param3": Input(type="str"),
+                }
+            ),
+        } => {
+            "group1.param1.param1_1": Input(type="str"),
+            "group1.param2": Input(type="int"),
+            "group2.param3": Input(type="str"),
+        }
+        inputs = {
+            "group1": {
+                "param1": {
+                    "param1_1": "value1",
+                },
+                "param2": 2,
+            },
+            "group2": {
+                "param3": "value3",
+            },
+        } => {
+            "group1.param1.param1_1": "value1",
+            "group1.param2": 2,
+            "group2.param3": "value3",
+        }
+        :param inputs: The inputs
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        :param input_definition_dict: The input definition dict
+        :type input_definition_dict: dict
+        :return: The flattened inputs and definition
+        :rtype: Tuple[Dict, Dict]
+        """
+        group_input_names = [key for key, val in input_definition_dict.items() if isinstance(val, GroupInput)]
+        flattened_inputs = flatten_dict(inputs, _GroupAttrDict, allow_dict_fields=group_input_names)
+        flattened_definition_dict = flatten_dict(input_definition_dict, GroupInput)
+        return flattened_inputs, flattened_definition_dict
+
+    def _build_inputs_dict(
+        self,
+        inputs: Dict[str, Union[Input, str, bool, int, float]],
+        *,
+        input_definition_dict: Optional[dict] = None,
+    ) -> InputsAttrDict:
+        """Build an input attribute dict so user can get/set inputs by
+        accessing attribute, eg: node1.inputs.xxx.
+
+        :param inputs: Provided kwargs when parameterizing component func.
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        :keyword input_definition_dict: Input definition dict from component entity.
+        :paramtype input_definition_dict: dict
+        :return: Built input attribute dict.
+        :rtype: InputsAttrDict
+        """
+
+        # TODO: should we support group input when there is no local input definition?
+        if input_definition_dict is not None:
+            # Validate group mismatch
+            self._validate_group_input_type(input_definition_dict, inputs)
+
+            # Flatten inputs and definition
+            flattened_inputs, flattened_definition_dict = self._flatten_inputs_and_definition(
+                inputs, input_definition_dict
+            )
+            # Build: zip all flattened parameter with definition
+            inputs = super()._build_inputs_dict(flattened_inputs, input_definition_dict=flattened_definition_dict)
+            return InputsAttrDict(GroupInput.restore_flattened_inputs(inputs))
+        return super()._build_inputs_dict(inputs)
+
+
+class PipelineJobIOMixin(NodeWithGroupInputMixin):
+    """Provides ability to wrap pipeline job inputs/outputs and build data bindings
+    dynamically."""
+
+    def _build_input(self, name: str, meta: Optional[Input], data: Any) -> "PipelineInput":
+        return PipelineInput(name=name, meta=meta, data=data, owner=self)
+
+    def _build_output(
+        self, name: str, meta: Optional[Union[Input, Output]], data: Optional[Union[Output, str]]
+    ) -> "PipelineOutput":
+        # TODO: settings data to None for un-configured outputs so we won't passing extra fields(eg: default mode)
+        result = PipelineOutput(port_name=name, meta=meta, data=data, owner=self)
+        return result
+
+    def _build_inputs_dict(
+        self,
+        inputs: Dict[str, Union[Input, str, bool, int, float]],
+        *,
+        input_definition_dict: Optional[dict] = None,
+    ) -> InputsAttrDict:
+        """Build an input attribute dict so user can get/set inputs by
+        accessing attribute, eg: node1.inputs.xxx.
+
+        :param inputs: Provided kwargs when parameterizing component func.
+        :type inputs: Dict[str, Union[Input, str, bool, int, float]]
+        :keyword input_definition_dict: Input definition dict from component entity.
+        :return: Built input attribute dict.
+        :rtype: InputsAttrDict
+        """
+        input_dict = super()._build_inputs_dict(inputs, input_definition_dict=input_definition_dict)
+        # TODO: should we do this when input_definition_dict is not None?
+        # TODO: should we put this in super()._build_inputs_dict?
+        if input_definition_dict is None:
+            return InputsAttrDict(GroupInput.restore_flattened_inputs(input_dict))
+        return input_dict
+
+    def _build_output_for_pipeline(self, name: str, data: Optional[Union[Output, NodeOutput]]) -> "PipelineOutput":
+        """Build an output object for pipeline and copy settings from source output.
+
+        :param name: Output name.
+        :type name: str
+        :param data: Output data.
+        :type data: Optional[Union[Output, NodeOutput]]
+        :return: Built output object.
+        :rtype: PipelineOutput
+        """
+        # pylint: disable=protected-access
+        if data is None:
+            # For None output, build an empty output builder
+            output_val = self._build_output(name=name, meta=None, data=None)
+        elif isinstance(data, Output):
+            # For output entity, build an output builder with data points to it
+            output_val = self._build_output(name=name, meta=data, data=data)
+        elif isinstance(data, NodeOutput):
+            # For output builder, build a new output builder and copy settings from it
+            output_val = self._build_output(name=name, meta=data._meta, data=None)
+            copy_output_setting(source=data, target=output_val)
+        else:
+            message = "Unsupported output type: {} for pipeline output: {}: {}"
+            raise ValidationException(
+                message=message.format(type(data), name, data),
+                no_personal_data_message=message,
+                target=ErrorTarget.PIPELINE,
+            )
+        return output_val
+
+    def _build_pipeline_outputs_dict(self, outputs: Dict) -> OutputsAttrDict:
+        """Build an output attribute dict without output definition metadata.
+        For pipeline outputs, its setting should be copied from node level outputs.
+
+        :param outputs: Node output dict or pipeline component's outputs.
+        :type outputs: Dict[str, Union[Output, NodeOutput]]
+        :return: Built dynamic output attribute dict.
+        :rtype: OutputsAttrDict
+        """
+        output_dict = {}
+        for key, val in outputs.items():
+            output_dict[key] = self._build_output_for_pipeline(name=key, data=val)
+        return OutputsAttrDict(output_dict)
+
+    def _build_outputs(self) -> Dict[str, Output]:
+        """Build outputs of this pipeline to a dict which maps output to actual
+        value.
+
+        The built dictionary's format aligns with component job's output yaml,
+        un-configured outputs will be None, eg:
+        {"eval_output": "${{jobs.eval.outputs.eval_output}}", "un_configured": None}
+
+        :return: The output dict
+        :rtype: Dict[str, Output]
+        """
+        outputs = {}
+        for name, output in self.outputs.items():  # type: ignore
+            if isinstance(output, NodeOutput):
+                output = output._to_job_output()  # pylint: disable=protected-access
+            outputs[name] = output
+        return outputs
+
+    def _get_default_input_val(self, val: Any):  # type: ignore
+        # use Default value as data placeholder for unfilled inputs.
+        # client side need to fill the default value for dsl.pipeline
+        if isinstance(val, GroupInput):
+            # Copy default value dict for group
+            return copy.deepcopy(val.default)
+        return val.default
+
+    def _update_output_types(self, rest_data_outputs: Dict) -> None:
+        """Won't clear output type for pipeline level outputs since it's required in rest object.
+
+        :param rest_data_outputs: The REST data outputs
+        :type rest_data_outputs: Dict
+        """
+
+
+class AutoMLNodeIOMixin(NodeIOMixin):
+    """Wrap outputs of automl node and build data bindings dynamically."""
+
+    def __init__(self, **kwargs):  # type: ignore
+        # add a inputs field to align with other nodes
+        self.inputs = {}
+        super(AutoMLNodeIOMixin, self).__init__(**kwargs)
+        if getattr(self, "outputs", None):
+            self._outputs = self._build_outputs_dict(self.outputs or {})
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_load_component.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_load_component.py
new file mode 100644
index 00000000..60c4cbe7
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_load_component.py
@@ -0,0 +1,313 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+from typing import Any, Callable, Dict, List, Mapping, Optional, Union, cast
+
+from marshmallow import INCLUDE
+
+from azure.ai.ml import Output
+from azure.ai.ml._schema import NestedField
+from azure.ai.ml._schema.pipeline.component_job import SweepSchema
+from azure.ai.ml.constants._common import BASE_PATH_CONTEXT_KEY, SOURCE_PATH_CONTEXT_KEY, CommonYamlFields
+from azure.ai.ml.constants._component import ControlFlowType, DataTransferTaskType, NodeType
+from azure.ai.ml.constants._compute import ComputeType
+from azure.ai.ml.dsl._component_func import to_component_func
+from azure.ai.ml.dsl._overrides_definition import OverrideDefinition
+from azure.ai.ml.entities._builders import (
+    BaseNode,
+    Command,
+    DataTransferCopy,
+    DataTransferExport,
+    DataTransferImport,
+    Import,
+    Parallel,
+    Spark,
+    Sweep,
+)
+from azure.ai.ml.entities._builders.condition_node import ConditionNode
+from azure.ai.ml.entities._builders.control_flow_node import ControlFlowNode
+from azure.ai.ml.entities._builders.do_while import DoWhile
+from azure.ai.ml.entities._builders.parallel_for import ParallelFor
+from azure.ai.ml.entities._builders.pipeline import Pipeline
+from azure.ai.ml.entities._component.component import Component
+from azure.ai.ml.entities._job.automl.automl_job import AutoMLJob
+from azure.ai.ml.entities._util import get_type_from_spec
+from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationException
+
+
+class _PipelineNodeFactory:
+    """A class to create pipeline node instances from yaml dict or rest objects without hard-coded type check."""
+
+    def __init__(self) -> None:
+        self._create_instance_funcs: dict = {}
+        self._load_from_rest_object_funcs: dict = {}
+
+        self.register_type(
+            _type=NodeType.COMMAND,
+            create_instance_func=lambda: Command.__new__(Command),
+            load_from_rest_object_func=Command._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.IMPORT,
+            create_instance_func=lambda: Import.__new__(Import),
+            load_from_rest_object_func=Import._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.PARALLEL,
+            create_instance_func=lambda: Parallel.__new__(Parallel),
+            load_from_rest_object_func=Parallel._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.PIPELINE,
+            create_instance_func=lambda: Pipeline.__new__(Pipeline),
+            load_from_rest_object_func=Pipeline._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.SWEEP,
+            create_instance_func=lambda: Sweep.__new__(Sweep),
+            load_from_rest_object_func=Sweep._from_rest_object,
+            nested_schema=NestedField(SweepSchema, unknown=INCLUDE),
+        )
+        self.register_type(
+            _type=NodeType.AUTOML,
+            create_instance_func=None,
+            load_from_rest_object_func=self._automl_from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.SPARK,
+            create_instance_func=lambda: Spark.__new__(Spark),
+            load_from_rest_object_func=Spark._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=ControlFlowType.DO_WHILE,
+            create_instance_func=None,
+            load_from_rest_object_func=DoWhile._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=ControlFlowType.IF_ELSE,
+            create_instance_func=None,
+            load_from_rest_object_func=ConditionNode._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=ControlFlowType.PARALLEL_FOR,
+            create_instance_func=None,
+            load_from_rest_object_func=ParallelFor._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type="_".join([NodeType.DATA_TRANSFER, DataTransferTaskType.COPY_DATA]),
+            create_instance_func=lambda: DataTransferCopy.__new__(DataTransferCopy),
+            load_from_rest_object_func=DataTransferCopy._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type="_".join([NodeType.DATA_TRANSFER, DataTransferTaskType.IMPORT_DATA]),
+            create_instance_func=lambda: DataTransferImport.__new__(DataTransferImport),
+            load_from_rest_object_func=DataTransferImport._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type="_".join([NodeType.DATA_TRANSFER, DataTransferTaskType.EXPORT_DATA]),
+            create_instance_func=lambda: DataTransferExport.__new__(DataTransferExport),
+            load_from_rest_object_func=DataTransferExport._from_rest_object,
+            nested_schema=None,
+        )
+        self.register_type(
+            _type=NodeType.FLOW_PARALLEL,
+            create_instance_func=lambda: Parallel.__new__(Parallel),
+            load_from_rest_object_func=None,
+            nested_schema=None,
+        )
+
+    @classmethod
+    def _get_func(cls, _type: str, funcs: Dict[str, Callable]) -> Callable:
+        if _type == NodeType._CONTAINER:
+            msg = (
+                "Component returned by 'list' is abbreviated and can not be used directly, "
+                "please use result from 'get'."
+            )
+            raise ValidationException(
+                message=msg,
+                no_personal_data_message=msg,
+                target=ErrorTarget.COMPONENT,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+        _type = get_type_from_spec({CommonYamlFields.TYPE: _type}, valid_keys=funcs)
+        return funcs[_type]
+
+    def get_create_instance_func(self, _type: str) -> Callable[..., BaseNode]:
+        """Get the function to create a new instance of the node.
+
+        :param _type: The type of the node.
+        :type _type: str
+        :return: The create instance function
+        :rtype: Callable[..., BaseNode]
+        """
+        return self._get_func(_type, self._create_instance_funcs)
+
+    def get_load_from_rest_object_func(self, _type: str) -> Callable:
+        """Get the function to load a node from a rest object.
+
+        :param _type: The type of the node.
+        :type _type: str
+        :return: The `_load_from_rest_object` function
+        :rtype: Callable[[Any], Union[BaseNode, AutoMLJob, ControlFlowNode]]
+        """
+        return self._get_func(_type, self._load_from_rest_object_funcs)
+
+    def register_type(
+        self,
+        _type: str,
+        *,
+        create_instance_func: Optional[Callable[..., Union[BaseNode, AutoMLJob]]] = None,
+        load_from_rest_object_func: Optional[Callable] = None,
+        nested_schema: Optional[Union[NestedField, List[NestedField]]] = None,
+    ) -> None:
+        """Register a type of node.
+
+        :param _type: The type of the node.
+        :type _type: str
+        :keyword create_instance_func: A function to create a new instance of the node
+        :paramtype create_instance_func: typing.Optional[typing.Callable[..., typing.Union[BaseNode, AutoMLJob]]]
+        :keyword load_from_rest_object_func: A function to load a node from a rest object
+        :paramtype load_from_rest_object_func: typing.Optional[typing.Callable[[Any], typing.Union[BaseNode, AutoMLJob\
+            , ControlFlowNode]]]
+        :keyword nested_schema: schema/schemas of corresponding nested field, will be used in \
+            PipelineJobSchema.jobs.value
+        :paramtype nested_schema: typing.Optional[typing.Union[NestedField, List[NestedField]]]
+        """
+        if create_instance_func is not None:
+            self._create_instance_funcs[_type] = create_instance_func
+        if load_from_rest_object_func is not None:
+            self._load_from_rest_object_funcs[_type] = load_from_rest_object_func
+        if nested_schema is not None:
+            from azure.ai.ml._schema.core.fields import TypeSensitiveUnionField
+            from azure.ai.ml._schema.pipeline.pipeline_component import PipelineComponentSchema
+            from azure.ai.ml._schema.pipeline.pipeline_job import PipelineJobSchema
+
+            for declared_fields in [
+                PipelineJobSchema._declared_fields,
+                PipelineComponentSchema._declared_fields,
+            ]:
+                jobs_value_field: TypeSensitiveUnionField = declared_fields["jobs"].value_field
+                if not isinstance(nested_schema, list):
+                    nested_schema = [nested_schema]
+                for nested_field in nested_schema:
+                    jobs_value_field.insert_type_sensitive_field(type_name=_type, field=nested_field)
+
+    def load_from_dict(self, *, data: dict, _type: Optional[str] = None) -> Union[BaseNode, AutoMLJob]:
+        """Load a node from a dict.
+
+        :keyword data: A dict containing the node's data.
+        :paramtype data: dict
+        :keyword _type: The type of the node. If not specified, it will be inferred from the data.
+        :paramtype _type: str
+        :return: The node
+        :rtype: Union[BaseNode, AutoMLJob]
+        """
+        if _type is None:
+            _type = data[CommonYamlFields.TYPE] if CommonYamlFields.TYPE in data else NodeType.COMMAND
+            # todo: refine Hard code for now to support different task type for DataTransfer node
+            if _type == NodeType.DATA_TRANSFER:
+                _type = "_".join([NodeType.DATA_TRANSFER, data.get("task", " ")])
+        else:
+            data[CommonYamlFields.TYPE] = _type
+
+        new_instance: Union[BaseNode, AutoMLJob] = self.get_create_instance_func(_type)()
+
+        if isinstance(new_instance, BaseNode):
+            # parse component
+            component_key = new_instance._get_component_attr_name()
+            if component_key in data and isinstance(data[component_key], dict):
+                data[component_key] = Component._load(
+                    data=data[component_key],
+                    yaml_path=data[component_key].pop(SOURCE_PATH_CONTEXT_KEY, None),
+                )
+        # TODO: Bug Item number: 2883415
+        new_instance.__init__(**data)  # type: ignore
+        return new_instance
+
+    def load_from_rest_object(
+        self, *, obj: dict, _type: Optional[str] = None, **kwargs: Any
+    ) -> Union[BaseNode, AutoMLJob, ControlFlowNode]:
+        """Load a node from a rest object.
+
+        :keyword obj: A rest object containing the node's data.
+        :paramtype obj: dict
+        :keyword _type: The type of the node. If not specified, it will be inferred from the data.
+        :paramtype _type: str
+        :return: The node
+        :rtype: Union[BaseNode, AutoMLJob, ControlFlowNode]
+        """
+
+        # TODO: Remove in PuP with native import job/component type support in MFE/Designer
+        if "computeId" in obj and obj["computeId"] and obj["computeId"].endswith("/" + ComputeType.ADF):
+            _type = NodeType.IMPORT
+
+        if _type is None:
+            _type = obj[CommonYamlFields.TYPE] if CommonYamlFields.TYPE in obj else NodeType.COMMAND
+            # todo: refine Hard code for now to support different task type for DataTransfer node
+            if _type == NodeType.DATA_TRANSFER:
+                _type = "_".join([NodeType.DATA_TRANSFER, obj.get("task", " ")])
+        else:
+            obj[CommonYamlFields.TYPE] = _type
+
+        res: Union[BaseNode, AutoMLJob, ControlFlowNode] = self.get_load_from_rest_object_func(_type)(obj, **kwargs)
+        return res
+
+    @classmethod
+    def _automl_from_rest_object(cls, node: Dict) -> AutoMLJob:
+        _outputs = cast(Dict[str, Union[str, dict]], node.get("outputs"))
+        # rest dict outputs -> Output objects
+        outputs = AutoMLJob._from_rest_outputs(_outputs)
+        # Output objects -> yaml dict outputs
+        parsed_outputs = {}
+        for key, val in outputs.items():
+            if isinstance(val, Output):
+                val = val._to_dict()
+            parsed_outputs[key] = val
+        node["outputs"] = parsed_outputs
+        return AutoMLJob._load_from_dict(
+            node,
+            context={BASE_PATH_CONTEXT_KEY: "./"},
+            additional_message="Failed to load automl task from backend.",
+            inside_pipeline=True,
+        )
+
+
+def _generate_component_function(
+    component_entity: Component,
+    override_definitions: Optional[Mapping[str, OverrideDefinition]] = None,  # pylint: disable=unused-argument
+) -> Callable[..., Union[Command, Parallel]]:
+    # Generate a function which returns a component node.
+    def create_component_func(**kwargs: Any) -> Union[BaseNode, AutoMLJob]:
+        # todo: refine Hard code for now to support different task type for DataTransfer node
+        _type = component_entity.type
+        if _type == NodeType.DATA_TRANSFER:
+            # TODO: Bug Item number: 2883431
+            _type = "_".join([NodeType.DATA_TRANSFER, component_entity.task])  # type: ignore
+            if component_entity.task == DataTransferTaskType.IMPORT_DATA:  # type: ignore
+                return pipeline_node_factory.load_from_dict(
+                    data={"component": component_entity, "_from_component_func": True, **kwargs},
+                    _type=_type,
+                )
+        return pipeline_node_factory.load_from_dict(
+            data={"component": component_entity, "inputs": kwargs, "_from_component_func": True},
+            _type=_type,
+        )
+
+    res: Callable = to_component_func(component_entity, create_component_func)
+    return res
+
+
+pipeline_node_factory = _PipelineNodeFactory()
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_expression.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_expression.py
new file mode 100644
index 00000000..49bb8a61
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_expression.py
@@ -0,0 +1,662 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+
+import re
+import tempfile
+from collections import namedtuple
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, cast
+
+from azure.ai.ml._utils.utils import dump_yaml_to_file, get_all_data_binding_expressions, load_yaml
+from azure.ai.ml.constants._common import AZUREML_PRIVATE_FEATURES_ENV_VAR, DefaultOpenEncoding
+from azure.ai.ml.constants._component import ComponentParameterTypes, IOConstants
+from azure.ai.ml.exceptions import UserErrorException
+
+if TYPE_CHECKING:
+    from azure.ai.ml.entities._builders import BaseNode
+
+ExpressionInput = namedtuple("ExpressionInput", ["name", "type", "value"])
+NONE_PARAMETER_TYPE = "None"
+
+
+class PipelineExpressionOperator:
+    """Support operator in native Python experience."""
+
+    ADD = "+"
+    SUB = "-"
+    MUL = "*"
+    DIV = "/"
+    MOD = "%"
+    POW = "**"
+    FLOORDIV = "//"
+    LT = "<"
+    GT = ">"
+    LTE = "<="
+    GTE = ">="
+    EQ = "=="
+    NE = "!="
+    AND = "&"
+    OR = "|"
+    XOR = "^"
+
+
+_SUPPORTED_OPERATORS = {
+    getattr(PipelineExpressionOperator, attr)
+    for attr in PipelineExpressionOperator.__dict__
+    if not attr.startswith("__")
+}
+
+
+def _enumerate_operation_combination() -> Dict[str, Union[str, Exception]]:
+    """Enumerate the result type of binary operations on types
+
+    Leverages `eval` to validate operation and get its result type.
+
+    :return: A dictionary that maps an operation to either:
+      * A result type
+      * An Exception
+    :rtype: Dict[str, Union[str, Exception]]
+    """
+    res: Dict = {}
+    primitive_types_values = {
+        NONE_PARAMETER_TYPE: repr(None),
+        ComponentParameterTypes.BOOLEAN: repr(True),
+        ComponentParameterTypes.INTEGER: repr(1),
+        ComponentParameterTypes.NUMBER: repr(1.0),
+        ComponentParameterTypes.STRING: repr("1"),
+    }
+    for type1, operand1 in primitive_types_values.items():
+        for type2, operand2 in primitive_types_values.items():
+            for operator in _SUPPORTED_OPERATORS:
+                k = f"{type1} {operator} {type2}"
+                try:
+                    eval_result = eval(f"{operand1} {operator} {operand2}")  # pylint: disable=eval-used # nosec
+                    res[k] = IOConstants.PRIMITIVE_TYPE_2_STR[type(eval_result)]
+                except TypeError:
+                    error_message = (
+                        f"Operator '{operator}' is not supported between instances of '{type1}' and '{type2}'."
+                    )
+                    res[k] = UserErrorException(message=error_message, no_personal_data_message=error_message)
+    return res
+
+
+# enumerate and store as a lookup table:
+#   key format is "<operand1_type> <operator> <operand2_type>"
+#   value can be either result type as str and UserErrorException for invalid operation
+_OPERATION_RESULT_TYPE_LOOKUP = _enumerate_operation_combination()
+
+
+class PipelineExpressionMixin:
+    _SUPPORTED_PRIMITIVE_TYPES = (bool, int, float, str)
+    _SUPPORTED_PIPELINE_INPUT_TYPES = (
+        ComponentParameterTypes.BOOLEAN,
+        ComponentParameterTypes.INTEGER,
+        ComponentParameterTypes.NUMBER,
+        ComponentParameterTypes.STRING,
+    )
+
+    def _validate_binary_operation(self, other: Any, operator: str) -> None:
+        from azure.ai.ml.entities._job.pipeline._io import NodeOutput, PipelineInput
+
+        if (
+            other is not None
+            and not isinstance(other, self._SUPPORTED_PRIMITIVE_TYPES)
+            and not isinstance(other, (PipelineInput, NodeOutput, PipelineExpression))
+        ):
+            error_message = (
+                f"Operator '{operator}' is not supported with {type(other)}; "
+                "currently only support primitive types (None, bool, int, float and str), "
+                "pipeline input, component output and expression."
+            )
+            raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+
+    def __add__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.ADD)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.ADD)
+
+    def __radd__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.ADD)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.ADD)
+
+    def __sub__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.SUB)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.SUB)
+
+    def __rsub__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.SUB)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.SUB)
+
+    def __mul__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.MUL)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.MUL)
+
+    def __rmul__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.MUL)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.MUL)
+
+    def __truediv__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.DIV)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.DIV)
+
+    def __rtruediv__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.DIV)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.DIV)
+
+    def __mod__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.MOD)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.MOD)
+
+    def __rmod__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.MOD)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.MOD)
+
+    def __pow__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.POW)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.POW)
+
+    def __rpow__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.POW)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.POW)
+
+    def __floordiv__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.FLOORDIV)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.FLOORDIV)
+
+    def __rfloordiv__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.FLOORDIV)
+        return PipelineExpression._from_operation(other, self, PipelineExpressionOperator.FLOORDIV)
+
+    def __lt__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.LT)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.LT)
+
+    def __gt__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.GT)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.GT)
+
+    def __le__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.LTE)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.LTE)
+
+    def __ge__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.GTE)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.GTE)
+
+    # TODO: Bug Item number: 2883354
+    def __eq__(self, other: Any) -> "PipelineExpression":  # type: ignore
+        self._validate_binary_operation(other, PipelineExpressionOperator.EQ)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.EQ)
+
+    # TODO: Bug Item number: 2883354
+    def __ne__(self, other: Any) -> "PipelineExpression":  # type: ignore
+        self._validate_binary_operation(other, PipelineExpressionOperator.NE)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.NE)
+
+    def __and__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.AND)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.AND)
+
+    def __or__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.OR)
+        return PipelineExpression._from_operation(self, other, PipelineExpressionOperator.OR)
+
+    def __xor__(self, other: Any) -> "PipelineExpression":
+        self._validate_binary_operation(other, PipelineExpressionOperator.XOR)
+        return PipelineExpression._from_operation(self, None, PipelineExpressionOperator.XOR)
+
+    def __bool__(self) -> bool:
+        """Python method that is used to implement truth value testing and the built-in operation bool().
+
+        This method is not supported as PipelineExpressionMixin is designed to record operation history,
+        while this method can only return False or True, leading to history breaks here.
+        As overloadable boolean operators PEP (refer to: https://www.python.org/dev/peps/pep-0335/)
+        was rejected, logical operations are also not supported.
+
+        :return: True if not inside dsl pipeline func, raises otherwise
+        :rtype: bool
+        """
+        from azure.ai.ml.dsl._pipeline_component_builder import _is_inside_dsl_pipeline_func
+
+        # note: unexpected bool test always be checking if the object is None;
+        # so for non-pipeline scenarios, directly return True to avoid unexpected breaking,
+        # and for pipeline scenarios, will use is not None to replace bool test.
+        if not _is_inside_dsl_pipeline_func():
+            return True
+
+        error_message = f"Type {type(self)} is not supported for operation bool()."
+        raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+
+
+class PipelineExpression(PipelineExpressionMixin):
+    """Pipeline expression entity.
+
+    Use PipelineExpression to support simple and trivial parameter transformation tasks with constants
+    or other parameters. Operations are recorded in this class during executions, and expected result
+    will be generated for corresponding scenario.
+    """
+
+    _PIPELINE_INPUT_PREFIX = ["parent", "inputs"]
+    _PIPELINE_INPUT_PATTERN = re.compile(pattern=r"parent.inputs.(?P<pipeline_input_name>[^.]+)")
+    _PIPELINE_INPUT_NAME_GROUP = "pipeline_input_name"
+    # AML type to Python type, for generated Python code
+    _TO_PYTHON_TYPE = {
+        ComponentParameterTypes.BOOLEAN: bool.__name__,
+        ComponentParameterTypes.INTEGER: int.__name__,
+        ComponentParameterTypes.NUMBER: float.__name__,
+        ComponentParameterTypes.STRING: str.__name__,
+    }
+
+    _INDENTATION = "    "
+    _IMPORT_MLDESIGNER_LINE = "from mldesigner import command_component, Output"
+    _DECORATOR_LINE = "@command_component(@@decorator_parameters@@)"
+    _COMPONENT_FUNC_NAME = "expression_func"
+    _COMPONENT_FUNC_DECLARATION_LINE = (
+        f"def {_COMPONENT_FUNC_NAME}(@@component_parameters@@)" " -> Output(type=@@return_type@@):"
+    )
+    _PYTHON_CACHE_FOLDER_NAME = "__pycache__"
+
+    def __init__(self, postfix: List[str], inputs: Dict[str, ExpressionInput]):
+        self._postfix = postfix
+        self._inputs = inputs.copy()  # including PiplineInput and Output, extra stored name and type
+        self._result_type: Optional[str] = None
+        self._created_component = None
+
+    @property
+    def expression(self) -> str:
+        """Infix expression string, wrapped with parentheses.
+
+        :return: The infix expression
+        :rtype: str
+        """
+        return self._to_infix()
+
+    def __str__(self) -> str:
+        return self._to_data_binding()
+
+    def _data_binding(self) -> str:
+        return self._to_data_binding()
+
+    def _to_infix(self) -> str:
+        stack = []
+        for token in self._postfix:
+            if token not in _SUPPORTED_OPERATORS:
+                stack.append(token)
+                continue
+            operand2, operand1 = stack.pop(), stack.pop()
+            stack.append(f"({operand1} {token} {operand2})")
+        return stack.pop()
+
+    # pylint: disable=too-many-statements
+    @staticmethod
+    def _handle_operand(
+        operand: "PipelineExpression",
+        postfix: List[str],
+        expression_inputs: Dict[str, ExpressionInput],
+        pipeline_inputs: dict,
+    ) -> Tuple[List[str], Dict[str, ExpressionInput]]:
+        """Handle operand in expression, update postfix expression and expression inputs.
+
+        :param operand: The operand
+        :type operand: "PipelineExpression"
+        :param postfix:
+        :type postfix: List[str]
+        :param expression_inputs: The expression inputs
+        :type expression_inputs: Dict[str, ExpressionInput]
+        :param pipeline_inputs: The pipeline inputs
+        :type pipeline_inputs: dict
+        :return: A 2-tuple of the updated postfix expression and expression inputs
+        :rtype: Tuple[List[str], Dict[str, ExpressionInput]]
+        """
+        from azure.ai.ml.entities._job.pipeline._io import NodeOutput, PipelineInput
+
+        def _update_postfix(_postfix: List[str], _old_name: str, _new_name: str) -> List[str]:
+            return list(map(lambda _x: _new_name if _x == _old_name else _x, _postfix))
+
+        def _get_or_create_input_name(
+            _original_name: str,
+            _operand: Union[PipelineInput, NodeOutput],
+            _expression_inputs: Dict[str, ExpressionInput],
+        ) -> str:
+            """Get or create expression input name as current operand may have appeared in expression.
+
+            :param _original_name: The original name
+            :type _original_name: str
+            :param _operand: The expression operand
+            :type _operand: Union[PipelineInput, NodeOutput]
+            :param _expression_inputs: The expression inputs
+            :type _expression_inputs: Dict[str, ExpressionInput]
+            :return: The input name
+            :rtype: str
+            """
+            _existing_id_to_name = {id(_v.value): _k for _k, _v in _expression_inputs.items()}
+            if id(_operand) in _existing_id_to_name:
+                return _existing_id_to_name[id(_operand)]
+            # use a counter to generate a unique name for current operand
+            _name, _counter = _original_name, 0
+            while _name in _expression_inputs:
+                _name = f"{_original_name}_{_counter}"
+                _counter += 1
+            return _name
+
+        def _handle_pipeline_input(
+            _pipeline_input: PipelineInput,
+            _postfix: List[str],
+            _expression_inputs: Dict[str, ExpressionInput],
+        ) -> Tuple[List[str], dict]:
+            _name = _pipeline_input._port_name
+            # 1. use name with counter for pipeline input; 2. add component's name to component output
+            if _name in _expression_inputs:
+                _seen_input = _expression_inputs[_name]
+                if isinstance(_seen_input.value, PipelineInput):
+                    _name = _get_or_create_input_name(_name, _pipeline_input, _expression_inputs)
+                else:
+                    _expression_inputs.pop(_name)
+                    _new_name = f"{_seen_input.value._owner.component.name}__{_seen_input.value._port_name}"
+                    _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, param_input[_pipeline_input._port_name].type, _pipeline_input
+            )
+            return _postfix, _expression_inputs
+
+        def _handle_component_output(
+            _component_output: NodeOutput,
+            _postfix: List[str],
+            _expression_inputs: Dict[str, ExpressionInput],
+        ) -> Tuple[List[str], dict]:
+            if _component_output._meta is not None and not _component_output._meta._is_primitive_type:
+                error_message = (
+                    f"Component output {_component_output._port_name} in expression must "
+                    f"be a primitive type with value {True!r}, "
+                    f"got {_component_output._meta._is_primitive_type!r}"
+                )
+                raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+            _name = _component_output._port_name
+            _has_prefix = False
+            # "output" is the default output name for command component, add component's name as prefix
+            if _name == "output":
+                if _component_output._owner is not None and not isinstance(_component_output._owner.component, str):
+                    _name = f"{_component_output._owner.component.name}__output"
+                _has_prefix = True
+            # following loop is expected to execute at most twice:
+            #   1. add component's name to output(s)
+            #   2. use name with counter
+            while _name in _expression_inputs:
+                _seen_input = _expression_inputs[_name]
+                if isinstance(_seen_input.value, PipelineInput):
+                    if not _has_prefix:
+                        if _component_output._owner is not None and not isinstance(
+                            _component_output._owner.component, str
+                        ):
+                            _name = f"{_component_output._owner.component.name}__{_component_output._port_name}"
+                        _has_prefix = True
+                        continue
+                    _name = _get_or_create_input_name(_name, _component_output, _expression_inputs)
+                else:
+                    if not _has_prefix:
+                        _expression_inputs.pop(_name)
+                        _new_name = f"{_seen_input.value._owner.component.name}__{_seen_input.value._port_name}"
+                        _postfix = _update_postfix(_postfix, _name, _new_name)
+                        _expression_inputs[_new_name] = ExpressionInput(_new_name, _seen_input.type, _seen_input)
+                        if _component_output._owner is not None and not isinstance(
+                            _component_output._owner.component, str
+                        ):
+                            _name = f"{_component_output._owner.component.name}__{_component_output._port_name}"
+                        _has_prefix = True
+                    _name = _get_or_create_input_name(_name, _component_output, _expression_inputs)
+            _postfix.append(_name)
+            _expression_inputs[_name] = ExpressionInput(_name, _component_output.type, _component_output)
+            return _postfix, _expression_inputs
+
+        if operand is None or isinstance(operand, PipelineExpression._SUPPORTED_PRIMITIVE_TYPES):
+            postfix.append(repr(operand))
+        elif isinstance(operand, PipelineInput):
+            postfix, expression_inputs = _handle_pipeline_input(operand, postfix, expression_inputs)
+        elif isinstance(operand, NodeOutput):
+            postfix, expression_inputs = _handle_component_output(operand, postfix, expression_inputs)
+        elif isinstance(operand, PipelineExpression):
+            postfix.extend(operand._postfix.copy())
+            expression_inputs.update(operand._inputs.copy())
+        return postfix, expression_inputs
+
+    @staticmethod
+    def _from_operation(operand1: Any, operand2: Any, operator: str) -> "PipelineExpression":
+        if operator not in _SUPPORTED_OPERATORS:
+            error_message = (
+                f"Operator '{operator}' is not supported operator, "
+                f"currently supported operators are {','.join(_SUPPORTED_OPERATORS)}."
+            )
+            raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+
+        # get all pipeline input types from builder stack
+        # TODO: check if there is pipeline input we cannot know its type (missing in `PipelineComponentBuilder.inputs`)?
+        from azure.ai.ml.dsl._pipeline_component_builder import _definition_builder_stack
+
+        res = _definition_builder_stack.top()
+        pipeline_inputs = res.inputs if res is not None else {}
+        postfix: List[str] = []
+        inputs: Dict[str, ExpressionInput] = {}
+        postfix, inputs = PipelineExpression._handle_operand(operand1, postfix, inputs, pipeline_inputs)
+        postfix, inputs = PipelineExpression._handle_operand(operand2, postfix, inputs, pipeline_inputs)
+        postfix.append(operator)
+        return PipelineExpression(postfix, inputs)
+
+    @property
+    def _string_concatenation(self) -> bool:
+        """If all operands are string and operations are addition, it is a string concatenation expression.
+
+        :return: Whether this represents string concatenation
+        :rtype: bool
+        """
+        for token in self._postfix:
+            # operator can only be "+" for string concatenation
+            if token in _SUPPORTED_OPERATORS:
+                if token != PipelineExpressionOperator.ADD:
+                    return False
+                continue
+            # constant and PiplineInput should be type string
+            if token in self._inputs:
+                if self._inputs[token].type != ComponentParameterTypes.STRING:
+                    return False
+            else:
+                if not isinstance(eval(token), str):  # pylint: disable=eval-used # nosec
+                    return False
+        return True
+
+    def _to_data_binding(self) -> str:
+        """Convert operands to data binding and concatenate them in the order of postfix expression.
+
+        :return: The data binding
+        :rtype: str
+        """
+        if not self._string_concatenation:
+            error_message = (
+                "Only string concatenation expression is supported to convert to data binding, "
+                f"current expression is '{self.expression}'."
+            )
+            raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+
+        stack = []
+        for token in self._postfix:
+            if token != PipelineExpressionOperator.ADD:
+                if token in self._inputs:
+                    stack.append(self._inputs[token].value._data_binding())
+                else:
+                    stack.append(eval(token))  # pylint: disable=eval-used # nosec
+                continue
+            operand2, operand1 = stack.pop(), stack.pop()
+            stack.append(operand1 + operand2)
+        res: str = stack.pop()
+        return res
+
+    def resolve(self) -> Union[str, "BaseNode"]:
+        """Resolve expression to data binding or component, depend on the operations.
+
+        :return: The data binding string or the component
+        :rtype: Union[str, BaseNode]
+        """
+        if self._string_concatenation:
+            return self._to_data_binding()
+        return cast(Union[str, "BaseNode"], self._create_component())
+
+    @staticmethod
+    def parse_pipeline_inputs_from_data_binding(data_binding: str) -> List[str]:
+        """Parse all PipelineInputs name from data binding expression.
+
+        :param data_binding: Data binding expression
+        :type data_binding: str
+        :return: List of PipelineInput's name from given data binding expression
+        :rtype: List[str]
+        """
+        pipeline_input_names = []
+        for single_data_binding in get_all_data_binding_expressions(
+            value=data_binding,
+            binding_prefix=PipelineExpression._PIPELINE_INPUT_PREFIX,
+            is_singular=False,
+        ):
+            m = PipelineExpression._PIPELINE_INPUT_PATTERN.match(single_data_binding)
+            # `get_all_data_binding_expressions` should work as pre-filter, so no need to concern `m` is None
+            if m is not None:
+                pipeline_input_names.append(m.group(PipelineExpression._PIPELINE_INPUT_NAME_GROUP))
+        return pipeline_input_names
+
+    @staticmethod
+    def _get_operation_result_type(type1: str, operator: str, type2: str) -> str:
+        def _validate_operand_type(_type: str) -> None:
+            if _type != NONE_PARAMETER_TYPE and _type not in PipelineExpression._SUPPORTED_PIPELINE_INPUT_TYPES:
+                error_message = (
+                    f"Pipeline input type {_type!r} is not supported in expression; "
+                    f"currently only support None, "
+                    + ", ".join(PipelineExpression._SUPPORTED_PIPELINE_INPUT_TYPES)
+                    + "."
+                )
+                raise UserErrorException(message=error_message, no_personal_data_message=error_message)
+
+        _validate_operand_type(type1)
+        _validate_operand_type(type2)
+        operation = f"{type1} {operator} {type2}"
+        lookup_value = _OPERATION_RESULT_TYPE_LOOKUP.get(operation)
+        if isinstance(lookup_value, str):
+            return lookup_value  # valid operation, return result type
+        _user_exception: UserErrorException = lookup_value
+        raise _user_exception  # invalid operation, raise UserErrorException
+
+    def _get_operand_type(self, operand: str) -> str:
+        if operand in self._inputs:
+            res: str = self._inputs[operand].type
+            return res
+        primitive_type = type(eval(operand))  # pylint: disable=eval-used # nosec
+        res_type: str = IOConstants.PRIMITIVE_TYPE_2_STR.get(primitive_type, NONE_PARAMETER_TYPE)
+        return res_type
+
+    @property
+    def _component_code(self) -> str:
+        def _generate_function_code_lines() -> Tuple[List[str], str]:
+            """Return lines of code and return type.
+
+            :return: A 2-tuple of (function body, return type name)
+            :rtype: Tuple[List[str], str]
+            """
+            _inter_id, _code, _stack = 0, [], []
+            _line_recorder: Dict = {}
+            for _token in self._postfix:
+                if _token not in _SUPPORTED_OPERATORS:
+                    _type = self._get_operand_type(_token)
+                    _stack.append((_token, _type))
+                    continue
+                _operand2, _type2 = _stack.pop()
+                _operand1, _type1 = _stack.pop()
+                _current_line = f"{_operand1} {_token} {_operand2}"
+                if _current_line in _line_recorder:
+                    _inter_var, _inter_var_type = _line_recorder[_current_line]
+                else:
+                    _inter_var = f"inter_var_{_inter_id}"
+                    _inter_id += 1
+                    _inter_var_type = self._get_operation_result_type(_type1, _token, _type2)
+                    _code.append(f"{self._INDENTATION}{_inter_var} = {_current_line}")
+                    _line_recorder[_current_line] = (_inter_var, _inter_var_type)
+                _stack.append((_inter_var, _inter_var_type))
+            _return_var, _result_type = _stack.pop()
+            _code.append(f"{self._INDENTATION}return {_return_var}")
+            return _code, _result_type
+
+        def _generate_function_decorator_and_declaration_lines(_return_type: str) -> List[str]:
+            # decorator parameters
+            _display_name = f'{self._INDENTATION}display_name="Expression: {self.expression}",'
+            _decorator_parameters = "\n" + "\n".join([_display_name]) + "\n"
+            # component parameters
+            _component_parameters = []
+            for _name in sorted(self._inputs):
+                _type = self._TO_PYTHON_TYPE[self._inputs[_name].type]
+                _component_parameters.append(f"{_name}: {_type}")
+            _component_parameters_str = (
+                "\n"
+                + "\n".join(
+                    [f"{self._INDENTATION}{_component_parameter}," for _component_parameter in _component_parameters]
+                )
+                + "\n"
+            )
+            return [
+                self._IMPORT_MLDESIGNER_LINE + "\n\n",
+                self._DECORATOR_LINE.replace("@@decorator_parameters@@", _decorator_parameters),
+                self._COMPONENT_FUNC_DECLARATION_LINE.replace(
+                    "@@component_parameters@@", _component_parameters_str
+                ).replace("@@return_type@@", f'"{_return_type}"'),
+            ]
+
+        lines, result_type = _generate_function_code_lines()
+        self._result_type = result_type
+        code = _generate_function_decorator_and_declaration_lines(result_type) + lines
+        return "\n".join(code) + "\n"
+
+    def _create_component(self) -> Any:
+        def _generate_python_file(_folder: Path) -> None:
+            _folder.mkdir()
+            with open(_folder / "expression_component.py", "w", encoding=DefaultOpenEncoding.WRITE) as _f:
+                _f.write(self._component_code)
+
+        def _generate_yaml_file(_path: Path) -> None:
+            _data_folder = Path(__file__).parent / "data"
+            # update YAML content from template and dump
+            with open(_data_folder / "expression_component_template.yml", "r", encoding=DefaultOpenEncoding.READ) as _f:
+                _data = load_yaml(_f)
+            _data["display_name"] = f"Expression: {self.expression}"
+            _data["inputs"] = {}
+            _data["outputs"]["output"]["type"] = self._result_type
+            _command_inputs_items = []
+            for _name in sorted(self._inputs):
+                _type = self._inputs[_name].type
+                _data["inputs"][_name] = {"type": _type}
+                _command_inputs_items.append(_name + '="${{inputs.' + _name + '}}"')
+            _command_inputs_string = " ".join(_command_inputs_items)
+            _command_output_string = 'output="${{outputs.output}}"'
+            _command = (
+                "mldesigner execute --source expression_component.py --name expression_func"
+                " --inputs " + _command_inputs_string + " --outputs " + _command_output_string
+            )
+            _data["command"] = _data["command"].format(command_placeholder=_command)
+            dump_yaml_to_file(_path, _data)
+
+        if self._created_component is None:
+            tmp_folder = Path(tempfile.mkdtemp())
+            code_folder = tmp_folder / "src"
+            yaml_path = tmp_folder / "component_spec.yml"
+            _generate_python_file(code_folder)
+            _generate_yaml_file(yaml_path)
+
+            from azure.ai.ml import load_component
+
+            component_func = load_component(yaml_path)
+            component_kwargs = {k: v.value for k, v in self._inputs.items()}
+            self._created_component = component_func(**component_kwargs)
+            if self._created_component is not None:
+                self._created_component.environment_variables = {AZUREML_PRIVATE_FEATURES_ENV_VAR: "true"}
+        return self._created_component
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_job_helpers.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_job_helpers.py
new file mode 100644
index 00000000..3a7d89e7
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/_pipeline_job_helpers.py
@@ -0,0 +1,182 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+import re
+from typing import Dict, List, Tuple, Type, Union
+
+from azure.ai.ml._restclient.v2023_04_01_preview.models import InputDeliveryMode
+from azure.ai.ml._restclient.v2023_04_01_preview.models import JobInput as RestJobInput
+from azure.ai.ml._restclient.v2023_04_01_preview.models import JobOutput as RestJobOutput
+from azure.ai.ml._restclient.v2023_04_01_preview.models import Mpi, PyTorch, Ray, TensorFlow
+from azure.ai.ml.constants._component import ComponentJobConstants
+from azure.ai.ml.entities._inputs_outputs import Input, Output
+from azure.ai.ml.entities._job._input_output_helpers import (
+    INPUT_MOUNT_MAPPING_FROM_REST,
+    INPUT_MOUNT_MAPPING_TO_REST,
+    OUTPUT_MOUNT_MAPPING_FROM_REST,
+    OUTPUT_MOUNT_MAPPING_TO_REST,
+)
+from azure.ai.ml.entities._util import normalize_job_input_output_type
+from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationException
+
+
+def process_sdk_component_job_io(
+    io: Dict,
+    io_binding_regex_list: List[str],
+) -> Tuple:
+    """Separates SDK ComponentJob inputs that are data bindings (i.e. string inputs prefixed with 'inputs.' or
+    'outputs.') and dataset and literal inputs/outputs.
+
+    :param io: Input or output dictionary of an SDK ComponentJob
+    :type io:  Dict[str, Union[str, float, bool, Input]]
+    :param io_binding_regex_list: A list of regexes for io bindings
+    :type io_binding_regex_list: List[str]
+    :return: A tuple of dictionaries:
+      * One mapping inputs to REST formatted ComponentJobInput/ComponentJobOutput for data binding io.
+      * The other dictionary contains any IO that is not a databinding that is yet to be turned into REST form
+    :rtype: Tuple[Dict[str, str], Dict[str, Union[str, float, bool, Input]]]
+    """
+    io_bindings: Dict = {}
+    dataset_literal_io: Dict = {}
+    legacy_io_binding_regex_list = [
+        ComponentJobConstants.LEGACY_INPUT_PATTERN,
+        ComponentJobConstants.LEGACY_OUTPUT_PATTERN,
+    ]
+    for io_name, io_value in io.items():
+        if isinstance(io_value, (Input, Output)) and isinstance(io_value.path, str):
+            mode = io_value.mode
+            path = io_value.path
+            name = io_value.name if hasattr(io_value, "name") else None
+            version = io_value.version if hasattr(io_value, "version") else None
+            if any(re.match(item, path) for item in io_binding_regex_list):
+                # Yaml syntax requires using ${{}} to enclose inputs and outputs bindings
+                # io_bindings[io_name] = io_value
+                io_bindings.update({io_name: {"value": path}})
+                # add mode to literal value for binding input
+                if mode:
+                    if isinstance(io_value, Input):
+                        io_bindings[io_name].update({"mode": INPUT_MOUNT_MAPPING_TO_REST[mode]})
+                    else:
+                        io_bindings[io_name].update({"mode": OUTPUT_MOUNT_MAPPING_TO_REST[mode]})
+                if name or version:
+                    assert isinstance(io_value, Output)
+                    if name:
+                        io_bindings[io_name].update({"name": name})
+                    if version:
+                        io_bindings[io_name].update({"version": version})
+                if isinstance(io_value, Output) and io_value.name is not None:
+                    # when the output should be registered,
+                    # we add io_value to dataset_literal_io for further to_rest_data_outputs
+                    dataset_literal_io[io_name] = io_value
+            elif any(re.match(item, path) for item in legacy_io_binding_regex_list):
+                new_format = path.replace("{{", "{{parent.")
+                msg = "{} has changed to {}, please change to use new format."
+                raise ValidationException(
+                    message=msg.format(path, new_format),
+                    no_personal_data_message=msg.format("[io_value]", "[io_value_new_format]"),
+                    target=ErrorTarget.PIPELINE,
+                    error_category=ErrorCategory.USER_ERROR,
+                )
+            else:
+                dataset_literal_io[io_name] = io_value
+        else:
+            # Collect non-input data inputs
+            dataset_literal_io[io_name] = io_value
+    return io_bindings, dataset_literal_io
+
+
+def from_dict_to_rest_io(
+    io: Dict[str, Union[str, dict]],
+    rest_object_class: Union[Type[RestJobInput], Type[RestJobOutput]],
+    io_binding_regex_list: List[str],
+) -> Tuple[Dict[str, str], Dict[str, Union[RestJobInput, RestJobOutput]]]:
+    """Translate rest JObject dictionary to rest inputs/outputs and bindings.
+
+    :param io: Input or output dictionary.
+    :type io: Dict[str, Union[str, dict]]
+    :param rest_object_class: RestJobInput or RestJobOutput
+    :type rest_object_class: Union[Type[RestJobInput], Type[RestJobOutput]]
+    :param io_binding_regex_list: A list of regexes for io bindings
+    :type io_binding_regex_list: List[str]
+    :return: Map from IO name to IO bindings and Map from IO name to IO objects.
+    :rtype: Tuple[Dict[str, str], Dict[str, Union[RestJobInput, RestJobOutput]]]
+    """
+    io_bindings: dict = {}
+    rest_io_objects = {}
+    DIRTY_MODE_MAPPING = {
+        "Mount": InputDeliveryMode.READ_ONLY_MOUNT,
+        "RoMount": InputDeliveryMode.READ_ONLY_MOUNT,
+        "RwMount": InputDeliveryMode.READ_WRITE_MOUNT,
+    }
+    for key, val in io.items():
+        if isinstance(val, dict):
+            # convert the input of camel to snake to be compatible with the Jun api
+            # todo: backend help convert node level input/output type
+            normalize_job_input_output_type(val)
+
+            # Add casting as sometimes we got value like 1(int)
+            io_value = str(val.get("value", ""))
+            io_mode = val.get("mode", None)
+            io_name = val.get("name", None)
+            io_version = val.get("version", None)
+            if any(re.match(item, io_value) for item in io_binding_regex_list):
+                io_bindings.update({key: {"path": io_value}})
+                # add mode to literal value for binding input
+                if io_mode:
+                    # deal with dirty mode data submitted before
+                    if io_mode in DIRTY_MODE_MAPPING:
+                        io_mode = DIRTY_MODE_MAPPING[io_mode]
+                        val["mode"] = io_mode
+                    if io_mode in OUTPUT_MOUNT_MAPPING_FROM_REST:
+                        io_bindings[key].update({"mode": OUTPUT_MOUNT_MAPPING_FROM_REST[io_mode]})
+                    else:
+                        io_bindings[key].update({"mode": INPUT_MOUNT_MAPPING_FROM_REST[io_mode]})
+                # add name and version for binding input
+                if io_name or io_version:
+                    assert rest_object_class.__name__ == "JobOutput"
+                    # current code only support dump name and version for JobOutput
+                    # this assert can be deleted if we need to dump name/version for JobInput
+                    if io_name:
+                        io_bindings[key].update({"name": io_name})
+                    if io_version:
+                        io_bindings[key].update({"version": io_version})
+                if not io_mode and not io_name and not io_version:
+                    io_bindings[key] = io_value
+            else:
+                if rest_object_class.__name__ == "JobOutput":
+                    # current code only support dump name and version for JobOutput
+                    # this condition can be deleted if we need to dump name/version for JobInput
+                    if "name" in val.keys():
+                        val["asset_name"] = val.pop("name")
+                    if "version" in val.keys():
+                        val["asset_version"] = val.pop("version")
+                rest_obj = rest_object_class.from_dict(val)
+                rest_io_objects[key] = rest_obj
+        else:
+            msg = "Got unsupported type of input/output: {}:" + f"{type(val)}"
+            raise ValidationException(
+                message=msg.format(val),
+                no_personal_data_message=msg.format("[val]"),
+                target=ErrorTarget.PIPELINE,
+                error_category=ErrorCategory.USER_ERROR,
+            )
+    return io_bindings, rest_io_objects
+
+
+def from_dict_to_rest_distribution(distribution_dict: Dict) -> Union[PyTorch, Mpi, TensorFlow, Ray]:
+    target_type = distribution_dict["distribution_type"].lower()
+    if target_type == "pytorch":
+        return PyTorch(**distribution_dict)
+    if target_type == "mpi":
+        return Mpi(**distribution_dict)
+    if target_type == "tensorflow":
+        return TensorFlow(**distribution_dict)
+    if target_type == "ray":
+        return Ray(**distribution_dict)
+    msg = "Distribution type must be pytorch, mpi, tensorflow or ray: {}".format(target_type)
+    raise ValidationException(
+        message=msg,
+        no_personal_data_message=msg,
+        target=ErrorTarget.PIPELINE,
+        error_category=ErrorCategory.USER_ERROR,
+    )
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/data/expression_component_template.yml b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/data/expression_component_template.yml
new file mode 100644
index 00000000..10d391aa
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/data/expression_component_template.yml
@@ -0,0 +1,16 @@
+$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json
+type: command
+
+name: expression_component
+version: 1
+
+outputs:
+  output:
+    is_control: true
+
+code: ./src
+
+environment: azureml://registries/azureml/environments/mldesigner/labels/latest
+
+command: >-
+  {command_placeholder}
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job.py
new file mode 100644
index 00000000..7ddbbc46
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job.py
@@ -0,0 +1,711 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+import itertools
+import logging
+import typing
+from functools import partial
+from pathlib import Path
+from typing import Any, Dict, Generator, List, Optional, Union, cast
+
+from typing_extensions import Literal
+
+from azure.ai.ml._restclient.v2024_01_01_preview.models import JobBase
+from azure.ai.ml._restclient.v2024_01_01_preview.models import PipelineJob as RestPipelineJob
+from azure.ai.ml._schema import PathAwareSchema
+from azure.ai.ml._schema.pipeline.pipeline_job import PipelineJobSchema
+from azure.ai.ml._utils._arm_id_utils import get_resource_name_from_arm_id_safe
+from azure.ai.ml._utils.utils import (
+    camel_to_snake,
+    is_data_binding_expression,
+    is_private_preview_enabled,
+    transform_dict_keys,
+)
+from azure.ai.ml.constants import JobType
+from azure.ai.ml.constants._common import AZUREML_PRIVATE_FEATURES_ENV_VAR, BASE_PATH_CONTEXT_KEY
+from azure.ai.ml.constants._component import ComponentSource
+from azure.ai.ml.constants._job.pipeline import ValidationErrorCode
+from azure.ai.ml.entities._builders import BaseNode
+from azure.ai.ml.entities._builders.condition_node import ConditionNode
+from azure.ai.ml.entities._builders.control_flow_node import LoopNode
+from azure.ai.ml.entities._builders.import_node import Import
+from azure.ai.ml.entities._builders.parallel import Parallel
+from azure.ai.ml.entities._builders.pipeline import Pipeline
+from azure.ai.ml.entities._component.component import Component
+from azure.ai.ml.entities._component.pipeline_component import PipelineComponent
+
+# from azure.ai.ml.entities._job.identity import AmlToken, Identity, ManagedIdentity, UserIdentity
+from azure.ai.ml.entities._credentials import (
+    AmlTokenConfiguration,
+    ManagedIdentityConfiguration,
+    UserIdentityConfiguration,
+    _BaseJobIdentityConfiguration,
+)
+from azure.ai.ml.entities._inputs_outputs import Input, Output
+from azure.ai.ml.entities._inputs_outputs.group_input import GroupInput
+from azure.ai.ml.entities._job._input_output_helpers import (
+    from_rest_data_outputs,
+    from_rest_inputs_to_dataset_literal,
+    to_rest_data_outputs,
+    to_rest_dataset_literal_inputs,
+)
+from azure.ai.ml.entities._job.import_job import ImportJob
+from azure.ai.ml.entities._job.job import Job
+from azure.ai.ml.entities._job.job_service import JobServiceBase
+from azure.ai.ml.entities._job.pipeline._io import PipelineInput, PipelineJobIOMixin
+from azure.ai.ml.entities._job.pipeline.pipeline_job_settings import PipelineJobSettings
+from azure.ai.ml.entities._mixins import YamlTranslatableMixin
+from azure.ai.ml.entities._system_data import SystemData
+from azure.ai.ml.entities._validation import MutableValidationResult, PathAwareSchemaValidatableMixin
+from azure.ai.ml.exceptions import ErrorTarget, UserErrorException, ValidationException
+
+module_logger = logging.getLogger(__name__)
+
+
+class PipelineJob(Job, YamlTranslatableMixin, PipelineJobIOMixin, PathAwareSchemaValidatableMixin):
+    """Pipeline job.
+
+    You should not instantiate this class directly. Instead, you should
+    use the `@pipeline` decorator to create a `PipelineJob`.
+
+    :param component: Pipeline component version. The field is mutually exclusive with 'jobs'.
+    :type component: Union[str, ~azure.ai.ml.entities._component.pipeline_component.PipelineComponent]
+    :param inputs: Inputs to the pipeline job.
+    :type inputs: dict[str, Union[~azure.ai.ml.entities.Input, str, bool, int, float]]
+    :param outputs: Outputs of the pipeline job.
+    :type outputs: dict[str, ~azure.ai.ml.entities.Output]
+    :param name: Name of the PipelineJob. Defaults to None.
+    :type name: str
+    :param description: Description of the pipeline job. Defaults to None
+    :type description: str
+    :param display_name: Display name of the pipeline job. Defaults to None
+    :type display_name: str
+    :param experiment_name: Name of the experiment the job will be created under.
+        If None is provided, the experiment will be set to the current directory. Defaults to None
+    :type experiment_name: str
+    :param jobs: Pipeline component node name to component object. Defaults to None
+    :type jobs: dict[str, ~azure.ai.ml.entities._builders.BaseNode]
+    :param settings: Setting of the pipeline job. Defaults to None
+    :type settings: ~azure.ai.ml.entities.PipelineJobSettings
+    :param identity: Identity that the training job will use while running on compute. Defaults to None
+    :type identity: Union[
+        ~azure.ai.ml.entities._credentials.ManagedIdentityConfiguration,
+        ~azure.ai.ml.entities._credentials.AmlTokenConfiguration,
+        ~azure.ai.ml.entities._credentials.UserIdentityConfiguration
+
+    ]
+    :param compute: Compute target name of the built pipeline. Defaults to None
+    :type compute: str
+    :param tags: Tag dictionary. Tags can be added, removed, and updated. Defaults to None
+    :type tags: dict[str, str]
+    :param kwargs: A dictionary of additional configuration parameters. Defaults to None
+    :type kwargs: dict
+
+    .. admonition:: Example:
+
+        .. literalinclude:: ../samples/ml_samples_pipeline_job_configurations.py
+            :start-after: [START configure_pipeline_job_and_settings]
+            :end-before: [END configure_pipeline_job_and_settings]
+            :language: python
+            :dedent: 8
+            :caption: Shows how to create a pipeline using this class.
+    """
+
+    def __init__(
+        self,
+        *,
+        component: Optional[Union[str, PipelineComponent, Component]] = None,
+        inputs: Optional[Dict[str, Union[Input, str, bool, int, float]]] = None,
+        outputs: Optional[Dict[str, Output]] = None,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        display_name: Optional[str] = None,
+        experiment_name: Optional[str] = None,
+        jobs: Optional[Dict[str, BaseNode]] = None,
+        settings: Optional[PipelineJobSettings] = None,
+        identity: Optional[
+            Union[ManagedIdentityConfiguration, AmlTokenConfiguration, UserIdentityConfiguration]
+        ] = None,
+        compute: Optional[str] = None,
+        tags: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> None:
+        # initialize io
+        inputs, outputs = inputs or {}, outputs or {}
+        if isinstance(component, PipelineComponent) and component._source in [
+            ComponentSource.DSL,
+            ComponentSource.YAML_COMPONENT,
+        ]:
+            self._inputs = self._build_inputs_dict(inputs, input_definition_dict=component.inputs)
+            # for pipeline component created pipeline jobs,
+            # it's output should have same value with the component outputs,
+            # then override it with given outputs (filter out None value)
+            pipeline_outputs = {k: v for k, v in (outputs or {}).items() if v}
+            self._outputs = self._build_pipeline_outputs_dict({**component.outputs, **pipeline_outputs})
+        else:
+            # Build inputs/outputs dict without meta when definition not available
+            self._inputs = self._build_inputs_dict(inputs)
+            # for node created pipeline jobs,
+            # it's output should have same value with the given outputs
+            self._outputs = self._build_pipeline_outputs_dict(outputs=outputs)
+        source = kwargs.pop("_source", ComponentSource.CLASS)
+        if component is None:
+            component = PipelineComponent(
+                jobs=jobs,
+                description=description,
+                display_name=display_name,
+                base_path=kwargs.get(BASE_PATH_CONTEXT_KEY),
+                _source=source,
+            )
+
+        # If component is Pipeline component, jobs will be component.jobs
+        self._jobs = (jobs or {}) if isinstance(component, str) else {}
+
+        self.component: Union[PipelineComponent, str] = cast(Union[PipelineComponent, str], component)
+        if "type" not in kwargs:
+            kwargs["type"] = JobType.PIPELINE
+        if isinstance(component, PipelineComponent):
+            description = component.description if description is None else description
+            display_name = component.display_name if display_name is None else display_name
+        super(PipelineJob, self).__init__(
+            name=name,
+            description=description,
+            tags=tags,
+            display_name=display_name,
+            experiment_name=experiment_name,
+            compute=compute,
+            **kwargs,
+        )
+
+        self._remove_pipeline_input()
+        self.compute = compute
+        self._settings: Any = None
+        self.settings = settings
+        self.identity = identity
+        # TODO: remove default code & environment?
+        self._default_code = None
+        self._default_environment = None
+
+    @property
+    def inputs(self) -> Dict:
+        """Inputs of the pipeline job.
+
+        :return: Inputs of the pipeline job.
+        :rtype: dict[str, Union[~azure.ai.ml.entities.Input, str, bool, int, float]]
+        """
+        return self._inputs
+
+    @property
+    def outputs(self) -> Dict[str, Union[str, Output]]:
+        """Outputs of the pipeline job.
+
+        :return: Outputs of the pipeline job.
+        :rtype: dict[str, Union[str, ~azure.ai.ml.entities.Output]]
+        """
+        return self._outputs
+
+    @property
+    def jobs(self) -> Dict:
+        """Return jobs of pipeline job.
+
+        :return: Jobs of pipeline job.
+        :rtype: dict
+        """
+        res: dict = self.component.jobs if isinstance(self.component, PipelineComponent) else self._jobs
+        return res
+
+    @property
+    def settings(self) -> Optional[PipelineJobSettings]:
+        """Settings of the pipeline job.
+
+        :return: Settings of the pipeline job.
+        :rtype: ~azure.ai.ml.entities.PipelineJobSettings
+        """
+        if self._settings is None:
+            self._settings = PipelineJobSettings()
+        res: Optional[PipelineJobSettings] = self._settings
+        return res
+
+    @settings.setter
+    def settings(self, value: Union[Dict, PipelineJobSettings]) -> None:
+        """Set the pipeline job settings.
+
+        :param value: The pipeline job settings.
+        :type value: Union[dict, ~azure.ai.ml.entities.PipelineJobSettings]
+        """
+        if value is not None:
+            if isinstance(value, PipelineJobSettings):
+                # since PipelineJobSettings inherit _AttrDict, we need add this branch to distinguish with dict
+                pass
+            elif isinstance(value, dict):
+                value = PipelineJobSettings(**value)
+            else:
+                raise TypeError("settings must be PipelineJobSettings or dict but got {}".format(type(value)))
+        self._settings = value
+
+    @classmethod
+    def _create_validation_error(cls, message: str, no_personal_data_message: str) -> ValidationException:
+        return ValidationException(
+            message=message,
+            no_personal_data_message=no_personal_data_message,
+            target=ErrorTarget.PIPELINE,
+        )
+
+    @classmethod
+    def _create_schema_for_validation(cls, context: Any) -> PathAwareSchema:
+        # import this to ensure that nodes are registered before schema is created.
+
+        return PipelineJobSchema(context=context)
+
+    @classmethod
+    def _get_skip_fields_in_schema_validation(cls) -> typing.List[str]:
+        # jobs validations are done in _customized_validate()
+        return ["component", "jobs"]
+
+    @property
+    def _skip_required_compute_missing_validation(self) -> Literal[True]:
+        return True
+
+    def _validate_compute_is_set(self) -> MutableValidationResult:
+        validation_result = self._create_empty_validation_result()
+        if self.compute is not None:
+            return validation_result
+        if self.settings is not None and self.settings.default_compute is not None:
+            return validation_result
+
+        if not isinstance(self.component, str):
+            validation_result.merge_with(self.component._validate_compute_is_set())
+        return validation_result
+
+    def _customized_validate(self) -> MutableValidationResult:
+        """Validate that all provided inputs and parameters are valid for current pipeline and components in it.
+
+        :return: The validation result
+        :rtype: MutableValidationResult
+        """
+        validation_result = super(PipelineJob, self)._customized_validate()
+
+        if isinstance(self.component, PipelineComponent):
+            # Merge with pipeline component validate result for structure validation.
+            # Skip top level parameter missing type error
+            validation_result.merge_with(
+                self.component._customized_validate(),
+                condition_skip=lambda x: x.error_code == ValidationErrorCode.PARAMETER_TYPE_UNKNOWN
+                and x.yaml_path.startswith("inputs"),
+            )
+            # Validate compute
+            validation_result.merge_with(self._validate_compute_is_set())
+        # Validate Input
+        validation_result.merge_with(self._validate_input())
+        # Validate initialization & finalization jobs
+        validation_result.merge_with(self._validate_init_finalize_job())
+
+        return validation_result
+
+    def _validate_input(self) -> MutableValidationResult:
+        validation_result = self._create_empty_validation_result()
+        if not isinstance(self.component, str):
+            # TODO(1979547): refine this logic: not all nodes have `_get_input_binding_dict` method
+            used_pipeline_inputs = set(
+                itertools.chain(
+                    *[
+                        self.component._get_input_binding_dict(node if not isinstance(node, LoopNode) else node.body)[0]
+                        for node in self.jobs.values()
+                        if not isinstance(node, ConditionNode)
+                        # condition node has no inputs
+                    ]
+                )
+            )
+        # validate inputs
+        if not isinstance(self.component, Component):
+            return validation_result
+        for key, meta in self.component.inputs.items():
+            if key not in used_pipeline_inputs:  # pylint: disable=possibly-used-before-assignment
+                # Only validate inputs certainly used.
+                continue
+            # raise error when required input with no default value not set
+            if (
+                self.inputs.get(key, None) is None  # input not provided
+                and meta.optional is not True  # and it's required
+                and meta.default is None  # and it does not have default
+            ):
+                name = self.name or self.display_name
+                name = f"{name!r} " if name else ""
+                validation_result.append_error(
+                    yaml_path=f"inputs.{key}",
+                    message=f"Required input {key!r} for pipeline {name}not provided.",
+                )
+        return validation_result
+
+    def _validate_init_finalize_job(self) -> MutableValidationResult:  # pylint: disable=too-many-statements
+        from azure.ai.ml.entities._job.pipeline._io import InputOutputBase, _GroupAttrDict
+
+        validation_result = self._create_empty_validation_result()
+        # subgraph (PipelineComponent) should not have on_init/on_finalize set
+        for job_name, job in self.jobs.items():
+            if job.type != "pipeline":
+                continue
+            if job.settings.on_init:
+                validation_result.append_error(
+                    yaml_path=f"jobs.{job_name}.settings.on_init",
+                    message="On_init is not supported for pipeline component.",
+                )
+            if job.settings.on_finalize:
+                validation_result.append_error(
+                    yaml_path=f"jobs.{job_name}.settings.on_finalize",
+                    message="On_finalize is not supported for pipeline component.",
+                )
+
+        on_init = None
+        on_finalize = None
+
+        if self.settings is not None:
+            # quick return if neither on_init nor on_finalize is set
+            if self.settings.on_init is None and self.settings.on_finalize is None:
+                return validation_result
+
+            on_init, on_finalize = self.settings.on_init, self.settings.on_finalize
+
+        append_on_init_error = partial(validation_result.append_error, "settings.on_init")
+        append_on_finalize_error = partial(validation_result.append_error, "settings.on_finalize")
+        # on_init and on_finalize cannot be same
+        if on_init == on_finalize:
+            append_on_init_error(f"Invalid on_init job {on_init}, it should be different from on_finalize.")
+            append_on_finalize_error(f"Invalid on_finalize job {on_finalize}, it should be different from on_init.")
+        # pipeline should have at least one normal node
+        if len(set(self.jobs.keys()) - {on_init, on_finalize}) == 0:
+            validation_result.append_error(yaml_path="jobs", message="No other job except for on_init/on_finalize job.")
+
+        def _is_control_flow_node(_validate_job_name: str) -> bool:
+            from azure.ai.ml.entities._builders.control_flow_node import ControlFlowNode
+
+            _validate_job = self.jobs[_validate_job_name]
+            return issubclass(type(_validate_job), ControlFlowNode)
+
+        def _is_isolated_job(_validate_job_name: str) -> bool:
+            def _try_get_data_bindings(
+                _name: str, _input_output_data: Union["_GroupAttrDict", "InputOutputBase"]
+            ) -> Optional[List]:
+                """Try to get data bindings from input/output data, return None if not found.
+                :param _name: The name to use when flattening GroupAttrDict
+                :type _name: str
+                :param _input_output_data: The input/output data
+                :type _input_output_data: Union[_GroupAttrDict, str, InputOutputBase]
+                :return: A list of data bindings, or None if not found
+                :rtype: Optional[List[str]]
+                """
+                # handle group input
+                if GroupInput._is_group_attr_dict(_input_output_data):
+                    _new_input_output_data: _GroupAttrDict = cast(_GroupAttrDict, _input_output_data)
+                    # flatten to avoid nested cases
+                    flattened_values: List[Input] = list(_new_input_output_data.flatten(_name).values())
+                    # handle invalid empty group
+                    if len(flattened_values) == 0:
+                        return None
+                    return [_value.path for _value in flattened_values]
+                _input_output_data = _input_output_data._data
+                if isinstance(_input_output_data, str):
+                    return [_input_output_data]
+                if not hasattr(_input_output_data, "_data_binding"):
+                    return None
+                return [_input_output_data._data_binding()]
+
+            _validate_job = self.jobs[_validate_job_name]
+            # no input to validate job
+            for _input_name in _validate_job.inputs:
+                _data_bindings = _try_get_data_bindings(_input_name, _validate_job.inputs[_input_name])
+                if _data_bindings is None:
+                    continue
+                for _data_binding in _data_bindings:
+                    if is_data_binding_expression(_data_binding, ["parent", "jobs"]):
+                        return False
+            # no output from validate job - iterate other jobs input(s) to validate
+            for _job_name, _job in self.jobs.items():
+                # exclude control flow node as it does not have inputs
+                if _is_control_flow_node(_job_name):
+                    continue
+                for _input_name in _job.inputs:
+                    _data_bindings = _try_get_data_bindings(_input_name, _job.inputs[_input_name])
+                    if _data_bindings is None:
+                        continue
+                    for _data_binding in _data_bindings:
+                        if is_data_binding_expression(_data_binding, ["parent", "jobs", _validate_job_name]):
+                            return False
+            return True
+
+        # validate on_init
+        if on_init is not None:
+            if on_init not in self.jobs:
+                append_on_init_error(f"On_init job name {on_init} not exists in jobs.")
+            else:
+                if _is_control_flow_node(on_init):
+                    append_on_init_error("On_init job should not be a control flow node.")
+                elif not _is_isolated_job(on_init):
+                    append_on_init_error("On_init job should not have connection to other execution node.")
+        # validate on_finalize
+        if on_finalize is not None:
+            if on_finalize not in self.jobs:
+                append_on_finalize_error(f"On_finalize job name {on_finalize} not exists in jobs.")
+            else:
+                if _is_control_flow_node(on_finalize):
+                    append_on_finalize_error("On_finalize job should not be a control flow node.")
+                elif not _is_isolated_job(on_finalize):
+                    append_on_finalize_error("On_finalize job should not have connection to other execution node.")
+        return validation_result
+
+    def _remove_pipeline_input(self) -> None:
+        """Remove None pipeline input.If not remove, it will pass "None" to backend."""
+        redundant_pipeline_inputs = []
+        for pipeline_input_name, pipeline_input in self._inputs.items():
+            if isinstance(pipeline_input, PipelineInput) and pipeline_input._data is None:
+                redundant_pipeline_inputs.append(pipeline_input_name)
+        for redundant_pipeline_input in redundant_pipeline_inputs:
+            self._inputs.pop(redundant_pipeline_input)
+
+    def _check_private_preview_features(self) -> None:
+        """Checks is private preview features included in pipeline.
+
+        If private preview environment not set, raise exception.
+        """
+        if not is_private_preview_enabled():
+            error_msg = (
+                "{} is a private preview feature, "
+                f"please set environment variable {AZUREML_PRIVATE_FEATURES_ENV_VAR} to true to use it."
+            )
+            # check has not supported nodes
+            for _, node in self.jobs.items():
+                # TODO: Remove in PuP
+                if isinstance(node, (ImportJob, Import)):
+                    msg = error_msg.format("Import job in pipeline")
+                    raise UserErrorException(message=msg, no_personal_data_message=msg)
+
+    def _to_node(self, context: Optional[Dict] = None, **kwargs: Any) -> "Pipeline":
+        """Translate a command job to a pipeline node when load schema.
+
+        (Write a pipeline job as node in yaml is not supported presently.)
+
+        :param context: Context of command job YAML file.
+        :type context: dict
+        :return: Translated command component.
+        :rtype: Pipeline
+        """
+        component = self._to_component(context, **kwargs)
+
+        return Pipeline(
+            component=component,
+            compute=self.compute,
+            # Need to supply the inputs with double curly.
+            inputs=self.inputs,
+            outputs=self.outputs,
+            description=self.description,
+            tags=self.tags,
+            display_name=self.display_name,
+            properties=self.properties,
+        )
+
+    def _to_rest_object(self) -> JobBase:
+        """Build current parameterized pipeline instance to a pipeline job object before submission.
+
+        :return: Rest pipeline job.
+        :rtype: JobBase
+        """
+        # Check if there are private preview features in it
+        self._check_private_preview_features()
+
+        # Build the inputs to dict. Handle both value & binding assignment.
+        # Example: {
+        #   "input_data": {"data": {"path": "path/to/input/data"},  "mode"="Mount"},
+        #   "input_value": 10,
+        #   "learning_rate": "${{jobs.step1.inputs.learning_rate}}"
+        # }
+        built_inputs = self._build_inputs()
+
+        # Build the outputs to dict
+        # example: {"eval_output": "${{jobs.eval.outputs.eval_output}}"}
+        built_outputs = self._build_outputs()
+
+        if self.settings is not None:
+            settings_dict = self.settings._to_dict()
+
+        if isinstance(self.component, PipelineComponent):
+            source = self.component._source
+            # Build the jobs to dict
+            rest_component_jobs = self.component._build_rest_component_jobs()
+        else:
+            source = ComponentSource.REMOTE_WORKSPACE_JOB
+            rest_component_jobs = {}
+        # add _source on pipeline job.settings
+        if "_source" not in settings_dict:  # pylint: disable=possibly-used-before-assignment
+            settings_dict.update({"_source": source})
+
+        # TODO: Revisit this logic when multiple types of component jobs are supported
+        rest_compute = self.compute
+        # This will be resolved in job_operations _resolve_arm_id_or_upload_dependencies.
+        component_id = self.component if isinstance(self.component, str) else self.component.id
+
+        # TODO remove it in the future.
+        # MFE not support pass None or empty input value. Remove the empty inputs in pipeline job.
+        built_inputs = {k: v for k, v in built_inputs.items() if v is not None and v != ""}
+
+        pipeline_job = RestPipelineJob(
+            compute_id=rest_compute,
+            component_id=component_id,
+            display_name=self.display_name,
+            tags=self.tags,
+            description=self.description,
+            properties=self.properties,
+            experiment_name=self.experiment_name,
+            jobs=rest_component_jobs,
+            inputs=to_rest_dataset_literal_inputs(built_inputs, job_type=self.type),
+            outputs=to_rest_data_outputs(built_outputs),
+            settings=settings_dict,
+            services={k: v._to_rest_object() for k, v in self.services.items()} if self.services else None,
+            identity=self.identity._to_job_rest_object() if self.identity else None,
+        )
+
+        rest_job = JobBase(properties=pipeline_job)
+        rest_job.name = self.name
+        return rest_job
+
+    @classmethod
+    def _load_from_rest(cls, obj: JobBase) -> "PipelineJob":
+        """Build a pipeline instance from rest pipeline object.
+
+        :param obj: The REST Pipeline Object
+        :type obj: JobBase
+        :return: pipeline job.
+        :rtype: PipelineJob
+        """
+        properties: RestPipelineJob = obj.properties
+        # Workaround for BatchEndpoint as these fields are not filled in
+        # Unpack the inputs
+        from_rest_inputs = from_rest_inputs_to_dataset_literal(properties.inputs) or {}
+        from_rest_outputs = from_rest_data_outputs(properties.outputs) or {}
+        # Unpack the component jobs
+        sub_nodes = PipelineComponent._resolve_sub_nodes(properties.jobs) if properties.jobs else {}
+        # backend may still store Camel settings, eg: DefaultDatastore, translate them to snake when load back
+        settings_dict = transform_dict_keys(properties.settings, camel_to_snake) if properties.settings else None
+        settings_sdk = PipelineJobSettings(**settings_dict) if settings_dict else PipelineJobSettings()
+        # Create component or use component id
+        if getattr(properties, "component_id", None):
+            component = properties.component_id
+        else:
+            component = PipelineComponent._load_from_rest_pipeline_job(
+                {
+                    "inputs": from_rest_inputs,
+                    "outputs": from_rest_outputs,
+                    "display_name": properties.display_name,
+                    "description": properties.description,
+                    "jobs": sub_nodes,
+                }
+            )
+
+        job = PipelineJob(
+            component=component,
+            inputs=from_rest_inputs,
+            outputs=from_rest_outputs,
+            name=obj.name,
+            id=obj.id,
+            jobs=sub_nodes,
+            display_name=properties.display_name,
+            tags=properties.tags,
+            properties=properties.properties,
+            experiment_name=properties.experiment_name,
+            status=properties.status,
+            creation_context=SystemData._from_rest_object(obj.system_data) if obj.system_data else None,
+            services=JobServiceBase._from_rest_job_services(properties.services) if properties.services else None,
+            compute=get_resource_name_from_arm_id_safe(properties.compute_id),
+            settings=settings_sdk,
+            identity=(
+                _BaseJobIdentityConfiguration._from_rest_object(properties.identity) if properties.identity else None
+            ),
+        )
+
+        return job
+
+    def _to_dict(self) -> Dict:
+        res: dict = self._dump_for_validation()
+        return res
+
+    @classmethod
+    def _component_items_from_path(cls, data: Dict) -> Generator:
+        if "jobs" in data:
+            for node_name, job_instance in data["jobs"].items():
+                potential_component_path = job_instance["component"] if "component" in job_instance else None
+                if isinstance(potential_component_path, str) and potential_component_path.startswith("file:"):
+                    yield node_name, potential_component_path
+
+    @classmethod
+    def _load_from_dict(cls, data: Dict, context: Dict, additional_message: str, **kwargs: Any) -> "PipelineJob":
+        path_first_occurrence: dict = {}
+        component_first_occurrence = {}
+        for node_name, component_path in cls._component_items_from_path(data):
+            if component_path in path_first_occurrence:
+                component_first_occurrence[node_name] = path_first_occurrence[component_path]
+                # set components to be replaced here may break the validation logic
+            else:
+                path_first_occurrence[component_path] = node_name
+
+        # use this instead of azure.ai.ml.entities._util.load_from_dict to avoid parsing
+        loaded_schema = cls._create_schema_for_validation(context=context).load(data, **kwargs)
+
+        # replace repeat component with first occurrence to reduce arm id resolution
+        # current load yaml file logic is in azure.ai.ml._schema.core.schema.YamlFileSchema.load_from_file
+        # is it possible to load the same yaml file only once in 1 pipeline loading?
+        for node_name, first_occurrence in component_first_occurrence.items():
+            job = loaded_schema["jobs"][node_name]
+            job._component = loaded_schema["jobs"][first_occurrence].component
+            # For Parallel job, should also align task attribute which is usually from component.task
+            if isinstance(job, Parallel):
+                job.task = job._component.task
+                # parallel.task.code is based on parallel._component.base_path, so need to update it
+                job._base_path = job._component.base_path
+        return PipelineJob(
+            base_path=context[BASE_PATH_CONTEXT_KEY],
+            _source=ComponentSource.YAML_JOB,
+            **loaded_schema,
+        )
+
+    def __str__(self) -> str:
+        try:
+            res_to_yaml: str = self._to_yaml()
+            return res_to_yaml
+        except BaseException:  # pylint: disable=W0718
+            res: str = super(PipelineJob, self).__str__()
+            return res
+
+    def _get_telemetry_values(self) -> Dict:
+        telemetry_values: dict = super()._get_telemetry_values()
+        if isinstance(self.component, PipelineComponent):
+            telemetry_values.update(self.component._get_telemetry_values())
+        else:
+            telemetry_values.update({"source": ComponentSource.REMOTE_WORKSPACE_JOB})
+        telemetry_values.pop("is_anonymous")
+        return telemetry_values
+
+    def _to_component(self, context: Optional[Dict] = None, **kwargs: Any) -> "PipelineComponent":
+        """Translate a pipeline job to pipeline component.
+
+        :param context: Context of pipeline job YAML file.
+        :type context: dict
+        :return: Translated pipeline component.
+        :rtype: PipelineComponent
+        """
+        ignored_keys = PipelineComponent._check_ignored_keys(self)
+        if ignored_keys:
+            name = self.name or self.display_name
+            name = f"{name!r} " if name else ""
+            module_logger.warning("%s ignored when translating PipelineJob %sto PipelineComponent.", ignored_keys, name)
+        pipeline_job_dict = kwargs.get("pipeline_job_dict", {})
+        context = context or {BASE_PATH_CONTEXT_KEY: Path("./")}
+
+        # Create anonymous pipeline component with default version as 1
+        return PipelineComponent(
+            base_path=context[BASE_PATH_CONTEXT_KEY],
+            display_name=self.display_name,
+            inputs=self._to_inputs(inputs=self.inputs, pipeline_job_dict=pipeline_job_dict),
+            outputs=self._to_outputs(outputs=self.outputs, pipeline_job_dict=pipeline_job_dict),
+            jobs=self.jobs,
+        )
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job_settings.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job_settings.py
new file mode 100644
index 00000000..0fe41e2e
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_job/pipeline/pipeline_job_settings.py
@@ -0,0 +1,75 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+from typing import Any, Dict, Generator, Optional
+
+from azure.ai.ml.entities._job.pipeline._attr_dict import _AttrDict
+
+
+class PipelineJobSettings(_AttrDict):
+    """Settings of PipelineJob.
+
+    :param default_datastore: The default datastore of the pipeline.
+    :type default_datastore: str
+    :param default_compute: The default compute target of the pipeline.
+    :type default_compute: str
+    :param continue_on_step_failure: Flag indicating whether to continue pipeline execution if a step fails.
+    :type continue_on_step_failure: bool
+    :param force_rerun: Flag indicating whether to force rerun pipeline execution.
+    :type force_rerun: bool
+
+    .. admonition:: Example:
+
+        .. literalinclude:: ../samples/ml_samples_pipeline_job_configurations.py
+            :start-after: [START configure_pipeline_job_and_settings]
+            :end-before: [END configure_pipeline_job_and_settings]
+            :language: python
+            :dedent: 8
+            :caption: Shows how to set pipeline properties using this class.
+    """
+
+    def __init__(
+        self,
+        default_datastore: Optional[str] = None,
+        default_compute: Optional[str] = None,
+        continue_on_step_failure: Optional[bool] = None,
+        force_rerun: Optional[bool] = None,
+        **kwargs: Any
+    ) -> None:
+        self._init = True
+        super().__init__()
+        self.default_compute: Any = default_compute
+        self.default_datastore: Any = default_datastore
+        self.continue_on_step_failure = continue_on_step_failure
+        self.force_rerun = force_rerun
+        self.on_init = kwargs.get("on_init", None)
+        self.on_finalize = kwargs.get("on_finalize", None)
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+        self._init = False
+
+    def _get_valid_keys(self) -> Generator[str, Any, None]:
+        for k, v in self.__dict__.items():
+            if v is None:
+                continue
+            # skip private attributes inherited from _AttrDict
+            if k in ["_logger", "_allowed_keys", "_init", "_key_restriction"]:
+                continue
+            yield k
+
+    def _to_dict(self) -> Dict:
+        result = {}
+        for k in self._get_valid_keys():
+            result[k] = self.__dict__[k]
+        result.update(self._get_attrs())
+        return result
+
+    def _initializing(self) -> bool:
+        return self._init
+
+    def __bool__(self) -> bool:
+        for _ in self._get_valid_keys():
+            return True
+        # _attr_dict will return False if no extra attributes are set
+        return self.__len__() > 0