morca

morca · API

orca.sensors

Runs started by datasets: sensors, and rebuilds downstream of a dataset.

orca rebuild raw.events 2024-03           every pipeline downstream, in order

Sensors. A pipeline declared sensor=True runs when a dataset it reads (and doesn't write itself) gets a snapshot. The scheduler learns of one in two ways (orca.scheduler): from orca.commits, for a snapshot a succeeded run of another pipeline committed, and by polling <binary> sense every ORCA_SENSE_SECONDS (default 60), for a snapshot something other than orca committed. orca.sensed holds a row per pipeline and snapshot seen, with the run it started, inserted in the transaction that queues the run, so a snapshot starts one run however many ticks (or schedulers) see it. The snapshots a tick sees for one pipeline start one run, over all the partitions they hold.

A sensor counts from when it's registered as one: the snapshots its datasets have then are recorded as seen without a run, and commits from before then are ignored. A backfill or a rebuild covers what came before.

Partitions. A snapshot an orca run committed holds the partition of the step that committed it, or the run's partitions for a gather step. One committed by something else holds what its summary's orca.partitions says, and without that the downstream pipeline runs the partition its schedule's next fire would fill now, or, unscheduled, the one now falls in. map_keys moves keys between kinds by containment: a day is the month it's in, and a month is each of its days. Keys before a pipeline's first partition are dropped.

Rebuilds. rebuild queues runs for every pipeline downstream of a dataset, whether or not it's a sensor, over the keys given, mapped to each pipeline's kind. A pipeline runs at its longest distance from the dataset (level 0 reads it), so one that reads two rebuilt datasets runs after both. The plan is orca.rebuild_pipelines; advance_rebuilds, on every tick, queues a level's runs once every run before it has succeeded, and stops at one that failed or was cancelled, which resuming the run from the UI continues. A rebuild's commits don't start sensors in pipelines the rebuild runs itself.

Structs

Planned

struct Planned

A pipeline a rebuild runs: its level, partitions and run.

Implements Copyable, Movable

Fields
pipeline: String
level: Int
partitions: String
run: Int: 0 until queued.
status: String: The run's status, or empty.

Rebuild

struct Rebuild

Implements Copyable, Movable

Fields
id: Int
dataset: String
partitions: String
status: String
created: String
ended: String
plan: List[Planned]

Link

struct Link

A pipeline downstream of a dataset, as a rebuild of it runs them.

Implements Copyable, Movable

Fields
pipeline: String
level: Int: Its run's place in the rebuild: 0 reads the dataset itself, and each other runs after every pipeline it reads from.
reads: List[String]: The datasets it reads that the rebuild changes: the dataset, or what a pipeline before it writes.

Functions

key_kind

def key_kind(key: StringSpan) -> String

monthly for a YYYY-MM key, daily for YYYY-MM-DD, hourly for YYYY-MM-DDTHH, or empty.

split_keys

def split_keys(s: StringSpan) -> List[String]

Comma-separated keys, without blanks.

map_keys

def map_keys(keys: List[String], kind: String, start: String = "") -> List[String]

keys, of any time kind, as keys of kind: a key as the longer one it falls in (an hour as its day or month, a day as its month), or as each shorter one it has (a month as its days or hours, a day as its hours). Those before start are dropped, and the rest are sorted, once each.

join_keys

def join_keys(keys: List[String]) -> String

parse_sense

def parse_sense(text: StringSpan) -> List[Sensed]

What <binary> sense printed, a line per dataset.

downstream

def downstream(mut db: State, dataset: String) -> List[Link]

The pipelines a rebuild of dataset runs, by level then name, each with what it reads that the rebuild changes (_levels).

plan_rebuild

def plan_rebuild(mut db: State, dataset: String, keys: String) -> Tuple[List[String], List[Planned]]

The partitions a rebuild of dataset over keys is given, and the run of each pipeline downstream it would queue, by level; nothing is written. Raises, saying why, if there's nothing to rebuild.

rebuild

def rebuild(mut db: State, dataset: String, keys: String) -> Int

Plans a rebuild downstream of dataset over keys (KEY[..KEY][,...] of either kind) and returns its ID; advance_rebuilds queues its runs. Raises, saying why, if there's nothing to rebuild.

rebuilds

def rebuilds(mut db: State, id: Int = Int(0), dataset: String = "", limit: Int = Int(20)) -> List[Rebuild]

Rebuilds, newest first: the one id names, or those of dataset, or all; each with its plan in run order.

advance_rebuilds

def advance_rebuilds(mut db: State)

Queues the next level of each rebuild whose levels before it have all succeeded, and brings each rebuild's status up to date.