ocf

package
v2.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package ocf implements encoding and decoding of Avro Object Container Files as defined by the Avro specification.

See the Avro specification for an understanding of Avro: http://avro.apache.org/docs/current/

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (

	// HeaderSchema is the Avro schema of a container file header.
	HeaderSchema = avro.MustParse(`{
	"type": "record",
	"name": "org.apache.avro.file.Header",
	"fields": [
		{"name": "magic", "type": {"type": "fixed", "name": "Magic", "size": 4}},
		{"name": "meta", "type": {"type": "map", "values": "bytes"}},
		{"name": "sync", "type": {"type": "fixed", "name": "Sync", "size": 16}}
	]
}`)

	// DefaultSchemaMarshaler calls the schema's String() method, to produce
	// a "canonical" schema.
	DefaultSchemaMarshaler = defaultMarshalSchema
	// FullSchemaMarshaler calls the schema's MarshalJSON() method, to produce
	// a schema with all details preserved. The "canonical" schema returned by
	// the default marshaler does not preserve a type's extra properties.
	FullSchemaMarshaler = fullMarshalSchema
)

Functions

This section is empty.

Types

type Codec 

type Codec interface {
	// Decode decodes the given bytes.
	Decode([]byte) ([]byte, error)
	// Encode encodes the given bytes.
	Encode([]byte) []byte
}

Codec represents a compression codec.

type CodecName 

type CodecName string

CodecName represents a compression codec name. 

const (
	Null      CodecName = "null"
	Deflate   CodecName = "deflate"
	Snappy    CodecName = "snappy"
	ZStandard CodecName = "zstandard"
)

Supported compression codecs.

type Decoder 

type Decoder struct {
	// contains filtered or unexported fields
}

Decoder reads and decodes Avro values from a container file.

func NewDecoder 

func NewDecoder(r io.Reader, opts ...DecoderFunc) (*Decoder, error)

NewDecoder returns a new decoder that reads from reader r.

Example 
package main

import (
	"log"
	"os"

	"github.com/arquivei/avro/v2/ocf"
)

func main() {
	type SimpleRecord struct {
		A int64  `avro:"a"`
		B string `avro:"b"`
	}

	f, err := os.Open("/your/avro/file.avro")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()

	dec, err := ocf.NewDecoder(f)
	if err != nil {
		log.Fatal(err)
	}

	for dec.HasNext() {
		var record SimpleRecord
		err = dec.Decode(&record)
		if err != nil {
			log.Fatal(err)
		}

		// Do something with the data
	}

	if err := dec.Error(); err != nil {
		log.Fatal(err)
	}
}

func (*Decoder) Close 

func (d *Decoder) Close() error

Close releases codec resources.

func (*Decoder) Decode 

func (d *Decoder) Decode(v any) error

Decode reads the next Avro encoded value from its input and stores it in the value pointed to by v.

func (*Decoder) Error 

func (d *Decoder) Error() error

Error returns the last reader error.

func (*Decoder) HasNext 

func (d *Decoder) HasNext() bool

HasNext determines if there is another value to read.

func (*Decoder) Metadata 

func (d *Decoder) Metadata() map[string][]byte

Metadata returns the header metadata.

func (*Decoder) Schema 

func (d *Decoder) Schema() avro.Schema

Schema returns the schema that was parsed from the file's metadata and that is used to interpret the file's contents.

type DecoderFunc 

type DecoderFunc func(cfg *decoderConfig)

DecoderFunc represents a configuration function for Decoder.

func WithDecoderConfig 

func WithDecoderConfig(wCfg avro.API) DecoderFunc

WithDecoderConfig sets the value decoder config on the OCF decoder.

func WithDecoderSchemaCache 

func WithDecoderSchemaCache(cache *avro.SchemaCache) DecoderFunc

WithDecoderSchemaCache sets the schema cache for the decoder. If not specified, defaults to avro.DefaultSchemaCache.

func WithMaxBlockSize 

func WithMaxBlockSize(size int) DecoderFunc

WithMaxBlockSize sets the maximum size, in bytes, that a single OCF block's declared (pre-decompression) data may have, and the maximum decompressed size the compression codec may produce for that block. Reading a block over that size returns an error instead of allocating memory for it.

