apache/beam · error

Flatten input type cannot be CoGBK

Error message

Flatten input type cannot be CoGBK: %v

What it means

After the type/window checks, NewFlatten rejects any input whose type is a CoGBK type (`typex.IsCoGBK`). CoGBK represents an unexploded multi-input grouping result, which Flatten cannot merge as a regular element stream. The graph builder throws this to prevent flattening grouped/deferred multi-input values.

Solutions

1. Expand the CoGBK result first: use beam.ParDo with a func over the CoGBK KV to emit per-key records, then flatten.
2. Merge the pre-CoGBK inputs (flatten before grouping) if the intent is to group them together.
3. Use beam.Explode (or a custom expansion DoFn) on each CoGBK collection prior to Flatten.
4. Restructure so Flatten operates on plain typed PCollections only.

Example fix

// before
merged := beam.Flatten(s, cogbk1, cogbk2) // CoGBK inputs rejected
// after
expand := func(k beam.KV, it func(*string) bool) string { /* emit value */ return "" }
flat1 := beam.ParDo(s, expand, cogbk1)
flat2 := beam.ParDo(s, expand, cogbk2)
merged := beam.Flatten(s, flat1, flat2)

// Go: expand CoGBK outputs before Flatten
expand := func(kv beam.KV, iter func(*T) bool) T { var v T; iter(&v); return v }
flattenable := beam.ParDo(s, expand, cogbkPC)

When it happens

Trigger: Passing the output of beam.CoGroupByKey (or a PCollection typed as CoGBK) directly into beam.Flatten.

Common situations: Trying to merge results of multiple CoGroupByKey operations; misunderstanding CoGBK outputs as ordinary PCollections and passing them to combinators that only accept main inputs.

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

• Reshuffle input type cannot be CoGBK
• Combine cannot follow multi-input CoGBK
• Combine requires CoGBK type
• Flatten needs at least 2 input, got
• mismatched Flatten input types

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

Appendix: source

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

	// TODO(herohde) 4/5/2018: is it fine mixing boundedness for flatten?
	// The output would be unbounded iff any input is.
	bounded := true
	for _, n := range in {
		if !n.Bounded() {
			bounded = false
			break
		}
	}
	for _, n := range in {
		if !typex.IsEqual(t, n.Type()) {
			return nil, addContext(errors.Errorf("mismatched Flatten input types: %v, want %v", n.Type(), t), s)
		}
		if !w.Equals(n.WindowingStrategy()) {
			return nil, addContext(errors.Errorf("mismatched Flatten window types: %v, want %v", n.WindowingStrategy(), w), s)
		}
	}
	if typex.IsCoGBK(t) {
		return nil, addContext(errors.Errorf("Flatten input type cannot be CoGBK: %v", t), s)
	}

	edge := g.NewEdge(s)
	edge.Op = Flatten
	for _, n := range in {
		edge.Input = append(edge.Input, &Inbound{Kind: Main, From: n, Type: t})
	}
	edge.Output = []*Outbound{{To: g.NewNode(t, w, bounded), Type: t}}
	return edge, nil
}

// NewCrossLanguage inserts a Cross-langugae External transform using initialized input and output nodes
func NewCrossLanguage(g *Graph, s *Scope, ext *ExternalTransform, ins []*Inbound, outs []*Outbound) (*MultiEdge, func(*Node, bool)) {
	edge := g.NewEdge(s)
	edge.Op = External
	edge.External = ext

	ws := window.DefaultWindowingStrategy()

View on GitHub (pinned to 12126d8942)