{"record":{"id":"f51cbe9daea6999b","repo":"apache/beam","slug":"input-value-coder-s-does-not-match-output-value-coder-s-for","errorCode":null,"errorMessage":"Input value coder %s does not match output value coder %s for transform %s","messagePattern":"Input value coder (.+?) does not match output value coder (.+?) for transform (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/pipeline_utils.py","lineNumber":116,"sourceCode":"      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]\n      output_values_coder = pipeline_proto.components.coders[\n          output_values_coder_id]\n      if output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn:\n        raise ValueError(\n            \"Output value coder %s for transform %s must be an iterable \"\n            \"coder, but uses URN %s\" % (\n                output_values_coder_id,\n                transform_id,\n                output_values_coder.spec.urn))\n      input_value_coder_id = input_coder.component_coder_ids[1]\n      output_value_coder_id = output_values_coder.component_coder_ids[0]\n      if output_value_coder_id != input_value_coder_id:\n        raise ValueError(\n            \"Input value coder %s does not match output value coder %s for \"\n            \"transform %s\" %\n            (input_value_coder_id, output_value_coder_id, transform_id))\n    elif transform_proto.spec.urn == common_urns.primitives.ASSIGN_WINDOWS.urn:\n      if not transform_proto.inputs:\n        raise ValueError(\"Missing input for transform: %s\" % transform_proto)\n    elif transform_proto.spec.urn == common_urns.primitives.PAR_DO.urn:\n      if not transform_proto.inputs:\n        raise ValueError(\"Missing input for transform: %s\" % transform_proto)\n\n    for t in transform_proto.subtransforms:\n      validate_transform(t)\n\n  for t in pipeline_proto.root_transform_ids:\n    validate_transform(t)\n\n\ndef _dep_key(dep):","sourceCodeStart":98,"sourceCodeEnd":134,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/pipeline_utils.py#L98-L134","documentation":"Raised by validate_transform when the value component coder of a GROUP_BY_KEY's output does not match the value coder of its input. GBK preserves value encodings (output is KV<K, Iterable<V>> over the same V encoding as the input KV<K,V>); a mismatch indicates corrupted or rewritten coder wiring in the proto.","triggerScenarios":"output_values_coder.component_coder_ids[0] (the iterable's element coder) != input_coder.component_coder_ids[1] for a GBK transform — usually after custom coder ID reassignment or graph merging.","commonSituations":"Pipeline graph rewrite/optimization tools reassigning coder IDs; manually stitched pipeline fragments; cross-SDK serialization with divergent coder registries.","solutions":["Set the GBK output iterable coder's element coder ID to the input KV's value coder ID","Rebuild the pipeline from Beam code so value coders are inferred identically on both sides","Review custom coder rewrite logic to preserve value coder identity through GBK","Re-serialize the pipeline with one consistent SDK version"],"exampleFix":"// before\niter_coder = make_iterable_coder(other_value_coder_id)\n// after\niter_coder = make_iterable_coder(input_coder.component_coder_ids[1])","handlingStrategy":"validation","validationCode":"def check_value_coders_match(pipeline, t) -> bool:\n    if t.spec.urn != common_urns.primitives.GROUP_BY_KEY.urn:\n        return True\n    in_c = get_coder(next(iter(t.inputs.values())))\n    out_c = get_coder(next(iter(t.outputs.values())))\n    values_coder = pipeline.components.coders[out_c.component_coder_ids[1]]\n    return values_coder.component_coder_ids[0] == in_c.component_coder_ids[1]","typeGuard":"def values_consistent(pipeline, t) -> bool:\n    in_c, out_c = gbk_coders(pipeline, t)\n    element_id = pipeline.components.coders[out_c.component_coder_ids[1]].component_coder_ids[0]\n    return element_id == in_c.component_coder_ids[1]","tryCatchPattern":"try:\n    validate_pipeline_graph(pipeline_proto)\nexcept ValueError as e:\n    if 'does not match output value coder' in str(e):\n        point_iterable_at_input_value_coder(pipeline_proto)\n    else:\n        raise","preventionTips":["Point the GBK output iterable coder's element at the input KV's value coder ID","Preserve coder IDs verbatim through graph rewrites","Unit-test graph-rewrite tools asserting value coder identity across GBK","Serialize and consume pipelines with one SDK version"],"tags":["python","apache-beam","pipeline-validation","coders"],"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"}