`orca.tables`

Iceberg tables from steps: pinned reads, and commits a retry recognises.

from orca.tables import append, scan

def load(mut ctx: Ctx, raw: RawFile) raises -> Loaded:
    var src = Table.load_local(raw.table)
    var batches = scan(ctx, src).to_batches()
    var dst = Table.load_local(OUT)
    var id = append(ctx, dst, transform(batches))
    return Loaded(raw.month, id)

scan reads a table at a pinned snapshot: the one this step's first attempt read, or its current snapshot if this is the first attempt, which is then recorded. A retry or a resume of the step sees the same data, whatever was committed in between.
append and commit put orca.* properties in the summary of the snapshot they commit (Ctx.summary): pipeline, run, step, partition, attempt, commit number, code, and the snapshots read. Before writing anything, they look through the table's snapshots for one this step already made in this run (Ctx.made). If an attempt committed and was killed before Postgres heard of it, the retry finds that snapshot and returns it instead of committing the data a second time.

A table is known by its location, metadata.location, and, when it's in the warehouse, by its dataset name. ORCA_WAREHOUSE is the root of a filesystem catalog: dataset nyc.trips is the table at <warehouse>/nyc/trips, where the warehouse is a directory or a location in object storage (s3://bucket/warehouse, as iceberg's FileIO reads it, from the AWS_* environment). table loads a dataset the step declared, and ensure_table creates one it writes if it's missing, and adds the columns the step's schema has and the table lacks. scan refuses a dataset the step didn't declare in reads=, and append and commit one it didn't declare in writes=. A table outside the warehouse (or with ORCA_WAREHOUSE unset) is used by location as before, and belongs to no dataset.

This module imports iceberg, so only pipelines that use tables build it; orca itself doesn't.

Functions

`warehouse`

def warehouse() -> String

ORCA_WAREHOUSE: the root datasets are found under, or empty.

`dataset_of`

def dataset_of(location: String) -> String

The dataset a table at location is, namespace.table, or empty if it isn't directly under a namespace in the warehouse.

`load_dataset`

def load_dataset(dataset: String) -> Table

A dataset's table, outside any step: nothing is checked or recorded. For reading what isn't a step's input, such as a keyed pipeline's keys (orca.keyed).

`table`

def table(mut ctx: Ctx, dataset: String) -> Table

The table of a dataset the step declared it reads or writes.

`ensure_table`

def ensure_table(mut ctx: Ctx, dataset: String, schema: Schema, spec: PartitionSpec = PartitionSpec.unpartitioned(Int(0))) -> Table

The table of a dataset the step writes, created empty with schema and spec if the warehouse doesn't have it yet.

A table that exists is brought up to schema: the columns schema has and the table lacks, matched by name at any depth, are added in one schema commit, and a column schema makes optional becomes optional. Rows already in the table read the new columns as null. Columns the table has and schema doesn't are kept, and written as null.

It refuses, with the dataset in the message, what that can't do: a column whose type differs, a new column marked required, a required column schema leaves out, and a field id in schema that isn't the id the table gave that column, since batches built from schema carry its ids and a column is matched by id first. A new column's id should be the next one up: the table numbers new fields from its highest id, in the order schema lists them, a struct's fields after the struct. spec is only used to create the table.

`scan`

def scan(mut ctx: Ctx, table: Table) -> TableScan

A scan of table at this step's pinned snapshot.

A table that had no snapshot when it was first read stays empty for the step: the scan sees no files even if some were committed since.

`find_commit`

def find_commit(ctx: Ctx, table: Table) -> Int64

The snapshot this step already committed as its next commit to table in this run, or -1.

append and commit check this themselves. A step that writes files before committing them, such as a compaction, checks it first, so a retry of an attempt that committed doesn't write the files again.

`append`

def append(mut ctx: Ctx, mut table: Table, batches: List[RecordBatch]) -> Int64

Writes batches to table as one append snapshot and returns its ID, or returns the snapshot an earlier attempt of this step committed.

`commit`

def commit(mut ctx: Ctx, mut table: Table, operation: String, var files: FileChanges) -> Int64

Commits files (data files added and removed, already written) to table as one snapshot of operation, e.g. replace for a compaction, and returns its ID, or returns the snapshot an earlier attempt of this step committed.