[ML][Pipeline] Fix sub graph job download

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_schedule/trigger.py

@@ -121,10 +121,17 @@ class CronTrigger(TriggerBase):
     :param expression: Specifies cron expression of schedule.
         The expression should follow NCronTab format.
     :type expression: str
-    :param start_time: Specifies start time of schedule in ISO 8601 format.
+    :param start_time: Accepts str or datetime object. The tzinfo should be none if a datetime object, use
+                       ``time_zone`` property to specify a time zone if needed. You can also specify this
+                       parameter as a string in this format: YYYY-MM-DDThh:mm:ss. If None is provided, the
+                       first workload is run instantly and the future workloads are run based on the schedule.
+                       If the start time is in the past, the first workload is run at the next calculated run time.
     :type start_time: Union[str, datetime]
-    :param end_time: Specifies end time of schedule in ISO 8601 format.
-        Note that end_time is not supported for compute schedules.
+    :param end_time: Accepts str or datetime object. The tzinfo should be none if a datetime object, use
+                     ``time_zone`` property to specify a time zone if needed. You can also specify this
+                     parameter as a string in this format: YYYY-MM-DDThh:mm:ss. End time in the past is invalid
+                     and will raise exception when creating schedule.
+                     Note that end_time is not supported for compute schedules.
     :type end_time: Union[str, datetime]
     :param time_zone: Time zone in which the schedule runs. Default to UTC(+00:00).
         This does apply to the start_time and end_time.

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py

@@ -646,7 +646,8 @@ def download(
         """
         job_details = self.get(name)
         # job is reused, get reused job to download
-        if job_details.properties.get(PipelineConstants.REUSED_FLAG_FIELD) == PipelineConstants.REUSED_FLAG_TRUE:
+        if job_details.properties.get(PipelineConstants.REUSED_FLAG_FIELD) == PipelineConstants.REUSED_FLAG_TRUE and \
+                PipelineConstants.REUSED_JOB_ID in job_details.properties:
             reused_job_name = job_details.properties[PipelineConstants.REUSED_JOB_ID]
             reused_job_detail = self.get(reused_job_name)
             module_logger.info("job %s reuses previous job %s, download from the reused job.", name, reused_job_name)

sdk/ml/azure-ai-ml/tests/component/e2etests/test_component.py

@@ -833,9 +833,6 @@ def test_helloworld_nested_pipeline_component(self, client: MLClient, randstr: C
         }
         assert component_dict == expected_dict
 
-    @pytest.mark.skip(
-        "Skip for Bug https://msdata.visualstudio.com/Vienna/_workitems/edit/1969753 not release to canary yet."
-    )
     def test_create_pipeline_component_from_job(self, client: MLClient, randstr: Callable[[str], str]):
         params_override = [{"name": randstr("component_name_0")}]
         pipeline_job = load_job(

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

@@ -607,3 +607,31 @@ def pipeline_with_use_node_with_multiple_output_as_input(component_in_number: in
         with pytest.raises(ValidationException) as ex:
             pipeline_with_use_node_with_multiple_output_as_input(10, "test")
             assert "Exactly 1 output is required, got 2. ({'component_out_path_1': <azure.ai.m" in ex.__str__()
+
+    @pytest.mark.usefixtures("enable_pipeline_private_preview_features")
+    def test_dsl_pipeline_with_compute_binding(self):
+        path = "./tests/test_configs/components/merge_outputs_component.yml"
+        component_func1 = load_component(path)
+
+        @dsl.pipeline
+        def sub_pipeline_with_compute_binding(compute_name: str):
+            node1 = component_func1(
+                component_in_number=1,
+                component_in_path_1="test",
+                component_in_path_2="test2",
+            )
+            node1.compute = compute_name
+
+        @dsl.pipeline
+        def pipeline_with_compute_binding(compute_name: str):
+            node1 = component_func1(
+                component_in_number=1,
+                component_in_path_1="test",
+                component_in_path_2="test2",
+            )
+            node1.compute = compute_name
+            sub_pipeline_with_compute_binding(compute_name)
+
+        pipeline_job = pipeline_with_compute_binding('cpu-cluster')
+        # Assert compute binding validate not raise error when validate
+        assert pipeline_job._validate().passed