`orca.cli`

orca_main: the command line every pipeline binary gets.

taxi describe                              the DAG as JSON
taxi run --partitions 2024-01..2024-03     starts a run and runs it
taxi resume <run> [--allow-new-code]       continues a run at what's unfinished
taxi status [<run>]                        the latest runs, or one run's steps
taxi sense                                 the current snapshot of each dataset it reads
labels keys                                a keyed pipeline's keys snapshot, then its keys

For a keyed pipeline (orca.keyed), --partitions takes names: a list, @all or @file:<path>. A name its keys dataset doesn't have now is refused, as a malformed key is, with exit code 2.

With ORCA_DATABASE_URL set, runs are recorded in Postgres and can be resumed; without it, run runs in memory and resume and status aren't available. ORCA_STALE_SECONDS (default 30) is how old a run's heartbeat must be before another process may resume it.

A pipeline with dista tasks gets a runtime per run, started before anything else so its forked processes inherit no connection or thread; run with DISTA_ROLE set, the binary is a dista head or worker instead (see orca.tasks). With ORCA_LOGS set, a directory or a location in object storage (orca.logs), attempts log there, the run's dista timeline is written beside them, and status <run> shows each step's latest log.

sense prints a line per dataset the pipeline reads and doesn't write, <dataset> <snapshot> <orca|other> <partitions|->: its table's current snapshot in ORCA_WAREHOUSE (-1 for none), whether an orca step made it, and the partition keys its summary says it holds. The scheduler polls it for a sensor pipeline (orca.sensors).

run and resume print one line per step and partition, then the run's status. Exit codes: 0 when the run succeeded, 1 when a step didn't or the run was cancelled, 2 for a usage error, 3 for any other error (the database, a run that can't be resumed).

Functions

`seed_random`

def seed_random()

Seeds std.random from the clock and the process ID.

Unseeded, every process starts from the same state, so pipeline processes committing to one table at once draw the same Iceberg snapshot IDs and file UUIDs (iceberg takes both from std.random and doesn't seed it), and overwrite each other's manifest lists.

`orca_main`

def orca_main[P: AnyStruct[Pipeline[P.name, P.Ts, tasks=P.tasks, partitions=P.partitions, schedule=P.schedule, fills=P.fills, catchup=P.catchup, max_runs=P.max_runs, run_retries=P.run_retries, sensor=P.sensor, keys_per_run=P.keys_per_run, workers=P.workers]]]()

Parses the command line and runs the command on P.

`report`

def report(res: RunResult)

One line per step and partition, then the run's status.

Aliases

comptime USAGE = "usage:\n  <pipeline> describe\n  <pipeline> run --partitions KEY[..KEY][,...]   (YYYY-MM, or YYYY-MM-DD daily)\n  <pipeline> run --partitions NAME[,...]|@all|@file:<path>   (keyed)\n  <pipeline> resume <run> [--allow-new-code]\n  <pipeline> status [<run>]\n  <pipeline> sense\n  <pipeline> keys\n  <pipeline> fires [<count>] [--after YYYY-MM-DDTHH:MM[:SS]Z]"