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.
      maybe_int = lambda s: int(s) if re.match(r'^\d+$', s) else s
      semver = tuple(maybe_int(s) for s in beam.__version__.split('.'))
      if semver >= (2, 63):
        extra_args = {
            'environment_id': item.environment_id,
            'annotations': item.annotations,
        }
      else:
        extra_args = {}
      copied = beam_pipeline.AppliedPTransform(
          item.parent, item.transform, new_label, new_inputs, **extra_args
      )

      # 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
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.
  
tensorflow_transform/beam/deep_copy.py