{"record":{"id":"8b0f02fc577bd8f7","repo":"apache/beam","slug":"computefacts-two-producers-for-one-pcollection-v-and-v","errorCode":null,"errorMessage":"computeFacts: two producers for one PCollection: %v and %v","messagePattern":"computeFacts: two producers for one PCollection: (.+?) and (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/preprocess.go","lineNumber":377,"sourceCode":"\n// computeFacts computes facts about the given set of transforms and components that\n// are useful for fusion.\nfunc computeFacts(topological []string, comps *pipepb.Components) (*fusionFacts, error) {\n\tret := &fusionFacts{\n\t\tPcolProducers:        map[string]link{},\n\t\tPcolConsumers:        map[string][]link{},\n\t\tUsedAsSideInput:      map[string]bool{},\n\t\tDirectSideInputs:     map[string]map[string]bool{}, // direct set\n\t\tDownstreamSideInputs: map[string]map[string]bool{}, // transitive set\n\t}\n\n\t// Use the topological ids so each PCollection only has a single\n\t// producer. We've already pruned out composites at this stage.\n\tfor _, tID := range topological {\n\t\tt := comps.GetTransforms()[tID]\n\t\tfor local, global := range t.GetOutputs() {\n\t\t\tif p, ok := ret.PcolProducers[global]; ok {\n\t\t\t\treturn nil, fmt.Errorf(\"computeFacts: two producers for one PCollection: %v and %v\", p, link{Transform: tID, Local: local, Global: global})\n\t\t\t}\n\t\t\tret.PcolProducers[global] = link{Transform: tID, Local: local, Global: global}\n\t\t}\n\t\tsis, err := getSideInputs(t)\n\t\tif err != nil {\n\t\t\treturn nil, fmt.Errorf(\"computeFacts: unable to check %q side inputs\", tID)\n\t\t}\n\t\tdirectSIs := map[string]bool{}\n\t\tret.DirectSideInputs[tID] = directSIs\n\t\tfor local, global := range t.GetInputs() {\n\t\t\tret.PcolConsumers[global] = append(ret.PcolConsumers[global], link{Transform: tID, Local: local, Global: global})\n\t\t\tif _, ok := sis[local]; ok {\n\t\t\t\tret.UsedAsSideInput[global] = true\n\t\t\t\tdirectSIs[global] = true\n\t\t\t}\n\t\t}\n\t}\n","sourceCodeStart":359,"sourceCodeEnd":395,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/preprocess.go#L359-L395","documentation":"computeFacts builds a map from each PCollection to its single producing transform. Encountering a second producer for the same PCollection breaks Prism's dataflow assumptions, so it fails with this error naming both producer links. This propagates up through preProcessGraph as 'error computing pipeline facts'.","triggerScenarios":"During computeFacts' pass over topological transform IDs, a transform output global ID already exists in ret.PcolProducers — i.e. two transforms both declare the same PCollection as an output.","commonSituations":"Malformed or hand-built pipeline protos with duplicated output wiring; graph-corrupting transforms or tests; SDK bugs producing shared output PCollections after composite expansion.","solutions":["Inspect both producer links in the message to find the transform IDs fighting over the PCollection.","Ensure each ParDo/transform declares unique output PCollections in the pipeline construction code.","Re-expand composites so outputs are not aliased across transforms.","If the pipeline comes from an SDK pipeline builder, validate the graph with DirectRunner and file a Beam issue."],"exampleFix":"// before: two transforms writing the same pc\nbeam.ParDo0(s, fnA, input) -> pc\nbeam.ParDo0(s, fnB, input) -> pc // illegal alias\n\n// after: distinct outputs\npcA := beam.ParDo(s, fnA, input)\npcB := beam.ParDo(s, fnB, input)","handlingStrategy":"validation","validationCode":"// construction-time check: never reuse one PCol for multiple transform outputs\noutputs := map[string]bool{}\nfor _, out := range transformOutputs {\n    if outputs[out.GetGlobal()] {\n        return errors.New(\"duplicate output PCollection: \" + out.GetGlobal())\n    }\n    outputs[out.GetGlobal()] = true\n}","typeGuard":null,"tryCatchPattern":"err := submitToPrism(pipeline)\nif err != nil && strings.Contains(err.Error(), \"two producers for one PCollection\") {\n    // inspect both producer links in the message and fix the graph\n}","preventionTips":["Give every transform its own distinct output PCollections.","Never alias a PCollection across multiple transforms in generated pipelines.","Run graph validation (DirectRunner) before submitting to prism."],"tags":["go","beam-prism","pipeline-graph","preprocessing"],"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-20T03:17:13.778Z"}