apache/beam · error

failed to update PTransform spec

Error message

failed to update PTransform spec: %v

What it means

Wrapped by updateIfCombineComposite when the accumulator coder for a Combine transform cannot be added to the marshaller's coder registry (m.coders.Add fails). The pipeline proto for the Combine PTransform cannot be built without registering the accumulator coder, so marshalling aborts with the transform name attached.

Solutions

1. Ensure the CombineFn accumulator type only contains coder-supported exported fields
2. Check the inner coder error from m.coders.Add to identify the offending type
3. Register a custom coder for the accumulator type (graphx.CustomCoder)
4. Replace the accumulator with an equivalent encodable type

Example fix

// before
type acc struct { mu sync.Mutex; sum int }
// after
type acc struct { Sum int64 }

// Verify accumulator type is derivable as a coder before building the pipeline
var acc MyAcc
if _, err := graphx.MakeCoderFromType(reflect.TypeOf(acc)); err != nil {
    return fmt.Errorf("MyAcc not codable: %w", err)
}

if _, err := graphx.Marshal(p); err != nil {
    if strings.Contains(err.Error(), "failed to update PTransform spec") {
        // fall back to logging pipeline and reporting coder failure
        log.Fatalf("combine coder failure: %v", err)
    }
    return err
}

When it happens

Trigger: Marshalling a pipeline containing beam.Combine whose edge.AccumCoder cannot be serialized into the coder registry — typically because the accumulator type has no derivable coder.

Common situations: CombineFns with accumulator types that are structs with unexported fields, channels, funcs, or other non-encodable Go types; use of beam.Combine with custom types lacking coder registration.

Related errors

• unexpected non-KV coder PCollection input to combine
• array len mismatch. decoding
• coder must not be nil
• coder type must be identical to node type
• combine does not support side inputs

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/translate.go:333

}

// updateIfCombineComposite examines the scope tree and sets the PTransform Spec
// to be a CombinePerKey with a CombinePayload if it's a liftable composite.
// Beam Portability requires that composites contain an implementation for runners
// that don't understand the URN and Payload, which this lightly checks for.
func (m *marshaller) updateIfCombineComposite(s *ScopeTree, transform *pipepb.PTransform) error {
	if s.Scope.Name != graph.CombinePerKeyScope ||
		len(s.Edges) != 2 ||
		len(s.Edges[0].Edge.Input) != 1 ||
		len(s.Edges[1].Edge.Output) != 1 ||
		s.Edges[1].Edge.Op != graph.Combine {
		return nil
	}

	edge := s.Edges[1].Edge
	acID, err := m.coders.Add(edge.AccumCoder)
	if err != nil {
		return errors.Wrapf(err, "failed to update PTransform spec: %v", transform)
	}
	mustEncodeMultiEdge, err := mustEncodeMultiEdgeBase64(edge)
	if err != nil {
		return errors.Wrapf(err, "failed to update PTransform spec: %v", transform)
	}
	payload := &pipepb.CombinePayload{
		CombineFn: &pipepb.FunctionSpec{
			Urn:     URNDoFn,
			Payload: []byte(mustEncodeMultiEdge),
		},
		AccumulatorCoderId: acID,
	}
	transform.Spec = &pipepb.FunctionSpec{Urn: URNCombinePerKey, Payload: protox.MustEncode(payload)}
	return nil
}

func getSideWindowMappingUrn(winFn *window.Fn) string {
	var mappingUrn string

View on GitHub (pinned to 12126d8942)