`orca.ctx`

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`

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`

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`

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

`read`

def read(mut self, t: TableSnapshot)

`commit`

def commit(mut self, t: TableSnapshot)

`Ctx`

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`

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`

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`

def log_to_file(self, msg: String)

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

`runtime`

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

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

`pin`

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`

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`

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

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

`check_declared`

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`

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`

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