decoder

type Avro ¶

type Avro struct {

	// ByTopic declares field selection per topic. A topic absent from the
	// map (including when ByTopic itself is nil, the zero value) keeps every
	// field the schema decodes.
	ByTopic map[string]TopicExtraction
	// Miss is called for every declared field a record does not carry (a
	// column declared that the registry schema does not have). Optional.
	Miss MissFn
	// contains filtered or unexported fields
}

Avro decodes Confluent-Avro messages: a 1-byte magic (0x00), a 4-byte big-endian schema id, then the binary Avro payload. The schema is resolved from the registry by id and cached (schemas are immutable per id) and the payload is decoded into the canonical value shape — records become map[string]any, arrays []any, scalars their Go types — which the sink's composite writer consumes directly. Every message is an insert: the record IS the row. (A Debezium-Avro envelope, where the message wraps before/after, is a future envelope layer on top.)

One Reader consumes every topic its spec tables name through a single shared decoder, so field selection is declared per topic: ByTopic projects the decoded record onto a subset for a topic that has an entry, and keeps every field for one that does not. The registry schema is the source of truth for a topic's record shape, and the spec is a projection of it, so a column present in the record but not declared is dropped rather than tripping schema drift downstream. There is no raw-blob equivalent of Raw's KeepPayload here — the pre-decode bytes are Confluent wire format (magic byte + schema id + binary), unreadable without the registry, so storing them would land a column nobody can query.

func NewAvroDecoder(registry SchemaRegistry) *Avro

func (d *Avro) Decode(rec *kgo.Record) ([]rowchange.Change, error)

type DebeziumJSON ¶

type DebeziumJSON struct {
	// TopicToTable maps a Kafka topic name to the pipeline target table.
	// If empty, the decoder uses the source.db + source.table from the
	// envelope.
	TopicToTable map[string]string
}

DebeziumJSON decodes Kafka records in the Debezium JSON envelope format.

func (d *DebeziumJSON) Decode(record *kgo.Record) ([]rowchange.Change, error)

type Decoder ¶

type Decoder interface {
	Decode(record *kgo.Record) ([]rowchange.Change, error)
}

Decoder turns a single Kafka record into zero or more pipeline changes. The decoder is responsible for mapping the source message's operation type, key, and payload into the Change struct. A single record may produce multiple changes (e.g. a debezium outbox envelope wrapping a batch).

type ErrBadWireFormat ¶

type ErrBadWireFormat struct{}

ErrBadWireFormat marks a value that does not carry the Confluent wire header — the topic is not Confluent-Avro, or the value is corrupt.

func (e *ErrBadWireFormat) Error() string

type ErrFieldMissing ¶ added in v0.3.0

type ErrFieldMissing struct {
	Column string
	Path   string
}

ErrFieldMissing marks a required field the payload does not carry.

func (e *ErrFieldMissing) Error() string

type ErrNotJSON ¶ added in v0.3.0

type ErrNotJSON struct{ Err error }

ErrNotJSON marks a payload that could not be parsed as JSON when field extraction is declared. This is a hard failure, not a per-field miss: a payload extraction was declared for is presumed to be JSON, and one that is not means the topic does not hold what the spec says it does.

func (e *ErrNotJSON) Error() string

func (e *ErrNotJSON) Unwrap() error

type ErrUnknownSchema ¶

type ErrUnknownSchema struct {
	ID int
}

ErrUnknownSchema marks a schema id the registry does not know — the producer pointed at a different registry, or the id is corrupt. There is no schema to retry, so this is a needs-human condition, not a transient one.

func (e *ErrUnknownSchema) Error() string

type Field ¶ added in v0.3.0

type Field struct {
	// Name is the destination column.
	Name string
	// Path is the payload path, dot-separated for nested documents. Empty
	// means Name is the path.
	Path string
	// Required fails the record when the field is absent, rather than
	// landing NULL.
	Required bool
}

Field is one declared column and where to read it from.

type HTTPRegistry ¶

type HTTPRegistry struct {
	// contains filtered or unexported fields
}

HTTPRegistry is a Confluent-compatible schema registry client. Schemas are cached by id for the life of the process (immutable per id).

func NewHTTPRegistry(base string) *HTTPRegistry

func (r *HTTPRegistry) Get(ctx context.Context, id int) (avro.Schema, error)

type MissFn ¶ added in v0.3.0

type MissFn func(column string)

MissFn is called when a non-required declared field is absent from a payload, so the caller can count it. An absent field is either contract optionality or a producer that changed shape, and the counter is what makes the second case visible. The decoders close over the record's topic before handing this to Extract, so the callback signature here stays a plain column name — the pure unit is "document + declared fields", the topic is the caller's context, not this function's.

type Raw ¶

type Raw struct {
	// ByTopic declares field extraction per topic. A topic absent from the
	// map (including when ByTopic itself is nil, the zero value) stays fully
	// opaque: one "payload" column, the raw bytes, unconditionally.
	ByTopic map[string]TopicExtraction
	// Miss is called for every declared field a message does not carry — see
	// MissFn. Optional.
	Miss MissFn
}

Raw is the passthrough decoder. Every message is an insert — raw landing has no update or delete semantics, the log is the data. A tombstone (nil value) becomes a row with a NULL payload, which is the faithful record of the fact.

func (d *Raw) Decode(r *kgo.Record) ([]rowchange.Change, error)

type SchemaRegistry ¶

type SchemaRegistry interface {
	// Get returns the schema registered under id. A registry that does not
	// know the id returns ErrUnknownSchema.
	Get(ctx context.Context, id int) (avro.Schema, error)
}

SchemaRegistry resolves Avro schemas by Confluent schema id. Schemas are immutable once registered under an id — evolution publishes a NEW id and never rewrites an old one — so a per-id cache has no staleness window and no expiry.

type TopicExtraction ¶ added in v0.3.0

type TopicExtraction struct {
	Fields      []Field
	KeepPayload bool
}

TopicExtraction is one topic's field-extraction declaration: which fields to keep, and (for raw only) whether to also keep the full payload alongside them.