{"record":{"id":"78f6215ffdf2d8e1","repo":"apache/beam","slug":"stage-requires-multiple-parallel-inputs-but-wasn-t-a-flatten","errorCode":null,"errorMessage":"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","messagePattern":"stage requires multiple parallel inputs but wasn't a flatten:\n\ttransforms\n\t(.+?)\n\tmain inputs\n\t(.+?)\n\tsidinputs\n\t(.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/preprocess.go","lineNumber":532,"sourceCode":"\n\t// Impulses won't have any inputs.\n\tif l := len(mainInputs); l == 1 {\n\t\tstg.primaryInput = getOnlyValue(mainInputs)\n\t} else if l > 1 {\n\t\t// Quick check that this is lead by a flatten node, and that it's handled runner side.\n\t\tt := comps.GetTransforms()[stg.transforms[0]]\n\t\tif !(t.GetSpec().GetUrn() == urns.TransformFlatten && t.GetEnvironmentId() == \"\") {\n\t\t\tformatMap := func(in map[string]string) string {\n\t\t\t\tvar b strings.Builder\n\t\t\t\tfor k, v := range in {\n\t\t\t\t\tb.WriteString(k)\n\t\t\t\t\tb.WriteString(\" : \")\n\t\t\t\t\tb.WriteString(v)\n\t\t\t\t\tb.WriteString(\"\\n\\t\")\n\t\t\t\t}\n\t\t\t\treturn b.String()\n\t\t\t}\n\t\t\treturn 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)\n\t\t}\n\t}\n\treturn nil\n}\n\n// greedyFusion produces a pipeline as tightly fused as possible.\n//\n// Fusion is a critical optimization for performance of pipeline execution.\n// Thus it's important for SDKs to be capable of executing transforms in a fused state.\n//\n// However, not all transforms can be fused into the same stage together.\n// Further, some transforms must be at the root of a stage.\n//\n// # Fusion Restrictions\n//\n// Environments: Transforms that aren't in the same environment can't be\n// fused together *unless* their environments can also be fused together.\n// Eg. Resource hints can often be ignored for local runners.","sourceCodeStart":514,"sourceCodeEnd":550,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/preprocess.go#L514-L550","documentation":"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.","triggerScenarios":"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.","commonSituations":"Custom or third-party transform authors building pipeline components manually; pipelines post-processed by other optimizers before reaching prism; bugs in upstream graph rewriting.","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"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"if len(mainInputs) > 1 && !isFlatten(stg) {\n    return fmt.Errorf(\"stage %v has multiple main inputs without Flatten\", stg.ID)\n}","typeGuard":null,"tryCatchPattern":"if err := finalizeStage(stg, comps); err != nil {\n    if strings.Contains(err.Error(), \"wasn't a flatten\") {\n        // dump stg.transforms and inputs for graph debugging\n    }\n    return err\n}","preventionTips":["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"],"tags":["go","beam","pipeline-graph","fusion"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}