.nornir/ingest-multi-guide.md

Cross-table single-fsync populate — `ingest_multi` — guide

Class: guide · Function: ingest_multi. The what/why is in ingest-multi-design.md.

When to use it

Reach for ingest_multi when one logical populate writes several tables at once — the canonical case is nornir's symbol scan, which writes ~4 tables per scan. The per-table paths commit each table separately, so that flush pays one fsync per table; ingest_multi commits all of them in one redb transaction = one fsync, while writing byte-identical data and giving each table exactly one new snapshot.

For a single table keep using append / ingest / ingest_parallel / ingest_pipelined — they are unchanged and already optimal for one table. ingest_multi only pays off across ≥2 tables (its win is fewer commits, not faster encode; the encode is the same all-core gatling fan-out).

Single-partition per table (same rule as append): every batch for a given table must share one identity-partition value, and that value must be non-null — a NULL identity-partition value is rejected rather than filed under "". Unpartitioned tables are the common case.

Each table may appear at most once per call. One batch commits at most one new snapshot per table, so listing the same table twice has no meaning and is rejected up front (before any encode) with table … appears more than once in one batch. If you grouped batches by table, merge that table's groups into a single (table, groups) entry — groups is already a Vec<Vec<RecordBatch>>.

Populate several tables in one fsync

use skade::{ingest_multi, WriteProps};

// One warehouse, several tables (create or load them however you like).
let symbols = wh.table_or_create("symbols", &symbols_schema).await?;
let refs    = wh.table_or_create("refs",    &refs_schema).await?;
let calls   = wh.table_or_create("calls",   &calls_schema).await?;

// Each table's contribution is Vec<Vec<RecordBatch>> — one inner Vec per output
// data file (same shape as ingest_parallel's `groups`).
let tables = vec![
    (symbols.inner().clone(), symbol_groups),
    (refs.inner().clone(),    ref_groups),
    (calls.inner().clone(),   call_groups),
];

// ONE redb transaction (one fsync) flips all three tables atomically.
let (advanced, stats) =
    ingest_multi(wh.catalog().as_ref(), tables, &WriteProps::default()).await?;

assert_eq!(stats.commits, 1);              // one commit for the whole flush
println!("{} rows across {} tables", stats.rows, advanced.len());

advanced is the updated iceberg table handles in input order (empty- contribution tables are skipped). stats.rows is the total across all tables; stats.commits is 1 (or 0 if every table's contribution was empty, in which case nothing is committed).

Compression / bloom / row-group knobs

Pass a tuned WriteProps just like the per-table paths — the codec, bloom columns, and row-group size all run inside the all-core encode, parallelised for free:

let props = WriteProps::new(skade::Compression::ZSTD(Default::default()))
    .row_group_size(128 * 1024);
let (advanced, stats) = ingest_multi(wh.catalog().as_ref(), tables, &props).await?;

Atomicity

The whole flush is all-or-nothing. If any table fails to encode/stage, the call returns an error and nothing is committed — no earlier table advances. On a catalog conflict the whole batch aborts (retryable). So a partial multi-table populate can never be observed: either every table advances or none does.

Memory

Like ingest_parallel, ingest_multi holds every table's encoded files in memory before the commit. For very large loads, size each call to your memory budget and chunk across calls (each ingest_multi call is its own one-fsync flush).

Confirming the collapse

skade_katalog::redb_commit_count() is the process-global fsync count of the write path. Measure a delta around your call to confirm one fsync:

let before = skade::skade_katalog::redb_commit_count();
let _ = ingest_multi(wh.catalog().as_ref(), tables, &WriteProps::default()).await?;
assert_eq!(skade::skade_katalog::redb_commit_count() - before, 1);

Cross-table single-fsync populate — ingest_multi — guide