{"record":{"id":"c58e10aa04ab2975","repo":"apache/beam","slug":"expected-single-input-pcollection-in-reshuffle","errorCode":null,"errorMessage":"Expected single input PCollection in reshuffle: ","messagePattern":"Expected single input PCollection in reshuffle: ","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/handlerunner.go","lineNumber":185,"sourceCode":"\n\tif h.config.SDKReshuffle {\n\t\tpanic(\"SDK side reshuffle not yet supported\")\n\t}\n\n\t// A Reshuffle, in principle, is a no-op on the pipeline structure, WRT correctness.\n\t// It could however affect performance, so it exists to tell the runner that this\n\t// point in the pipeline needs a fusion break, to enable the pipeline to change it's\n\t// degree of parallelism.\n\t//\n\t// The change of parallelism goes both ways. It could allow for larger batch sizes\n\t// enable smaller batch sizes downstream if it is infact paralleizable.\n\t//\n\t// But for a single transform node per stage runner, we can elide it entirely,\n\t// since the input collection and output collection types match.\n\n\t// Get the input and output PCollections, there should only be 1 each.\n\tif len(t.GetInputs()) != 1 {\n\t\tpanic(\"Expected single input PCollection in reshuffle: \" + prototext.Format(t))\n\t}\n\tif len(t.GetOutputs()) != 1 {\n\t\tpanic(\"Expected single output PCollection in reshuffle: \" + prototext.Format(t))\n\t}\n\n\tinColID := getOnlyValue(t.GetInputs())\n\toutColID := getOnlyValue(t.GetOutputs())\n\n\t// We need to find all Transforms that consume the output collection and\n\t// replace them so they consume the input PCollection directly.\n\n\t// We need to remove the consumers of the output PCollection.\n\ttoRemove := []string{}\n\t// We need to force the consumers to be stage root,\n\t// because reshuffle should be a fusion break.\n\tforcedRoots := []string{}\n\n\tfor tid, t := range comps.GetTransforms() {","sourceCodeStart":167,"sourceCodeEnd":203,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/handlerunner.go#L167-L203","documentation":"A Reshuffle transform must consume exactly one input PCollection. When the submitted PTransform's inputs map does not have exactly one entry, the runner panics with the transform's prototext dump, because the elision optimization (rewiring input to downstream consumers) only works 1:1.","triggerScenarios":"A pipeline graph reaches handleReshuffle with a Reshuffle/RedistributeByKey/RedistributeArbitrarily transform that has zero or multiple inputs — typically from a malformed or hand-built graph rather than SDK-generated code.","commonSituations":"Custom pipeline builders or graph-rewriting tools that wire extra inputs into a reshuffle node; internal prism composites that were flattened incorrectly; corrupted job-submission payloads.","solutions":["Inspect the prototext dump in the panic to see the actual inputs wired into the reshuffle transform.","Regenerate the pipeline with the standard beam.Reshuffle() API so the SDK produces a single-input reshuffle.","Fix any custom graph construction/rewriting code to keep reshuffle transforms strictly single-input/single-output.","Report upstream if a standard pipeline produced a multi-input reshuffle."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"if t.GetUrnReshuffle() && len(t.GetInputs()) != 1 { /* fix graph before submission */ }","typeGuard":null,"tryCatchPattern":"defer func(){ if r := recover(); r != nil && strings.Contains(fmt.Sprint(r), \"Expected single input PCollection in reshuffle\") { /* inspect transform dump */ } }()","preventionTips":["Build reshuffles only via beam.Reshuffle()","Keep custom graph rewriters from adding inputs to reshuffle nodes"],"tags":["go","apache-beam","prism-runner","reshuffle"],"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"}