[ml][pipeline] Fix pipeline job outputs not load correctly when referring local component (#34026)

* Fix pipeline job outputs not load correctly when local component
reference exists

Signed-off-by: Brynn Yin <[email protected]>

* Filter None value out from pipeline output

Signed-off-by: Brynn Yin <[email protected]>

---------

Signed-off-by: Brynn Yin <[email protected]>

Author: brynn-code

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -6,6 +6,7 @@
 - Remove `experimental` tag for  `ml_client.jobs.validate`.
 
 ### Bugs Fixed
+- Fix pipeline job `outputs` not load correctly when `component: <local-file>` exists in pipeline job yaml.
 
 ### Breaking Changes
 

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

@@ -140,8 +140,10 @@ def __init__(
         ]:
             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
-            self._outputs = self._build_pipeline_outputs_dict(component.outputs)
+            # 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)
@@ -228,6 +230,11 @@ def settings(self) -> Optional[PipelineJobSettings]:
 
     @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

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

@@ -28,12 +28,12 @@
 from azure.ai.ml.entities._assets import Code
 from azure.ai.ml.entities._component.datatransfer_component import DataTransferComponent
 from azure.ai.ml.entities._component.parallel_component import ParallelComponent
+from azure.ai.ml.entities._credentials import UserIdentityConfiguration
 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,
     validate_pipeline_input_key_characters,
 )
-from azure.ai.ml.entities._credentials import UserIdentityConfiguration
 from azure.ai.ml.entities._job.automl.search_space_utils import _convert_sweep_dist_dict_to_str_dict
 from azure.ai.ml.entities._job.job_service import (
     JobService,
@@ -1841,6 +1841,25 @@ def test_pipeline_component_job(self):
         assert "jobs" in job_dict
         assert "component_a_job" in job_dict["jobs"]
 
+        test_path = "./tests/test_configs/pipeline_jobs/pipeline_component_job_with_overrides.yml"
+        job: PipelineJob = load_job(source=test_path)
+        assert job._validate().passed
+        job_dict = job._to_dict()
+        assert job_dict["outputs"] == {
+            "not_exists": {"path": "azureml://datastores/mock/paths/not_exists.txt", "type": "uri_file"},
+            "output_path": {"path": "azureml://datastores/mock/paths/my_output_file.txt", "type": "uri_file"},
+        }
+
+        # Assert the output_path:None not override original component output value
+        test_path = "./tests/test_configs/pipeline_jobs/pipeline_component_job_with_overrides2.yml"
+        job: PipelineJob = load_job(source=test_path)
+        assert job._validate().passed
+        job_dict = job._to_dict()
+        assert job_dict["outputs"] == {
+            "not_exists": {"path": "azureml://datastores/mock/paths/not_exists.txt", "type": "uri_file"},
+            "output_path": {"type": "uri_folder"},  # uri_folder from component output
+        }
+
     def test_invalid_pipeline_component_job(self):
         test_path = "./tests/test_configs/pipeline_jobs/invalid/invalid_pipeline_component_job.yml"
         with pytest.raises(Exception) as e:

sdk/ml/azure-ai-ml/tests/test_configs/pipeline_jobs/pipeline_component_job_with_overrides.yml

@@ -0,0 +1,27 @@
+type: pipeline
+
+description: The hello world pipeline job
+tags:
+  tag: tagvalue
+  owner: sdkteam
+
+settings:
+  default_compute: azureml:cpu-cluster
+
+inputs:
+  # examples of inputs that take values such as int, string, etc.
+  component_in_number: 20
+  component_in_path:
+    path: https://dprepdata.blob.core.windows.net/demo/Titanic.csv
+    type: uri_file
+
+
+outputs:
+  output_path:
+    type: uri_file
+    path: azureml://datastores/mock/paths/my_output_file.txt
+  not_exists:
+    type: uri_file
+    path: azureml://datastores/mock/paths/not_exists.txt
+
+component: ../components/helloworld_pipeline_component.yml

sdk/ml/azure-ai-ml/tests/test_configs/pipeline_jobs/pipeline_component_job_with_overrides2.yml

@@ -0,0 +1,25 @@
+type: pipeline
+
+description: The hello world pipeline job
+tags:
+  tag: tagvalue
+  owner: sdkteam
+
+settings:
+  default_compute: azureml:cpu-cluster
+
+inputs:
+  # examples of inputs that take values such as int, string, etc.
+  component_in_number: 20
+  component_in_path:
+    path: https://dprepdata.blob.core.windows.net/demo/Titanic.csv
+    type: uri_file
+
+
+outputs:
+  output_path:
+  not_exists:
+    type: uri_file
+    path: azureml://datastores/mock/paths/not_exists.txt
+
+component: ../components/helloworld_pipeline_component.yml