{"record":{"id":"55cf34e1579f828b","repo":"apache/beam","slug":"unwindowed-coder-v-on-datasink-v-v","errorCode":null,"errorMessage":"unwindowed coder %v on DataSink %v: %v","messagePattern":"unwindowed coder (.+?) on DataSink (.+?): (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/translate.go","lineNumber":826,"sourceCode":"\t\t// Use the same flatten instance for all the inputs links to this transform.\n\t\tfor i := 0; i < len(transform.Inputs); i++ {\n\t\t\tb.links[linkID{id.to, i}] = u\n\t\t}\n\n\tcase urnDataSink:\n\t\tport, cid, err := unmarshalPort(payload)\n\t\tif err != nil {\n\t\t\treturn nil, err\n\t\t}\n\n\t\tsink := &DataSink{UID: b.idgen.New()}\n\t\tsink.SID = StreamID{PtransformID: id.to, Port: port}\n\t\tsink.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(sink.Coder) {\n\t\t\treturn nil, errors.Errorf(\"unwindowed coder %v on DataSink %v: %v\", cid, id, sink.Coder)\n\t\t}\n\t\tu = sink\n\n\tcase graphx.URNToString:\n\t\tu = &ToString{UID: b.idgen.New(), Out: out[0]}\n\n\tdefault:\n\t\tpanic(fmt.Sprintf(\"Unexpected transform URN: %v\", urn))\n\t}\n\n\tb.links[id] = u\n\tb.units = append(b.units, u)\n\treturn u, nil\n}\n\nfunc unmarshalReshuffleCoders(mainID string, payloads map[string][]byte) (*coder.Coder, error) {\n\tm := map[string]*pipepb.Coder{}\n\tfor id, v := range payloads {","sourceCodeStart":808,"sourceCodeEnd":844,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/translate.go#L808-L844","documentation":"The DataSink (bundle write) coder must be a windowed coder (Kind W) so window metadata can be encoded alongside elements. makeLink resolves the coder ID from the port payload and, if coder.IsW is false, fails with this error because writing unwindowed elements would lose windowing information.","triggerScenarios":"The coder ID referenced by the DataSink port payload resolves to a non-windowed coder — e.g. a custom runner or hand-built job proto attaching a raw element coder instead of coder.NewW(...), or custom coder logic that unwraps the windowed coder.","commonSituations":"Custom Go runners implementing the FnAPI; pipelines whose PCollection was re-coded after windowing, stripping the W wrapper; hand-edited pipeline protos used in tests.","solutions":["Wrap the sink coder as windowed: use coder.NewW(elemCoder, windowCoder) when building the pipeline/port payload","If using a custom runner, ensure it sends the same windowed coder ID the Go SDK expects on urnDataSink","Resubmit the unmodified pipeline from the Beam SDK so coder inference adds the W wrapper automatically"],"exampleFix":"// before\ncid := elemCoderID\n// after\nwc, _ := beam.WindowingStrategyOf(coll).WindowFn().Coder()\nc := coder.NewW(elemCoder, wc)\npipelineProto.Coders[c.GetId()] = c","handlingStrategy":"validation","validationCode":"sinkCoder, err := coders.Coder(cid)\nif err != nil {\n\treturn err\n}\nif !coder.IsW(sinkCoder) {\n\treturn fmt.Errorf(\"datasink coder %q must be windowed (W), got %v\", cid, sinkCoder)\n}","typeGuard":"null","tryCatchPattern":"if _, err := graph.MakePipeline(pipelineProto); err != nil {\n\tif strings.Contains(err.Error(), \"unwindowed coder\") {\n\t\t// wrap the sink coder via coder.NewW(elemCoder, windowCoder) and rebuild\n\t}\n}","preventionTips":["Always build sink coders with coder.NewW over the element coder","Never re-code a windowed PCollection with a bare element coder before writing","For custom runners, mirror the Go SDK's windowed-coder construction on urnDataSink"],"tags":["go","apache-beam","coder","windowing","datasink"],"backgroundTag":"type-mismatch","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"}