magmalake iceberg

iceberg · API · catalog

iceberg.catalog.filesystem

Filesystem tables: a table is a directory, no catalog service involved.

<table>/metadata/version-hint.text -> "4" <table>/metadata/v4.metadata.json (or 00004-<uuid>.metadata.json) <table>/metadata/snap-<id>-<n>-<uuid>.avro <table>/data/...

version-hint.text is the fast path. When it is absent — the layout every catalog-backed writer produces, including the fixtures in this repo — the newest metadata file is found by listing the directory and taking the highest version prefix, which is the same rule the reference implementations use.

Gzipped metadata (*.gz.metadata.json) is supported: gzip is a nine-byte header, a raw deflate stream and an eight-byte trailer, and the header and trailer are parsed here — deflate.mojo decodes the stream between them.

Structs

Table

struct Table

A loaded table: its metadata, where that metadata came from, and the FileIO its locations resolve through.

Implements Copyable, Movable

Fields
metadata: TableMetadata
metadata_location: String
io: FileIO
name: String

__init__

def __init__(out self, var metadata: TableMetadata, var metadata_location: String, var io: FileIO, var name: String)

load

def load(location: String, var io: FileIO) -> Self

Load a filesystem table from a directory or a metadata file path.

load_local

def load_local(location: String) -> Self

scan

def scan(self) -> TableScan

refresh

def refresh(mut self)

new_append

def new_append(self) -> AppendFiles

Start a fast-append transaction against this table.

append

def append(mut self, batches: List[RecordBatch]) -> Int64

Write batches and commit them as one snapshot. Rows appended.

delete_where

def delete_where(mut self, filter_dsl: String, mode: String = "") -> Int64

DELETE FROM t WHERE <filter>. Returns the rows removed.

The strategy comes from write.delete.mode unless mode overrides it: copy-on-write (the default) rewrites the affected data files, merge-on-read writes deletion vectors on v3 and position delete files on v2.

delete_by_equality

def delete_by_equality(mut self, rows: RecordBatch, equality_ids: List[Int]) -> Int64

Write one equality delete file. Returns the delete rows written.

The rows are values, not positions: every table row whose equality_ids columns match one of them is gone from then on. v2 and unpartitioned only — see iceberg.delete.write_equality_deletes.

overwrite

def overwrite(mut self, batches: List[RecordBatch], filter_dsl: String = "[\22true\22]") -> Int64

Delete every row the filter matches and add batches, in one overwrite snapshot. With no filter this replaces the whole table. Returns the rows added.

dynamic_partition_overwrite

def dynamic_partition_overwrite(mut self, batches: List[RecordBatch]) -> Int64

Replace exactly the partitions the new rows land in. Rows added.

expire_snapshots

def expire_snapshots(mut self, older_than_ms: Int64 = Int64(-1), keep_last: Int = Int(1), dry_run: Bool = False) -> ExpireResult

Drop old snapshots and the files nothing points at any more.

The pruned metadata is committed first and the files go after, so an interruption leaves orphans rather than a snapshot with holes in it. dry_run stops after the decision and returns the list.

update_schema

def update_schema(mut self, update: UpdateSchema) -> Schema

Commit schema changes; returns the new current schema.

No data file is touched: one new metadata file with the new schema appended and made current. A concurrent commit makes this reload and apply the same changes to whatever schema is current then, so a column another writer added in the meantime is kept, and a clash (both adding note) raises instead of committing twice.

commit

def commit(mut self, result: AppendResult)

Persist a prepared commit as a new <V>-<uuid>.metadata.json.

commit_metadata

def commit_metadata(mut self, var metadata: TableMetadata)

Persist table metadata as a new <V>-<uuid>.metadata.json.

The new file is written with create semantics — it must not already exist — which is what makes two writers racing on the same version detectable. A local filesystem really provides that; an object store does not, and the spec says so, so this is unsafe over s3:// in exactly the way every other filesystem-table writer is.

AppendFiles

struct AppendFiles

A fast-append transaction: collect batches or data files, then commit.

var tx = table.new_append()
tx.add(batch)
_ = tx.commit()

Implements Movable

Fields
table: Table
batches: List[RecordBatch]
files: List[DataFile]
retries: Int: How many times to reload and retry when another writer got there first.

__init__

def __init__(out self, var table: Table)

add

def add(mut self, batch: RecordBatch)

add_batches

def add_batches(mut self, batches: List[RecordBatch])

add_data_file

def add_data_file(mut self, file: DataFile)

Append a Parquet file that is already written and already in place.

commit

def commit(mut self) -> Int64

Write the data files, then commit, reloading and retrying if another writer committed in between. Returns the rows appended.

FilesystemCatalog

struct FilesystemCatalog

Tables addressed by path under a warehouse root: <warehouse>/<ns>/<tbl>.

Implements Copyable, Movable

Fields
warehouse: String
io: FileIO

__init__

def __init__(out self, var warehouse: String, var io: FileIO)

local

def local(var warehouse: String) -> Self

table_location

def table_location(self, namespace: String, name: String) -> String

load_table

def load_table(self, namespace: String, name: String) -> Table

table_exists

def table_exists(self, namespace: String, name: String) -> Bool

list_tables

def list_tables(self, namespace: String) -> List[String]

Directories under the namespace that look like Iceberg tables.

create_table

def create_table(self, namespace: String, name: String, schema: Schema, spec: PartitionSpec = PartitionSpec.unpartitioned(Int(0)), var properties: Dict[String, String] = Dict(), format_version: Int = Int(2)) -> Table

Create an empty table: one 00000-<uuid>.metadata.json, no snapshot.

drop_table

def drop_table(self, namespace: String, name: String)

Remove every file of a table. Nothing else references them.

list_namespaces

def list_namespaces(self) -> List[String]

Functions

gunzip

def gunzip(data: List[UInt8]) -> List[UInt8]

Decompress a gzip member. Only the deflate method is defined by RFC 1952.

read_metadata_file

def read_metadata_file(io: FileIO, location: String) -> TableMetadata

Read one metadata file, transparently un-gzipping a *.gz.metadata.json.

read_version_hint

def read_version_hint(io: FileIO, metadata_dir: String) -> Int

The integer in version-hint.text, or -1 when there is no hint.

find_latest_metadata

def find_latest_metadata(io: FileIO, table_location: String) -> String

Locate the current metadata file for a filesystem table.

table_location may be the table directory, its metadata/ directory, or the metadata file itself.

new_table_metadata

def new_table_metadata(location: String, schema: Schema, spec: PartitionSpec, var properties: Dict[String, String], format_version: Int) -> TableMetadata

The metadata of a brand new table: schema, spec, sort order, no data.

Aliases

comptime VERSION_HINT = String("version-hint.text")
comptime METADATA_DIR = String("metadata")
comptime METADATA_SUFFIX = String(".metadata.json")
comptime GZ_METADATA_SUFFIX = String(".gz.metadata.json")