osm-katana/src/reader.rs
#![allow(clippy::float_arithmetic, clippy::arithmetic_side_effects)]
use crate::shared::gpu_backend::{GpuBackend, RawWay};
use serde::Serialize;
use crate::{Clip, clip_keeps};
use crate::{geometry, node_store::NodeStore, pbf_decomp, pbf_io, tags};
// ── Records ───────────────────────────────────────────────────────────────────
pub struct NodeRecord {
pub id: i64,
pub lon_lat: (f64, f64), // (longitude, latitude) — WKB encoded at write time
pub tags_json: String,
pub version: Option<i32>,
}
pub struct WayRecord {
pub id: i64,
pub geometry: Option<Vec<u8>>,
pub tags_json: String,
pub node_refs: Vec<i64>,
pub version: Option<i32>,
}
pub struct RelationRecord {
pub id: i64,
pub tags_json: String,
pub members_json: String,
pub version: Option<i32>,
}
/// Which kinds of group a decoded `PrimitiveBlock` carried — the pass-1 probe that
/// lets pass 2 leave node-only blobs closed, and a relations-only sub-pass leave the
/// way blobs closed too. An unreadable blob reports BOTH `true`: if we could not look
/// inside it we must not claim there is nothing in it.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct BlobKinds {
pub ways: bool,
pub rels: bool,
}
impl BlobKinds {
/// The conservative answer for a blob that could not be inflated or parsed.
pub const UNKNOWN: BlobKinds = BlobKinds {
ways: true,
rels: true,
};
/// Carries a way OR a relation group — the original pass-2 question.
#[inline]
pub fn any(self) -> bool {
self.ways || self.rels
}
}
// ── the relation clip prefilter ──────────────────────────────────────────────
/// **The ONE keep-decision for a relation under a region clip** (LAW 5).
///
/// MEASURED on the live Sweden cut 2026-08-31: `relations.parquet` held 14 632 472
/// rows — the whole planet's relation set, `geometry` 100 % null — against 274 206 in
/// `relations.resolved.parquet`. The clip filters NODES at decode and a way is dropped
/// when none of its nodes survived, but a relation carries no coordinates, so every
/// relation passed through: ~1.005 GB of `members` + `tags` per cut for rows that can
/// never resolve, and 14.36 M rows the resolve stage scans to find 274 k.
///
/// The rule here is the admission rule `crate::relgeom::resolve_relations` already
/// applies, moved to the writer: **a relation is kept when ANY member survived the
/// clip** — a way present in the cut's `ways.parquet`, a node present in the node
/// store, or a relation that is itself kept. That is a strict SUPERSET of what
/// resolves (resolve additionally needs the geometry to assemble), so the resolve
/// stage's output cannot change; only the rows it scans do.
///
/// Way membership is a sorted id list collected by the way sub-pass — it exists only
/// because ways are decided in the same pass-2 that decides relations, and a planet
/// file orders every way before every relation, so the relation sub-pass runs AFTER
/// the way ids are complete. Node membership is the node store itself. Relation
/// membership cannot be answered on a worker (a `route_master`'s routes are in
/// other blobs, and may be kept only via their own relation members), so a relation
/// whose only surviving chance is a relation member is [`RelVerdict::Pending`] and is
/// settled by a fixpoint on the collector — the same 8-lap fixpoint `relgeom` runs.
pub struct RelationClipOracle {
/// Sorted, deduplicated ids of the ways the clip kept.
way_ids: Vec<i64>,
}
/// What the oracle says about one relation's members.
#[derive(Debug, PartialEq, Eq)]
pub enum RelVerdict {
/// A node or way member survived the clip — write it.
Keep,
/// No node or way member survived, but these relation members might have; the
/// collector decides once every worker has reported its kept relation ids.
Pending(Vec<i64>),
/// No member of any kind survived — the row could never resolve.
Drop,
}
/// A relation the oracle could not settle on the worker, carried to the collector
/// with the relation ids its fate depends on.
pub struct PendingRelation {
pub record: RelationRecord,
pub rel_members: Vec<i64>,
}
impl RelationClipOracle {
/// `way_ids` need not be sorted; it is sorted and deduplicated here, once.
pub fn new(mut way_ids: Vec<i64>) -> Self {
way_ids.sort_unstable();
way_ids.dedup();
Self { way_ids }
}
/// How many ways the clip kept — the oracle's whole knowledge of the way table.
pub fn way_count(&self) -> usize {
self.way_ids.len()
}
/// The verdict for one relation, from its members as `(type, ref)` pairs. `store`
/// is the clip's node store: it holds exactly the nodes that survived.
pub fn verdict<'m>(
&self,
store: &NodeStore,
members: impl IntoIterator<Item = (&'m str, i64)>,
) -> RelVerdict {
let mut rel_members: Vec<i64> = Vec::new();
for (kind, id) in members {
match kind {
"way" => {
if self.way_ids.binary_search(&id).is_ok() {
return RelVerdict::Keep;
}
}
"node" => {
if store.lookup(id).is_some() {
return RelVerdict::Keep;
}
}
"relation" => rel_members.push(id),
_ => {}
}
}
if rel_members.is_empty() {
RelVerdict::Drop
} else {
RelVerdict::Pending(rel_members)
}
}
/// **The collector's half**: settle every pending relation against the ids the
/// workers kept, to a fixpoint. Returns the promoted records; `kept_ids` grows
/// with them so a caller can chain further batches.
///
/// Bounded at 8 laps like `relgeom`'s — OSM nests `route_master` → `route`, and
/// `superroute` one deeper; 8 is far past anything real and stops a cycle dead.
pub fn settle(
pending: Vec<PendingRelation>,
kept_ids: &mut std::collections::HashSet<i64>,
) -> Vec<RelationRecord> {
let mut pending: Vec<Option<PendingRelation>> = pending.into_iter().map(Some).collect();
let mut promoted: Vec<RelationRecord> = Vec::new();
for _ in 0..8 {
let mut moved = 0usize;
for slot in pending.iter_mut() {
let promote = slot
.as_ref()
.is_some_and(|p| p.rel_members.iter().any(|m| kept_ids.contains(m)));
if promote {
if let Some(p) = slot.take() {
kept_ids.insert(p.record.id);
promoted.push(p.record);
moved += 1;
}
}
}
if moved == 0 {
break;
}
}
promoted
}
}
/// What one blob's relation decode left behind besides the kept rows: the relations
/// the oracle could not settle, and the count it dropped. Empty without an oracle.
#[derive(Default)]
pub struct RelSieve {
pub pending: Vec<PendingRelation>,
pub dropped: u64,
}
impl RelSieve {
/// Fold another chunk's sieve into this one.
pub fn absorb(&mut self, other: RelSieve) {
self.pending.extend(other.pending);
self.dropped += other.dropped;
}
}
impl RelationClipOracle {
/// **Where a decoded relation meets its verdict** — the one seam for BOTH decoders
/// (a PBF `PrimitiveBlock` in `parse_ways_rels`, an XML `<relation>` in
/// `xml_reader::parse_chunk`), so the two readers cannot disagree about what a
/// clip keeps. With no oracle the relation is kept, as it always was.
///
/// Returns the record to WRITE. A `Drop` is counted in `sieve` and never built —
/// `make` runs only for a row that leaves this function, so a dropped relation
/// costs one verdict and no JSON. A `Pending` is built and parked in
/// `sieve.pending` for the collector's fixpoint ([`Self::settle`]).
pub(crate) fn admit(
oracle: Option<&Self>,
store: &NodeStore,
members: &[MemberEntry],
make: impl FnOnce() -> RelationRecord,
sieve: &mut RelSieve,
) -> Option<RelationRecord> {
let Some(o) = oracle else {
return Some(make());
};
match o.verdict(store, members.iter().map(|m| (m.member_type, m.member_ref))) {
RelVerdict::Keep => Some(make()),
RelVerdict::Pending(rel_members) => {
sieve.pending.push(PendingRelation {
record: make(),
rel_members,
});
None
}
RelVerdict::Drop => {
sieve.dropped += 1;
None
}
}
}
}
#[derive(Serialize)]
pub(crate) struct MemberEntry {
#[serde(rename = "type")]
pub(crate) member_type: &'static str,
#[serde(rename = "ref")]
pub(crate) member_ref: i64,
pub(crate) role: String,
}
impl MemberEntry {
/// The `members` column's spelling of a member list — one serialiser for the
/// PBF and XML decoders.
pub(crate) fn json(members: &[MemberEntry]) -> String {
serde_json::to_string(members).unwrap_or_else(|_| String::from("[]"))
}
}
/// **Way-resolution loss counters** — the 2026-08 extraction audit's finding (a).
///
/// A way whose node ref resolves to nothing used to be SILENTLY shortened (the
/// `filter_map` in [`parse_ways_rels`] dropped the vertex and nothing counted
/// it). The way is still emitted shortened — the schema already carries the
/// per-way signal by construction, because `node_refs` keeps the FULL authored
/// list beside the shorter geometry, and dropping the way would discard the
/// vertices that DID resolve — but the loss is now a counted fact that every
/// resolved convert — [`crate::xml_reader::read_resolved_pbf`] AND its XML-side
/// twins ([`crate::xml_reader::read_resolved_bz2`] / `read_resolved_gz` /
/// `read_resolved_osm`, which resolve in their own seams and were the disclosed
/// uncounted gap of this fix's first cut) — carries onto the convert summary
/// (`pass2` phase-log keys `unresolved_node_refs` / `shortened_ways`).
///
/// Counted only when NO clip is active: with a clip, the node store holds only
/// in-region nodes and boundary-crossing ways are truncated as a documented
/// semantic (see the region-clip note at the resolve site), not as data loss.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ResolveLoss {
/// Node refs that resolved to nothing (each one is a missing vertex).
pub unresolved_refs: u64,
/// Ways emitted with fewer coords than node refs (≥1 unresolved ref).
pub shortened_ways: u64,
}
impl ResolveLoss {
pub fn add(&mut self, other: ResolveLoss) {
self.unresolved_refs += other.unresolved_refs;
self.shortened_ways += other.shortened_ways;
}
/// Count ONE way's resolve outcome — the single counting rule every resolve
/// seam shares (PBF [`parse_ways_rels`], XML
/// [`crate::xml_reader`]'s `parse_chunk` and `Pass2ResolvedCodec`), so the
/// paths cannot drift apart. A shortfall WITHOUT a clip is a silently
/// missing vertex; with a clip active the node store is deliberately
/// partial and truncation is the documented node-membership semantics, not
/// data loss.
pub fn note_way(&mut self, n_refs: usize, n_resolved: usize, clip_active: bool) {
if !clip_active && n_resolved < n_refs {
self.unresolved_refs += (n_refs - n_resolved) as u64;
self.shortened_ways += 1;
}
}
}
pub(crate) struct CollectedWay {
pub(crate) id: i64,
pub(crate) coords: Vec<(f64, f64)>,
pub(crate) tags_json: String,
pub(crate) node_refs: Vec<i64>,
pub(crate) version: Option<i32>,
pub(crate) is_area: bool,
}
// ── Per-blob helpers for streaming PBF pipeline ───────────────────────────────
//
// One function call = one compressed blob. The Gatling PBF codec fans these
// across all cores (one blob per self-dispatched worker) — no rayon, no nesting.
/// Decompress one blob and extract full NodeRecords + coords in one decode.
/// Way/relation blobs return empty vecs. Used by the Gatling PBF pass-1 codec to
/// fuse coord extraction and node-record encoding (single protobuf parse).
#[inline]
pub fn blob_nodes_and_coords(
data: &[u8],
pos: &pbf_io::BlobPos,
clip: Option<&Clip>,
) -> (Vec<NodeRecord>, Vec<(i64, f32, f32)>) {
let (records, coords, _) = blob_nodes_and_coords_probed(data, pos, clip);
(records, coords)
}
/// [`blob_nodes_and_coords`] plus the pass-2 probe: does this blob carry any way
/// or relation group?
///
/// The bool is the same question [`read_pass1_mmap_planned`] asks on the mmap
/// path, asked here on the STREAMING path so `convert()` can build a pass-2 plan
/// too. It is free: pass 1 already inflates the blob and parses the
/// `PrimitiveBlock`, so the walk over `primitivegroup` costs one pointer chase
/// per group and no extra decode.
///
/// A blob we could not inflate or could not parse reports `true` — if we never
/// saw inside it we must not claim there is nothing in it.
#[inline]
pub fn blob_nodes_and_coords_probed(
data: &[u8],
pos: &pbf_io::BlobPos,
clip: Option<&Clip>,
) -> (Vec<NodeRecord>, Vec<(i64, f32, f32)>, BlobKinds) {
data.get(pos.offset..pos.offset.saturating_add(pos.length))
.and_then(|blob| pbf_decomp::decompress(blob).ok().flatten())
.map(|bytes| parse_nodes_probed(&bytes, clip))
// Unreadable blob → empty, and a pass-2 CANDIDATE.
.unwrap_or((Vec::new(), Vec::new(), BlobKinds::UNKNOWN))
}
/// Decompress one blob, parse ways+rels, encode WKB.
/// Node blobs return empty vecs.
#[inline]
pub fn blob_ways_rels(
data: &[u8],
pos: &pbf_io::BlobPos,
store: &NodeStore,
clip_active: bool,
oracle: Option<&RelationClipOracle>,
) -> (Vec<WayRecord>, Vec<RelationRecord>, ResolveLoss, RelSieve) {
let bytes = match data
.get(pos.offset..pos.offset.saturating_add(pos.length))
.and_then(|blob| pbf_decomp::decompress(blob).ok().flatten())
{
Some(b) => b,
None => {
return (
Vec::new(),
Vec::new(),
ResolveLoss::default(),
RelSieve::default(),
);
}
};
let (collected, rels, loss, sieve) = parse_ways_rels(&bytes, store, clip_active, oracle);
let raw: Vec<RawWay> = collected
.iter()
.map(|w| RawWay {
coords: w.coords.clone(),
is_area: w.is_area,
})
.collect();
let wkb_list = crate::shared::gpu_backend::CpuBackend.encode_wkb(&raw);
let ways = collected
.into_iter()
.zip(wkb_list)
.map(|(w, geom)| WayRecord {
id: w.id,
geometry: geom,
tags_json: w.tags_json,
node_refs: w.node_refs,
version: w.version,
})
.collect();
(ways, rels, loss, sieve)
}
// ── mmap single-pass variants ────────────────────────────────────────────────
/// Pass 1 over a mmap'd PBF: extract only (id, lat, lon) from every blob.
/// Way/relation blobs find no node groups and return empty — fast skip.
pub fn read_pass1_mmap(
data: &[u8],
positions: &[crate::pbf_io::BlobPos],
) -> anyhow::Result<NodeStore> {
read_pass1_mmap_planned(data, positions).map(|(store, _)| store)
}
/// Blob indices that carry ways or relations — the ONLY blobs pass 2 needs.
///
/// Pass 2 used to re-read and re-inflate **every** blob, including the node-only
/// blobs it was always going to find nothing in. On a Sweden clip that was measured
/// at **1693 s of a 2726 s run — 62 %** — and it cost 64 % LONGER than pass 1 to find
/// 8.6 M ways where pass 1 found 96.9 M nodes. The work was not unbalanced; it was
/// done twice.
///
/// The design doc recorded this as blocked: *"the streaming codec's `transform` is
/// handed a slot-relative slice and does not know the blob's absolute file offset, so
/// the map has no key yet."* That is true of the STREAMING path (`.bz2`/`.gz`/XML via
/// `run_typed`) and NOT of the mmap path, which is what a `.pbf` convert actually
/// takes: `BlobPos` carries the absolute `offset`, and the blob's index in
/// `positions` is already a perfectly good key.
///
/// The flag is free. Pass 1 decompresses and protobuf-parses each block anyway, and
/// walks its `primitivegroup`s looking for nodes — so it can see a `ways`/`relations`
/// group in the same walk, with no extra decode.
pub type Pass2Plan = Vec<u32>;
/// Pass 1 that also reports which blobs pass 2 must visit.
///
/// The predicate is *carries ways or relations*, NOT *carries no nodes*. A PBF
/// `PrimitiveBlock` may legally hold groups of different kinds, and although the
/// common writers emit homogeneous blocks, skipping on "this blob had nodes" would
/// silently drop ways out of any mixed block. Asking the question directly costs
/// nothing and cannot be wrong.
pub fn read_pass1_mmap_planned(
data: &[u8],
positions: &[crate::pbf_io::BlobPos],
) -> anyhow::Result<(NodeStore, Pass2Plan)> {
let per_block: Vec<(Vec<(i64, f32, f32)>, BlobKinds)> = crate::par::par_map(positions.len(), |i| {
let p = &positions[i];
data.get(p.offset..p.offset.saturating_add(p.length))
.and_then(|blob| pbf_decomp::decompress(blob).ok().flatten())
.map(|bytes| parse_node_coords_probed(&bytes, None))
// A blob we could not read at all stays a pass-2 CANDIDATE: the
// conservative direction is to look again, never to skip unseen.
.unwrap_or((Vec::new(), BlobKinds::UNKNOWN))
});
let mut plan: Pass2Plan = Vec::new();
let mut all_coords: Vec<(i64, f32, f32)> = Vec::new();
for (i, (coords, kinds)) in per_block.into_iter().enumerate() {
if kinds.any() {
plan.push(i as u32);
}
all_coords.extend(coords);
}
Ok((NodeStore::from_coords(all_coords), plan))
}
/// Pass 2 over a mmap'd PBF: resolve ways against the completed NodeStore.
/// Node blobs find no way groups and return empty — fast skip.
///
/// **PLAN-LESS: this visits EVERY blob** — including the node-only blobs that were
/// 62 % of a Sweden clip's wall time (see [`read_pass1_mmap_planned`]). It also
/// requires a `node_store` that already holds every referenced node, i.e. the
/// whole-input in-RAM [`NodeStore::from_coords`] store — the exact combination that
/// OOM-killed a planet index (507 GiB). It is therefore **clip-scale only**; on a
/// planet-scale input call [`read_pass2_mmap_planned`] with a way/rel-only plan
/// instead (and never build a whole-planet `from_coords` store). Kept for the
/// small-input unit tests and the (clip-scale) [`crate::load_admin_boundaries_pbf`].
pub fn read_pass2_mmap(
data: &[u8],
positions: &[crate::pbf_io::BlobPos],
node_store: &NodeStore,
backend: &dyn GpuBackend,
) -> anyhow::Result<(Vec<WayRecord>, Vec<RelationRecord>, ResolveLoss)> {
// No plan ⇒ visit every blob, exactly as before.
let all: Pass2Plan = (0..positions.len() as u32).collect();
read_pass2_mmap_planned(data, positions, &all, node_store, backend)
}
/// Pass 2 restricted to the blobs [`read_pass1_mmap_planned`] said carry ways or
/// relations.
///
/// Same work, same order, same output — just without re-inflating the node-only
/// blobs. Those were 62 % of a Sweden clip's wall time and by construction contain
/// nothing pass 2 can use.
pub fn read_pass2_mmap_planned(
data: &[u8],
positions: &[crate::pbf_io::BlobPos],
plan: &[u32],
node_store: &NodeStore,
backend: &dyn GpuBackend,
) -> anyhow::Result<(Vec<WayRecord>, Vec<RelationRecord>, ResolveLoss)> {
// Parse + WKB-encode each blob ON ITS WORKER. The old shape parsed in
// parallel but then (a) encoded every way's WKB serially on one core and
// (b) materialized the WayRecords serially — two 1-core tails after the
// fan-out. Here each worker resolves its blob's ways AND encodes their WKB
// (via the shared `backend`, `Send + Sync`), emitting finished WayRecords, so
// the whole parse→encode→materialize path saturates every core. Balanced by
// blob byte length (LPT) so a fat blob can't leave the tail on one core.
let per_block: Vec<(Vec<WayRecord>, Vec<RelationRecord>, ResolveLoss)> =
gatling::gatling_forkjoin::gatling_for_each_balanced(
plan.len(),
0,
1,
|i| positions[plan[i] as usize].length as u64,
|i| {
let p = &positions[plan[i] as usize];
// No clip on this path (`clip_active` is false), so no oracle: every
// relation is kept, exactly as before.
let (collected, rels, loss, _sieve) = data
.get(p.offset..p.offset.saturating_add(p.length))
.and_then(|blob| pbf_decomp::decompress(blob).ok().flatten())
.map(|bytes| parse_ways_rels(&bytes, node_store, false, None))
.unwrap_or_default();
let raw: Vec<RawWay> = collected
.iter()
.map(|w| RawWay {
coords: w.coords.clone(),
is_area: w.is_area,
})
.collect();
let wkb_list = backend.encode_wkb(&raw);
let ways: Vec<WayRecord> = collected
.into_iter()
.zip(wkb_list)
.map(|(w, geom)| WayRecord {
id: w.id,
geometry: geom,
tags_json: w.tags_json,
node_refs: w.node_refs,
version: w.version,
})
.collect();
(ways, rels, loss)
},
);
// Materialize = concatenation of finished per-block records (moves only).
let mut all_ways: Vec<WayRecord> = Vec::new();
let mut all_rels: Vec<RelationRecord> = Vec::new();
let mut all_loss = ResolveLoss::default();
for (ways, rels, loss) in per_block {
all_ways.extend(ways);
all_rels.extend(rels);
all_loss.add(loss);
}
Ok((all_ways, all_rels, all_loss))
}
// ── Stage 3: parse PrimitiveBlock ─────────────────────────────────────────────
/// Extract only (id, lat, lon) from a PrimitiveBlock — no geometry encoding, no tag
/// serialisation. Used by `read_pass1_store_only` to avoid ~2 GB of throw-away allocations.
fn parse_node_coords(bytes: &[u8], clip: Option<&Clip>) -> Vec<(i64, f32, f32)> {
parse_node_coords_probed(bytes, clip).0
}
/// [`parse_node_coords`] plus the pass-2 probe: does this block carry any way or
/// relation group?
///
/// One extra `bool` off a walk pass 1 already performs, so pass 2 can skip the
/// node-only blobs instead of re-inflating them. An unparseable block reports `true`
/// — if we could not read it, we must not claim there is nothing in it.
fn parse_node_coords_probed(
bytes: &[u8],
clip: Option<&Clip>,
) -> (Vec<(i64, f32, f32)>, BlobKinds) {
use crate::proto::osmformat::PrimitiveBlock;
use protobuf::Message as _;
let block = match PrimitiveBlock::parse_from_bytes(bytes) {
Ok(b) => b,
Err(_) => return (Vec::new(), BlobKinds::UNKNOWN),
};
(parse_node_coords_inner(&block, clip), block_kinds(&block))
}
/// **The pass-2 predicate — the ONE place it is decided**, so the mmap path
/// ([`parse_node_coords_probed`]) and the streaming path
/// ([`parse_nodes_probed`]) can never disagree about which blobs pass 2 needs.
///
/// The question is *carries ways or relations*, NOT *carries no nodes*. A PBF
/// `PrimitiveBlock` may legally hold groups of different kinds; skipping on
/// "this blob had nodes" would silently drop the ways of any mixed block.
#[inline]
fn block_kinds(block: &crate::proto::osmformat::PrimitiveBlock) -> BlobKinds {
BlobKinds {
ways: block.primitivegroup.iter().any(|g| !g.ways.is_empty()),
rels: block.primitivegroup.iter().any(|g| !g.relations.is_empty()),
}
}
fn parse_node_coords_inner(
block: &crate::proto::osmformat::PrimitiveBlock,
clip: Option<&Clip>,
) -> Vec<(i64, f32, f32)> {
let gran = block.granularity.unwrap_or(100) as i64;
let lat_off = block.lat_offset.unwrap_or(0);
let lon_off = block.lon_offset.unwrap_or(0);
let mut coords: Vec<(i64, f32, f32)> = Vec::new();
for group in &block.primitivegroup {
if let Some(dense) = group.dense.as_ref() {
let mut id: i64 = 0;
let mut lat: i64 = 0;
let mut lon: i64 = 0;
let mut kv_pos = 0usize;
let has_tags = !dense.keys_vals.is_empty();
for ((&di, &dlat), &dlon) in dense.id.iter().zip(dense.lat.iter()).zip(dense.lon.iter())
{
id += di;
lat += dlat;
lon += dlon;
// skip tags — advance kv_pos past this node's key-value pairs
// (still done for clipped-out nodes so the shared cursor stays
// in sync, but it is a cheap pointer walk — no alloc, no JSON)
if has_tags {
while let Some(&k) = dense.keys_vals.get(kv_pos) {
kv_pos = kv_pos.saturating_add(1);
if k == 0 {
break;
}
kv_pos = kv_pos.saturating_add(1); // skip value
}
}
let lat_deg = (lat_off + gran * lat) as f64 / 1_000_000_000.0;
let lon_deg = (lon_off + gran * lon) as f64 / 1_000_000_000.0;
if !clip_keeps(clip, lon_deg, lat_deg) {
continue;
}
coords.push((id, lat_deg as f32, lon_deg as f32));
}
}
for node in &group.nodes {
let lat_deg = (lat_off + gran * node.lat.unwrap_or(0)) as f64 / 1_000_000_000.0;
let lon_deg = (lon_off + gran * node.lon.unwrap_or(0)) as f64 / 1_000_000_000.0;
if !clip_keeps(clip, lon_deg, lat_deg) {
continue;
}
coords.push((node.id.unwrap_or(0), lat_deg as f32, lon_deg as f32));
}
}
coords
}
/// Full node records + coords out of one `PrimitiveBlock`, plus
/// [`block_has_way_or_rel`] off the same parse. See
/// [`blob_nodes_and_coords_probed`].
fn parse_nodes_probed(
bytes: &[u8],
clip: Option<&Clip>,
) -> (Vec<NodeRecord>, Vec<(i64, f32, f32)>, BlobKinds) {
use crate::proto::osmformat::PrimitiveBlock;
use protobuf::Message as _;
let block = match PrimitiveBlock::parse_from_bytes(bytes) {
Ok(b) => b,
// Unparseable → nothing extracted, and a pass-2 CANDIDATE.
Err(_) => return (Vec::new(), Vec::new(), BlobKinds::UNKNOWN),
};
let has_way_or_rel = block_kinds(&block);
let gran = block.granularity.unwrap_or(100) as i64;
let lat_off = block.lat_offset.unwrap_or(0);
let lon_off = block.lon_offset.unwrap_or(0);
let strtab = string_table(&block);
let mut records: Vec<NodeRecord> = Vec::new();
let mut coords: Vec<(i64, f32, f32)> = Vec::new();
for group in &block.primitivegroup {
// Dense nodes (MessageField<DenseNodes> — use .as_deref() to get Option<&DenseNodes>)
if let Some(dense) = group.dense.as_ref() {
let mut id: i64 = 0;
let mut lat: i64 = 0;
let mut lon: i64 = 0;
let mut kv_pos = 0usize;
let has_tags = !dense.keys_vals.is_empty();
for (i, ((&di, &dlat), &dlon)) in dense
.id
.iter()
.zip(dense.lat.iter())
.zip(dense.lon.iter())
.enumerate()
{
id += di;
lat += dlat;
lon += dlon;
let lat_deg = (lat_off + gran * lat) as f64 / 1_000_000_000.0;
let lon_deg = (lon_off + gran * lon) as f64 / 1_000_000_000.0;
// Region clip at decode: a clipped-out node still advances the
// SHARED key/value cursor (a cheap pointer walk, no alloc/JSON),
// then is skipped before any tag serialisation or record push.
if !clip_keeps(clip, lon_deg, lat_deg) {
if has_tags {
kv_pos = skip_dense_tags(&dense.keys_vals, kv_pos);
}
continue;
}
let version = dense
.denseinfo
.as_ref()
.and_then(|di| di.version.get(i))
.copied();
let tags_json = if has_tags {
let (json, new_pos) = collect_dense_tags(&dense.keys_vals, kv_pos, &strtab);
kv_pos = new_pos;
json
} else {
String::from("{}")
};
coords.push((id, lat_deg as f32, lon_deg as f32));
records.push(NodeRecord {
id,
lon_lat: (lon_deg, lat_deg),
tags_json,
version,
});
}
}
// Regular (non-dense) nodes
for node in &group.nodes {
let lat_deg = (lat_off + gran * node.lat.unwrap_or(0)) as f64 / 1_000_000_000.0;
let lon_deg = (lon_off + gran * node.lon.unwrap_or(0)) as f64 / 1_000_000_000.0;
// Region clip — regular nodes carry their own tags, no shared cursor.
if !clip_keeps(clip, lon_deg, lat_deg) {
continue;
}
let version = node.info.as_ref().and_then(|i| i.version);
let tags_json =
tags::serialize_tags(node.keys.iter().zip(node.vals.iter()).filter_map(
|(&k, &v)| Some((*strtab.get(k as usize)?, *strtab.get(v as usize)?)),
));
let id = node.id.unwrap_or(0);
coords.push((id, lat_deg as f32, lon_deg as f32));
records.push(NodeRecord {
id,
lon_lat: (lon_deg, lat_deg),
tags_json,
version,
});
}
}
(records, coords, has_way_or_rel)
}
/// Parse one `PrimitiveBlock` into resolved ways and relations. With an `oracle` the
/// relations are sieved through [`RelationClipOracle::verdict`] as they are decoded —
/// kept rows go to the second element, unsettled ones and the drop count to the
/// fourth; without one every relation is kept and the sieve is empty.
fn parse_ways_rels(
bytes: &[u8],
store: &NodeStore,
clip_active: bool,
oracle: Option<&RelationClipOracle>,
) -> (Vec<CollectedWay>, Vec<RelationRecord>, ResolveLoss, RelSieve) {
use crate::proto::osmformat::PrimitiveBlock;
use protobuf::Message as _;
let block = match PrimitiveBlock::parse_from_bytes(bytes) {
Ok(b) => b,
Err(_) => {
return (
Vec::new(),
Vec::new(),
ResolveLoss::default(),
RelSieve::default(),
);
}
};
let strtab = string_table(&block);
let mut ways: Vec<CollectedWay> = Vec::new();
let mut rels: Vec<RelationRecord> = Vec::new();
let mut sieve = RelSieve::default();
// Collect all way metadata + refs, then batch-resolve all node IDs at once.
struct WayPending {
id: i64,
version: Option<i32>,
tags_json: String,
refs: Vec<i64>,
is_area: bool,
ref_offset: usize, // index into all_refs
}
let mut pending: Vec<WayPending> = Vec::new();
let mut all_refs: Vec<i64> = Vec::new();
// One scratch buffer for the whole block, reused per way. The pairs are
// slices into the block's own string table — no copy, and after the first
// way no allocation either (stinky `katana-5`; guarded by
// `tests::way_tag_parsing_does_not_allocate_per_tag`).
let mut tag_pairs: Vec<(&str, &str)> = Vec::new();
for group in &block.primitivegroup {
for way in &group.ways {
let id = way.id.unwrap_or(0);
let version = way.info.as_ref().and_then(|i| i.version);
tag_pairs.clear();
tag_pairs.extend(
way.keys.iter().zip(way.vals.iter()).filter_map(|(&k, &v)| {
Some((*strtab.get(k as usize)?, *strtab.get(v as usize)?))
}),
);
// `serialize_tags` sorts into a `BTreeMap` internally, so feeding it
// pairs in file order instead of `HashMap` order changes nothing in
// the emitted JSON — see `tags::serialize_tags`.
let tags_json = tags::serialize_tags(tag_pairs.iter().copied());
let mut r: i64 = 0;
let refs: Vec<i64> = way
.refs
.iter()
.map(|&delta| {
r += delta;
r
})
.collect();
let is_area = geometry::is_closed_area(&refs, &tag_pairs);
let ref_offset = all_refs.len();
all_refs.extend_from_slice(&refs);
pending.push(WayPending {
id,
version,
tags_json,
refs,
is_area,
ref_offset,
});
}
for rel in &group.relations {
let id = rel.id.unwrap_or(0);
let version = rel.info.as_ref().and_then(|i| i.version);
let tags_json =
tags::serialize_tags(rel.keys.iter().zip(rel.vals.iter()).filter_map(
|(&k, &v)| Some((*strtab.get(k as usize)?, *strtab.get(v as usize)?)),
));
// Delta-decode member IDs
let mut memid: i64 = 0;
let members: Vec<MemberEntry> = rel
.memids
.iter()
.zip(rel.roles_sid.iter())
.zip(rel.types.iter())
.filter_map(|((&dm, &role_sid), mtype)| {
memid += dm;
let role = strtab
.get(role_sid as usize)
.copied()
.unwrap_or("")
.to_string();
use crate::proto::osmformat::relation::MemberType;
let member_type = match mtype.enum_value_or_default() {
MemberType::NODE => "node",
MemberType::WAY => "way",
MemberType::RELATION => "relation",
};
Some(MemberEntry {
member_type,
member_ref: memid,
role,
})
})
.collect();
// The clip prefilter, BEFORE the members are serialised: a dropped
// relation costs one verdict and no JSON (`RelationClipOracle::admit`).
let kept = RelationClipOracle::admit(
oracle,
store,
&members,
|| RelationRecord {
id,
tags_json,
members_json: MemberEntry::json(&members),
version,
},
&mut sieve,
);
if let Some(record) = kept {
rels.push(record);
}
}
}
// Batch-resolve all node refs at once using Ragnar's pipelined prefetch
let resolved = store.lookup_batch(&all_refs);
let mut loss = ResolveLoss::default();
for wp in pending {
let n = wp.refs.len();
let coords: Vec<(f64, f64)> = resolved[wp.ref_offset..wp.ref_offset + n]
.iter()
.filter_map(|opt| opt.map(|(lat, lon)| (lon as f64, lat as f64)))
.collect();
// A ref that resolved to nothing WITHOUT a clip is a silently missing
// vertex (audit finding (a)) — count it so the convert summary can say
// so. `note_way` is the ONE counting rule, shared with the XML seams.
loss.note_way(n, coords.len(), clip_active);
// Region-clip way semantics (node membership): with a clip active the
// node store holds ONLY in-region nodes, so a way with zero resolvable
// coords lies entirely outside the region → drop it. A way that crosses
// the boundary keeps its in-region endpoints (truncated, not edge-clipped
// — documented). Without a clip every way is kept (unchanged behaviour).
if clip_active && coords.is_empty() {
continue;
}
ways.push(CollectedWay {
id: wp.id,
coords,
tags_json: wp.tags_json,
node_refs: wp.refs,
version: wp.version,
is_area: wp.is_area,
});
}
(ways, rels, loss, sieve)
}
// ── Helpers ───────────────────────────────────────────────────────────────────
pub(crate) fn string_table<'a>(block: &'a crate::proto::osmformat::PrimitiveBlock) -> Vec<&'a str> {
block
.stringtable
.as_ref()
.map(|st| {
st.s.iter()
.map(|b| std::str::from_utf8(b).unwrap_or(""))
.collect()
})
.unwrap_or_default()
}
/// Consume dense node key-value pairs up to (and including) the 0 separator.
/// Returns (tags_json, new_kv_pos).
/// Advance the shared dense key/value cursor past one node's tags WITHOUT
/// building any string — used for clipped-out dense nodes so the next node's
/// cursor stays in sync at zero JSON/alloc cost.
#[inline]
fn skip_dense_tags(kv: &[i32], mut pos: usize) -> usize {
while let Some(&k) = kv.get(pos) {
pos = pos.saturating_add(1);
if k == 0 {
break;
}
pos = pos.saturating_add(1); // skip value
}
pos
}
fn collect_dense_tags(kv: &[i32], mut pos: usize, strtab: &[&str]) -> (String, usize) {
let mut pairs: Vec<(&str, &str)> = Vec::new();
loop {
let k = kv.get(pos).copied().unwrap_or(0);
pos = pos.saturating_add(1);
if k == 0 {
break;
}
let v = kv.get(pos).copied().unwrap_or(0);
pos = pos.saturating_add(1);
if let (Some(&key), Some(&val)) = (strtab.get(k as usize), strtab.get(v as usize)) {
pairs.push((key, val));
}
}
(tags::serialize_tags(pairs.into_iter()), pos)
}
#[cfg(test)]
mod relation_clip_oracle_law {
use super::*;
fn store_of(ids: &[i64]) -> NodeStore {
NodeStore::from_coords(ids.iter().map(|&i| (i, 0.0, 0.0)).collect())
}
/// The verdict, member by member, with the fixpoint — the unit half of
/// `tests/relation_clip_prefilter.rs`. Every arm is a literal.
#[test]
fn the_oracle_keeps_on_any_surviving_member_and_defers_on_relations_only() {
let o = RelationClipOracle::new(vec![30, 10, 10, 20]);
assert_eq!(o.way_count(), 3, "sorted + deduplicated");
let store = store_of(&[1, 2, 3]);
let v = |m: &[(&str, i64)]| o.verdict(&store, m.iter().copied());
assert_eq!(v(&[("way", 10)]), RelVerdict::Keep);
assert_eq!(v(&[("way", 99)]), RelVerdict::Drop);
assert_eq!(v(&[("node", 2)]), RelVerdict::Keep);
assert_eq!(v(&[("node", 9)]), RelVerdict::Drop);
assert_eq!(v(&[("way", 99), ("node", 9), ("way", 20)]), RelVerdict::Keep);
assert_eq!(v(&[("relation", 5)]), RelVerdict::Pending(vec![5]));
assert_eq!(
v(&[("way", 99), ("relation", 5), ("relation", 6)]),
RelVerdict::Pending(vec![5, 6])
);
assert_eq!(v(&[]), RelVerdict::Drop, "no members at all is nothing to draw");
// The fixpoint: 7 → 4 → 1, with 1 kept; 5 → 2, with 2 dropped.
let rec = |id: i64| RelationRecord {
id,
tags_json: String::from("{}"),
members_json: String::from("[]"),
version: None,
};
let pending = vec![
PendingRelation { record: rec(7), rel_members: vec![4] },
PendingRelation { record: rec(5), rel_members: vec![2] },
PendingRelation { record: rec(4), rel_members: vec![1] },
];
let mut kept: std::collections::HashSet<i64> = [1].into_iter().collect();
let mut promoted: Vec<i64> = RelationClipOracle::settle(pending, &mut kept)
.into_iter()
.map(|r| r.id)
.collect();
promoted.sort_unstable();
assert_eq!(promoted, vec![4, 7], "4 via 1, then 7 via 4; 5 never");
assert!(kept.contains(&7) && !kept.contains(&5));
}
/// **MEASURED over the live Sweden cut** — the number the design doc quotes.
///
/// Replays the oracle over the real `relations.parquet` (the planet's 14.6 M
/// relations, written by the pre-prefilter cut) against the real `ways.parquet`
/// and `nodes.parquet` id columns, writes the kept rows as a real parquet with the
/// same compression, and reports rows and bytes before and after. Then the proof
/// that the prefilter is a SUPERSET of resolve: every id in the live
/// `relations.resolved.parquet` must be in the kept set.
///
/// `#[ignore]` because it needs the cut on disk (`OSM_KATANA_LIVE_CUT`, default
/// `/home/korp/.korp/work/sweden/unpacked-backup`) and ~2 GB resident:
/// `cargo test --release -p osm-katana --lib -- the_prefilter_over_the_live_sweden_cut --ignored --nocapture`
#[test]
#[ignore]
fn the_prefilter_over_the_live_sweden_cut() {
use arrow::array::{Array, Int32Array, Int64Array, StringArray};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use std::path::PathBuf;
let dir = PathBuf::from(
std::env::var("OSM_KATANA_LIVE_CUT")
.unwrap_or_else(|_| String::from("/home/korp/.korp/work/sweden/unpacked-backup")),
);
let t0 = std::time::Instant::now();
let way_ids = crate::relgeom::read_id_column(&dir.join("ways.parquet")).unwrap();
let oracle = RelationClipOracle::new(way_ids);
let node_ids = crate::relgeom::read_id_column(&dir.join("nodes.parquet")).unwrap();
let n_nodes = node_ids.len();
let store = store_of(&node_ids);
drop(node_ids);
println!(
"oracle: {} ways, {} nodes, built in {:.1} s",
oracle.way_count(),
n_nodes,
t0.elapsed().as_secs_f64()
);
#[derive(serde::Deserialize)]
struct M {
#[serde(rename = "type")]
kind: String,
#[serde(rename = "ref")]
id: i64,
}
let rels_path = dir.join("relations.parquet");
let before_bytes = std::fs::metadata(&rels_path).unwrap().len();
let f = std::fs::File::open(&rels_path).unwrap();
let reader = ParquetRecordBatchReaderBuilder::try_new(f)
.unwrap()
.with_batch_size(65_536)
.build()
.unwrap();
let mut before_rows = 0u64;
let mut kept: Vec<RelationRecord> = Vec::new();
let mut kept_ids: std::collections::HashSet<i64> = std::collections::HashSet::new();
let mut pending: Vec<PendingRelation> = Vec::new();
let mut dropped = 0u64;
for b in reader {
let b = b.unwrap();
let ids = b.column_by_name("id").unwrap().as_any().downcast_ref::<Int64Array>().unwrap();
let tags = b.column_by_name("tags").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
let mems = b.column_by_name("members").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
let vers = b.column_by_name("version").unwrap().as_any().downcast_ref::<Int32Array>().unwrap();
for i in 0..b.num_rows() {
before_rows += 1;
let members: Vec<M> = serde_json::from_str(mems.value(i)).unwrap_or_default();
let verdict = oracle.verdict(&store, members.iter().map(|m| (m.kind.as_str(), m.id)));
let record = || RelationRecord {
id: ids.value(i),
tags_json: tags.value(i).to_string(),
members_json: mems.value(i).to_string(),
version: (!vers.is_null(i)).then(|| vers.value(i)),
};
match verdict {
RelVerdict::Keep => {
kept_ids.insert(ids.value(i));
kept.push(record());
}
RelVerdict::Pending(rel_members) => pending.push(PendingRelation {
record: record(),
rel_members,
}),
RelVerdict::Drop => dropped += 1,
}
}
}
let n_pending = pending.len() as u64;
let promoted = RelationClipOracle::settle(pending, &mut kept_ids);
let via_relations = promoted.len() as u64;
kept.extend(promoted);
let after_rows = kept.len() as u64;
let out = PathBuf::from(
std::env::var("OSM_KATANA_PREFILTER_OUT")
.unwrap_or_else(|_| String::from("/tmp/relations.prefiltered.parquet")),
);
crate::writer::write_relations(&out, &kept, "zstd").unwrap();
let after_bytes = std::fs::metadata(&out).unwrap().len();
println!("relations.parquet BEFORE: {before_rows} rows, {before_bytes} bytes");
println!(
"relations.parquet AFTER : {after_rows} rows, {after_bytes} bytes → {}",
out.display()
);
println!(
" dropped on the worker {dropped} · pending {n_pending} → promoted via a relation \
member {via_relations}, dropped by the fixpoint {} · rows kept {:.3} % · bytes kept \
{:.3} % · {:.1} s",
n_pending - via_relations,
after_rows as f64 * 100.0 / before_rows as f64,
after_bytes as f64 * 100.0 / before_bytes as f64,
t0.elapsed().as_secs_f64()
);
// The SUPERSET proof, on the live resolve output.
let resolved = crate::relgeom::read_id_column(&dir.join("relations.resolved.parquet")).unwrap();
let lost: Vec<i64> = resolved.iter().copied().filter(|id| !kept_ids.contains(id)).collect();
assert!(
lost.is_empty(),
"{} of {} resolved relations would have been dropped by the prefilter: {:?}…",
lost.len(),
resolved.len(),
&lost[..lost.len().min(10)]
);
println!(" superset: all {} resolved relations are in the kept set", resolved.len());
assert!(
after_rows * 10 < before_rows,
"the prefilter kept {after_rows} of {before_rows} rows — less than a 90 % saving"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pbf_enc::fixtures::{block as blob, pbf_file};
use crate::shared::gpu_backend::CpuBackend;
/// Core-saturation correctness for `read_pass2_mmap`: the reworked path
/// (parse + WKB-encode + build WayRecords ON THE WORKER via a balanced
/// fan-out, then concatenate) must yield byte-identical WKB geometries, in
/// the same order, as a SERIAL reference that parses every blob in order,
/// encodes WKB once on one core, and zips. Pushing the encode into the worker
/// only changes WHICH core does each blob — never the bytes or the order.
#[test]
fn read_pass2_parallel_wkb_matches_serial() {
// SKEWED across three blobs: blob 0 carries one fat way (60 nodes),
// blobs 1 and 2 carry several tiny ways — the exact shape that tail-chokes
// a plain fan-out and that the LPT balance is meant to spread.
let heavy_ids: Vec<i64> = (1..=60).collect();
let b0 = blob(&heavy_ids, &[heavy_ids.clone()]);
let b1 = blob(
&[100, 101, 102, 103],
&[vec![100, 101, 102], vec![101, 102, 103]],
);
let b2 = blob(&[200, 201, 202], &[vec![200, 201, 202], vec![200, 202]]);
let data = pbf_file(&[b0, b1, b2]);
let positions = crate::pbf_io::scan_blob_positions_par(&data);
assert!(
positions.len() >= 3,
"expected ≥3 blobs, got {}",
positions.len()
);
let store = read_pass1_mmap(&data, &positions).unwrap();
// Parallel path under test.
let backend = CpuBackend;
let (par_ways, _rels, _loss) = read_pass2_mmap(&data, &positions, &store, &backend).unwrap();
let par_pairs: Vec<(i64, Option<Vec<u8>>)> = par_ways
.iter()
.map(|w| (w.id, w.geometry.clone()))
.collect();
// Serial reference: parse each blob in order, encode WKB once, zip.
let mut all_collected: Vec<CollectedWay> = Vec::new();
for p in &positions {
if let Some(bytes) = data
.get(p.offset..p.offset.saturating_add(p.length))
.and_then(|blob| pbf_decomp::decompress(blob).ok().flatten())
{
let (c, _r, _loss, _sieve) = parse_ways_rels(&bytes, &store, false, None);
all_collected.extend(c);
}
}
let raw: Vec<RawWay> = all_collected
.iter()
.map(|w| RawWay {
coords: w.coords.clone(),
is_area: w.is_area,
})
.collect();
let wkb = CpuBackend.encode_wkb(&raw);
let serial_pairs: Vec<(i64, Option<Vec<u8>>)> = all_collected
.iter()
.zip(wkb)
.map(|(w, g)| (w.id, g))
.collect();
assert!(!serial_pairs.is_empty(), "reference produced no ways");
assert!(
serial_pairs.iter().any(|(_, g)| g.is_some()),
"reference produced no WKB geometry"
);
assert_eq!(
par_pairs, serial_pairs,
"parallel in-worker WKB must equal the serial-WKB reference (id + bytes + order)"
);
}
// ── stinky katana-5: the per-way tag map ─────────────────────────────────
//
// `parse_ways_rels` used to build a `HashMap<String, String>` per way — two
// `to_string()` per tag off string-table slices that are already `&str` —
// for the sole purpose of answering `is_closed_area`, and then handed the
// same map to `serialize_tags`, which sorts into a `BTreeMap` regardless.
// MEASURED on the sweden extract: 8 594 851 ways, so 8,59 M throwaway maps
// and ~2 String allocations per tag pair, per convert, inside the gatling
// workers where memory bandwidth is the shared resource.
//
// It came BACK once, in the pass-2-plan commit, together with the deletion
// of this guard — which is the only reason the revert was invisible. The
// guard lives here permanently now.
//
// The guard is a SCALING law, not a magic constant: hold the way count
// fixed and raise the tag count, and the allocation count must barely move.
// A constant would drift with every unrelated allocation in the parser; the
// slope is a property of the defect itself.
/// Decompress one block and drive the REAL `parse_ways_rels` on this
/// thread, returning (ways parsed, allocations made).
fn parse_allocs(n_ways: usize, tags_per_way: usize) -> (usize, u64) {
let data = pbf_file(&[crate::pbf_enc::fixtures::tagged_block(n_ways, tags_per_way)]);
let positions = crate::pbf_io::scan_blob_positions_par(&data);
let store = read_pass1_mmap(&data, &positions).unwrap();
let p = positions.last().expect("no blobs");
let bytes = pbf_decomp::decompress(&data[p.offset..p.offset + p.length])
.unwrap()
.expect("blob did not decompress");
let ((ways, _rels, _loss, _sieve), allocs) =
crate::alloc_probe::count_allocs(|| parse_ways_rels(&bytes, &store, false, None));
(ways.len(), allocs)
}
#[test]
fn way_tag_parsing_does_not_allocate_per_tag() {
const WAYS: usize = 400;
const LEAN: usize = 2;
const FAT: usize = 18;
/// Allocations a single extra tag on a single way may cost. The PBF tag
/// keys and values are already `&str` slices into the block's string
/// table, so the honest answer is **zero**; the budget is 0.25 only
/// because `serialize_tags` still builds a `BTreeMap` whose nodes hold
/// 11 entries, which is a per-way step, not a per-tag one. The defect
/// costs 2 `to_string()` per tag.
const BUDGET_PER_TAG: f64 = 0.25;
let (lean_ways, lean) = parse_allocs(WAYS, LEAN);
let (fat_ways, fat) = parse_allocs(WAYS, FAT);
assert_eq!(lean_ways, WAYS, "lean corpus lost ways");
assert_eq!(fat_ways, WAYS, "fat corpus lost ways");
assert!(
lean > 0,
"the allocation probe read zero — the instrument is dead"
);
let per_tag = (fat as f64 - lean as f64) / (WAYS * (FAT - LEAN)) as f64;
println!(
"katana-5 tag path: {lean} allocations at {LEAN} tags/way, {fat} at {FAT} tags/way, \
over {WAYS} ways = {per_tag:.3} allocations per tag (budget {BUDGET_PER_TAG:.2})"
);
assert!(
per_tag <= BUDGET_PER_TAG,
"parsing {WAYS} ways cost {lean} allocations at {LEAN} tags/way and {fat} at \
{FAT} tags/way — {per_tag:.2} allocations per tag, budget {BUDGET_PER_TAG:.2}. \
The tag path is allocating per tag again (katana-5).",
);
}
}