# orca.step > A module of Orca'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"], ] ``` `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 ``` #### `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] ``` #### `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), reads: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None)), writes: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None))] ``` 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 ``` #### `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] ``` #### `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. #### `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=`. #### `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. ## 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.