{"record":{"id":"b91e35c56eb3f84a","repo":"apache/beam","slug":"failed-to-update-ptransform-spec-v","errorCode":null,"errorMessage":"failed to update PTransform spec: %v","messagePattern":"failed to update PTransform spec: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/graphx/translate.go","lineNumber":333,"sourceCode":"}\n\n// updateIfCombineComposite examines the scope tree and sets the PTransform Spec\n// to be a CombinePerKey with a CombinePayload if it's a liftable composite.\n// Beam Portability requires that composites contain an implementation for runners\n// that don't understand the URN and Payload, which this lightly checks for.\nfunc (m *marshaller) updateIfCombineComposite(s *ScopeTree, transform *pipepb.PTransform) error {\n\tif s.Scope.Name != graph.CombinePerKeyScope ||\n\t\tlen(s.Edges) != 2 ||\n\t\tlen(s.Edges[0].Edge.Input) != 1 ||\n\t\tlen(s.Edges[1].Edge.Output) != 1 ||\n\t\ts.Edges[1].Edge.Op != graph.Combine {\n\t\treturn nil\n\t}\n\n\tedge := s.Edges[1].Edge\n\tacID, err := m.coders.Add(edge.AccumCoder)\n\tif err != nil {\n\t\treturn errors.Wrapf(err, \"failed to update PTransform spec: %v\", transform)\n\t}\n\tmustEncodeMultiEdge, err := mustEncodeMultiEdgeBase64(edge)\n\tif err != nil {\n\t\treturn errors.Wrapf(err, \"failed to update PTransform spec: %v\", transform)\n\t}\n\tpayload := &pipepb.CombinePayload{\n\t\tCombineFn: &pipepb.FunctionSpec{\n\t\t\tUrn:     URNDoFn,\n\t\t\tPayload: []byte(mustEncodeMultiEdge),\n\t\t},\n\t\tAccumulatorCoderId: acID,\n\t}\n\ttransform.Spec = &pipepb.FunctionSpec{Urn: URNCombinePerKey, Payload: protox.MustEncode(payload)}\n\treturn nil\n}\n\nfunc getSideWindowMappingUrn(winFn *window.Fn) string {\n\tvar mappingUrn string","sourceCodeStart":315,"sourceCodeEnd":351,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/graphx/translate.go#L315-L351","documentation":"Wrapped by updateIfCombineComposite when the accumulator coder for a Combine transform cannot be added to the marshaller's coder registry (m.coders.Add fails). The pipeline proto for the Combine PTransform cannot be built without registering the accumulator coder, so marshalling aborts with the transform name attached.","triggerScenarios":"Marshalling a pipeline containing beam.Combine whose edge.AccumCoder cannot be serialized into the coder registry — typically because the accumulator type has no derivable coder.","commonSituations":"CombineFns with accumulator types that are structs with unexported fields, channels, funcs, or other non-encodable Go types; use of beam.Combine with custom types lacking coder registration.","solutions":["Ensure the CombineFn accumulator type only contains coder-supported exported fields","Check the inner coder error from m.coders.Add to identify the offending type","Register a custom coder for the accumulator type (graphx.CustomCoder)","Replace the accumulator with an equivalent encodable type"],"exampleFix":"// before\ntype acc struct { mu sync.Mutex; sum int }\n// after\ntype acc struct { Sum int64 }","handlingStrategy":"validation","validationCode":"// Verify accumulator type is derivable as a coder before building the pipeline\nvar acc MyAcc\nif _, err := graphx.MakeCoderFromType(reflect.TypeOf(acc)); err != nil {\n    return fmt.Errorf(\"MyAcc not codable: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"if _, err := graphx.Marshal(p); err != nil {\n    if strings.Contains(err.Error(), \"failed to update PTransform spec\") {\n        // fall back to logging pipeline and reporting coder failure\n        log.Fatalf(\"combine coder failure: %v\", err)\n    }\n    return err\n}","preventionTips":["Avoid funcs, channels and unexported fields in accumulator structs","Register custom coders for complex accumulator types","Marshal pipelines in CI to catch coder issues early"],"tags":["go","apache-beam","coder","combine"],"backgroundTag":"coder-serialization-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"}