code: def deep_copy(pcollection):
      """Create a deep copy of a PCollection up to materialization boundaries."""
      if not isinstance(pcollection, pvalue.PCollection):
          raise ValueError("Input to deep_copy must be a PCollection.")
  
      # AppliedPTransform.update_input_refcounts() is a vestigial method that
      # uses an incorrect heuristic; it will be removed in a future version of
      # Beam, since its results aren't used anyway.  Until then, we work around
      # this (see https://issues.apache.org/jira/browse/BEAM-4593).
      if (
          getattr(
              beam_pipeline.AppliedPTransform,
              "update_input_refcounts",
              None,
          )
          is not None
      ):
          beam_pipeline.AppliedPTransform.update_input_refcounts = (
              lambda _: None
          )
  
      to_clone = _get_items_to_clone(pcollection)
      pcollection_replacements = _clone_items(
          pcollection.pipeline, to_clone
      )
  
      return pcollection_replacements[pcollection]
  
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 4 for calling a global function with example usage: deep_copy(pcollection) and returns: pcollection_replacements