apache/beam · error · RuntimeError

Could not find subtransform to copy:

Error message

Could not find subtransform to copy: 

What it means

When rebuilding a pipeline proto from optimized stages, copy_subtransforms walks each transform's subtransforms and copies them into the output components. If a listed subtransform id is absent from the source pipeline's components.transforms map, the graph is inconsistent and a RuntimeError is thrown rather than silently dropping the node.

Solutions

1. Check for a Beam bug — inspect the pipeline proto before optimization (use --save_main_session / pipeline-to-proto dump) and file an issue with the repro
2. Verify no custom translation/optimization pass removes transforms from components.transforms without removing them from parents' subtransforms lists
3. Ensure external transforms (cross-language) are fully expanded before stage fusion/pipeline reconstruction
4. As a diagnostic, iterate all transforms and assert every subtransform id exists in components.transforms before invoking the optimization

Example fix

# before (custom pass breaks graph)
del components.transforms[dead_id]  # parent still lists dead_id
# after: keep graph consistent
for parent in components.transforms.values():
    if dead_id in parent.subtransforms:
        parent.subtransforms.remove(dead_id)
del components.transforms[dead_id]

missing = [sid for t in pipeline.components.transforms.values()
          for sid in t.subtransforms
          if sid not in pipeline.components.transforms]
assert not missing, f'Dangling subtransforms: {missing}'

try:
    proto = translations.pipeline_from_stages(stages)
except RuntimeError as e:
    if 'Could not find subtransform' in str(e):
        log.error('Broken pipeline graph: %s', e)  # custom pass left a dangling ref
    raise

When it happens

Trigger: A stage's transform references a subtransform id that was never added (or was already pruned) from pipeline_proto.components.transforms — typically a bug in a GraphOptimizer/ptransform translation that mutates subtransform lists without keeping the components map in sync.

Common situations: Custom Beam runner/optimization passes that reorder or delete transforms; cross-language expansion producing dangling transform references; upgrading Beam and hitting an internal invariant in pipeline_from_stages.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• Found more than one renderer for option
• Initializing SDFBoundedSourceRestrictionTracker requires a…
• Input to Impulse transform must be a PBegin but found
• Stage has no main inputs. At least one main input is…
• A BigQuery table or a query must be specified

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/5d08ae254f2d5aa8. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py:690

      if (parent not in components.transforms and
          parent in pipeline_proto.components.transforms):
        components.transforms[parent].CopyFrom(
            pipeline_proto.components.transforms[parent])
        copy_output_pcollections(components.transforms[parent])
        del components.transforms[parent].subtransforms[:]
      # Ensure that child is the last item in the parent's subtransforms.
      # If the stages were previously sorted into topological order using
      # sort_stages, this ensures that the parent transforms are also
      # added in topological order.
      if child in components.transforms[parent].subtransforms:
        components.transforms[parent].subtransforms.remove(child)
      components.transforms[parent].subtransforms.append(child)
      add_parent(parent, parents.get(parent))

  def copy_subtransforms(transform):
    for subtransform_id in transform.subtransforms:
      if subtransform_id not in pipeline_proto.components.transforms:
        raise RuntimeError(
            'Could not find subtransform to copy: ' + subtransform_id)
      subtransform = pipeline_proto.components.transforms[subtransform_id]
      components.transforms[subtransform_id].CopyFrom(subtransform)
      copy_output_pcollections(components.transforms[subtransform_id])
      copy_subtransforms(subtransform)

  all_consumers = collections.defaultdict(
      set)  # type: DefaultDict[str, Set[int]]
  for stage in stages:
    for transform in stage.transforms:
      for pcoll in transform.inputs.values():
        all_consumers[pcoll].add(id(transform))

  for stage in stages:
    if partial:
      transform = only_element(stage.transforms)
      copy_subtransforms(transform)
    else:

View on GitHub (pinned to 12126d8942)