apache/beam · error

expected CoGBK, got

Error message

expected CoGBK, got %v

What it means

Produced by MakeKVUnionCoder in graphx/cogbk.go, which builds the reshuffle union coder KV<K,KV<int,[]byte>> for a CoGBK node. If the supplied MultiEdge's Op is not graph.CoGBK, the function refuses to build the coder, since the union layout only makes sense for CoGBK outputs.

Solutions

1. Ensure the MultiEdge passed in was produced by a CoGBK (gbk.Op == graph.CoGBK) before calling.
2. Check that you are passing gbk *graph.MultiEdge from the correct expansion step (expandCoGBK), not another transform's edge.
3. Log the MultiEdge (the %v in the message shows the wrong op) and trace where it came from.
4. If you need a union coder for another op, use or write the appropriate builder rather than reusing MakeKVUnionCoder.

Example fix

// before
c, err := graphx.MakeKVUnionCoder(parDoEdge)
// after
if parDoEdge.Op == graph.CoGBK {
    c, err = graphx.MakeKVUnionCoder(parDoEdge)
}

if gbk == nil || gbk.Op != graph.CoGBK {
    return errors.New("MakeKVUnionCoder requires a MultiEdge with Op == graph.CoGBK")
}

func isCoGBKEdge(e *graph.MultiEdge) bool {
    return e != nil && e.Op == graph.CoGBK
}

c, err := graphx.MakeKVUnionCoder(edge)
if err != nil {
    return fmt.Errorf("kv union coder for %v: %w", edge.Op, err)
}

When it happens

Trigger: Calling MakeKVUnionCoder(gbk) with a *graph.MultiEdge whose Op is, e.g., ParDo, Flatten, or CoGBK-replacement operations — typically from a custom reshuffle/expansion hook passing the wrong edge.

Common situations: Custom reshuffle implementations or internal transforms that hand a non-GBK MultiEdge to the union-coder builder; refactored pipeline code where the operation type changed; unit tests exercising the bad-input path.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• Expected single input PCollection in reshuffle:
• Expected single output PCollection in reshuffle:
• failed to expand CoGBK transform for edge
• failed to expand Reshuffle transform for edge
• failed to make GBK Union coder.

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/cogbk.go:74

//                        |
//                 U4: CoGBK<K,RUV>
//                        |
//                      Expand
//                        |
//                Out: CoGBK<K,X,Y>
//
// Inject and Expand are system-defined functions. This expansion cannot be
// expressed as Go user code.

const (
	URNInject = "beam:go:transform:inject:v1"
	URNExpand = "beam:go:transform:expand:v1"
)

// MakeKVUnionCoder returns KV<K,KV<int,[]byte>> for a given CoGBK.
func MakeKVUnionCoder(gbk *graph.MultiEdge) (*coder.Coder, error) {
	if gbk.Op != graph.CoGBK {
		err := errors.Errorf("expected CoGBK, got %v", gbk)
		return nil, errors.WithContext(err, "failed to make KV Union coder")
	}

	from := gbk.Input[0].From
	key := from.Coder.Components[0]
	kvCoder, err := makeUnionCoder()
	if err != nil {
		return nil, errors.Wrapf(err, "failed to make KV Union coder.")
	}
	return coder.NewKV([]*coder.Coder{key, kvCoder}), nil
}

// MakeGBKUnionCoder returns CoGBK<K,KV<int,[]byte>> for a given CoGBK.
func MakeGBKUnionCoder(gbk *graph.MultiEdge) (*coder.Coder, error) {
	if gbk.Op != graph.CoGBK {
		err := errors.Errorf("expected CoGBK, got %v", gbk)
		return nil, errors.WithContext(err, "failed to make GBK Union coder")
	}

View on GitHub (pinned to 12126d8942)