unsupported collection type - only normal structs supported…

Error message

unsupported collection type - only normal structs supported for writing.

What it means

Panic raised by spannerio.Write when the input PCollection's type is CoGBK or KV. The Spanner writer only accepts plain element collections of schema structs; KV/CoGBK-typed data means the pipeline is feeding grouped data directly to the sink instead of extracting values first.

Solutions

Reshape the data into a plain struct: use beam.ParDo to map KV pairs into a struct type whose fields match the table schema.
Re-run the producing transform so it emits structs instead of KV.
Define a Go struct mirroring the Spanner table columns and emit that from your DoFns.

Example fix

// before
spannerio.Write(s, db, "users", kvCol)
// after
structs := beam.ParDo(s, func(k string, v []byte) User { return User{Key: k, Data: string(v)} }, kvCol)
spannerio.Write(s, db, "users", structs)

Defensive patterns

Strategy: type-guard

Validate before calling

if typex.IsCoGBK(col.Type()) || typex.IsKV(col.Type()) {
    return fmt.Errorf("spannerio.Write requires a struct-typed PCollection, got %v", col.Type())
}

Type guard

func isStructPCollection(col beam.PCollection) bool {
    return !typex.IsCoGBK(col.Type()) && !typex.IsKV(col.Type())
}

Try / catch

defer func() {
    if r := recover(); r != nil {
        err = fmt.Errorf("spannerio.Write type mismatch: %v", r)
    }
}()

Prevention

Always map keyed/KV data through a struct-emitting ParDo before writing to Spanner.
Define one canonical struct per Spanner table and use it as the pipeline's record type.
Check col.Type() during development with a log statement when wiring new sinks.

When it happens

Trigger: Passing a PCollection<K,V> (e.g. output of beam.CoGBK or beam.ParDo emitting KV) or a CoGBK result as the col argument to spannerio.Write.

Common situations: Piping the result of a group-by/keyed transform directly into the Spanner sink, or reading a KV-shaped source (like some text/BigQuery reads) and writing it without reshaping.

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

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

Appendix: source

Thrown at sdks/go/pkg/beam/io/spannerio/write.go:55

}

// UseBatchSize explicitly sets the batch size per transaction for writes.
func UseBatchSize(batchSize int) WriteOptionsFn {
	return func(qo *writeOptions) error {
		qo.BatchSize = batchSize
		return nil
	}
}

// Write writes the elements of the given PCollection<T> to spanner. T is required
// to be the schema type.
func Write(s beam.Scope, db string, table string, col beam.PCollection, options ...WriteOptionsFn) {
	if db == "" {
		panic("no database provided!")
	}

	if typex.IsCoGBK(col.Type()) || typex.IsKV(col.Type()) {
		panic("unsupported collection type - only normal structs supported for writing.")
	}

	s = s.Scope("spanner.Write")

	beam.ParDo0(s, newWriteFn(db, table, col.Type().Type(), options...), col)
}

type writeFn struct {
	spannerFn
	Table     string           `json:"table"`   // The table to write to
	Type      beam.EncodedType `json:"type"`    // Type is the encoded schema type.
	Options   writeOptions     `json:"options"` // Spanner write options
	mutations []*spanner.Mutation
}

func newWriteFn(db string, table string, t reflect.Type, options ...WriteOptionsFn) *writeFn {
	writeOptions := writeOptions{
		BatchSize: 1000, // default

View on GitHub (pinned to 12126d8942)