{"record":{"id":"c25a5001320e5e29","repo":"apache/beam","slug":"invalid-pipeline","errorCode":null,"errorMessage":"invalid pipeline","messagePattern":"invalid pipeline","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/direct/direct.go","lineNumber":65,"sourceCode":"\tif !beam.Initialized() {\n\t\tlog.Warn(ctx, \"Beam has not been initialized. Call beam.Init() before pipeline construction.\")\n\t}\n\n\tlog.Info(ctx, \"Pipeline:\")\n\tlog.Info(ctx, p)\n\tctx = metrics.SetBundleID(ctx, \"direct\") // Ensure a metrics.Store exists.\n\n\tif *jobopts.Strict {\n\t\tlog.Info(ctx, \"Strict mode enabled, applying additional validation.\")\n\t\tif _, err := vet.Execute(ctx, p); err != nil {\n\t\t\treturn nil, errors.Wrap(err, \"strictness check failed\")\n\t\t}\n\t\tlog.Info(ctx, \"Strict mode validation passed.\")\n\t}\n\n\tedges, _, err := p.Build()\n\tif err != nil {\n\t\treturn nil, errors.Wrap(err, \"invalid pipeline\")\n\t}\n\tplan, err := Compile(edges)\n\tif err != nil {\n\t\treturn nil, errors.Wrap(err, \"translation failed\")\n\t}\n\tbeam.PipelineOptions.LoadOptionsFromFlags(nil)\n\tlog.Info(ctx, plan)\n\n\tif err = plan.Execute(ctx, \"\", exec.DataContext{}); err != nil {\n\t\tplan.Down(ctx) // ignore any teardown errors\n\t\treturn nil, err\n\t}\n\tif err = plan.Down(ctx); err != nil {\n\t\treturn nil, err\n\t}\n\n\treturn newDirectPipelineResult(ctx)\n}","sourceCodeStart":47,"sourceCodeEnd":83,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/direct/direct.go#L47-L83","documentation":"After optional strict validation, the direct runner calls p.Build() to turn the pipeline into a graph of edges. If graph construction fails (invalid transform wiring, nil inputs, duplicate consumption of a PCollection), the error is wrapped as 'invalid pipeline'. It signals the pipeline could not be compiled into an execution graph at all.","triggerScenarios":"Calling beam.Run (direct runner) on a pipeline whose graph is invalid: consuming a PCollection after it was already consumed, using a nil/zero PCollection as input, building transforms outside beam.TryNewPipeline scope, or misuse of composite transforms.","commonSituations":"Copy-pasted transform code reusing the same output twice; pipelines constructed with a pipeline value from a different package instance; panics recovered into errors inside custom graph manipulations; tests that partially build pipelines.","solutions":["Inspect the inner error from p.Build() — it identifies the edge/node that failed.","Ensure each PCollection is consumed exactly once; fork it via beam.ParDo if branching is needed.","Check that all transforms are added to the same beam.Pipeline instance and inputs are non-nil.","If building programmatically, switch to beam.TryNewPipeline-style error-checked construction to catch the failure at the right place."],"exampleFix":"// before\nout1 := beam.ParDo(s, extractFn, col)\nout2 := beam.ParDo(s, otherFn, col) // col consumed twice -> invalid pipeline\n// after\nout1 := beam.ParDo(s, extractFn, col)\nout2 := beam.ParDo(s, otherFn, out1)","handlingStrategy":"try-catch","validationCode":"if _, err := p.Build(); err != nil {\n    return fmt.Errorf(\"pipeline graph invalid: %w\", err)\n}","typeGuard":null,"tryCatchPattern":"if _, err := beam.Run(ctx, runner, p); err != nil && strings.Contains(err.Error(), \"invalid pipeline\") {\n    log.Printf(\"check PCollection reuse/nil inputs: %v\", err)\n}","preventionTips":["Consume each PCollection exactly once; fork with explicit transforms.","Never pass nil PCollections as transform inputs.","Construct all transforms on one pipeline instance."],"tags":["go","beam","direct-runner","pipeline","graph"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}