package columnarschema
import "github.com/krabka-io/krabka-streams-go/columnarschema"
Package columnarschema connects the schema registry serdes to columnar topologies.
AvroBatchCodec and ProtobufBatchCodec implement columnar.BatchCodec, so they plug into AddSource and AddSink directly, and their columns follow the record schema instead of JSON text: nested records become Struct, arrays and repeated fields List, maps Map, and decimals and timestamps their native Arrow types.
A few shapes have no finite or faithful Arrow form and fall back, tagged with field metadata so the write-back path can reverse them: a multi-branch Avro union becomes a struct with one nullable child per branch (krabka.avro.union), a recursive Avro record becomes its JSON text (krabka.json), and a recursive Protobuf message — or the dynamic google.protobuf.Struct, Value, and ListValue — becomes its canonical Protobuf JSON text (krabka.json plus krabka.proto.message naming the type).
Index
- Constants
- func AvroArrowField
- func AvroArrowSchema
- func ParseAvroSchema
- func ProtobufArrowField
- func ProtobufArrowSchema
- type AvroBatchCodec
- type AvroRowBridge
- type ProtobufBatchCodec
- type ProtobufRowBridge
Constants
const ( // MetadataJSON marks a Utf8 column that holds serialized JSON text. MetadataJSON = "krabka.json" // MetadataAvroEnum names the Avro enum type of a Utf8 symbol column. MetadataAvroEnum = "krabka.avro.enum" // MetadataAvroEnumSymbols lists the Avro enum symbols, comma separated. MetadataAvroEnumSymbols = "krabka.avro.enum.symbols" // MetadataAvroUnion marks a struct column that spreads a multi-branch // Avro union across one nullable child per branch. MetadataAvroUnion = "krabka.avro.union" // MetadataAvroFixed names the Avro fixed type of a fixed-size binary // column. MetadataAvroFixed = "krabka.avro.fixed" // MetadataAvroLogical names an Avro logical type carried by the column. MetadataAvroLogical = "krabka.avro.logical" // MetadataProtoEnum names the Protobuf enum type of a Utf8 symbol column. MetadataProtoEnum = "krabka.proto.enum" // MetadataProtoOneof names the Protobuf oneof a column belongs to. MetadataProtoOneof = "krabka.proto.oneof" // MetadataProtoWrapper names the well-known wrapper type a column // unwraps. MetadataProtoWrapper = "krabka.proto.wrapper" // MetadataProtoMessage names the Protobuf message type of a JSON text // column. MetadataProtoMessage = "krabka.proto.message" )
Field metadata keys that tag fallback and provenance information on bridge columns.
Functions
func AvroArrowField source
func AvroArrowField(name string, schema avro.Schema) (arrow.Field, error)
AvroArrowField translates one named Avro schema into an Arrow field.
func AvroArrowSchema source
func AvroArrowSchema(schema avro.Schema) (*arrow.Schema, error)
AvroArrowSchema translates an Avro schema into the Arrow schema its bridge produces, without touching data. A record schema spreads its fields across columns; any other schema becomes a single column named "value".
func ParseAvroSchema source
func ParseAvroSchema(schemaText string) (avro.Schema, error)
ParseAvroSchema parses Avro schema text, for callers that build bridges directly.
func ProtobufArrowField source
func ProtobufArrowField(field protoreflect.FieldDescriptor) arrow.Field
ProtobufArrowField translates one message field into an Arrow field.
func ProtobufArrowSchema source
func ProtobufArrowSchema(descriptor protoreflect.MessageDescriptor) *arrow.Schema
ProtobufArrowSchema translates a message descriptor into the Arrow schema its bridge produces, without touching data.
Types
type AvroBatchCodec source
type AvroBatchCodec struct { // contains filtered or unexported fields }
AvroBatchCodec is a columnar.BatchCodec for Avro topics backed by the schema registry.
The Arrow schema derives from the fixed reader schema once, at construction. Records written with other registered writer schemas are resolved onto that reader view by the embedded serde, so a schema change mid-stream never changes the columns, and an unknown writer schema ID surfaces as the cache's retriable pending-fetch error, which the group runner rethrows instead of skipping or dead-lettering.
func NewAvroBatchCodec source
func NewAvroBatchCodec(schemaText string, cache *registry.SchemaCache, mem memory.Allocator) (*AvroBatchCodec, error)
NewAvroBatchCodec creates a codec for a reader schema over generic Avro values.
func (*AvroBatchCodec) ArrowSchema source
func (c *AvroBatchCodec) ArrowSchema() *arrow.Schema
ArrowSchema returns the payload schema of every decoded batch.
func (*AvroBatchCodec) Decode source
func (c *AvroBatchCodec) Decode(topic string, records []columnar.ConsumedRecord) (arrow.Record, error)
Decode implements columnar.BatchCodec.
func (*AvroBatchCodec) Encode source
func (c *AvroBatchCodec) Encode(topic string, batch arrow.Record) ([]columnar.ProduceRecord, error)
Encode implements columnar.BatchCodec.
func (*AvroBatchCodec) RegisterSubject source
func (c *AvroBatchCodec) RegisterSubject(topic string)
RegisterSubject interns the codec's schema under the topic's subject for the next prewarm.
type AvroRowBridge source
type AvroRowBridge struct { // contains filtered or unexported fields }
AvroRowBridge converts between Avro generic values and Arrow batches.
Rows use hamba/avro's generic representation: records are map[string]any, enums strings, fixed values byte arrays, decimals *big.Rat, date and timestamp logical types time.Time, time logical types time.Duration, and non-nullable unions map[string]any keyed by the branch name.
The Arrow schema derives from the fixed reader schema once, at construction. The bridge is safe for concurrent use.
func NewAvroRowBridge source
func NewAvroRowBridge(schema avro.Schema) (*AvroRowBridge, error)
NewAvroRowBridge creates a bridge for a top-level Avro record schema.
func (*AvroRowBridge) ArrowSchema source
func (b *AvroRowBridge) ArrowSchema() *arrow.Schema
ArrowSchema returns the Arrow schema of every batch the bridge produces.
func (*AvroRowBridge) BatchToRows source
func (b *AvroRowBridge) BatchToRows(batch arrow.Record) ([]any, error)
BatchToRows implements columnar.RowBridge.
func (*AvroRowBridge) RowsToBatch source
func (b *AvroRowBridge) RowsToBatch(rows []any, mem memory.Allocator) (arrow.Record, error)
RowsToBatch implements columnar.RowBridge.
type ProtobufBatchCodec source
type ProtobufBatchCodec[T proto.Message] struct { // contains filtered or unexported fields }
ProtobufBatchCodec is a columnar.BatchCodec for Protobuf topics backed by the schema registry.
The Arrow schema derives from the message descriptor once, at construction, and the embedded serde enforces the registry messageType check on read.
func NewProtobufBatchCodec source
func NewProtobufBatchCodec[T proto.Message](prototype T, cache *registry.SchemaCache, mem memory.Allocator) *ProtobufBatchCodec[T]
NewProtobufBatchCodec creates a codec for the message type of prototype, which is typically the zero value of a generated message type.
func (*ProtobufBatchCodec[T]) ArrowSchema source
func (c *ProtobufBatchCodec[T]) ArrowSchema() *arrow.Schema
ArrowSchema returns the payload schema of every decoded batch.
func (*ProtobufBatchCodec[T]) Decode source
func (c *ProtobufBatchCodec[T]) Decode(topic string, records []columnar.ConsumedRecord) (arrow.Record, error)
Decode implements columnar.BatchCodec.
func (*ProtobufBatchCodec[T]) Encode source
func (c *ProtobufBatchCodec[T]) Encode(topic string, batch arrow.Record) ([]columnar.ProduceRecord, error)
Encode implements columnar.BatchCodec.
func (*ProtobufBatchCodec[T]) RegisterSubject source
func (c *ProtobufBatchCodec[T]) RegisterSubject(topic string)
RegisterSubject interns the codec's schema under the topic's subject for the next prewarm.
type ProtobufRowBridge source
type ProtobufRowBridge[T proto.Message] struct { // contains filtered or unexported fields }
ProtobufRowBridge converts between Protobuf messages and Arrow batches.
The Arrow schema derives from the message descriptor once, at construction. Enums become symbol names, uint64 stays exact as an unsigned 64-bit column, google.protobuf.Timestamp becomes a UTC microsecond timestamp with sub-microsecond nanos truncated, wrapper types unwrap to their value, and google.protobuf.Struct, Value, ListValue, and recursive messages fall back to canonical Protobuf JSON text.
func NewProtobufRowBridge source
func NewProtobufRowBridge[T proto.Message](prototype T) *ProtobufRowBridge[T]
NewProtobufRowBridge creates a bridge for the message type of prototype.
func (*ProtobufRowBridge[T]) ArrowSchema source
func (b *ProtobufRowBridge[T]) ArrowSchema() *arrow.Schema
ArrowSchema returns the Arrow schema of every batch the bridge produces.
func (*ProtobufRowBridge[T]) BatchToRows source
func (b *ProtobufRowBridge[T]) BatchToRows(batch arrow.Record) ([]T, error)
BatchToRows implements columnar.RowBridge.
func (*ProtobufRowBridge[T]) RowsToBatch source
func (b *ProtobufRowBridge[T]) RowsToBatch(rows []T, mem memory.Allocator) (arrow.Record, error)
RowsToBatch implements columnar.RowBridge.