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)

func (d *Avro) DecodeInto(rec *kgo.Record, e Encoder, position string) (int, error)

type ColumnarDecoder ¶ added in v0.5.0

type ColumnarDecoder interface {
	DecodeInto(rec *kgo.Record, enc Encoder, position string) (int, error)
}

ColumnarDecoder is the optional direct-to-Arrow sibling of Decoder (#733). DecodeInto resolves the record's source key (a Debezium envelope source, or "" for a topic-keyed decoder), asks enc for that table's encoder, and appends each decoded row straight into its Arrow builders — no per-row map[string]any buffered between decode and encode.

It returns the number of rows appended: 0 for a tombstone, a skipped op, or an unmapped source. Every row is stamped with its operation, commit time and the given position. A record the direct path cannot encode without changing behavior (a shape the schema cannot represent, a missing target schema, or a delete that needs the map path's key backfill) returns ErrShapeDrift, and the caller re-encodes that record on the map path — which reproduces the row puller exactly.

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)

func (d *DebeziumJSON) DecodeInto(record *kgo.Record, e Encoder, position string) (int, 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 Encoder ¶ added in v0.5.0

type Encoder func(sourceKey string) (enc *transport.RowEncoder, mapped bool, err error)

Encoder resolves a decoded record's source key to the Arrow encoder for its target table, built from that table's canonical schema. It reports whether the source maps to a configured target (mapped=false means the record is skipped, as the row path skips an unmapped change). A nil encoder with mapped=true means the target has no canonical schema; the decoder then returns ErrShapeDrift so the caller falls back to the map path, which infers the shape from the rows.

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 ErrNotEnvelope ¶ added in v0.5.0

type ErrNotEnvelope struct {
	Op  string
	Err error
}

ErrNotEnvelope marks a message that is not a Debezium envelope — an empty op, an unknown op, or a payload that does not parse. It is fatal: the topic's format is an assertion, and silently skipping every record is total data loss.

func (e *ErrNotEnvelope) Error() string

func (e *ErrNotEnvelope) Unwrap() error

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 NewHTTPRegistryWithAuth(base, username, password string, tlsCfg *tls.Config) *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)

func (d *Raw) DecodeInto(rec *kgo.Record, e Encoder, position string) (int, 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.