Implemented skipping of cached task results via execution config

CatalogClient.Put can now create or update/overwrite artifacts and their data

Signed-off-by: Nick Müller <[email protected]>

go.mod

@@ -2,6 +2,12 @@ module github.com/flyteorg/flytepropeller
 
 go 1.18
 
+replace (
+	github.com/flyteorg/flyteidl => github.com/blackshark-ai/flyteidl v0.24.22-0.20221005103053-1172311340ca
+	github.com/flyteorg/flyteplugins => github.com/blackshark-ai/flyteplugins v1.0.2-0.20220927114241-fb57ce261bb5
+	github.com/flyteorg/flytestdlib => github.com/blackshark-ai/flytestdlib v1.0.1-0.20220927112514-285faf0b16b4
+)
+
 require (
 	github.com/DiSiqueira/GoTree v1.0.1-0.20180907134536-53a8e837f295
 	github.com/benlaurie/objecthash v0.0.0-20180202135721-d1e3d6079fc1
@@ -26,7 +32,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.2
 	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8
 	google.golang.org/grpc v1.46.0
 	google.golang.org/protobuf v1.28.0
@@ -43,12 +49,12 @@ require (
 	cloud.google.com/go/compute v1.6.1 // indirect
 	cloud.google.com/go/iam v0.3.0 // indirect
 	cloud.google.com/go/storage v1.22.0 // indirect
-	github.com/Azure/azure-sdk-for-go v62.3.0+incompatible // indirect
-	github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 // indirect
-	github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3 // indirect
-	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0 // indirect
+	github.com/Azure/azure-sdk-for-go v63.4.0+incompatible // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/azcore v0.23.1 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/internal v0.9.2 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.4.0 // indirect
 	github.com/Azure/go-autorest v14.2.0+incompatible // indirect
-	github.com/Azure/go-autorest/autorest v0.11.25 // indirect
+	github.com/Azure/go-autorest/autorest v0.11.27 // indirect
 	github.com/Azure/go-autorest/autorest/adal v0.9.18 // indirect
 	github.com/Azure/go-autorest/autorest/date v0.3.0 // indirect
 	github.com/Azure/go-autorest/logger v0.2.1 // indirect
@@ -74,15 +80,15 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful v2.9.6+incompatible // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
-	github.com/flyteorg/stow v0.3.4 // indirect
+	github.com/flyteorg/stow v0.3.6 // indirect
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.14 // indirect
 	github.com/gofrs/uuid v4.2.0+incompatible // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang-jwt/jwt/v4 v4.2.0 // indirect
+	github.com/golang-jwt/jwt/v4 v4.4.1 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/gnostic v0.5.7-v3refs // indirect
 	github.com/google/go-cmp v0.5.8 // indirect
@@ -122,9 +128,9 @@ require (
 	github.com/subosito/gotenv v1.2.0 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect
-	golang.org/x/net v0.0.0-20220607020251-c690dde0001d // indirect
+	golang.org/x/net v0.0.0-20220722155237-a158d28d115b // indirect
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect
-	golang.org/x/sys v0.0.0-20220608164250-635b8c9b7f68 // indirect
+	golang.org/x/sys v0.0.0-20220829200755-d48e67d00261 // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect

pkg/apis/flyteworkflow/v1alpha1/execution_config.go

@@ -30,6 +30,8 @@ type ExecutionConfig struct {
 	TaskResources TaskResources
 	// Defines whether a workflow has been flagged as interruptible.
 	Interruptible *bool
+	// Defines whether a workflow should skip all its cached results and re-compute its output, overwriting any already stored data.
+	SkipCache bool
 }
 
 type TaskPluginOverride struct {

pkg/compiler/test/testdata/snacks-core/k8s/002_core.containerization.multi_images.my_workflow_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/010_core.containerization.raw_container.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/015_core.containerization.use_secrets.my_secret_workflow_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/019_core.control_flow.chain_tasks.chain_tasks_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/022_core.control_flow.checkpoint.example_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/026_core.control_flow.conditions.multiplier_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/028_core.control_flow.conditions.multiplier_2_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/030_core.control_flow.conditions.multiplier_3_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/035_core.control_flow.conditions.basic_boolean_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/037_core.control_flow.conditions.bool_input_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/039_core.control_flow.conditions.nested_conditions_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/042_core.control_flow.conditions.consume_outputs_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/048_core.control_flow.dynamics.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/053_core.control_flow.map_task.my_map_workflow_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/059_core.control_flow.merge_sort.merge_sort_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/062_core.control_flow.subworkflows.my_subwf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/069_core.control_flow.subworkflows.ext_workflow_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/077_core.extend_flyte.custom_task_plugin.my_workflow_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/081_core.extend_flyte.custom_types.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/085_core.flyte_basics.basic_workflow.my_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/089_core.flyte_basics.decorating_tasks.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/095_core.flyte_basics.decorating_workflows.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/098_core.flyte_basics.documented_workflow.sphinx_docstring_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/100_core.flyte_basics.documented_workflow.numpy_docstring_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/102_core.flyte_basics.documented_workflow.google_docstring_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/105_core.flyte_basics.files.normalize_csv_file_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/109_core.flyte_basics.folders.download_and_normalize_csv_files_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/112_core.flyte_basics.hello_world.my_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/117_my.imperative.workflow.example_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/120_core.flyte_basics.lp.my_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/125_core.flyte_basics.lp.go_greet_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/129_core.flyte_basics.named_outputs.my_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/140_core.flyte_basics.shell_task.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/147_core.flyte_basics.task_cache.cached_dataframe_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/151_core.scheduled_workflows.lp_schedules.date_formatter_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/155_core.scheduled_workflows.lp_schedules.positive_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/162_core.type_system.custom_objects.wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/166_core.type_system.enums.enum_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/169_core.type_system.flyte_pickle.welcome_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits:

pkg/compiler/test/testdata/snacks-core/k8s/173_core.type_system.schema.df_wf_2_wf.yaml

@@ -3,6 +3,7 @@ executionConfig:
   Interruptible: null
   MaxParallelism: 0
   RecoveryExecution: {}
+  SkipCache: false
   TaskPluginImpls: null
   TaskResources:
     Limits: