nordisk/skade

Design: S3 ingest + read concurrency (plan G4)

Status: implemented (2026-06-12, feat/g4-s3-concurrency). Scope: the detached benchmark crate (bench/) — the data-plane ingest/scan paths it drives over its aegir S3 FileIO. Local NVMe/RAM paths are unchanged.

The regression

Bench run bb49d14a showed data_pipe throughput collapsing on S3:

(tpch read on S3 ≈ 68% of the file path.) The vendored iceberg-datafusion per-file-partition scan patch was already active and is not the cause — S3 burned only ~23/32 cores because the I/O, not the CPU, was the wall. The root cause is sequential object-store I/O in the bench:

• WRITE (bench/src/pipeline.rs::run_single_writer_ingest): one writer wrote files one at a time. On S3 each PUT is a full HTTP round-trip (≈ tens of ms), so the device sat idle for the round-trip of every file — the writer was latency-bound, not bandwidth-bound.
• READ (bench/src/s3_storage.rs::S3FileRead::read): one blocking ranged GET per Parquet column chunk, issued serially, no read-ahead/coalesce. A scan of N column chunks = N sequential round-trips per file.

Why local stays sequential (and must)

The single-writer / many-processor topology (znippy pattern) is optimal on local NVMe/RAM: there a write is a memcpy-class syscall with no round-trip latency to hide. One uncontended writer already saturates the device; adding writers only buys lock/seek/allocator contention and lowers throughput. So the local path keeps exactly one in-flight write. The S3 problem is the opposite — the cost is per-op latency, which only concurrency hides. Hence the fix is target-selective, not a global change:

target ─► is_s3_target(target)?
            ├─ no  (skade-file/-nvme/-ram) → sequential single writer  (1 PUT/GET in flight)
            └─ yes (skade-s3/nessie/polaris) → concurrent writer + coalesced reads

is_s3_target lives in bench/src/main.rs; it is the single discriminator.

WRITE — concurrent multi-PUT (S3)

pipeline.rs now has two public entry points over one shared run_ingest:

• run_single_writer_ingest(...) — write_concurrency = 1, the original sequential writer, byte-for-byte the old behaviour. Local.
• run_concurrent_writer_ingest(..., write_concurrency) — drains the same encode channel and drives the PUT fan-out through znippy's gatling::io no-barrier async task pool, keeping up to write_concurrency PUTs in flight. Each job moves one pre-encoded Parquet blob's Bytes straight into the S3 write (zero-copy); gatling::io::run_ordered re-sequences the landed DataFiles back into seq order so the files_per_commit fast_append batches stay deterministic — exactly as before. S3. Default fan-out 16, override BENCH_PIPE_WRITE_CONCURRENCY.

The encode stage (N CPU workers → Parquet bytes → bounded mpsc) is identical for both; only the writer's in-flight depth differs. Because both paths write the same files and commit them the same way, they are interchangeable for correctness — see the test below.

PipelineStats gained write_concurrency (the cap) and peak_inflight (the largest number of PUTs that actually overlapped), so a run/test can prove the concurrency happened rather than was merely requested.

Why a new gatling variant (gatling::io) was needed

The concurrent writer previously hand-rolled a FuturesUnordered admission loop. The obvious "reuse znippy" move would be gatling::run/run_typed — but that is the wrong substrate. The sync gatling engine is a std::thread::scope CPU thread pool that pulls a Read byte stream, splits it into zero-copy slot slices, and runs a CPU transform per byte range. Our G4 save is the opposite shape on every axis:

A thread blocked on an S3 socket is wasted; there is no byte stream to split (the unit is already a whole Parquet file). So we added gatling::io — the async sibling that keeps every gatling invariant (no barrier, bounded N-in-flight, ordered output, zero-copy Bytes) but swaps the substrate to tokio tasks for I/O-bound fan-out. It lives in znippy-zoomies/src/gatling_io.rs, re-exported as gatling::io; its module doc carries the full side-by-side rationale. The bench depends on it via znippy-zoomies = { path = "../../znippy-zoomies" }.

READ — whole-file coalescing read-ahead (S3)

S3FileRead (the iceberg FileRead over aegir) now coalesces. The Parquet reader issues many small ranged reads against the same object (footer, then one per column chunk). Instead of one blocking GET per range, on the first read we fetch the whole object once (a single GET) when it fits BENCH_S3_WHOLE_FILE_CAP (default 64 MiB; bench files are far under it) and serve every subsequent range as a zero-copy slice of that cached Bytes. This turns N sequential round-trips into 1. Concurrent reads for the same file all await the single shared fetch via a tokio::sync::OnceCell, so distinct ranges never re-fetch. Objects over the cap transparently fall back to per-range GETs. This is the in-the-limit form of object_store::coalesce_ranges — there is no benefit to merging adjacent sub-ranges when the whole small file is one cheap GET. Counters (object_gets, range_reads) make the collapse observable.

Correctness assertion (LAW: inject-and-assert)

pipeline.rs test concurrent_writer_matches_sequential_and_overlaps: feeds the same real synthetic batches (24 files × 500 rows) through both the sequential and the concurrent writer into a local LocalFs catalog, then asserts:

• both ingest the expected row count and scan back the same rows — the concurrent path produces the same data as the sequential one;
• conc.peak_inflight > 1 (concurrency truly overlapped) while seq.peak_inflight == 1 and seq.write_concurrency == 1 (local stays serial).

