nordisk/skade

All-core footer-stat build — encode_and_stat_parallel — design

Class: design · Function: ingest_parallel / ingest_multi / ingest_pipelined[_overlap] · Uses: write.rs, znippy-zoomies/gatling_forkjoin, parquet. Only the shape that runs. This is the write-side complement to the read-path local-fd streaming: the read fix took a serial pre-slurp off the decode; this takes the serial footer-parse off the write.

f0 — the hot cost: a serial footer parse behind an all-core encode

skade's bulk write (ingest_parallel, the mashup's 35.4 M rows/s column, and the per-table half of ingest_multi) has two stages:

1. Encode — every group's Parquet file encoded in memory across all cores via the ONE fork-join engine (znippy-zoomies gatling_forkjoin). This stage is memory-bandwidth-bound and saturates ~11/12 cores.
2. Write — a single sequential loop that, for each encoded file, (a) calls build_data_file to parse the file's Parquet footer and build its file-prunable DataFile (per-column lower/upper_bounds, value_counts, null_value_counts, column_sizes, split_offsets), then (b) writes the bytes to FileIO, then commits every files_per_commit files.

Step (2a) is CPU work — parsing the footer and converting every row group's column statistics into Iceberg Datums — and it ran serially, one file at a time, with the other ~11 cores idle (the encode had already finished). By the core-saturation law that idle is a bug: a parallel write path should keep ~11/12 cores busy, and this stage dropped to one. It is the exact write-side twin of the read-path bug where the whole dataset was slurped serially before the decode.

f1 — the fix: fold the footer parse into the encode worker

write.rs · encode_and_stat_parallel(target, ice_schema, data_dir, prefix, partition, groups, props) -> Result<Vec<(Bytes, DataFile, u64)>>

Each group is paired with its submission index (enumerate) and handed to gatling_map_owned. Inside each worker, on its own core, the file is both:

• encoded (encode_parquet → the same bytes as before), and
• stat-built (build_data_file over those bytes → its prunable DataFile),

so the footer parse now rides the all-core burn instead of the serial tail. The worker returns (bytes, DataFile, record_count). gatling_map_owned returns results in input order, so the file name {prefix}-{seq:08}.parquet (and hence the whole DataFile) is deterministic regardless of which worker ran which group.

The serial writer stage is now pure I/O: it just streams each file's bytes to the DataFile's own path (FileIO.write) and commits every files_per_commit. No CPU parse strands the cores anymore.

Every call site uses one shared per-file core — encode_stat_one(target, ice_schema, data_dir, prefix, seq, partition, batches, props) — which does the whole CPU half for one file (encode + footer parse + DataFile build). The only thing that differs between the paths is where partition comes from, so it is a parameter, never twinned logic (L5):

• encode_and_stat_parallel → ingest_parallel_props (the default bulk write) and encode_and_write_files (the encode+write half of ingest_multi, the cross-table single-fsync populate). One partition for the whole call (the single-partition rule).
• encode_pipelined → ingest_pipelined_props (serial-commit tail) and ingest_pipelined_overlap_props (overlapped-commit tail). Partition derived per group (a pipelined ingest may span partitions). See f1a.

f1a — the pipelined paths: parse off the tokio runtime, onto gatling

ingest_pipelined and ingest_pipelined_overlap share the encode stage encode_pipelined, then run a write stage on the async runtime — the serial tail drives the FileIO writes through gatling::io::run_ordered; the overlap tail runs a writer future beside a committer future. Previously encode_pipelined returned only (bytes, record_count, partition) and each file's build_data_file (the footer parse) ran inside those async write jobs on the tokio runtime — off the strictly-serial block, but a CPU parse driven by the runtime's reorder task (one at a time), not on the all-core gatling_forkjoin burn. By the core-saturation law that is the same bug as the non-pipelined serial tail, just relocated to the runtime.

Now encode_pipelined builds each file's prunable DataFile inside its gatling worker (via the shared encode_stat_one), stamping {prefix}-{seq:08}.parquet from the group's input index — so the footer parse rides the all-core encode burn. Both write stages become pure I/O: they take the already-built DataFile, write its bytes to data_file.file_path(), and commit. Same reorder/overlap logic, same file names, same submission order — only the placement of the parse moved.

f2 — why it is byte-identical (L2)

Nothing about the output changes — only where the parse runs:

• Data-file bytes: produced by the same encode_parquet (deterministic Parquet encode — no wall-clock in the footer), byte-for-byte identical.
• DataFile manifest entry: built by the same build_data_file over the same bytes and the same {prefix}-{seq:08} path; DataFile derives Eq, and the manifest is a deterministic function of it, so the manifest bytes are identical.
• Snapshot shape: unchanged — same fast_append, same files_per_commit grouping, same one-snapshot-per-table result.

f3 — what is asserted (RED-when-broken)

skade/src/write.rs #[cfg(test)]:

• write_output_is_byte_identical — over a 4-file × 50k-row, 5-row-group-per- file fixture, the parallel encode+stat path is compared directly against the old serial build (encode, then build_data_file in a serial loop) using a fixed prefix + data_dir so the paths line up: per file, the Parquet Bytes must be equal (data-file byte-identity) and the DataFile must be equal (assert_eq! on the Eq-deriving struct = manifest-entry byte-identity). Verified RED: XOR-ing the seq inside the helper (files mis-numbered) makes the DataFile paths diverge and the compare fails; reverted.
• pipelined_output_is_byte_identical — the same proof for the pipelined path: encode_pipelined (footer parse now in the gatling worker) is compared per file against the OLD placement — per-group recast + partition derivation, then build_data_file in a serial loop (as the old async write jobs did) — over the same 4-file × 50k-row / 5-row-group fixture with a fixed prefix. Bytes and DataFile must be equal. Verified RED: seq ^ 1 inside encode_pipelined diverges the DataFile.file_path → the struct compare fails; reverted.
• ingest_parallel_lands_all_rows — a full ingest_parallel over a 6-file input reads back the exact id multiset (no group dropped / mis-stated / lost). The pipelined paths' functional round-trips are covered by tests/pipelined_order.rs, tests/ingest_overlap.rs, tests/data_skip_* and tests/partition.rs.

f4 — bench arm (needs the Loki HEAVY pass)

bench/examples/nornir-bench.rs · skade.write_stat_build: both lanes encode identically across cores; they differ ONLY in where the footer parse runs — serial_stat (parse in a serial loop after the encode, the old placement) vs parallel_stat (parse inside each gatling worker, the new placement). Reports rows/sec + MB/s + the core-saturation (*_cores_busy) of each lane: the parallel-stat lane keeps more cores busy and is ≥ as fast. LIGHT here; the authoritative numbers come from the serial Loki HEAVY pass (SKADE_WSB_FILES/SKADE_WSB_ROWS/SKADE_WSB_RGSIZE). The real function is also covered by skade.ingest_parallel_scaling, whose cores_busy should rise.

skade.pipelined_stat_build is the pipelined complement: both lanes encode across cores, then runtime_stat parses each footer inside async jobs driven on the tokio runtime by run_ordered (the OLD placement — the parses serialise on the reorder task, ~1 core busy) vs gatling_stat parsing inside each gatling worker (the NEW placement). Reports rows/sec + MB/s + *_stat_cores_busy: gatling_stat_cores_busy should be well above runtime_stat_cores_busy and ≥ as fast. Env SKADE_PSB_FILES/SKADE_PSB_ROWS/SKADE_PSB_RGSIZE/SKADE_PSB_DEPTH.

The write-path footer parse is now fully on-core across all four bulk write tails (ingest_parallel, ingest_multi, ingest_pipelined, ingest_pipelined_overlap) — no CPU parse strands cores in a serial tail or on the async runtime anymore.