apache/beam · error

Combine cannot follow multi-input CoGBK

Error message

Combine cannot follow multi-input CoGBK: %v

What it means

After confirming the input is CoGBK, NewCombine checks that it has at most 2 components (key + one value). Combines cannot follow multi-input CoGBK (results of multiple-input CoGroupByKey), so a CoGBK with more than 2 components aborts graph construction. Combining is only defined for grouped K/V pairs, not wider grouped tuples.

Solutions

1. Expand the multi-input CoGBK with a ParDo that extracts the single value component you want, re-group if needed, then CombinePerKey.
2. Combine each input PCollection before the CoGroupByKey, then merge the aggregated results.
3. Use a ParDo over the multi-input CoGBK to implement the per-key aggregation manually.
4. If only two inputs are truly needed, replace the multi-input CoGBK with a plain GroupByKey on a KV PCollection.

Example fix

// before
combined := beam.CombinePerKey(s, sumFn, multiCogbk) // CoGBK has 3 components
// after
extract := func(k beam.KV, a, b iter.Interface) beam.KV { /* merge side values */ return k }
kv := beam.ParDo(s, extract, multiCogbk)
combined := beam.CombinePerKey(s, sumFn, kv)

// Go: aggregate before cogrouping instead of combining a multi-input CoGBK
summedA := beam.CombinePerKey(s, sumFn, a)
summedB := beam.CombinePerKey(s, sumFn, b)
merged := beam.CoGroupByKey(s, summedA, summedB)

When it happens

Trigger: Applying beam.CombinePerKey to the output of a multi-input CoGroupByKey (CoGBK with 3+ components), e.g. cogrouping 3+ PCollections and then trying to combine per key.

Common situations: Co-grouping several streams and then attempting CombinePerKey on the merged grouped result; chaining Combine after a join-like CoGBK instead of expanding it first.

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 requires CoGBK type
• Flatten input type cannot be CoGBK
• Flatten needs at least 2 input, got
• Reshuffle input type cannot be CoGBK
• bad CoGBK

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

Appendix: source

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

// 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 {
		// drop accumulator and irrelevant parameters
		synth.Param = funcx.SubParams(f.Param, f.Params(funcx.FnValue)[1:]...)
	} else {

View on GitHub (pinned to 12126d8942)