{"record":{"id":"8656f262e9d63210","repo":"apache/beam","slug":"input-schema-does-not-match-coder-schema","errorCode":null,"errorMessage":"input schema does not match coder schema","messagePattern":"input schema does not match coder schema","errorType":"exception","errorClass":"CoderException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGbkResult.java","lineNumber":295,"sourceCode":"    @Override\n    public List<? extends Coder<?>> getCoderArguments() {\n      return ImmutableList.of(unionCoder);\n    }\n\n    public CoGbkResultSchema getSchema() {\n      return schema;\n    }\n\n    public UnionCoder getUnionCoder() {\n      return unionCoder;\n    }\n\n    @Override\n    @SuppressWarnings(\"unchecked\")\n    public void encode(CoGbkResult value, OutputStream outStream)\n        throws CoderException, IOException {\n      if (!schema.equals(value.getSchema())) {\n        throw new CoderException(\"input schema does not match coder schema\");\n      }\n      if (schema.size() == 0) {\n        return;\n      }\n      for (int unionTag = 0; unionTag < schema.size(); unionTag++) {\n        tagListCoder(unionTag).encode(value.valueMap.get(unionTag), outStream);\n      }\n    }\n\n    @Override\n    public CoGbkResult decode(InputStream inStream) throws CoderException, IOException {\n      if (schema.size() == 0) {\n        return new CoGbkResult(schema, ImmutableList.<Iterable<?>>of());\n      }\n      List<Iterable<?>> valueMap = Lists.newArrayListWithExpectedSize(schema.size());\n      for (int unionTag = 0; unionTag < schema.size(); unionTag++) {\n        valueMap.add(tagListCoder(unionTag).decode(inStream));\n      }","sourceCodeStart":277,"sourceCodeEnd":313,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGbkResult.java#L277-L313","documentation":"CoGroupByKey's result coder verifies that the schema of the CoGbkResult being encoded exactly equals the schema the CoGbkResultCoder was constructed with before encoding. If the value's CoGbkResultSchema differs (different TupleTagList), encoding is aborted with a CoderException because the union tags would be serialized inconsistently between workers.","triggerScenarios":"Encoding a CoGbkResult whose schema was built from a different TupleTagList than the one used to create the coder, e.g. after changing the set of TupleTags passed to KeyedPCollectionTuple.withoutCoderConfirmation or mixing results from differently-shaped CoGroupByKey transforms.","commonSituations":"Refactoring a CoGroupByKey pipeline by adding/removing a TupleTag from the input while cached or serialized coder metadata still reflects the old schema; deserializing old written results with a new schema.","solutions":["Ensure the TupleTagList passed to CoGbkResultCoder.of matches the schema of every CoGbkResult being encoded","Re-run the pipeline so all coder inference happens against the current KeyedPCollectionTuple tags","If results were persisted, re-generate them with the pipeline version that produces the matching schema"],"exampleFix":"// before\nCoGbkResultCoder.of(oldSchema).encode(result, out);\n// after\nif (result.getSchema().equals(newSchema)) {\n  CoGbkResultCoder.of(newSchema).encode(result, out);\n}","handlingStrategy":"validation","validationCode":"if (!result.getSchema().equals(coderSchema)) { throw new IllegalStateException(\"CoGbkResult schema mismatch before encode\"); }","typeGuard":null,"tryCatchPattern":"try { coder.encode(result, out); } catch (CoderException e) { throw new IOException(\"Schema mismatch encoding CoGbkResult\", e); }","preventionTips":["Always derive the coder from the same TupleTagList used to build results","Re-run coder inference after changing join inputs","Avoid persisting CoGbkResults across schema refactors without regeneration"],"tags":["beam","coder","schema-mismatch","serialization"],"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-14T16:17:12.679Z"}