SDK - Compiler - Fix large data passing

Stop outputting parameters unless they're consumed as parameters downstream.
This prevents the situaltion when component outputs a big file, but DSL compiler instructs Argo to pick it up as parameter (parameters only hold few kilobytes of data).

As byproduct, this change fixes some minor compiler data passing bugs where some parameters were being passed around, but never consumed (happened with `ResourceOp`, `dsl.Condition` and recursion).

sdk/python/kfp/compiler/compiler.py

@@ -809,6 +809,9 @@ def create_workflow(self,
         dsl_pipeline,
         op_transformers)
 
+    from ._data_passing_rewriter import fix_big_data_passing
+    workflow = fix_big_data_passing(workflow)
+
     import json
     workflow.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/pipeline_spec'] = json.dumps(pipeline_meta.to_dict(), sort_keys=True)
 

sdk/python/tests/compiler/testdata/coin.yaml

@@ -29,22 +29,13 @@ spec:
           parameters:
           - name: flip-again-output
             value: '{{tasks.flip-again.outputs.parameters.flip-again-output}}'
-          - name: flip-output
-            value: '{{inputs.parameters.flip-output}}'
         dependencies:
         - flip-again
         name: condition-2
         template: condition-2
         when: '"{{tasks.flip-again.outputs.parameters.flip-again-output}}" == "tails"'
-      - arguments:
-          parameters:
-          - name: flip-output
-            value: '{{inputs.parameters.flip-output}}'
-        name: flip-again
+      - name: flip-again
         template: flip-again
-    inputs:
-      parameters:
-      - name: flip-output
     name: condition-1
     outputs:
       parameters:
@@ -57,29 +48,23 @@ spec:
           parameters:
           - name: flip-again-output
             value: '{{inputs.parameters.flip-again-output}}'
-          - name: flip-output
-            value: '{{inputs.parameters.flip-output}}'
         name: print1
         template: print1
     inputs:
       parameters:
       - name: flip-again-output
-      - name: flip-output
     name: condition-2
   - dag:
       tasks:
       - arguments:
           parameters:
           - name: flip-again-output
             value: '{{inputs.parameters.flip-again-output}}'
-          - name: flip-output
-            value: '{{inputs.parameters.flip-output}}'
         name: print2
         template: print2
     inputs:
       parameters:
       - name: flip-again-output
-      - name: flip-output
     name: condition-3
   - container:
       args:
@@ -94,10 +79,6 @@ spec:
       artifacts:
       - name: flip-output
         path: /tmp/output
-      parameters:
-      - name: flip-output
-        valueFrom:
-          path: /tmp/output
   - container:
       args:
       - python -c "import random; result = 'heads' if random.randint(0,1) == 0 else
@@ -117,11 +98,7 @@ spec:
           path: /tmp/output
   - dag:
       tasks:
-      - arguments:
-          parameters:
-          - name: flip-output
-            value: '{{tasks.flip.outputs.parameters.flip-output}}'
-        dependencies:
+      - dependencies:
         - flip
         name: condition-1
         template: condition-1
@@ -130,8 +107,6 @@ spec:
           parameters:
           - name: flip-again-output
             value: '{{tasks.condition-1.outputs.parameters.flip-again-output}}'
-          - name: flip-output
-            value: '{{tasks.flip.outputs.parameters.flip-output}}'
         dependencies:
         - condition-1
         - flip

sdk/python/tests/compiler/testdata/imagepullsecrets.yaml

@@ -29,10 +29,6 @@ spec:
       artifacts:
       - name: get-frequent-word
         path: /tmp/message.txt
-      parameters:
-      - name: get-frequent-word
-        valueFrom:
-          path: /tmp/message.txt
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/preemptible_tpu_gpu.yaml

@@ -28,10 +28,6 @@ spec:
         artifacts:
           - name: flip-output
             path: /tmp/output
-        parameters:
-          - name: flip-output
-            valueFrom:
-              path: /tmp/output
       retryStrategy:
         limit: 5
       tolerations:

sdk/python/tests/compiler/testdata/recursive_do_while.yaml

@@ -53,10 +53,6 @@ spec:
       artifacts:
       - name: flip-2-output
         path: /tmp/output
-      parameters:
-      - name: flip-2-output
-        valueFrom:
-          path: /tmp/output
   - container:
       args:
       - python -c "import random; result = 'heads' if random.randint(0,1) == 0 else

sdk/python/tests/compiler/testdata/recursive_while.yaml

@@ -3,8 +3,7 @@ kind: Workflow
 metadata:
   annotations:
     pipelines.kubeflow.org/pipeline_spec: '{"description": "shows how to use dsl.Condition.",
-      "inputs": [{"default": "12", "name": "maxVal"}],
-      "name": "pipeline flip coin"}'
+      "inputs": [{"default": "12", "name": "maxVal"}], "name": "pipeline flip coin"}'
   generateName: pipeline-flip-coin-
 spec:
   arguments:
@@ -16,20 +15,14 @@ spec:
   templates:
   - dag:
       tasks:
