{"record":{"id":"be706c72d9d7a500","repo":"apache/beam","slug":"input-type-must-be-kv-v","errorCode":null,"errorMessage":"input type must be KV: %v","messagePattern":"input type must be KV: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/graph/edge.go","lineNumber":209,"sourceCode":"func (e *MultiEdge) String() string {\n\treturn fmt.Sprintf(\"%v: %v %v -> %v\", e.id, e.Op, e.Input, e.Output)\n}\n\n// NOTE(herohde) 4/28/2017: In general, we have no good coder guess for outgoing\n// nodes, unless we add a notion of default coder for arbitrary types. We leave\n// that to the beam layer.\n\n// NewCoGBK inserts a new CoGBK edge into the graph.\nfunc NewCoGBK(g *Graph, s *Scope, ns []*Node) (*MultiEdge, error) {\n\taddContext := func(err error, s *Scope) error {\n\t\treturn errors.WithContextf(err, \"creating new CoGBK in scope %v\", s)\n\t}\n\n\tif len(ns) == 0 {\n\t\treturn nil, addContext(errors.New(\"needs at least 1 input\"), s)\n\t}\n\tif !typex.IsKV(ns[0].Type()) {\n\t\treturn nil, addContext(errors.Errorf(\"input type must be KV: %v\", ns[0]), s)\n\t}\n\n\t// (1) Create CoGBK result type: KV<T,U>, .., KV<T,Z> -> CoGBK<T,U,..,Z>.\n\n\tc := ns[0].Coder.Components[0]\n\tw := inputWindow(ns)\n\tbounded := inputBounded(ns)\n\tcomp := []typex.FullType{c.T, ns[0].Type().Components()[1]}\n\n\tfor i := 1; i < len(ns); i++ {\n\t\tn := ns[i]\n\t\tif !typex.IsKV(n.Type()) {\n\t\t\treturn nil, addContext(errors.Errorf(\"input type must be KV: %v\", n), s)\n\t\t}\n\t\tif !n.Coder.Components[0].Equals(c) {\n\t\t\treturn nil, addContext(errors.Errorf(\"key coder for %v is %v, want %v\", n, n.Coder.Components[0], c), s)\n\t\t}\n\t\tif !w.Equals(n.WindowingStrategy()) {","sourceCodeStart":191,"sourceCodeEnd":227,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/graph/edge.go#L191-L227","documentation":"Error (wrapped with scope context) returned by NewCoGBK when any input node to a CoGBK edge is not of KV type. CoGBK groups by key across all inputs, so every input PCollection must be KV<K,V>; the message names the offending node. Usually surfaced from beam.TryCoGroupByKey when one of the joined collections lacks a KV type.","triggerScenarios":"Calling beam.CoGroupByKey (TryCoGroupByKey → NewCoGBK) with a first input of non-KV type, e.g. a plain PCollection<string> or a PCollection of structs.","commonSituations":"Forgetting beam.KV when building the input to a CoGroupByKey, or changing upstream transforms so the element type is no longer KV.","solutions":["Wrap inputs in beam.KV via beam.ParDo returning KV values before CoGroupByKey","Use beam.KV helpers to make the key/value structure explicit","Check upstream transform output types with typex.IsKV in tests"],"exampleFix":"// before\nout := beam.ParDo(s, func(w Word) string { return w.Text }, words)\nbeam.CoGroupByKey(s, counts) // counts not KV\n// after\nkeyed := beam.ParDo(s, func(w Word) (string, Word) { return w.Key, w }, words)\nbeam.CoGroupByKey(s, keyed)","handlingStrategy":"validation","validationCode":"if !typex.IsKV(col.Type()) {\n    return fmt.Errorf(\"input must be KV, got %v\", col.Type())\n}","typeGuard":"func isKVInput(n *beam.Node) bool { return n != nil && typex.IsKV(n.Type()) }","tryCatchPattern":"if err := beam.TryCoGroupByKey(s, in); err != nil {\n    return fmt.Errorf(\"CoGroupByKey needs KV inputs: %w\", err)\n}","preventionTips":["Key all inputs with beam.KV before CoGroupByKey","Add typex.IsKV assertions after keying transforms","Re-check types after refactoring upstream DoFns"],"tags":["go","beam","cogbk","type-check"],"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-14T21:17:11.552Z"}