<!-- nornir:class: design -->
# Cross-table single-fsync populate — `ingest_multi` — design
> **Class:** `design` · **Function:** `ingest_multi` · **Uses:** `write`,
> `commit`, `catalog`, `atomic`. Only the shape that runs; the populate
> bottleneck it removes (one fsync **per table**) is the motivation, not an
> alternative — there is no second design here.
## f0 — why one commit for many tables
skade's encode is already memory-bandwidth-bound and all-core-saturated (`write`
≈ 35 M rows/s, 12/12 cores). The remaining populate cost is **downstream of the
encode**: the snapshot-commit fsync. Every `fast_append` is one Iceberg snapshot
= one `redb` write transaction = **one fsync** (`Store::group_commit`, at
`Durability::Immediate`). A populate that writes several tables as one logical
unit — e.g. nornir's symbol scan writing 4 tables per scan — therefore pays
**4 snapshots = 4 fsyncs** per scan, one per table, even though the four are a
single flush.
The atomic multi-table primitive already existed: `RedbCatalog::atomic_release_raw`
flips N table pointers in **one** `redb` transaction (all-or-nothing, one fsync),
and it already backs `Warehouse::atomic_release_with_lineage`. `ingest_multi` is
the *write path's* use of it: all-core-encode every table's data files exactly as
`ingest_parallel` does, then commit them all through `atomic_release_raw` so the
whole flush costs **one fsync** instead of N. Each table still gains **exactly one**
new snapshot — only the commit *transactions* collapse from N to 1.
This is purely additive. The per-table `append` / `ingest_parallel` /
`ingest_pipelined` paths are untouched; single-table callers keep their exact
behaviour. `ingest_multi` is the opt-in multi-table fast lane.
## f1 — `write::ingest_multi` (the orchestration)
`write.rs` · `ingest_multi(catalog: &RedbCatalog, tables: Vec<(IceTable, Vec<Vec<RecordBatch>>)>, props) -> Result<(Vec<IceTable>, IngestStats)>`
For each `(table, groups)` in input order:
1. **Encode + write** (`encode_and_write_files`, §f2): all-core-encode the table's
groups into Parquet files and write the bytes to the table's `FileIO` — no
commit. Returns the file-prunable `DataFile`s + a row count. A table whose
contribution is empty produces no files and is skipped (matching the per-table
paths, which commit nothing for empty input).
2. **Stage the fast append** (`Transaction::stage_fast_append`, §f3): produce the
`(updates, requirements)` this table's `fast_append` would commit — writing the
same manifest + manifest-list to storage — **without** touching the catalog
pointer. Push `(ident, requirements, updates)` onto the batch.
Then **one** `catalog.atomic_release_raw(batch)` flips every staged table in a
single `redb` transaction (one fsync). Returns the advanced handles + `IngestStats`
whose `commits` is `1` for a non-empty flush (`0` if every table was empty — in
which case no `redb` write happens at all; an empty `atomic_release_raw` would
otherwise fsync an empty txn).
**All-or-nothing.** A failure in step 1 or 2 for any table returns before
`atomic_release_raw` is called, so **nothing** is committed — no earlier table
advances. A catalog conflict inside `atomic_release_raw` aborts the whole batch.
Data + manifest files written for the failed flush are left as orphan blobs
(standard Iceberg orphan-file semantics), exactly as a failed `atomic_release`.
Single-partition per table (same rule as `append`): the encode derives one
partition key per table via `partition_key_for`; a batch spanning partitions, or
carrying a NULL identity-partition value, errors loudly rather than mis-tagging
files.
**Distinct idents per batch.** Every staged entry is built against the state
loaded *before* the batch, so two entries for the same table would each fold
their `AddSnapshot` onto the identical base instead of chaining. Worse, the
conflict-check loop in `atomic_release_raw` completes before any insert, so both
entries observe `current == base` and pass — the two inserts then hit the same
catalog key and the second wins, leaving the first entry's snapshot referenced
by nothing while `IngestStats` counts its rows. `ingest_multi` therefore rejects
a repeated `TableIdent` up front (before any encode), and `atomic_release` /
`atomic_release_raw` reject one too, so no other caller of the shared primitive
can reach the same corruption.
## f2 — `encode_and_write_files` (encode + write, no commit)
`write.rs` · the encode+write half of `ingest_parallel_props`, hoisted so the
multi-table lane can gather **every** table's files before a single commit. It
recasts each group to the table's field-id Arrow schema, derives the single
partition key, all-core-encodes via the one fork-join engine
(`znippy_zoomies::gatling_forkjoin::gatling_map_owned` over `encode_parquet`),
then writes each in-memory Parquet file to the table's `FileIO` and builds its
prunable `DataFile` from the file's own footer stats (`build_data_file`). Every
primitive is shared with the per-table path, so the files it emits are
**byte-identical** to what `ingest_parallel` writes — the collapse is in the
commit, never in the bytes.
## f3 — `Transaction::stage_fast_append` (the vendored seam)
`vendor/iceberg-arrow58` · `stage_fast_append(table, data_files) -> (updates, requirements)`
iceberg-rust 0.9.1 keeps `FastAppendAction::commit` and the `TransactionAction`
trait `pub(crate)`, and `TableCommit`'s builder is crate-private — so an external
caller cannot obtain a fast-append's `(updates, requirements)` to hand to
`atomic_release_raw`. `stage_fast_append` is the public projection of exactly that
path: it runs the identical `FastAppendAction` (writing the same manifest +
manifest-list files, returning the same `AddSnapshot` / `SetSnapshotRef` updates
and `UuidMatch` / `RefSnapshotIdMatch` requirements) and only leaves the catalog
step to the caller. The produced snapshot is therefore byte-identical to a
standalone fast-append; several tables' appends can then be folded into one atomic
catalog commit. (Additive to the fork, matching the existing public `refs()`
accessor precedent.)
## f4 — observing the collapse: the fsync counter
`skade_katalog::redb_commit_count()` is a process-global count of durable `redb`
write transactions on the **data / pointer-advance** paths — `group_commit`,
`atomic_release{,_raw}`, and `swap_table`. Each is one fsync at `Immediate`
durability, so this **is** the write path's fsync count. It advances by **1**
across an `ingest_multi` of N tables and by **N** across N separate
`ingest_parallel` calls — the observable that proves (and regression-guards) the
collapse. DDL commits (create/drop namespace/table) do not bump it. It is a
metrics-class `Relaxed` atomic, negligible next to the fsync it counts.
## f5 — what is asserted (RED-when-broken)
`skade/tests/ingest_multi.rs`:
- **one-fsync + byte-identical**: a 3-table `ingest_multi` advances
`redb_commit_count` by exactly **1** (vs **3** for three separate
`ingest_parallel` calls); every table gets **exactly one** snapshot; and each
table's rows (`id → score-bits + name`) are byte/row-identical to the
separate-ingest baseline. Verified RED: a per-table commit loop makes the delta
**3**, failing the assertion.
- **all-or-nothing**: a mid-flush encode failure (a later table whose batch has
the wrong column count) fails the whole call and commits **nothing** — the
earlier, already-staged table has **0** snapshots and **0** rows.