s3_storage.rs test coalesced_reads_return_exact_bytes_and_issue_no_ranged_gets: seeds the whole-object cache with a known 4 KiB body and drives the footer-then-column-chunk access pattern, asserting every range returns the exact bytes and zero ranged GETs are issued (the coalesce served them all). read_past_eof_is_clamped covers the EOF clamp. None of these are "didn't panic" — each injects real input and asserts real output.

Measured (live RustFS :9002, synthetic, scan-back included)

Scanned row count matched ingested in every run, so the read-coalescing path is correct end-to-end. Larger file counts saturate the 16-wide window; the heavy orchestrated bench (Nessie/Polaris) will measure the full curve.

CPU saturation (the 90-100% goal)

Measured on a 32-core box (live RustFS S3 :9002), 1 Hz /proc/stat sampler:

Conclusion. Uncompressed columnar ingest is memory-bandwidth bound — Arrow→Parquet is memcpy-class, so a few cores already exceed what the writer absorbs and there is no CPU work to parallelize (more cores don't help; that's not a skade limit). With real CPU work (BENCH_PIPE_COMPRESSION=zstd:9) the all-core encode stage is CPU-bound and saturates the box; on S3 the gatling::io concurrent writer (32 PUTs in flight) overlaps the async saves with the encode, sustaining 96-100% on all 32 cores. So skade ingest hits the 90-100% goal whenever the workload carries CPU work (compression, or real OSM parse — planet_pbf_to_geo likewise saturates 31/32). Knobs: BENCH_PIPE_COMPRESSION, BENCH_PIPE_DEPTH, BENCH_PIPE_WRITE_CONCURRENCY, BENCH_PIPE_FILES_PER_COMMIT.

zstd via the user's own static libzstd (zstd-sys-rs)

The zstd:<level> codec above runs zstd inside the Parquet writer, i.e. through the zstd crate that the parquet crate pulls in transitively (parquet → zstd → zstd-safe → zstd-sys 2.0.x). To route the heavy zstd CPU work through the user's own bindings — zstd-sys-rs (statically-linked libzstd 1.5.7, raw C FFI: ZSTD_compressBound/ZSTD_compress/ZSTD_getFrameContentSize/ ZSTD_decompress/ZSTD_isError) — the bench adds a separate codec:

BENCH_PIPE_COMPRESSION=zstd-sys        # level 9 (CPU-bound, like zstd:9)
BENCH_PIPE_COMPRESSION=zstd-sys:N      # explicit level N

Why not [patch.crates-io] zstd-sys?

The obvious-looking route — patch zstd-sys so parquet's Compression::ZSTD transparently links the user's static libzstd — is not viable cleanly. zstd-safe 7.x consumes ~160 symbols from zstd-sys: the whole seekable API (ZSTD_seekable_*, ZSTD_seekTable_*), the ZDICT dictionary trainer (ZDICT_trainFromBuffer, ZDICT_getDictID), the full advanced parameter/streaming surface, and dozens of typed constants/enums (ZSTD_cParameter, ZSTD_ParamSwitch_e, ZSTD_SEEKABLE_*, the ZSTD_*LOG_* bounds, …). zstd-sys-rs deliberately compiles only the core lib/{common,compress,decompress} sources and bindgen-allowlists ZSTD_.* from zstd.h — it ships no seekable, no ZDICT, and not the typed constant set zstd-safe references. A pub use zstd_sys_rs::* shim named zstd-sys would fail to resolve those missing items, so the patch can't satisfy zstd-safe without re-implementing that whole surface. Rejected.

Chosen approach: per-file round-trip through zstd-sys-rs

pipeline.rs::encode_parquet detects the zstd-sys codec, writes the Parquet file UNCOMPRESSED, then runs the encoded bytes through pipeline.rs::zstd_sys_roundtrip:

ZSTD_compressBound → ZSTD_compress (level N)  →  ZSTD_getFrameContentSize → ZSTD_decompress

The round-trip returns the same bytes it was given, so what gets written to the table is still a valid Parquet file — the iceberg scan reads it back unchanged and scan_count == rows (no silent correctness break). Meanwhile the CPU-heavy compress + decompress on every file went through the user's static libzstd 1.5.7, so the encode stage saturates the cores through the user's bindings instead of parquet's internal zstd crate. (This measures user-bindings zstd throughput, not on-disk size — the stored file is uncompressed Parquet; use zstd:N when you want parquet's own on-disk zstd compression.)

Proof the user's bindings are exercised (cargo tree -i zstd-sys-rs → zstd-sys-rs v1.5.11 is a direct dep of the bench; encode_parquet calls zstd_sys_rs::ZSTD_* directly) and correctness are both pinned by inject-and-assert unit tests in pipeline.rs::tests:

• zstd_sys_rs_roundtrip_recovers_known_bytes — compresses a known structured buffer and asserts it decompresses back to the exact original bytes, plus asserts the frame is actually smaller than the input (libzstd really compressed, not passed through).
• zstd_sys_encode_path_yields_valid_parquet — runs encode_parquet with BENCH_PIPE_COMPRESSION=zstd-sys:9 and asserts the result parses as a Parquet file with the right row count (so the iceberg scan round-trip holds).