apache/beam · error · CoderException

input schema does not match coder schema

Error message

input schema does not match coder schema

What it means

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.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGbkResult.java:295

    @Override
    public List<? extends Coder<?>> getCoderArguments() {
      return ImmutableList.of(unionCoder);
    }

    public CoGbkResultSchema getSchema() {
      return schema;
    }

    public UnionCoder getUnionCoder() {
      return unionCoder;
    }

    @Override
    @SuppressWarnings("unchecked")
    public void encode(CoGbkResult value, OutputStream outStream)
        throws CoderException, IOException {
      if (!schema.equals(value.getSchema())) {
        throw new CoderException("input schema does not match coder schema");
      }
      if (schema.size() == 0) {
        return;
      }
      for (int unionTag = 0; unionTag < schema.size(); unionTag++) {
        tagListCoder(unionTag).encode(value.valueMap.get(unionTag), outStream);
      }
    }

    @Override
    public CoGbkResult decode(InputStream inStream) throws CoderException, IOException {
      if (schema.size() == 0) {
        return new CoGbkResult(schema, ImmutableList.<Iterable<?>>of());
      }
      List<Iterable<?>> valueMap = Lists.newArrayListWithExpectedSize(schema.size());
      for (int unionTag = 0; unionTag < schema.size(); unionTag++) {
        valueMap.add(tagListCoder(unionTag).decode(inStream));
      }

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the TupleTagList passed to CoGbkResultCoder.of matches the schema of every CoGbkResult being encoded
2. Re-run the pipeline so all coder inference happens against the current KeyedPCollectionTuple tags
3. If results were persisted, re-generate them with the pipeline version that produces the matching schema

Example fix

// before
CoGbkResultCoder.of(oldSchema).encode(result, out);
// after
if (result.getSchema().equals(newSchema)) {
  CoGbkResultCoder.of(newSchema).encode(result, out);
}

if (!result.getSchema().equals(coderSchema)) { throw new IllegalStateException("CoGbkResult schema mismatch before encode"); }

try { coder.encode(result, out); } catch (CoderException e) { throw new IOException("Schema mismatch encoding CoGbkResult", e); }

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• cannot encode a null String
• cannot encode a null Integer
• cannot encode a null ValueKind
• cannot encode a null Integer
• cannot encode a null Long

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/8656f262e9d63210. Report an issue: GitHub.