# orca.tables

> A module of Orca's API.

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

```mojo
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`

```mojo
def warehouse() -> String
```

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

### `dataset_of`

```mojo
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`

```mojo
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`

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

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

### `ensure_table`

```mojo
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`

```mojo
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`

```mojo
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`

```mojo
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`

```mojo
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.