# orca.ctx

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

**Commits are fenced.** A process whose heartbeat went stale while it was
alive (the database unreachable for a while, a paused VM) may have lost its
run to another process, which reruns the step. `orca.tables` calls `fence`
right before each commit to Iceberg: it goes ahead only if this process
still owns the run and its heartbeat is younger than half the stale limit,
so no other process can have claimed the run, nor claim it during the
commit unless the commit takes that long. Otherwise the attempt fails
without committing.

## Structs

### `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]`: The datasets it declared it writes, in `writes=` or `replaces=`.
- `declared_replaces: List[String]`: The datasets whose partitions a new run of it replaces (`replaces=`): `orca.tables.append` and `append_files` replace rather than append.
- `attempt: Int`
- `code: String`: The pipeline's registry hash, in hex.
- `token: String`: The run's random token (`Store.run_token`), which its commits carry beside its number; empty for a run without one.
- `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(), log_path: String = "", runtime: Int = Int(0), runtime_tasks: String = "", partitions: String = "", var declared_reads: List[String] = List(), var declared_writes: List[String] = List(), var declared_replaces: List[String] = List(), token: String = "")
```

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

#### `fence`

```mojo
def fence(mut self)
```

Raises unless this process may still commit for its run (see the module's doc); `orca.tables` calls it before each commit.

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

#### `replaces`

```mojo
def replaces(self, dataset: String) -> Bool
```

Whether the step declared it replaces `dataset`'s partitions.

#### `same_run`

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

Whether a snapshot with `summary` was committed by any step of this run: the same pipeline and run. Never for an in-memory run.

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

And not by another run that had this run's number: one a restore of
orca's database lost, or one of another database over the same
tables (#292). A snapshot that carries a run token (`orca.run-token`)
is this run's only if the token is this run's; a snapshot without
one, or a run without one, is told by the number alone, as before
tokens.