apache/beam · error

unexpected non-KV coder PCollection input to combine

Error message

unexpected non-KV coder PCollection input to combine: %v

What it means

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.

Solutions

1. Ensure all custom coders used by the input PCollection are registered identically on the worker side
2. Verify the input coder payload in the job proto decodes to a KV coder; fix coder inference if needed
3. Check that beam.CombinePerKey (not a mis-typed Combine over non-KV elements) is used in user code

Example fix

// before
coll := beam.Create(s, "a", "b")
beam.CombinePerKey(s, combineFn, coll)
// after
kv := beam.ParDo(s, emitKVFn, coll)
beam.CombinePerKey(s, combineFn, kv)

ec, wc, err := b.makeCoderForPCollection(inputs[0])
if err != nil {
	return err
}
if !coder.IsKV(ec) {
	return fmt.Errorf("combine input PCollection must be KV-coded, got %v", ec)
}

null

null

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• failed to update PTransform spec
• array len mismatch. decoding
• coder must not be nil
• coder type must be identical to node type
• combine does not support side inputs

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/404a6a258b87eb15. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/translate.go:678

				if err != nil {
					return nil, err
				}
				cn.UsesKey = typex.IsKV(in[0].Type)

				cn.PID = id.to

				switch urn {
				case urnPerKeyCombinePre:
					inputs := unmarshalKeyedValues(transform.GetInputs())
					if len(inputs) != 1 {
						return nil, errors.Errorf("unexpected sideinput to combine: got %d, want 1", len(inputs))
					}
					ec, wc, err := b.makeCoderForPCollection(inputs[0])
					if err != nil {
						return nil, err
					}
					if !coder.IsKV(ec) {
						return nil, errors.Errorf("unexpected non-KV coder PCollection input to combine: %v", ec)
					}
					u = &LiftedCombine{Combine: cn, KeyCoder: ec.Components[0], WindowCoder: wc}
				case urnPerKeyCombineMerge:
					ma := &MergeAccumulators{Combine: cn}
					if pc, ok := ma.Out.(*PCollection); ok {
						if eo, ok := pc.Out.(*ExtractOutput); ok {
							// Strip PCollections from between MergeAccumulators and ExtractOutputs
							// as it's a synthetic PCollection.
							b.units = b.units[:len(b.units)-1]
							ma.Out = eo
						}
					}
					u = ma
				case urnPerKeyCombineExtract:
					u = &ExtractOutput{Combine: cn}
				case urnPerKeyCombineConvert:
					u = &ConvertToAccumulators{Combine: cn}
				default: // For unlifted combines

View on GitHub (pinned to 12126d8942)