apache/beam · error

preprocess validation failure of stage

Error message

preprocess validation failure of stage %v: %v

What it means

After greedy fusion, Prism validates each fused stage via finalizeStage; a failure means the fused stage is internally inconsistent (bad wiring, unsupported combine, etc.). The error is wrapped with the stage index, reported to the job's message stream, and the job is Failed.

Solutions

1. Read the %v inner error for the specific stage and the stage index to locate the offending fused stage.
2. Restructure the pipeline (e.g. split complex composite DoFns, avoid problematic transform adjacency) so stages fuse differently.
3. Try forcing stage boundaries (e.g. reshuffles) to prevent the bad fusion.
4. Verify with DirectRunner and file a Prism bug with the pipeline if it's a valid graph.

// validate the graph on a permissive runner before prism stage fusion
if err := directRun(pipeline); err != nil {
    return fmt.Errorf("pipeline won't validate: %w", err)
}

err := submitToPrism(pipeline)
if err != nil && strings.Contains(err.Error(), "preprocess validation failure of stage") {
    // restructure pipeline to change fusion, then retry once
}

When it happens

Trigger: executePipeline → preProcessGraph loops over stages produced by greedyFusion and finalizeStage returns an error for stage i — the pipeline graph yields a stage Prism cannot finalize.

Common situations: Pipelines with transform combinations that fuse into invalid stages (e.g. unsupported GBK/flatten placements after fusion); SDK pipeline shapes not exercised by Prism tests.

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

• computeFacts: two producers for one PCollection
• error computing pipeline facts
• expected single value map, had
• found uses of features unimplemented in prism in job
• GetJobMetrics: unknown jobID

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/987b7d5ad9829028. Report an issue: GitHub.

Appendix: source

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

	facts, err := computeFacts(topological, comps)
	if err != nil {
		err = fmt.Errorf("error computing pipeline facts: %w", err)
		j.SendMsg(err.Error())
		j.Failed(err)
		return nil
	}
	facts.ForcedRoots = forcedRoots

	// avoid "unused" warnings while keeping the older default approach available.
	_ = greedyFusion
	_ = defaultFusion

	stages := greedyFusion(topological, comps, facts)

	for i, stg := range stages {
		err := finalizeStage(stg, comps, facts)
		if err != nil {
			err = fmt.Errorf("preprocess validation failure of stage %v: %v", i, err)
			j.SendMsg(err.Error())
			j.Failed(err)
			return nil
		}
	}
	var stageDetails []any
	for i, stg := range stages {
		var transformNames []string
		for _, tid := range stg.transforms {
			transformNames = append(transformNames, comps.GetTransforms()[tid].GetUniqueName())
		}
		stageDetails = append(stageDetails,
			slog.Group(fmt.Sprintf("stage-%03d", i),
				slog.String("environment", stg.envID),
				slog.Any("transforms", transformNames),
			),
		)
	}

View on GitHub (pinned to 12126d8942)