apache/beam · error

Combine requires CoGBK type

Error message

Combine requires CoGBK type: %v

What it means

NewCombine requires the incoming edge's type to be a CoGBK type, since CombinePerKey operates on grouped key/value collections. If `in.Type()` is not CoGBK (e.g., a plain PCollection), graph construction fails with this error. It enforces that Combine is always preceded by a grouping operation.

Solutions

1. Use beam.CombinePerKey (the public API), which performs the required CoGBK internally.
2. If building graphs manually, insert a GroupByKey/CoGBK node before the Combine edge.
3. If the intent is element-wise aggregation without keys, use beam.Combine instead of beam.CombinePerKey.
4. Verify the input PCollection actually comes from a grouping transform before getCombineEdge runs.

Example fix

// before
combined := graph.NewCombine(...) // input is ungrouped PCollection<string>
// after
// at pipeline level:
combined := beam.CombinePerKey(s, combineFn, kvPC) // kvPC is PCollection<KV<K,V>>; CombinePerKey creates the CoGBK

// Go: use the public API so the CoGBK is created for you
// combined := beam.CombinePerKey(s, combineFn, kvPC)  // kvPC: PCollection<KV<K,V>>

When it happens

Trigger: Calling beam.CombinePerKey on a PCollection that was not grouped (no GroupByKey/CoGBK), or applying a combineFn via NewCombine to a raw element stream.

Common situations: Manually building a graph and skipping the GroupByKey before Combine; calling low-level graph APIs instead of beam.Combine; a refactor removed the preceding GroupByKey.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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

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

Appendix: source

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

// CombinePerKeyScope is the Go SDK canonical name for the combine composite
// scope. With Beam Portability, "primitive" composite transforms like
// combine have their URNs & payloads attached to a high level scope, with a
// default representation beneath. The use of this const permits the
// translation layer to confirm the SDK expects this combine to be liftable
// by a runner and should set this scope's URN and Payload accordingly.
const CombinePerKeyScope = "CombinePerKey"

// NewCombine inserts a new Combine edge into the graph. Combines cannot have side
// input.
func NewCombine(g *Graph, s *Scope, u *CombineFn, in *Node, ac *coder.Coder, typedefs map[string]reflect.Type) (*MultiEdge, error) {
	addContext := func(err error, s *Scope) error {
		return errors.WithContextf(err, "creating new Combine in scope %v", s)
	}

	inT := in.Type()
	if !typex.IsCoGBK(inT) {
		return nil, addContext(errors.Errorf("Combine requires CoGBK type: %v", inT), s)
	}
	if len(inT.Components()) > 2 {
		return nil, addContext(errors.Errorf("Combine cannot follow multi-input CoGBK: %v", inT), s)
	}

	// Create a synthetic function for binding purposes. It takes main input
	// and returns the output type -- but hides the accumulator.
	//
	//  (1) If AddInput exists, then it lists the main inputs. If not,
	//      the only main input type is the accumulator type.
	//  (2)	If ExtractOutput exists then it returns the output type. If not,
	//      then the accumulator is the output type.
	//
	// MergeAccumulators is guaranteed to exist. We do not allow the accumulator
	// to be a tuple type (i.e., so one can't define a inline KV merge function).

	synth := &funcx.Fn{}
	if f := u.AddInputFn(); f != nil {

View on GitHub (pinned to 12126d8942)