# orca.step > A module of morca's API. Steps: the functions a pipeline runs, and the edges between them. A step wraps a top-level `def` with a name, its input and output types, its kind, and the step it follows: ```mojo def download(mut ctx: Ctx, month: Month) raises -> RawFile: ... def load(mut ctx: Ctx, raw: RawFile) raises -> Loaded: ... def compact(mut ctx: Ctx, loaded: Seq[Loaded]) raises -> Int64: ... comptime Download = Step["download", Month, RawFile, download, Mapped] comptime Load = Step["load", RawFile, Loaded, load, Mapped, after=Download] comptime Compact = Step["compact", Seq[Loaded], Int64, compact, Gather, after=Load] ``` A mapped step runs once per partition of a run, a gather step once per run. The edge is checked when the pipeline is compiled: | Step | Its input must be | |---|---| | first, mapped | the pipeline's partition key, such as `Month` | | first, gather | a `Seq` of the key, every partition of the run | | mapped, after a mapped step | that step's output | | gather, after a mapped step | `Seq` of that step's output, one per partition in run order | | gather, after a gather step | that step's output | A mapped step can't follow a gather step. Types are compared by their reflected names, which include their module; the first step's input is checked by `Pipeline.check`, which knows the key. A step declares the datasets it reads and writes, as `namespace.table` names in the warehouse (`orca.tables`): ```mojo comptime Clean = Step[ "clean", Month, Int64, clean, Mapped, reads=["raw.events"], writes=["lake.events"], ] ``` A dataset in `replaces=` instead is one the step writes whose partitions a new run *replaces* rather than appends to: `orca.tables.append` and `append_files` to it make the run's first commit to each partition an `overwrite` that removes what earlier runs left there, so `orca backfill --rerun` leaves one copy of a key's rows (`orca.tables`, "Replacing partitions"). A dataset in `replaces=` counts as written; it needn't be in `writes=` as well. ```mojo comptime Load = Step[ "load", Day, Int64, load, Mapped, replaces=["gharchive.events"], ] ``` A step that fails is tried again up to `retries` times in the same run, at once, or after `backoff_seconds` doubling with each retry: ```mojo comptime Fetch = Step[ "fetch", Month, RawFile, fetch, Mapped, retries=3, backoff_seconds=30, ] ``` A step that can't do its work yet, such as a download that isn't published or an API that's out of requests, raises `defer(seconds, why)` instead of failing: the attempt ends `deferred`, spending no retry, and its run ends `deferred` and is started again by the scheduler that many seconds later, without counting against `run_retries` (`orca.scheduler`). `timeout_seconds=N` bounds an attempt: one still going N seconds after it started is stopped (`orca.heartbeat`), as a failed attempt, and its run fails, to be retried by the scheduler as any failed run is. A step can't be interrupted where it is, so the run's process exits: its other partitions and the step's own retries wait for the run's retry. `orca.tables` refuses a scan of a dataset the step didn't declare in `reads`, and a commit to one it didn't declare in `writes`, so the graph of pipelines and datasets that `describe` gives is what the steps do. ## Structs ### `StepKind` ```mojo struct StepKind ``` Whether a step runs once per partition or once per run. Implements: `Copyable`, `Equatable`, `ImplicitlyCopyable`, `Movable`, `RegisterPassable`, `TrivialRegisterPassable`, `Writable` Fields: - `value: Int` #### `__eq__` ```mojo def __eq__(self, other: Self) -> Bool ``` #### `write_to` ```mojo def write_to(self, mut writer: T) ``` ### `NoStep` ```mojo struct NoStep ``` What a first step follows. Implements: `Movable`, `StepLike` #### `A` ```mojo comptime A = NoneType ``` #### `R` ```mojo comptime R = NoneType ``` #### `step_name` ```mojo def step_name() -> StaticString ``` #### `kind` ```mojo def kind() -> StepKind ``` #### `upstream` ```mojo def upstream() -> StaticString ``` #### `max_retries` ```mojo def max_retries() -> Int ``` #### `backoff` ```mojo def backoff() -> Int ``` #### `timeout` ```mojo def timeout() -> Int ``` #### `code_version` ```mojo def code_version() -> Int ``` #### `upstream_function` ```mojo def upstream_function() -> StaticString ``` #### `function` ```mojo def function() -> StaticString ``` #### `input_type` ```mojo def input_type() -> StaticString ``` #### `output_type` ```mojo def output_type() -> StaticString ``` #### `dataset_reads` ```mojo def dataset_reads() -> List[String] ``` #### `dataset_writes` ```mojo def dataset_writes() -> List[String] ``` #### `dataset_replaces` ```mojo def dataset_replaces() -> List[String] ``` #### `check` ```mojo def check() ``` #### `run` ```mojo def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8]) ``` ### `Step` ```mojo struct Step[name: StringSpan[ImmStaticOrigin], A_: Copyable & Deinitable, R_: Copyable & Deinitable, f: def(mut Ctx, A_) raises thin -> R_, kind_: StepKind, *, after: StepLike = NoStep, retries: Int = Int(0), backoff_seconds: Int = Int(0), timeout_seconds: Int = Int(0), reads: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None)), writes: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None)), replaces: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None)), version: Int = Int(0)] ``` A step named `name` running `f` on an `A_` to make an `R_`. Implements: `Movable`, `StepLike` #### `A` ```mojo comptime A = A_ ``` #### `R` ```mojo comptime R = R_ ``` #### `step_name` ```mojo def step_name() -> StaticString ``` #### `kind` ```mojo def kind() -> StepKind ``` #### `upstream` ```mojo def upstream() -> StaticString ``` #### `max_retries` ```mojo def max_retries() -> Int ``` #### `backoff` ```mojo def backoff() -> Int ``` #### `timeout` ```mojo def timeout() -> Int ``` #### `code_version` ```mojo def code_version() -> Int ``` #### `upstream_function` ```mojo def upstream_function() -> StaticString ``` #### `function` ```mojo def function() -> StaticString ``` #### `input_type` ```mojo def input_type() -> StaticString ``` #### `output_type` ```mojo def output_type() -> StaticString ``` #### `dataset_reads` ```mojo def dataset_reads() -> List[String] ``` #### `dataset_writes` ```mojo def dataset_writes() -> List[String] ``` #### `dataset_replaces` ```mojo def dataset_replaces() -> List[String] ``` #### `check` ```mojo def check() ``` #### `run` ```mojo def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8]) ``` ## Traits ### `StepLike` ```mojo undefined ``` What a pipeline needs from a step, without knowing its function. #### `A` The input type. #### `R` The output type. #### `step_name` ```mojo def step_name() -> StaticString ``` The step's name in its pipeline; empty for `NoStep`. #### `kind` ```mojo def kind() -> StepKind ``` #### `upstream` ```mojo def upstream() -> StaticString ``` The name of the step this one follows; empty for a first step. #### `max_retries` ```mojo def max_retries() -> Int ``` How many times a failed attempt is retried, in the same run. #### `backoff` ```mojo def backoff() -> Int ``` Seconds before the first retry, doubling before each after it. #### `timeout` ```mojo def timeout() -> Int ``` Seconds an attempt may take before it's stopped; 0 for no limit. #### `code_version` ```mojo def code_version() -> Int ``` The step's `version=`: raising it says its outputs must be made again, so its pipeline's keys are no longer done; 0 when not set. #### `upstream_function` ```mojo def upstream_function() -> StaticString ``` The function of the step this one follows; empty for a first step. #### `function` ```mojo def function() -> StaticString ``` The step function's linkage name. #### `input_type` ```mojo def input_type() -> StaticString ``` #### `output_type` ```mojo def output_type() -> StaticString ``` #### `dataset_reads` ```mojo def dataset_reads() -> List[String] ``` The datasets the step declared in `reads=`. #### `dataset_writes` ```mojo def dataset_writes() -> List[String] ``` The datasets the step declared in `writes=` or `replaces=`. #### `dataset_replaces` ```mojo def dataset_replaces() -> List[String] ``` The datasets the step declared in `replaces=`: written, with a new run replacing their partitions. #### `check` ```mojo def check() ``` Asserts at compile time that the step's input matches its edge. #### `run` ```mojo def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8]) ``` Decodes the input, calls the function, appends the encoded output. ## Functions ### `defer` ```mojo def defer(seconds: Int, why: String = "") -> Error ``` An error a step raises to say it can't do its work yet: try again in `seconds` (at least 1). Raised, it ends the attempt `deferred` instead of failed, and spends none of the step's retries or the run's. ### `deferred_for` ```mojo def deferred_for(error: StringSpan) -> Int ``` The seconds a `defer` error asks to wait, or 0 for another error. ### `deferred_why` ```mojo def deferred_why(error: StringSpan) -> String ``` The reason a `defer` error gives. ## Aliases ### `Mapped` ```mojo comptime Mapped = StepKind(Int(0)) ``` Runs once per partition of the run. ### `Gather` ```mojo comptime Gather = StepKind(Int(1)) ``` Runs once per run, over every partition's output of the step before.