Merge 4f54647 into 389b585

sdk/python/kfp/compiler/compiler.py

@@ -709,7 +709,7 @@ def _sanitize_and_inject_artifact(self, pipeline: dsl.Pipeline):
         param.name = K8sHelper.sanitize_k8s_name(param.name)
         if param.op_name:
           param.op_name = K8sHelper.sanitize_k8s_name(param.op_name)
-      if op.output is not None:
+      if op.output is not None and not isinstance(op.output, dsl._container_op._MultipleOutputsError):
         op.output.name = K8sHelper.sanitize_k8s_name(op.output.name)
         op.output.op_name = K8sHelper.sanitize_k8s_name(op.output.op_name)
       if op.dependent_names:

sdk/python/kfp/dsl/_container_op.py

@@ -1099,9 +1099,10 @@ def _decorated(*args, **kwargs):
                 for name in file_outputs.keys()
             }
 
-        self.output = None
         if len(self.outputs) == 1:
             self.output = list(self.outputs.values())[0]
+        else:
+            self.output = _MultipleOutputsError()
 
         self.pvolumes = {}
         self.add_pvolumes(pvolumes)
@@ -1163,7 +1164,6 @@ def _set_metadata(self, metadata):
                         output_type = output_meta.type
                 self.outputs[output].param_type = output_type
 
-            self.output = None
             if len(self.outputs) == 1:
                 self.output = list(self.outputs.values())[0]
 
@@ -1198,3 +1198,15 @@ def add_pvolumes(self,
 # proxy old ContainerOp properties to ContainerOp.container
 # with PendingDeprecationWarning.
 ContainerOp = _proxy_container_op_props(ContainerOp)
+
+
+class _MultipleOutputsError:
+    @staticmethod
+    def raise_error():
+        raise RuntimeError('This task has multiple outputs. Use `task.outputs[<output name>]` dictionary to refer to the one you need.')
+
+    def __getattribute__(self, name):
+        _MultipleOutputsError.raise_error()
+
+    def __str__(self):
+        _MultipleOutputsError.raise_error()

sdk/python/tests/compiler/compiler_tests.py

@@ -683,6 +683,28 @@ def some_pipeline():
       if container:
         self.assertEqual(template['retryStrategy']['limit'], 5)
 
+  def test_container_op_output_error_when_no_or_multiple_outputs(self):
+
+    def no_outputs_pipeline():
+      no_outputs_op = dsl.ContainerOp(name='dummy', image='dummy')
+      dsl.ContainerOp(name='dummy', image='dummy', arguments=[no_outputs_op.output])
+
+    def one_output_pipeline():
+      one_output_op = dsl.ContainerOp(name='dummy', image='dummy', file_outputs={'out1': 'path1'})
+      dsl.ContainerOp(name='dummy', image='dummy', arguments=[one_output_op.output])
+
+    def two_outputs_pipeline():
+      two_outputs_op = dsl.ContainerOp(name='dummy', image='dummy', file_outputs={'out1': 'path1', 'out2': 'path2'})
+      dsl.ContainerOp(name='dummy', image='dummy', arguments=[two_outputs_op.output])
+
+    with self.assertRaises(RuntimeError):
+      compiler.Compiler()._compile(no_outputs_pipeline)
+
+    compiler.Compiler()._compile(one_output_pipeline)
+
+    with self.assertRaises(RuntimeError):
+      compiler.Compiler()._compile(two_outputs_pipeline)
+
   def test_withitem_basic(self):
     self._test_py_compile_yaml('withitem_basic')