code: def _clone_items(pipeline, to_clone):
      """Clones dependency-sorted list of PCollections and PTransforms.
  
      Returns mappings of PCollection and PTransform replacements.
  
      Args:
        pipeline: The beam.Pipeline.
        to_clone: A dependency-sorted list of PCollections and PTransforms.
  
      Returns:
        pcollection_replacements: a dict mapping original to cloned PCollections.
  
      Raises:
        ValueError: if a clone is requested of an invalid object.
      """
      pcollection_replacements = {}
      ptransform_replacements = {}
      for item in to_clone:
          if isinstance(item, pvalue.PCollection):
              assert item not in pcollection_replacements
              copied = pvalue.PCollection(
                  pipeline,
                  tag=item.tag,
                  element_type=item.element_type,
                  windowing=item.windowing,
              )
              copied.producer = item.producer
              # Update copied PCollection producer if its producer was copied as well.
              if copied.producer in ptransform_replacements:
                  original_producer = copied.producer
                  copied.producer = ptransform_replacements[
                      original_producer
                  ]
                  # Update producer outputs,
                  for tag, output in original_producer.outputs.items():
                      if output == item:
                          copied.producer.outputs[tag] = copied
              assert copied.producer.transform is not None
              pcollection_replacements[item] = copied
          elif isinstance(item, beam_pipeline.AppliedPTransform):
              assert item.transform is not None
              assert item not in ptransform_replacements
              # The Beam pipeline graph keeps track of composite PTransforms by having
              # AppliedPTransform.parts be a list of "children" AppliedPTransforms that
              # are part of the "parent" AppliedPTransform. Any of these "composite
              # wrapper" AppliedPTransforms does not actually produce output independent
              # of the child non-composite transform. We therefore shouldn't ever clone
              # AppliedPTransforms with non-empty parts, since such AppliedPTransforms
              # are not reachable by tracing outputs in the pipeline graph.
              assert not item.parts, (
                  "Reached invalid composite AppliedPTransform: %r."
                  % item
              )
  
              # TODO(b/217271822): Implement resource hint 'close to resources' for
              # Beam/Dataflow, as when CSE makes it to Dataflow, 'close to resources'
              # cannot be recognized. Once this is fixed, we can change the tag prefix
              # to 'beam'.
              # TODO(b/238243699): Obviate the need for setting 'close to resources'
              # hints.
              close_to_resources_available = (
                  resources.ResourceHint.is_registered(
                      "close_to_resources"
                  )
              )
  
              if close_to_resources_available:
                  # Assign close_to_resources resource hint to the orginal PTransforms.
                  # The reason of adding this annotation is to prevent root Reads that are
                  # generated from deep copy being merged due to common subexpression
                  # elimination (CSE).
                  item.resource_hints[
                      "beam:resources:close_to_resources:v1"
                  ] = b"/fake/DeepCopy.Original[0]"
  
              # Assign new label.
              count = 0
              copy_suffix = f"Copy{count}"
              new_label = f"{item.full_label}.{copy_suffix}"
              while new_label in pipeline.applied_labels:
                  count += 1
                  copy_suffix = f"Copy{count}"
                  new_label = f"{item.full_label}.{copy_suffix}"
              pipeline.applied_labels.add(new_label)
  
              # Update inputs.
              new_inputs = {
                  tag: pcollection_replacements.get(
                      old_input, old_input
                  )
                  for tag, old_input in item.main_inputs.items()
              }
  
              # Create the copy. Note that in the copy, copied.outputs will start out
              # empty. Any outputs that are used will be repopulated in the PCollection
              # copy branch above.
              copied = beam_pipeline.AppliedPTransform(
                  item.parent, item.transform, new_label, new_inputs
              )
  
              # Add a 'close to resource' resource hint to the copied PTransforms. The
              # PTransforms that are generated from each deep copy have the same unique
              # 'close to resource' resource hint. This is to make sure that the
              # PTransforms that are cloned from each deep copy can be fused together,
              # but not across copies nor with the original.
              if close_to_resources_available:
                  copied.resource_hints[
                      "beam:resources:close_to_resources:v1"
                  ] = f"/fake/DeepCopy.{copy_suffix}[0]".encode()
  
              ptransform_replacements[item] = copied
  
              # Update composite transform parent to include this copy.
              # TODO(b/111366378): Reconcile the composite PTransform nesting hierarchy,
              # especially in the case where copied PTransforms should be copied in an
              # "all-or-nothing" manner. This would allow the deep copy operation to be
              # safe in the case runners replace well-known composite PTransforms in
              # their entirety during execution.
              copied.parent.parts.append(copied)
          else:
              raise ValueError("Invalid object to clone: %s" % item)
  
      return pcollection_replacements
  
file_path: transform/tensorflow_transform/beam/deep_copy.py
imports: import queue
  import apache_beam
  from apache_beam import pipeline
  from apache_beam import pvalue
  from apache_beam.pvalue import PCollection
  from apache_beam.transforms import resources
label_desc: Use image node_id 3 for calling a global function with example usage: _clone_items(pipeline, to_clone) and returns: pcollection_replacements