<!-- nornir:class: design -->
# Un-serialized committer — `ingest_pipelined_overlap` — design
> **Class:** `design` · **Function:** `ingest_pipelined_overlap` · **Uses:**
> `write`, `commit`, `gatling::io`. Only the shape that runs; the serial-commit
> tail it overlaps (`ingest_pipelined`) is the untouched fallback, not an
> alternative design.
## f0 — why overlap the committer with the writer
skade's encode is memory-bandwidth-bound and all-core-saturated; the remaining
populate cost is **downstream of the encode** — the FileIO writes and the
snapshot-commit fsyncs. `ingest_pipelined` already overlaps the *encode* with the
*writes* (the async `gatling::io::run_ordered` keeps ≤ `channel_depth` FileIO
writes in flight), but its three stages still run as **sequential blocks**:
encode ALL → write ALL → **then** a serial commit loop, each `fast_append` fsync
strictly after the last write has landed. So the tail of the ingest is a run of
commit fsyncs with nothing to overlap them — pure serial fsync latency.
For a **single table** the commits are *inherently* serial: Iceberg snapshot
history is linear, each `fast_append`'s parent is the current snapshot, so you
cannot parallelise or reorder them. What you **can** do — safely, byte-identically
— is start committing the first `files_per_commit` files the moment they are
written, **while the later files are still being written**. The commit fsync of
group *k* then overlaps the FileIO writes of groups *k+1…*, hiding the serial
fsync tail behind the ongoing write I/O (and, under backpressure, hiding write
latency behind the fsync). That is the un-serialization: the committer stops being
a stage that runs *after* the writer and becomes a consumer that runs *beside* it.
This is purely additive. `append` / `ingest_parallel` / `ingest_pipelined` and the
one-fsync multi-table `ingest_multi` are all untouched. `ingest_pipelined_overlap`
is the opt-in overlap-commit fast lane; `ingest_pipelined` stays the proven
serial-commit path.
## f1 — `write::ingest_pipelined_overlap` (the pipeline)
`write.rs` · `ingest_pipelined_overlap(_props)(catalog, table, groups, files_per_commit, channel_depth, …) -> Result<(IceTable, IngestStats)>`
1. **Encode stage** (`encode_pipelined`, §f2): the shared all-core burn — every
group encoded into an in-memory `EncodedFile` in input (submission) order.
Byte-identical to `ingest_pipelined`'s encode.
2. **Writer future**: drives ≤ `channel_depth` FileIO writes in flight (no-barrier
`FuturesUnordered`), each carrying its submission index. A small reorder buffer
(≤ `channel_depth` held) re-sequences completions into submission order and
streams each in-order `(DataFile, rows)` into a bounded channel. Identical write
payloads, file names (`-{seq:08}.parquet`) and prunable `DataFile` footer stats
to the serial path — only the *delivery* is streamed instead of collected.
3. **Committer future**: one sequential consumer draining the channel, committing
every `files_per_commit` files via `commit_files` → `fast_append` (each its own
ordered snapshot), advancing its own `table` handle.
The writer and committer run **concurrently on one task** via `tokio::try_join!`
(no `spawn` → no `Send`/`'static` bound, so the `&dyn Catalog` borrow is free):
while the committer awaits an fsync, the runtime polls the writer's in-flight
FileIO writes; while the writer awaits channel admission (a full bounded channel),
the runtime polls the committer. That cooperative interleave **is** the overlap.
The bounded channel (`capacity = channel_depth`) is the backpressure: the writer
runs at most `channel_depth` finished files ahead of the committer.
## f2 — `encode_pipelined` (shared encode stage)
`write.rs` · `encode_pipelined(table, groups, props) -> Result<Vec<EncodedFile>>`.
The encode half of the pipeline, hoisted so **both** the serial
(`ingest_pipelined_props`) and overlapped (`ingest_pipelined_overlap_props`) tails
share it verbatim. One `spawn_blocking` runs the ONE fork-join engine
(`znippy_zoomies::gatling_forkjoin::gatling_map_owned`) so the async runtime stays
free during the all-core burn; recast + `partition_key_for` + `encode_parquet`
happen inside each worker (parallelised) against an immutable table handle.
`gatling_map_owned` returns results in **input order**, so `encoded[i]` is group
`i` — the submission order both write tails preserve. Because the encode is
literally the same code, the two paths emit **byte-identical** files.
## f3 — what stays identical, what changes
**Identical** to `ingest_pipelined` (the guarantees the tests pin): the file
bytes, the submission-order file sequence, the per-`files_per_commit` snapshot
grouping, the snapshot count, `IngestStats.rows`, and `IngestStats.commits`. The
resulting table is indistinguishable — same rows, same order, same snapshots.
**Changed**: only the *timing* — commit fsyncs overlap writes instead of running
strictly after all writes.
**Failure contract** (the one honest caveat): like `ingest_pipelined`'s multi-
commit contract, this is **not** atomic across its commits. In the serial path a
mid-loop commit error already leaves the earlier snapshots durable; here, because
commits stream, a *write* failure on a later group can also land after an earlier
group's snapshot is durable — a prefix may be committed when the ingest errors. An
error short-circuits `try_join!`, which drops (cancels) the committer at its next
await; an in-flight `redb` write txn dropped un-committed simply rolls back, so no
*partial* snapshot is ever produced. For all-or-nothing semantics use
`ingest_multi` (one atomic redb txn) instead. This is why the overlap path is a
separate opt-in function, not a silent change to `ingest_pipelined`.
## f4 — what is asserted (RED-when-broken)
`skade/tests/ingest_overlap.rs` pins the overlap path to the proven serial path by
**direct equality**, so any drop / duplicate / reorder / mis-group in the pipeline
diverges the two and fails the test:
- **matches-serial**: the same 40-file / `fpc`=4 / `depth`=4 workload ingested via
`ingest_pipelined` and `ingest_pipelined_overlap` yields identical row content
(`id → score-bits + name`), identical read-back **row order** (the reorder-buffer
guard), identical snapshot count (one per commit group), and identical
`IngestStats.commits`. Verified RED: mis-grouping the committer (never flushing
mid-loop → 1 commit instead of 10) fails the commit-count / snapshot-count
assertions.
- **empty + single**: an empty ingest commits nothing (0 snapshots, no deadlock);
a single-file ingest at `depth`=1 is exactly one snapshot with all rows.
## f5 — bench arm (needs the Loki HEAVY pass)
`bench/examples/nornir-bench.rs` · `skade.committer_overlap`: `ingest_pipelined`
(serial) vs `ingest_pipelined_overlap` on a **commit-bound** shape (many small
files, small `files_per_commit` → many fsyncs), reporting rows/sec per lane, the
commit count of each (equal — same grouping), and the speedup. LIGHT here. On
local NVMe the win is small — writes are fast relative to fsync — and grows as
commit/fsync latency dominates (slow or remote FileIO, e.g. S3). The authoritative
numbers come from the serial Loki HEAVY pass
(`SKADE_OVL_FILES`/`SKADE_OVL_ROWS`/`SKADE_OVL_FPC`/`SKADE_OVL_DEPTH`).