magmalake objectstore

objectstore · API

objectstore.fileio

objectstore.fileio — the InputFile / OutputFile / FileIO traits and the scheme resolver.

This is the seam iceberg.mojo already anticipates. Its src/iceberg/io.mojo declares an InputFile trait with location(), exists() and read_all(), and a local-only FileIO; the traits here are a superset of that, so an objectstore input file satisfies the Iceberg trait unchanged, while adding the two things a lazy Parquet reader needs: length() and read_range(offset, length).

FileIOResolver is the dynamic half. Mojo dispatches traits statically, so a "pick a backend by URI scheme" API cannot return an arbitrary trait implementation; AnyInputFile and AnyOutputFile are small tagged unions over the three backends, and they conform to the traits themselves. Generic code can keep using the traits; code that resolves locations at runtime uses the resolver.

Per-scheme configuration follows Iceberg's property names (s3.endpoint, s3.access-key-id, …) so a LoadTableResult's config map can be handed straight over, and storage-credentials — the REST spec's list of {prefix, config} entries — is matched longest prefix first, exactly as the spec requires when a catalog vends different credentials for different paths within one table.

Structs

AnyInputFile

struct AnyInputFile

An InputFile whose backend is chosen at runtime by URI scheme.

Implements Copyable, InputFile, Movable, RangeReader

Fields
backend: Int
uri: String

__init__

def __init__(out self, backend: Int, var uri: String, var local: LocalInputFile, var http: HttpInputFile, var s3: S3Client, var bucket: String, var key: String)

location

def location(self) -> String

exists

def exists(self) -> Bool

length

def length(self) -> Int

read_all

def read_all(self) -> List[UInt8]

read_range

def read_range(self, offset: Int, length: Int) -> List[UInt8]

read_ranges

def read_ranges(self, ranges: List[ByteRange]) -> List[List[UInt8]]

One seek per range for local files. For HTTP and S3, nearby ranges are coalesced and the groups fetched at once.

AnyOutputFile

struct AnyOutputFile

An OutputFile whose backend is chosen at runtime by URI scheme.

The HTTP backend covers three cases that all reduce to a single PUT: a presigned S3 upload URL, GCS's XML endpoint with a bearer token, and an Azure blob with a SAS token. create there is "check, then write", which is racy — HTTP has no atomic create.

Implements Copyable, Movable, OutputFile

Fields
backend: Int
uri: String

__init__

def __init__(out self, backend: Int, var uri: String, var local: LocalOutputFile, var http: HttpOutputFile, var s3: S3Client, var bucket: String, var key: String)

location

def location(self) -> String

exists

def exists(self) -> Bool

create

def create(self, data: Span[UInt8])

overwrite

def overwrite(self, data: Span[UInt8])

StorageCredential

struct StorageCredential

One entry of a REST LoadTableResult's storage-credentials.

Same shape as iceberg.catalog.rest.StorageCredential, so a parsed response can be forwarded without translation.

Implements Copyable, Movable

Fields
prefix: String
config: Dict[String, String]

FileIOResolver

struct FileIOResolver

Picks a backend by URI scheme and configures it from properties.

Property resolution for a given location, most specific first:

  1. the longest matching storage-credentials prefix,
  2. the resolver's own properties (an Iceberg table's config),
  3. AWS_* environment variables.

Location rewrites (rebase) are applied before any of that, because Iceberg metadata records absolute locations and a warehouse that has been copied elsewhere — or a test fixture checked into a repo — needs them redirected without rewriting the metadata.

Implements Copyable, Movable

Fields
properties: Dict[String, String]
storage_credentials: List[StorageCredential]
from_prefixes: List[String]
to_prefixes: List[String]
http: HttpClient

__init__

def __init__(out self)
def __init__(out self, var properties: Dict[String, String])

set

def set(mut self, var key: String, var value: String)

add_storage_credential

def add_storage_credential(mut self, var prefix: String, var config: Dict[String, String])

rebase

def rebase(mut self, var old_prefix: String, var new_prefix: String)

Redirect every location under old_prefix to new_prefix.

Byte-compatible with iceberg.io.FileIO.rebase.

resolve

def resolve(self, location: String) -> String

The location this one should actually be read from.

properties_for

