osm-katana/src/relgeom.rs
#![allow(clippy::arithmetic_side_effects)]
//! **The relation path.** Turns the `relations.parquet` a convert produces —
//! every relation on the planet, with a NULL geometry column — into a regional,
//! geometry-carrying table the tiler can read.
//!
//! # What was measured before this existed (oden, 2026-08-03, korp's `sweden` cut)
//!
//! | fact | value |
//! |---|---|
//! | rows in `relations.parquet` | 14 300 803 |
//! | rows whose `geometry` is NULL | 14 300 803 — **100 %** |
//! | share of the extract's bytes | 1 012.7 MB of 2 607.8 MB, **38.8 %** |
//! | distinct relation ids / max id | 14 300 803 / 20 593 321 |
//! | occurrences of the word `relations` in `tiles.rs` | **0** |
//!
//! So the largest single table in the extract was the planet's whole relation
//! space, carried through a *regional* cut, holding no geometry, read by nothing.
//!
//! # Why the order is clip → assemble and not assemble → clip
//!
//! Assembly is a member join: it needs every member way's resolved coordinates in
//! memory at once. Doing that for 14.3 M relations to keep the ~2 % a regional cut
//! can even reference is 50× the residency for the same answer — and residency is
//! the thing that took this box down (an `ldeflate` test binary reached 474.9 GiB
//! RSS on 2026-08-03). The clip is a sorted-`i64` membership test against
//! `ways.parquet`'s `id` column (74 MB for Sweden's 9 204 689 ways); it runs first,
//! and only what survives it is ever joined. Filed `katana-12` before `katana-11`
//! for exactly this reason.
//!
//! # The taxonomy (shared with the `ways`-side network layer)
//!
//! [`RelationKind`] is the classification both sides agree on, so a route that
//! exists only as a relation and a route carried by tagged ways are countable in
//! the same buckets. Route relations name their mode with the SAME `route=` values
//! the way-side layer sees on `highway=path` members — see [`RelationKind::route`].
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::Path;
use anyhow::Context as _;
use arrow::array::{
Array, BinaryArray, Int32Array, Int64Array, LargeListArray, ListArray, StringArray,
};
use arrow::record_batch::RecordBatch;
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ParquetRecordBatchReaderBuilder};
use crate::relations::{MemberWay, build_multipolygon, is_assembled_relation};
use crate::shared::geometry::decode_coords;
use crate::writer::{RelationRow, RelationWriter};
/// **What a relation is, for the map.** The one classification the relation path
/// and the way-side network layer share.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum RelationKind {
/// `type=multipolygon` / `type=boundary` / `boundary=administrative` — an
/// AREA, assembled into a WKB MultiPolygon with holes. Lakes with islands,
/// forest with clearings, farmyards, building outlines.
Area,
/// `type=route` / `type=route_master` / `type=superroute` — an ordered set of
/// member ways that is one named thing. The payload is the `route=` value:
/// `bus`, `hiking`, `bicycle`, `foot`, `mtb`, `ski`, `train`, `tram`, `ferry`…
///
/// This is the field the way-side network layer joins on. A `route=hiking`
/// relation over `highway=path` members is a trail that exists ONLY here —
/// the member ways carry no name and no network of their own.
Route(String),
/// `type=public_transport` — a stop area / stop-area group. Not geometry;
/// it is what binds a `route=bus` to its platforms.
PublicTransport,
/// Everything else, keyed by its `type=` value (`restriction`,
/// `associatedStreet`, `destination_sign`, `site`, `TMC`, …). Carried through
/// with members and tags but no geometry — a turn restriction is a rule, not
/// a shape.
Other(String),
}
impl RelationKind {
/// Classify from the relation's decoded tag map.
pub fn of(tags: &serde_json::Value) -> Self {
let ty = tags
.get("type")
.and_then(serde_json::Value::as_str)
.unwrap_or("");
if is_assembled_relation(tags) {
return Self::Area;
}
match ty {
"route" | "route_master" | "superroute" => Self::Route(
tags.get("route")
.or_else(|| tags.get("route_master"))
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string(),
),
"public_transport" => Self::PublicTransport,
other => Self::Other(other.to_string()),
}
}
/// The `route=` mode, for a route relation. `None` for anything else.
/// The join key with the way-side network layer.
pub fn route(&self) -> Option<&str> {
match self {
Self::Route(m) => Some(m.as_str()),
_ => None,
}
}
/// True when this kind gets an assembled MultiPolygon geometry.
pub fn is_area(&self) -> bool {
matches!(self, Self::Area)
}
/// A short stable label for stats/reporting.
pub fn label(&self) -> String {
match self {
Self::Area => "area".to_string(),
Self::Route(m) if m.is_empty() => "route=<none>".to_string(),
Self::Route(m) => format!("route={m}"),
Self::PublicTransport => "public_transport".to_string(),
Self::Other(t) if t.is_empty() => "<no type>".to_string(),
Self::Other(t) => format!("type={t}"),
}
}
}
/// Knobs for [`resolve_relations`].
#[derive(Debug, Clone)]
pub struct RelationResolveOptions {
/// Parquet compression for the output (`zstd` / `snappy` / `none`).
pub compression: String,
/// **Hard ceiling on the member-way coordinate table**, in vertices. The join
/// holds `(node_id, lon, lat)` per member vertex = 24 B, so the default 400 M
/// vertices is a ~9.6 GiB ceiling and the run FAILS at it rather than pages
/// the box. There is no version of this that is allowed to be "however much
/// the data needs": an unbounded member join on planet input is an OOM with a
/// plan.
pub max_member_vertices: usize,
/// Keep relations that are not areas (routes, restrictions, …). They carry no
/// geometry, but a bus route with no row is a bus route that does not exist.
pub keep_non_areas: bool,
/// Also admit a relation whose only members inside the cut are **nodes**.
///
/// MEASURED cost/benefit on the 2026-08-03 Sweden cut: the node-id oracle is
/// `nodes.parquet`'s 102 518 979 ids = **820 MB resident**, and it admits
/// 9 882 further relations — 3.7 % of the kept set, almost all
/// `type=public_transport` stop areas whose members are platform NODES. Worth
/// it for a regional cut, not obviously worth it for a planet one, so it is a
/// knob and its price is written down.
pub members_via_nodes: bool,
/// Also write the way-grain bridge table `route_members.parquet` here.
pub route_members: Option<std::path::PathBuf>,
/// Emit the per-relation ring-assembly complaints `build_multipolygon` logs.
/// Off by default: Sweden produces ~10⁵ of them and they are not news.
pub verbose_ring_errors: bool,
/// Hilbert-pack `relations.resolved.parquet` before returning, through
/// [`crate::spatial::pack_in_place`].
///
/// Default `true`, and the same reason `ConvertOptions::pack` is: the streaming
/// [`RelationWriter`] emits in `relations.parquet` order, so every row group spans
/// the whole extract. MEASURED on the Sweden cut — 274 206 rows, 3.96 s, 190.67 MB
/// → 192.78 MB (+1.1 %), and the central-Stockholm viewport then reads 10 of 34
/// row groups instead of all of them.
pub pack: bool,
}
impl Default for RelationResolveOptions {
fn default() -> Self {
Self {
compression: String::from("zstd"),
max_member_vertices: 400_000_000,
keep_non_areas: true,
members_via_nodes: true,
route_members: None,
verbose_ring_errors: false,
pack: true,
}
}
}
/// What one [`resolve_relations`] run did — every number the report needs.
#[derive(Debug, Clone, Default)]
pub struct RelationResolveStats {
pub relations_in: u64,
/// Survived the clip: at least one member way is present in `ways.parquet`.
pub kept: u64,
/// Of `kept`, the ones classified [`RelationKind::Area`].
pub areas: u64,
/// Of `areas`, the ones that produced a non-empty MultiPolygon.
pub assembled: u64,
/// Of `assembled`, the ones with at least one hole.
pub with_holes: u64,
pub outer_rings: u64,
pub inner_rings: u64,
/// Kept only because a member **node** is in the cut (no member way is).
pub kept_via_nodes: u64,
/// Kept only because a member **relation** is kept — the `route_master` case.
pub kept_via_relations: u64,
/// Distinct member way ids the area relations asked for.
pub member_ways_wanted: u64,
/// …and how many were found with usable geometry in `ways.parquet`.
pub member_ways_loaded: u64,
pub member_vertices: u64,
pub geometry_bytes: u64,
/// Kept-count per [`RelationKind::label`].
pub by_kind: BTreeMap<String, u64>,
/// Kept-count per `route=` mode, for the network-layer join.
pub by_route: BTreeMap<String, u64>,
/// The bridge table, when one was asked for.
pub route_members: Option<RouteMemberStats>,
}
impl RelationResolveStats {
/// One human line per the numbers that decide whether this is a small or a
/// large problem. Ratios, not raw counts (LAW 7).
pub fn summary(&self) -> String {
let pct = |a: u64, b: u64| {
if b == 0 {
0.0
} else {
100.0 * a as f64 / b as f64
}
};
format!(
"relations {} in → {} kept ({:.2} % survive the clip) · areas {} of kept ({:.1} %), \
assembled {} ({:.1} % of areas), {} with holes ({:.1} % of assembled); \
rings {} outer / {} inner; member ways {} wanted → {} loaded ({:.1} %), \
{} vertices; geometry {:.1} MB; admitted via node members {} / via member \
relations {}",
self.relations_in,
self.kept,
pct(self.kept, self.relations_in),
self.areas,
pct(self.areas, self.kept),
self.assembled,
pct(self.assembled, self.areas),
self.with_holes,
pct(self.with_holes, self.assembled),
self.outer_rings,
self.inner_rings,
self.member_ways_wanted,
self.member_ways_loaded,
pct(self.member_ways_loaded, self.member_ways_wanted),
self.member_vertices,
self.geometry_bytes as f64 / 1e6,
self.kept_via_nodes,
self.kept_via_relations,
)
}
}
/// The member reference shape in `relations.parquet`'s `members` JSON.
#[derive(serde::Deserialize)]
struct RawMember {
#[serde(rename = "type")]
member_type: String,
#[serde(rename = "ref")]
member_ref: i64,
#[serde(default)]
role: String,
}
/// **Clip, then assemble.** Reads `<dir>/relations.parquet` and `<dir>/ways.parquet`
/// and writes `<out>` — the relations that a member way in this extract can reach,
/// with `type=multipolygon` / `type=boundary` ones carrying an assembled WKB
/// MultiPolygon.
///
/// Three passes, each cheap in the thing the one before it made small:
/// 1. `ways.parquet`'s `id` column → one sorted `Vec<i64>` (the clip oracle).
/// 2. `relations.parquet` (`id`, `tags`, `members`) → the kept set + the member
/// way ids the AREAS want. Nothing is joined yet.
/// 3. `ways.parquet` again (`id`, `geometry`, `node_refs`), keeping only wanted
/// ids → the member table; then `relations.parquet` a second time to write.
pub fn resolve_relations(
dir: &Path,
out: &Path,
opts: &RelationResolveOptions,
) -> anyhow::Result<RelationResolveStats> {
let rels_path = dir.join("relations.parquet");
let ways_path = dir.join("ways.parquet");
anyhow::ensure!(rels_path.exists(), "no relations.parquet under {dir:?}");
anyhow::ensure!(
ways_path.exists(),
"no ways.parquet under {dir:?} — without it there is no clip oracle and no \
member geometry, and 'keep everything' is the 14.3 M-row planet table this \
stage exists to remove"
);
let mut stats = RelationResolveStats::default();
// ── 1. The clip oracle ───────────────────────────────────────────────────────
let mut way_ids = read_id_column(&ways_path)?;
way_ids.sort_unstable();
way_ids.dedup();
eprintln!(
"osm-katana/relations: clip oracle = {} way ids from {:?} ({:.1} MB resident)",
way_ids.len(),
ways_path,
(way_ids.len() * 8) as f64 / 1e6
);
// The optional node oracle. Its price is the whole point of the flag.
let node_ids: Vec<i64> = if opts.members_via_nodes && dir.join("nodes.parquet").exists() {
let mut v = read_id_column(&dir.join("nodes.parquet"))?;
v.sort_unstable();
v.dedup();
eprintln!(
"osm-katana/relations: node oracle = {} node ids ({:.0} MB resident)",
v.len(),
(v.len() * 8) as f64 / 1e6
);
v
} else {
Vec::new()
};
// ── 2. Decide what survives, and what geometry it will need ──────────────────
struct Kept {
id: i64,
area: bool,
outers: Vec<i64>,
inners: Vec<i64>,
kind: RelationKind,
}
let mut kept: Vec<Kept> = Vec::new();
let mut wanted: HashSet<i64> = HashSet::new();
// Relations with NO member way/node in the cut but with member RELATIONS:
// the `type=route_master` shape. Held for the fixpoint below.
let mut pending: Vec<(usize, Vec<i64>)> = Vec::new();
// Parallel to `kept`, so a relation promoted by the fixpoint can be finished
// without re-reading the file. `None` ⇒ not (yet) kept.
struct Held {
id: i64,
kind: RelationKind,
area: bool,
outers: Vec<i64>,
inners: Vec<i64>,
via_node: bool,
}
let mut held: Vec<Held> = Vec::new();
let mut kept_flag: Vec<bool> = Vec::new();
for_each_relation_batch(&rels_path, &["id", "tags", "members"], |b| {
let ids = col_i64(b, "id")?;
let tags = col_str(b, "tags")?;
let mems = col_str(b, "members")?;
for i in 0..b.num_rows() {
stats.relations_in += 1;
let tag_str = if tags.is_null(i) { "" } else { tags.value(i) };
let mem_str = if mems.is_null(i) { "" } else { mems.value(i) };
let tag_val: serde_json::Value =
serde_json::from_str(tag_str).unwrap_or(serde_json::Value::Null);
let kind = RelationKind::of(&tag_val);
let area = kind.is_area();
if !area && !opts.keep_non_areas {
continue;
}
let members: Vec<RawMember> = serde_json::from_str(mem_str).unwrap_or_default();
let mut outers = Vec::new();
let mut inners = Vec::new();
let mut via_way = false;
let mut via_node = false;
let mut rel_members: Vec<i64> = Vec::new();
for m in &members {
match m.member_type.as_str() {
// The CLIP: a member way present in this extract's ways.parquet.
"way" => {
if way_ids.binary_search(&m.member_ref).is_ok() {
via_way = true;
if area {
if m.role == "inner" {
inners.push(m.member_ref);
} else {
outers.push(m.member_ref);
}
wanted.insert(m.member_ref);
}
}
}
"node" => {
if !via_node && node_ids.binary_search(&m.member_ref).is_ok() {
via_node = true;
}
}
// A relation of relations. `type=route_master` binds the two
// directions of a bus line together and has NO way members at
// all — a way-only clip kept 7 of Sweden's 109 928 route_masters
// (measured), i.e. it deleted the bus NETWORK while keeping the
// individual runs. Resolved by a fixpoint, below.
"relation" => rel_members.push(m.member_ref),
_ => {}
}
}
let idx = held.len();
let in_cut = via_way || via_node;
if !in_cut && rel_members.is_empty() {
continue;
}
held.push(Held {
id: ids.value(i),
kind,
area,
outers,
inners,
via_node: via_node && !via_way,
});
kept_flag.push(in_cut);
if !in_cut {
pending.push((idx, rel_members));
}
}
Ok(())
})?;
// The fixpoint: a relation is in the cut when a relation it contains is. Runs
// to stability (bounded — OSM nests route_master → route, and `superroute`
// one deeper; 8 laps is far past anything real and stops a cycle dead).
{
let mut in_cut_ids: HashSet<i64> = held
.iter()
.zip(&kept_flag)
.filter(|&(_, &k)| k)
.map(|(h, _)| h.id)
.collect();
for _ in 0..8 {
let mut promoted = 0u64;
for (idx, members) in &pending {
if kept_flag[*idx] {
continue;
}
if members.iter().any(|m| in_cut_ids.contains(m)) {
kept_flag[*idx] = true;
in_cut_ids.insert(held[*idx].id);
promoted += 1;
}
}
stats.kept_via_relations += promoted;
if promoted == 0 {
break;
}
}
}
for (h, keep) in held.into_iter().zip(&kept_flag) {
if !keep {
continue;
}
stats.kept += 1;
if h.via_node {
stats.kept_via_nodes += 1;
}
*stats.by_kind.entry(h.kind.label()).or_default() += 1;
if let Some(r) = h.kind.route() {
*stats.by_route.entry(r.to_string()).or_default() += 1;
}
if h.area {
stats.areas += 1;
}
kept.push(Kept {
id: h.id,
area: h.area,
outers: h.outers,
inners: h.inners,
kind: h.kind,
});
}
drop(node_ids);
stats.member_ways_wanted = wanted.len() as u64;
eprintln!(
"osm-katana/relations: clip kept {} of {} relations ({:.3} %); {} of them are areas \
wanting {} distinct member ways",
stats.kept,
stats.relations_in,
if stats.relations_in == 0 {
0.0
} else {
100.0 * stats.kept as f64 / stats.relations_in as f64
},
stats.areas,
stats.member_ways_wanted,
);
// ── 3a. The member table — ONLY the ways the surviving areas asked for ───────
let mut way_map: HashMap<i64, MemberWay> = HashMap::with_capacity(wanted.len());
let mut vertices: usize = 0;
// **Row 22: the ids may live beside the table rather than in it.** Read ONCE, and
// only for the ways the surviving areas actually asked for — `wanted` is the member
// set, typically a few hundred thousand ids out of 9.3 M rows, so this map is a
// fraction of the column it replaces.
let sidecar = if wanted.is_empty() {
None
} else {
read_node_refs_sidecar(&ways_path, &wanted)?
};
if !wanted.is_empty() {
for_each_way_batch(&ways_path, |b| {
let ids = col_i64(b, "id")?;
let geo = col_bin(b, "geometry")?;
let refs = col_i64_list(b, "node_refs");
// Neither in the table nor beside it. Before row 22 this fell through to
// `nodes.len() != coords.len()` and skipped every member way in SILENCE —
// a cut made with `--node-refs false` resolved zero areas and reported
// success. `assemble_rings` matches ring endpoints on node id; without the
// ids there is no assembly to do and saying so is the only honest answer.
anyhow::ensure!(
refs.is_some() || sidecar.is_some(),
"{:?} has no `node_refs` column and there is no {:?} beside it, but {} \
member ways are wanted. Ring assembly matches endpoints on NODE ID, so \
every one of them would be skipped and this cut would resolve zero \
areas. Re-pack with `--node-refs-sidecar true`, or cut with the column.",
ways_path,
crate::spatial::node_refs_sidecar_path(&ways_path),
wanted.len(),
);
for i in 0..b.num_rows() {
let id = ids.value(i);
if !wanted.contains(&id) {
continue;
}
if geo.is_null(i) {
continue;
}
let Some(coords) = decode_coords(geo.value(i)) else {
continue;
};
let nodes = match (&refs, &sidecar) {
(Some(f), _) => f(i),
(None, Some(m)) => m.get(&id).cloned().unwrap_or_default(),
(None, None) => unreachable!("refused above"),
};
// `assemble_rings` matches ring endpoints on NODE ID, so a way with
// no node_refs cannot be stitched. Synthesising ids from coordinates
// would join two ways that merely touch, which is a different — and
// silently wrong — map.
if nodes.len() != coords.len() || coords.len() < 2 {
continue;
}
vertices += coords.len();
anyhow::ensure!(
vertices <= opts.max_member_vertices,
"member-way coordinate table exceeded the {} vertex ceiling \
(~{:.1} GiB) while loading {} of {} wanted ways — raise \
--max-member-vertices deliberately or clip harder; this stage \
will not page the box",
opts.max_member_vertices,
(opts.max_member_vertices * 24) as f64 / (1u64 << 30) as f64,
way_map.len(),
wanted.len(),
);
way_map.insert(id, MemberWay { nodes, coords });
}
Ok(())
})?;
}
stats.member_ways_loaded = way_map.len() as u64;
stats.member_vertices = vertices as u64;
eprintln!(
"osm-katana/relations: member table {} ways / {} vertices (~{:.2} GiB) — assembling",
stats.member_ways_loaded,
stats.member_vertices,
(vertices * 24) as f64 / (1u64 << 30) as f64,
);
// ── 3b. Assemble, then write ────────────────────────────────────────────────
let mut geom_of: HashMap<i64, Vec<u8>> = HashMap::new();
for k in &kept {
if !k.area {
continue;
}
let gather = |refs: &[i64]| -> Vec<MemberWay> {
refs.iter()
.filter_map(|r| way_map.get(r).cloned())
.collect()
};
let outer_ways = gather(&k.outers);
let inner_ways = gather(&k.inners);
if outer_ways.is_empty() {
continue;
}
let assembled = if opts.verbose_ring_errors {
build_multipolygon(outer_ways, inner_ways, k.id)
} else {
crate::relations::build_multipolygon_quiet(outer_ways, inner_ways)
};
if let Some((wkb, outer, inner)) = assembled {
stats.assembled += 1;
stats.outer_rings += outer as u64;
stats.inner_rings += inner as u64;
if inner > 0 {
stats.with_holes += 1;
}
stats.geometry_bytes += wkb.len() as u64;
geom_of.insert(k.id, wkb);
}
}
drop(way_map);
let keep_ids: HashSet<i64> = kept.iter().map(|k| k.id).collect();
drop(kept);
// The `geo` declaration is derived from what we actually assembled — an empty
// list when nothing did, which is the TRUE statement and what
// `RelationWriter::finish` checks against the rows.
let declared: &[&str] = if geom_of.is_empty() {
&[]
} else {
&["MultiPolygon"]
};
let mut w = RelationWriter::create(out, &opts.compression, declared)?;
for_each_relation_batch(&rels_path, &["id", "tags", "members", "version"], |b| {
let ids = col_i64(b, "id")?;
let tags = col_str(b, "tags")?;
let mems = col_str(b, "members")?;
let vers = col_i32(b, "version").ok();
for i in 0..b.num_rows() {
let id = ids.value(i);
if !keep_ids.contains(&id) {
continue;
}
w.push(RelationRow {
id,
geometry: geom_of.get(&id).map(Vec::as_slice),
tags: if tags.is_null(i) { "" } else { tags.value(i) },
members: if mems.is_null(i) { "" } else { mems.value(i) },
version: vers
.as_ref()
.and_then(|v| (!v.is_null(i)).then(|| v.value(i))),
})?;
}
Ok(())
})?;
let (rows, with_geom) = w.finish()?;
anyhow::ensure!(
rows == stats.kept,
"wrote {rows} rows but the clip kept {} — the two passes over relations.parquet \
disagreed about the same file",
stats.kept
);
anyhow::ensure!(
with_geom == stats.assembled,
"wrote {with_geom} geometries but assembled {}",
stats.assembled
);
// The layout, through the one packer — same Hilbert order, same `covering.bbox`,
// same row-group size as `networks/roads.parquet` and as convert's base tables.
// The row-count checks above run FIRST: a table this stage rejected must never be
// rewritten into the shape a reader trusts.
if opts.pack {
let pack = crate::spatial::PackOptions {
compression: opts.compression.clone(),
..crate::spatial::PackOptions::default()
};
if let Some(p) = crate::spatial::pack_in_place(out, &pack)? {
eprintln!(
"osm-katana/relations: packed {out:?} — {} rows, {} row groups, {} bytes",
p.rows, p.row_groups, p.bytes
);
}
}
if let Some(rm_path) = &opts.route_members {
let rm = write_route_members(&rels_path, &ways_path, rm_path, &keep_ids, opts)?;
eprintln!(
"osm-katana/relations: route_members {:?} — {} route relations → {} distinct \
member ways, {} present in this cut ({:.1} %), {} (way,route) edges, {:.1} MB WKB",
rm_path,
rm.route_relations,
rm.ways_referenced,
rm.ways_written,
if rm.ways_referenced == 0 {
0.0
} else {
100.0 * rm.ways_written as f64 / rm.ways_referenced as f64
},
rm.memberships,
rm.geometry_bytes as f64 / 1e6,
);
stats.route_members = Some(rm);
}
eprintln!("osm-katana/relations: {}", stats.summary());
Ok(stats)
}
// ── Parquet plumbing ────────────────────────────────────────────────────────────
pub(crate) fn read_id_column(path: &Path) -> anyhow::Result<Vec<i64>> {
let f = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let md = ArrowReaderMetadata::load(&f, Default::default())?;
let mask = ProjectionMask::columns(md.metadata().file_metadata().schema_descr(), ["id"]);
let rows = md.metadata().file_metadata().num_rows().max(0) as usize;
let mut out = Vec::with_capacity(rows);
let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(f, md)
.with_projection(mask)
.with_batch_size(65_536)
.build()?;
for b in reader {
let b = b?;
let ids = col_i64(&b, "id")?;
out.extend(ids.values().iter().copied());
}
Ok(out)
}
fn for_each_relation_batch(
path: &Path,
cols: &[&str],
mut f: impl FnMut(&RecordBatch) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let file = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let md = ArrowReaderMetadata::load(&file, Default::default())?;
let mask = ProjectionMask::columns(
md.metadata().file_metadata().schema_descr(),
cols.iter().copied(),
);
let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(file, md)
.with_projection(mask)
.with_batch_size(16_384)
.build()?;
for b in reader {
f(&b?)?;
}
Ok(())
}
/// [`for_each_way_batch`] with `tags` as well — the bridge carries the way's own
/// tags through so a reader can see `highway=path` beside the route class.
fn for_each_way_batch_full(
path: &Path,
mut f: impl FnMut(&RecordBatch) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let file = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let md = ArrowReaderMetadata::load(&file, Default::default())?;
let mask = ProjectionMask::columns(
md.metadata().file_metadata().schema_descr(),
["id", "geometry", "tags", "node_refs"],
);
let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(file, md)
.with_projection(mask)
.with_batch_size(8_192)
.build()?;
for b in reader {
f(&b?)?;
}
Ok(())
}
fn for_each_way_batch(
path: &Path,
mut f: impl FnMut(&RecordBatch) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let file = std::fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let md = ArrowReaderMetadata::load(&file, Default::default())?;
let mask = ProjectionMask::columns(
md.metadata().file_metadata().schema_descr(),
["id", "geometry", "node_refs"],
);
let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(file, md)
.with_projection(mask)
.with_batch_size(8_192)
.build()?;
for b in reader {
f(&b?)?;
}
Ok(())
}
fn col_i64<'a>(b: &'a RecordBatch, name: &str) -> anyhow::Result<&'a Int64Array> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<Int64Array>())
.with_context(|| format!("column `{name}` is missing or not Int64"))
}
fn col_i32<'a>(b: &'a RecordBatch, name: &str) -> anyhow::Result<&'a Int32Array> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.with_context(|| format!("column `{name}` is missing or not Int32"))
}
fn col_str<'a>(b: &'a RecordBatch, name: &str) -> anyhow::Result<&'a StringArray> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.with_context(|| format!("column `{name}` is missing or not Utf8"))
}
fn col_bin<'a>(b: &'a RecordBatch, name: &str) -> anyhow::Result<&'a BinaryArray> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<BinaryArray>())
.with_context(|| format!("column `{name}` is missing or not Binary"))
}
/// `node_refs` as a per-row accessor, or `None` when the column is absent.
#[allow(clippy::type_complexity)]
/// **Read `ways.refs.parquet` — but only the ways the areas asked for** (row 22).
///
/// `Ok(None)` when there is no sidecar, which is the normal case for a table that still
/// carries its own `node_refs`; the caller refuses only when BOTH are absent.
///
/// # Why it filters, and why that is the whole point
///
/// The served `ways.parquet` carried 437.3 MB of `node_refs` (28.4 % of the file,
/// measured on the live Sweden cut) so that ring assembly could look up **the member
/// ways of the surviving relations** — 223 382 multipolygons' worth, not 9.3 M rows.
/// `wanted` is already computed by the time this runs, so the map built here holds the
/// ids that will actually be used and nothing else. Moving the column out of the served
/// table therefore costs korp nothing and costs this step less resident memory than
/// projecting the column did.
fn read_node_refs_sidecar(
ways_path: &Path,
wanted: &HashSet<i64>,
) -> anyhow::Result<Option<HashMap<i64, Vec<i64>>>> {
let path = crate::spatial::node_refs_sidecar_path(ways_path);
if !path.exists() {
return Ok(None);
}
let file = std::fs::File::open(&path).with_context(|| format!("open {path:?}"))?;
let md = ArrowReaderMetadata::load(&file, Default::default())?;
let mask = ProjectionMask::columns(
md.metadata().file_metadata().schema_descr(),
["id", "node_refs"],
);
let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(file, md)
.with_projection(mask)
.with_batch_size(8_192)
.build()?;
let mut out: HashMap<i64, Vec<i64>> = HashMap::with_capacity(wanted.len());
for b in reader {
let b = b?;
let ids = col_i64(&b, "id")?;
let refs = col_i64_list(&b, "node_refs").ok_or_else(|| {
anyhow::anyhow!("{path:?} has no `node_refs` column — it is not a refs sidecar")
})?;
for i in 0..b.num_rows() {
let id = ids.value(i);
if wanted.contains(&id) {
out.insert(id, refs(i));
}
}
}
eprintln!(
"osm-katana/relations: {} of {} wanted member ways found in {:?}",
out.len(),
wanted.len(),
path.file_name().unwrap_or_default(),
);
Ok(Some(out))
}
fn col_i64_list<'a>(b: &'a RecordBatch, name: &str) -> Option<Box<dyn Fn(usize) -> Vec<i64> + 'a>> {
let c = b.column_by_name(name)?;
if let Some(l) = c.as_any().downcast_ref::<ListArray>() {
return Some(Box::new(move |i| {
if l.is_null(i) {
return Vec::new();
}
let v = l.value(i);
v.as_any()
.downcast_ref::<Int64Array>()
.map(|a| a.values().to_vec())
.unwrap_or_default()
}));
}
if let Some(l) = c.as_any().downcast_ref::<LargeListArray>() {
return Some(Box::new(move |i| {
if l.is_null(i) {
return Vec::new();
}
let v = l.value(i);
v.as_any()
.downcast_ref::<Int64Array>()
.map(|a| a.values().to_vec())
.unwrap_or_default()
}));
}
None
}
// ── The bridge to the way-side network layer ────────────────────────────────────
/// **`route_members.parquet` — the ways that only a RELATION knows are a route.**
///
/// # Why this file exists
///
/// The two sides of the network measure the same country and disagree, because
/// they are looking at different objects. MEASURED on korp's 2026-08-03 `sweden`
/// cut, cross-referenced against the 9 204 689 Swedish way ids:
///
/// | | ways | raw WKB |
/// |---|---|---|
/// | carry trail tags of their own (`trails.parquet`) | 74 398 | 27.9 MB |
/// | member of a trail relation touching Sweden | 122 889 | 37.0 MB |
/// | **both** | **16 211** | |
/// | **union** | **181 076** | **58.7 MB** |
///
/// The way-tagged file covers 41 % of the union, the relation side 68 %, and the
/// overlap is 9.0 %. Neither is the trail network. A `highway=path` segment in a
/// marked hiking route usually carries no `route=`, no `network=` and no name —
/// the route is the relation, and the segment is just a path.
///
/// # Why it is written HERE and not by the way side
///
/// The member lists live in `relations.parquet`, which is 1 012.7 MB and 98.1 %
/// non-Swedish. Deriving this on the way side means re-reading that file for the
/// 1.9 % that matters. This stage has already read it and already knows which
/// relations survived the clip.
///
/// # The grain, and the collision it avoids
///
/// One row per **way**, keyed `id` — way-id space, NOT the relation-id space of
/// `relations.resolved.parquet`. Both are `Int64 not null` in the same schema
/// position, so a join across them compiles, runs and returns rows. The `network`
/// column (`"route_members"`) is what makes the namespace visible; it is the same
/// column and the same purpose the way side added for its per-network toggles.
pub const ROUTE_MEMBERS_NETWORK: &str = "route_members";
/// The columns of `route_members.parquet`: the way side's own
/// [`crate::network::network_fields`] — unchanged, in its order — plus the two
/// this bridge adds.
///
/// Derived rather than restated (LAW 5): a column added or retyped on the way side
/// appears here with no edit, and the shared prefix cannot drift. It is NOT the
/// same list, though, so it does not move the way side's `NetworkRecipe` — those
/// tables are theirs to describe.
pub fn route_member_fields() -> Vec<arrow::datatypes::Field> {
use arrow::datatypes::{DataType, Field};
let mut f = crate::network::network_fields();
// The relation ids this way is a member of. A way in a bus route AND a hiking
// route has both, so it is a list — collapsing it to one id is how a bridge
// silently drops the second network.
f.push(Field::new(
"route_ids",
DataType::List(std::sync::Arc::new(Field::new(
"item",
DataType::Int64,
true,
))),
true,
));
// `role` per entry of `route_ids`, same order: `forward` / `backward` / "".
// JSON rather than a second list because it is read for display, not filtered on.
f.push(Field::new("roles", DataType::Utf8, true));
f
}
/// The way side's class for a `route=` mode, or the mode itself when it is not a
/// trail.
///
/// **Asks [`crate::network::TRAILS`] rather than restating its table**, so
/// `route=piste → ski` is true here because it is true there, and a change on that
/// side moves this with no edit. `route=bus` / `road` / `ferry` are not trails and
/// keep their own name — the file carries every kind of route, and a reader that
/// only wants trails filters `class IN ('hiking','cycling','mtb','ski','snowmobile')`.
pub fn route_class(mode: &str) -> String {
if mode.is_empty() {
return String::from("unknown");
}
// The one tag a `Rule::KeyIn { key: "route", .. }` can match on.
let probe = format!("{{\"route\":\"{}\"}}", mode.replace('"', ""));
match crate::network::TRAILS.class(&probe) {
Some(c) => c.into_owned(),
None => mode.to_string(),
}
}
/// What one [`write_route_members`] run produced.
#[derive(Debug, Clone, Default)]
pub struct RouteMemberStats {
/// Route relations whose members were walked.
pub route_relations: u64,
/// Distinct member ways referenced by them.
pub ways_referenced: u64,
/// …of which present in this extract's `ways.parquet` with geometry.
pub ways_written: u64,
/// Total `(way, route)` edges — always ≥ `ways_written`.
pub memberships: u64,
/// Raw WKB bytes carried.
pub geometry_bytes: u64,
/// Rows per class.
pub by_class: BTreeMap<String, u64>,
}
/// Write the bridge table. `keep` is the relation-id set the clip produced.
///
/// Two passes and no new residency ceiling of its own: one over
/// `relations.parquet` for the kept ROUTE relations' member lists, one over
/// `ways.parquet` for the geometry of the ways they name. The way table is the
/// same one the clip oracle came from, so nothing new is opened.
fn write_route_members(
rels_path: &Path,
ways_path: &Path,
out: &Path,
keep: &HashSet<i64>,
opts: &RelationResolveOptions,
) -> anyhow::Result<RouteMemberStats> {
use arrow::array::{BinaryBuilder, Int64Builder, ListBuilder, StringBuilder};
use arrow::datatypes::Schema;
use std::sync::Arc;
let mut st = RouteMemberStats::default();
// way id → (route relation ids, roles, class)
let mut edges: HashMap<i64, (Vec<i64>, Vec<String>, String)> = HashMap::new();
for_each_relation_batch(rels_path, &["id", "tags", "members"], |b| {
let ids = col_i64(b, "id")?;
let tags = col_str(b, "tags")?;
let mems = col_str(b, "members")?;
for i in 0..b.num_rows() {
let id = ids.value(i);
if !keep.contains(&id) {
continue;
}
let tag_str = if tags.is_null(i) { "" } else { tags.value(i) };
let tag_val: serde_json::Value =
serde_json::from_str(tag_str).unwrap_or(serde_json::Value::Null);
let Some(mode) = RelationKind::of(&tag_val).route().map(str::to_string) else {
continue;
};
st.route_relations += 1;
let class = route_class(&mode);
let mem_str = if mems.is_null(i) { "" } else { mems.value(i) };
let members: Vec<RawMember> = serde_json::from_str(mem_str).unwrap_or_default();
for m in members {
if m.member_type != "way" {
continue;
}
let e = edges
.entry(m.member_ref)
.or_insert_with(|| (Vec::new(), Vec::new(), class.clone()));
e.0.push(id);
e.1.push(m.role.clone());
st.memberships += 1;
}
}
Ok(())
})?;
st.ways_referenced = edges.len() as u64;
// Pass 2: the geometry, taken byte for byte out of ways.parquet. Nothing is
// decoded or re-encoded — a MultiPolygon that is right in the source is right
// here without this stage having an opinion about it.
let fields = route_member_fields();
let geo = crate::metadata::GeoMeta::new(&["LineString", "Polygon", "MultiPolygon"]).to_json();
let mut meta = std::collections::HashMap::new();
meta.insert(String::from(crate::metadata::GEO_KEY), geo);
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
let mut batches: Vec<RecordBatch> = Vec::new();
let mut ids = Int64Builder::new();
let mut nets = StringBuilder::new();
let mut classes = StringBuilder::new();
let mut geoms = BinaryBuilder::new();
let mut tagcol = StringBuilder::new();
let mut refs = ListBuilder::new(Int64Builder::new());
let mut routes = ListBuilder::new(Int64Builder::new());
let mut roles = StringBuilder::new();
let mut staged = 0usize;
let mut flush = |ids: &mut Int64Builder,
nets: &mut StringBuilder,
classes: &mut StringBuilder,
geoms: &mut BinaryBuilder,
tagcol: &mut StringBuilder,
refs: &mut ListBuilder<Int64Builder>,
routes: &mut ListBuilder<Int64Builder>,
roles: &mut StringBuilder,
batches: &mut Vec<RecordBatch>|
-> anyhow::Result<()> {
batches.push(RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(ids.finish()),
Arc::new(nets.finish()),
Arc::new(classes.finish()),
Arc::new(geoms.finish()),
Arc::new(tagcol.finish()),
Arc::new(refs.finish()),
Arc::new(routes.finish()),
Arc::new(roles.finish()),
],
)?);
Ok(())
};
for_each_way_batch_full(ways_path, |b| {
let wid = col_i64(b, "id")?;
let geo = col_bin(b, "geometry")?;
let tg = col_str(b, "tags").ok();
let nr = col_i64_list(b, "node_refs");
for i in 0..b.num_rows() {
let id = wid.value(i);
let Some((rids, rroles, class)) = edges.get(&id) else {
continue;
};
if geo.is_null(i) {
continue;
}
ids.append_value(id);
nets.append_value(ROUTE_MEMBERS_NETWORK);
classes.append_value(class);
geoms.append_value(geo.value(i));
st.geometry_bytes += geo.value(i).len() as u64;
match &tg {
Some(t) if !t.is_null(i) => tagcol.append_value(t.value(i)),
_ => tagcol.append_null(),
}
match &nr {
Some(f) => {
for v in f(i) {
refs.values().append_value(v);
}
refs.append(true);
}
None => refs.append(false),
}
for r in rids {
routes.values().append_value(*r);
}
routes.append(true);
roles
.append_value(serde_json::to_string(rroles).unwrap_or_else(|_| String::from("[]")));
st.ways_written += 1;
*st.by_class.entry(class.clone()).or_default() += 1;
staged += 1;
if staged >= 65_536 {
flush(
&mut ids,
&mut nets,
&mut classes,
&mut geoms,
&mut tagcol,
&mut refs,
&mut routes,
&mut roles,
&mut batches,
)?;
staged = 0;
}
}
Ok(())
})?;
if staged > 0 {
flush(
&mut ids,
&mut nets,
&mut classes,
&mut geoms,
&mut tagcol,
&mut refs,
&mut routes,
&mut roles,
&mut batches,
)?;
}
// Hilbert-packed + `covering.bbox`, through the way side's own layout writer —
// one packer in the tree, so this file prunes exactly like `trails.parquet`.
let pack = crate::spatial::PackOptions {
compression: opts.compression.clone(),
row_group_rows: 8192,
covering: true,
// Assembled multipolygons are the single densest display geometry in the cut —
// coastlines, lakes, forests — so this is the table the generalisation columns
// were designed for (DELTA 2).
generalise: true,
// Row 22 is about the SERVED ways table; a resolved relation has no node_refs.
node_refs_sidecar: false,
};
crate::spatial::pack_batches(batches, &schema, out, &pack, &[])?;
Ok(st)
}
#[cfg(test)]
mod bridge_law {
//! FAIL-ON-BUG for the two ways `route_members.parquet` could be wrong in a
//! shape no row count would show.
use super::*;
/// ★ **RED-when-broken: the class table is ASKED of the way side, not copied
/// from it.**
///
/// The way side and this one agreed on `route=piste → ski` in conversation.
/// An agreement restated as a literal here is a copy, and a copy drifts the
/// first time `network::TRAILS` changes — silently, because both files would
/// still be full of valid rows in valid classes, just not the same ones.
///
/// Asserted against `TRAILS`'s OWN rules rather than against a literal, so
/// this cannot rot into the very hard-coding it forbids: for every `route=`
/// value `TRAILS` admits, `route_class` must return exactly what `TRAILS`
/// returns.
#[test]
fn the_trail_classes_come_from_the_way_sides_own_rules() {
use crate::network::{Rule, TRAILS};
let mut checked = 0;
for rule in TRAILS.rules {
let Rule::KeyIn {
key: "route",
values,
name,
} = rule
else {
continue;
};
for v in *values {
assert_eq!(
&route_class(v),
name,
"route={v} must classify as the way side classifies it"
);
checked += 1;
}
}
assert!(
checked >= 5,
"only {checked} route→class rules found in network::TRAILS — this guard \
would be nearly vacuous; the rule list moved"
);
// The specific pair the two sides had to settle by hand.
assert_eq!(
route_class("piste"),
"ski",
"piste is a ski trail, agreed 2026-08-03"
);
// A route that is NOT a trail keeps its own name rather than being forced
// into the trail namespace — 5 943 Swedish bus routes must not become
// `hiking` because the fallback was a default rather than a passthrough.
assert_eq!(route_class("bus"), "bus");
assert_eq!(route_class("road"), "road");
assert_eq!(route_class(""), "unknown");
}
/// ★ **RED-when-broken: the bridge carries the way side's schema, unchanged,
/// as its prefix — and says which id space it is in.**
///
/// `route_members.id` is a WAY id; `relations.resolved.parquet`'s `id` is a
/// RELATION id. Both are `Int64 not null` in position 0, so a join across them
/// compiles, runs and returns rows — it is the join that is wrong, not the
/// query. The `network` column is the only thing in either file that says so,
/// so it must be non-nullable and it must be set.
#[test]
fn the_bridge_extends_the_way_side_schema_and_names_its_id_space() {
let theirs = crate::network::network_fields();
let mine = route_member_fields();
assert!(
mine.len() > theirs.len(),
"the bridge must ADD to the way side's columns, not replace them"
);
for (i, t) in theirs.iter().enumerate() {
assert_eq!(
(&mine[i]).clone(),
t.clone(),
"column {i} diverged from the way side's network_fields() — the shared \
prefix is derived precisely so it cannot"
);
}
let net = mine
.iter()
.find(|f| f.name() == "network")
.expect("network column");
assert!(
!net.is_nullable(),
"`network` is what tells a reader whether `id` is a way or a relation — a \
NULL there is an unnamed id space"
);
assert_eq!(ROUTE_MEMBERS_NETWORK, "route_members");
// The two columns that make it a bridge and not just another network table.
assert!(mine.iter().any(|f| f.name() == "route_ids"));
assert!(mine.iter().any(|f| f.name() == "roles"));
}
}