{"record":{"id":"8c3cba208bc6830d","repo":"apache/beam","slug":"pcollection-must-be-of-kv-type-v","errorCode":null,"errorMessage":"pcollection must be of KV type: %v","messagePattern":"pcollection must be of KV type: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/validate.go","lineNumber":31,"sourceCode":"// See the License for the specific language governing permissions and\n// limitations under the License.\n\npackage beam\n\nimport (\n\t\"fmt\"\n\t\"reflect\"\n\n\t\"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex\"\n\t\"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors\"\n)\n\n// ValidateKVType panics if the type of the PCollection is not KV<A,B>.\n// It returns (A,B).\nfunc ValidateKVType(col PCollection) (typex.FullType, typex.FullType) {\n\tt := col.Type()\n\tif !typex.IsKV(t) {\n\t\tpanic(fmt.Sprintf(\"pcollection must be of KV type: %v\", col))\n\t}\n\treturn t.Components()[0], t.Components()[1]\n}\n\n// ValidateNonCompositeType panics if the type of the PCollection is not a\n// composite type. It returns the type.\nfunc ValidateNonCompositeType(col PCollection) typex.FullType {\n\tt := col.Type()\n\tif typex.IsComposite(t.Type()) {\n\t\tpanic(fmt.Sprintf(\"pcollection must be of non-composite type: %v\", col))\n\t}\n\treturn t\n}\n\n// validate validates and processes the input collection and options. Private convenience\n// function.\nfunc validate(s Scope, col PCollection, opts []Option) ([]SideInput, map[string]reflect.Type, error) {\n\tif !s.IsValid() {","sourceCodeStart":13,"sourceCodeEnd":49,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/validate.go#L13-L49","documentation":"beam.ValidateKVType asserts that a PCollection's element type is KV<A,B> and panics with \"pcollection must be of KV type\" if not. It exists because keyed transforms (CombinePerKey, MeanPerKey, per-key aggregations) only operate on key/value pairs. The message includes the offending PCollection's type so the mismatch is visible.","triggerScenarios":"Calling beam.CombinePerKey, MeanPerKey, LargestPerKey, SmallestPerKey, ApproximateWeightedQuantiles (or combinePerKey internally) on a PCollection that emits plain values instead of beam.KV pairs.","commonSituations":"Forgetting beam.KV() when building the input collection; applying a per-key stats transform directly to a PCollection<string> or PCollection<int>; upstream ParDo changed output type from KV to a plain value.","solutions":["Wrap elements as beam.KV before the transform: beam.KV{k, v} via ParDo emitting KV pairs.","Insert a Map/ParDo that converts your element type to beam.KV<K,V>.","Check col.Type() / print the PCollection type and confirm typex.IsKV holds.","If you don't need keying, use the non-keyed variants (Mean, Largest, etc.) instead."],"exampleFix":"// before\nsummed := beam.CombinePerKey(s, ints) // ints is PCollection<int>\n// after\npairs := beam.ParDo(s, func(i int) beam.KV { return beam.KV{\"key\", i} }, ints)\nsummed := beam.CombinePerKey(s, pairs)","handlingStrategy":"validation","validationCode":"col := beam.ParDo(s, func(x T) beam.KV { return beam.KV{k, x} }, input)\n// ensure the collection is KV before CombinePerKey\n// (in tests: if !typex.IsKV(col.Type()) { t.Fatal(...) })","typeGuard":"func isKV(col beam.PCollection) bool {\n    return typex.IsKV(col.Type())\n}","tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if s, ok := r.(string); ok && strings.Contains(s, \"must be of KV type\") {\n            log.Fatalf(\"wrong transform input: %s\", s)\n        }\n        panic(r)\n    }\n}()","preventionTips":["Emit beam.KV explicitly from the upstream ParDo","Check the PCollection type with typex.IsKV in tests","Use non-keyed transforms when data has no key","Re-verify types after refactors that change DoFn outputs"],"tags":["go","panic","type-mismatch","beam-transform"],"backgroundTag":"type-mismatch","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"}