{"record":{"id":"090b87c9ab43045e","repo":"apache/beam","slug":"combine-requires-cogbk-type-v","errorCode":null,"errorMessage":"Combine requires CoGBK type: %v","messagePattern":"Combine requires CoGBK type: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/edge.go","lineNumber":459,"sourceCode":"\n// CombinePerKeyScope is the Go SDK canonical name for the combine composite\n// scope. With Beam Portability, \"primitive\" composite transforms like\n// combine have their URNs & payloads attached to a high level scope, with a\n// default representation beneath. The use of this const permits the\n// translation layer to confirm the SDK expects this combine to be liftable\n// by a runner and should set this scope's URN and Payload accordingly.\nconst CombinePerKeyScope = \"CombinePerKey\"\n\n// NewCombine inserts a new Combine edge into the graph. Combines cannot have side\n// input.\nfunc NewCombine(g *Graph, s *Scope, u *CombineFn, in *Node, ac *coder.Coder, typedefs map[string]reflect.Type) (*MultiEdge, error) {\n\taddContext := func(err error, s *Scope) error {\n\t\treturn errors.WithContextf(err, \"creating new Combine in scope %v\", s)\n\t}\n\n\tinT := in.Type()\n\tif !typex.IsCoGBK(inT) {\n\t\treturn nil, addContext(errors.Errorf(\"Combine requires CoGBK type: %v\", inT), s)\n\t}\n\tif len(inT.Components()) > 2 {\n\t\treturn nil, addContext(errors.Errorf(\"Combine cannot follow multi-input CoGBK: %v\", inT), s)\n\t}\n\n\t// Create a synthetic function for binding purposes. It takes main input\n\t// and returns the output type -- but hides the accumulator.\n\t//\n\t//  (1) If AddInput exists, then it lists the main inputs. If not,\n\t//      the only main input type is the accumulator type.\n\t//  (2)\tIf ExtractOutput exists then it returns the output type. If not,\n\t//      then the accumulator is the output type.\n\t//\n\t// MergeAccumulators is guaranteed to exist. We do not allow the accumulator\n\t// to be a tuple type (i.e., so one can't define a inline KV merge function).\n\n\tsynth := &funcx.Fn{}\n\tif f := u.AddInputFn(); f != nil {","sourceCodeStart":441,"sourceCodeEnd":477,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/edge.go#L441-L477","documentation":"NewCombine requires the incoming edge's type to be a CoGBK type, since CombinePerKey operates on grouped key/value collections. If `in.Type()` is not CoGBK (e.g., a plain PCollection), graph construction fails with this error. It enforces that Combine is always preceded by a grouping operation.","triggerScenarios":"Calling beam.CombinePerKey on a PCollection that was not grouped (no GroupByKey/CoGBK), or applying a combineFn via NewCombine to a raw element stream.","commonSituations":"Manually building a graph and skipping the GroupByKey before Combine; calling low-level graph APIs instead of beam.Combine; a refactor removed the preceding GroupByKey.","solutions":["Use beam.CombinePerKey (the public API), which performs the required CoGBK internally.","If building graphs manually, insert a GroupByKey/CoGBK node before the Combine edge.","If the intent is element-wise aggregation without keys, use beam.Combine instead of beam.CombinePerKey.","Verify the input PCollection actually comes from a grouping transform before getCombineEdge runs."],"exampleFix":"// before\ncombined := graph.NewCombine(...) // input is ungrouped PCollection<string>\n// after\n// at pipeline level:\ncombined := beam.CombinePerKey(s, combineFn, kvPC) // kvPC is PCollection<KV<K,V>>; CombinePerKey creates the CoGBK","handlingStrategy":"validation","validationCode":"// Go: use the public API so the CoGBK is created for you\n// combined := beam.CombinePerKey(s, combineFn, kvPC)  // kvPC: PCollection<KV<K,V>>","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Prefer beam.CombinePerKey/beam.Combine over manual graph construction.","Only use low-level graph.NewCombine when a CoGBK node provably precedes the edge.","For keyless aggregation use beam.Combine on a plain PCollection."],"tags":["beam-go","combine","cogbk","graph-construction"],"backgroundTag":"unsupported-operation","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"}