SDK - Compiler - Fix large data passing (#2173)

* 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).

* Replaced ... with `raise AssertionError`

* Fixed small bug

* Removed unused variables

* Fixed names of the mark_upstream_ios_of_* functions

* Fixed detection of parameter output references

* Fixed handling of volumes

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:
@@ -117,11 +102,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 +111,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
-