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"],
]

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 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 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), reads: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None)), writes: List[StringSpan[ImmStaticOrigin]] = List(__list_literal__=NoneType(None))]

comptime A = A_

comptime R = R_

def step_name() -> StaticString

def kind() -> StepKind

def upstream() -> StaticString

def max_retries() -> 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 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 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 check()

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

comptime Mapped = StepKind(Int(0))

comptime Gather = StepKind(Int(1))