apache/beam · error

empty pipeline

Error message

empty pipeline

What it means

Normalize validates that a pipeline proto has at least one transform before recomputing derivative information (roots, composite inputs/outputs, topological ordering). An empty Components.Transforms map cannot be meaningfully normalized, so it is rejected.

Source

Thrown at sdks/go/pkg/beam/core/runtime/pipelinex/replace.go:58

	reflectx.UpdateMap(ret.Components.Pcollections, values.Pcollections)
	reflectx.UpdateMap(ret.Components.WindowingStrategies, values.WindowingStrategies)
	reflectx.UpdateMap(ret.Components.Coders, values.Coders)
	reflectx.UpdateMap(ret.Components.Environments, values.Environments)
	return Normalize(ret)
}

// IdempotentNormalize determines whether to use the idempotent version
// of ensureUniqueNames or the legacy version.
// TODO(BEAM-12341): Cleanup once nothing depends on the legacy implementation.
var IdempotentNormalize bool = true

// Normalize recomputes derivative information in the pipeline, such
// as roots and input/output for composite transforms. It also
// ensures that unique names are so and topologically sorts each
// subtransform list.
func Normalize(p *pipepb.Pipeline) (*pipepb.Pipeline, error) {
	if len(p.GetComponents().GetTransforms()) == 0 {
		return nil, errors.New("empty pipeline")
	}

	ret := shallowClonePipeline(p)
	if IdempotentNormalize {
		ret.Components.Transforms = ensureUniqueNames(ret.Components.Transforms)
	} else {
		ret.Components.Transforms = ensureUniqueNamesLegacy(ret.Components.Transforms)
	}
	ret.Components.Transforms = computeCompositeInputOutput(ret.Components.Transforms)
	ret.RootTransformIds = computeRoots(ret.Components.Transforms)
	return ret, nil
}

// TrimCoders returns the transitive closure of the given coders ids.
func TrimCoders(coders map[string]*pipepb.Coder, ids ...string) map[string]*pipepb.Coder {
	ret := make(map[string]*pipepb.Coder)
	for _, id := range ids {
		walkCoders(coders, ret, id)

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure at least one transform is added to the pipeline before normalizing/marshaling
2. Verify the pipeline proto was constructed correctly if building it programmatically (use pipeline.New or the model pipelinex helpers)
3. Re-create the pipeline if it came from a corrupt or empty serialized source

Example fix

// before
p := &pipepb.Pipeline{Components: &pipepb.Components{}}
_, err := pipelinex.Normalize(p)
// after
if len(p.GetComponents().GetTransforms()) == 0 {
    return errors.New("pipeline has no transforms; construct it via beam.NewPipeline() before normalizing")
}
_, err := pipelinex.Normalize(p)

if p == nil || len(p.GetComponents().GetTransforms()) == 0 {
    return errors.New("cannot normalize: pipeline has no transforms")
}

out, err := pipelinex.Normalize(p)
if err != nil {
    if strings.Contains(err.Error(), "empty pipeline") { return fmt.Errorf("pipeline was never populated with transforms") }
    return err
}

When it happens

Trigger: Calling Normalize (directly or via Marshal, Update, or Expand) on a *pipepb.Pipeline whose components contain no transforms.

Common situations: Programmatic pipeline construction where transforms were never added before marshaling; deserializing a truncated or hand-crafted pipeline proto; cross-language payload construction mistakes.

Understand the failure class

Background: "must not be empty", "cannot be empty" — required-field validation errors across open-source libraries — this error's family across 41 libraries.

Related errors

• Params.MaxBufferingDuration must be >= 0; got %s
• OnTimer function is defined for the DoFn but no TimerProvide
• create has no values
• invalid scope
• invalid scope

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