This guards against files that declare an implausibly large or negative block size, and against decompression-bomb blocks (a small compressed payload that expands to a huge decompressed size).

Defaults to 100 MiB. A value <= 0 disables the limit - not recommended for untrusted input. See WithZStandardDecoder for a caveat when reusing a pre-created ZStandard decoder.

func WithZStandardDecoder 

func WithZStandardDecoder(dec *zstd.Decoder) DecoderFunc

WithZStandardDecoder sets a pre-created ZStandard decoder to be reused. This allows sharing a single decoder across multiple OCF decoders for efficiency. The caller is responsible for closing the decoder after all OCF decoders are done.

Because the decoder is created outside of this package, WithMaxBlockSize's decompressed-size limit cannot be enforced for ZStandard when this option is used; only the pre-decompression block-size check still applies. Size the shared decoder's own options (e.g. zstd.WithDecoderMaxMemory) accordingly if it will process untrusted input.

func WithZStandardDecoderOptions 

func WithZStandardDecoderOptions(opts ...zstd.DOption) DecoderFunc

WithZStandardDecoderOptions sets the options for the ZStandard decoder.

type DeflateCodec 

type DeflateCodec struct {
	// contains filtered or unexported fields
}

DeflateCodec is a flate compression codec.

func (*DeflateCodec) Decode 

func (c *DeflateCodec) Decode(b []byte) ([]byte, error)

Decode decodes the given bytes.

func (*DeflateCodec) Encode 

func (c *DeflateCodec) Encode(b []byte) []byte

Encode encodes the given bytes.

type Encoder 

type Encoder struct {
	// contains filtered or unexported fields
}

Encoder writes Avro container file to an output stream.

func NewEncoder 

func NewEncoder(s string, w io.Writer, opts ...EncoderFunc) (*Encoder, error)

NewEncoder returns a new encoder that writes to w using schema s.

If the writer is an existing ocf file, it will append data using the existing schema.

Example 
package main

import (
	"log"
	"os"

	"github.com/arquivei/avro/v2/ocf"
)

func main() {
	schema := `{
	    "type": "record",
	    "name": "simple",
	    "namespace": "org.hamba.avro",
	    "fields" : [
	        {"name": "a", "type": "long"},
	        {"name": "b", "type": "string"}
	    ]
	}`

	type SimpleRecord struct {
		A int64  `avro:"a"`
		B string `avro:"b"`
	}

	f, err := os.Open("/your/avro/file.avro")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()

	enc, err := ocf.NewEncoder(schema, f)
	if err != nil {
		log.Fatal(err)
	}

	var record SimpleRecord
	err = enc.Encode(record)
	if err != nil {
		log.Fatal(err)
	}

	if err := enc.Flush(); err != nil {
		log.Fatal(err)
	}

	if err := f.Sync(); err != nil {
		log.Fatal(err)
	}
}

func NewEncoderWithSchema 

func NewEncoderWithSchema(schema avro.Schema, w io.Writer, opts ...EncoderFunc) (*Encoder, error)

NewEncoderWithSchema returns a new encoder that writes to w using schema s.

If the writer is an existing ocf file, it will append data using the existing schema.

func (*Encoder) Close 

func (e *Encoder) Close() error

Close closes the encoder, flushing the writer and releasing codec resources.

func (*Encoder) Encode 

func (e *Encoder) Encode(v any) error

Encode writes the Avro encoding of v to the stream.

func (*Encoder) Flush 

func (e *Encoder) Flush() error

Flush flushes the underlying writer.

func (*Encoder) Reset 

func (e *Encoder) Reset(w io.Writer) error

Reset flushes any pending data, resets the encoder to write to a new io.Writer, and writes a fresh header with a new sync marker. The schema, codec, and other settings are preserved from the original encoder. This allows reusing the encoder for multiple files without reallocating buffers.

func (*Encoder) Write 

func (e *Encoder) Write(p []byte) (n int, err error)

Write v to the internal buffer. This method skips the internal encoder and therefore the caller is responsible for encoding the bytes. No error will be thrown if the bytes does not conform to the schema given to NewEncoder, but the final ocf data will be corrupted.

