# orca.queueing > A module of morca's API. Queueing runs over keys: backfills, the keys a pipeline has done or holds, and what a rerun or an append refuses. ## Structs ### `KeptKeys` ```mojo 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` ```mojo 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__` ```mojo def __init__(out self) ``` #### `skipped` ```mojo def skipped(self) -> List[String] ``` A line for each reason keys were left, or none. ## Functions ### `dataset_cycle` ```mojo 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` ```mojo def parse_keys_output(binary: String, text: String) -> Tuple[Int64, List[String]] ``` What ` keys` printed: its keys snapshot, then its keys. ### `current_keys` ```mojo def current_keys(p: Registered, timeout: Float64 = Float64("-1")) -> Tuple[Int64, List[String]] ``` A keyed pipeline's keys snapshot and keys, as ` keys` prints them, killed past `timeout` seconds (by default `ORCA_BINARY_SECONDS`). ### `keep_keys` ```mojo def keep_keys(mut db: State, pipeline: String, keys: Tuple[Int64, List[String]]) ``` Records the keys ` keys` printed, in `orca.keys`. ### `kept_keys` ```mojo def kept_keys(mut db: State, pipeline: String) -> Optional[KeptKeys] ``` ### `file_names` ```mojo def file_names(path: String) -> String ``` The names in the file at `path`, a line each, as `NAME,NAME,...` (`@file:`). ### `kind_mismatch` ```mojo 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` ```mojo 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:`, each a current key. ### `done_keys` ```mojo 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` ```mojo 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` ```mojo def appending_writes(mut db: State, pipeline: String, from_step: Int = Int(0)) -> List[String] ``` `step writes ` for each dataset a step of `pipeline` writes without replacing it (in `writes=`, not `replaces=`), from its `from_step`th 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` ```mojo 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` ```mojo 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` ```mojo 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` ```mojo 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 ### `DONE_SQL` ```mojo 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). ### `APPENDING_SQL` ```mojo 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=`. ### `ALLOW_APPEND_HOW` ```mojo comptime ALLOW_APPEND_HOW = "pass --allow-append to append anyway" ```