feat: enable_caching in PipelineJob to compile time settings (#557)

google/cloud/aiplatform/pipeline_jobs.py

@@ -96,7 +96,7 @@ def __init__(
         job_id: Optional[str] = None,
         pipeline_root: Optional[str] = None,
         parameter_values: Optional[Dict[str, Any]] = None,
-        enable_caching: Optional[bool] = True,
+        enable_caching: Optional[bool] = None,
         encryption_spec_key_name: Optional[str] = None,
         labels: Optional[Dict[str, str]] = None,
         credentials: Optional[auth_credentials.Credentials] = None,
@@ -121,7 +121,15 @@ def __init__(
                 Optional. The mapping from runtime parameter names to its values that
                 control the pipeline run.
             enable_caching (bool):
-                Optional. Whether to turn on caching for the run. Defaults to True.
+                Optional. Whether to turn on caching for the run.
+
+                If this is not set, defaults to the compile time settings, which
+                are True for all tasks by default, while users may specify
+                different caching options for individual tasks.
+
+                If this is set, the setting applies to all tasks in the pipeline.
+
+                Overrides the compile time settings.
             encryption_spec_key_name (str):
                 Optional. The Cloud KMS resource identifier of the customer
                 managed encryption key used to protect the job. Has the
@@ -198,7 +206,8 @@ def __init__(
         runtime_config = gca_pipeline_job_v1beta1.PipelineJob.RuntimeConfig()._pb
         json_format.ParseDict(runtime_config_dict, runtime_config)
 
-        _set_enable_caching_value(pipeline_job["pipelineSpec"], enable_caching)
+        if enable_caching is not None:
+            _set_enable_caching_value(pipeline_job["pipelineSpec"], enable_caching)
 
         self._gca_resource = gca_pipeline_job_v1beta1.PipelineJob(
             display_name=display_name,