{"record":{"id":"5a55c1baef3f4c52","repo":"apache/beam","slug":"unexpected-number-of-inputs-s","errorCode":null,"errorMessage":"Unexpected number of inputs: %s","messagePattern":"Unexpected number of inputs: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":85,"sourceCode":"\n  return GroupByKeyInputVisitor(deterministic_key_coders)\n\n\ndef validate_pipeline_graph(pipeline_proto):\n  \"\"\"Ensures this is a correctly constructed Beam pipeline.\n  \"\"\"\n  def get_coder(pcoll_id):\n    return pipeline_proto.components.coders[\n        pipeline_proto.components.pcollections[pcoll_id].coder_id]\n\n  def validate_transform(transform_id):\n    transform_proto = pipeline_proto.components.transforms[transform_id]\n\n    # Currently the only validation we perform is that GBK operations have\n    # their coders set properly.\n    if transform_proto.spec.urn == common_urns.primitives.GROUP_BY_KEY.urn:\n      if len(transform_proto.inputs) != 1:\n        raise ValueError(\"Unexpected number of inputs: %s\" % transform_proto)\n      if len(transform_proto.outputs) != 1:\n        raise ValueError(\"Unexpected number of outputs: %s\" % transform_proto)\n      input_coder = get_coder(next(iter(transform_proto.inputs.values())))\n      output_coder = get_coder(next(iter(transform_proto.outputs.values())))\n      if input_coder.spec.urn != common_urns.coders.KV.urn:\n        raise ValueError(\n            \"Bad coder for input of %s: %s\" % (transform_id, input_coder))\n      if output_coder.spec.urn != common_urns.coders.KV.urn:\n        raise ValueError(\n            \"Bad coder for output of %s: %s\" % (transform_id, output_coder))\n      input_key_coder_id = input_coder.component_coder_ids[0]\n      output_key_coder_id = output_coder.component_coder_ids[0]\n      if input_key_coder_id != output_key_coder_id:\n        raise ValueError(\n            \"Input key coder %s does not match output key coder %s for \"\n            \"transform %s\" %\n            (input_key_coder_id, output_key_coder_id, transform_id))\n      output_values_coder_id = output_coder.component_coder_ids[1]","sourceCodeStart":67,"sourceCodeEnd":103,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L67-L103","documentation":"Raised by validate_transform while validating a pipeline's model proto: a GROUP_BY_KEY (GroupByKey) transform must have exactly one input, and the pipeline's proto declares a different number. Beam validates the serialized pipeline graph before handing it to a runner. This indicates a malformed or hand-assembled pipeline proto rather than a normal user-level Beam error.","triggerScenarios":"Submitting a pipeline whose proto contains a GROUP_BY_KEY transform with 0 or 2+ entries in transform_proto.inputs — typically from manually constructed protos, cross-language portability graphs, or corrupted/older serialized pipeline JSON passed through validate_pipeline_graph.","commonSituations":"Building Pipeline protos programmatically or in tooling; round-tripping pipeline graphs through external storage; mixing SDK versions where a runner or translator produced a nonstandard GBK node.","solutions":["Inspect the transform_proto.inputs of the failing GBK transform (ID is in the message) and correct it to exactly one input PCollection","Regenerate the pipeline from Pipeline/Beam Python code instead of editing or constructing protos by hand","Verify all SDK/runner components use compatible Beam versions; recompile/re-serialize the pipeline proto","If writing custom translators, ensure your GBK replacement has exactly one input map entry"],"exampleFix":"// before\ngbk.spec.inputs['main'] = pc_in\ngbk.spec.inputs['side'] = pc_extra  # 2 inputs -> raises\n// after\ndel gbk.spec.inputs['side']  # exactly one input\nassert len(gbk.spec.inputs) == 1","handlingStrategy":"validation","validationCode":"from apache_beam.portability.api import beam_runner_api_pb2\ndef check_gbk_inputs(t: beam_runner_api_pb2.PTransform) -> bool:\n    return (t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn\n            or len(t.inputs) == 1)","typeGuard":"def is_wellformed_gbk(t) -> bool:\n    return t.spec.urn != 'beam:transform:group_by_key:v1' or len(t.inputs) == 1","tryCatchPattern":"try:\n    validate_pipeline_graph(pipeline_proto)\nexcept ValueError as e:\n    if 'Unexpected number of inputs' in str(e):\n        rebuild_pipeline_graph(pipeline_proto)\n    else:\n        raise","preventionTips":["Always build pipelines via the Beam Python API, not by hand-writing protos","Run validate_pipeline_graph on generated protos in CI tests","Never edit serialized pipeline JSON/protos with ad-hoc tooling","Pin one Beam SDK version across producer and consumer of pipeline protos"],"tags":["python","apache-beam","pipeline-validation","group-by-key"],"backgroundTag":"schema-validation-failed","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"}