# orca.pipeline

> A module of Orca's API.

Pipelines: a named list of steps, checked and run in order.

```mojo
comptime Taxi = Pipeline["taxi", Download, Load, Compact]

var result = Taxi.run(parse_months("2024-01..2024-03"))
```

A pipeline is partitioned by month unless it says otherwise with
`partitions=` (`orca.partitions`): its first step takes the partition key,
or a `Seq` of them for a gather step.

Steps are listed after the step they follow, so the declared order is a
valid run order; `check` asserts it at compile time, along with every edge's
types and unique names. Every function that runs a pipeline calls `check`,
so a pipeline that breaks a rule doesn't compile.

`execute` is the executor: it runs the steps that haven't succeeded yet and
reports each attempt to a `Store`, either `MemoryStore` (what `run` uses: a
run lives as long as the process) or `orca.state.PgStore`, which makes a run
resumable. Each step's output is kept encoded by dista's codec, as Postgres
holds it. A step that raises is retried up to its `retries`, then fails its
partition: the steps after it for that partition, and any gather step over
it, are skipped, while other partitions carry on.

## Structs

### `StepRun`

```mojo
struct StepRun
```

One step for one partition (or for the run, for a gather step).

Implements: `Copyable`, `Movable`

Fields:

- `step: String`
- `partition: String`: Empty for a gather step.
- `status: String`
- `output: List[UInt8]`: The encoded output, when the step succeeded.
- `error: String`
- `attempt: Int`: The number of the attempt that settled it; 0 for a skipped step.

#### `__init__`

```mojo
def __init__(out self, step: String, partition: String, status: String, var output: List[UInt8], error: String, attempt: Int = Int(0))
```

### `RunResult`

```mojo
struct RunResult
```

What a run did: every step and partition, in the order they ran.

Implements: `Movable`

Fields:

- `steps: List[StepRun]`

#### `__init__`

```mojo
def __init__(out self)
```

#### `succeeded`

```mojo
def succeeded(self) -> Bool
```

#### `find`

```mojo
def find(self, step: StringSpan, partition: StringSpan) -> Int
```

The index of `step` for `partition`, or -1.

#### `status`

```mojo
def status(self, step: StringSpan, partition: StringSpan) -> String
```

#### `output`

```mojo
def output[T: Movable](self, step: StringSpan, partition: StringSpan = "") -> T
```

The decoded output of `step` for `partition`.

### `Pipeline`

```mojo
struct Pipeline[name: StringSpan[ImmStaticOrigin], *Ts: StepLike, *, tasks: TaskSet = NoTasks, partitions: Partitions = Monthly, schedule: ScheduleLike = NoSchedule, fills: Fill = Previous, catchup: Bool = True, max_runs: Int = Int(1), run_retries: Int = Int(3), sensor: Bool = False, keys_per_run: Int = Int(1), workers: Int = Int(2)]
```

A pipeline named `name` of the steps `Ts`, in run order, whose steps call the dista `tasks`, over the keys of `partitions`; run by `schedule` for the partition its fires `fills` (`orca.schedule`), and, with `sensor`, when a dataset it reads gets a new snapshot (`orca.sensors`). A backfill puts `keys_per_run` keys in each run it queues: one for time keys by default, tens to hundreds for keyed partitions (`orca.keyed`), whose runs then each go through many keys. On Kubernetes, a run of a pipeline with `tasks` gets a cluster of `workers` dista workers (`orca.kube`); locally it forks `ORCA_WORKERS`.

The scheduler runs at most `max_runs` of its runs at once, and starts a
failed scheduled or backfilled run again up to `run_retries` times, with
backoff (`orca.scheduler`).

Implements: `Movable`

#### `K`

```mojo
comptime K = partitions.K
```

The partition key.

#### `check`

```mojo
def check()
```

Asserts at compile time that the pipeline is well formed.

#### `registry_hash`

```mojo
def registry_hash() -> UInt64
```

A hash over the pipeline's name, its steps' names, kinds, edges, functions and datasets, and its dista tasks: two binaries with the same hash run the same DAG.

#### `describe`

```mojo
def describe() -> String
```

The DAG as JSON: the pipeline's name, its registry hash, its partitions, each step's name, kind, upstream, function, types and datasets in run order, and its dista tasks.

#### `sensed_datasets`

```mojo
def sensed_datasets() -> List[String]
```

The datasets its steps read and none of them writes, once each, in step order, and a keyed pipeline's keys dataset: what a sensor watches.

#### `scheduled`

```mojo
def scheduled() -> Bool
```

#### `load_schedule`

```mojo
def load_schedule() -> Schedule
```

The schedule, parsed, with its zone loaded; raises for a pipeline without one, or one whose expression or zone is bad.

#### `fills_for`

```mojo
def fills_for(fire: Int64, s: Schedule) -> Self.K
```

The partition a fire at `fire` fills.

#### `run`

```mojo
def run(keys: List[Self.K], run: Int = Int(0)) -> RunResult
```

Runs every step over `keys`, in order, recording nothing, with a dista runtime of its own if the pipeline has tasks.

#### `execute`

```mojo
def execute[S: Store](mut store: S, var prior: RunResult, keys: List[Self.K], run: Int, runtime: Int = Int(0)) -> RunResult
```

Runs the steps over `keys` that haven't succeeded in `prior`, in order, reporting each attempt to `store`.

`prior` holds the steps a run had already finished when it's resumed,
with their outputs; they aren't run again. A step that raises is
retried up to its `retries`, then fails its partition. An error from
the store itself is raised: the run stops where it is, and can be
resumed from there. `runtime` is the address of the run's
`tasks.RT`, which steps reach through `Ctx.runtime`.

## Aliases

### `SUCCEEDED`

```mojo
comptime SUCCEEDED = "succeeded"
```

### `FAILED`

```mojo
comptime FAILED = "failed"
```

### `SKIPPED`

```mojo
comptime SKIPPED = "skipped"
```