magmalake avro

avro · API

avro.datafile

Avro Object Container Files — the reader and the writer.

An OCF is Obj\x01, a map<bytes> of file metadata (carrying at least avro.schema, and usually avro.codec), a 16-byte sync marker, and then a run of blocks, each count · size · data · sync.

Apache Iceberg stores its own keys in that same metadata map — schema, partition-spec, format-version, content — so the metadata is exposed both as raw bytes (metadata_keys / metadata_vals) and as text (metadata_string), alongside the parsed avro.schema.

The metadata is kept as two parallel lists rather than a Dict, in the order the file wrote them. A file has a handful of entries, so a linear scan over StringSlices beats hashing — and, unlike a Dict, looking one up does not have to build a String first. metadata_map() still hands back a Dict for callers that want one.

Structs

DataFileReader

struct DataFileReader[C: CodecSet = DefaultCodecs]

Reads an Avro Object Container File held in memory.

Implements Copyable, Movable

Fields
data: List[UInt8]
schema: Schema: The parsed avro.schema. Empty until ensure_schema() when the reader was opened with defer_schema=True.
schema_ready: Bool: False only between a defer_schema=True open and ensure_schema().
metadata_keys: List[String]: Metadata keys in the order the file wrote them.
metadata_vals: List[List[UInt8]]: Metadata values, parallel to metadata_keys.
codec: String
sync_marker: List[UInt8]
pos: Int: Offset of the next block header in data.
block: List[UInt8]
block_pos: Int
block_left: Int
block_index: Int

__init__

def __init__(out self, var data: List[UInt8], *, defer_schema: Bool = False)

Read the header. defer_schema skips parsing the avro.schema JSON, which is the expensive half of opening a small file.

It is for a caller that already holds a compiled DecodePlan for this exact schema — see avro.cursor.PlanCache. Nothing else changes: the schema is parsed on demand the moment anything needs it, so next() and read_all() behave the same either way.

open

def open(path: StringSpan) -> Self

Read the whole file, then parse its header.

from_bytes

def from_bytes(data: Span[UInt8]) -> Self

ensure_schema

def ensure_schema(mut self)

Parse avro.schema if a defer_schema=True open left it unparsed.

metadata_index

def metadata_index(self, key: StringSpan) -> Int

Where key sits in metadata_keys/metadata_vals, or -1.

Span(reader.metadata_vals[i]) then reads the entry's raw bytes without copying them.

A linear scan, and deliberately so: a file has a handful of metadata entries, and this compares StringSlices rather than allocating a String to hash the way a Dict lookup would.

metadata_string

def metadata_string(self, key: StringSpan) -> String

A metadata entry decoded as UTF-8; "" when the key is absent.

has_metadata

def has_metadata(self, key: StringSpan) -> Bool

metadata_map

def metadata_map(self) -> Dict[String, List[UInt8]]

The metadata as a Dict, for callers that want that shape.

schema_json

def schema_json(self) -> String

The schema exactly as the file spelled it.

has_next

def has_next(mut self) -> Bool

next

def next(mut self) -> Value

Decode the next record; raises when the file is exhausted.

read_all

def read_all(mut self) -> List[Value]

count_blocks

def count_blocks(self) -> Int

Walk every block header, without decompressing anything.

Independent of how far iteration has got: it re-reads the header and then strides over the blocks.

DataFileWriter

struct DataFileWriter[C: CodecSet = DefaultCodecs]

Builds an Avro Object Container File in memory.

Implements Copyable, Movable

Fields
schema: Schema
codec: String
sync_marker: List[UInt8]
sync_interval: Int: Approximate uncompressed bytes per block before an automatic flush.
meta_keys: List[String]
meta_vals: List[List[UInt8]]
out: List[UInt8]
buf: Encoder
block_count: Int
started: Bool
closed: Bool
schema_json: String: What to write as avro.schema; the printed schema unless overridden.

__init__

def __init__(out self, var schema: Schema, codec: StringSpan = "null", sync_interval: Int = Int(64000))

set_sync_marker

def set_sync_marker(mut self, marker: Span[UInt8])

Pin the sync marker — the tests need reproducible files.

set_metadata

def set_metadata(mut self, key: StringSpan, value: Span[UInt8])

Add a file-level metadata entry. avro.* keys are reserved.

set_metadata_string

def set_metadata_string(mut self, key: StringSpan, value: StringSpan)

set_schema_json

def set_schema_json(mut self, text: StringSpan)

Write text verbatim as avro.schema.

to_json() prints an equivalent schema, not a byte-identical one. Apache Iceberg writers that want their manifests to match another implementation's bytes can hand over the exact JSON here; it is the caller's job to keep it equivalent to the schema being written.

append

def append(mut self, value: Value)

flush

def flush(mut self)

Close off the current block, if it has anything in it.

close

def close(mut self)

bytes

def bytes(mut self) -> List[UInt8]

Finish the file and hand back its bytes.

save

def save(mut self, path: StringSpan)

Functions

read_file_bytes

def read_file_bytes(path: StringSpan) -> List[UInt8]

write_file_bytes

def write_file_bytes(path: StringSpan, data: Span[UInt8])

random_sync_marker

def random_sync_marker() -> List[UInt8]

Aliases

comptime MAGIC0 = UInt8(79)
comptime MAGIC1 = UInt8(98)
comptime MAGIC2 = UInt8(106)
comptime MAGIC3 = UInt8(1)
comptime SYNC_SIZE = Int(16)
comptime DEFAULT_SYNC_INTERVAL = Int(64000)