{"record":{"id":"404a6a258b87eb15","repo":"apache/beam","slug":"unexpected-non-kv-coder-pcollection-input-to-combine-v","errorCode":null,"errorMessage":"unexpected non-KV coder PCollection input to combine: %v","messagePattern":"unexpected non-KV coder PCollection input to combine: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/translate.go","lineNumber":678,"sourceCode":"\t\t\t\tif err != nil {\n\t\t\t\t\treturn nil, err\n\t\t\t\t}\n\t\t\t\tcn.UsesKey = typex.IsKV(in[0].Type)\n\n\t\t\t\tcn.PID = id.to\n\n\t\t\t\tswitch urn {\n\t\t\t\tcase urnPerKeyCombinePre:\n\t\t\t\t\tinputs := unmarshalKeyedValues(transform.GetInputs())\n\t\t\t\t\tif len(inputs) != 1 {\n\t\t\t\t\t\treturn nil, errors.Errorf(\"unexpected sideinput to combine: got %d, want 1\", len(inputs))\n\t\t\t\t\t}\n\t\t\t\t\tec, wc, err := b.makeCoderForPCollection(inputs[0])\n\t\t\t\t\tif err != nil {\n\t\t\t\t\t\treturn nil, err\n\t\t\t\t\t}\n\t\t\t\t\tif !coder.IsKV(ec) {\n\t\t\t\t\t\treturn nil, errors.Errorf(\"unexpected non-KV coder PCollection input to combine: %v\", ec)\n\t\t\t\t\t}\n\t\t\t\t\tu = &LiftedCombine{Combine: cn, KeyCoder: ec.Components[0], WindowCoder: wc}\n\t\t\t\tcase urnPerKeyCombineMerge:\n\t\t\t\t\tma := &MergeAccumulators{Combine: cn}\n\t\t\t\t\tif pc, ok := ma.Out.(*PCollection); ok {\n\t\t\t\t\t\tif eo, ok := pc.Out.(*ExtractOutput); ok {\n\t\t\t\t\t\t\t// Strip PCollections from between MergeAccumulators and ExtractOutputs\n\t\t\t\t\t\t\t// as it's a synthetic PCollection.\n\t\t\t\t\t\t\tb.units = b.units[:len(b.units)-1]\n\t\t\t\t\t\t\tma.Out = eo\n\t\t\t\t\t\t}\n\t\t\t\t\t}\n\t\t\t\t\tu = ma\n\t\t\t\tcase urnPerKeyCombineExtract:\n\t\t\t\t\tu = &ExtractOutput{Combine: cn}\n\t\t\t\tcase urnPerKeyCombineConvert:\n\t\t\t\t\tu = &ConvertToAccumulators{Combine: cn}\n\t\t\t\tdefault: // For unlifted combines","sourceCodeStart":660,"sourceCodeEnd":696,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/translate.go#L660-L696","documentation":"A lifted per-key combine (urnPerKeyCombinePre) must operate on KV-encoded input so the key coder can be extracted for the LiftedCombine. If makeCoderForPCollection returns a non-KV element coder, translation fails with this error.","triggerScenarios":"A Combine PerKey whose input PCollection coder is not KV — e.g. the coder registry failed to reconstruct the KV coder, coder payloads were stripped, or a custom runner attached the wrong coder ID to the combine input.","commonSituations":"Custom coders registered on one side but not the other, making the coder fall back to a non-KV representation; version mismatches altering coder inference; manually edited pipeline protos.","solutions":["Ensure all custom coders used by the input PCollection are registered identically on the worker side","Verify the input coder payload in the job proto decodes to a KV coder; fix coder inference if needed","Check that beam.CombinePerKey (not a mis-typed Combine over non-KV elements) is used in user code"],"exampleFix":"// before\ncoll := beam.Create(s, \"a\", \"b\")\nbeam.CombinePerKey(s, combineFn, coll)\n// after\nkv := beam.ParDo(s, emitKVFn, coll)\nbeam.CombinePerKey(s, combineFn, kv)","handlingStrategy":"validation","validationCode":"ec, wc, err := b.makeCoderForPCollection(inputs[0])\nif err != nil {\n\treturn err\n}\nif !coder.IsKV(ec) {\n\treturn fmt.Errorf(\"combine input PCollection must be KV-coded, got %v\", ec)\n}","typeGuard":"null","tryCatchPattern":"null","preventionTips":["Register custom coders identically on both pipeline builder and worker","Use beam.CombinePerKey only on KV PCollections","Verify coder payloads in the job proto before submission"],"tags":["go","apache-beam","coder","combine"],"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"}