orca.step

A step wraps a top-level def with a name, its input and output types, its kind, and the step it follows:

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):

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.

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:

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.

struct StepKind

def __eq__(self, other: Self) -> Bool

def write_to(self, mut writer: T)

struct NoStep

comptime A = NoneType

comptime R = NoneType

def step_name() -> StaticString

def kind() -> StepKind

def upstream() -> StaticString

def max_retries() -> Int

def backoff() -> Int

def timeout() -> Int

def code_version() -> Int

def upstream_function() -> StaticString

def function() -> StaticString

def input_type() -> StaticString

def output_type() -> StaticString

def dataset_reads() -> List[String]

def dataset_writes() -> List[String]

def dataset_replaces() -> List[String]

def check()

def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8])

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)]

comptime A = A_

comptime R = R_

def step_name() -> StaticString

def kind() -> StepKind

def upstream() -> StaticString

def max_retries() -> Int

def backoff() -> Int

def timeout() -> Int

def code_version() -> Int

def upstream_function() -> StaticString

def function() -> StaticString

def input_type() -> StaticString

def output_type() -> StaticString

def dataset_reads() -> List[String]

def dataset_writes() -> List[String]

def dataset_replaces() -> List[String]

def check()

def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8])

comptime A

comptime R

def step_name() -> StaticString

def kind() -> StepKind

def upstream() -> StaticString

def max_retries() -> Int

def backoff() -> Int

def timeout() -> Int

def code_version() -> Int

def upstream_function() -> StaticString

def function() -> StaticString

def input_type() -> StaticString

def output_type() -> StaticString

def dataset_reads() -> List[String]

def dataset_writes() -> List[String]

def dataset_replaces() -> List[String]

def check()

def run(mut ctx: Ctx, input: Span[UInt8], mut out: List[UInt8])

def defer(seconds: Int, why: String = "") -> Error

def deferred_for(error: StringSpan) -> Int

def deferred_why(error: StringSpan) -> String

comptime Mapped = StepKind(Int(0))

comptime Gather = StepKind(Int(1))