Repository: kubeflow/kfp-tekton (language: python)

Issue:
Migrate components away from containerOp
/kind feature

**Description:**
We need to update all our helper functions and examples to use component.yaml instead of containerOp because containerOp is going to be removed in the v2 release.

- [ ] [Hyperparameter tuning using Katib](/samples/katib)
- [ ] [Training and Serving Models with Watson Machine Learning](/samples/watson-train-serve#training-and-serving-models-with-watson-machine-learning)
- [x] [Lightweight python components example](/samples/lightweight-component)
- [x] [Pipeline with Nested loops](/samples/nested-loops)
- [x] [Using Tekton Custom Task on KFP](/samples/tekton-custom-task)
- [x] [The flip-coin pipeline using custom task](/samples/flip-coin-custom-task)

**Additional information:**
[Miscellaneous information that will assist in solving the issue.]



Discussion / hints:
/assign 

/assign @yhwang 
@yhwang can you take care of 
- [x] https://github.com/kubeflow/kfp-tekton/tree/master/samples/nested-pipeline
- [x] https://github.com/kubeflow/kfp-tekton/tree/master/samples/flip-coin
- [x] https://github.com/kubeflow/kfp-tekton/tree/master/samples/trusted-ai

some of them should be partially using component.yaml already

Fix patch:
diff --git a/sdk/FEATURES.md b/sdk/FEATURES.md
index 092b3add49..8372f9109c 100644
--- a/sdk/FEATURES.md
+++ b/sdk/FEATURES.md
@@ -26,7 +26,7 @@ and test pipelines found in the KFP repository.
   - [Features with the Same Behavior as Argo](#features-with-the-same-behavior-as-argo)
     - [InitContainers](#initcontainers)
     - [Conditions](#conditions)
-    - [ResourceOp, VolumeOp, and VolumeSnapshotOp](#resourceop-volumeop-and-volumesnapshotop)
+    - [ResourceOp and VolumeOp](#resourceop-and-volumeop)
     - [Output Parameters](#output-parameters)
     - [Input Artifacts](#input-artifacts)
     - [Output Artifacts](#output-artifacts)
@@ -214,17 +214,10 @@ is an example of how to use this feature.
 Conditions are used for determining whether to execute certain components based on the output of the condition checks. In KFP Argo, each condition is represented as an Argo DAG template so it can be used as a dependency for other Argo templates. To replicate this in KFP Tekton, we put our condition into a dedicated Tekton task so that conditions can be treated as a dependency for other Tekton tasks. Another advantage of creating conditions using Tekton tasks is that we can have more flexible conditions such as comparing an integer and a float number, which currently is not available in Tekton. We are using the Tekton [when expression](https://github.com/tektoncd/pipeline/blob/master/docs/pipelines.md#guard-task-execution-using-whenexpressions) to check whether the condition task has succeeded or not. We created a custom python image to replicate the same condition checks that are in Argo and made it as the default in our compiler. The
 [flip-coin](/samples/flip-coin) example demonstrates how to use multiple conditions within the same pipeline.
 
-### ResourceOp, VolumeOp, and VolumeSnapshotOp
+### ResourceOp and VolumeOp
 
-[ResourceOp, VolumeOp, and VolumeSnapshotOp](https://www.kubeflow.org/docs/pipelines/sdk/manipulate-resources/) are special operations for
-creating Kubernetes resources on the pipeline cluster. ResourceOp is a basic operation for manipulating any Kubernetes resource. VolumeOp
-and VolumeSnapshotOp are operations for creating a unique Volume/VolumeSnapshot per pipeline and can be used as volume/snapshot with any pipeline ContainerOp. Because Tekton has no support for these features natively, we have to implement our own custom task called kubectl-wrapper for creating these Kubernetes resources. The [resourceop_basic](/sdk/python/tests/compiler/testdata/resourceop_basic.py),
-[volume_op](/sdk/python/tests/compiler/testdata/volume_op.py), and
-[volume_snapshot_op](/sdk/python/tests/compiler/testdata/volume_snapshot_op.py) python tests are examples of how to use these features.
-
-One thing to pay attention to is that VolumeSnapshot won't be available by default on all the supported Kubernetes versions. Therefore,
-[VolumeSnapshotDataSource feature gate](https://kubernetes.io/docs/reference/command-line-tools-reference/feature-gates/) must be enabled
-in order to use VolumeSnapshotOp.
+[ResourceOp and VolumeOp](https://www.kubeflow.org/docs/pipelines/sdk/manipulate-resources/) are special operations for
+creating Kubernetes resources on the pipeline cluster. ResourceOp is a basic operation for manipulating any Kubernetes resource. VolumeOps are operations for creating a unique Volume per pipeline and can be used as volume with any pipeline ContainerOp. Because Tekton has no support for these features natively, we have to implement our own custom task called kubectl-wrapper for creating these Kubernetes resources. The [resourceop_basic](/sdk/python/tests/compiler/testdata/resourceop_basic.py) and [volume_op](/sdk/python/tests/compiler/testdata/volume_op.py) python tests are examples of how to use these features.
 
 ### Output Parameters
 


Test patch:
diff --git a/sdk/python/tests/compiler/compiler_tests.py b/sdk/python/tests/compiler/compiler_tests.py
index 2bae8c1341..7b12b7224e 100644
--- a/sdk/python/tests/compiler/compiler_tests.py
+++ b/sdk/python/tests/compiler/compiler_tests.py
@@ -388,13 +388,6 @@ def test_volumeOp_workflow(self):
     from .testdata.volume_op import volumeop_basic
     self._test_pipeline_workflow(volumeop_basic, 'volume_op.yaml')
 
-  def test_volumeSnapshotOp_workflow(self):
-    """
-    Test compiling a volumeSnapshotOp basic workflow.
-    """
-    from .testdata.volume_snapshot_op import volume_snapshotop_sequential
-    self._test_pipeline_workflow(volume_snapshotop_sequential, 'volume_snapshot_op.yaml')
-
   def test_hidden_output_file_workflow(self):
     """
     Test compiling a workflow with non configurable output file.
diff --git a/sdk/python/tests/compiler/testdata/volume.py b/sdk/python/tests/compiler/testdata/volume.py
index 868a4ea22f..1698aed2bf 100644
--- a/sdk/python/tests/compiler/testdata/volume.py
+++ b/sdk/python/tests/compiler/testdata/volume.py
@@ -15,18 +15,53 @@
 
 import kfp.dsl as dsl
 from kubernetes.client import V1Volume, V1SecretVolumeSource, V1VolumeMount, V1EnvVar
+from kfp import components
+
+OP1_STR = """
+name: download
+outputs:
+  - {name: downloaded, type: String}
+implementation:
+  container:
+    image: google/cloud-sdk
+    command:
+    - sh
+    - -c
+    args:
+    - |
+      set -e
+      ls | tee $0
+    - {outputPath: downloaded}
+"""
+
+OP2_STR = """
+name: echo
+inputs:
+  - {name: msg, type: String}
+implementation:
+  container:
+    image: library/bash
+    command:
+    - sh
+    - -c
+    args:
+    - |
+      set -e
+      echo
+    - {inputValue: msg}
+"""
+
+op1_op = components.load_component_from_text(OP1_STR)
+op2_op = components.load_component_from_text(OP2_STR)
 
 
 @dsl.pipeline(
-    name='Volume',
+    name='volume',
     description='A pipeline with volume.'
 )
 def volume_pipeline():
-    op1 = dsl.ContainerOp(name='download',
-                          image='google/cloud-sdk',
-                          command=['sh', '-c'],
-                          arguments=['ls | tee /tmp/results.txt'],
-                          file_outputs={'downloaded': '/tmp/results.txt'})
+    op1 = op1_op()
+
     op1.add_volume(V1Volume(name='gcp-credentials',
                             secret=V1SecretVolumeSource(secret_name='user-gcp-sa')))
     op1.container.add_volume_mount(V1VolumeMount(mount_path='/secret/gcp-credentials',
@@ -35,10 +70,7 @@ def volume_pipeline():
                                             value='/secret/gcp-credentials/user-gcp-sa.json'))
     op1.container.add_env_variable(V1EnvVar(name='Foo', value='bar'))
 
-    op2 = dsl.ContainerOp(name='echo',
-                          image='library/bash',
-                          command=['sh', '-c'],
-                          arguments=['echo %s' % op1.output])
+    op2 = op2_op(op1.output)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/volume.yaml b/sdk/python/tests/compiler/testdata/volume.yaml
index b05fb15f59..e98b1c987d 100644
--- a/sdk/python/tests/compiler/testdata/volume.yaml
+++ b/sdk/python/tests/compiler/testdata/volume.yaml
@@ -18,7 +18,7 @@ metadata:
   name: volume
   annotations:
     tekton.dev/output_artifacts: '{"download": [{"key": "artifacts/$PIPELINERUN/download/downloaded.tgz",
-      "name": "download-downloaded", "path": "/tmp/results.txt"}]}'
+      "name": "download-downloaded", "path": "/tmp/outputs/downloaded/data"}]}'
     tekton.dev/input_artifacts: '{"echo": [{"name": "download-downloaded", "parent_task":
       "download"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -28,7 +28,7 @@ metadata:
       "echo": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A pipeline with volume.",
-      "name": "Volume"}'
+      "name": "volume"}'
 spec:
   pipelineSpec:
     tasks:
@@ -37,7 +37,10 @@ spec:
         steps:
         - name: main
           args:
-          - ls | tee $(results.downloaded.path)
+          - |
+            set -e
+            ls | tee $0
+          - $(results.downloaded.path)
           command:
           - sh
           - -c
@@ -52,7 +55,7 @@ spec:
             name: gcp-credentials
         results:
         - name: downloaded
-          description: /tmp/results.txt
+          description: /tmp/outputs/downloaded/data
         volumes:
         - name: gcp-credentials
           secret:
@@ -63,6 +66,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\nls | tee $0\n", {"outputPath": "downloaded"}], "command":
+              ["sh", "-c"], "image": "google/cloud-sdk"}}, "name": "download", "outputs":
+              [{"name": "downloaded", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: echo
@@ -73,7 +80,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo $(inputs.params.download-downloaded)
+          - |
+            set -e
+            echo
+          - $(inputs.params.download-downloaded)
           command:
           - sh
           - -c
@@ -86,6 +96,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho\n", {"inputValue": "msg"}], "command": ["sh",
+              "-c"], "image": "library/bash"}}, "inputs": [{"name": "msg", "type":
+              "String"}], "name": "echo"}'
             tekton.dev/template: ''
       timeout: 0s
   timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/volume_noninlined.yaml b/sdk/python/tests/compiler/testdata/volume_noninlined.yaml
index b05fb15f59..e98b1c987d 100644
--- a/sdk/python/tests/compiler/testdata/volume_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/volume_noninlined.yaml
@@ -18,7 +18,7 @@ metadata:
   name: volume
   annotations:
     tekton.dev/output_artifacts: '{"download": [{"key": "artifacts/$PIPELINERUN/download/downloaded.tgz",
-      "name": "download-downloaded", "path": "/tmp/results.txt"}]}'
+      "name": "download-downloaded", "path": "/tmp/outputs/downloaded/data"}]}'
     tekton.dev/input_artifacts: '{"echo": [{"name": "download-downloaded", "parent_task":
       "download"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -28,7 +28,7 @@ metadata:
       "echo": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A pipeline with volume.",
-      "name": "Volume"}'
+      "name": "volume"}'
 spec:
   pipelineSpec:
     tasks:
@@ -37,7 +37,10 @@ spec:
         steps:
         - name: main
           args:
-          - ls | tee $(results.downloaded.path)
+          - |
+            set -e
+            ls | tee $0
+          - $(results.downloaded.path)
           command:
           - sh
           - -c
@@ -52,7 +55,7 @@ spec:
             name: gcp-credentials
         results:
         - name: downloaded
-          description: /tmp/results.txt
+          description: /tmp/outputs/downloaded/data
         volumes:
         - name: gcp-credentials
           secret:
@@ -63,6 +66,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\nls | tee $0\n", {"outputPath": "downloaded"}], "command":
+              ["sh", "-c"], "image": "google/cloud-sdk"}}, "name": "download", "outputs":
+              [{"name": "downloaded", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: echo
@@ -73,7 +80,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo $(inputs.params.download-downloaded)
+          - |
+            set -e
+            echo
+          - $(inputs.params.download-downloaded)
           command:
           - sh
           - -c
@@ -86,6 +96,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho\n", {"inputValue": "msg"}], "command": ["sh",
+              "-c"], "image": "library/bash"}}, "inputs": [{"name": "msg", "type":
+              "String"}], "name": "echo"}'
             tekton.dev/template: ''
       timeout: 0s
   timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/volume_op.py b/sdk/python/tests/compiler/testdata/volume_op.py
index 737eea16e9..2939dfc210 100644
--- a/sdk/python/tests/compiler/testdata/volume_op.py
+++ b/sdk/python/tests/compiler/testdata/volume_op.py
@@ -13,13 +13,28 @@
 # limitations under the License.
 
 import kfp.dsl as dsl
+from kfp import components
+
+COP_STR = """
+name: cop
+implementation:
+  container:
+    image: library/bash:4.4.23
+    command:
+    - sh
+    - -c
+    args:
+    - echo foo > /mnt/file1
+"""
+
+cop_op = components.load_component_from_text(COP_STR)
 
 
 @dsl.pipeline(
-    name="VolumeOp Basic",
+    name="volumeop-basic",
     description="A Basic Example on VolumeOp Usage."
 )
-def volumeop_basic(size="10M"):
+def volumeop_basic(size: str = "10M"):
     vop = dsl.VolumeOp(
         name="create-pvc",
         resource_name="my-pvc",
@@ -29,13 +44,7 @@ def volumeop_basic(size="10M"):
         # failure_condition="status.phase = Failed"
     )
 
-    cop = dsl.ContainerOp(
-        name="cop",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo foo > /mnt/file1"],
-        pvolumes={"/mnt": vop.volume}
-    )
+    cop = cop_op().add_pvolumes({"/mnt": vop.volume})
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/volume_op.yaml b/sdk/python/tests/compiler/testdata/volume_op.yaml
index 837217d09a..2233fca33d 100644
--- a/sdk/python/tests/compiler/testdata/volume_op.yaml
+++ b/sdk/python/tests/compiler/testdata/volume_op.yaml
@@ -26,8 +26,8 @@ metadata:
     tekton.dev/artifact_items: '{"cop": [], "create-pvc": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A Basic Example on VolumeOp
-      Usage.", "inputs": [{"default": "10M", "name": "size", "optional": true}], "name":
-      "VolumeOp Basic"}'
+      Usage.", "inputs": [{"default": "10M", "name": "size", "optional": true, "type":
+      "String"}], "name": "volumeop-basic"}'
 spec:
   params:
   - name: size
@@ -155,6 +155,9 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["echo foo > /mnt/file1"], "command": ["sh", "-c"], "image":
+              "library/bash:4.4.23"}}, "name": "cop"}'
             tekton.dev/template: ''
       timeout: 0s
   timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/volume_op_noninlined.yaml b/sdk/python/tests/compiler/testdata/volume_op_noninlined.yaml
index 837217d09a..2233fca33d 100644
--- a/sdk/python/tests/compiler/testdata/volume_op_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/volume_op_noninlined.yaml
@@ -26,8 +26,8 @@ metadata:
     tekton.dev/artifact_items: '{"cop": [], "create-pvc": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"description": "A Basic Example on VolumeOp
-      Usage.", "inputs": [{"default": "10M", "name": "size", "optional": true}], "name":
-      "VolumeOp Basic"}'
+      Usage.", "inputs": [{"default": "10M", "name": "size", "optional": true, "type":
+      "String"}], "name": "volumeop-basic"}'
 spec:
   params:
   - name: size
@@ -155,6 +155,9 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["echo foo > /mnt/file1"], "command": ["sh", "-c"], "image":
+              "library/bash:4.4.23"}}, "name": "cop"}'
             tekton.dev/template: ''
       timeout: 0s
   timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshot_op.log b/sdk/python/tests/compiler/testdata/volume_snapshot_op.log
deleted file mode 100644
index 6739720de7..0000000000
--- a/sdk/python/tests/compiler/testdata/volume_snapshot_op.log
+++ /dev/null
@@ -1,4 +0,0 @@
-[create-volume : main] map[apiVersion:v1 kind:PersistentVolumeClaim metadata:map[annotations:map[ibm.io/provisioning-status:{"status":"inprogress","time":"2021-04-23T23:58:25Z","attempt":1,"retry":false,"pluginid":"ibm-file-plugin-1619200208","pvcid":"19fb44e6-8599-431f-84f4-3eb93cdef7e0"} volume.beta.kubernetes.io/storage-provisioner:ibm.io/ibmc-file] creationTimestamp:2021-04-23T23:58:25Z finalizers:[kubernetes.io/pvc-protection] labels:map[region:us-south zone:sjc04] managedFields:[map[apiVersion:v1 fieldsType:FieldsV1 fieldsV1:map[f:metadata:map[f:annotations:map[f:ibm.io/provisioning-status:map[]] f:labels:map[.:map[] f:region:map[] f:zone:map[]]]] manager:armada-storage-file-plugin operation:Update time:2021-04-23T23:58:25Z] map[apiVersion:v1 fieldsType:FieldsV1 fieldsV1:map[f:metadata:map[f:annotations:map[.:map[] f:volume.beta.kubernetes.io/storage-provisioner:map[]]]] manager:kube-controller-manager operation:Update time:2021-04-23T23:58:25Z] map[apiVersion:v1 fieldsType:FieldsV1 fieldsV1:map[f:spec:map[f:accessModes:map[] f:resources:map[f:requests:map[.:map[] f:storage:map[]]] f:volumeMode:map[]] f:status:map[f:phase:map[]]] manager:kubectl operation:Update time:2021-04-23T23:58:25Z]] name:volumesnapshotop-sequential-cfd3c-vol1 namespace:kubeflow resourceVersion:210076 selfLink:/api/v1/namespaces/kubeflow/persistentvolumeclaims/volumesnapshotop-sequential-cfd3c-vol1 uid:19fb44e6-8599-431f-84f4-3eb93cdef7e0] spec:map[accessModes:[ReadWriteOnce] resources:map[requests:map[storage:1Gi]] storageClassName:ibmc-file-gold-gid volumeMode:Filesystem] status:map[phase:Pending]]volumesnapshotop-sequential-cfd3c-vol1
-
-[step1-ingest : main] mkdir: cannot create directory '/data/step1': File exists
-
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshot_op.py b/sdk/python/tests/compiler/testdata/volume_snapshot_op.py
deleted file mode 100644
index 561b2955a3..0000000000
--- a/sdk/python/tests/compiler/testdata/volume_snapshot_op.py
+++ /dev/null
@@ -1,95 +0,0 @@
-# Copyright 2020 kubeflow.org
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-"""This sample uses Rok as an example to show case how VolumeOp accepts
-annotations as an extra argument, and how we can use arbitrary PipelineParams
-to determine their contents.
-
-The specific annotation is Rok-specific, but the use of annotations in such way
-is widespread in storage systems integrated with K8s.
-"""
-
-import kfp.dsl as dsl
-
-
-@dsl.pipeline(
-    name="VolumeSnapshotOp Sequential",
-    description="""The fourth example of the design doc. Please enable the
-    volume snapshot feature gate in order to run this pipeline."""
-)
-def volume_snapshotop_sequential(url):
-    vop = dsl.VolumeOp(
-        name="create_volume",
-        resource_name="vol1",
-        size="1Gi",
-        modes=dsl.VOLUME_MODE_RWO
-    )
-
-    step1 = dsl.ContainerOp(
-        name="step1_ingest",
-        image="google/cloud-sdk:279.0.0",
-        command=["sh", "-c"],
-        arguments=["mkdir /data/step1 && "
-                   "gsutil cat %s | gzip -c >/data/step1/file1.gz" % url],
-        pvolumes={"/data": vop.volume}
-    )
-
-    step1_snap = dsl.VolumeSnapshotOp(
-        name="step1_snap",
-        resource_name="step1_snap",
-        volume=step1.pvolume
-    )
-
-    step2 = dsl.ContainerOp(
-        name="step2_gunzip",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["mkdir /data/step2 && "
-                   "gunzip /data/step1/file1.gz -c >/data/step2/file1"],
-        pvolumes={"/data": step1.pvolume}
-    )
-
-    step2_snap = dsl.VolumeSnapshotOp(
-        name="step2_snap",
-        resource_name="step2_snap",
-        volume=step2.pvolume
-    )
-
-    step3 = dsl.ContainerOp(
-        name="step3_copy",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["mkdir /data/step3 && "
-                   "cp -av /data/step2/file1 /data/step3/file3"],
-        pvolumes={"/data": step2.pvolume}
-    )
-
-    step3_snap = dsl.VolumeSnapshotOp(
-        name="step3_snap",
-        resource_name="step3_snap",
-        volume=step3.pvolume
-    )
-
-    step4 = dsl.ContainerOp(
-        name="step4_output",
-        image="library/bash:4.4.23",
-        command=["cat", "/data/step2/file1", "/data/step3/file3"],
-        pvolumes={"/data": step3.pvolume}
-    )
-
-
-if __name__ == '__main__':
-    from kfp_tekton.compiler import TektonCompiler
-    TektonCompiler().compile(volume_snapshotop_sequential, __file__.replace('.py', '.yaml'))
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshot_op.yaml b/sdk/python/tests/compiler/testdata/volume_snapshot_op.yaml
deleted file mode 100644
index 6d58c860d9..0000000000
--- a/sdk/python/tests/compiler/testdata/volume_snapshot_op.yaml
+++ /dev/null
@@ -1,539 +0,0 @@
-# Copyright 2021 kubeflow.org
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-apiVersion: tekton.dev/v1beta1
-kind: PipelineRun
-metadata:
-  name: volumesnapshotop-sequential
-  annotations:
-    tekton.dev/output_artifacts: '{}'
-    tekton.dev/input_artifacts: '{"step1-ingest": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step1-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step2-gunzip": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step2-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step3-copy": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step3-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step4-output": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}]}'
-    tekton.dev/artifact_bucket: mlpipeline
-    tekton.dev/artifact_endpoint: minio-service.kubeflow:9000
-    tekton.dev/artifact_endpoint_scheme: http://
-    tekton.dev/artifact_items: '{"create-volume": [], "step1-ingest": [], "step1-snap":
-      [], "step2-gunzip": [], "step2-snap": [], "step3-copy": [], "step3-snap": [],
-      "step4-output": []}'
-    sidecar.istio.io/inject: "false"
-    pipelines.kubeflow.org/pipeline_spec: '{"description": "The fourth example of
-      the design doc. Please enable the\n    volume snapshot feature gate in order
-      to run this pipeline.", "inputs": [{"name": "url"}], "name": "VolumeSnapshotOp
-      Sequential"}'
-spec:
-  params:
-  - name: url
-    value: ''
-  pipelineSpec:
-    params:
-    - name: url
-    tasks:
-    - name: create-volume
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.capacity.storage}'
-      - name: set-ownerreference
-        value: "false"
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: v1
-            kind: PersistentVolumeClaim
-            metadata:
-              name: $(PIPELINERUN)-vol1
-            spec:
-              accessModes:
-              - ReadWriteOnce
-              resources:
-                requests:
-                  storage: 1Gi
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.capacity.storage}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      timeout: 0s
-    - name: step1-ingest
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      - name: url
-        value: $(params.url)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step1 && gsutil cat $(inputs.params.url) | gzip -c >/data/step1/file1.gz
-          command:
-          - sh
-          - -c
-          image: google/cloud-sdk:279.0.0
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        - name: url
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      timeout: 0s
-    - name: step1-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step1-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step1-ingest
-      timeout: 0s
-    - name: step2-gunzip
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step2 && gunzip /data/step1/file1.gz -c >/data/step2/file1
-          command:
-          - sh
-          - -c
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step1-ingest
-      timeout: 0s
-    - name: step2-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step2-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step2-gunzip
-      timeout: 0s
-    - name: step3-copy
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step3 && cp -av /data/step2/file1 /data/step3/file3
-          command:
-          - sh
-          - -c
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step2-gunzip
-      timeout: 0s
-    - name: step3-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step3-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step3-copy
-      timeout: 0s
-    - name: step4-output
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          command:
-          - cat
-          - /data/step2/file1
-          - /data/step3/file3
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step3-copy
-      timeout: 0s
-  timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshot_op_noninlined.yaml b/sdk/python/tests/compiler/testdata/volume_snapshot_op_noninlined.yaml
deleted file mode 100644
index 6d58c860d9..0000000000
--- a/sdk/python/tests/compiler/testdata/volume_snapshot_op_noninlined.yaml
+++ /dev/null
@@ -1,539 +0,0 @@
-# Copyright 2021 kubeflow.org
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-apiVersion: tekton.dev/v1beta1
-kind: PipelineRun
-metadata:
-  name: volumesnapshotop-sequential
-  annotations:
-    tekton.dev/output_artifacts: '{}'
-    tekton.dev/input_artifacts: '{"step1-ingest": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step1-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step2-gunzip": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step2-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step3-copy": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step3-snap": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}], "step4-output": [{"name": "create-volume-name",
-      "parent_task": "create-volume"}]}'
-    tekton.dev/artifact_bucket: mlpipeline
-    tekton.dev/artifact_endpoint: minio-service.kubeflow:9000
-    tekton.dev/artifact_endpoint_scheme: http://
-    tekton.dev/artifact_items: '{"create-volume": [], "step1-ingest": [], "step1-snap":
-      [], "step2-gunzip": [], "step2-snap": [], "step3-copy": [], "step3-snap": [],
-      "step4-output": []}'
-    sidecar.istio.io/inject: "false"
-    pipelines.kubeflow.org/pipeline_spec: '{"description": "The fourth example of
-      the design doc. Please enable the\n    volume snapshot feature gate in order
-      to run this pipeline.", "inputs": [{"name": "url"}], "name": "VolumeSnapshotOp
-      Sequential"}'
-spec:
-  params:
-  - name: url
-    value: ''
-  pipelineSpec:
-    params:
-    - name: url
-    tasks:
-    - name: create-volume
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.capacity.storage}'
-      - name: set-ownerreference
-        value: "false"
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: v1
-            kind: PersistentVolumeClaim
-            metadata:
-              name: $(PIPELINERUN)-vol1
-            spec:
-              accessModes:
-              - ReadWriteOnce
-              resources:
-                requests:
-                  storage: 1Gi
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.capacity.storage}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      timeout: 0s
-    - name: step1-ingest
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      - name: url
-        value: $(params.url)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step1 && gsutil cat $(inputs.params.url) | gzip -c >/data/step1/file1.gz
-          command:
-          - sh
-          - -c
-          image: google/cloud-sdk:279.0.0
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        - name: url
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      timeout: 0s
-    - name: step1-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step1-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step1-ingest
-      timeout: 0s
-    - name: step2-gunzip
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step2 && gunzip /data/step1/file1.gz -c >/data/step2/file1
-          command:
-          - sh
-          - -c
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step1-ingest
-      timeout: 0s
-    - name: step2-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step2-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step2-gunzip
-      timeout: 0s
-    - name: step3-copy
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          args:
-          - mkdir /data/step3 && cp -av /data/step2/file1 /data/step3/file3
-          command:
-          - sh
-          - -c
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step2-gunzip
-      timeout: 0s
-    - name: step3-snap
-      params:
-      - name: action
-        value: create
-      - name: output
-        value: |
-          - name: manifest
-            valueFrom: '{}'
-          - name: name
-            valueFrom: '{.metadata.name}'
-          - name: size
-            valueFrom: '{.status.restoreSize}'
-      - name: success-condition
-        value: status.readyToUse == true
-      - name: set-ownerreference
-        value: "false"
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        params:
-        - description: Action on the resource
-          name: action
-          type: string
-        - default: strategic
-          description: Merge strategy when using action patch
-          name: merge-strategy
-          type: string
-        - default: ''
-          description: An express to retrieval data from resource.
-          name: output
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is success.
-          name: success-condition
-          type: string
-        - default: ''
-          description: A label selector express to decide if the action on resource
-            is failure.
-          name: failure-condition
-          type: string
-        - default: aipipeline/kubectl-wrapper:0.9.0
-          description: Kubectl wrapper image
-          name: image
-          type: string
-        - default: "false"
-          description: Enable set owner reference for created resource.
-          name: set-ownerreference
-          type: string
-        - name: create-volume-name
-        steps:
-        - args:
-          - --action=$(params.action)
-          - --merge-strategy=$(params.merge-strategy)
-          - |
-            --manifest=apiVersion: snapshot.storage.k8s.io/v1alpha1
-            kind: VolumeSnapshot
-            metadata:
-              name: $(PIPELINERUN)-step3-snap
-            spec:
-              source:
-                kind: PersistentVolumeClaim
-                name: $(inputs.params.create-volume-name)
-          - --output=$(params.output)
-          - --success-condition=$(params.success-condition)
-          - --failure-condition=$(params.failure-condition)
-          - --set-ownerreference=$(params.set-ownerreference)
-          image: $(params.image)
-          name: main
-          resources: {}
-          env:
-          - name: PIPELINERUN
-            valueFrom:
-              fieldRef:
-                fieldPath: metadata.labels['tekton.dev/pipelineRun']
-        results:
-        - name: manifest
-          description: '{}'
-        - name: name
-          description: '{.metadata.name}'
-        - name: size
-          description: '{.status.restoreSize}'
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step3-copy
-      timeout: 0s
-    - name: step4-output
-      params:
-      - name: create-volume-name
-        value: $(tasks.create-volume.results.name)
-      taskSpec:
-        steps:
-        - name: main
-          command:
-          - cat
-          - /data/step2/file1
-          - /data/step3/file3
-          image: library/bash:4.4.23
-          volumeMounts:
-          - mountPath: /data
-            name: create-volume
-        params:
-        - name: create-volume-name
-        volumes:
-        - name: create-volume
-          persistentVolumeClaim:
-            claimName: $(inputs.params.create-volume-name)
-        metadata:
-          labels:
-            pipelines.kubeflow.org/pipelinename: ''
-            pipelines.kubeflow.org/generation: ''
-            pipelines.kubeflow.org/cache_enabled: "true"
-          annotations:
-            tekton.dev/template: ''
-      runAfter:
-      - step3-copy
-      timeout: 0s
-  timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested.py b/sdk/python/tests/compiler/testdata/withitem_multi_nested.py
index 2d0398e256..5976437993 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested.py
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested.py
@@ -15,6 +15,7 @@
 import kfp.dsl as dsl
 from kfp_tekton.compiler import TektonCompiler
 from typing import List
+from kfp import components
 
 
 class Coder:
@@ -24,47 +25,109 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op1_yaml = '''\
+name: 'my-in-coop1'
+inputs:
+- {name: item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op1 "$0"
+        - {inputValue: item}
+'''
 
-@dsl.pipeline(name='withitem multiple nesting pipeline')
+op11_yaml = '''\
+name: 'my-inner-inner-coop'
+inputs:
+- {name: item, type: Integer}
+- {name: inner_item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op11 "$0" "$1"
+        - {inputValue: item}
+        - {inputValue: inner_item}
+'''
+
+op12_yaml = '''\
+name: 'my-inner-inner-coop'
+inputs:
+- {name: item, type: Integer}
+- {name: inner_item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op12 "$0" "$1"
+        - {inputValue: item}
+        - {inputValue: inner_item}
+'''
+
+op13_yaml = '''\
+name: 'my-inner-inner-coop'
+inputs:
+- {name: item, type: Integer}
+- {name: inner_item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op13 "$0" "$1"
+        - {inputValue: item}
+        - {inputValue: inner_item}
+'''
+
+op2_yaml = '''\
+name: 'my-in-coop2'
+inputs:
+- {name: item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op2 "$0"
+        - {inputValue: item}
+'''
+
+
+@dsl.pipeline(name='withitem-multiple-nesting-pipeline')
 def pipeline(my_pipe_param: list = [100, 200], my_pipe_param3: list = [1, 2]):
     loop_args = [{'a': 1, 'b': 2}, {'a': 10, 'b': 20}]
     with dsl.ParallelFor(loop_args) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-coop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo op1 %s" % (item.a)],
-        )
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item.a)
 
         with dsl.ParallelFor(my_pipe_param) as inner_item:
-            op11 = dsl.ContainerOp(
-                name="my-inner-inner-coop",
-                image="library/bash:4.4.23",
-                command=["sh", "-c"],
-                arguments=["echo op1 %s %s" % (item.a, inner_item)],
-            )
+            op11_template = components.load_component_from_text(op11_yaml)
+            op11 = op11_template(item.a, inner_item)
+
             my_pipe_param2: List[int] = [4, 5]
             with dsl.ParallelFor(my_pipe_param2) as inner_item:
-                op12 = dsl.ContainerOp(
-                    name="my-inner-inner-coop",
-                    image="library/bash:4.4.23",
-                    command=["sh", "-c"],
-                    arguments=["echo op1 %s %s" % (item.b, inner_item)],
-                )
+                op12_template = components.load_component_from_text(op12_yaml)
+                op12 = op11_template(item.b, inner_item)
                 with dsl.ParallelFor(my_pipe_param3) as inner_item:
-                    op13 = dsl.ContainerOp(
-                        name="my-inner-inner-coop",
-                        image="library/bash:4.4.23",
-                        command=["sh", "-c"],
-                        arguments=["echo op1 %s %s" % (item.b, inner_item)],
-                    )
-
-        op2 = dsl.ContainerOp(
-            name="my-in-coop2",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo op2 %s" % item.b],
-        )
+                    op13_template = components.load_component_from_text(op13_yaml)
+                    op13 = op11_template(item.b, inner_item)
+
+        op2_template = components.load_component_from_text(op2_yaml)
+        op2 = op2_template(item.b)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested.yaml
index cca9691351..935c480f63 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested.yaml
@@ -27,8 +27,7 @@ metadata:
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "[100, 200]", "name":
       "my_pipe_param", "optional": true, "type": "JsonArray"}, {"default": "[1, 2]",
-      "name": "my_pipe_param3", "optional": true, "type": "JsonArray"}], "name": "withitem
-      multiple nesting pipeline"}'
+      "name": "my_pipe_param3", "optional": true, "type": "JsonArray"}], "name": "withitem-multiple-nesting-pipeline"}'
 spec:
   params:
   - name: my_pipe_param
@@ -73,7 +72,10 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - echo op1 $(inputs.params.loop-item-param-1-subvar-a)
+                  - |
+                    set -e
+                    echo op1 "$0"
+                  - $(inputs.params.loop-item-param-1-subvar-a)
                   command:
                   - sh
                   - -c
@@ -87,6 +89,10 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho op1 \"$0\"\n", {"inputValue": "item"}],
+                      "command": ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs":
+                      [{"name": "item", "type": "Integer"}], "name": "my-in-coop1"}'
                     tekton.dev/template: ''
               timeout: 0s
             - name: my-in-coop2
@@ -97,7 +103,10 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - echo op2 $(inputs.params.loop-item-param-1-subvar-b)
+                  - |
+                    set -e
+                    echo op2 "$0"
+                  - $(inputs.params.loop-item-param-1-subvar-b)
                   command:
                   - sh
                   - -c
@@ -111,6 +120,10 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho op2 \"$0\"\n", {"inputValue": "item"}],
+                      "command": ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs":
+                      [{"name": "item", "type": "Integer"}], "name": "my-in-coop2"}'
                     tekton.dev/template: ''
               timeout: 0s
             - name: withitem-multiple-nesting-pipeline-for-loop-3
@@ -148,7 +161,11 @@ spec:
                         steps:
                         - name: main
                           args:
-                          - echo op1 $(inputs.params.loop-item-param-1-subvar-a) $(inputs.params.my_pipe_param-loop-item)
+                          - |
+                            set -e
+                            echo op11 "$0" "$1"
+                          - $(inputs.params.loop-item-param-1-subvar-a)
+                          - $(inputs.params.my_pipe_param-loop-item)
                           command:
                           - sh
                           - -c
@@ -164,6 +181,12 @@ spec:
                             pipelines.kubeflow.org/generation: ''
                             pipelines.kubeflow.org/cache_enabled: "true"
                           annotations:
+                            pipelines.kubeflow.org/component_spec: '{"implementation":
+                              {"container": {"args": ["set -e\necho op11 \"$0\" \"$1\"\n",
+                              {"inputValue": "item"}, {"inputValue": "inner_item"}],
+                              "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+                              "inputs": [{"name": "item", "type": "Integer"}, {"name":
+                              "inner_item", "type": "Integer"}], "name": "my-inner-inner-coop"}'
                             tekton.dev/template: ''
                       timeout: 0s
                     - name: withitem-multiple-nesting-pipeline-for-loop-5
@@ -197,8 +220,11 @@ spec:
                                 steps:
                                 - name: main
                                   args:
-                                  - echo op1 $(inputs.params.loop-item-param-1-subvar-b)
-                                    $(inputs.params.loop-item-param-4)
+                                  - |
+                                    set -e
+                                    echo op11 "$0" "$1"
+                                  - $(inputs.params.loop-item-param-1-subvar-b)
+                                  - $(inputs.params.loop-item-param-4)
                                   command:
                                   - sh
                                   - -c
@@ -214,6 +240,13 @@ spec:
                                     pipelines.kubeflow.org/generation: ''
                                     pipelines.kubeflow.org/cache_enabled: "true"
                                   annotations:
+                                    pipelines.kubeflow.org/component_spec: '{"implementation":
+                                      {"container": {"args": ["set -e\necho op11 \"$0\"
+                                      \"$1\"\n", {"inputValue": "item"}, {"inputValue":
+                                      "inner_item"}], "command": ["sh", "-c"], "image":
+                                      "library/bash:4.4.23"}}, "inputs": [{"name":
+                                      "item", "type": "Integer"}, {"name": "inner_item",
+                                      "type": "Integer"}], "name": "my-inner-inner-coop"}'
                                     tekton.dev/template: ''
                               timeout: 0s
                             - name: withitem-multiple-nesting-pipeline-for-loop-6
@@ -243,8 +276,11 @@ spec:
                                         steps:
                                         - name: main
                                           args:
-                                          - echo op1 $(inputs.params.loop-item-param-1-subvar-b)
-                                            $(inputs.params.my_pipe_param3-loop-item)
+                                          - |
+                                            set -e
+                                            echo op11 "$0" "$1"
+                                          - $(inputs.params.loop-item-param-1-subvar-b)
+                                          - $(inputs.params.my_pipe_param3-loop-item)
                                           command:
                                           - sh
                                           - -c
@@ -260,6 +296,14 @@ spec:
                                             pipelines.kubeflow.org/generation: ''
                                             pipelines.kubeflow.org/cache_enabled: "true"
                                           annotations:
+                                            pipelines.kubeflow.org/component_spec: '{"implementation":
+                                              {"container": {"args": ["set -e\necho
+                                              op11 \"$0\" \"$1\"\n", {"inputValue":
+                                              "item"}, {"inputValue": "inner_item"}],
+                                              "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+                                              "inputs": [{"name": "item", "type":
+                                              "Integer"}, {"name": "inner_item", "type":
+                                              "Integer"}], "name": "my-inner-inner-coop"}'
                                             tekton.dev/template: ''
                                       timeout: 0s
                                   iterateParam: my_pipe_param3-loop-item
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested_noninlined.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested_noninlined.yaml
index b9c1a8f1a7..af5b76cf3d 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested_noninlined.yaml
@@ -27,8 +27,7 @@ metadata:
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "[100, 200]", "name":
       "my_pipe_param", "optional": true, "type": "JsonArray"}, {"default": "[1, 2]",
-      "name": "my_pipe_param3", "optional": true, "type": "JsonArray"}], "name": "withitem
-      multiple nesting pipeline"}'
+      "name": "my_pipe_param3", "optional": true, "type": "JsonArray"}], "name": "withitem-multiple-nesting-pipeline"}'
     tekton.dev/resource_templates: '[{"apiVersion": "custom.tekton.dev/v1alpha1",
       "kind": "PipelineLoop", "metadata": {"name": "withitem-multiple-nesting-pipeline-for-loop-2"},
       "spec": {"iterateParam": "loop-item-param-1", "pipelineSpec": {"params": [{"name":
@@ -36,17 +35,25 @@ metadata:
       "type": "string"}, {"name": "my_pipe_param-loop-item", "type": "string"}, {"name":
       "my_pipe_param3-loop-item", "type": "string"}], "tasks": [{"name": "my-in-coop1",
       "params": [{"name": "loop-item-param-1-subvar-a", "value": "$(params.loop-item-param-1-subvar-a)"}],
-      "taskSpec": {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels":
-      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
-      "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loop-item-param-1-subvar-a",
-      "type": "string"}], "steps": [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-a)"],
+      "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho op1 \\\"$0\\\"\\n\",
+      {\"inputValue\": \"item\"}], \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}},
+      \"inputs\": [{\"name\": \"item\", \"type\": \"Integer\"}], \"name\": \"my-in-coop1\"}",
+      "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
+      ""}}, "params": [{"name": "loop-item-param-1-subvar-a", "type": "string"}],
+      "steps": [{"args": ["set -e\necho op1 \"$0\"\n", "$(inputs.params.loop-item-param-1-subvar-a)"],
       "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
       "timeout": "0s"}, {"name": "my-in-coop2", "params": [{"name": "loop-item-param-1-subvar-b",
       "value": "$(params.loop-item-param-1-subvar-b)"}], "taskSpec": {"metadata":
-      {"annotations": {"tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho op2 \\\"$0\\\"\\n\", {\"inputValue\":
+      \"item\"}], \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}},
+      \"inputs\": [{\"name\": \"item\", \"type\": \"Integer\"}], \"name\": \"my-in-coop2\"}",
+      "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
       "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
       ""}}, "params": [{"name": "loop-item-param-1-subvar-b", "type": "string"}],
-      "steps": [{"args": ["echo op2 $(inputs.params.loop-item-param-1-subvar-b)"],
+      "steps": [{"args": ["set -e\necho op2 \"$0\"\n", "$(inputs.params.loop-item-param-1-subvar-b)"],
       "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
       "timeout": "0s"}, {"name": "withitem-multiple-nesting-pipeline-for-loop-3",
       "params": [{"name": "loop-item-param-1-subvar-a", "value": "$(params.loop-item-param-1-subvar-a)"},
@@ -62,13 +69,18 @@ metadata:
       "my_pipe_param3-loop-item", "type": "string"}], "tasks": [{"name": "my-inner-inner-coop",
       "params": [{"name": "loop-item-param-1-subvar-a", "value": "$(params.loop-item-param-1-subvar-a)"},
       {"name": "my_pipe_param-loop-item", "value": "$(params.my_pipe_param-loop-item)"}],
-      "taskSpec": {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels":
-      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
+      "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho op11 \\\"$0\\\"
+      \\\"$1\\\"\\n\", {\"inputValue\": \"item\"}, {\"inputValue\": \"inner_item\"}],
+      \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\":
+      [{\"name\": \"item\", \"type\": \"Integer\"}, {\"name\": \"inner_item\", \"type\":
+      \"Integer\"}], \"name\": \"my-inner-inner-coop\"}", "tekton.dev/template": ""},
+      "labels": {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
       "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loop-item-param-1-subvar-a",
       "type": "string"}, {"name": "my_pipe_param-loop-item", "type": "string"}], "steps":
-      [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-a) $(inputs.params.my_pipe_param-loop-item)"],
-      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "0s"}, {"name": "withitem-multiple-nesting-pipeline-for-loop-5",
+      [{"args": ["set -e\necho op11 \"$0\" \"$1\"\n", "$(inputs.params.loop-item-param-1-subvar-a)",
+      "$(inputs.params.my_pipe_param-loop-item)"], "command": ["sh", "-c"], "image":
+      "library/bash:4.4.23", "name": "main"}]}, "timeout": "0s"}, {"name": "withitem-multiple-nesting-pipeline-for-loop-5",
       "params": [{"name": "loop-item-param-1-subvar-b", "value": "$(params.loop-item-param-1-subvar-b)"},
       {"name": "loop-item-param-4", "value": "[4, 5]"}, {"name": "my_pipe_param3-loop-item",
       "value": "$(params.my_pipe_param3-loop-item)"}], "taskRef": {"apiVersion": "custom.tekton.dev/v1alpha1",
@@ -80,12 +92,18 @@ metadata:
       "my_pipe_param3-loop-item", "type": "string"}], "tasks": [{"name": "my-inner-inner-coop-2",
       "params": [{"name": "loop-item-param-1-subvar-b", "value": "$(params.loop-item-param-1-subvar-b)"},
       {"name": "loop-item-param-4", "value": "$(params.loop-item-param-4)"}], "taskSpec":
-      {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho op11 \\\"$0\\\" \\\"$1\\\"\\n\",
+      {\"inputValue\": \"item\"}, {\"inputValue\": \"inner_item\"}], \"command\":
+      [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\": [{\"name\":
+      \"item\", \"type\": \"Integer\"}, {\"name\": \"inner_item\", \"type\": \"Integer\"}],
+      \"name\": \"my-inner-inner-coop\"}", "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
       "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
       ""}}, "params": [{"name": "loop-item-param-1-subvar-b", "type": "string"}, {"name":
-      "loop-item-param-4", "type": "string"}], "steps": [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-b)
-      $(inputs.params.loop-item-param-4)"], "command": ["sh", "-c"], "image": "library/bash:4.4.23",
-      "name": "main"}]}, "timeout": "0s"}, {"name": "withitem-multiple-nesting-pipeline-for-loop-6",
+      "loop-item-param-4", "type": "string"}], "steps": [{"args": ["set -e\necho op11
+      \"$0\" \"$1\"\n", "$(inputs.params.loop-item-param-1-subvar-b)", "$(inputs.params.loop-item-param-4)"],
+      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
+      "timeout": "0s"}, {"name": "withitem-multiple-nesting-pipeline-for-loop-6",
       "params": [{"name": "loop-item-param-1-subvar-b", "value": "$(params.loop-item-param-1-subvar-b)"},
       {"name": "my_pipe_param3-loop-item", "value": "$(params.my_pipe_param3-loop-item)"}],
       "taskRef": {"apiVersion": "custom.tekton.dev/v1alpha1", "kind": "PipelineLoop",
@@ -96,13 +114,18 @@ metadata:
       "type": "string"}], "tasks": [{"name": "my-inner-inner-coop-3", "params": [{"name":
       "loop-item-param-1-subvar-b", "value": "$(params.loop-item-param-1-subvar-b)"},
       {"name": "my_pipe_param3-loop-item", "value": "$(params.my_pipe_param3-loop-item)"}],
-      "taskSpec": {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels":
-      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
+      "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho op11 \\\"$0\\\"
+      \\\"$1\\\"\\n\", {\"inputValue\": \"item\"}, {\"inputValue\": \"inner_item\"}],
+      \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\":
+      [{\"name\": \"item\", \"type\": \"Integer\"}, {\"name\": \"inner_item\", \"type\":
+      \"Integer\"}], \"name\": \"my-inner-inner-coop\"}", "tekton.dev/template": ""},
+      "labels": {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
       "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loop-item-param-1-subvar-b",
       "type": "string"}, {"name": "my_pipe_param3-loop-item", "type": "string"}],
-      "steps": [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-b) $(inputs.params.my_pipe_param3-loop-item)"],
-      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "0s"}]}}}]'
+      "steps": [{"args": ["set -e\necho op11 \"$0\" \"$1\"\n", "$(inputs.params.loop-item-param-1-subvar-b)",
+      "$(inputs.params.my_pipe_param3-loop-item)"], "command": ["sh", "-c"], "image":
+      "library/bash:4.4.23", "name": "main"}]}, "timeout": "0s"}]}}}]'
 spec:
   params:
   - name: my_pipe_param
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr1.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr1.yaml
index 9cf1b400b0..5eb8452098 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr1.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr1.yaml
@@ -36,7 +36,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo op1 $(inputs.params.loop-item-param-1-subvar-a)
+          - |
+            set -e
+            echo op1 "$0"
+          - $(inputs.params.loop-item-param-1-subvar-a)
           command:
           - sh
           - -c
@@ -50,6 +53,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho op1 \"$0\"\n", {"inputValue": "item"}], "command":
+              ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs": [{"name":
+              "item", "type": "Integer"}], "name": "my-in-coop1"}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-in-coop2
@@ -60,7 +67,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo op2 $(inputs.params.loop-item-param-1-subvar-b)
+          - |
+            set -e
+            echo op2 "$0"
+          - $(inputs.params.loop-item-param-1-subvar-b)
           command:
           - sh
           - -c
@@ -74,6 +84,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho op2 \"$0\"\n", {"inputValue": "item"}], "command":
+              ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs": [{"name":
+              "item", "type": "Integer"}], "name": "my-in-coop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - name: withitem-multiple-nesting-pipeline-for-loop-3
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr2.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr2.yaml
index fd2e643ed4..2d2183c57f 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr2.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr2.yaml
@@ -38,7 +38,11 @@ spec:
         steps:
         - name: main
           args:
-          - echo op1 $(inputs.params.loop-item-param-1-subvar-a) $(inputs.params.my_pipe_param-loop-item)
+          - |
+            set -e
+            echo op11 "$0" "$1"
+          - $(inputs.params.loop-item-param-1-subvar-a)
+          - $(inputs.params.my_pipe_param-loop-item)
           command:
           - sh
           - -c
@@ -54,6 +58,11 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho op11 \"$0\" \"$1\"\n", {"inputValue": "item"},
+              {"inputValue": "inner_item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item", "type": "Integer"}, {"name": "inner_item",
+              "type": "Integer"}], "name": "my-inner-inner-coop"}'
             tekton.dev/template: ''
       timeout: 0s
     - name: withitem-multiple-nesting-pipeline-for-loop-5
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr3.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr3.yaml
index 9fe9fbc27d..2b70498d31 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr3.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr3.yaml
@@ -36,7 +36,11 @@ spec:
         steps:
         - name: main
           args:
-          - echo op1 $(inputs.params.loop-item-param-1-subvar-b) $(inputs.params.loop-item-param-4)
+          - |
+            set -e
+            echo op11 "$0" "$1"
+          - $(inputs.params.loop-item-param-1-subvar-b)
+          - $(inputs.params.loop-item-param-4)
           command:
           - sh
           - -c
@@ -52,6 +56,11 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho op11 \"$0\" \"$1\"\n", {"inputValue": "item"},
+              {"inputValue": "inner_item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item", "type": "Integer"}, {"name": "inner_item",
+              "type": "Integer"}], "name": "my-inner-inner-coop"}'
             tekton.dev/template: ''
       timeout: 0s
     - name: withitem-multiple-nesting-pipeline-for-loop-6
diff --git a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr4.yaml b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr4.yaml
index 47b28d6b7d..c40f1f32a4 100644
--- a/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr4.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_multi_nested_pipelineloop_cr4.yaml
@@ -19,10 +19,10 @@ metadata:
 spec:
   pipelineSpec:
     params:
-    - name: my_pipe_param3-loop-item
-      type: string
     - name: loop-item-param-1-subvar-b
       type: string
+    - name: my_pipe_param3-loop-item
+      type: string
     tasks:
     - name: my-inner-inner-coop-3
       params:
@@ -34,7 +34,11 @@ spec:
         steps:
         - name: main
           args:
-          - echo op1 $(inputs.params.loop-item-param-1-subvar-b) $(inputs.params.my_pipe_param3-loop-item)
+          - |
+            set -e
+            echo op11 "$0" "$1"
+          - $(inputs.params.loop-item-param-1-subvar-b)
+          - $(inputs.params.my_pipe_param3-loop-item)
           command:
           - sh
           - -c
@@ -50,6 +54,11 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho op11 \"$0\" \"$1\"\n", {"inputValue": "item"},
+              {"inputValue": "inner_item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item", "type": "Integer"}, {"name": "inner_item",
+              "type": "Integer"}], "name": "my-inner-inner-coop"}'
             tekton.dev/template: ''
       timeout: 0s
   iterateParam: my_pipe_param3-loop-item
diff --git a/sdk/python/tests/compiler/testdata/withitem_nested.py b/sdk/python/tests/compiler/testdata/withitem_nested.py
index 593b10ce98..067a6d2169 100644
--- a/sdk/python/tests/compiler/testdata/withitem_nested.py
+++ b/sdk/python/tests/compiler/testdata/withitem_nested.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import kfp.dsl as dsl
+from kfp import components
 from kfp_tekton.compiler import TektonCompiler
 
 
@@ -23,39 +24,89 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op1_yaml = '''\
+name: 'my-in-coop1'
+inputs:
+- {name: item, type: Integer}
+- {name: my_pipe_param, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op1 "$0" "$1"
+        - {inputValue: item}
+        - {inputValue: my_pipe_param}
+'''
 
-@dsl.pipeline(name='with-item-nested-pipeline')
+op11_yaml = '''\
+name: 'my-inner-inner-coop'
+inputs:
+- {name: item, type: Integer}
+- {name: inner_item, type: Integer}
+- {name: my_pipe_param, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op11 "$0" "$1" "$2"
+        - {inputValue: item}
+        - {inputValue: inner_item}
+        - {inputValue: my_pipe_param}
+'''
+
+op2_yaml = '''\
+name: 'my-in-coop2'
+inputs:
+- {name: item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo op2 "$0"
+        - {inputValue: item}
+'''
+
+op_out_yaml = '''\
+name: 'my-out-cop'
+inputs:
+- {name: my_pipe_param, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo "$0"
+        - {inputValue: my_pipe_param}
+'''
+
+
+@dsl.pipeline(name='my-pipeline')
 def pipeline(my_pipe_param: int = 10):
-    loop_args = [{'a': 1, 'b': 2}, {'a': 10, 'b': 20}]
+    loop_args = [1, 2]
     with dsl.ParallelFor(loop_args) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-coop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo op1 %s %s" % (item.a, my_pipe_param)],
-        )
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item, my_pipe_param)
 
         with dsl.ParallelFor([100, 200, 300]) as inner_item:
-            op11 = dsl.ContainerOp(
-                name="my-inner-inner-coop",
-                image="library/bash:4.4.23",
-                command=["sh", "-c"],
-                arguments=["echo op1 %s %s %s" % (item.a, inner_item, my_pipe_param)],
-            )
-
-        op2 = dsl.ContainerOp(
-            name="my-in-coop2",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo op2 %s" % item.b],
-        )
-
-    op_out = dsl.ContainerOp(
-        name="my-out-cop",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo %s" % my_pipe_param],
-    )
+            op11_template = components.load_component_from_text(op11_yaml)
+            op11 = op11_template(item, inner_item, my_pipe_param)
+
+        op2_template = components.load_component_from_text(op2_yaml)
+        op2 = op2_template(item)
+
+    op_out_template = components.load_component_from_text(op_out_yaml)
+    op_out = op_out_template(my_pipe_param)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withitem_nested.yaml b/sdk/python/tests/compiler/testdata/withitem_nested.yaml
index ab7e441014..80132e2efe 100644
--- a/sdk/python/tests/compiler/testdata/withitem_nested.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_nested.yaml
@@ -15,7 +15,7 @@
 apiVersion: tekton.dev/v1beta1
 kind: PipelineRun
 metadata:
-  name: with-item-nested-pipeline
+  name: my-pipeline
   annotations:
     tekton.dev/output_artifacts: '{}'
     tekton.dev/input_artifacts: '{}'
@@ -26,7 +26,7 @@ metadata:
       [], "my-out-cop": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "my_pipe_param",
-      "optional": true, "type": "Integer"}], "name": "with-item-nested-pipeline"}'
+      "optional": true, "type": "Integer"}], "name": "my-pipeline"}'
 spec:
   params:
   - name: my_pipe_param
@@ -44,7 +44,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo $(inputs.params.my_pipe_param)
+          - |
+            set -e
+            echo "$0"
+          - $(inputs.params.my_pipe_param)
           command:
           - sh
           - -c
@@ -57,12 +60,16 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho \"$0\"\n", {"inputValue": "my_pipe_param"}],
+              "command": ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs":
+              [{"name": "my_pipe_param", "type": "Integer"}], "name": "my-out-cop"}'
             tekton.dev/template: ''
       timeout: 0s
-    - name: with-item-nested-pipeline-for-loop-2
+    - name: my-pipeline-for-loop-2
       params:
       - name: loop-item-param-1
-        value: '[{"a": 1, "b": 2}, {"a": 10, "b": 20}]'
+        value: '[1, 2]'
       - name: my_pipe_param
         value: $(params.my_pipe_param)
       taskSpec:
@@ -71,30 +78,32 @@ spec:
         spec:
           pipelineSpec:
             params:
-            - name: loop-item-param-1-subvar-a
-              type: string
-            - name: loop-item-param-1-subvar-b
+            - name: loop-item-param-1
               type: string
             - name: my_pipe_param
               type: string
             tasks:
             - name: my-in-coop1
               params:
-              - name: loop-item-param-1-subvar-a
-                value: $(params.loop-item-param-1-subvar-a)
+              - name: loop-item-param-1
+                value: $(params.loop-item-param-1)
               - name: my_pipe_param
                 value: $(params.my_pipe_param)
               taskSpec:
                 steps:
                 - name: main
                   args:
-                  - echo op1 $(inputs.params.loop-item-param-1-subvar-a) $(inputs.params.my_pipe_param)
+                  - |
+                    set -e
+                    echo op1 "$0" "$1"
+                  - $(inputs.params.loop-item-param-1)
+                  - $(inputs.params.my_pipe_param)
                   command:
                   - sh
                   - -c
                   image: library/bash:4.4.23
                 params:
-                - name: loop-item-param-1-subvar-a
+                - name: loop-item-param-1
                   type: string
                 - name: my_pipe_param
                   type: string
@@ -104,23 +113,32 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho op1 \"$0\" \"$1\"\n", {"inputValue":
+                      "item"}, {"inputValue": "my_pipe_param"}], "command": ["sh",
+                      "-c"], "image": "library/bash:4.4.23"}}, "inputs": [{"name":
+                      "item", "type": "Integer"}, {"name": "my_pipe_param", "type":
+                      "Integer"}], "name": "my-in-coop1"}'
                     tekton.dev/template: ''
               timeout: 0s
             - name: my-in-coop2
               params:
-              - name: loop-item-param-1-subvar-b
-                value: $(params.loop-item-param-1-subvar-b)
+              - name: loop-item-param-1
+                value: $(params.loop-item-param-1)
               taskSpec:
                 steps:
                 - name: main
                   args:
-                  - echo op2 $(inputs.params.loop-item-param-1-subvar-b)
+                  - |
+                    set -e
+                    echo op2 "$0"
+                  - $(inputs.params.loop-item-param-1)
                   command:
                   - sh
                   - -c
                   image: library/bash:4.4.23
                 params:
-                - name: loop-item-param-1-subvar-b
+                - name: loop-item-param-1
                   type: string
                 metadata:
                   labels:
@@ -128,12 +146,16 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho op2 \"$0\"\n", {"inputValue": "item"}],
+                      "command": ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs":
+                      [{"name": "item", "type": "Integer"}], "name": "my-in-coop2"}'
                     tekton.dev/template: ''
               timeout: 0s
-            - name: with-item-nested-pipeline-for-loop-4
+            - name: my-pipeline-for-loop-4
               params:
-              - name: loop-item-param-1-subvar-a
-                value: $(params.loop-item-param-1-subvar-a)
+              - name: loop-item-param-1
+                value: $(params.loop-item-param-1)
               - name: loop-item-param-3
                 value: '[100, 200, 300]'
               - name: my_pipe_param
@@ -144,7 +166,7 @@ spec:
                 spec:
                   pipelineSpec:
                     params:
-                    - name: loop-item-param-1-subvar-a
+                    - name: loop-item-param-1
                       type: string
                     - name: loop-item-param-3
                       type: string
@@ -153,8 +175,8 @@ spec:
                     tasks:
                     - name: my-inner-inner-coop
                       params:
-                      - name: loop-item-param-1-subvar-a
-                        value: $(params.loop-item-param-1-subvar-a)
+                      - name: loop-item-param-1
+                        value: $(params.loop-item-param-1)
                       - name: loop-item-param-3
                         value: $(params.loop-item-param-3)
                       - name: my_pipe_param
@@ -163,14 +185,18 @@ spec:
                         steps:
                         - name: main
                           args:
-                          - echo op1 $(inputs.params.loop-item-param-1-subvar-a) $(inputs.params.loop-item-param-3)
-                            $(inputs.params.my_pipe_param)
+                          - |
+                            set -e
+                            echo op11 "$0" "$1" "$2"
+                          - $(inputs.params.loop-item-param-1)
+                          - $(inputs.params.loop-item-param-3)
+                          - $(inputs.params.my_pipe_param)
                           command:
                           - sh
                           - -c
                           image: library/bash:4.4.23
                         params:
-                        - name: loop-item-param-1-subvar-a
+                        - name: loop-item-param-1
                           type: string
                         - name: loop-item-param-3
                           type: string
@@ -182,6 +208,14 @@ spec:
                             pipelines.kubeflow.org/generation: ''
                             pipelines.kubeflow.org/cache_enabled: "true"
                           annotations:
+                            pipelines.kubeflow.org/component_spec: '{"implementation":
+                              {"container": {"args": ["set -e\necho op11 \"$0\" \"$1\"
+                              \"$2\"\n", {"inputValue": "item"}, {"inputValue": "inner_item"},
+                              {"inputValue": "my_pipe_param"}], "command": ["sh",
+                              "-c"], "image": "library/bash:4.4.23"}}, "inputs": [{"name":
+                              "item", "type": "Integer"}, {"name": "inner_item", "type":
+                              "Integer"}, {"name": "my_pipe_param", "type": "Integer"}],
+                              "name": "my-inner-inner-coop"}'
                             tekton.dev/template: ''
                       timeout: 0s
                   iterateParam: loop-item-param-3
diff --git a/sdk/python/tests/compiler/testdata/withitem_nested_noninlined.yaml b/sdk/python/tests/compiler/testdata/withitem_nested_noninlined.yaml
index 49b08a913b..a43624cde6 100644
--- a/sdk/python/tests/compiler/testdata/withitem_nested_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withitem_nested_noninlined.yaml
@@ -15,7 +15,7 @@
 apiVersion: tekton.dev/v1beta1
 kind: PipelineRun
 metadata:
-  name: with-item-nested-pipeline
+  name: my-pipeline
   annotations:
     tekton.dev/output_artifacts: '{}'
     tekton.dev/input_artifacts: '{}'
@@ -26,44 +26,58 @@ metadata:
       [], "my-out-cop": []}'
     sidecar.istio.io/inject: "false"
     pipelines.kubeflow.org/pipeline_spec: '{"inputs": [{"default": "10", "name": "my_pipe_param",
-      "optional": true, "type": "Integer"}], "name": "with-item-nested-pipeline"}'
+      "optional": true, "type": "Integer"}], "name": "my-pipeline"}'
     tekton.dev/resource_templates: '[{"apiVersion": "custom.tekton.dev/v1alpha1",
-      "kind": "PipelineLoop", "metadata": {"name": "with-item-nested-pipeline-for-loop-2"},
-      "spec": {"iterateParam": "loop-item-param-1", "pipelineSpec": {"params": [{"name":
-      "loop-item-param-1-subvar-a", "type": "string"}, {"name": "loop-item-param-1-subvar-b",
+      "kind": "PipelineLoop", "metadata": {"name": "my-pipeline-for-loop-2"}, "spec":
+      {"iterateParam": "loop-item-param-1", "pipelineSpec": {"params": [{"name": "loop-item-param-1",
       "type": "string"}, {"name": "my_pipe_param", "type": "string"}], "tasks": [{"name":
-      "my-in-coop1", "params": [{"name": "loop-item-param-1-subvar-a", "value": "$(params.loop-item-param-1-subvar-a)"},
+      "my-in-coop1", "params": [{"name": "loop-item-param-1", "value": "$(params.loop-item-param-1)"},
       {"name": "my_pipe_param", "value": "$(params.my_pipe_param)"}], "taskSpec":
-      {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho op1 \\\"$0\\\" \\\"$1\\\"\\n\",
+      {\"inputValue\": \"item\"}, {\"inputValue\": \"my_pipe_param\"}], \"command\":
+      [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\": [{\"name\":
+      \"item\", \"type\": \"Integer\"}, {\"name\": \"my_pipe_param\", \"type\": \"Integer\"}],
+      \"name\": \"my-in-coop1\"}", "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
       "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
-      ""}}, "params": [{"name": "loop-item-param-1-subvar-a", "type": "string"}, {"name":
-      "my_pipe_param", "type": "string"}], "steps": [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-a)
-      $(inputs.params.my_pipe_param)"], "command": ["sh", "-c"], "image": "library/bash:4.4.23",
-      "name": "main"}]}, "timeout": "0s"}, {"name": "my-in-coop2", "params": [{"name":
-      "loop-item-param-1-subvar-b", "value": "$(params.loop-item-param-1-subvar-b)"}],
-      "taskSpec": {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels":
-      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
-      "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loop-item-param-1-subvar-b",
-      "type": "string"}], "steps": [{"args": ["echo op2 $(inputs.params.loop-item-param-1-subvar-b)"],
+      ""}}, "params": [{"name": "loop-item-param-1", "type": "string"}, {"name": "my_pipe_param",
+      "type": "string"}], "steps": [{"args": ["set -e\necho op1 \"$0\" \"$1\"\n",
+      "$(inputs.params.loop-item-param-1)", "$(inputs.params.my_pipe_param)"], "command":
+      ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout":
+      "0s"}, {"name": "my-in-coop2", "params": [{"name": "loop-item-param-1", "value":
+      "$(params.loop-item-param-1)"}], "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho op2 \\\"$0\\\"\\n\",
+      {\"inputValue\": \"item\"}], \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}},
+      \"inputs\": [{\"name\": \"item\", \"type\": \"Integer\"}], \"name\": \"my-in-coop2\"}",
+      "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
+      ""}}, "params": [{"name": "loop-item-param-1", "type": "string"}], "steps":
+      [{"args": ["set -e\necho op2 \"$0\"\n", "$(inputs.params.loop-item-param-1)"],
       "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "0s"}, {"name": "with-item-nested-pipeline-for-loop-4", "params":
-      [{"name": "loop-item-param-1-subvar-a", "value": "$(params.loop-item-param-1-subvar-a)"},
-      {"name": "loop-item-param-3", "value": "[100, 200, 300]"}, {"name": "my_pipe_param",
-      "value": "$(params.my_pipe_param)"}], "taskRef": {"apiVersion": "custom.tekton.dev/v1alpha1",
-      "kind": "PipelineLoop", "name": "with-item-nested-pipeline-for-loop-4"}}]}}},
-      {"apiVersion": "custom.tekton.dev/v1alpha1", "kind": "PipelineLoop", "metadata":
-      {"name": "with-item-nested-pipeline-for-loop-4"}, "spec": {"iterateParam": "loop-item-param-3",
-      "pipelineSpec": {"params": [{"name": "loop-item-param-1-subvar-a", "type": "string"},
-      {"name": "loop-item-param-3", "type": "string"}, {"name": "my_pipe_param", "type":
-      "string"}], "tasks": [{"name": "my-inner-inner-coop", "params": [{"name": "loop-item-param-1-subvar-a",
-      "value": "$(params.loop-item-param-1-subvar-a)"}, {"name": "loop-item-param-3",
-      "value": "$(params.loop-item-param-3)"}, {"name": "my_pipe_param", "value":
-      "$(params.my_pipe_param)"}], "taskSpec": {"metadata": {"annotations": {"tekton.dev/template":
-      ""}, "labels": {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
-      "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loop-item-param-1-subvar-a",
+      "timeout": "0s"}, {"name": "my-pipeline-for-loop-4", "params": [{"name": "loop-item-param-1",
+      "value": "$(params.loop-item-param-1)"}, {"name": "loop-item-param-3", "value":
+      "[100, 200, 300]"}, {"name": "my_pipe_param", "value": "$(params.my_pipe_param)"}],
+      "taskRef": {"apiVersion": "custom.tekton.dev/v1alpha1", "kind": "PipelineLoop",
+      "name": "my-pipeline-for-loop-4"}}]}}}, {"apiVersion": "custom.tekton.dev/v1alpha1",
+      "kind": "PipelineLoop", "metadata": {"name": "my-pipeline-for-loop-4"}, "spec":
+      {"iterateParam": "loop-item-param-3", "pipelineSpec": {"params": [{"name": "loop-item-param-1",
       "type": "string"}, {"name": "loop-item-param-3", "type": "string"}, {"name":
-      "my_pipe_param", "type": "string"}], "steps": [{"args": ["echo op1 $(inputs.params.loop-item-param-1-subvar-a)
-      $(inputs.params.loop-item-param-3) $(inputs.params.my_pipe_param)"], "command":
+      "my_pipe_param", "type": "string"}], "tasks": [{"name": "my-inner-inner-coop",
+      "params": [{"name": "loop-item-param-1", "value": "$(params.loop-item-param-1)"},
+      {"name": "loop-item-param-3", "value": "$(params.loop-item-param-3)"}, {"name":
+      "my_pipe_param", "value": "$(params.my_pipe_param)"}], "taskSpec": {"metadata":
+      {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho op11 \\\"$0\\\" \\\"$1\\\" \\\"$2\\\"\\n\",
+      {\"inputValue\": \"item\"}, {\"inputValue\": \"inner_item\"}, {\"inputValue\":
+      \"my_pipe_param\"}], \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}},
+      \"inputs\": [{\"name\": \"item\", \"type\": \"Integer\"}, {\"name\": \"inner_item\",
+      \"type\": \"Integer\"}, {\"name\": \"my_pipe_param\", \"type\": \"Integer\"}],
+      \"name\": \"my-inner-inner-coop\"}", "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
+      ""}}, "params": [{"name": "loop-item-param-1", "type": "string"}, {"name": "loop-item-param-3",
+      "type": "string"}, {"name": "my_pipe_param", "type": "string"}], "steps": [{"args":
+      ["set -e\necho op11 \"$0\" \"$1\" \"$2\"\n", "$(inputs.params.loop-item-param-1)",
+      "$(inputs.params.loop-item-param-3)", "$(inputs.params.my_pipe_param)"], "command":
       ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout":
       "0s"}]}}}]'
 spec:
@@ -83,7 +97,10 @@ spec:
         steps:
         - name: main
           args:
-          - echo $(inputs.params.my_pipe_param)
+          - |
+            set -e
+            echo "$0"
+          - $(inputs.params.my_pipe_param)
           command:
           - sh
           - -c
@@ -96,16 +113,20 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho \"$0\"\n", {"inputValue": "my_pipe_param"}],
+              "command": ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs":
+              [{"name": "my_pipe_param", "type": "Integer"}], "name": "my-out-cop"}'
             tekton.dev/template: ''
       timeout: 0s
-    - name: with-item-nested-pipeline-for-loop-2
+    - name: my-pipeline-for-loop-2
       taskRef:
         apiVersion: custom.tekton.dev/v1alpha1
         kind: PipelineLoop
-        name: with-item-nested-pipeline-for-loop-2
+        name: my-pipeline-for-loop-2
       params:
       - name: loop-item-param-1
-        value: '[{"a": 1, "b": 2}, {"a": 10, "b": 20}]'
+        value: '[1, 2]'
       - name: my_pipe_param
         value: $(params.my_pipe_param)
   timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withparam_global.py b/sdk/python/tests/compiler/testdata/withparam_global.py
index 3f0597a3d2..df61bacd53 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global.py
+++ b/sdk/python/tests/compiler/testdata/withparam_global.py
@@ -14,6 +14,7 @@
 
 import kfp.dsl as dsl
 from kfp_tekton.compiler import TektonCompiler
+from kfp import components
 
 
 class Coder:
@@ -23,32 +24,63 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op0_yaml = '''\
+name: 'my-out-cop0'
+outputs:
+  - {name: out, type: String}
+implementation:
+    container:
+        image: python:alpine3.6
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+        - {outputPath: out}
+'''
+
+op1_yaml = '''\
+name: 'my-in-cop1'
+inputs:
+- {name: item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op1, item: "$0"
+        - {inputValue: item}
+'''
+
+op_out_yaml = '''\
+name: 'my-out-cop2'
+inputs:
+- {name: output, type: String}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op2, outp: "$0"
+        - {inputValue: output}
+'''
+
 
 @dsl.pipeline(name='withparam-global')
 def pipeline(loopidy_doop: list = [3, 5, 7, 9]):
-    op0 = dsl.ContainerOp(
-        name="my-out-cop0",
-        image='python:alpine3.6',
-        command=["sh", "-c"],
-        arguments=[
-            'python -c "import json; import sys; json.dump([i for i in range(20, 31)], open(\'/tmp/out.json\', \'w\'))"'],
-        file_outputs={'out': '/tmp/out.json'},
-    )
+    op0_template = components.load_component_from_text(op0_yaml)
+    op0 = op0_template()
 
     with dsl.ParallelFor(loopidy_doop) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-cop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo no output global op1, item: %s" % item],
-        ).after(op0)
-
-    op_out = dsl.ContainerOp(
-        name="my-out-cop2",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo no output global op2, outp: %s" % op0.output],
-    ).after(op1)
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item).after(op0)
+
+    op_out_template = components.load_component_from_text(op_out_yaml)
+    op_out = op_out_template(op0.output).after(op1)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withparam_global.yaml b/sdk/python/tests/compiler/testdata/withparam_global.yaml
index 3a0c8360f2..9fbb52faa3 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-global
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -43,21 +43,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -68,7 +75,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -81,6 +91,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       runAfter:
       - withparam-global-for-loop-1
@@ -108,7 +122,10 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - 'echo no output global op1, item: $(inputs.params.loopidy_doop-loop-item)'
+                  - |
+                    set -e
+                    echo no output global op1, item: "$0"
+                  - $(inputs.params.loopidy_doop-loop-item)
                   command:
                   - sh
                   - -c
@@ -122,6 +139,10 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho no output global op1, item: \"$0\"\n",
+                      {"inputValue": "item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+                      "inputs": [{"name": "item", "type": "Integer"}], "name": "my-in-cop1"}'
                     tekton.dev/template: ''
               runAfter: []
               timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_dict.py b/sdk/python/tests/compiler/testdata/withparam_global_dict.py
index 6de8b8f048..8d26b251cc 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_dict.py
+++ b/sdk/python/tests/compiler/testdata/withparam_global_dict.py
@@ -14,6 +14,7 @@
 
 import kfp.dsl as dsl
 from kfp_tekton.compiler import TektonCompiler
+from kfp import components
 
 
 class Coder:
@@ -23,31 +24,65 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op0_yaml = '''\
+name: 'my-out-cop0'
+outputs:
+  - {name: out, type: String}
+implementation:
+    container:
+        image: python:alpine3.6
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+        - {outputPath: out}
+'''
+
+op1_yaml = '''\
+name: 'my-in-cop1'
+inputs:
+- {name: item-a, type: Integer}
+- {name: item-b, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op1, item.a: "$0" , item.b: "$1"
+        - {inputValue: item-a}
+        - {inputValue: item-b}
+'''
+
+op_out_yaml = '''\
+name: 'my-out-cop2'
+inputs:
+- {name: output, type: String}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op2, outp: "$0"
+        - {inputValue: output}
+'''
+
 
 @dsl.pipeline(name='withparam-global-dict')
 def pipeline(loopidy_doop: dict = [{'a': '1', 'b': '2'}, {'a': '10', 'b': '20'}]):
-    op0 = dsl.ContainerOp(
-        name="my-out-cop0",
-        image='python:alpine3.6',
-        command=["sh", "-c"],
-        arguments=['python -c "import json; import sys; json.dump([i for i in range(20, 31)], open(\'/tmp/out.json\', \'w\'))"'],
-        file_outputs={'out': '/tmp/out.json'},
-    )
+    op0_template = components.load_component_from_text(op0_yaml)
+    op0 = op0_template()
 
     with dsl.ParallelFor(loopidy_doop) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-cop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo no output global op1, item.a: %s, item.b: %s" % (item.a, item.b)],
-        ).after(op0)
-
-    op_out = dsl.ContainerOp(
-        name="my-out-cop2",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo no output global op2, outp: %s" % op0.output],
-    )
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item.a, item.b).after(op0)
+
+    op_out_template = components.load_component_from_text(op_out_yaml)
+    op_out = op_out_template(op0.output)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_dict.yaml b/sdk/python/tests/compiler/testdata/withparam_global_dict.yaml
index 5b2e452bd9..b1e2fc0c1a 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_dict.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global_dict.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-global-dict
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -44,21 +44,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -69,7 +76,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -82,6 +92,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
@@ -111,8 +125,11 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - 'echo no output global op1, item.a: $(inputs.params.loopidy_doop-loop-item-subvar-a),
-                    item.b: $(inputs.params.loopidy_doop-loop-item-subvar-b)'
+                  - |
+                    set -e
+                    echo no output global op1, item.a: "$0" , item.b: "$1"
+                  - $(inputs.params.loopidy_doop-loop-item-subvar-a)
+                  - $(inputs.params.loopidy_doop-loop-item-subvar-b)
                   command:
                   - sh
                   - -c
@@ -128,6 +145,12 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho no output global op1, item.a: \"$0\"
+                      , item.b: \"$1\"\n", {"inputValue": "item-a"}, {"inputValue":
+                      "item-b"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+                      "inputs": [{"name": "item-a", "type": "Integer"}, {"name": "item-b",
+                      "type": "Integer"}], "name": "my-in-cop1"}'
                     tekton.dev/template: ''
               runAfter: []
               timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_dict_noninlined.yaml b/sdk/python/tests/compiler/testdata/withparam_global_dict_noninlined.yaml
index e4a0d44844..cffd8b0a05 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_dict_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global_dict_noninlined.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-global-dict
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -37,13 +37,19 @@ metadata:
       "type": "string"}], "tasks": [{"name": "my-in-cop1", "params": [{"name": "loopidy_doop-loop-item-subvar-a",
       "value": "$(params.loopidy_doop-loop-item-subvar-a)"}, {"name": "loopidy_doop-loop-item-subvar-b",
       "value": "$(params.loopidy_doop-loop-item-subvar-b)"}], "runAfter": [], "taskSpec":
-      {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
-      "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
-      ""}}, "params": [{"name": "loopidy_doop-loop-item-subvar-a", "type": "string"},
-      {"name": "loopidy_doop-loop-item-subvar-b", "type": "string"}], "steps": [{"args":
-      ["echo no output global op1, item.a: $(inputs.params.loopidy_doop-loop-item-subvar-a),
-      item.b: $(inputs.params.loopidy_doop-loop-item-subvar-b)"], "command": ["sh",
-      "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout": "0s"}]}}}]'
+      {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho no output global op1, item.a: \\\"$0\\\"
+      , item.b: \\\"$1\\\"\\n\", {\"inputValue\": \"item-a\"}, {\"inputValue\": \"item-b\"}],
+      \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\":
+      [{\"name\": \"item-a\", \"type\": \"Integer\"}, {\"name\": \"item-b\", \"type\":
+      \"Integer\"}], \"name\": \"my-in-cop1\"}", "tekton.dev/template": ""}, "labels":
+      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
+      "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loopidy_doop-loop-item-subvar-a",
+      "type": "string"}, {"name": "loopidy_doop-loop-item-subvar-b", "type": "string"}],
+      "steps": [{"args": ["set -e\necho no output global op1, item.a: \"$0\" , item.b:
+      \"$1\"\n", "$(inputs.params.loopidy_doop-loop-item-subvar-a)", "$(inputs.params.loopidy_doop-loop-item-subvar-b)"],
+      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
+      "timeout": "0s"}]}}}]'
 spec:
   params:
   - name: loopidy_doop
@@ -58,21 +64,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -83,7 +96,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -96,6 +112,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_dict_pipelineloop_cr1.yaml b/sdk/python/tests/compiler/testdata/withparam_global_dict_pipelineloop_cr1.yaml
index 193cd160cd..1909dafbdc 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_dict_pipelineloop_cr1.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global_dict_pipelineloop_cr1.yaml
@@ -34,8 +34,11 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op1, item.a: $(inputs.params.loopidy_doop-loop-item-subvar-a),
-            item.b: $(inputs.params.loopidy_doop-loop-item-subvar-b)'
+          - |
+            set -e
+            echo no output global op1, item.a: "$0" , item.b: "$1"
+          - $(inputs.params.loopidy_doop-loop-item-subvar-a)
+          - $(inputs.params.loopidy_doop-loop-item-subvar-b)
           command:
           - sh
           - -c
@@ -51,6 +54,12 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op1, item.a: \"$0\" , item.b:
+              \"$1\"\n", {"inputValue": "item-a"}, {"inputValue": "item-b"}], "command":
+              ["sh", "-c"], "image": "library/bash:4.4.23"}}, "inputs": [{"name":
+              "item-a", "type": "Integer"}, {"name": "item-b", "type": "Integer"}],
+              "name": "my-in-cop1"}'
             tekton.dev/template: ''
       runAfter: []
       timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_noninlined.yaml b/sdk/python/tests/compiler/testdata/withparam_global_noninlined.yaml
index 9e25c6aa05..01a9cda352 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global_noninlined.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-global
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -34,12 +34,16 @@ metadata:
       "spec": {"iterateParam": "loopidy_doop-loop-item", "pipelineSpec": {"params":
       [{"name": "loopidy_doop-loop-item", "type": "string"}], "tasks": [{"name": "my-in-cop1",
       "params": [{"name": "loopidy_doop-loop-item", "value": "$(params.loopidy_doop-loop-item)"}],
-      "runAfter": [], "taskSpec": {"metadata": {"annotations": {"tekton.dev/template":
+      "runAfter": [], "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho no output
+      global op1, item: \\\"$0\\\"\\n\", {\"inputValue\": \"item\"}], \"command\":
+      [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\": [{\"name\":
+      \"item\", \"type\": \"Integer\"}], \"name\": \"my-in-cop1\"}", "tekton.dev/template":
       ""}, "labels": {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
       "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "loopidy_doop-loop-item",
-      "type": "string"}], "steps": [{"args": ["echo no output global op1, item: $(inputs.params.loopidy_doop-loop-item)"],
-      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "0s"}]}}}]'
+      "type": "string"}], "steps": [{"args": ["set -e\necho no output global op1,
+      item: \"$0\"\n", "$(inputs.params.loopidy_doop-loop-item)"], "command": ["sh",
+      "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout": "0s"}]}}}]'
 spec:
   params:
   - name: loopidy_doop
@@ -54,21 +58,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -79,7 +90,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -92,6 +106,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       runAfter:
       - withparam-global-for-loop-1
diff --git a/sdk/python/tests/compiler/testdata/withparam_global_pipelineloop_cr1.yaml b/sdk/python/tests/compiler/testdata/withparam_global_pipelineloop_cr1.yaml
index 3f4c7af0f2..2dfe8290ce 100644
--- a/sdk/python/tests/compiler/testdata/withparam_global_pipelineloop_cr1.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_global_pipelineloop_cr1.yaml
@@ -30,7 +30,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo no output global op1, item: $(inputs.params.loopidy_doop-loop-item)'
+          - |
+            set -e
+            echo no output global op1, item: "$0"
+          - $(inputs.params.loopidy_doop-loop-item)
           command:
           - sh
           - -c
@@ -44,6 +47,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op1, item: \"$0\"\n", {"inputValue":
+              "item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item", "type": "Integer"}], "name": "my-in-cop1"}'
             tekton.dev/template: ''
       runAfter: []
       timeout: 0s
diff --git a/sdk/python/tests/compiler/testdata/withparam_output.py b/sdk/python/tests/compiler/testdata/withparam_output.py
index 7d64dabfd4..f39f0a9904 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output.py
+++ b/sdk/python/tests/compiler/testdata/withparam_output.py
@@ -14,6 +14,7 @@
 
 import kfp.dsl as dsl
 from kfp_tekton.compiler import TektonCompiler
+from kfp import components
 
 
 class Coder:
@@ -23,32 +24,63 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op0_yaml = '''\
+name: 'my-out-cop0'
+outputs:
+  - {name: out, type: String}
+implementation:
+    container:
+        image: python:alpine3.6
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+        - {outputPath: out}
+'''
+
+op1_yaml = '''\
+name: 'my-in-cop1'
+inputs:
+- {name: item, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op1, item: "$0"
+        - {inputValue: item}
+'''
+
+op_out_yaml = '''\
+name: 'my-out-cop2'
+inputs:
+- {name: output, type: String}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op2, outp: "$0"
+        - {inputValue: output}
+'''
+
 
 @dsl.pipeline(name='withparam-output')
 def pipeline():
-    op0 = dsl.ContainerOp(
-        name="my-out-cop0",
-        image='python:alpine3.6',
-        command=["sh", "-c"],
-        arguments=[
-            'python -c "import json; import sys; json.dump([i for i in range(20, 31)], open(\'/tmp/out.json\', \'w\'))"'],
-        file_outputs={'out': '/tmp/out.json'},
-    )
+    op0_template = components.load_component_from_text(op0_yaml)
+    op0 = op0_template()
 
     with dsl.ParallelFor(op0.output) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-cop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo do output op1 item: %s" % item],
-        )
-
-    op_out = dsl.ContainerOp(
-        name="my-out-cop2",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo do output op2, outp: %s" % op0.output],
-    )
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item)
+
+    op_out_template = components.load_component_from_text(op_out_yaml)
+    op_out = op_out_template(op0.output)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withparam_output.yaml b/sdk/python/tests/compiler/testdata/withparam_output.yaml
index c508ce3188..923a5dcbe0 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-output
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -36,21 +36,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -61,7 +68,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -74,6 +84,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
@@ -99,7 +113,10 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - 'echo do output op1 item: $(inputs.params.my-out-cop0-out-loop-item)'
+                  - |
+                    set -e
+                    echo no output global op1, item: "$0"
+                  - $(inputs.params.my-out-cop0-out-loop-item)
                   command:
                   - sh
                   - -c
@@ -113,6 +130,10 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho no output global op1, item: \"$0\"\n",
+                      {"inputValue": "item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+                      "inputs": [{"name": "item", "type": "Integer"}], "name": "my-in-cop1"}'
                     tekton.dev/template: ''
               timeout: 0s
           iterateParam: my-out-cop0-out-loop-item
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_dict.py b/sdk/python/tests/compiler/testdata/withparam_output_dict.py
index d55b93bb9e..d8b431a094 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_dict.py
+++ b/sdk/python/tests/compiler/testdata/withparam_output_dict.py
@@ -14,6 +14,7 @@
 
 import kfp.dsl as dsl
 from kfp_tekton.compiler import TektonCompiler
+from kfp import components
 
 
 class Coder:
@@ -23,33 +24,63 @@ def empty(self):
 
 TektonCompiler._get_unique_id_code = Coder.empty
 
+op0_yaml = '''\
+name: 'my-out-cop0'
+outputs:
+  - {name: out, type: String}
+implementation:
+    container:
+        image: python:alpine3.6
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          python -c "import json; import sys; json.dump([{\'a\': 1, \'b\': 2}, {\'a\': 10, \'b\': 20}], open('$0', 'w'))"
+        - {outputPath: out}
+'''
+
+op1_yaml = '''\
+name: 'my-in-cop1'
+inputs:
+- {name: item-a, type: Integer}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op1, item.a: "$0"
+        - {inputValue: item-a}
+'''
+
+op_out_yaml = '''\
+name: 'my-out-cop2'
+inputs:
+- {name: output, type: String}
+implementation:
+    container:
+        image: library/bash:4.4.23
+        command: ['sh', '-c']
+        args:
+        - |
+          set -e
+          echo no output global op2, outp: "$0"
+        - {inputValue: output}
+'''
+
 
 @dsl.pipeline(name='withparam-output-dict')
 def pipeline():
-    op0 = dsl.ContainerOp(
-        name="my-out-cop0",
-        image='python:alpine3.6',
-        command=["sh", "-c"],
-        arguments=['python -c "import json; import sys; '
-                   'json.dump([{\'a\': 1, \'b\': 2}, {\'a\': 10, \'b\': 20}], '
-                   'open(\'/tmp/out.json\', \'w\'))"'],
-        file_outputs={'out': '/tmp/out.json'},
-    )
+    op0_template = components.load_component_from_text(op0_yaml)
+    op0 = op0_template()
 
     with dsl.ParallelFor(op0.output) as item:
-        op1 = dsl.ContainerOp(
-            name="my-in-cop1",
-            image="library/bash:4.4.23",
-            command=["sh", "-c"],
-            arguments=["echo do output op1 item.a: %s" % item.a],
-        )
-
-    op_out = dsl.ContainerOp(
-        name="my-out-cop2",
-        image="library/bash:4.4.23",
-        command=["sh", "-c"],
-        arguments=["echo do output op2, outp: %s" % op0.output],
-    )
+        op1_template = components.load_component_from_text(op1_yaml)
+        op1 = op1_template(item.a)
+
+    op_out_template = components.load_component_from_text(op_out_yaml)
+    op_out = op_out_template(op0.output)
 
 
 if __name__ == '__main__':
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_dict.yaml b/sdk/python/tests/compiler/testdata/withparam_output_dict.yaml
index 2d4a0ec306..386299887e 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_dict.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output_dict.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-output-dict
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -36,21 +36,28 @@ spec:
         steps:
         - name: main
           args:
-          - 'python -c "import json; import sys; json.dump([{''a'': 1, ''b'': 2},
-            {''a'': 10, ''b'': 20}], open(''$(results.out.path)'', ''w''))"'
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([{'a': 1, 'b': 2}, {'a': 10, 'b': 20}], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([{''a'':
+              1, ''b'': 2}, {''a'': 10, ''b'': 20}], open(''$0'', ''w''))\"\n", {"outputPath":
+              "out"}], "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name":
+              "my-out-cop0", "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -61,7 +68,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -74,6 +84,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
@@ -99,7 +113,10 @@ spec:
                 steps:
                 - name: main
                   args:
-                  - 'echo do output op1 item.a: $(inputs.params.my-out-cop0-out-loop-item-subvar-a)'
+                  - |
+                    set -e
+                    echo no output global op1, item.a: "$0"
+                  - $(inputs.params.my-out-cop0-out-loop-item-subvar-a)
                   command:
                   - sh
                   - -c
@@ -113,6 +130,11 @@ spec:
                     pipelines.kubeflow.org/generation: ''
                     pipelines.kubeflow.org/cache_enabled: "true"
                   annotations:
+                    pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+                      {"args": ["set -e\necho no output global op1, item.a: \"$0\"\n",
+                      {"inputValue": "item-a"}], "command": ["sh", "-c"], "image":
+                      "library/bash:4.4.23"}}, "inputs": [{"name": "item-a", "type":
+                      "Integer"}], "name": "my-in-cop1"}'
                     tekton.dev/template: ''
               timeout: 0s
           iterateParam: my-out-cop0-out-loop-item
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_dict_noninlined.yaml b/sdk/python/tests/compiler/testdata/withparam_output_dict_noninlined.yaml
index 9759f6bb90..0d12375912 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_dict_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output_dict_noninlined.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-output-dict
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -34,10 +34,14 @@ metadata:
       [{"name": "my-out-cop0-out-loop-item-subvar-a", "type": "string"}], "tasks":
       [{"name": "my-in-cop1", "params": [{"name": "my-out-cop0-out-loop-item-subvar-a",
       "value": "$(params.my-out-cop0-out-loop-item-subvar-a)"}], "taskSpec": {"metadata":
-      {"annotations": {"tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
+      {"annotations": {"pipelines.kubeflow.org/component_spec": "{\"implementation\":
+      {\"container\": {\"args\": [\"set -e\\necho no output global op1, item.a: \\\"$0\\\"\\n\",
+      {\"inputValue\": \"item-a\"}], \"command\": [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}},
+      \"inputs\": [{\"name\": \"item-a\", \"type\": \"Integer\"}], \"name\": \"my-in-cop1\"}",
+      "tekton.dev/template": ""}, "labels": {"pipelines.kubeflow.org/cache_enabled":
       "true", "pipelines.kubeflow.org/generation": "", "pipelines.kubeflow.org/pipelinename":
       ""}}, "params": [{"name": "my-out-cop0-out-loop-item-subvar-a", "type": "string"}],
-      "steps": [{"args": ["echo do output op1 item.a: $(inputs.params.my-out-cop0-out-loop-item-subvar-a)"],
+      "steps": [{"args": ["set -e\necho no output global op1, item.a: \"$0\"\n", "$(inputs.params.my-out-cop0-out-loop-item-subvar-a)"],
       "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
       "timeout": "0s"}]}}}]'
 spec:
@@ -48,21 +52,28 @@ spec:
         steps:
         - name: main
           args:
-          - 'python -c "import json; import sys; json.dump([{''a'': 1, ''b'': 2},
-            {''a'': 10, ''b'': 20}], open(''$(results.out.path)'', ''w''))"'
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([{'a': 1, 'b': 2}, {'a': 10, 'b': 20}], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([{''a'':
+              1, ''b'': 2}, {''a'': 10, ''b'': 20}], open(''$0'', ''w''))\"\n", {"outputPath":
+              "out"}], "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name":
+              "my-out-cop0", "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -73,7 +84,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -86,6 +100,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_dict_pipelineloop_cr1.yaml b/sdk/python/tests/compiler/testdata/withparam_output_dict_pipelineloop_cr1.yaml
index d3ee6b4312..c1ab6c1619 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_dict_pipelineloop_cr1.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output_dict_pipelineloop_cr1.yaml
@@ -30,7 +30,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op1 item.a: $(inputs.params.my-out-cop0-out-loop-item-subvar-a)'
+          - |
+            set -e
+            echo no output global op1, item.a: "$0"
+          - $(inputs.params.my-out-cop0-out-loop-item-subvar-a)
           command:
           - sh
           - -c
@@ -44,6 +47,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op1, item.a: \"$0\"\n", {"inputValue":
+              "item-a"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item-a", "type": "Integer"}], "name": "my-in-cop1"}'
             tekton.dev/template: ''
       timeout: 0s
   iterateParam: my-out-cop0-out-loop-item
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_noninlined.yaml b/sdk/python/tests/compiler/testdata/withparam_output_noninlined.yaml
index 16d171162a..3127847f1c 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_noninlined.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output_noninlined.yaml
@@ -18,7 +18,7 @@ metadata:
   name: withparam-output
   annotations:
     tekton.dev/output_artifacts: '{"my-out-cop0": [{"key": "artifacts/$PIPELINERUN/my-out-cop0/out.tgz",
-      "name": "my-out-cop0-out", "path": "/tmp/out.json"}]}'
+      "name": "my-out-cop0-out", "path": "/tmp/outputs/out/data"}]}'
     tekton.dev/input_artifacts: '{"my-out-cop2": [{"name": "my-out-cop0-out", "parent_task":
       "my-out-cop0"}]}'
     tekton.dev/artifact_bucket: mlpipeline
@@ -33,12 +33,16 @@ metadata:
       "spec": {"iterateParam": "my-out-cop0-out-loop-item", "pipelineSpec": {"params":
       [{"name": "my-out-cop0-out-loop-item", "type": "string"}], "tasks": [{"name":
       "my-in-cop1", "params": [{"name": "my-out-cop0-out-loop-item", "value": "$(params.my-out-cop0-out-loop-item)"}],
-      "taskSpec": {"metadata": {"annotations": {"tekton.dev/template": ""}, "labels":
-      {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
+      "taskSpec": {"metadata": {"annotations": {"pipelines.kubeflow.org/component_spec":
+      "{\"implementation\": {\"container\": {\"args\": [\"set -e\\necho no output
+      global op1, item: \\\"$0\\\"\\n\", {\"inputValue\": \"item\"}], \"command\":
+      [\"sh\", \"-c\"], \"image\": \"library/bash:4.4.23\"}}, \"inputs\": [{\"name\":
+      \"item\", \"type\": \"Integer\"}], \"name\": \"my-in-cop1\"}", "tekton.dev/template":
+      ""}, "labels": {"pipelines.kubeflow.org/cache_enabled": "true", "pipelines.kubeflow.org/generation":
       "", "pipelines.kubeflow.org/pipelinename": ""}}, "params": [{"name": "my-out-cop0-out-loop-item",
-      "type": "string"}], "steps": [{"args": ["echo do output op1 item: $(inputs.params.my-out-cop0-out-loop-item)"],
-      "command": ["sh", "-c"], "image": "library/bash:4.4.23", "name": "main"}]},
-      "timeout": "0s"}]}}}]'
+      "type": "string"}], "steps": [{"args": ["set -e\necho no output global op1,
+      item: \"$0\"\n", "$(inputs.params.my-out-cop0-out-loop-item)"], "command": ["sh",
+      "-c"], "image": "library/bash:4.4.23", "name": "main"}]}, "timeout": "0s"}]}}}]'
 spec:
   pipelineSpec:
     tasks:
@@ -47,21 +51,28 @@ spec:
         steps:
         - name: main
           args:
-          - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
-            open('$(results.out.path)', 'w'))"
+          - |
+            set -e
+            python -c "import json; import sys; json.dump([i for i in range(20, 31)], open('$0', 'w'))"
+          - $(results.out.path)
           command:
           - sh
           - -c
           image: python:alpine3.6
         results:
         - name: out
-          description: /tmp/out.json
+          description: /tmp/outputs/out/data
         metadata:
           labels:
             pipelines.kubeflow.org/pipelinename: ''
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\npython -c \"import json; import sys; json.dump([i
+              for i in range(20, 31)], open(''$0'', ''w''))\"\n", {"outputPath": "out"}],
+              "command": ["sh", "-c"], "image": "python:alpine3.6"}}, "name": "my-out-cop0",
+              "outputs": [{"name": "out", "type": "String"}]}'
             tekton.dev/template: ''
       timeout: 0s
     - name: my-out-cop2
@@ -72,7 +83,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op2, outp: $(inputs.params.my-out-cop0-out)'
+          - |
+            set -e
+            echo no output global op2, outp: "$0"
+          - $(inputs.params.my-out-cop0-out)
           command:
           - sh
           - -c
@@ -85,6 +99,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op2, outp: \"$0\"\n", {"inputValue":
+              "output"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "output", "type": "String"}], "name": "my-out-cop2"}'
             tekton.dev/template: ''
       timeout: 0s
     - runAfter:
diff --git a/sdk/python/tests/compiler/testdata/withparam_output_pipelineloop_cr1.yaml b/sdk/python/tests/compiler/testdata/withparam_output_pipelineloop_cr1.yaml
index 0ce25da73a..bc74be91a3 100644
--- a/sdk/python/tests/compiler/testdata/withparam_output_pipelineloop_cr1.yaml
+++ b/sdk/python/tests/compiler/testdata/withparam_output_pipelineloop_cr1.yaml
@@ -30,7 +30,10 @@ spec:
         steps:
         - name: main
           args:
-          - 'echo do output op1 item: $(inputs.params.my-out-cop0-out-loop-item)'
+          - |
+            set -e
+            echo no output global op1, item: "$0"
+          - $(inputs.params.my-out-cop0-out-loop-item)
           command:
           - sh
           - -c
@@ -44,6 +47,10 @@ spec:
             pipelines.kubeflow.org/generation: ''
             pipelines.kubeflow.org/cache_enabled: "true"
           annotations:
+            pipelines.kubeflow.org/component_spec: '{"implementation": {"container":
+              {"args": ["set -e\necho no output global op1, item: \"$0\"\n", {"inputValue":
+              "item"}], "command": ["sh", "-c"], "image": "library/bash:4.4.23"}},
+              "inputs": [{"name": "item", "type": "Integer"}], "name": "my-in-cop1"}'
             tekton.dev/template: ''
       timeout: 0s
   iterateParam: my-out-cop0-out-loop-item