type EncoderFunc 

type EncoderFunc func(cfg *encoderConfig)

EncoderFunc represents a configuration function for Encoder.

func WithBlockLength 

func WithBlockLength(length int) EncoderFunc

WithBlockLength sets the block length on the encoder.

func WithBlockSize 

func WithBlockSize(size int) EncoderFunc

WithBlockSize sets the maximum uncompressed size of a buffered block before it is written and flushed to the underlying io.Writer (after compression).

func WithCodec 

func WithCodec(codec CodecName) EncoderFunc

WithCodec sets the compression codec on the encoder.

func WithCompressionLevel 

func WithCompressionLevel(compLvl int) EncoderFunc

WithCompressionLevel sets the compression codec to deflate and the compression level on the encoder.

func WithEncoderSchemaCache 

func WithEncoderSchemaCache(cache *avro.SchemaCache) EncoderFunc

WithEncoderSchemaCache sets the schema cache for the encoder. If not specified, defaults to avro.DefaultSchemaCache.

func WithEncodingConfig 

func WithEncodingConfig(wCfg avro.API) EncoderFunc

WithEncodingConfig sets the value encoder config on the OCF encoder.

func WithMetadata 

func WithMetadata(meta map[string][]byte) EncoderFunc

WithMetadata sets the metadata on the encoder header.

func WithMetadataKeyVal 

func WithMetadataKeyVal(key string, val []byte) EncoderFunc

WithMetadataKeyVal sets a single key-value pair for the metadata on the encoder header.

func WithSchemaMarshaler 

func WithSchemaMarshaler(m func(avro.Schema) ([]byte, error)) EncoderFunc

WithSchemaMarshaler sets the schema marshaler for the encoder. If not specified, defaults to DefaultSchemaMarshaler.

func WithSyncBlock 

func WithSyncBlock(sync [16]byte) EncoderFunc

WithSyncBlock sets the sync block.

func WithZStandardEncoder 

func WithZStandardEncoder(enc *zstd.Encoder) EncoderFunc

WithZStandardEncoder sets a pre-created ZStandard encoder to be reused. This allows sharing a single encoder across multiple OCF encoders for efficiency. The caller is responsible for closing the encoder after all OCF encoders are done.

func WithZStandardEncoderOptions 

func WithZStandardEncoderOptions(opts ...zstd.EOption) EncoderFunc

WithZStandardEncoderOptions sets the options for the ZStandard encoder. 

type Header struct {
	Magic [4]byte           `avro:"magic"`
	Meta  map[string][]byte `avro:"meta"`
	Sync  [16]byte          `avro:"sync"`
}

Header represents an Avro container file header.

type NullCodec 

type NullCodec struct{}

NullCodec is a no op codec.

func (*NullCodec) Decode 

func (*NullCodec) Decode(b []byte) ([]byte, error)

Decode decodes the given bytes.

func (*NullCodec) Encode 

func (*NullCodec) Encode(b []byte) []byte

Encode encodes the given bytes.

type SnappyCodec 

type SnappyCodec struct {
	// contains filtered or unexported fields
}

SnappyCodec is a snappy compression codec.

func (*SnappyCodec) Decode 

func (c *SnappyCodec) Decode(b []byte) ([]byte, error)

Decode decodes the given bytes.

func (*SnappyCodec) Encode 

func (*SnappyCodec) Encode(b []byte) []byte

Encode encodes the given bytes.

type ZStandardCodec 

type ZStandardCodec struct {
	// contains filtered or unexported fields
}

ZStandardCodec is a zstandard compression codec.

func (*ZStandardCodec) Close 

func (zstdCodec *ZStandardCodec) Close() error

Close closes the zstandard encoder and decoder, releasing resources. Shared instances (provided via WithZStandardEncoder/WithZStandardDecoder) are not closed.

func (*ZStandardCodec) Decode 

func (zstdCodec *ZStandardCodec) Decode(b []byte) ([]byte, error)

Decode decodes the given bytes.

func (*ZStandardCodec) Encode 

func (zstdCodec *ZStandardCodec) Encode(b []byte) []byte

Encode encodes the given bytes.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.