apache/beam · error

unwindowed coder %v on DataSink %v: %v

Error message

unwindowed coder %v on DataSink %v: %v

What it means

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.

Source

Thrown at sdks/go/pkg/beam/core/runtime/exec/translate.go:826

		// Use the same flatten instance for all the inputs links to this transform.
		for i := 0; i < len(transform.Inputs); i++ {
			b.links[linkID{id.to, i}] = u
		}

	case urnDataSink:
		port, cid, err := unmarshalPort(payload)
		if err != nil {
			return nil, err
		}

		sink := &DataSink{UID: b.idgen.New()}
		sink.SID = StreamID{PtransformID: id.to, Port: port}
		sink.Coder, err = b.coders.Coder(cid) // Expected to be windowed coder
		if err != nil {
			return nil, err
		}
		if !coder.IsW(sink.Coder) {
			return nil, errors.Errorf("unwindowed coder %v on DataSink %v: %v", cid, id, sink.Coder)
		}
		u = sink

	case graphx.URNToString:
		u = &ToString{UID: b.idgen.New(), Out: out[0]}

	default:
		panic(fmt.Sprintf("Unexpected transform URN: %v", urn))
	}

	b.links[id] = u
	b.units = append(b.units, u)
	return u, nil
}

func unmarshalReshuffleCoders(mainID string, payloads map[string][]byte) (*coder.Coder, error) {
	m := map[string]*pipepb.Coder{}
	for id, v := range payloads {

View on GitHub (pinned to 12126d8942)

Solutions

1. Wrap the sink coder as windowed: use coder.NewW(elemCoder, windowCoder) when building the pipeline/port payload
2. If using a custom runner, ensure it sends the same windowed coder ID the Go SDK expects on urnDataSink
3. Resubmit the unmodified pipeline from the Beam SDK so coder inference adds the W wrapper automatically

Example fix

// before
cid := elemCoderID
// after
wc, _ := beam.WindowingStrategyOf(coll).WindowFn().Coder()
c := coder.NewW(elemCoder, wc)
pipelineProto.Coders[c.GetId()] = c

sinkCoder, err := coders.Coder(cid)
if err != nil {
	return err
}
if !coder.IsW(sinkCoder) {
	return fmt.Errorf("datasink coder %q must be windowed (W), got %v", cid, sinkCoder)
}

null

if _, err := graph.MakePipeline(pipelineProto); err != nil {
	if strings.Contains(err.Error(), "unwindowed coder") {
		// wrap the sink coder via coder.NewW(elemCoder, windowCoder) and rebuild
	}
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• unwindowed coder %v on DataSource %v: %v
• could not unmarshal window coder for pcollection %v: %w
• could not unmarshal window coder for timer: %w
• could not unmarshal global window coder: %w
• window must not be nil

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