Schema Versions

A message’s schema version is a column on its row, written from the Go type that produced it. A stream is one name with one log; rows of different versions sit in it side by side, and a consumer group reads only the rows at its own type’s version.

The type declares its version

type OrderPlacedV2 struct {
	OrderId string `json:"order_id"`
	Total   Money  `json:"total"`
}

func (OrderPlacedV2) SchemaVersion() int { return 2 }

instance, err := client.Stream[OrderPlacedV2]("orders.placed").Producer().Register(ctx, nil)
if err != nil {
	return err
}

The producer and consumer require sqlstreams.Versioned, whose method is SchemaVersion(). The stream handle carries the type: client.Stream[OrderPlacedV2]("orders.placed"). Its producer and consumer inherit that type, so a handler cannot accidentally accept a different Go payload type. The method supplies the version stored on each message; there is no config override.

Every produce writes the method’s value into the row:

message_log_<id>.schema_version INTEGER NOT NULL   -- what this row's payload is

stream_config identifies streams by unique name. Versions share one log and one id space.

A group reads its own version and steps over the rest

instance, err := client.Stream[OrderPlacedV2]("orders.placed").Consumer("fraud-scoring").Register(ctx, nil)
if err != nil {
	return err
}

The claim adds m.schema_version = $version beside the binding predicate it already has. A v1 row under a v2 group is treated the way a routing key that matches no binding is treated: the cursor passes over it without decoding it. The per-version health read counts only rows at the selected version; raw cursor distance still spans skipped ids. The exception claim filters the same way, so a v1 row that was dead-lettered or delayed never replays weeks later into a v2 struct.

A rolling deploy is then two groups on one stream. Say orders.placed has been live a month at v1 with Total int64 cents and the change is Total Money. Deploy producers built on OrderPlacedV2 -> new rows carry schema_version = 2 -> the existing fraud-scoring group on OrderPlacedV1 keeps draining v1 rows and steps over v2 -> a second group fraud-scoring-v2 on OrderPlacedV2 does the opposite. Stop v1 producers before retiring v1 consumers. Check the health read for unread and unresolved v1 deliveries, and investigate dead rows separately before stopping the old group. Retained v1 history can remain after processing has finished.

Compacted streams

Compaction is per key per stream, and the winner compare reads the version first: (schema_version, compaction_rank, id). A v2 write for acct-1 becomes the key’s head whatever its rank, and the v1 row stops being claimable. Inside one version, rank then id decide as before.

That ordering is what makes the bridge pattern work on one stream. A group bound to the old type consumes each key’s current head and re-produces it as the new type at CompactionRank -1. Being v2, the copy beats the v1 head it was made from; being rank -1, it loses to any live rank-0 v2 write for the same key, in either arrival order. .tests/e2e/schemaevolution runs that end to end, including a crash mid-drain.

Retirement is a query, not a guess. client.Stream[sqlstreams.RawPayload]("orders.placed").Health lists every version present in the log with the compaction heads still at it and each group’s unread rows and unresolved exceptions at it. No heads, unread rows, or unresolved exceptions is the retirement verdict. A version with heads at it reads “compaction heads remain” no matter how drained the groups are.

What it costs

• A v1 row lives until retention drains it, not until you drop its family. Retiring v1 is “no v1 rows or heads remain,” not DROP TABLE.
• Partitions and retention are shared across versions. A version cannot be sized or retained on its own.
• Four bytes per message for the version column.
• A stream handle’s Get, Destroy, Rename and the CLI take a name only; there is no --schema-version flag anywhere.

Not shipped

Reading an older version through a converter (Confluent’s reader schema, Temporal’s data converter) is not shipped. Consumers currently skip other payload versions; plan separate groups or an explicit conversion handler when evolving a payload.