Documentation
¶
Overview ¶
Package ocf implements Avro Object Container Files (OCF).
An OCF is a self-describing binary file format: it embeds the Avro schema in the file header so readers do not need out-of-band schema information. Data is stored in compressed blocks separated by sync markers, making files splittable for parallel processing. OCF is the standard format for storing Avro data on disk; for sending 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. Custom codecs can be provided via WithCodec.
Block size limits ¶
The reader caps both the compressed block it reads off the wire (WithMaxBlockBytes) and the size that block decompresses to (WithMaxDecompressedBlockBytes), each defaulting to 64 MiB, to bound memory and decode time on untrusted input. The writer has no such cap (matching Java's DataFileWriter and fastavro): it writes whatever blocks it is given.
A single Avro datum cannot be split across blocks, so a value larger than the reader default — e.g. an 80 MiB blob — is written as one block that a default reader then refuses, with an error naming the option to raise. The caps are a reader-side defense, so they live on the reader; to read a file whose blocks exceed the default, raise the matching cap there:
r, err := ocf.NewReader(f, ocf.WithMaxDecompressedBlockBytes(128<<20))
Configure the cap to match the largest block your writer produces (which, for single large values, is governed by 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 capability a Codec may implement. When it does, the Reader calls DecompressBounded with its per-block cap (from WithMaxDecompressedBlockBytes) instead of Codec.Decompress, letting the codec refuse early or stream-limit BEFORE allocating the block. That is the only effective defense against a decompression bomb; checking the size afterward is false comfort, because the allocation already happened. A Codec without it decompresses unbounded, so for untrusted data supply one that bounds itself. All built-in codecs do.
A wrapper that embeds Codec, such as a NopCloser result, does NOT inherit the capability — embedding an interface promotes only that interface's methods — so it must forward DecompressBounded explicitly or it silently disables bounding for the codec it wraps.
max <= 0 means no limit. max is constant across all calls for a given Reader, so a codec that caches a configured decoder (e.g. 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 any resources held by the codec. Codecs that hold no
// resources may 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 but has a no-op Close method, so that an individual Writer.Close or Reader.Close — or a constructor that fails and releases the codec it was handed — does not release resources shared with another writer or reader. The caller is responsible for closing the underlying codec when it is no longer needed.
If c implements BoundedDecompressor, so does the returned Codec (the reader's decompression bound is forwarded to c), so wrapping a built-in codec for sharing does not silently drop its bounding.
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. Encoder options (eopts) and decoder options (dopts) are passed to zstd.NewWriter and zstd.NewReader respectively. Both may be nil for defaults.
zstd.WithEncoderConcurrency(1) and zstd.WithDecoderConcurrency(1) are prepended to the options; pass a different concurrency to override.
A single ZstdCodec is safe to share across multiple readers and writers via NopCloser.
ZstdCodec implements BoundedDecompressor, so a reader applies its WithMaxDecompressedBlockBytes cap to a supplied ZstdCodec the same as a name-resolved one: the decoder is built lazily on first read with zstd.WithDecoderMaxMemory set from the cap. The decoder is then cached, so a ZstdCodec shared across readers with different caps honors the first reader's cap (set zstd.WithDecoderMaxMemory in dopts to pin a specific bound regardless of reader). A cap below zstd.MinWindowSize (1 KiB) is raised up to it — the smallest window the decoder accepts; a smaller limit would reject every frame, since each frame's window is at least MinWindowSize. At or above that bound a sub-MiB cap is honored exactly.
type Opt ¶
Opt is an option that applies to both NewWriter and NewReader.
func WithCodec ¶
WithCodec sets the compression codec. The default is null (no compression). WithCodec can be used as both a WriterOpt and a ReaderOpt. The four built-in codecs (null, deflate, snappy, zstandard) do not need to be registered for reading. A custom codec whose name matches a built-in overrides it.
Passing a codec hands it over, and it is closed exactly once whatever happens: by Writer.Close or Reader.Close when the constructor returns a usable one, by the constructor when it fails, and by the constructor when it succeeds without using the codec at all. That last case is easy to reach and gives no sign — NewReader and NewAppendWriter take a codec only when its Name matches the header's avro.codec, and NewWriter takes only the last WithCodec written — so a declined offer is released rather than dropped.
A caller sharing one codec across several writers, readers, or files must therefore give it a Close that returns nil, or wrap it in NopCloser. The rule holds whether or not the offer was taken.
A nil codec is ignored, on every constructor, in both spellings: a nil Codec, and a non-nil Codec holding a nil pointer. Such an offer is never named, never adopted, and never closed; the constructor behaves as though it were not written. It is still a caller mistake, and if it is the ONLY offer for a codec the file names, the constructor reports an unknown codec rather than silently reading nothing.
For reader-side decompression bounding of a supplied codec, see WithMaxDecompressedBlockBytes: it reaches any supplied codec implementing BoundedDecompressor (every built-in does, including one wrapped by NopCloser). A custom codec that does not implement BoundedDecompressor decompresses unbounded — supply one that bounds itself for untrusted data.
func WithSchemaOpts ¶ added in v1.4.0
WithSchemaOpts passes avro.SchemaOpt values (such as avro.CustomType or avro.WithLaxNames) to the avro.Parse call that parses the file 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: its schema is already parsed by the caller.
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. The header is read 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 any resources it holds. Close is idempotent: subsequent calls return nil without re-closing the codec. After Close, Decode returns an error.
func (*Reader) Decode ¶
Decode reads the next datum into v, returning io.EOF at end of file. io.EOF is returned only at a clean end of stream — the file ends exactly at a block boundary; a stream truncated mid-block (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 WithMaxBlockBytes ¶
WithMaxBlockBytes sets the maximum compressed block size in bytes that the reader will accept. The default is 64 MiB. This 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 that the reader will accept. The default is 64 MiB. WithMaxBlockBytes bounds the compressed size read off the wire; this bounds what that compressed block inflates to, guarding against decompression-amplification ("zip bomb") inputs where a tiny compressed block declares or expands to a huge output. Because a block's record count is bounded by its decompressed length, this also bounds the per-block decode loop. Pass a larger value if you legitimately write blocks (via WithBlockBytes) that decompress beyond the default.
Enforcement scope: this limit is applied UP FRONT (the over-cap allocation is prevented, not merely caught afterward) by passing it to the codec's BoundedDecompressor.DecompressBounded at decode time. It therefore reaches every codec implementing that capability uniformly — a codec resolved by name from the file header (the common case), AND a codec supplied as an instance via WithCodec, including one wrapped by NopCloser. All four built-in codecs (null, deflate, snappy, zstandard) implement it. A custom codec that does NOT implement BoundedDecompressor decompresses unbounded — this limit does not apply to it (no post-decompression backstop; that is false comfort once the allocation has happened), so supply a self-bounding codec for untrusted data.
func WithReaderSchema ¶
WithReaderSchema provides the reader schema to resolve the file's writer schema against via avro.Resolve. Subsequent Reader.Decode calls use the resolved schema. Fields added in the reader schema must have defaults; writer fields absent from the reader schema are skipped.
Use WithReaderSchemaFunc when the reader schema must be chosen based on the file's header (metadata or writer-schema shape).
At most one of WithReaderSchema and WithReaderSchemaFunc may be used.
func WithReaderSchemaFunc ¶ added in v1.6.0
WithReaderSchemaFunc is the dynamic counterpart to WithReaderSchema. The callback is invoked by NewReader after the OCF header has been parsed, so it can inspect the file's writer schema and metadata via rd.Schema() and rd.Metadata() before deciding which reader schema to resolve against.
If the callback returns a non-nil schema, the writer schema is resolved against it via avro.Resolve and subsequent Reader.Decode calls use the resolved schema.
If the callback returns (nil, nil), no resolution is performed and records decode against the writer schema directly — equivalent to not passing any reader-schema option at all.
If the callback returns a non-nil error, NewReader returns that error.
The callback must not call rd.Decode or rd.Close; rd is only valid for read-only header inspection during the callback.
At most one of WithReaderSchema and WithReaderSchemaFunc may be used.
Example ¶
ExampleWithReaderSchemaFunc demonstrates choosing the reader schema based on state that's only available after the OCF header is parsed — for example, a metadata key that distinguishes between old and new file variants, 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 for resolution 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)
}
// Our application reads with two reader schemas — one per producer
// version — each using the spec-correct field name "ts" but declaring
// 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. Values are buffered into blocks that are compressed and flushed automatically. Close must be called to flush remaining items.
func NewAppendWriter ¶
func NewAppendWriter(rws io.ReadWriteSeeker, opts ...WriterOpt) (*Writer, error)
NewAppendWriter opens an existing OCF for appending. It reads the header to recover the schema, codec, and sync marker, then seeks to the end.
WithBlockCount and WithBlockBytes are honored. WithCodec can provide a codec implementation for non-built-in codecs (matched by name against the header). WithSchemaOpts passes schema options to the header-schema parse — required when the header schema needs an option to parse at all (e.g. avro.WithLaxNames for a file written with a lax-named schema). WithSchema, WithSyncMarker, and WithMetadata are ignored: the header is already on disk and is never rewritten, so the schema, sync marker, and metadata always come from the existing file. (Reference implementations behave the same on append — neither Java's DataFileWriter.appendTo nor fastavro's append mode lands new metadata in the file.) Any remaining options are likewise ignored.
func NewWriter ¶
NewWriter creates a Writer that writes an OCF to w. The file header is written 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. The codec is closed even if the writer is in a poisoned state — zstd and similar codecs hold goroutines and buffers whose lifetime must be bounded. (Deliberately more careful than Java: DataFileWriter.close is a plain flush-then-close sequence with no finally, so a failing flush skips its close — DataFileWriter.java:483-489.)
Close is idempotent: subsequent 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. The block is flushed automatically when it hits the count or byte limit, or when a run of zero-byte datums reaches the Reader's per-block bound (so files of zero-byte datums — "null", all-null records, size-0 fixed — are always readable back).
A value error (v does not fit the schema) discards the failed datum and leaves the Writer usable: the datum was only ever appended to the in-memory block buffer, never the file, so previously accepted datums are intact and continue to flush.
After an I/O or compression error — where the sink's state is not knowable — the Writer is poisoned: all subsequent calls return the same error.
func (*Writer) Reset ¶
Reset flushes buffered items to the current destination, then starts a new OCF on dst reusing the original schema, codec, and options. If the Writer is in an error state the flush is skipped and the error is cleared. Reset returns an error if the Writer has been closed — its codec is no longer usable.
If Reset fails after it has repointed to dst — a sync-marker generation error or a header write error — the Writer is poisoned exactly as a failed Writer.Encode or Writer.Flush is: every subsequent Encode/Flush/Close returns the sticky error until a later successful Reset clears it. (The flush of the OLD destination, which runs before the repoint, also poisons on failure.) Without this a caller that ignores Reset's returned error and keeps writing would emit a silent headerless byte stream onto dst.
type WriterOpt ¶
type WriterOpt interface {
// contains filtered or unexported methods
}
WriterOpt is an option for NewWriter.
func WithBlockBytes ¶
WithBlockBytes sets the maximum uncompressed size of a block in bytes before it is flushed. The default is 64 KiB. If both WithBlockCount and WithBlockBytes are set, whichever limit is hit first triggers a flush.
func WithBlockCount ¶
WithBlockCount sets the maximum number of items per block. The default is 0 (unlimited). If both WithBlockCount and WithBlockBytes are set, whichever limit is hit first triggers a flush.
func WithMetadata ¶
WithMetadata adds custom metadata to the file header. Keys starting with "avro." are reserved by the spec. Multiple calls are cumulative.
func WithSchema ¶
WithSchema overrides the schema JSON written to the file header. By default avro.Schema.String is used (the original JSON passed to avro.Parse with all properties preserved — logicalType, precision, scale, doc, aliases, default — matching Java's DataFileWriter and fastavro). Use this only to write a deliberately-different schema text (e.g. the Parsing Canonical Form via avro.Schema.Canonical for strict-PCF downstream consumers).
func WithSyncMarker ¶
WithSyncMarker sets the 16-byte sync marker written between blocks. By default a random marker is generated. This is primarily useful for deterministic test output.
Source Files