{"record":{"id":"9ae28f2b03500ca6","repo":"apache/beam","slug":"error-computing-pipeline-facts-w","errorCode":null,"errorMessage":"error computing pipeline facts: %w","messagePattern":"error computing pipeline facts: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/preprocess.go","lineNumber":163,"sourceCode":"\t\t// It's unlikely for these to change, but better to handle them now, to save a headache later.\n\t\tfor wid, w := range prepResult.SubbedComps.GetWindowingStrategies() {\n\t\t\tcomps.GetWindowingStrategies()[wid] = w\n\t\t}\n\t\tfor envid, env := range prepResult.SubbedComps.GetEnvironments() {\n\t\t\tcomps.GetEnvironments()[envid] = env\n\t\t}\n\t}\n\n\t// Extract URNs for the given transform.\n\n\tkeptLeaves := maps.Keys(leaves)\n\tsort.Strings(keptLeaves)\n\ttopological := pipelinex.TopologicalSort(ts, keptLeaves)\n\tslog.Debug(\"topological transform ordering\", slog.Any(\"topological\", topological))\n\n\tfacts, err := computeFacts(topological, comps)\n\tif err != nil {\n\t\terr = fmt.Errorf(\"error computing pipeline facts: %w\", err)\n\t\tj.SendMsg(err.Error())\n\t\tj.Failed(err)\n\t\treturn nil\n\t}\n\tfacts.ForcedRoots = forcedRoots\n\n\t// avoid \"unused\" warnings while keeping the older default approach available.\n\t_ = greedyFusion\n\t_ = defaultFusion\n\n\tstages := greedyFusion(topological, comps, facts)\n\n\tfor i, stg := range stages {\n\t\terr := finalizeStage(stg, comps, facts)\n\t\tif err != nil {\n\t\t\terr = fmt.Errorf(\"preprocess validation failure of stage %v: %v\", i, err)\n\t\t\tj.SendMsg(err.Error())\n\t\t\tj.Failed(err)","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/preprocess.go#L145-L181","documentation":"In preProcessGraph, Prism runs computeFacts over the topologically sorted transforms to derive pipeline facts (producers, side inputs, etc.). If that computation fails, the error is wrapped with this prefix, sent to the job's message stream, and the job is marked Failed.","triggerScenarios":"executePipeline invokes preProcessGraph and computeFacts returns an error — such as a PCollection with two producers, or a transform whose side inputs can't be interpreted — before any stage fusion happens.","commonSituations":"Pipelines with malformed or unusual transform graphs (composite expansion leftovers, duplicate output wiring); SDK-emitted pipelines that violate Prism's single-producer assumptions.","solutions":["Inspect the wrapped cause message to see which pipeline invariant computeFacts violated.","Simplify or re-expand composites in the pipeline so each PCollection has exactly one producer.","Validate the pipeline graph with the DirectRunner, which tolerates graphs Prism rejects, to isolate the construct.","Report the pipeline graph to the Beam project if a valid pipeline triggers this."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// client-side: run the pipeline through DirectRunner first to surface graph problems\nif err := directRun(pipeline); err != nil {\n    return fmt.Errorf(\"pipeline graph invalid, prism will reject it: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"err := submitToPrism(pipeline)\nif err != nil && strings.Contains(err.Error(), \"error computing pipeline facts\") {\n    log.Printf(\"prism rejected graph: %v\", err) // fix graph then resubmit\n}","preventionTips":["Keep transform graphs simple; avoid hand-editing pipeline protos.","Ensure composites are fully expanded so each PCollection has one producer.","Smoke-test pipelines on the DirectRunner before 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"}