feat(sdk.v2): enable_caching in v2 client defaults to compile time se…

…ttings. (#6052)

sdk/python/kfp/_client.py

@@ -631,10 +631,11 @@ def run_pipeline(
         dsl.PipelineExecutionMode.V2_ENGINGE mode.
       enable_caching: Optional. Whether or not to enable caching for the run.
         This setting affects v2 compatible mode and v2 mode only.
-        If not set, defaults to the compile time settings, which could vary for
-        different tasks -- users may set different caching options for different
-        task. If not specified, the defaults for all tasks are True.
-        If set, the value overrides the compile time settings.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
 
     Returns:
       A run object. Most important field is id.
@@ -711,10 +712,11 @@ def create_recurring_run(
       enabled: A bool indicating whether the recurring run is enabled or disabled.
       enable_caching: Optional. Whether or not to enable caching for the run.
         This setting affects v2 compatible mode and v2 mode only.
-        If not set, defaults to the compile time settings, which could vary for
-        different tasks -- users may set different caching options for different
-        task. If not specified, the defaults for all tasks are True.
-        If set, the value overrides the compile time settings.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
 
     Returns:
       A Job object. Most important field is id.
@@ -773,10 +775,11 @@ def _create_job_config(
         If only pipeline_id is specified, the default version of this pipeline is used to create the run.
       enable_caching: Whether or not to enable caching for the run.
         This setting affects v2 compatible mode and v2 mode only.
-        If not set, defaults to the compile time settings, which could vary for
-        different tasks -- users may set different caching options for different
-        task. If not specified, the defaults for all tasks are True.
-        If set, the value overrides the compile time settings.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
 
     Returns:
       A JobConfig object with attributes spec and resource_reference.
@@ -858,10 +861,11 @@ def create_run_from_pipeline_func(
         dsl.PipelineExecutionMode.V2_ENGINGE mode.
       enable_caching: Optional. Whether or not to enable caching for the run.
         This setting affects v2 compatible mode and v2 mode only.
-        If not set, defaults to the compile time settings, which could vary for
-        different tasks -- users may set different caching options for different
-        task. If not specified, the defaults for all tasks are True.
-        If set, the value overrides the compile time settings.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
     """
     if pipeline_root is not None and mode == dsl.PipelineExecutionMode.V1_LEGACY:
       raise ValueError('`pipeline_root` should not be used with '
@@ -916,10 +920,11 @@ def create_run_from_pipeline_package(
         dsl.PipelineExecutionMode.V2_ENGINGE mode.
       enable_caching: Optional. Whether or not to enable caching for the run.
         This setting affects v2 compatible mode and v2 mode only.
-        If not set, defaults to the compile time settings, which could vary for
-        different tasks -- users may set different caching options for different
-        task. If not specified, the defaults for all tasks are True.
-        If set, the value overrides the compile time settings.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
     """
 
     class RunPipelineResult:

sdk/python/kfp/v2/google/client/client.py

@@ -261,7 +261,7 @@ def create_run_from_job_spec(
       job_id: Optional[str] = None,
       pipeline_root: Optional[str] = None,
       parameter_values: Optional[Mapping[str, Any]] = None,
-      enable_caching: bool = True,
+      enable_caching: Optional[bool] = None,
       cmek: Optional[str] = None,
       service_account: Optional[str] = None,
       network: Optional[str] = None,
@@ -276,7 +276,12 @@ def create_run_from_job_spec(
       pipeline_root: Optionally the user can override the pipeline root
         specified during the compile time.
       parameter_values: The mapping from runtime parameter names to its values.
-      enable_caching: Whether to turn on caching for the run. Defaults to True.
+      enable_caching: Whether or not to enable caching for the run.
+        If 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 set, the setting applies to all tasks in the pipeline -- overrides
+        the compile time settings.
       cmek: The customer-managed encryption key for a pipelineJob. If set, the
         pipeline job and all of its sub-resources will be secured by this key.
       service_account: The service account that the pipeline workload runs as.
@@ -317,7 +322,8 @@ def create_run_from_job_spec(
     runtime_config = builder.build()
     job_spec['runtimeConfig'] = runtime_config
 
-    _set_enable_caching_value(job_spec['pipelineSpec'], enable_caching)
+    if enable_caching is not None:
+      _set_enable_caching_value(job_spec['pipelineSpec'], enable_caching)
 
     if cmek is not None:
       job_spec['encryptionSpec'] = {'kmsKeyName': cmek}