fix(sdk): Compiler - Fixed input artifact name sanitization when usin…

…g raw string arguments. Fixes #4110 (#4120)

sdk/python/kfp/compiler/compiler.py

@@ -759,6 +759,11 @@ def _sanitize_and_inject_artifact(self, pipeline: dsl.Pipeline, pipeline_conf=No
           sanitized_attribute_outputs[sanitize_k8s_name(key, True)] = \
             op.attribute_outputs[key]
         op.attribute_outputs = sanitized_attribute_outputs
+      if isinstance(op, dsl.ContainerOp):
+        if op.input_artifact_paths:
+          op.input_artifact_paths = {sanitize_k8s_name(key, True): value for key, value in op.input_artifact_paths.items()}
+        if op.artifact_arguments:
+          op.artifact_arguments = {sanitize_k8s_name(key, True): value for key, value in op.artifact_arguments.items()}
       sanitized_ops[sanitized_name] = op
     pipeline.ops = sanitized_ops
 

sdk/python/tests/compiler/compiler_tests.py

@@ -943,3 +943,32 @@ def some_pipeline(pipeline_in1, pipeline_in2):
           'Wrong argument mapping: "{}" passed to "{}"'.format(argument['value'], argument['name']))
       else:
         self.fail('Unexpected input name: ' + argument['name'])
+
+  def test_input_name_sanitization(self):
+    # Verifying that the recursive call arguments are passed correctly when specified out of order
+    component_2_in_1_out_op = kfp.components.load_component_from_text('''
+inputs:
+- name: Input 1
+- name: Input 2
+outputs:
+- name: Output 1
+implementation:
+  container:
+    image: busybox
+    command:
+    - echo
+    - inputValue: Input 1
+    - inputPath: Input 2
+    - outputPath: Output 1
+    ''')
+    def some_pipeline():
+      task1 = component_2_in_1_out_op('value 1', 'value 2')
+      component_2_in_1_out_op(task1.output, task1.output)
+
+    workflow_dict = kfp.compiler.Compiler()._compile(some_pipeline)
+    container_templates = [template for template in workflow_dict['spec']['templates'] if 'container' in template]
+    for template in container_templates:
+      for argument in template['inputs'].get('parameters', []):
+        self.assertNotIn(' ', argument['name'], 'The input name "{}" of template "{}" was not sanitized.'.format(argument['name'], template['name']))
+      for argument in template['inputs']['artifacts']:
+        self.assertNotIn(' ', argument['name'], 'The input name "{}" of template "{}" was not sanitized.'.format(argument['name'], template['name']))