nordisk/skade

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