computeFacts: two producers for one PCollection

Error message

computeFacts: two producers for one PCollection: %v and %v

What it means

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'.

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.

Example fix

// before: two transforms writing the same pc
beam.ParDo0(s, fnA, input) -> pc
beam.ParDo0(s, fnB, input) -> pc // illegal alias

// after: distinct outputs
pcA := beam.ParDo(s, fnA, input)
pcB := beam.ParDo(s, fnB, input)

Defensive patterns

Strategy: validation

Validate before calling

// construction-time check: never reuse one PCol for multiple transform outputs
outputs := map[string]bool{}
for _, out := range transformOutputs {
    if outputs[out.GetGlobal()] {
        return errors.New("duplicate output PCollection: " + out.GetGlobal())
    }
    outputs[out.GetGlobal()] = true
}

Try / catch

err := submitToPrism(pipeline)
if err != nil && strings.Contains(err.Error(), "two producers for one PCollection") {
    // inspect both producer links in the message and fix the graph
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: Malformed or hand-built pipeline protos with duplicated output wiring; graph-corrupting transforms or tests; SDK bugs producing shared output PCollections after composite expansion.

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/8b0f02fc577bd8f7. Report an issue: GitHub.

Appendix: source

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

// computeFacts computes facts about the given set of transforms and components that
// are useful for fusion.
func computeFacts(topological []string, comps *pipepb.Components) (*fusionFacts, error) {
	ret := &fusionFacts{
		PcolProducers:        map[string]link{},
		PcolConsumers:        map[string][]link{},
		UsedAsSideInput:      map[string]bool{},
		DirectSideInputs:     map[string]map[string]bool{}, // direct set
		DownstreamSideInputs: map[string]map[string]bool{}, // transitive set
	}

	// Use the topological ids so each PCollection only has a single
	// producer. We've already pruned out composites at this stage.
	for _, tID := range topological {
		t := comps.GetTransforms()[tID]
		for local, global := range t.GetOutputs() {
			if p, ok := ret.PcolProducers[global]; ok {
				return nil, fmt.Errorf("computeFacts: two producers for one PCollection: %v and %v", p, link{Transform: tID, Local: local, Global: global})
			}
			ret.PcolProducers[global] = link{Transform: tID, Local: local, Global: global}
		}
		sis, err := getSideInputs(t)
		if err != nil {
			return nil, fmt.Errorf("computeFacts: unable to check %q side inputs", tID)
		}
		directSIs := map[string]bool{}
		ret.DirectSideInputs[tID] = directSIs
		for local, global := range t.GetInputs() {
			ret.PcolConsumers[global] = append(ret.PcolConsumers[global], link{Transform: tID, Local: local, Global: global})
			if _, ok := sis[local]; ok {
				ret.UsedAsSideInput[global] = true
				directSIs[global] = true
			}
		}
	}

View on GitHub (pinned to 12126d8942)