SDK - Restored module decoupling (#4625)

A recent PR has added changes architecturally belonged to a different module (the component bridge). This has introduced unintended dependencies and couplings between the modules. This PR restores the module separation. It also makes the code simpler.

sdk/python/kfp/components/_components.py

@@ -390,7 +390,6 @@ def _resolve_command_line_and_paths(
     arguments: Mapping[str, str],
     input_path_generator=_generate_input_file_name,
     output_path_generator=_generate_output_file_name,
-    output_value_generator=None,
     argument_serializer=serialize_value,
 ) -> _ResolvedCommandLineAndPaths:
     """Resolves the command line argument placeholders. Also produces the maps of the generated inpuit/output paths."""
@@ -400,7 +399,6 @@ def _resolve_command_line_and_paths(
         raise TypeError('Only container components have command line to resolve')
 
     inputs_dict = {input_spec.name: input_spec for input_spec in component_spec.inputs or []}
-    outputs_dict = {output_spec.name: output_spec for output_spec in component_spec.outputs or []}
     container_spec = component_spec.implementation.container
 
     output_paths = OrderedDict() #Preserving the order to make the kubernetes output names deterministic
@@ -457,15 +455,6 @@ def expand_command_part(arg) -> Union[str, List[str], None]:
             else:
                 output_paths[output_name] = output_filename
 
-            # output_value_generator is only used by the v2 (experimental) compiler.
-            # TODO: maybe fork the file so that we can split the logic for v1 vs. v2.
-            if output_value_generator:
-                output_spec = outputs_dict[output_name]
-                # TODO: move the import to the top of the file (once fixed circular dependency).
-                from kfp.v2.dsl import type_utils
-                if type_utils.is_parameter_type(output_spec.type):
-                    return output_value_generator(output_name)
-
             return output_filename
 
         elif isinstance(arg, ConcatPlaceholder):

sdk/python/kfp/v2/dsl/component_bridge.py

@@ -116,6 +116,8 @@ def create_container_op_from_component_and_arguments(
           'Unsupported output type: "{}". The type must be one of the following: {}.'
           .format(output.type, type_utils.all_types()))
 
+  outputs_dict = {output_spec.name: output_spec for output_spec in component_spec.outputs or []}
+
   def _input_artifact_placeholder(input_key: str) -> str:
     return "{{{{$.inputs.artifacts['{}'].uri}}}}".format(input_key)
 
@@ -128,6 +130,12 @@ def _output_artifact_placeholder(output_key: str) -> str:
   def _output_parameter_placeholder(output_key: str) -> str:
     return "{{{{$.outputs.parameters['{}'].output_file}}}}".format(output_key)
 
+  def _resolve_output_path_placeholder(output_key: str) -> str:
+    if type_utils.is_parameter_type(outputs_dict[output_key].type):
+      return _output_parameter_placeholder(output_key)
+    else:
+      return _output_artifact_placeholder(output_key)
+
   # IR placeholders are decided merely based on the declared type of the input.
   # It doesn't matter wether it's InputValuePlaceholder or InputPathPlaceholder
   # from component_spec.
@@ -142,8 +150,7 @@ def _output_parameter_placeholder(output_key: str) -> str:
       component_spec=component_spec,
       arguments=placeholder_arguments,
       input_path_generator=_input_artifact_placeholder,
-      output_path_generator=_output_artifact_placeholder,
-      output_value_generator=_output_parameter_placeholder,
+      output_path_generator=_resolve_output_path_placeholder,
   )
 
   resolved_cmd = _resolve_command_line_and_paths(