nordisk/skade

Streaming Parquet windows — the OSM/GeoParquet reuse surface

Class: design · Function: parquet-stream · Uses: parquet_io, bridge. Only the shape that runs. This is the single-call, bounded-memory streaming read that a map/geo consumer (korp's MapLayer OSM source) binds instead of re-parsing OSM/GeoParquet files.

Why this exists (consolidation law)

Per .nornir/consolidation-panels-maps-design.md §3: "OSM source → a MapLayer reading OSM/PMTiles (the constellation already decodes these: osm-katana / znippy-zoomies geo path, skade osm parquet). Reuse, don't re-parse."

skade already decodes OSM/GeoParquet at the record-holding ingest rates the Loki bencher measures (osm_ingest_skade_nvme ~1077 MB/s / 103M rows/s; zstd_decode_sys_zerocopy ~16058 MB/s @ 12/12 cores). korp must NOT grow a second parquet decoder — it binds skade's.

Before this change the reuse required a three-call dance (parquet_layout → rowgroup_windows → loop read_row_groups). That is the right primitive set for skade's own ingest, but it is not a clean seam for a downstream consumer: korp would have to re-implement the windowing loop, which is where a duplicate slowly grows. ParquetWindows collapses the dance into one iterator so the seam is a single bind.

The shape (one iterator, bounded memory)

// skade::ParquetWindows — public, in skade/src/parquet_io.rs
let (ice_schema, arrow_schema, windows) =
    skade::ParquetWindows::open(path, window_rows, max_rows, rows_per_batch, threads)?;

for window in windows {          // ExactSizeIterator: count known up front
    let batches: Vec<RecordBatch> = window?;   // ONE window, all-core (gatling) decoded
    // korp: project geometry cols, build DrawFeatures for the viewport, drop.
    // Peak memory is one window — never the whole nodes.parquet.
}

• open inspects the file (no row data read), derives the Iceberg + Arrow schemas, and precomputes the windows. Returns them up front so the consumer can build its column projection / draw-feature mapping before streaming.
• Each next() decodes exactly one ~window_rows-sized window across up to threads no-barrier workers of the ONE fork-join engine (znippy-zoomies gatling_forkjoin) — the SAME hot path read_row_groups uses, so it inherits skade's benched throughput with zero added copy. ParquetWindows is a thin composition; it changes no hot code.
• max_rows caps the stream (whole-row-group granularity) for a bounded viewport prefetch; usize::MAX streams the whole file.
• Batches keep the file's own Arrow schema — call skade::recast before ingesting into a field-id Iceberg table. A pure map consumer that only draws geometry does not need recast.

The footer is parsed once, not once per worker

read_row_groups fans a window's row groups across up to threads no-barrier gatling workers. Each worker decodes only its bucket's row groups, but every worker needs the file's Parquet footer/metadata (schema + row-group offsets) to build a reader. The naive shape — ParquetRecordBatchReaderBuilder::try_new per worker — re-opens the file and re-reads + re-parses the footer once per bucket (threads parses for a single window). On the bandwidth / light-CPU windowed decode that repeated parse is pure overhead: it buys nothing, since the metadata is identical for every worker.

The shape that runs parses the footer exactly once up front (ArrowReaderMetadata::load), then hands each worker the shared, Arc-backed metadata via ParquetRecordBatchReaderBuilder::new_with_metadata:

let meta = ArrowReaderMetadata::load(&File::open(path)?, Default::default())?; // ONCE
// per worker (own fd for an independent seek cursor; footer reused, not re-parsed):
let rdr = ParquetRecordBatchReaderBuilder::new_with_metadata(File::open(path)?, meta.clone())
    .with_row_groups(bucket).with_batch_size(rpb).build()?;

Each worker still opens its own file descriptor (an independent seek cursor for its lazy per-row-group page reads), so peak memory stays one window — the file is never slurped whole. A single-bucket read (small input / bandwidth-bound) skips the fork-join fan-out entirely and streams on the calling thread. The decode itself is byte-identical; a white-box unit test asserts the footer is parsed once regardless of bucket count (FOOTER_PARSES == 1) and that the fan-out returns the exact same rows as a lean single stream (RED-when-broken).

What korp binds (and what it must NOT do)

korp's MapLayer::features(view) maps each streamed RecordBatch window into Vec<DrawFeature> for the viewport. The renderer stays source-agnostic; the OSM source is just "stream windows from skade + project to draw primitives". Another agent owns korp's map — this doc only guarantees the seam is clean and stable.

Deferred / noted

• Column projection. ParquetWindows currently decodes all columns of the selected row groups (as read_row_groups does). A map consumer often wants only id, lon, lat, tags. A with_projection(&[&str]) builder on ParquetWindows (feeding ProjectionMask into the row-group reader) is the natural next additive step; it is NOT built here because it touches the decode hot path and would need a HEAVY before/after bench. Filed for the korp-map agent to request once the seam is in use.
• Duplicate primitive. bench/src/data.rs carries its own byte-identical parquet_layout / rowgroup_windows / read_row_groups (the bench crate is a detached workspace that predates the skade lib extraction; skade/src/parquet_io.rs is the "lifted" copy). Not consolidated here (the bench crate does not depend on the skade lib, and rewiring it risks perturbing the benched numbers) — noted so a future pass can point the bench at skade::ParquetWindows and delete the twin. The parse-once optimization above was applied to both copies so the osm_ingest_* bench (data::read_row_groups) measures the improved path; the bench arm's read_rows_per_sec sub-metric is the observable and needs the Loki serial HEAVY pass (OSM GeoParquet file) to record the before/after.