stage requires multiple parallel inputs but wasn't a flatten

Error message

stage requires multiple parallel inputs but wasn't a flatten:
	transforms
	%v
	main inputs
	%v
	sidinputs
	%v

What it means

prism only allows a stage to consume multiple parallel (main) inputs if it is a Flatten; otherwise the graph would need data joining that the fusion planner cannot express. finalizeStage detects a stage with several main inputs and no Flatten and fails with a diagnostic dump of transforms, main inputs, and side inputs.

Source

Thrown at sdks/go/pkg/beam/runners/prism/internal/preprocess.go:532

	// Impulses won't have any inputs.
	if l := len(mainInputs); l == 1 {
		stg.primaryInput = getOnlyValue(mainInputs)
	} else if l > 1 {
		// Quick check that this is lead by a flatten node, and that it's handled runner side.
		t := comps.GetTransforms()[stg.transforms[0]]
		if !(t.GetSpec().GetUrn() == urns.TransformFlatten && t.GetEnvironmentId() == "") {
			formatMap := func(in map[string]string) string {
				var b strings.Builder
				for k, v := range in {
					b.WriteString(k)
					b.WriteString(" : ")
					b.WriteString(v)
					b.WriteString("\n\t")
				}
				return b.String()
			}
			return fmt.Errorf("stage requires multiple parallel inputs but wasn't a flatten:\n\ttransforms\n\t%v\n\tmain inputs\n\t%v\n\tsidinputs\n\t%v", strings.Join(stg.transforms, "\n\t\t"), formatMap(mainInputs), sideInputs)
		}
	}
	return nil
}

// greedyFusion produces a pipeline as tightly fused as possible.
//
// Fusion is a critical optimization for performance of pipeline execution.
// Thus it's important for SDKs to be capable of executing transforms in a fused state.
//
// However, not all transforms can be fused into the same stage together.
// Further, some transforms must be at the root of a stage.
//
// # Fusion Restrictions
//
// Environments: Transforms that aren't in the same environment can't be
// fused together *unless* their environments can also be fused together.
// Eg. Resource hints can often be ignored for local runners.

View on GitHub (pinned to 12126d8942)

Solutions

Inspect the dumped transforms/main inputs to find which transform received multiple main inputs
Replace manual multi-input wiring with a legitimate Flatten or CoGroupByKey
Check whether any custom graph rewriting is duplicating input edges into one transform
Verify with a standard SDK-constructed pipeline to isolate custom graph manipulation

Defensive patterns

Strategy: validation

Validate before calling

if len(mainInputs) > 1 && !isFlatten(stg) {
    return fmt.Errorf("stage %v has multiple main inputs without Flatten", stg.ID)
}

Try / catch

if err := finalizeStage(stg, comps); err != nil {
    if strings.Contains(err.Error(), "wasn't a flatten") {
        // dump stg.transforms and inputs for graph debugging
    }
    return err
}

Prevention

Use Flatten/CoGroupByKey instead of manual multi-input wiring
Test any custom graph rewriting against prism
Inspect the error dump of transforms/inputs to find the offending edge

When it happens

Trigger: A pipeline graph where a non-Flatten transform ends up with 2+ PCollections as main inputs — typically from a malformed or hand-assembled pipeline, or a graph transformation bug that fused incompatible transforms into one stage.

Common situations: Custom or third-party transform authors building pipeline components manually; pipelines post-processed by other optimizers before reaching prism; bugs in upstream graph rewriting.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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