morca

morca · API

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

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

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__

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

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.

fence

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

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.

replaces

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

Whether the step declared it replaces dataset's partitions.

same_run

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

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.