class_docstr: Compiles a TFX pipeline or a component into a uDSL IR proto.
docstr: Compiles a PipelineBegin node for a composable pipeline.
code: def _compile_pipeline_begin_node(
      self,
      p: pipeline.Pipeline,
      pipeline_ctx: compiler_context.PipelineContext,
  ) -> pipeline_pb2.PipelineNode:
      """Compiles a PipelineBegin node for a composable pipeline."""
      node = pipeline_pb2.PipelineNode()
  
      # Step 1: Node info
      node.node_info.type.name = (
          compiler_utils.pipeline_begin_node_type_name(p)
      )
      node.node_info.id = compiler_utils.pipeline_begin_node_id(p)
  
      # Step 2: Node Context
      # Inner pipeline's contexts.
      _set_node_context(node, pipeline_ctx)
  
      # Step 3: Node inputs
      # Pipeline node inputs are stored as the inputs of the PipelineBegin node.
      node_inputs_compiler.compile_node_inputs(
          pipeline_ctx.parent, p, node.inputs
      )
  
      # Step 4: Node outputs
      # PipeineBegin node's outputs are the same as its inputs,
      # i.e., the composable pipeline's inputs.
      internal_inputs = (
          p._inputs.inputs if p._inputs else {}
      )  # pylint: disable=protected-access
      _set_node_outputs(node, internal_inputs)
      pipeline_ctx.channels.update(
          _generate_input_spec_for_outputs(node, internal_inputs)
      )
  
      # Pipeline Begin node has no parameters.
  
      # Step 5: Upstream/Downstream nodes
      # PipelineBegin node's upstreams nodes are the inner pipeline's upstream
      # nodes, i.e., the producer nodes of inner pipeline's inputs.
      # Outmost pipeline's PipelineBegin node does not has upstream nodes.
      if pipeline_ctx.is_subpipeline:
          upstreams = set(
              _find_runtime_upstream_node_ids(pipeline_ctx.parent, p)
          )
          if _begin_node_is_upstream(p, pipeline_ctx.parent.pipeline):
              upstreams.add(
                  compiler_utils.pipeline_begin_node_id(
                      pipeline_ctx.parent.pipeline
                  )
              )
          # Sort node ids so that compiler generates consistent results.
          node.upstream_nodes.extend(sorted(upstreams))
  
      # Set node execution options based on pipeline-level NodeExecutionOptions.
      # Because _compile_pipeline_begin_node may be called if either the pipeline
      # is a subpipeline OR the pipeline has inputs we can *only* pass in the
      # parent context if this begin node is for a subpipeline.
      if pipeline_ctx.is_subpipeline:
          execution_options_ctx = pipeline_ctx.parent
      else:
          execution_options_ctx = pipeline_ctx
      _set_node_execution_options(
          node, p, execution_options_ctx, p.enable_cache
      )
  
      # PipelineBegin node's downstream nodes are the nodes in the inner pipeline
      # that consumes pipeline's input channels.
      result = set()
      for inner_node in p.components:
          if _begin_node_is_upstream(inner_node, p):
              result.add(inner_node.id)
      # Sort node ids so that compiler generates consistent results.
      node.downstream_nodes.extend(sorted(result))
  
      return node
  
imports: import itertools
  from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, 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.compiler import node_inputs_compiler
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_driver
  from tfx.dsl.components.base import base_node
  from tfx.dsl.experimental.node_execution_options import utils
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import pipeline
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
label_desc: Use image node_id 2 for calling the Compiler obj's underlying member method code with example usage: obj._compile_pipeline_begin_node(p, pipeline_ctx) and returns: node