class_docstr: Compiles a TFX pipeline or a component into a uDSL IR proto.
docstr: Compiles a PipelineEnd node for a composable pipeline.
code: def _compile_pipeline_end_node(
      self,
      p: pipeline.Pipeline,
      pipeline_ctx: compiler_context.PipelineContext,
  ) -> pipeline_pb2.PipelineNode:
      """Compiles a PipelineEnd node for a composable pipeline."""
      node = pipeline_pb2.PipelineNode()
  
      # Step 1: Node info
      node.node_info.type.name = (
          compiler_utils.pipeline_end_node_type_name(p)
      )
      node.node_info.id = compiler_utils.pipeline_end_node_id(p)
  
      # Step 2: Node Context
      # Inner pipeline's contexts.
      _set_node_context(node, pipeline_ctx)
  
      # Step 3: Node inputs
      node_inputs_compiler.compile_node_inputs(
          pipeline_ctx,
          compiler_utils.create_pipeline_end_node(p),
          node.inputs,
      )
  
      # Step 4: Node outputs
      # PipeineEnd node's outputs are the same as inner pipeline's outputs.
      _set_node_outputs(node, p.outputs)
      if pipeline_ctx.is_subpipeline:
          pipeline_ctx.parent.channels.update(
              _generate_input_spec_for_pipeline_outputs(node, p)
          )
  
      # PipelineEnd node does not have parameters.
  
      # Step 5: Upstream/Downstream nodes
      # PipelineEnd node's upstream nodes are the nodes in the inner pipeline
      # that produces pipeline's output channels.
      result = set()
      for pipeline_output in p.outputs.values():
          for inner_node in p.components:
              if pipeline_output.wrapped in inner_node.outputs.values():
                  result.add(inner_node.id)
      # Sort node ids so that compiler generates consistent results.
      node.upstream_nodes.extend(sorted(result))
  
      # PipelineEnd node's downstream nodes are the inner pipeline's downstream
      # nodes, i.e., the consumer nodes of inner pipeline's outputs.
      downstreams = set(
          _find_runtime_downstream_node_ids(pipeline_ctx.parent, p)
      )
      if pipeline_ctx.is_subpipeline and _end_node_is_downstream(
          p, pipeline_ctx.parent.pipeline
      ):
          downstreams.add(
              compiler_utils.pipeline_end_node_id(
                  pipeline_ctx.parent.pipeline
              )
          )
      # Sort node ids so that compiler generates consistent results.
      node.downstream_nodes.extend(sorted(downstreams))
  
      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 3 for calling the Compiler obj's underlying member method code with example usage: obj._compile_pipeline_end_node(p, pipeline_ctx) and returns: node