from orca.tables import append, scan
def load(mut ctx: Ctx, raw: RawFile) raises -> Loaded:
var src = Table.load_local(raw.table)
var batches = scan(ctx, src).to_batches()
var dst = Table.load_local(OUT)
var id = append(ctx, dst, transform(batches))
return Loaded(raw.month, id)
scan reads a table at a pinned snapshot: the one this step's first
attempt read, or its current snapshot if this is the first attempt, which
is then recorded. A retry or a resume of the step sees the same data,
whatever was committed in between.append and commit put orca.* properties in the summary of the
snapshot they commit (Ctx.summary): pipeline, run, step, partition,
attempt, commit number, code, and the snapshots read. Before writing
anything, they look through the table's snapshots for one this step
already made in this run (Ctx.made). If an attempt committed and was
killed before Postgres heard of it, the retry finds that snapshot and
returns it instead of committing the data a second time.
A table is known by its location, metadata.location, and, when it's in
the warehouse, by its dataset name. ORCA_WAREHOUSE is the root of a
filesystem catalog: dataset nyc.trips is the table at
<warehouse>/nyc/trips, where the warehouse is a directory or a location
in object storage (s3://bucket/warehouse, as iceberg's FileIO reads it,
from the AWS_* environment). table loads a dataset the step declared, and
ensure_table creates one it writes if it's missing, and adds the
columns the step's schema has and the table lacks. scan refuses a
dataset the step didn't declare in reads=, and append and commit one it
didn't declare in writes=. A table outside the warehouse (or with
ORCA_WAREHOUSE unset) is used by location as before, and belongs to no
dataset.
Files written on a worker. A dista task can write a table's data
files itself and send back only their entries, so the step commits
metadata rather than receiving the rows:
def write_hour(day: String, hour: Int) raises -> DataFiles:
# On a worker: no Ctx, nothing checked or recorded.
return write_files(load_dataset(EVENTS), fetch(day, hour))
comptime Hour = Task2[String, Int, DataFiles, write_hour]
def load_day(mut ctx: Ctx, d: Day) raises -> Int64:
var t = ensure_table(ctx, EVENTS, schema())
if find_commit(ctx, t) >= 0: # a retry: don't write the files again
return append_files(ctx, t, DataFiles())
ref rt = ctx.runtime[Tasks[Hour]]()
var got = rt.get(rt.remote[Hour](String(d), 0))
return append_files(ctx, t, got)
write_files (or a FileWriter, a batch at a time) writes
batches into a table's data/ without committing them, and returns their
entries as DataFiles, a dista Codec over iceberg's
DataFile.encode. A struct holding a DataFiles and plain fields (row
counts, timings) crosses processes by dista's reflection, with no codec
of its own.append_files commits entries written elsewhere as one append
snapshot, as append commits the files it writes: the same writes=
check, summary and recognition. A retry that finds its commit returns it
and commits nothing; a step checks find_commit first, as above, so a
retry doesn't have the workers write the files again. Files written by an
attempt that died before committing them are left unreferenced in
data/, as with an interrupted append.
Replacing partitions. A new run of a key (orca backfill --rerun)
isn't a retry: find_commit doesn't recognise an earlier run's commits, so
append adds the key's rows a second time. A step that recomputes a key's
rows declares the dataset in replaces= rather than writes=, and its
append and append_files to it then replace:
comptime LoadDay = Step[
"load_day", Day, Int64, load_day, Mapped, replaces=[EVENTS],
]
- The run's first commit to a partition is an
overwrite that removes the
files this run didn't commit from the partitions its new files land in,
and adds them; its later commits find only the run's files there and
append. A retry recognises its commits as before, so nothing is
overwritten twice or lost.
- Another writer's file in such a partition, added after the run's first
commit to the table, means two runs are replacing it at once: the commit
raises, committing nothing, and only one run keeps its rows.
replace_partitions and replace_partition_files do the same for one
call, to any table. The summary has orca.replaced-files and
orca.replaced-records.
commit refuses to remove files that aren't all in the current snapshot,
on every try: a removal planned before another writer replaced them
would otherwise keep both writers' rows.
This module imports iceberg, so only pipelines that use tables build it;
orca itself doesn't.