feat(sdk): Allow disabling default caching via a CLI flag and env var

[DharmitD]

Description of your changes:
Allow setting a default of execution caching disabled via a compiler CLI flag and env var.
Resolves #11092

Key Changes:

CLI Update (kfp dsl compile):
A new flag --disable-execution-caching-by-default is added. When set, this flag disables caching for all pipeline tasks by default. Behavior specified in pipeline code still takes precedence.

A corresponding Environment Variable KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT is added as well.

Details:

By default, tasks default to caching enabled if the pipeline code doesn't specify a desired behavior via set_caching_options. For example:

@dsl.pipeline(name='my-pipeline')
def my_pipeline():
    task_1 = create_dataset()
    task_2 = create_dataset()
    task_1.set_caching_options(False)

In this example, task 1 won't try to use the cache, but task 2 will, even though the author didn't specify anything about task 2. The resulting pipeline spec will look like this:

root:
  dag:
    tasks:
      task-1:
        cachingOptions: {}
      task-2:
        cachingOptions:
          enableCache: true

This PR enhances the DSL compiler to allow the user to override that default behavior via a command line flag. This change is done:

• in an additive way, using an optional feature flag

• in a way that doesn't alter the DSL or pipeline spec interfaces

• In a way that doesn't change the default behavior for those who prefer the default of caching enabled

After this change, this command will continue to result in the default behavior:

kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml

Whereas this command will cause tasks to default to caching disabled if no desired caching behavior is specified in code:

kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml --disable-execution-caching-by-default

For example, this pipeline compiled with --disable-execution-caching-by-default

@dsl.pipeline(name='my-pipeline')
def my_pipeline():
    task_1 = create_dataset()
    task_2 = create_dataset()
    task_1.set_caching_options(False)

would then result in both tasks having caching disabled:

root:
  dag:
    tasks:
      task-1:
        cachingOptions: {}
      task-2:
        cachingOptions: {}

even though nothing was specified about task_2 in the code.

Usage

CLI flag for dsl compile.

A new flag --disable-execution-caching-by-default is added. When set, this flag disables caching for all pipeline tasks by default.

Example:

kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml --disable-execution-caching-by-default

Environment Variable KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT. 

You can also set the default caching behavior by using the environment variable. When set to true, 1, or other truthy values, it will disable execution caching by default for all pipelines. When set to false or when absent, the default of caching enabled remains.
Example:

KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT=true \
kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml

This environment variable also works for Compiler().compile().

Given the following pipeline file my_pipeline.py:

ffrom kfp import dsl
from kfp.compiler import Compiler

@dsl.component
def create_dataset():
    pass

@dsl.pipeline(name='my-pipeline')
def my_pipeline():
    task_1 = create_dataset()
    task_2 = create_dataset()
    task_1.set_caching_options(False)

Compiler().compile(
    pipeline_func=my_pipeline,
    package_path='my_pipeline.yaml',
)

Executing this

KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT=true \
python my_pipeline.py

will result in task_2 having caching disabled.

This PR is made in collaboration with @gregsheremeta. Thank you, Greg for your contributions!

Checklist:

• You have signed off your commits
• The title for your pull request (PR) should follow our title convention. Learn more about the pull request title convention used in this repository.

[diegolovison]

/hold

[diegolovison]

Wrote my thoughts on this PR as a general feedback without explicit approval.

[diegolovison]

/lgtm

Despite having a different opinion about how this can be implemented, I support this PR

[DharmitD]

/unhold

[chensun]

/lgtm
/approve

[google-oss-prow]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: chensun

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• sdk/OWNERS [chensun]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment