fix: resolve jobs.xxx.inputs for subgraph (Azure#32168)

Author: elliotzh

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

@@ -760,7 +760,7 @@ def _try_resolve_environment_for_component(cls, component, _: str, resolver: _As
     def _resolve_azureml_id(self, component: Component, jobs_only: bool = False) -> None:
         # TODO: remove the parameter `jobs_only`. Some tests are expecting an arm id after resolving for now.
         resolver = self._orchestrators.resolve_azureml_id
-        self._resolve_dependencies_for_component(component, resolver, resolve_jobs_inputs=True, jobs_only=jobs_only)
+        self._resolve_dependencies_for_component(component, resolver, jobs_only=jobs_only)
 
     def _resolve_arm_id_or_upload_dependencies(self, component: Component) -> None:
         resolver = OperationOrchestrator(
@@ -774,7 +774,6 @@ def _resolve_dependencies_for_component(
         component: Component,
         resolver: Callable,
         *,
-        resolve_jobs_inputs: bool = False,
         jobs_only: bool = False,
     ) -> None:
         # for now, many tests are expecting long arm id instead of short id for environment and code
@@ -805,7 +804,6 @@ def _resolve_dependencies_for_component(
         self._resolve_dependencies_for_pipeline_component_jobs(
             component,
             resolver=resolver,
-            resolve_inputs=resolve_jobs_inputs,
         )
 
     def _resolve_inputs_for_pipeline_component_jobs(self, jobs: Dict[str, Any], base_path: str):
@@ -1049,7 +1047,9 @@ def _get_client_key(self) -> str:
         return self._client_key
 
     def _resolve_dependencies_for_pipeline_component_jobs(
-        self, component: Union[Component, str], resolver: _AssetResolver, *, resolve_inputs: bool = True
+        self,
+        component: Union[Component, str],
+        resolver: _AssetResolver,
     ):
         """Resolve dependencies for pipeline component jobs.
         Will directly return if component is not a pipeline component.
@@ -1066,8 +1066,7 @@ def _resolve_dependencies_for_pipeline_component_jobs(
 
         from azure.ai.ml.entities._job.automl.automl_job import AutoMLJob
 
-        if resolve_inputs:
-            self._resolve_inputs_for_pipeline_component_jobs(component.jobs, component._base_path)
+        self._resolve_inputs_for_pipeline_component_jobs(component.jobs, component._base_path)
 
         # This is a preparation for concurrent resolution. Nodes will be resolved later layer by layer
         # from bottom to top, as hash calculation of a parent node will be impacted by resolution

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

@@ -832,7 +832,7 @@ def test_simple_pipeline_component_create(self, client: MLClient, randstr: Calla
         )
         # Assert binding on compute not changed after resolve dependencies
         client.components._resolve_dependencies_for_pipeline_component_jobs(
-            component, resolver=client.components._orchestrators.get_asset_arm_id, resolve_inputs=False
+            component, resolver=client.components._orchestrators.get_asset_arm_id
         )
         assert component.jobs["component_a_job"].compute == "${{parent.inputs.node_compute}}"
         # Assert E2E

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

@@ -1132,7 +1132,7 @@ def valid_pipeline_func(
 
         # Assert binding on compute not changed after resolve dependencies
         client.components._resolve_dependencies_for_pipeline_component_jobs(
-            component, resolver=client.components._orchestrators.get_asset_arm_id, resolve_inputs=False
+            component, resolver=client.components._orchestrators.get_asset_arm_id
         )
         assert component.jobs["node2"].compute == "${{parent.inputs.node_compute}}"
 

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

@@ -1,10 +1,9 @@
-from dataclasses import field
 from functools import partial
 from pathlib import Path
 from typing import Callable, Union
 
 import pytest
-from devtools_testutils import AzureRecordedTestCase
+from devtools_testutils import AzureRecordedTestCase, is_live
 from mock import mock
 from pytest_mock import MockFixture
 from test_utilities.utils import (
@@ -394,3 +393,27 @@ def pipeline_with_command(input_data):
             "component_in_group.number": {"job_input_type": "literal", "value": "1.0"},
             "component_in_group.sub1.integer": {"job_input_type": "literal", "value": "1"},
         }
+
+    @pytest.mark.skipif(not is_live(), reason="Met some problem in recording")
+    def test_dsl_pipeline_component_with_static_local_data_input(self, client: MLClient) -> None:
+        path = "./tests/test_configs/components/helloworld_component.yml"
+        input_data_path = "./tests/test_configs/data/"
+
+        @dsl.pipeline()
+        def pipeline_no_arg():
+            component_func = load_component(source=path)
+            component_func(
+                component_in_path=Input(
+                    path=input_data_path,
+                    type=AssetTypes.URI_FOLDER,
+                ),
+                component_in_number=1,
+            )
+
+        @dsl.pipeline
+        def pipeline_func():
+            pipeline_no_arg()
+
+        pipeline_job: PipelineJob = pipeline_func()
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        client.jobs.create_or_update(pipeline_job)