magmalake iceberg

iceberg · API · catalog

iceberg.catalog.sql

A JDBC-style SQL catalog, over sqlite.mojo or postgres.mojo — the schema and semantics PyIceberg's SqlCatalog uses.

Which backend answers is the URI's business and nothing else's: sqlite:///catalog.db opens a file, postgresql://user@host/db opens a server, and every method below is the same either way. See _sqldriver for what that costs (placeholder rewriting, and one predicate over SQLSTATEs).

Over sqlite this is a development catalog: local, single-process, and test parity with PyIceberg, whose quickstart default is exactly this — a SQLite-backed SqlCatalog — and which is what iceberg-rs.mojo's own fixtures already use as an oracle. Over PostgreSQL the same catalog is deployable: iceberg_tables and iceberg_namespace_properties are the tables PyIceberg's SqlCatalog and the Java JdbcCatalog both use, the guarded UPDATE is a real transaction against a real server, and many writers may share it. A database this catalog creates opens cleanly in PyIceberg's SqlCatalog and vice versa, on either engine: same two tables, same column names, same optimistic-concurrency guard.

iceberg_tables (
    catalog_name, table_namespace, table_name,   -- primary key
    metadata_location, previous_metadata_location
)
iceberg_namespace_properties (
    catalog_name, namespace, property_key,        -- primary key
    property_value
)

Namespaces are dot-joined strings ('db.sub'); a nested namespace is a prefix of a longer one, matched with a LIKE ... ESCAPE '!' pattern the same way PyIceberg escapes %, _ and the escape character itself.

The commit path is the one part of this file that is not shared with the sibling catalogs: unlike FilesystemCatalog (whose optimistic concurrency compares the newest file a directory listing turns up) and RestCatalog (whose server enforces assert-ref-snapshot-id), a SQL catalog's source of truth is one row, and the swap is a guarded

UPDATE iceberg_tables SET metadata_location = <new>
WHERE ... AND metadata_location = <the value this attempt read>

exactly PyIceberg's guard. Zero rows affected means somebody else's commit already moved the pointer; this reloads and retries like the REST catalog's 409 handling, and raises a commit-failed error when retries run out. Manifest writing, data-file writing and metadata JSON serialization are the same prepare_append / prepare_delete / prepare_overwrite / prepare_dynamic_partition_overwrite machinery RestCatalog uses — this file only adds the row that names the current metadata file.

Structs

NamespacePropertiesUpdateSummary

struct NamespacePropertiesUpdateSummary

What update_namespace_properties actually did — PyIceberg's PropertiesUpdateSummary shape.

Implements Copyable, Movable

Fields
removed: List[String]
updated: List[String]
missing: List[String]

__init__

def __init__(out self, var removed: List[String], var updated: List[String], var missing: List[String])

SqlCatalog

struct SqlCatalog

A JDBC/SQL catalog: tables and namespaces live as rows in a SQLite file or a PostgreSQL database, table data and metadata JSON live under warehouse — the same split PyIceberg's SqlCatalog makes.

var cat = SqlCatalog.local("default", "sqlite:///catalog.db", "warehouse")
cat.create_namespace("db")
var t = cat.create_table("db", "orders", schema)
_ = cat.append("db", "orders", batches)

The URI is the only thing that changes between the two backends:

var shared = SqlCatalog.local(
    "default", "postgresql://iceberg@db.internal/catalog?connect_timeout=5",
    "s3://warehouse",
)

Implements Movable

Fields
name: String
warehouse: String
io: FileIO
driver: SqlDriver

__init__

def __init__(out self, var name: String, uri: String, var warehouse: String, var io: FileIO, create_tables: Bool = True)

local

def local(var name: String, uri: String, var warehouse: String) -> Self

create_tables

def create_tables(mut self)

CREATE TABLE IF NOT EXISTS for both catalog tables. Idempotent; opening a database another SQL catalog (ours or PyIceberg's) already initialized is a no-op.

destroy_tables

def destroy_tables(mut self)

Drop both catalog tables. Mirrors PyIceberg's destroy_tables.

table_location

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

namespace_exists

def namespace_exists(self, namespace: String) -> Bool

create_namespace

def create_namespace(mut self, namespace: String, var properties: Dict[String, String] = Dict())

drop_namespace

def drop_namespace(mut self, namespace: String)

list_namespaces

def list_namespaces(self, parent: String = "") -> List[String]

load_namespace_properties

def load_namespace_properties(self, namespace: String) -> Dict[String, String]

update_namespace_properties

def update_namespace_properties(mut self, namespace: String, var removals: List[String] = List(), var updates: Dict[String, String] = Dict()) -> NamespacePropertiesUpdateSummary

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]

create_table

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

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

drop_table

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

Remove the catalog row. Matches PyIceberg's SqlCatalog: the data and metadata files are left in place — use purge_table semantics at a higher layer if the files themselves need to go.

rename_table

def rename_table(mut self, from_namespace: String, from_name: String, to_namespace: String, to_name: String) -> Table

commit_append

def commit_append(mut self, namespace: String, name: String, data_files: List[DataFile], retries: Int = Int(4)) -> Table

Commit already-written data files as one append snapshot.

Each attempt writes a fresh metadata JSON file (and, inside prepare_append, a fresh manifest and manifest list); a refused attempt's files are orphaned, exactly as in every other implementation here, and are nobody's to read.

append

def append(mut self, namespace: String, name: String, batches: List[RecordBatch], retries: Int = Int(4)) -> Table

Write batches into the table's location, then commit them.

commit_change

def commit_change(mut self, namespace: String, name: String, kind: Int, filter_dsl: String, batches: List[RecordBatch], mode: String, retries: Int = Int(4)) -> Table

A delete or an overwrite, committed through the same guarded swap an append uses — the plan is entirely in the manifest list, so there is no per-operation update to the catalog row.

update_schema

def update_schema(mut self, namespace: String, name: String, update: UpdateSchema, retries: Int = Int(4)) -> Table

Commit schema changes: a new metadata file and the guarded swap.

A swap that finds the pointer moved reloads and applies the same changes to the schema that is current then. Changes that leave the schema as it is commit nothing.

delete_where

def delete_where(mut self, namespace: String, name: String, filter_dsl: String, mode: String = "", retries: Int = Int(4)) -> Table

DELETE FROM t WHERE <filter>, committed through the catalog.

overwrite

def overwrite(mut self, namespace: String, name: String, batches: List[RecordBatch], filter_dsl: String = "[\22true\22]", retries: Int = Int(4)) -> Table

Delete what the filter matches and add batches, in one snapshot.

dynamic_partition_overwrite

def dynamic_partition_overwrite(mut self, namespace: String, name: String, batches: List[RecordBatch], retries: Int = Int(4)) -> Table

Replace exactly the partitions the new rows land in.

Aliases

comptime NAMESPACE_EXISTS_PROPERTY = String("exists")