# orca.tables

> A module of morca'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.

**Files written on a worker.** A dista task can write a table's data
files itself and send back only their entries, so the step commits
metadata rather than receiving the rows:

```mojo
def write_hour(day: String, hour: Int) raises -> DataFiles:
    # On a worker: no Ctx, nothing checked or recorded.
    return write_files(load_dataset(EVENTS), fetch(day, hour))

comptime Hour = Task2[String, Int, DataFiles, write_hour]

def load_day(mut ctx: Ctx, d: Day) raises -> Int64:
    var t = ensure_table(ctx, EVENTS, schema())
    if find_commit(ctx, t) >= 0:  # a retry: don't write the files again
        return append_files(ctx, t, DataFiles())
    ref rt = ctx.runtime[Tasks[Hour]]()
    var got = rt.get(rt.remote[Hour](String(d), 0))
    return append_files(ctx, t, got)
```

- **`write_files`** (or a **`FileWriter`**, a batch at a time) writes
  batches into a table's `data/` without committing them, and returns their
  entries as **`DataFiles`**, a dista `Codec` over iceberg's
  `DataFile.encode`. A struct holding a `DataFiles` and plain fields (row
  counts, timings) crosses processes by dista's reflection, with no codec
  of its own.
- **`append_files`** commits entries written elsewhere as one `append`
  snapshot, as `append` commits the files it writes: the same `writes=`
  check, summary and recognition. A retry that finds its commit returns it
  and commits nothing; a step checks `find_commit` first, as above, so a
  retry doesn't have the workers write the files again. Files written by an
  attempt that died before committing them are left unreferenced in
  `data/`, as with an interrupted `append`.

**Replacing partitions.** A new run of a key (`orca backfill --rerun`)
isn't a retry: `find_commit` doesn't recognise an earlier run's commits, so
`append` adds the key's rows a second time. A step that recomputes a key's
rows declares the dataset in `replaces=` rather than `writes=`, and its
`append` and `append_files` to it then replace:

```mojo
comptime LoadDay = Step[
    "load_day", Day, Int64, load_day, Mapped, replaces=[EVENTS],
]
```

- The run's first commit to a partition is an `overwrite` that removes the
  files this run didn't commit from the partitions its new files land in,
  and adds them; its later commits find only the run's files there and
  append. A retry recognises its commits as before, so nothing is
  overwritten twice or lost.
- Another writer's file in such a partition, added after the run's first
  commit to the table, means two runs are replacing it at once: the commit
  raises, committing nothing, and only one run keeps its rows.
- `replace_partitions` and `replace_partition_files` do the same for one
  call, to any table. The summary has `orca.replaced-files` and
  `orca.replaced-records`.

`commit` refuses to remove files that aren't all in the current snapshot,
on every try: a removal planned before another writer replaced them
would otherwise keep both writers' rows.

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

## Structs

### `DataFiles`

```mojo
struct DataFiles
```

Data file entries, written but not committed, as a dista task returns them: `encode`/`decode` are iceberg's `encode_data_files` / `decode_data_files`, which carry every field of each entry.

Implements: `Codec`, `Copyable`, `Movable`, `Sized`

Fields:

- `files: List[DataFile]`

#### `__init__`

```mojo
def __init__(out self)
```

```mojo
def __init__(out self, var files: List[DataFile])
```

#### `__len__`

```mojo
def __len__(self) -> Int
```

#### `add`

```mojo
def add(mut self, other: Self)
```

Appends `other`'s entries, e.g. from another task.

#### `encode`

```mojo
def encode(self, mut out: List[UInt8])
```

#### `decode`

```mojo
def decode(buf: Span[UInt8], mut pos: Int) -> Self
```

### `FileWriter`

```mojo
struct FileWriter
```

Writes batches into a table's data files as they come, without committing them, and keeps their entries. For a worker that produces a table's rows a batch at a time; `write_files` writes a list at once.

It holds the table's schema, spec, sort order and write properties as
they were when it was made: a schema change committed meanwhile isn't
seen, and `append_files` commits the entries all the same.

Implements: `Movable`

Fields:

- `io: FileIO`
- `location: String`
- `schema: Schema`
- `spec: PartitionSpec`
- `sort_order_id: Int`
- `options: WriteOptions`
- `files: DataFiles`
- `rows: Int`: The rows written so far.

#### `__init__`

```mojo
def __init__(out self, table: Table)
```

#### `write`

```mojo
def write(mut self, batches: List[RecordBatch])
```

Writes `batches`: a file per partition they hold.

#### `put`

```mojo
def put(mut self, var batch: RecordBatch)
```

Writes one batch.

#### `finish`

```mojo
def finish(deinit self) -> DataFiles
```

