Add an environment variable and CLI option to disable default caching

Signed-off-by: ddalvi <ddalvi@redhat.com>

Author: DharmitD

sdk/python/kfp/cli/compile_.py

@@ -133,14 +133,32 @@ def parse_parameters(parameters: Optional[str]) -> Dict:
     is_flag=True,
     default=False,
     help='Whether to disable type checking.')
+@click.option(
+    '--execution-caching-enabled-by-default',
+    type=click.Choice(['enabled', 'disabled'], case_sensitive=False),
+    default=None,
+    help="Enable task-level caching. Enabled by default, set it to disabled to disable caching."
+)
 def compile_(
     py: str,
     output: str,
     function_name: Optional[str] = None,
     pipeline_parameters: Optional[str] = None,
     disable_type_check: bool = False,
+    execution_caching_enabled_by_default: Optional[bool] = None,
 ) -> None:
     """Compiles a pipeline or component written in a .py file."""
+
+    env_enable_caching = os.getenv('KFP_EXECUTION_CACHING_ENABLED_BY_DEFAULT', 'enabled').lower() == 'enabled'
+    if execution_caching_enabled_by_default is None:
+        execution_caching_enabled_by_default = env_enable_caching
+    else:
+        execution_caching_enabled_by_default = execution_caching_enabled_by_default.lower() == 'enabled'
+        if execution_caching_enabled_by_default:
+            os.environ['KFP_EXECUTION_CACHING_ENABLED_BY_DEFAULT'] = 'enabled'
+        else:
+            os.environ['KFP_EXECUTION_CACHING_ENABLED_BY_DEFAULT'] = 'disabled'
+
     pipeline_func = collect_pipeline_or_component_func(
         python_file=py, function_name=function_name)
     parsed_parameters = parse_parameters(parameters=pipeline_parameters)

sdk/python/kfp/dsl/pipeline_task.py

@@ -18,6 +18,7 @@
 import functools
 import inspect
 import itertools
+import os
 import re
 from typing import Any, Dict, List, Mapping, Optional, Union
 import warnings
@@ -130,7 +131,7 @@ def __init__(
             inputs=dict(args.items()),
             dependent_tasks=[],
             component_ref=component_spec.name,
-            enable_caching=True)
+            enable_caching=os.getenv('KFP_EXECUTION_CACHING_ENABLED_BY_DEFAULT', 'enabled').lower() == 'enabled')
         self._run_after: List[str] = []
 
         self.importer_spec = None

sdk/python/kfp/dsl/structures.py

@@ -17,6 +17,7 @@
 import collections
 import dataclasses
 import itertools
+import os
 import re
 from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 import uuid
@@ -420,11 +421,10 @@ class TaskSpec:
     trigger_strategy: Optional[str] = None
     iterator_items: Optional[Any] = None
     iterator_item_input: Optional[str] = None
-    enable_caching: bool = True
+    enable_caching: bool = os.getenv('KFP_EXECUTION_CACHING_ENABLED_BY_DEFAULT', 'enabled').lower() == 'enabled'
     display_name: Optional[str] = None
     retry_policy: Optional[RetryPolicy] = None
 
-
 @dataclasses.dataclass
 class ImporterSpec:
     """ImporterSpec definition.