fix(PipelineJob): use name as output only field (#719)

Author: ji-yaqi

google/cloud/aiplatform/pipeline_jobs.py

@@ -48,9 +48,6 @@
     [gca_pipeline_state_v1beta1.PipelineState.PIPELINE_STATE_FAILED]
 )
 
-# Vertex AI Pipelines service API job name relative name prefix pattern.
-_JOB_NAME_PATTERN = "{parent}/pipelineJobs/{job_id}"
-
 # Pattern for valid names used as a Vertex resource name.
 _VALID_NAME_PATTERN = re.compile("^[a-z][-a-z0-9]{0,127}$")
 
@@ -178,19 +175,18 @@ def __init__(
         )
 
         pipeline_name = pipeline_job["pipelineSpec"]["pipelineInfo"]["name"]
-        job_id = job_id or "{pipeline_name}-{timestamp}".format(
+        self.job_id = job_id or "{pipeline_name}-{timestamp}".format(
             pipeline_name=re.sub("[^-0-9a-z]+", "-", pipeline_name.lower())
             .lstrip("-")
             .rstrip("-"),
             timestamp=_get_current_time().strftime("%Y%m%d%H%M%S"),
         )
-        if not _VALID_NAME_PATTERN.match(job_id):
+        if not _VALID_NAME_PATTERN.match(self.job_id):
             raise ValueError(
                 "Generated job ID: {} is illegal as a Vertex pipelines job ID. "
                 "Expecting an ID following the regex pattern "
                 '"[a-z][-a-z0-9]{{0,127}}"'.format(job_id)
             )
-        job_name = _JOB_NAME_PATTERN.format(parent=self._parent, job_id=job_id)
 
         builder = pipeline_utils.PipelineRuntimeConfigBuilder.from_job_spec_json(
             pipeline_job
@@ -206,7 +202,6 @@ def __init__(
 
         self._gca_resource = gca_pipeline_job_v1beta1.PipelineJob(
             display_name=display_name,
-            name=job_name,
             pipeline_spec=pipeline_job["pipelineSpec"],
             labels=labels,
             runtime_config=runtime_config,
@@ -215,18 +210,6 @@ def __init__(
             ),
         )
 
-    def _assert_gca_resource_is_available(self) -> None:
-        # TODO(b/193800063) Change this to name after this fix
-        if not getattr(self._gca_resource, "create_time", None):
-            raise RuntimeError(
-                f"{self.__class__.__name__} resource has not been created."
-                + (
-                    f" Resource failed with: {self._exception}"
-                    if self._exception
-                    else ""
-                )
-            )
-
     @base.optional_sync()
     def run(
         self,
@@ -257,12 +240,10 @@ def run(
 
         _LOGGER.log_create_with_lro(self.__class__)
 
-        # PipelineJob.name is not used by pipeline service
-        pipeline_job_id = self._gca_resource.name.split("/")[-1]
         self._gca_resource = self.api_client.create_pipeline_job(
             parent=self._parent,
             pipeline_job=self._gca_resource,
-            pipeline_job_id=pipeline_job_id,
+            pipeline_job_id=self.job_id,
         )
 
         _LOGGER.log_create_complete_with_getter(

tests/unit/aiplatform/test_pipeline_jobs.py

@@ -236,7 +236,6 @@ def test_run_call_pipeline_service_create(
         # Construct expected request
         expected_gapic_pipeline_job = gca_pipeline_job_v1beta1.PipelineJob(
             display_name=_TEST_PIPELINE_JOB_DISPLAY_NAME,
-            name=_TEST_PIPELINE_JOB_NAME,
             pipeline_spec={
                 "components": {},
                 "pipelineInfo": _TEST_PIPELINE_JOB_SPEC["pipelineSpec"]["pipelineInfo"],