{"record":{"id":"0123b9e706db0a47","repo":"apache/beam","slug":"reshuffle-input-type-cannot-be-cogbk-v","errorCode":null,"errorMessage":"Reshuffle input type cannot be CoGBK: %v","messagePattern":"Reshuffle input type cannot be CoGBK: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/edge.go","lineNumber":577,"sourceCode":"\nfunc inputBounded(in []*Node) bool {\n\tif len(in) == 0 {\n\t\treturn true\n\t}\n\treturn in[0].Bounded()\n}\n\n// NewReshuffle inserts a new Reshuffle edge into the graph.\nfunc NewReshuffle(g *Graph, s *Scope, in *Node) (*MultiEdge, error) {\n\taddContext := func(err error, s *Scope) error {\n\t\treturn errors.WithContextf(err, \"creating new Reshuffle in scope %v\", s)\n\t}\n\tn := g.NewNode(in.Type(), in.WindowingStrategy(), in.Bounded())\n\tn.Coder = in.Coder\n\n\tt := in.Type()\n\tif typex.IsCoGBK(t) {\n\t\treturn nil, addContext(errors.Errorf(\"Reshuffle input type cannot be CoGBK: %v\", t), s)\n\t}\n\tedge := g.NewEdge(s)\n\tedge.Op = Reshuffle\n\tedge.Input = []*Inbound{{Kind: Main, From: in, Type: t}}\n\tedge.Output = []*Outbound{{To: n, Type: t}}\n\treturn edge, nil\n}\n","sourceCodeStart":559,"sourceCodeEnd":585,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/edge.go#L559-L585","documentation":"NewReshuffle rejects input whose type is CoGBK: a reshuffle (injecting a GBK round-trip to break fusion) only makes sense for ordinary keyed or plain element streams, not unexploded multi-input grouping results. The graph builder throws this when TryReshuffle is given a CoGBK-typed PCollection.","triggerScenarios":"Calling beam.Reshuffle on the output of beam.CoGroupByKey (CoGBK type), typically to redistribute work after a cogroup.","commonSituations":"Inserting Reshuffle after joins/cogroups for load balancing; fusion-breaking passes added by hand that assume plain element types.","solutions":["Expand the CoGBK first with a ParDo emitting plain or KV values, then apply beam.Reshuffle.","Reshuffle the pre-cogroup inputs instead of the CoGBK output.","If the CoGBK came from a single-input group, use GroupByKey output (KV) rather than multi-input CoGBK.","Remove the Reshuffle if it is not required (Beam inserts reshuffles automatically on some runners)."],"exampleFix":"// before\nbeam.Reshuffle(s, cogbkPC) // CoGBK input rejected\n// after\nflat := beam.ParDo(s, func(k beam.KV, v iter.Interface) beam.KV { return beam.KV{k, v} }, cogbkPC)\nreshuffled := beam.Reshuffle(s, flat)","handlingStrategy":"validation","validationCode":"// Go: expand CoGBK before reshuffling\nflat := beam.ParDo(s, expandCoGBKFn, cogbkPC)\nreshuffled := beam.Reshuffle(s, flat)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Reshuffle only plain or KV PCollections, never CoGBK outputs.","Reshuffle inputs before a cogroup if load balancing is needed there.","Let the runner handle fusion-breaking rather than hand-inserting reshuffles on grouped types."],"tags":["beam-go","reshuffle","cogbk","unsupported-type"],"backgroundTag":"incompatible-source-type","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"}