magmalake objectstore

objectstore · API

objectstore.s3

objectstore.s3 — an S3 client over HttpClient, signed with SigV4.

Covers what an Iceberg reader and writer actually issue: GetObject (with Range, because a Parquet reader wants the footer before anything else), PutObject (multipart above a threshold), HeadObject, DeleteObject and a paginated ListObjectsV2.

Addressing works both ways. Virtual-host style (bucket.s3.region.amazonaws.com) is what AWS prefers; path style (endpoint/bucket/key) is what MinIO, Ceph and most S3-compatible servers default to, and what a bucket name with dots forces even on AWS. S3Config.path_style picks, and a custom endpoint makes this work against MinIO, R2 or anything else that speaks the protocol.

Credentials come from five places, in the order a caller should expect: explicit properties, the AWS_* environment, the EC2 instance role (IMDSv2, when neither of those set a key and no custom endpoint is configured), REST-catalog vended credentials (which are just properties under a location prefix — see fileio.FileIOResolver), and presigned URLs, which carry the signature in the query string and need no credentials at the point of use at all.

The instance role is what AWS SDKs fall back to as well. The shim does the lookup and caches it for the process, "not on EC2" included, until five minutes before the credentials expire. AWS_EC2_METADATA_DISABLED=true turns it off and AWS_EC2_METADATA_SERVICE_ENDPOINT points it elsewhere. From a container, the instance's metadata hop limit must be at least 2.

Structs

S3Credentials

struct S3Credentials

Implements Copyable, Movable

Fields
access_key_id: String
secret_access_key: String
session_token: String

__init__

def __init__(out self)

is_set

def is_set(self) -> Bool

S3Config

struct S3Config

Everything needed to address and sign against one S3 endpoint.

Implements Copyable, Movable

Fields
endpoint: String: https://minio.local:9000, or "" for the AWS regional endpoint.
region: String
path_style: Bool
credentials: S3Credentials
sign_payload: Bool: When false, PUT bodies are signed as UNSIGNED-PAYLOAD — a real option for a large object over TLS, where hashing the payload twice is pure overhead. Set from the s3.unsigned-payload property; default on, since dropping it also drops the guarantee that what S3 stored is what was signed.
anonymous: Bool: Skip signing entirely: public buckets, and presigned URLs, need no credentials and are rejected by some servers if signed anyway.
multipart_threshold: Int: Objects larger than this go up as a multipart upload. 0 disables multipart entirely, which is the 0.1.x behaviour and still correct up to S3's 5 GB single-PUT ceiling.
multipart_part_size: Int: Bytes per part, clamped up to S3's 5 MiB floor.

__init__

def __init__(out self)
def __init__(out self, var endpoint: String, var region: String, path_style: Bool, var credentials: S3Credentials, sign_payload: Bool = True, anonymous: Bool = False, multipart_threshold: Int = Int(8388608), multipart_part_size: Int = Int(8388608))

from_env

def from_env() -> Self

The standard AWS_* variables, plus AWS_ENDPOINT_URL[_S3].

from_properties

def from_properties(props: Dict[String, String]) -> Self

Iceberg property names, which is what a REST catalog hands back.

