18473514217

Committed 13 Oct 2025 05:24PM UTC coverage: 94.737% (-0.4%) from 95.141%

Build # 18473514217

Build Type

push

github

Committed by

web-flow

Commit Message

feat: add SOE codecs (#514)

Avro describes a very simple Single Object Encoding in
https://avro.apache.org/docs/1.12.0/specification/#single-object-encoding-specification

Encoders put a schema fingerprint in the header, which decoders can use to look
up a schema definition for deserialization.

There are two codecs:

* soe.Codec: follows avro.Marshal/Unmarshal but adds/parses SOE header
* soe.TypedCodec: a type-safe generic version for avrogen-generated types

Both of them can encode/decode SOE, using a single schema.

Also add soe.DynamicDecoder, which has a more liberal view of types and schemas:
for each record, try to look up the associated schema from a configured
resolver, and decode anything for which a schema is found. This can be used for
full dynamic encoding of streams of records from different encoders, as long as
their schemas are available for decoding. This is similar to how the Avro Java
implementation does it.

Include a minimal in-memory schema store.

Co-authored-by: Nicholas Wiersma <nick@wiersma.co.za>

Run Details

96 of 129 new or added lines in 5 files covered. (74.42%)

6264 of 6612 relevant lines covered (94.74%)

34613.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.0

/soe/basic.go

package soe

import (
        "bytes"
        "fmt"

        "github.com/hamba/avro/v2"
)

// Codec marshals values to/from bytes, with the Avro binary payload wrapped in
// an SOE frame containing the writer schema fingerprint.
type Codec struct {
        api    avro.API
        schema avro.Schema
        header []byte
}

// NewCodec creates a new Codec for a Schema and the default config.
func NewCodec(schema avro.Schema) (*Codec, error) {
        return NewCodecWithAPI(schema, avro.DefaultConfig)
}

// NewCodecWithAPI creates a new Codec for a Schema and an API.
func NewCodecWithAPI(schema avro.Schema, api avro.API) (*Codec, error) {
        // Precompute SOE header
        header, err := BuildHeader(schema)
        if err != nil {
                return nil, err
        }
        return &Codec{
                schema: schema,
                api:    api,
                header: header,
        }, nil
}

// Encode marshals a value to SOE-encoded Avro binary.
func (c *Codec) Encode(v any) ([]byte, error) {
        data, err := c.api.Marshal(c.schema, v)
        if err != nil {
                return nil, err
        }
        return append(c.header, data...), nil
}

// Decode unmarshals a value from SOE-encoded Avro binary, and fails if
// the schema fingerprint doesn't match the held schema.
func (c *Codec) Decode(data []byte, v any) error {
        fingerprint, data, err := ParseHeader(data)
        if err != nil {
                return err
        }
        expected := c.getFingerprint()
        if !bytes.Equal(fingerprint, expected) {
                return fmt.Errorf("bad fingerprint %x, expected %x", fingerprint, expected)
        }
        return c.api.Unmarshal(c.schema, data, v)
}

// DecodeUnverified unmarshals a value from SOE-encoded Avro binary without
// validating the schema fingerprint.
func (c *Codec) DecodeUnverified(data []byte, v any) error {
        _, data, err := ParseHeader(data)
        if err != nil {
                return err
        }
        return c.api.Unmarshal(c.schema, data, v)
}

func (c *Codec) getFingerprint() []byte {
        // We know the header is well-formed here, so make assumptions.
        return c.header[2:]
}

1	package soe
2
3	import (
4	"bytes"
5	"fmt"
6
7	"github.com/hamba/avro/v2"
8	)
9
10	// Codec marshals values to/from bytes, with the Avro binary payload wrapped in
11	// an SOE frame containing the writer schema fingerprint.
12	type Codec struct {
13	api avro.API
14	schema avro.Schema
15	header []byte
16	}
17
18	// NewCodec creates a new Codec for a Schema and the default config.
19	func NewCodec(schema avro.Schema) (*Codec, error) {	28✔
20	return NewCodecWithAPI(schema, avro.DefaultConfig)	28✔
21	}	28✔
22
23	// NewCodecWithAPI creates a new Codec for a Schema and an API.
24	func NewCodecWithAPI(schema avro.Schema, api avro.API) (*Codec, error) {	38✔
25	// Precompute SOE header	38✔
26	header, err := BuildHeader(schema)	38✔
27	if err != nil {	38✔
NEW 28	return nil, err	×
NEW 29	}	×
30	return &Codec{	38✔
31	schema: schema,	38✔
32	api: api,	38✔
33	header: header,	38✔
34	}, nil	38✔
35	}
36
37	// Encode marshals a value to SOE-encoded Avro binary.
38	func (c *Codec) Encode(v any) ([]byte, error) {	26✔
39	data, err := c.api.Marshal(c.schema, v)	26✔
40	if err != nil {	26✔
NEW 41	return nil, err	×
NEW 42	}	×
43	return append(c.header, data...), nil	26✔
44	}
45
46	// Decode unmarshals a value from SOE-encoded Avro binary, and fails if
47	// the schema fingerprint doesn't match the held schema.
48	func (c *Codec) Decode(data []byte, v any) error {	16✔
49	fingerprint, data, err := ParseHeader(data)	16✔
50	if err != nil {	24✔
51	return err	8✔
52	}	8✔
53	expected := c.getFingerprint()	8✔
54	if !bytes.Equal(fingerprint, expected) {	12✔
55	return fmt.Errorf("bad fingerprint %x, expected %x", fingerprint, expected)	4✔
56	}	4✔
57	return c.api.Unmarshal(c.schema, data, v)	4✔
58	}
59
60	// DecodeUnverified unmarshals a value from SOE-encoded Avro binary without
61	// validating the schema fingerprint.
62	func (c *Codec) DecodeUnverified(data []byte, v any) error {	16✔
63	_, data, err := ParseHeader(data)	16✔
64	if err != nil {	24✔
65	return err	8✔
66	}	8✔
67	return c.api.Unmarshal(c.schema, data, v)	8✔
68	}
69
70	func (c *Codec) getFingerprint() []byte {	8✔
71	// We know the header is well-formed here, so make assumptions.	8✔
72	return c.header[2:]	8✔
73	}	8✔

hamba / avro / 18473514217

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous