nordisk/skade

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 DataFiles + 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.