-      - arguments:
-          parameters:
-          - name: flip-output
-            value: '{{inputs.parameters.flip-output}}'
-        dependencies:
+      - dependencies:
         - print
         name: flip-3
         template: flip-3
       - arguments:
           parameters:
           - name: flip-output
             value: '{{tasks.flip-3.outputs.parameters.flip-3-output}}'
-          - name: maxval
-            value: '{{inputs.parameters.maxval}}'
         dependencies:
         - flip-3
         name: graph-flip-component-1
@@ -43,7 +36,6 @@ spec:
     inputs:
       parameters:
       - name: flip-output
-      - name: maxval
     name: condition-2
   - container:
       args:
@@ -75,10 +67,6 @@ spec:
       artifacts:
       - name: flip-2-output
         path: /tmp/output
-      parameters:
-      - name: flip-2-output
-        valueFrom:
-          path: /tmp/output
   - container:
       args:
       - python -c "import random; result = 'heads' if random.randint(0,1) == 0 else
@@ -102,15 +90,12 @@ spec:
           parameters:
           - name: flip-output
             value: '{{inputs.parameters.flip-output}}'
-          - name: maxval
-            value: '{{inputs.parameters.maxval}}'
         name: condition-2
         template: condition-2
         when: '"{{inputs.parameters.flip-output}}" == "heads"'
     inputs:
       parameters:
       - name: flip-output
-      - name: maxval
     name: graph-flip-component-1
   - dag:
       tasks:
@@ -122,8 +107,6 @@ spec:
           parameters:
           - name: flip-output
             value: '{{tasks.flip.outputs.parameters.flip-output}}'
-          - name: maxval
-            value: '{{inputs.parameters.maxval}}'
         dependencies:
         - flip
         - flip-2
@@ -138,9 +121,6 @@ spec:
         - graph-flip-component-1
         name: print-2
         template: print-2
-    inputs:
-      parameters:
-      - name: maxval
     name: pipeline-flip-coin
   - container:
       command:
@@ -151,7 +131,6 @@ spec:
       parameters:
       - name: flip-output
     name: print
-
   - container:
       command:
       - echo
@@ -161,4 +140,3 @@ spec:
       parameters:
       - name: flip-output
     name: print-2
-

sdk/python/tests/compiler/testdata/resourceop_basic.yaml

@@ -3,7 +3,8 @@ kind: Workflow
 metadata:
   annotations:
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A Basic Example on ResourceOp
-      Usage.", "inputs": [{"name": "username"}, {"name": "password"}], "name": "ResourceOp Basic"}'
+      Usage.", "inputs": [{"name": "username"}, {"name": "password"}], "name": "ResourceOp
+      Basic"}'
   generateName: resourceop-basic-
 spec:
   arguments:
@@ -23,9 +24,6 @@ spec:
       volumeMounts:
       - mountPath: /etc/secret-volume
         name: my-secret
-    inputs:
-      parameters:
-      - name: create-my-secret-name
     name: cop
   - inputs:
       parameters:
@@ -47,11 +45,7 @@ spec:
         \ generateName: my-secret-\ntype: Opaque\n"
   - dag:
       tasks:
-      - arguments:
-          parameters:
-          - name: create-my-secret-name
-            value: '{{tasks.create-my-secret.outputs.parameters.create-my-secret-name}}'
-        dependencies:
+      - dependencies:
         - create-my-secret
         name: cop
         template: cop

sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml

@@ -129,9 +129,6 @@ spec:
       volumeMounts:
       - mountPath: /data
         name: create-volume-1
-    inputs:
-      parameters:
-      - name: create-volume-1-name
     name: step1-concat
   - container:
       command:
@@ -142,9 +139,6 @@ spec:
       volumeMounts:
       - mountPath: /data
         name: create-volume-2
-    inputs:
-      parameters:
-      - name: create-volume-2-name
     name: step2-gunzip
   - container:
       command:
@@ -154,9 +148,6 @@ spec:
       volumeMounts:
       - mountPath: /data
         name: create-volume-3
-    inputs:
-      parameters:
-      - name: create-volume-3-name
     name: step3-output
   - dag:
       tasks:
@@ -204,27 +195,15 @@ spec:
         - create-snapshot-2
         name: create-volume-3
         template: create-volume-3
-      - arguments:
-          parameters:
-          - name: create-volume-1-name
-            value: '{{tasks.create-volume-1.outputs.parameters.create-volume-1-name}}'
-        dependencies:
+      - dependencies:
         - create-volume-1
         name: step1-concat
         template: step1-concat
-      - arguments:
-          parameters:
-          - name: create-volume-2-name
-            value: '{{tasks.create-volume-2.outputs.parameters.create-volume-2-name}}'
-        dependencies:
+      - dependencies:
         - create-volume-2
         name: step2-gunzip
         template: step2-gunzip
-      - arguments:
-          parameters:
-          - name: create-volume-3-name
-            value: '{{tasks.create-volume-3.outputs.parameters.create-volume-3-name}}'
-        dependencies:
+      - dependencies:
         - create-volume-3
         name: step3-output
         template: step3-output