{"record":{"id":"8d0026469395e740","repo":"apache/beam","slug":"expected-one-output-from-datasource-got-v","errorCode":null,"errorMessage":"expected one output from DataSource, got %v","messagePattern":"expected one output from DataSource, got (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/translate.go","lineNumber":64,"sourceCode":"\turnPerKeyCombineConvert                = \"beam:transform:combine_per_key_convert_to_accumulators:v1\"\n\turnPairWithRestriction                 = \"beam:transform:sdf_pair_with_restriction:v1\"\n\turnSplitAndSizeRestrictions            = \"beam:transform:sdf_split_and_size_restrictions:v1\"\n\turnProcessSizedElementsAndRestrictions = \"beam:transform:sdf_process_sized_element_and_restrictions:v1\"\n\turnTruncateSizedRestrictions           = \"beam:transform:sdf_truncate_sized_restrictions:v1\"\n)\n\n// UnmarshalPlan converts a model bundle descriptor into an execution Plan.\nfunc UnmarshalPlan(desc *fnpb.ProcessBundleDescriptor, dataSampler *DataSampler) (*Plan, error) {\n\tb, err := newBuilder(desc, dataSampler)\n\tif err != nil {\n\t\treturn nil, err\n\t}\n\tfor id, transform := range desc.GetTransforms() {\n\t\tif transform.GetSpec().GetUrn() != urnDataSource {\n\t\t\tcontinue\n\t\t}\n\t\tif len(transform.GetOutputs()) != 1 {\n\t\t\treturn nil, errors.Errorf(\"expected one output from DataSource, got %v\", transform.GetOutputs())\n\t\t}\n\n\t\tport, cid, err := unmarshalPort(transform.GetSpec().GetPayload())\n\t\tif err != nil {\n\t\t\treturn nil, err\n\t\t}\n\n\t\tu := &DataSource{UID: b.idgen.New()}\n\t\tu.Coder, err = b.coders.Coder(cid) // Expected to be windowed coder\n\t\tif err != nil {\n\t\t\treturn nil, err\n\t\t}\n\t\tif !coder.IsW(u.Coder) {\n\t\t\treturn nil, errors.Errorf(\"unwindowed coder %v on DataSource %v: %v\", cid, id, u.Coder)\n\t\t}\n\n\t\t// There's only a single pair in this map, but a for loop range statement\n\t\t// is the easiest way to extract it, so this loop will iterate only once.","sourceCodeStart":46,"sourceCodeEnd":82,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/translate.go#L46-L82","documentation":"UnmarshalPlan translates a pipeline description into an executable plan. The DataSource transform must have exactly one output; if the harness receives a DataSource transform with zero or multiple outputs, the plan cannot be linked and this error is returned.","triggerScenarios":"Unmarshalling a runner-provided pipeline description (desc) whose DataSource transform (urnDataSource) has an Outputs map with a length other than 1, via getOrCreatePlan/UnmarshalPlan.","commonSituations":"Buggy or incompatible runner/harness protocol negotiation; hand-crafted or mutated pipeline protos in tests; SDK/runner version mismatch producing malformed graph payloads.","solutions":["Inspect the pipeline description proto and ensure the DataSource transform declares exactly one output","Upgrade/align the Beam Go SDK and runner (e.g. Flink/Spark/Dataflow) versions so the graph encoding matches","Reproduce with a minimal pipeline and file an issue if the runner emits multi-output DataSource transforms"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Pre-validate the incoming pipeline description\nfor _, t := range desc.GetTransforms() {\n    if t.GetSpec().GetUrn() == \"beam:runner:source:v1\" && len(t.GetOutputs()) != 1 {\n        return fmt.Errorf(\"bad DataSource outputs: %v\", t.GetOutputs())\n    }\n}","typeGuard":null,"tryCatchPattern":"plan, err := exec.UnmarshalPlan(desc)\nif err != nil {\n    return fmt.Errorf(\"unmarshal plan failed: %w\", err) // surface for runner-side diagnosis\n}","preventionTips":["Keep runner and SDK versions aligned","Avoid hand-editing pipeline protos","Validate pipeline descriptions in harness integration tests"],"tags":["go","apache-beam","pipeline-translation"],"backgroundTag":"unexpected-response-shape","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}