{"record":{"id":"5d08ae254f2d5aa8","repo":"apache/beam","slug":"could-not-find-subtransform-to-copy","errorCode":null,"errorMessage":"Could not find subtransform to copy: ","messagePattern":"Could not find subtransform to copy: ","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py","lineNumber":690,"sourceCode":"      if (parent not in components.transforms and\n          parent in pipeline_proto.components.transforms):\n        components.transforms[parent].CopyFrom(\n            pipeline_proto.components.transforms[parent])\n        copy_output_pcollections(components.transforms[parent])\n        del components.transforms[parent].subtransforms[:]\n      # Ensure that child is the last item in the parent's subtransforms.\n      # If the stages were previously sorted into topological order using\n      # sort_stages, this ensures that the parent transforms are also\n      # added in topological order.\n      if child in components.transforms[parent].subtransforms:\n        components.transforms[parent].subtransforms.remove(child)\n      components.transforms[parent].subtransforms.append(child)\n      add_parent(parent, parents.get(parent))\n\n  def copy_subtransforms(transform):\n    for subtransform_id in transform.subtransforms:\n      if subtransform_id not in pipeline_proto.components.transforms:\n        raise RuntimeError(\n            'Could not find subtransform to copy: ' + subtransform_id)\n      subtransform = pipeline_proto.components.transforms[subtransform_id]\n      components.transforms[subtransform_id].CopyFrom(subtransform)\n      copy_output_pcollections(components.transforms[subtransform_id])\n      copy_subtransforms(subtransform)\n\n  all_consumers = collections.defaultdict(\n      set)  # type: DefaultDict[str, Set[int]]\n  for stage in stages:\n    for transform in stage.transforms:\n      for pcoll in transform.inputs.values():\n        all_consumers[pcoll].add(id(transform))\n\n  for stage in stages:\n    if partial:\n      transform = only_element(stage.transforms)\n      copy_subtransforms(transform)\n    else:","sourceCodeStart":672,"sourceCodeEnd":708,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py#L672-L708","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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","Verify no custom translation/optimization pass removes transforms from components.transforms without removing them from parents' subtransforms lists","Ensure external transforms (cross-language) are fully expanded before stage fusion/pipeline reconstruction","As a diagnostic, iterate all transforms and assert every subtransform id exists in components.transforms before invoking the optimization"],"exampleFix":"# before (custom pass breaks graph)\ndel components.transforms[dead_id]  # parent still lists dead_id\n# after: keep graph consistent\nfor parent in components.transforms.values():\n    if dead_id in parent.subtransforms:\n        parent.subtransforms.remove(dead_id)\ndel components.transforms[dead_id]","handlingStrategy":"try-catch","validationCode":"missing = [sid for t in pipeline.components.transforms.values()\n          for sid in t.subtransforms\n          if sid not in pipeline.components.transforms]\nassert not missing, f'Dangling subtransforms: {missing}'","typeGuard":null,"tryCatchPattern":"try:\n    proto = translations.pipeline_from_stages(stages)\nexcept RuntimeError as e:\n    if 'Could not find subtransform' in str(e):\n        log.error('Broken pipeline graph: %s', e)  # custom pass left a dangling ref\n    raise","preventionTips":["When mutating the pipeline proto, always keep components.transforms and parent subtransforms lists consistent","Dump the pre-optimization pipeline proto when developing custom runners/optimizers","Fully expand external/cross-language transforms before graph rewriting","Test custom translation passes against pipelines containing composite transforms"],"tags":["python","apache-beam","pipeline-graph","internal-invariant"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}