<!-- nornir:class: design -->
# 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)
```rust
// 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`:
```rust
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 needs | binds | must not |
|---|---|---|
| stream OSM node/way geometry from `nodes.parquet` | `skade::ParquetWindows::open` | re-open the file with its own `ParquetRecordBatchReaderBuilder` |
| all-core decode without a private thread pool | inherited (gatling) | spawn its own rayon/thread pool (rayon-free law) |
| bounded memory over billions of rows | window-at-a-time iteration | collect the whole file into one `Vec<RecordBatch>` |
| file schema before drawing | `open`'s returned `(Schema, ArrowSchemaRef)` | guess column layout |
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.