{"record":{"id":"bdcfb3e11202b708","repo":"apache/beam","slug":"combine-cannot-follow-multi-input-cogbk-v","errorCode":null,"errorMessage":"Combine cannot follow multi-input CoGBK: %v","messagePattern":"Combine cannot follow multi-input CoGBK: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/edge.go","lineNumber":462,"sourceCode":"// 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 {\n\t\t// drop accumulator and irrelevant parameters\n\t\tsynth.Param = funcx.SubParams(f.Param, f.Params(funcx.FnValue)[1:]...)\n\t} else {","sourceCodeStart":444,"sourceCodeEnd":480,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/edge.go#L444-L480","documentation":"After confirming the input is CoGBK, NewCombine checks that it has at most 2 components (key + one value). Combines cannot follow multi-input CoGBK (results of multiple-input CoGroupByKey), so a CoGBK with more than 2 components aborts graph construction. Combining is only defined for grouped K/V pairs, not wider grouped tuples.","triggerScenarios":"Applying beam.CombinePerKey to the output of a multi-input CoGroupByKey (CoGBK with 3+ components), e.g. cogrouping 3+ PCollections and then trying to combine per key.","commonSituations":"Co-grouping several streams and then attempting CombinePerKey on the merged grouped result; chaining Combine after a join-like CoGBK instead of expanding it first.","solutions":["Expand the multi-input CoGBK with a ParDo that extracts the single value component you want, re-group if needed, then CombinePerKey.","Combine each input PCollection before the CoGroupByKey, then merge the aggregated results.","Use a ParDo over the multi-input CoGBK to implement the per-key aggregation manually.","If only two inputs are truly needed, replace the multi-input CoGBK with a plain GroupByKey on a KV PCollection."],"exampleFix":"// before\ncombined := beam.CombinePerKey(s, sumFn, multiCogbk) // CoGBK has 3 components\n// after\nextract := func(k beam.KV, a, b iter.Interface) beam.KV { /* merge side values */ return k }\nkv := beam.ParDo(s, extract, multiCogbk)\ncombined := beam.CombinePerKey(s, sumFn, kv)","handlingStrategy":"validation","validationCode":"// Go: aggregate before cogrouping instead of combining a multi-input CoGBK\nsummedA := beam.CombinePerKey(s, sumFn, a)\nsummedB := beam.CombinePerKey(s, sumFn, b)\nmerged := beam.CoGroupByKey(s, summedA, summedB)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never apply CombinePerKey to CoGroupByKey output; expand it or aggregate beforehand.","Combine each input stream separately, then cogroup the summaries.","Use ParDo for custom per-key aggregation over multi-input CoGBK results."],"tags":["beam-go","combine","cogbk","arity"],"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"}