apache/beam · error

malformed transform;

Error message

malformed transform; %v lacks a conforming pipeline ptransform

What it means

ExpandedTransform type-asserts an ExpandedTransform's Transform field to *pipepb.PTransform. This error is returned when the stored value is nil or another type, so callers (expandCrossLanguage, mergeExpandedWithPipeline, VerifyNamedOutputs, etc.) cannot read the expanded transform proto.

Solutions

1. Populate ExpandedTransform.Transform with a *pipepb.PTransform from the expansion response.
2. Check that the expansion call succeeded (error handled) before accessing the expanded transform.
3. Fix test fixtures to use the correct proto type.

Example fix

// before
exp := &graph.ExpandedTransform{Components: &pipepb.Components{}}
// after
exp := &graph.ExpandedTransform{Transform: &pipepb.PTransform{}, Components: &pipepb.Components{}}

func hasValidTransform(exp *graph.ExpandedTransform) bool {
    _, ok := exp.Transform.(*pipepb.PTransform)
    return ok
}

if t, err := graphx.ExpandedTransform(exp); err != nil {
    return fmt.Errorf("expansion lacks transform: %w", err)
}

When it happens

Trigger: Calling ExpandedTransform on a graph.ExpandedTransform whose Transform is nil or not a *pipepb.PTransform.

Common situations: Manually built or fake ExpandedTransform objects in tests; expansion failures that leave Transform unset; custom code populating Transform with the wrong proto type.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

• Parsing and encoding errors: unexpected token, malformed input — why parsers reject input and how to find the real culprit.

Related errors

• malformed components;
• attempted to add namespace to missing coder id
• attempted to add namespace to missing windowing strategy id
• failed to add output coder to coder registry
• failed to expand cross language transform for edge

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/graphx/xlang.go:250

	}
}

// ExpandedComponents type asserts the Components field with any type
// and returns its pipeline component proto representation
func ExpandedComponents(exp *graph.ExpandedTransform) (*pipepb.Components, error) {
	if c, ok := exp.Components.(*pipepb.Components); ok {
		return c, nil
	}
	return nil, errors.Errorf("malformed components; %v lacks a conforming pipeline component", exp)
}

// ExpandedTransform type asserts the Transform field with any type
// and returns its pipeline ptransform proto representation
func ExpandedTransform(exp *graph.ExpandedTransform) (*pipepb.PTransform, error) {
	if t, ok := exp.Transform.(*pipepb.PTransform); ok {
		return t, nil
	}
	return nil, errors.Errorf("malformed transform; %v lacks a conforming pipeline ptransform", exp)
}

// ExternalInputs returns the map (tag -> graph node representing the
// pcollection) of input nodes with respect to the map (tag -> index of Inbound
// in MultiEdge.Input) of named inputs
func ExternalInputs(e *graph.MultiEdge) map[string]*graph.Node {
	return InboundTagToNode(e.External.InputsMap, e.Input)
}

// InboundTagToNode relates the tags from inbound links to their respective nodes.
func InboundTagToNode(inputsMap map[string]int, inbound []*graph.Inbound) map[string]*graph.Node {
	inputs := make(map[string]*graph.Node)
	for tag, id := range inputsMap {
		inputs[tag] = inbound[id].From
	}
	return inputs
}

View on GitHub (pinned to 12126d8942)