Support remaining issues in register node output (Azure#28859)

* add name and version for output

* add asset_name, asset_version, merge 2022-12-01 pr

* resolve comments

* add test and modify related codes

* refine code

* fix failed test

* fix comments

* fix condition._port_name

* fix test and some comments

* revert the modification of compute

* fix comments

* fix pylint error, add tests for CommandJob, Spark, Parallel, Sweep

* delete one test, will add in another pr

* modify error message

* support Output() for origin node

* delete useless change

* modify test, update recording

* fix test_register_output_sdk

* modify test_register_output_sdk_succeed, add content of another pr

* run black with 22.3.0

---------

Co-authored-by: chenyangliao <chenyangliao@microsoft.com>

Author: lllilithyang

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/attr_dict.py

@@ -160,7 +160,7 @@ def __getitem__(self, item) -> NodeOutput:
     def __setattr__(self, key: str, value: Union[Data, Output]):
         if isinstance(value, Output):
             mode = value.mode
-            value = Output(type=value.type, path=value.path, mode=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)
 

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/base.py

@@ -384,8 +384,8 @@ def __init__(
         super().__init__(meta=meta, data=data, **kwargs)
         self._port_name = port_name
         self._owner = owner
-        self._name = data.name if isinstance(data, Output) else None
-        self._version = data.version if isinstance(data, Output) else None
+        self._name = self._data.name if isinstance(self._data, Output) else None
+        self._version = self._data.version if isinstance(self._data, Output) else None
 
         self._assert_name_and_version()
 
@@ -510,8 +510,8 @@ def _to_job_output(self):
                 path=self._data._data_binding(),
                 mode=self.mode,
                 is_control=is_control,
-                name=self.name,
-                version=self.version,
+                name=self._data.name,
+                version=self._data.version,
                 description=self.description,
             )
         else:

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/mixin.py

@@ -207,6 +207,7 @@ def _to_rest_outputs(self) -> Dict[str, Dict]:
                 rest_output_bindings[key].update({"version": binding["version"]})
 
         def _rename_name_and_version(output_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():

sdk/ml/azure-ai-ml/tests/dsl/e2etests/test_dsl_pipeline.py

@@ -59,6 +59,11 @@
 ]
 
 
+def check_name_and_version(output, output_name, output_version):
+    assert output.name == output_name
+    assert output.version == output_version
+
+
 @pytest.mark.usefixtures(
     "enable_environment_id_arm_expansion",
     "enable_pipeline_private_preview_features",
@@ -2692,8 +2697,8 @@ def register_both_output():
         assert pipeline_output.name == "b_output"
         assert pipeline_output.version == "2"
         node_output = pipeline_job.jobs["node"].outputs.component_out_path
-        assert node_output.name == "a_output"
-        assert node_output.version == "1"
+        assert node_output.name == None
+        assert node_output.version == None
 
     def test_dsl_pipeline_with_data_transfer_copy_2urifolder(self, client: MLClient) -> None:
         from test_configs.dsl_pipeline.data_transfer_job_in_pipeline.copy_data.pipeline import (
@@ -2833,6 +2838,10 @@ def register_both_output():
             node_2.outputs.component_out_path.name = "n2_output"
             node_2.outputs.component_out_path.version = "v1"
 
+            # register NodeOutput without version, in this case the run result can be reused
+            node_3 = component(component_in_path=component_input)
+            node_3.outputs.component_out_path.name = "n3_output"
+
             # register NodeOutput of subgraph
             sub_node = sub_pipeline()
             sub_node.outputs.sub_pipeine_a_output.name = "sub_pipeline"
@@ -2844,18 +2853,90 @@ def register_both_output():
         pipeline.outputs.pipeine_a_output.name = "p1_output"
         pipeline.outputs.pipeine_a_output.version = "v1"
         pipeline.settings.default_compute = "cpu-cluster"
-        pipeline_job = client.jobs.create_or_update(pipeline)
-        client.jobs.stream(pipeline_job.name)
+        pipeline_job = assert_job_cancel(pipeline, client)
 
-        def check_name_version_and_register_succeed(output, output_name, output_version):
-            assert output.name == output_name
-            assert output.version == output_version
-            assert client.data.get(name=output_name, version=output_version)
+        check_name_and_version(pipeline_job.outputs.pipeine_a_output, "p1_output", "v1")
+        check_name_and_version(pipeline_job.jobs["node_2"].outputs.component_out_path, "n2_output", "v1")
+        assert pipeline_job.jobs["node_3"].outputs.component_out_path.name == "n3_output"
+        check_name_and_version(pipeline_job.jobs["sub_node"].outputs.sub_pipeine_a_output, "sub_pipeline", "v1")
 
-        check_name_version_and_register_succeed(pipeline_job.outputs.pipeine_a_output, "p1_output", "v1")
-        check_name_version_and_register_succeed(
-            pipeline_job.jobs["node_2"].outputs.component_out_path, "n2_output", "v1"
-        )
-        check_name_version_and_register_succeed(
-            pipeline_job.jobs["sub_node"].outputs.sub_pipeine_a_output, "sub_pipeline", "v1"
-        )
+    @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skip(reason="TODO (269646): Internal Server Error")
+    def test_register_output_for_pipeline_component(self, client: MLClient):
+        component = load_component(source="./tests/test_configs/components/helloworld_component.yml")
+        component_input = Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv")
+
+        @dsl.pipeline()
+        def sub_pipeline():
+            node_1 = component(component_in_path=component_input)  # test use Output to initialize subgraph.jobs.output
+            node_1.outputs.component_out_path = Output(name="sub_pipeline_1_output", version="v1")
+
+            node_2 = component(component_in_path=component_input)  # test we can pass NodeOutput in PipelineComponent
+            node_2.outputs.component_out_path.name = "sub_pipeline_2_output"
+            node_2.outputs.component_out_path.version = "v2"
+
+            return {
+                "sub_node_1": node_1.outputs.component_out_path,
+            }
+
+        @dsl.pipeline()
+        def register_both_output():
+            subgraph = sub_pipeline()
+
+        pipeline = register_both_output()
+        pipeline.settings.default_compute = "cpu-cluster"
+        pipeline_job = assert_job_cancel(pipeline, client)
+
+        check_name_and_version(pipeline_job.jobs["subgraph"].outputs["sub_node_1"], "sub_pipeline_1_output", "v1")
+
+        subgraph_id = pipeline_job.jobs["subgraph"].component
+        subgraph_id = subgraph_id.split(":")
+        subgraph = client.components.get(name=subgraph_id[0], version=subgraph_id[1])
+        check_name_and_version(subgraph.jobs["node_2"].outputs["component_out_path"], "sub_pipeline_2_output", "v2")
+
+    @pytest.mark.skipif(
+        condition=not is_live(),
+        reason="TODO (2235034) x-ms-meta-name header masking fixture isn't working, so playback fails",
+    )
+    @pytest.mark.disable_mock_code_hash
+    # without this mark, the code would be passed with different id even when we upload the same component,
+    # add this mark to reuse node and further reuse pipeline
+    def test_register_with_output_format(self, client: MLClient):
+        component = load_component(source="./tests/test_configs/components/helloworld_component.yml")
+        component_input = Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv")
+
+        @dsl.pipeline()
+        def sub_pipeline():
+            node = component(component_in_path=component_input)
+            node.outputs.component_out_path = Output(name="sub_pipeline_o_output", version="v1")
+            return {"sub_pipeine_a_output": node.outputs.component_out_path}
+
+        @dsl.pipeline()
+        def register_both_output():
+            # register NodeOutput which is binding to PipelineOutput
+            node = component(component_in_path=component_input)  # binding and re-define name and version
+            node.outputs.component_out_path = Output(name="n1_o_output", version="1")
+
+            node_2 = component(component_in_path=component_input)  # binding
+            node_2.outputs.component_out_path = Output(name="n2_o_output", version="2")
+
+            node_3 = component(component_in_path=component_input)  # isn't binding
+            node_3.outputs.component_out_path = Output(name="n3_o_output", version="4")
+
+            sub_node = sub_pipeline()  # test set Output for PipelineComponent
+            sub_node.outputs.sub_pipeine_a_output = Output(name="subgraph_o_output", version="1")
+            return {
+                "pipeine_a_output": node.outputs.component_out_path,
+                "pipeine_b_output": node_2.outputs.component_out_path,
+            }
+
+        pipeline = register_both_output()
+        pipeline.settings.default_compute = "cpu-cluster"
+        pipeline.outputs.pipeine_a_output.name = "np_output"
+        pipeline.outputs.pipeine_a_output.version = "1"
+        pipeline_job = assert_job_cancel(pipeline, client)
+
+        check_name_and_version(pipeline_job.outputs.pipeine_a_output, "np_output", "1")
+        check_name_and_version(pipeline_job.outputs.pipeine_b_output, "n2_o_output", "2")
+        check_name_and_version(pipeline_job.jobs["node_3"].outputs["component_out_path"], "n3_o_output", "4")
+        check_name_and_version(pipeline_job.jobs["sub_node"].outputs["sub_pipeine_a_output"], "subgraph_o_output", "1")