apache/beam · error

expected WV coder for side input

Error message

expected WV coder for side input %v: %v

What it means

NewSideInputAdapter builds the encoding/decoding machinery for a side input and requires the coder to be a windowed coder (W<V> or W<KV<K,V>>), because the side-input wire protocol carries windowing information. If coder.IsW(c) is false the function panics with this message naming the StreamID and coder. This is a construction-time contract check protecting the protocol.

Solutions

1. Wrap the coder in a windowed coder: coder.NewW(inner, window.Window{}) before creating the adapter
2. Ensure the side input PCollection was created through normal beam.ParDo/beam.SideInput plumbing, not hand-built coders
3. Check custom translation code for coder.SkipW calls leaking into adapter construction
4. If hit in SDK-internal translation of a normal pipeline, file a bug with the graph JSON

Example fix

// before
adapter := exec.NewSideInputAdapter(sid, sideID, kvCoder, wm) // kvCoder is not windowed
// after
adapter := exec.NewSideInputAdapter(sid, sideID, coder.NewW(kvCoder, window.GlobalWindows{}.Coder()), wm)

if !coder.IsW(c) {
    c = coder.NewW(c, window.GlobalWindows{}) // wrap before adapter creation
}

func isWindowed(c *coder.Coder) bool { return coder.IsW(c) }

// Go: recover in custom translation/test harnesses
defer func() {
    if r := recover(); r != nil {
        if strings.Contains(fmt.Sprint(r), "expected WV coder") {
            log.Fatalf("side input coder not windowed: %v", r)
        }
        panic(r)
    }
}()

When it happens

Trigger: Calling NewSideInputAdapter with a non-windowed coder; a PCollection whose coder lost its window wrapper (e.g. built with coder.New without NewW); translation bugs where the window was skipped before adapter creation.

Common situations: Custom exec/translation code or tests constructing side input adapters manually with raw coders; pipeline graph manipulation that strips windowing; writing unit tests (like TestNewIterable_BadMapper) that pass bad coders intentionally.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• array len mismatch. decoding
• broken stream
• broken stream
• coder must not be nil
• coder type must be identical to node type

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/sideinput.go:53

	NewIterable(ctx context.Context, reader StateReader, w typex.Window) (ReStream, error)
	NewKeyedIterable(ctx context.Context, reader StateReader, w typex.Window, iterKey any) (ReStream, error)
}

type sideInputAdapter struct {
	sid         StreamID
	sideInputID string
	wc          WindowEncoder
	kc          ElementEncoder
	ec          ElementDecoder
	wm          WindowMapper
	c           *coder.Coder
}

// NewSideInputAdapter returns a side input adapter for the given StreamID and coder.
// It expects a W<V> or W<KV<K,V>> coder, because the protocol requires windowing information.
func NewSideInputAdapter(sid StreamID, sideInputID string, c *coder.Coder, wm WindowMapper) SideInputAdapter {
	if !coder.IsW(c) {
		panic(fmt.Sprintf("expected WV coder for side input %v: %v", sid, c))
	}

	wc := MakeWindowEncoder(c.Window)
	var kc ElementEncoder
	var ec ElementDecoder
	if coder.IsKV(coder.SkipW(c)) {
		kc = MakeElementEncoder(coder.SkipW(c).Components[0])
		ec = MakeElementDecoder(coder.SkipW(c).Components[1])
	} else {
		ec = MakeElementDecoder(coder.SkipW(c))
	}
	return &sideInputAdapter{sid: sid, sideInputID: sideInputID, wc: wc, kc: kc, ec: ec, wm: wm, c: c}
}

// NewIterable returns a ReStream of an iterable side input from the runner, either by getting the ReStream from
// the side input cache or by opening a new stream and reading it in.
func (s *sideInputAdapter) NewIterable(ctx context.Context, reader StateReader, w typex.Window) (ReStream, error) {
	key := []byte(iterableSideInputKey)

View on GitHub (pinned to 12126d8942)