Recognised: s3.endpoint, s3.access-key-id, s3.secret-access-key, s3.session-token, s3.region, s3.path-style-access, s3.multipart.part-size-bytes (Iceberg's own name and units), client.region as a fallback for the region, and s3.unsigned-payload.

Iceberg's s3.multipart.threshold is a factor of the part size expressed as a decimal, which needs a float parser this module does not have; the threshold follows the part size instead, and S3Config.multipart_threshold sets it directly.

ObjectInfo

struct ObjectInfo

Implements Copyable, Movable

Fields
key: String
size: Int
last_modified: String
etag: String

ListResult

struct ListResult

Implements Copyable, Movable

Fields
objects: List[ObjectInfo]
common_prefixes: List[String]
is_truncated: Bool
next_continuation_token: String

__init__

def __init__(out self)

S3Client

struct S3Client

Implements Copyable, Movable

Fields
config: S3Config
http: HttpClient

__init__

def __init__(out self, var config: S3Config)
def __init__(out self, var config: S3Config, var http: HttpClient)

set_max_retries

def set_max_retries(mut self, n: Int)

Retries per request; 3 by default, 0 to disable.

The full policy — backoff window, ceiling, whether a non-idempotent request may be repeated — is self.http.retry. S3 needs this more than most callers do: SlowDown is how a bucket says "you are going too fast", and it arrives at exactly the moment a scan is issuing its hundredth range read.

url_for

def url_for(self, bucket: String, key: String) -> Tuple[String, String, String]

Returns (url, host header value, signing path).

The signing path is the request target before the query string, and it must match byte for byte what the server reconstructs — hence building it here rather than re-deriving it from the URL later.

request

def request(self, method: String, bucket: String, key: String, params: List[QueryParam] = List(), body: Span[UInt8] = Span(), extra_headers: List[Header] = List(), range_start: Int = Int(-1), range_end: Int = Int(-1)) -> Response

get_object

def get_object(self, bucket: String, key: String, range_start: Int = Int(-1), range_end: Int = Int(-1)) -> List[UInt8]

Whole object, or the inclusive byte range [start, end].

signed_get

def signed_get(self, bucket: String, key: String, range_start: Int = Int(-1), range_end: Int = Int(-1)) -> BatchRequest

A signed GET of the object or the inclusive range [start, end], to send in a batch.

get_many

def get_many(self, reqs: List[BatchRequest], what: List[String]) -> List[List[UInt8]]

Runs signed GETs at once; raises naming what[i] for the first that failed.

get_ranges

def get_ranges(self, bucket: String, key: String, ranges: List[Tuple[Int, Int]]) -> List[List[UInt8]]

Several inclusive byte ranges (start, end) of one object, fetched at once.

head_object

def head_object(self, bucket: String, key: String) -> Response

object_length

def object_length(self, bucket: String, key: String) -> Int

object_exists

def object_exists(self, bucket: String, key: String) -> Bool

put_object

def put_object(self, bucket: String, key: String, body: Span[UInt8], content_type: String = "")

Stores the object, as one PUT or as a multipart upload.

The switch is S3Config.multipart_threshold. Nothing about the call changes: AnyOutputFile.overwrite and FileIOResolver.write get multipart for free, which is the point — a Parquet writer should not have to know how big its own output turned out to be.

create_multipart_upload

def create_multipart_upload(self, bucket: String, key: String, content_type: String = "") -> String

POST /key?uploads — returns the upload id.

upload_part

def upload_part(self, bucket: String, key: String, upload_id: String, part_number: Int, body: Span[UInt8]) -> String

PUT /key?partNumber=N&uploadId=… — returns the part's ETag.

Signed like any other request, so every part pays its own SigV4; with sign_payload = False it pays UNSIGNED-PAYLOAD instead, which for a multi-hundred-megabyte object over TLS is the difference between hashing the data twice and hashing it once.

complete_multipart_upload

def complete_multipart_upload(self, bucket: String, key: String, upload_id: String, etags: List[String])

POST /key?uploadId=… with the part list, in order.

abort_multipart_upload

def abort_multipart_upload(self, bucket: String, key: String, upload_id: String)

DELETE /key?uploadId=…. Unaborted parts are billed until a lifecycle rule collects them, so this runs on every failure path.

put_object_multipart

def put_object_multipart(self, bucket: String, key: String, body: Span[UInt8], content_type: String = "")

Create, upload each part, complete — aborting if anything fails.

Parts are Span slices of the caller's buffer, so the object is never held twice: peak memory is the caller's own bytes plus one part's worth of SigV4 hashing state.

delete_object

def delete_object(self, bucket: String, key: String)

list_objects_v2

def list_objects_v2(self, bucket: String, prefix: String = "", delimiter: String = "", max_keys: Int = Int(0), continuation_token: String = "") -> ListResult

One page of ListObjectsV2. See list_all for every page.

list_all

def list_all(self, bucket: String, prefix: String = "", delimiter: String = "") -> ListResult

Every page, concatenated. Bounded at 10 000 requests so a server that keeps handing back the same token cannot spin forever.

presign_get

def presign_get(self, bucket: String, key: String, expires_seconds: Int = Int(3600)) -> String

A URL that grants GET on one object for expires_seconds.

Everything travels in the query string, so the holder needs no credentials — which is the whole point, and why HttpClient keeps redirects off: re-issuing a signed URL elsewhere would leak it.

Functions

instance_credentials

def instance_credentials() -> Optional[S3Credentials]

The EC2 instance role's credentials, or None off EC2, without a role, or with AWS_EC2_METADATA_DISABLED=true.

xml_unescape

def xml_unescape(s: String) -> String

xml_escape

def xml_escape(s: String) -> String

The three characters that cannot appear in element text.

Only CompleteMultipartUpload needs this, and only for an ETag, which in practice is hex — but an ETag is server-chosen and this document decides whether a 40 MB upload lands.

xml_element

def xml_element(body: String, tag: String, start: Int = Int(0)) -> Int

Byte offset of <tag>'s content, or -1. Feeds xml_text.

xml_text

def xml_text(body: String, tag: String, start: Int = Int(0)) -> String

The text of the first <tag> at or after start; "" if absent.

xml_blocks

def xml_blocks(body: String, tag: String) -> List[String]

Every <tag> … </tag> body, in document order.

parse_list_objects

def parse_list_objects(body: String) -> ListResult

s3_error_message

def s3_error_message(resp: Response) -> String

<Error><Code>…</Code><Message>…</Message></Error> if that is what came back, else the first bytes of whatever did.

split_s3_uri

def split_s3_uri(location: String) -> Tuple[String, String]

s3://bucket/a/b -> ("bucket", "a/b").

s3a:// and s3n:// (the Hadoop schemes, still common in older Iceberg metadata) address the same objects and are accepted here.

Aliases

comptime SERVICE = String("s3")
comptime DEFAULT_REGION = String("us-east-1")
comptime MIN_PART_SIZE = 5242880

S3's floor for every part but the last. Not a suggestion: a smaller part is rejected at CompleteMultipartUpload, after the bytes have been sent.

comptime DEFAULT_PART_SIZE = 8388608

Big enough that a 5 GB object stays inside the 10 000-part limit with room to spare, small enough that a failed part is cheap to resend.