apache/beam · error

input type must be KV: %v

Error message

input type must be KV: %v

What it means

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.

Source

Thrown at sdks/go/pkg/beam/core/graph/edge.go:209

func (e *MultiEdge) String() string {
	return fmt.Sprintf("%v: %v %v -> %v", e.id, e.Op, e.Input, e.Output)
}

// NOTE(herohde) 4/28/2017: In general, we have no good coder guess for outgoing
// nodes, unless we add a notion of default coder for arbitrary types. We leave
// that to the beam layer.

// NewCoGBK inserts a new CoGBK edge into the graph.
func NewCoGBK(g *Graph, s *Scope, ns []*Node) (*MultiEdge, error) {
	addContext := func(err error, s *Scope) error {
		return errors.WithContextf(err, "creating new CoGBK in scope %v", s)
	}

	if len(ns) == 0 {
		return nil, addContext(errors.New("needs at least 1 input"), s)
	}
	if !typex.IsKV(ns[0].Type()) {
		return nil, addContext(errors.Errorf("input type must be KV: %v", ns[0]), s)
	}

	// (1) Create CoGBK result type: KV<T,U>, .., KV<T,Z> -> CoGBK<T,U,..,Z>.

	c := ns[0].Coder.Components[0]
	w := inputWindow(ns)
	bounded := inputBounded(ns)
	comp := []typex.FullType{c.T, ns[0].Type().Components()[1]}

	for i := 1; i < len(ns); i++ {
		n := ns[i]
		if !typex.IsKV(n.Type()) {
			return nil, addContext(errors.Errorf("input type must be KV: %v", n), s)
		}
		if !n.Coder.Components[0].Equals(c) {
			return nil, addContext(errors.Errorf("key coder for %v is %v, want %v", n, n.Coder.Components[0], c), s)
		}
		if !w.Equals(n.WindowingStrategy()) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Wrap inputs in beam.KV via beam.ParDo returning KV values before CoGroupByKey
2. Use beam.KV helpers to make the key/value structure explicit
3. Check upstream transform output types with typex.IsKV in tests

Example fix

// before
out := beam.ParDo(s, func(w Word) string { return w.Text }, words)
beam.CoGroupByKey(s, counts) // counts not KV
// after
keyed := beam.ParDo(s, func(w Word) (string, Word) { return w.Key, w }, words)
beam.CoGroupByKey(s, keyed)

if !typex.IsKV(col.Type()) {
    return fmt.Errorf("input must be KV, got %v", col.Type())
}

func isKVInput(n *beam.Node) bool { return n != nil && typex.IsKV(n.Type()) }

if err := beam.TryCoGroupByKey(s, in); err != nil {
    return fmt.Errorf("CoGroupByKey needs KV inputs: %w", err)
}

When it happens

Trigger: Calling beam.CoGroupByKey (TryCoGroupByKey → NewCoGBK) with a first input of non-KV type, e.g. a plain PCollection<string> or a PCollection of structs.

Common situations: Forgetting beam.KV when building the input to a CoGroupByKey, or changing upstream transforms so the element type is no longer KV.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• key coder for %v is %v, want %v
• mismatched CoGBK windowing strategies: %v, want %v
• failed to decode union value '%v' for key %v
• failed to marshal CoGBK coder %v
• bad CoGBK: %v

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