The entries of every file written.

## 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.

### `table_defaults`

```mojo
def table_defaults() -> Dict[String, String]
```

The properties `ensure_table` gives every table it creates, and adds to one that lacks them.

- `write.metadata.delete-after-commit.enabled=true` and
  `write.metadata.previous-versions-max=100`: each commit deletes the
  metadata files older than the last hundred, so `metadata/` stops
  growing by one file per commit.
- `history.expire.max-snapshot-age-ms` (30 days): the history the
  scheduler's expiry keeps (`orca expire`, `ORCA_EXPIRE_HOURS`). Its
  presence is also what makes the scheduler expire the table at all;
  a table orca didn't create keeps all its snapshots.

### `ensure_table`

```mojo
def ensure_table(mut ctx: Ctx, dataset: String, schema: Schema, spec: PartitionSpec = PartitionSpec.unpartitioned(Int(0)), properties: Dict[String, String] = Dict()) -> 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.

`properties` are Iceberg table properties: a table is created with
them on top of `table_defaults()`, and an existing one gets those it
lacks or has with another value (one metadata commit; none when all
are as given). `history.expire.max-snapshot-age-ms` sets how much
history the scheduler's expiry keeps.

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

### `write_files`

```mojo
def write_files(table: Table, batches: List[RecordBatch]) -> DataFiles
```

Writes `batches` into `table`'s data files without committing them, and returns their entries, for `append_files` to commit, here or in another process. Nothing is checked or recorded: it's for a dista task, which has no `Ctx`, with the table from `load_dataset`.

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

To a dataset the step declared in `replaces=`, it's `replace_partitions`:
the partitions the rows land in lose what earlier runs left there.

### `append_files`

```mojo
def append_files(mut ctx: Ctx, mut table: Table, files: List[DataFile]) -> Int64
```

Commits data files already written, here or on a worker (`write_files`), to `table` as one `append` snapshot and returns its ID, or returns the snapshot an earlier attempt of this step committed, as `append` does; the files are then not committed again. No files commits an empty snapshot, as `append` of no batches does, so a step's commits keep their numbers.

Refuses a dataset the step didn't declare in `writes=`, and an entry
that isn't a data file (a delete file goes through `commit`).

To a dataset the step declared in `replaces=`, it's
`replace_partition_files`.

```mojo
def append_files(mut ctx: Ctx, mut table: Table, files: DataFiles) -> Int64
```

`append_files` of the entries a task returned.

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

Every file `files` removes must be in the table's current snapshot, on
the first try and on each retry after another writer committed first:
if one isn't, another writer removed or replaced it since the caller
read it, and the commit raises, committing nothing, rather than remove
the rest and keep both writers' rows. The step's retry plans the change
again. (`replace_partitions` plans it again itself on each try.)

### `replace_partitions`

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

Writes `batches` to `table` as one snapshot that also removes the files earlier runs left in the partitions the rows land in, and returns its ID, or the snapshot an earlier attempt of this step committed.

What `append` does to a dataset the step declared in `replaces=`; this
is for a table outside the warehouse, or a step that replaces one table
and appends to another it declares in `writes=`.

- The run's first commit to a partition is an `overwrite` that removes
  every data and delete file in it that this run didn't commit, and
  adds the new files. Later commits of the run to the partition find
  only its own files there and append; files of this run, from any of
  its steps and attempts, are never removed.
- The partitions are those of the new files under the table's current
  spec: no rows replace nothing. Files written under an older spec are
  left alone. An unpartitioned table is one partition: the run replaces
  the whole table.
- What to remove is planned against the snapshot each try commits on
  top of, again on every retry after another writer committed first.
- A file in a partition being replaced that another writer committed
  after this run's first commit to the table means two writers are
  replacing (or writing) the partition at once. The commit raises,
  committing nothing, so only one of them keeps its rows: the other's
  attempt fails, and a rerun of it once the first has finished replaces
  the partition again.

The snapshot's summary has `orca.replaced-files` and
`orca.replaced-records`: the data files and rows it removed.

### `replace_partition_files`

```mojo
def replace_partition_files(mut ctx: Ctx, mut table: Table, files: List[DataFile]) -> Int64
```

`replace_partitions` of data files already written, here or on a worker (`write_files`), as `append_files` commits them.

```mojo
def replace_partition_files(mut ctx: Ctx, mut table: Table, files: DataFiles) -> Int64
```

`replace_partition_files` of the entries a task returned.

## Aliases

### `DAY_MS`

```mojo
comptime DAY_MS = Int64(86400000)
```

### `DEFAULT_HISTORY_DAYS`

```mojo
comptime DEFAULT_HISTORY_DAYS = 30
```

How long an orca table keeps its snapshots by default.