def properties_for(self, location: String) -> Dict[String, String]

Merged properties for one location, most specific last.

storage-credentials entries are matched by prefix and the longest match wins: a catalog may vend one credential for a table's data prefix and a narrower one for a single partition, and the narrower entry is the one that was meant.

s3_config_for

def s3_config_for(self, location: String) -> S3Config

gcs_config_for

def gcs_config_for(self, location: String) -> GcsConfig

azure_config_for

def azure_config_for(self, location: String, account: String) -> AzureConfig

backend_for

def backend_for(self, location: String) -> Int

new_input

def new_input(self, location: String) -> AnyInputFile

new_output

def new_output(self, location: String) -> AnyOutputFile

delete

def delete(self, location: String)

list

def list(self, prefix: String) -> List[String]

Locations under prefix, as full URIs.

Local listing is recursive; S3 listing is flat by nature — an object store has no directories, only key prefixes.

read_all

def read_all(self, location: String) -> List[UInt8]

read_range

def read_range(self, location: String, offset: Int, length: Int) -> List[UInt8]

The bytes [offset, offset+length).

This is the entry point a lazy Parquet reader wants: fetch the last 64 KiB for the footer, then exactly the row groups the scan needs.

read_ranges

def read_ranges(self, location: String, ranges: List[ByteRange]) -> List[List[UInt8]]

Several ranges of one object, coalesced where that saves requests.

One new_input for the lot, so an S3 location is signed and addressed once rather than once per range.

read_all_many

def read_all_many(self, locations: List[String]) -> List[List[UInt8]]

Several whole objects, in order. Those on S3 are fetched at once; the rest one by one. For many small files read together, such as an Iceberg table's manifests.

read_ranges_many

def read_ranges_many(self, locations: List[String], ranges: List[List[ByteRange]]) -> List[List[List[UInt8]]]

ranges[i] of locations[i], for every i, in order.

Each file's ranges are coalesced as read_ranges does, and then every group of every file on S3 is fetched in one batch: reading several files' footers, or several files' column chunks, costs one round of requests rather than one per file. Other backends read file by file.

read_text

def read_text(self, location: String) -> String

exists

def exists(self, location: String) -> Bool

write

def write(self, location: String, data: Span[UInt8])

Traits

InputFile

A readable byte source addressed by a location.

location, exists and read_all are byte-compatible with iceberg.io.InputFile; length and read_range are the additions that let a reader fetch a Parquet footer without pulling the whole file.

Implements Copyable, Movable

location

def location(self) -> String

exists

def exists(self) -> Bool

length

def length(self) -> Int

read_all

def read_all(self) -> List[UInt8]

read_range

def read_range(self, offset: Int, length: Int) -> List[UInt8]

read_ranges

def read_ranges(self, ranges: List[ByteRange]) -> List[List[UInt8]]

Several ranges at once, in the order asked for.

The default is one read_range per range, which is exactly right for a file the process can seek. Backends where a range costs a round trip override it to coalesce (see objectstore.ranges), which is what turns a Parquet scan's per-column-chunk reads into a handful of requests.

OutputFile

A writable destination.

create fails if the location already exists; overwrite does not. That is Iceberg's own distinction — metadata files are written with create so that two writers racing on the same version number cannot both believe they won.

Implements Copyable, Movable

location

def location(self) -> String

exists

def exists(self) -> Bool

create

def create(self, data: Span[UInt8])

overwrite

def overwrite(self, data: Span[UInt8])

FileIO

A namespace of locations: open, delete, list.

Implements Copyable, Movable

comptime InputType
comptime OutputType

new_input

def new_input(self, location: String) -> Self.InputType

new_output

def new_output(self, location: String) -> Self.OutputType

delete

def delete(self, location: String)

list

def list(self, prefix: String) -> List[String]

Aliases

comptime BACKEND_LOCAL = 0
comptime BACKEND_HTTP = 1
comptime BACKEND_S3 = 2
comptime BACKEND_GCS = 3
comptime BACKEND_AZURE = 4

GCS and Azure reduce to BACKEND_HTTP once a location has been turned into a URL plus headers — a bearer token for GCS, a SAS query string and x-ms-version for Azure — so only the resolver distinguishes them. The open files themselves are HttpInputFile/HttpOutputFile.