Documentation
¶
Overview ¶
Package ocf implements Avro Object Container Files (OCF).
An OCF is self-describing: the schema lives in the file header, so you do not need it out of band. Data sits in compressed blocks separated by sync markers, which makes files splittable for parallel processing. OCF is the standard way to store Avro on disk. To send individual values over the wire, see avro.AppendSingleObject instead.
See the Avro specification for the full format definition.
Writing ¶
schema := avro.MustParse(`{
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "age", "type": "int"}
]
}`)
f, err := os.Create("users.avro")
if err != nil { ... }
w, err := ocf.NewWriter(f, schema, ocf.WithCodec(ocf.SnappyCodec()))
if err != nil { ... }
for _, u := range users {
if err := w.Encode(&u); err != nil { ... }
}
if err := w.Close(); err != nil { ... }
Reading ¶
f, err := os.Open("users.avro")
if err != nil { ... }
r, err := ocf.NewReader(f)
if err != nil { ... }
for {
var u User
if err := r.Decode(&u); err != nil {
if err == io.EOF { break }
...
}
fmt.Println(u)
}
Appending ¶
Use NewAppendWriter to add records to an existing file without rewriting it.
Codecs ¶
Null, deflate, snappy, and zstandard are built in. You can supply your own via WithCodec.
Block size limits ¶
We cap both the compressed block we read off the wire (WithMaxBlockBytes) and what that block decompresses to (WithMaxDecompressedBlockBytes), each 64 MiB by default, to bound memory and decode time on untrusted input. We do not cap the writer, matching Java's DataFileWriter and fastavro: we write whatever blocks you give us.
A single Avro datum cannot be split across blocks, so a value larger than the reader's default cap (say an 80 MiB blob) is written as one block that a default reader refuses. The error names the option to raise:
r, err := ocf.NewReader(f, ocf.WithMaxDecompressedBlockBytes(128<<20))
Match the cap to the largest block your writer produces. For single large values that is the datum size, not WithBlockBytes.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BoundedDecompressor ¶ added in v1.8.0
BoundedDecompressor is an optional interface a Codec can implement. If yours does, we call DecompressBounded with our per-block cap (from WithMaxDecompressedBlockBytes) rather than Codec.Decompress, so you can refuse an oversized block *before* allocating it. That is the only real defense against a decompression bomb, since checking the size afterward means the allocation already happened. A Codec without this method decompresses unbounded, so for untrusted data supply one that bounds itself. All built-in codecs do.
Note that a wrapper embedding Codec does not inherit this method, because embedding an interface promotes only that interface's methods. Your wrapper must forward DecompressBounded itself, or it silently disables bounding for the codec it wraps. NopCloser forwards it.
max <= 0 means no limit. max is constant across all calls for a given Reader, so a codec that caches a configured decoder (say a zstd decoder built with a memory limit) may honor only the first call's max.
type Codec ¶
type Codec interface {
// Name returns the codec identifier for the "avro.codec" metadata key
// (e.g. "null", "deflate", "snappy", "zstandard").
Name() string
// Compress encodes a raw data block for storage.
Compress(src []byte) ([]byte, error)
// Decompress decodes a stored data block back to raw bytes.
Decompress(src []byte) ([]byte, error)
// Close releases whatever the codec holds. A codec that holds nothing
// can return nil.
Close() error
}
Codec compresses and decompresses OCF data blocks.
func DeflateCodec ¶
DeflateCodec returns a Codec using raw DEFLATE compression at the given level (e.g. flate.DefaultCompression).
func MustZstdCodec ¶
MustZstdCodec is like ZstdCodec but panics on error.
func NopCloser ¶
NopCloser returns a Codec that wraps c with a no-op Close, so that closing a Writer or Reader (or a constructor failing and releasing the codec it was handed) does not close a codec you share with another writer or reader. You close the underlying codec yourself once you are done with it.
If c implements BoundedDecompressor, so does the result: we forward the reader's decompression bound to c.
func SnappyCodec ¶
func SnappyCodec() Codec
SnappyCodec returns a Codec using Snappy compression with a trailing CRC-32 checksum per block, as required by the Avro spec.
func ZstdCodec ¶
ZstdCodec returns a Codec using Zstandard compression. We pass eopts to zstd.NewWriter and dopts to zstd.NewReader; both may be nil for defaults. We prepend zstd.WithEncoderConcurrency(1) and zstd.WithDecoderConcurrency(1), so pass a different concurrency to override. You can share one ZstdCodec across readers and writers via NopCloser.
ZstdCodec implements BoundedDecompressor: we build the decoder lazily on first read with zstd.WithDecoderMaxMemory set from the reader's WithMaxDecompressedBlockBytes cap, and then cache that decoder. Note that a ZstdCodec shared across readers with different caps therefore honors the first reader's cap; set zstd.WithDecoderMaxMemory in dopts to pin a bound regardless of reader. We raise a cap below zstd.MinWindowSize (1 KiB) up to it, since a smaller limit would reject every frame.
type Opt ¶
Opt is an option that applies to both NewWriter and NewReader.
func WithCodec ¶
WithCodec sets the compression codec, overriding the default of null (no compression). It is both a WriterOpt and a ReaderOpt. You do not need to register the four built-in codecs (null, deflate, snappy, zstandard) to read them. A custom codec whose name matches a built-in overrides it.
We take ownership of the codec you pass and close it exactly once: in Writer.Close or Reader.Close when the constructor succeeds, or in the constructor itself when it fails or does not use the codec. Note that the last case is easy to hit: NewReader and NewAppendWriter only use a codec whose Name matches the header's avro.codec, NewWriter only uses the last WithCodec you pass, and we close every codec we do not use.
If you share one codec across several writers, readers, or files, give it a Close that returns nil, or wrap it in NopCloser.
We ignore a nil codec, whether a nil Codec or a non-nil Codec holding a nil pointer, on every constructor: we never name, adopt, or close it, and the constructor behaves as though you had not passed it. If it is the only codec supplied for a name the file uses, the constructor reports an unknown codec.
For reader-side decompression bounding, see WithMaxDecompressedBlockBytes. It applies to any codec you supply that implements BoundedDecompressor, which every built-in does, including one wrapped by NopCloser.
func WithSchemaOpts ¶ added in v1.4.0
WithSchemaOpts passes avro.SchemaOpt values, such as avro.CustomType or avro.WithLaxNames, to the avro.Parse of the header's embedded schema. NewReader uses it to register custom type conversions and to accept lax-named header schemas. NewAppendWriter needs it whenever the header schema requires an option to parse at all. NewWriter ignores it: you already parsed its schema.
type Reader ¶
type Reader struct {
// contains filtered or unexported fields
}
Reader decodes Avro values from an OCF.
func NewReader ¶
NewReader creates a Reader that decodes an OCF from r. We read the header immediately. Use WithCodec if the file uses a non-built-in codec.
Example (Evolution) ¶
package main
import (
"bytes"
"fmt"
"log"
"github.com/twmb/avro"
"github.com/twmb/avro/ocf"
)
func main() {
// Write v1 data (name only).
v1Schema := avro.MustParse(`{
"type": "record", "name": "User",
"fields": [{"name": "name", "type": "string"}]
}`)
var buf bytes.Buffer
w, err := ocf.NewWriter(&buf, v1Schema)
if err != nil {
log.Fatal(err)
}
for _, name := range []string{"Alice", "Bob"} {
if err := w.Encode(map[string]any{"name": name}); err != nil {
log.Fatal(err)
}
}
if err := w.Close(); err != nil {
log.Fatal(err)
}
// Read with a v2 schema that added an age field with a default.
v2Schema := avro.MustParse(`{
"type": "record", "name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "age", "type": "int", "default": 0}
]
}`)
type User struct {
Name string `avro:"name"`
Age int32 `avro:"age"`
}
r, err := ocf.NewReader(bytes.NewReader(buf.Bytes()), ocf.WithReaderSchema(v2Schema))
if err != nil {
log.Fatal(err)
}
defer r.Close()
for {
var u User
if err := r.Decode(&u); err != nil {
break
}
fmt.Printf("%s age=%d\n", u.Name, u.Age)
}
}
Output: Alice age=0 Bob age=0
func (*Reader) Close ¶
Close closes the codec, releasing what it holds. Close is idempotent: later calls return nil without re-closing. After Close, Decode returns an error.
func (*Reader) Decode ¶
Decode reads the next datum into v, returning io.EOF at end of file. We return io.EOF only at a clean end of stream, where the file ends exactly at a block boundary. A stream truncated mid-block, with a promised block header, data, or sync marker cut short, returns an error matching io.ErrUnexpectedEOF instead, never one matching io.EOF.
type ReaderOpt ¶
type ReaderOpt interface {
// contains filtered or unexported methods
}
ReaderOpt is an option for NewReader.
func WithDecodeOpts ¶ added in v1.9.0
WithDecodeOpts passes avro.Opt values to the avro.Schema.Decode behind every Reader.Decode, overriding the default of no options. Without it you cannot use avro.TaggedUnions or avro.TagLogicalTypes, which change what a union decodes to in an *any target. Repeated calls are cumulative. NewWriter and NewAppendWriter ignore it.
We drop any option that would make decoded values point into the decode input, such as avro.AliasInput. A Reader decodes out of a block buffer it owns, and we do not promise that buffer outlives the next read.
func WithMaxBlockBytes ¶
WithMaxBlockBytes sets the maximum compressed block size in bytes we accept when reading, overriding the default of 64 MiB. It guards against malicious or corrupt files that declare very large blocks.
func WithMaxDecompressedBlockBytes ¶ added in v1.8.0
WithMaxDecompressedBlockBytes sets the maximum decompressed size in bytes of a single block we accept when reading, overriding the default of 64 MiB. WithMaxBlockBytes bounds the compressed size we read off the wire; this bounds what that block inflates to, guarding against "zip bomb" inputs where a tiny compressed block expands to a huge output. A block's record count is bounded by its decompressed length, so this also bounds the per-block decode loop. Raise it if you write blocks (via WithBlockBytes) that decompress beyond the default.
We pass the limit to the codec's BoundedDecompressor.DecompressBounded, which refuses an over-cap block before allocating it. This applies to every codec implementing that interface: one we resolve by name from the file header, and one you supply via WithCodec, including one wrapped by NopCloser. All four built-in codecs implement it. A codec that does not decompresses unbounded, and there is no post-decompression check, since the allocation would already have happened. Supply a self-bounding codec for untrusted data.
func WithReaderSchema ¶
WithReaderSchema gives us a reader schema to resolve the file's writer schema against via avro.Resolve. Reader.Decode then uses the resolved schema. Fields you add in the reader schema must have defaults, and we skip writer fields your reader schema omits.
Use WithReaderSchemaFunc if you must pick the reader schema from the file's header. You can use at most one of the two.
func WithReaderSchemaFunc ¶ added in v1.6.0
WithReaderSchemaFunc is the dynamic counterpart to WithReaderSchema. We call fn from NewReader once the OCF header is parsed, so it can inspect the file's writer schema and metadata through rd.Schema() and rd.Metadata() before choosing what to resolve against.
A non-nil schema is resolved against the writer schema via avro.Resolve, and Reader.Decode then uses the result. (nil, nil) means no resolution: records decode against the writer schema directly, exactly as if you passed no reader-schema option at all. A non-nil error is returned from NewReader.
Your fn must not call rd.Decode or rd.Close; rd is valid only for read-only header inspection during the call. You can use at most one of WithReaderSchema and WithReaderSchemaFunc.
Example ¶
ExampleWithReaderSchemaFunc demonstrates choosing the reader schema based on state that is only available after the OCF header is parsed: a metadata key that distinguishes old from new file variants, say, or a writer-schema shape that changed between versions of the producer. The callback runs after NewReader has read the header, so rd.Schema() and rd.Metadata() are populated. Whatever schema it returns becomes the reader schema, resolved against the writer schema.
package main
import (
"bytes"
"fmt"
"log"
"github.com/twmb/avro"
"github.com/twmb/avro/ocf"
)
func main() {
// Producer v1 wrote records with a legacy field name:
v1Schema := avro.MustParse(`{
"type": "record", "name": "Event",
"fields": [{"name": "legacy_ts", "type": "long"}]
}`)
var buf bytes.Buffer
w, err := ocf.NewWriter(&buf, v1Schema,
ocf.WithMetadata(map[string][]byte{"producer-version": []byte("1")}))
if err != nil {
log.Fatal(err)
}
if err := w.Encode(map[string]any{"legacy_ts": int64(1700000000)}); err != nil {
log.Fatal(err)
}
if err := w.Close(); err != nil {
log.Fatal(err)
}
// We read with two reader schemas, one per producer version. Each uses
// the spec-correct field name "ts" but declares the old name as an
// alias, so records from either version decode into the same struct
// without coalescing.
v1Reader := avro.MustParse(`{
"type": "record", "name": "Event",
"fields": [{"name": "ts", "type": "long", "aliases": ["legacy_ts"]}]
}`)
v2Reader := avro.MustParse(`{
"type": "record", "name": "Event",
"fields": [{"name": "ts", "type": "long"}]
}`)
type Event struct {
TS int64 `avro:"ts"`
}
r, err := ocf.NewReader(bytes.NewReader(buf.Bytes()),
ocf.WithReaderSchemaFunc(func(rd *ocf.Reader) (*avro.Schema, error) {
// Header has been parsed. Pick the reader schema based on
// whichever producer wrote the file.
if string(rd.Metadata()["producer-version"]) == "1" {
return v1Reader, nil
}
return v2Reader, nil
}))
if err != nil {
log.Fatal(err)
}
defer r.Close()
var e Event
if err := r.Decode(&e); err != nil {
log.Fatal(err)
}
fmt.Printf("ts=%d\n", e.TS)
}
Output: ts=1700000000
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer encodes Avro values into an OCF. We buffer values into blocks and compress and flush them for you. You must Close to flush what remains.
func NewAppendWriter ¶
func NewAppendWriter(rws io.ReadWriteSeeker, opts ...WriterOpt) (*Writer, error)
NewAppendWriter opens an existing OCF for appending. We read the header to recover the schema, codec, and sync marker, then seek to the end.
We honor WithBlockCount and WithBlockBytes. WithCodec supplies an implementation for a non-built-in codec, matched by name against the header. WithSchemaOpts applies to the header-schema parse, which you need whenever that schema requires an option to parse at all (say avro.WithLaxNames for a file written with a lax-named schema). We ignore WithSchema, WithSyncMarker, and WithMetadata: the header is already on disk and we never rewrite it, so the schema, sync marker, and metadata always come from the existing file. Java's DataFileWriter.appendTo and fastavro's append mode behave the same. We likewise ignore any other option.
func NewWriter ¶
NewWriter creates a Writer that writes an OCF to w. We write the file header immediately.
Example ¶
package main
import (
"bytes"
"fmt"
"log"
"github.com/twmb/avro"
"github.com/twmb/avro/ocf"
)
func main() {
schema := avro.MustParse(`{
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "age", "type": "int"}
]
}`)
type User struct {
Name string `avro:"name"`
Age int32 `avro:"age"`
}
var buf bytes.Buffer
w, err := ocf.NewWriter(&buf, schema)
if err != nil {
log.Fatal(err)
}
for _, u := range []User{
{Name: "Alice", Age: 30},
{Name: "Bob", Age: 25},
} {
if err := w.Encode(&u); err != nil {
log.Fatal(err)
}
}
if err := w.Close(); err != nil {
log.Fatal(err)
}
// Read back.
r, err := ocf.NewReader(bytes.NewReader(buf.Bytes()))
if err != nil {
log.Fatal(err)
}
defer r.Close()
for {
var u User
if err := r.Decode(&u); err != nil {
break
}
fmt.Printf("%s is %d\n", u.Name, u.Age)
}
}
Output: Alice is 30 Bob is 25
func (*Writer) Close ¶
Close flushes any remaining items and closes the codec. We close the codec even when the writer is poisoned, because zstd and similar codecs hold goroutines and buffers that must be released.
Close is idempotent: later calls return nil without re-closing the codec. After Close, Encode, Write, Flush, and Reset all return an error rather than silently extending the file.
func (*Writer) Encode ¶
Encode serializes v and appends it to the current block. We flush the block when it hits the count or byte limit, or when a run of zero-byte datums reaches the Reader's per-block bound, so a file of zero-byte datums ("null", all-null records, size-0 fixed) always reads back.
If v does not fit the schema, we discard that datum and the Writer remains usable: we had only appended it to the in-memory block, never the file, so datums we already accepted are intact and still flush.
After an I/O or compression error, where we cannot know the sink's state, we poison the Writer: every subsequent call returns the same error.
func (*Writer) Reset ¶
Reset flushes buffered items to the current destination, then starts a new OCF on dst with the original schema, codec, and options. If the Writer is in an error state we skip the flush and clear the error. Reset errors if you already closed the Writer, since its codec is no longer usable.
If Reset fails after switching to dst, on either a sync-marker generation error or a header write error, we poison the Writer exactly as a failed Writer.Encode or Writer.Flush does: every subsequent Encode, Flush, and Close returns the error until a later successful Reset clears it. (A failed flush of the old destination also poisons.) Otherwise, ignoring Reset's error and writing on would emit a headerless byte stream onto dst.
type WriterOpt ¶
type WriterOpt interface {
// contains filtered or unexported methods
}
WriterOpt is an option for NewWriter.
func WithBlockBytes ¶
WithBlockBytes caps a block's uncompressed bytes, overriding the default of 64 KiB. We flush a block when either this or WithBlockCount is reached.
func WithBlockCount ¶
WithBlockCount caps the number of items per block, overriding the default of 0 (unlimited). We flush a block when either this or WithBlockBytes is reached.
func WithMetadata ¶
WithMetadata adds custom metadata to the file header. The spec reserves keys starting with "avro.". Repeated calls are cumulative.
func WithSchema ¶
WithSchema sets the schema JSON we write to the file header, overriding the default of avro.Schema.String: the original JSON you passed to avro.Parse, with logicalType, precision, scale, doc, aliases, and default all preserved, matching Java's DataFileWriter and fastavro. Use this if you want different header text, say the Parsing Canonical Form from avro.Schema.Canonical.
func WithSyncMarker ¶
WithSyncMarker sets the 16-byte sync marker between blocks, overriding the random one we generate. Mostly useful for deterministic test output.
Source Files