{"record":{"id":"cd76921d4cfaf7dd","repo":"apache/beam","slug":"f-ambiguous-output-at-line-safelineloader-get-line-name-name","errorCode":null,"errorMessage":"f'Ambiguous output at line {SafeLineLoader.get_line(name)}: {name} has outputs {list(outputs.keys())}'","messagePattern":"f'Ambiguous output at line (.+?): (.+?) has outputs (.+?)'","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_transform.py","lineNumber":236,"sourceCode":"        return outputs[output]\n      elif len(outputs) == 1 and outputs[next(iter(outputs))].tag == output:\n        return outputs[next(iter(outputs))]\n      else:\n        raise ValueError(\n            f'Unknown output {repr(output)} '\n            f'at line {SafeLineLoader.get_line(name)}: '\n            f'{transform} only has outputs {list(outputs.keys())}')\n    else:\n      outputs = self.get_outputs(name)\n      if len(outputs) == 1:\n        return only_element(outputs.values())\n      else:\n        error_output = self._transforms_by_uuid[self.get_transform_id(\n            name)]['config'].get('error_handling', {}).get('output')\n        if error_output and error_output in outputs and len(outputs) == 2:\n          return next(\n              output for tag, output in outputs.items() if tag != error_output)\n        raise ValueError(\n            f'Ambiguous output at line {SafeLineLoader.get_line(name)}: '\n            f'{name} has outputs {list(outputs.keys())}')\n\n  def get_outputs(self, transform_name):\n    return self.compute_outputs(self.get_transform_id(transform_name))\n\n  @memoize_method\n  def compute_outputs(self, transform_id):\n    return expand_transform(self._transforms_by_uuid[transform_id], self)\n\n  def best_provider(\n      self, t, input_providers: yaml_provider.Iterable[yaml_provider.Provider]):\n    if isinstance(t, dict):\n      spec = t\n    else:\n      spec = self._transforms_by_uuid[self.get_transform_id(t)]\n    possible_providers = []\n    unavailable_provider_messages = []","sourceCodeStart":218,"sourceCodeEnd":254,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_transform.py#L218-L254","documentation":"Scope.get_pcollection, when given a bare transform name (no `.output` suffix), can pick a PCollection automatically only if the transform has exactly one output — or exactly two where one is a configured error-handling output. Otherwise the choice is ambiguous and Beam raises this ValueError listing all output tags.","triggerScenarios":"Using a bare `input: MyTransform` reference where the transform produces multiple outputs (e.g. a partition-style transform or any transform with error_handling configured) without specifying which output tag to consume.","commonSituations":"Multi-output transforms feeding a downstream transform without a dotted tag; forgetting that adding error_handling made a previously single-output transform ambiguous; generated references that omit tags.","solutions":["Reference the specific output with a dotted name: `MyTransform.<tag>` using a tag from the error message.","Check the listed output tags in the message and pick the intended one.","If only the main output is wanted and exactly one error output exists, the auto-disambiguation already handles it — verify the error_handling.output config.","Restructure the pipeline so each consumer names its input output explicitly."],"exampleFix":"# before\n- name: consume\n  input: split_and_validate\n  type: WriteToJson\n# after\n- name: consume\n  input: split_and_validate.valid_rows\n  type: WriteToJson","handlingStrategy":"validation","validationCode":"def check_reference(name, transform):\n    n_outputs = len(transform.get('outputs', []))\n    if n_outputs > 1 and '.' not in name:\n        err_out = transform.get('config', {}).get('error_handling', {}).get('output')\n        if not (err_out and n_outputs == 2):\n            raise ValueError(f'{name} is multi-output; use name.tag')","typeGuard":"def needs_explicit_output(transform_spec):\n    return len(transform_spec.get('outputs', [])) > 2","tryCatchPattern":"try:\n    pcoll = scope.get_pcollection(name)\nexcept ValueError as e:\n    if 'Ambiguous output' in str(e):\n        raise UserPipelineError(f'{name}: specify output tag, e.g. {name}.<tag>') from e","preventionTips":["Always use dotted references for multi-output transforms","Remember error_handling adds a second output","Prefer explicit output tags over bare names in generated pipelines"],"tags":["yaml","beam-yaml","pipeline-definition","ambiguity"],"backgroundTag":"invalid-argument-value","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"}