code: def _compile_input_spec(
      *,
      pipeline_ctx: compiler_context.PipelineContext,
      tfx_node: base_node.BaseNode,
      input_key: str,
      channel: channel_types.BaseChannel,
      hidden: bool,
      min_count: int,
      result: pipeline_pb2.NodeInputs,
  ) -> None:
      """Compiles `BaseChannel` into `InputSpec` at `result.inputs[input_key]`.
  
      Args:
        pipeline_ctx: A `PipelineContext`.
        tfx_node: A `BaseNode` instance from pipeline DSL.
        input_key: An input key that the compiled `InputSpec` would be stored with.
        channel: A `BaseChannel` instance to compile.
        hidden: If true, this sets `InputSpec.hidden = True`. If the same channel
          instances have been called multiple times with different `hidden` value,
          then `hidden` will be `False`. In other words, if the channel is ever
          compiled with `hidden=False`, it will ignore other `hidden=True`.
        min_count: Minimum number of artifacts that should be resolved for this
          input key. If min_count is not met during the input resolution, it is
          considered as an error.
        result: A `NodeInputs` proto to which the compiled result would be written.
      """
      if input_key in result.inputs:
          # Already compiled. This can happen during compiling another input channel
          # from the same resolver function output.
          if not hidden:
              # Overwrite hidden = False even for already compiled channel, this is
              # because we don't know the input should truely be hidden until the
              # channel turns out not to be.
              result.inputs[input_key].hidden = False
          return
  
      if channel in pipeline_ctx.channels:
          # OutputChannel or PipelineInputChannel from the same pipeline has already
          # compiled IR in context.channels
          result.inputs[input_key].channels.append(
              pipeline_ctx.channels[channel]
          )
  
      elif isinstance(channel, channel_types.PipelineOutputChannel):
          # This is the case when PipelineInputs uses pipeline.outputs where the
          # pipeline is external (i.e. not a parent or sibling pipeline) thus
          # pipeline run cannot be synced.
          channel = cast(channel_types.PipelineOutputChannel, channel)
          _compile_channel_pb(
              artifact_type=channel.type,
              pipeline_name=channel.pipeline.id,
              node_id=channel.wrapped.producer_component_id,
              output_key=channel.output_key,
              result=result.inputs[input_key].channels.add(),
          )
  
      elif isinstance(channel, channel_types.ExternalPipelineChannel):
          channel = cast(channel_types.ExternalPipelineChannel, channel)
          result_input_channel = result.inputs[input_key].channels.add()
          _compile_channel_pb(
              artifact_type=channel.type,
              pipeline_name=channel.pipeline_name,
              node_id=channel.producer_component_id,
              output_key=channel.output_key,
              result=result_input_channel,
          )
  
          if channel.pipeline_run_id:
              ctx = result_input_channel.context_queries.add()
              ctx.type.name = constants.PIPELINE_RUN_CONTEXT_TYPE_NAME
              ctx.name.field_value.string_value = (
                  channel.pipeline_run_id
              )
  
          if pipeline_ctx.pipeline.platform_config:
              project_config = (
                  pipeline_ctx.pipeline.platform_config.project_platform_config
              )
              if (
                  channel.owner != project_config.owner
                  or channel.pipeline_name
                  != project_config.project_name
              ):
                  config = metadata_pb2.MLMDServiceConfig(
                      owner=channel.owner,
                      name=channel.pipeline_name,
                  )
                  result_input_channel.metadata_connection_config.Pack(
                      config
                  )
          else:
              config = metadata_pb2.MLMDServiceConfig(
                  owner=channel.owner,
                  name=channel.pipeline_name,
              )
              result_input_channel.metadata_connection_config.Pack(
                  config
              )
  
      elif isinstance(channel, channel_types.Channel):
          channel = cast(channel_types.Channel, channel)
          _compile_channel_pb(
              artifact_type=channel.type,
              pipeline_name=pipeline_ctx.pipeline_info.pipeline_name,
              node_id=channel.producer_component_id,
              output_key=channel.output_key,
              result=result.inputs[input_key].channels.add(),
          )
  
      elif isinstance(channel, channel_types.UnionChannel):
          channel = cast(channel_types.UnionChannel, channel)
          mixed_inputs = result.inputs[input_key].mixed_inputs
          mixed_inputs.method = (
              pipeline_pb2.InputSpec.Mixed.Method.UNION
          )
          for sub_channel in channel.channels:
              sub_key = pipeline_ctx.get_node_context(
                  tfx_node
              ).get_input_key(sub_channel)
              mixed_inputs.input_keys.append(sub_key)
              _compile_input_spec(
                  pipeline_ctx=pipeline_ctx,
                  tfx_node=tfx_node,
                  input_key=sub_key,
                  channel=sub_channel,
                  hidden=True,
                  min_count=0,
                  result=result,
              )
  
      elif isinstance(channel, resolved_channel.ResolvedChannel):
          channel = cast(resolved_channel.ResolvedChannel, channel)
          input_graph_ref = result.inputs[input_key].input_graph_ref
          input_graph_ref.graph_id = _compile_input_graph(
              pipeline_ctx, tfx_node, channel, result
          )
          if channel.output_key:
              input_graph_ref.key = channel.output_key
  
      elif isinstance(channel, channel_utils.ChannelForTesting):
          channel = cast(channel_utils.ChannelForTesting, channel)
          # Access result.inputs[input_key] to create an empty `InputSpec`. If the
          # testing channel does not point to static artifact IDs, empty `InputSpec`
          # is enough for testing.
          input_spec = result.inputs[input_key]
          if channel.artifact_ids:
              input_spec.static_inputs.artifact_ids.extend(
                  channel.artifact_ids
              )
  
      else:
          raise NotImplementedError(
              f"Node {tfx_node.id} got unsupported channel type {channel!r} for "
              f"inputs[{input_key!r}]."
          )
  
      if hidden:
          result.inputs[input_key].hidden = True
      if min_count:
          result.inputs[input_key].min_count = min_count
  
imports: from typing import Type, cast
  from tfx import types
  from tfx.dsl.compiler import compiler_context
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_node
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.input_resolution import resolver_op
  from tfx.dsl.placeholder import artifact_placeholder
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types_utils
  from tfx.proto.orchestration import metadata_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.types import resolved_channel
  from tfx.types import value_artifact
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
  from tfx.utils import typing_utils
label_desc: Use image node_id 3 for calling a global function with example usage: _compile_input_spec() without return types