apache/beam · error

Reshuffle input type cannot be CoGBK

Error message

Reshuffle input type cannot be CoGBK: %v

What it means

NewReshuffle rejects input whose type is CoGBK: a reshuffle (injecting a GBK round-trip to break fusion) only makes sense for ordinary keyed or plain element streams, not unexploded multi-input grouping results. The graph builder throws this when TryReshuffle is given a CoGBK-typed PCollection.

Solutions

1. Expand the CoGBK first with a ParDo emitting plain or KV values, then apply beam.Reshuffle.
2. Reshuffle the pre-cogroup inputs instead of the CoGBK output.
3. If the CoGBK came from a single-input group, use GroupByKey output (KV) rather than multi-input CoGBK.
4. Remove the Reshuffle if it is not required (Beam inserts reshuffles automatically on some runners).

Example fix

// before
beam.Reshuffle(s, cogbkPC) // CoGBK input rejected
// after
flat := beam.ParDo(s, func(k beam.KV, v iter.Interface) beam.KV { return beam.KV{k, v} }, cogbkPC)
reshuffled := beam.Reshuffle(s, flat)

// Go: expand CoGBK before reshuffling
flat := beam.ParDo(s, expandCoGBKFn, cogbkPC)
reshuffled := beam.Reshuffle(s, flat)

When it happens

Trigger: Calling beam.Reshuffle on the output of beam.CoGroupByKey (CoGBK type), typically to redistribute work after a cogroup.

Common situations: Inserting Reshuffle after joins/cogroups for load balancing; fusion-breaking passes added by hand that assume plain element types.

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

• Flatten input type cannot be CoGBK
• Combine cannot follow multi-input CoGBK
• Combine requires CoGBK type
• expected CoGBK, got
• Any not yet supported

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

Appendix: source

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

func inputBounded(in []*Node) bool {
	if len(in) == 0 {
		return true
	}
	return in[0].Bounded()
}

// NewReshuffle inserts a new Reshuffle edge into the graph.
func NewReshuffle(g *Graph, s *Scope, in *Node) (*MultiEdge, error) {
	addContext := func(err error, s *Scope) error {
		return errors.WithContextf(err, "creating new Reshuffle in scope %v", s)
	}
	n := g.NewNode(in.Type(), in.WindowingStrategy(), in.Bounded())
	n.Coder = in.Coder

	t := in.Type()
	if typex.IsCoGBK(t) {
		return nil, addContext(errors.Errorf("Reshuffle input type cannot be CoGBK: %v", t), s)
	}
	edge := g.NewEdge(s)
	edge.Op = Reshuffle
	edge.Input = []*Inbound{{Kind: Main, From: in, Type: t}}
	edge.Output = []*Outbound{{To: n, Type: t}}
	return edge, nil
}

View on GitHub (pinned to 12126d8942)