# orca.ctx

> A module of Orca's API.

`Ctx`: what a step gets besides its input.

The step's identity (pipeline, run, step, partition, attempt, the code's
registry hash), a logger, and what makes reruns safe: the table snapshots
the step's first attempt read, which every later attempt reads again, and
the record of what each attempt read and committed. `orca.tables` reads and
commits Iceberg tables through it, and checks the datasets a step touches
against the ones it declared (`reads=` and `writes=` on its `Step`). `runtime` is the run's dista runtime, for a
pipeline with `tasks=`.

`log` writes to stdout and, when `ORCA_LOGS` is set, appends to the attempt's
log file as well (`orca.logs.attempt_log`).

## Structs

### `TableSnapshot`

```mojo
struct TableSnapshot
```

A table, by its location, at a snapshot.

Implements: `Copyable`, `Movable`

Fields:

- `table: String`
- `snapshot_id: Int64`: -1 for a table that had no snapshot yet.
- `dataset: String`: The table's dataset name, `namespace.table`, or empty for a table outside the warehouse.

### `Recorder`

```mojo
struct Recorder
```

Writes an attempt's reads and commits to Postgres as they happen, so a process killed right after a commit still leaves it on record.

Implements: `Movable`

Fields:

- `url: String`: Empty when nothing is recorded (an in-memory run).
- `attempt_id: Int64`

#### `__init__`

```mojo
def __init__(out self, url: String = "", attempt_id: Int64 = Int64(0))
```

#### `read`

```mojo
def read(mut self, t: TableSnapshot)
```

#### `commit`

```mojo
def commit(mut self, t: TableSnapshot)
```

### `Ctx`

```mojo
struct Ctx
```

The run, step and partition a step is running for.

Implements: `Movable`

Fields:

- `pipeline: String`
- `run: Int`: 0 for an in-memory run: nothing is recorded, and commits aren't recognised across processes.
- `step: String`
- `partition: String`: The partition key, e.g. `2024-01`; empty for a gather step.
- `partitions: String`: The keys this step's commits hold, comma-separated: its partition for a mapped step, the run's keys for a gather step.
- `declared_reads: List[String]`: The datasets the step declared it reads.
- `declared_writes: List[String]`
- `attempt: Int`
- `code: String`: The pipeline's registry hash, in hex.
- `pinned: Dict[String, Int64]`: Table location to the snapshot an earlier attempt of this step read.
- `reads: List[TableSnapshot]`
- `commits: List[TableSnapshot]`
- `recorder: Recorder`
- `log_path: String`: The attempt's log file; empty when logs go to stdout only.

#### `__init__`

```mojo
def __init__(out self, pipeline: String, run: Int, step: String, partition: String, attempt: Int = Int(1), code: String = "", var pinned: Dict[String, Int64] = Dict(), var recorder: Recorder = Recorder(String(""), Int64(0)), log_path: String = "", runtime: Int = Int(0), runtime_tasks: String = "", partitions: String = "", var declared_reads: List[String] = List(), var declared_writes: List[String] = List())
```

#### `log`

```mojo
def log(self, msg: String)
```

Writes a line tagged with the step and partition to stdout, and with a timestamp to the attempt's log file.

#### `log_to_file`

```mojo
def log_to_file(self, msg: String)
```

Writes a line to the attempt's log file only, if it has one.

#### `runtime`

```mojo
def runtime[T: TaskSet](self) -> ref[MutUnsafeAnyOrigin] T.RT
```

The run's dista runtime, for a pipeline declared with `tasks=T`.

#### `pin`

```mojo
def pin(mut self, table: String, current: Int64, dataset: String = "") -> Int64
```

The snapshot of `table` this step reads: the one an earlier attempt read, or else `current`, which is then recorded as read.

#### `committed`

```mojo
def committed(mut self, table: String, snapshot_id: Int64, dataset: String = "")
```

Records that this attempt committed (or found it had committed) `snapshot_id` to `table`.

#### `commits_to`

```mojo
def commits_to(self, table: String) -> Int
```

How many commits to `table` this attempt has made so far.

#### `check_declared`

```mojo
def check_declared(self, dataset: String, write: Bool)
```

Raises unless the step declared it reads (or writes) `dataset`; a table outside the warehouse (`dataset` empty) needs nothing.

#### `summary`

```mojo
def summary(self, table: String) -> Dict[String, String]
```

The snapshot summary properties of this step's next commit to `table`: who made it, from which code, over which partitions, having read what.

#### `made`

```mojo
def made(self, table: String, summary: Dict[String, String]) -> Bool
```

Whether a snapshot with `summary` is this step's next commit to `table`, made by any attempt of this run: the same pipeline, run, step, partition and commit number. Never for an in-memory run.

Every key must be there: a missing one doesn't match an empty value,
such as a gather step's partition (found by fuzz/fuzz_summary.mojo).