`orca.queueing`

Queueing runs over keys: backfills, the keys a pipeline has done or holds, and what a rerun or an append refuses.

Structs

`KeptKeys`

struct KeptKeys

A pipeline's keys as orca.keys has them.

Implements Copyable, Movable

Fields: snapshot: Int64; keys: List[String]; read_at: String; age: Float64: Seconds since they were read.

`Queued`

struct Queued

What a backfill or a keyed sensor queued, and the keys it left.

Implements Movable

Fields: runs: List[Int]; done: List[String]: Keys left because they're done.; live: List[String]: Keys left because a live run has them, that run in live_runs.; live_runs: List[Int]; keys: List[String]: The keys queued (or, in a preview, that would be), sorted.; batches: Int: How many runs those keys make, keys_per_run to a run; a preview counts them without queueing any.

`init`

def __init__(out self)

`skipped`

def skipped(self) -> List[String]

A line for each reason keys were left, or none.

Functions

`dataset_cycle`

def dataset_cycle(mut db: State, describe: String) -> String

How the pipeline describe gives would close a cycle across pipelines, through the datasets they write and read, or empty if it wouldn't. A pipeline reading a dataset it writes itself isn't one.

`parse_keys_output`

def parse_keys_output(binary: String, text: String) -> Tuple[Int64, List[String]]

What <binary> keys printed: its keys snapshot, then its keys.

`current_keys`

def current_keys(p: Registered, timeout: Float64 = Float64("-1")) -> Tuple[Int64, List[String]]

A keyed pipeline's keys snapshot and keys, as <binary> keys prints them, killed past timeout seconds (by default ORCA_BINARY_SECONDS).

`keep_keys`

def keep_keys(mut db: State, pipeline: String, keys: Tuple[Int64, List[String]])

Records the keys <binary> keys printed, in orca.keys.

`kept_keys`

def kept_keys(mut db: State, pipeline: String) -> Optional[KeptKeys]

`file_names`

def file_names(path: String) -> String

The names in the file at path, a line each, as NAME,NAME,... (@file:<path>).

`kind_mismatch`

def kind_mismatch(mut db: State, describe: String) -> String

How the pipeline describe gives would make a sensor run keys of the other kind, a keyed sensor reading a dataset a time-partitioned pipeline writes or the other way round, or empty if it wouldn't.

A keyed sensor's own keys dataset needs no mapping, whoever writes it: a new snapshot of it offers the keys it holds, which are names already. So a dated pipeline may write the keys a keyed sensor runs.

`partition_keys`

def partition_keys(p: Registered, keys: String) -> List[String]

keys as the pipeline's partition keys, each checked: for time keys KEY[..KEY][,...] from its first partition on; for keyed partitions a list of names, @all or @file:<path>, each a current key.

`done_keys`

def done_keys(mut db: State, pipeline: String, keys: List[String]) -> Dict[String, Bool]

Which of keys are done for pipeline (DONE_SQL).

`queue_keys`

def queue_keys(mut db: State, p: Registered, keys: List[String], trigger: String, rerun: Bool = False) -> Queued

Queues runs of p over those of keys that aren't done and aren't in a live run, sorted, keys_per_run to a run; returns the runs, and the keys it left and why. With rerun, keys that are done are queued too, but never one in a live run. Under a lock on the pipeline's row, so two backfills don't both queue a key.

`appending_writes`

def appending_writes(mut db: State, pipeline: String, from_step: Int = Int(0)) -> List[String]

step <name> writes <dataset> for each dataset a step of pipeline writes without replacing it (in writes=, not replaces=), from its from_stepth step on (0, the first): what a new run over a key that already has rows appends to again. A pipeline registered before replaces= existed appends to all it writes.

`append_again`

def append_again(pipeline: String, what: String, writes: List[String], how: String) -> String

Why a rerun of what is refused: it appends again (writes, from appending_writes). how says how to rerun anyway.

`rerun_refusal`

def rerun_refusal(mut db: State, p: Registered, keys: String) -> String

Why orca backfill --rerun of keys is refused without --allow-append, or empty: some are done, and the pipeline's steps append to datasets (appending_writes), so a new run of them would leave their rows twice.

`backfill`

def backfill(mut db: State, p: Registered, keys: String, rerun: Bool = False, allow_append: Bool = False) -> Queued

Queues runs for the keys in keys that aren't done and aren't in a live run, keys_per_run to a run; returns the runs it queued and the keys it left. With rerun (orca backfill --rerun) the done ones are queued again too, as for a day whose source has changed since; unless allow_append, that's refused for a pipeline whose steps append to what they write (rerun_refusal).

`preview_backfill`

def preview_backfill(mut db: State, p: Registered, keys: String, rerun: Bool = False, allow_append: Bool = False) -> Queued

What backfill would queue now, queueing nothing: the keys, how many runs they make, and the keys it would leave and why; or its refusal.

Aliases

comptime DONE_SQL = "WITH k AS (SELECT DISTINCT unnest(string_to_array($2, ',')) AS key), p AS (SELECT registry_hash, describe FROM orca.pipelines WHERE name = $1), m AS (SELECT x->>'name' AS step FROM p, jsonb_array_elements(p.describe->'steps') AS x WHERE x->>'kind' = 'mapped') SELECT k.key FROM k WHERE CASE WHEN EXISTS (SELECT 1 FROM m) THEN NOT EXISTS (SELECT 1 FROM m WHERE NOT EXISTS (SELECT 1 FROM orca.steps st JOIN orca.runs r ON r.id = st.run JOIN p ON r.registry_hash = p.registry_hash WHERE r.pipeline = $1 AND st.step = m.step AND st.partition = k.key AND st.status = 'succeeded')) ELSE false END UNION ALL SELECT k.key FROM k WHERE k.key IN (SELECT unnest(string_to_array(r.partitions, ',')) FROM orca.runs r JOIN p ON r.registry_hash = p.registry_hash WHERE NOT EXISTS (SELECT 1 FROM m) AND r.pipeline = $1 AND r.status = 'succeeded')"

Of the comma-separated keys $2, those done for pipeline $1: every mapped step has succeeded for the key in a run of the pipeline's current code. A pipeline with no mapped step has a key done when a run of its current code over it succeeded. The two are separate halves, each empty when the other applies. The first looks each key's steps up by index (the CASE keeps Postgres from hashing every step of every pipeline instead); the second joins the keys to its runs' keys, read once, rather than scanning the runs again for each key (1.2 s for 1,500 keys over 30,000 runs when it was a CASE branch too, 10 ms now).

comptime APPENDING_SQL = "SELECT x.st->>'name' AS step, w.ds FROM orca.pipelines p, jsonb_array_elements(p.describe->'steps') WITH ORDINALITY AS x(st, n), jsonb_array_elements_text(coalesce(x.st->'writes', '[]'::jsonb)) AS w(ds) WHERE p.name = $1 AND x.n > $2 AND NOT coalesce(x.st->'replaces', '[]'::jsonb) ? w.ds ORDER BY x.n, w.ds"

step and ds for each dataset a step of pipeline $1 after its first $2 declares in writes= and not in replaces=.

comptime ALLOW_APPEND_HOW = "pass --allow-append to append anyway"