<!-- nornir:class: guide -->
# Cross-table single-fsync populate — `ingest_multi` — guide
> **Class:** `guide` · **Function:** `ingest_multi`. The *what/why* is in
> [`ingest-multi-design.md`](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
```rust
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:
```rust
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:
```rust
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);
```