{"record":{"id":"3289b42b71e7fb1d","repo":"apache/beam","slug":"groupintobatches-input-pcollection-must-be-kv-typed-got-v","errorCode":null,"errorMessage":"GroupIntoBatches: input PCollection must be KV-typed; got %v","messagePattern":"GroupIntoBatches: input PCollection must be KV-typed; got (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/transforms/batch/batch.go","lineNumber":583,"sourceCode":"}\n\n// GroupIntoBatches groups the values of the input PCollection<KV<K, V>>\n// into batches of up to params.BatchSize elements (or\n// params.BatchSizeBytes bytes) per key and emits them as\n// PCollection<KV<K, []V>>.\n//\n// The input must be KV-typed. The key coder must be deterministic;\n// non-deterministic key coders would corrupt state keying. Panics at\n// pipeline build time on invalid params, non-KV input, zero limits, or\n// a non-deterministic key coder.\nfunc GroupIntoBatches(s beam.Scope, params Params, col beam.PCollection) beam.PCollection {\n\ts = s.Scope(\"batch.GroupIntoBatches\")\n\n\tif err := params.validate(); err != nil {\n\t\tpanic(fmt.Errorf(\"GroupIntoBatches: %w\", err))\n\t}\n\tif !typex.IsKV(col.Type()) {\n\t\tpanic(fmt.Errorf(\n\t\t\t\"GroupIntoBatches: input PCollection must be KV-typed; got %v\", col.Type()))\n\t}\n\n\tkeyFT := col.Type().Components()[0]\n\tvalFT := col.Type().Components()[1]\n\n\tif !beam.NewCoder(keyFT).IsDeterministic() {\n\t\tpanic(fmt.Errorf(\n\t\t\t\"GroupIntoBatches: key coder for type %v is not deterministic. \"+\n\t\t\t\t\"Register a deterministic custom coder with \"+\n\t\t\t\t\"coder.RegisterDeterministicCoder, or use a deterministic key \"+\n\t\t\t\t\"type (string, []byte, bool, integer, float)\", keyFT.Type()))\n\t}\n\n\tsizerKind := sizerNone\n\tif params.BatchSizeBytes > 0 {\n\t\tif !isBuiltinSizeable(valFT.Type()) {\n\t\t\tpanic(fmt.Errorf(","sourceCodeStart":565,"sourceCodeEnd":601,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/transforms/batch/batch.go#L565-L601","documentation":"GroupIntoBatches requires a KV-typed PCollection because it groups elements by key. A non-KV input cannot be sharded/batched, so the transform panics with the actual input type in the message. Re-shape your data into KV pairs first.","triggerScenarios":"Passing a PCollection of plain values (e.g. beam.Create of ints/strings) or a non-KV composite type directly to GroupIntoBatches, failing typex.IsKV(col.Type()).","commonSituations":"Forgetting a preceding beam.ParDo/Map that wraps elements into KV pairs; feeding output of a transform whose type changed after a refactor.","solutions":["Map elements to KV pairs with beam.AddFixedKey or a ParDo producing KV<K,V>","Check the PCollection type with col.Type() before the call","Ensure the upstream emit uses KV, not a custom struct","Assert typex.IsKV(col.Type()) in tests"],"exampleFix":"// before\nbatched := batch.GroupIntoBatches(s, params, words)\n// after\nkvs := beam.ParDo(s, func(w string) (string, int) { return w, 1 }, words)\nbatched := batch.GroupIntoBatches(s, params, kvs)","handlingStrategy":"type-guard","validationCode":"if !typex.IsKV(col.Type()) {\n    return fmt.Errorf(\"need KV input, got %v\", col.Type())\n}","typeGuard":"func isKV(col beam.PCollection) bool { return typex.IsKV(col.Type()) }","tryCatchPattern":null,"preventionTips":["Map to KV pairs before grouping","Assert col.Type() in pipeline tests","Keep upstream transform output types documented"],"tags":["go","apache-beam","type-mismatch","pipeline-construction"],"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-14T16:17:12.679Z"}