osm-katana/src/network.rs
//! **Networks as their own datasets — stored ONCE, not duplicated into every tile.**
//!
//! # The measurement this module exists to answer
//!
//! The `sweden` tile pyramid is **2.7 GB out of a 1.5 GB source** — larger than the data
//! it comes from — because [`crate::tiles`] duplicates a way into every tile its bbox
//! overlaps (`tiles::way_tiles`). A motorway through Sweden lives in hundreds of z12
//! tiles and would live in thousands at z15.
//!
//! MEASURED on `~/.korp/work/sweden/ways.parquet`, **all 32 row groups / 9 204 689 ways,
//! 2026-08-03**:
//!
//! | | ways | share | raw WKB |
//! |---|---|---|---|
//! | `highway=*` | 2 252 829 | 24.5 % | 472.8 MB (22.7 %) |
//! | `railway=*` | 36 321 | 0.39 % | 7.6 MB (0.37 %) |
//! | trail-tagged ways | 75 566 | 0.82 % | 29.2 MB (1.4 %) |
//! | every way | 9 204 689 | 100 % | 2 078.8 MB |
//!
//! So the whole road network is ~473 MB of raw WKB — **less than the z9 pyramid level
//! alone (398 MB) and a sixth of z12 (2 342 MB)** — and stored once it serves every zoom.
//!
//! # Why a network never needs re-cutting for a new zoom
//!
//! **LOD for a network is a filter on CLASS, not a geometric simplification.** At z6 you
//! want motorway and trunk; that is a `WHERE class IN (…)`, not a pyramid. The geometry
//! is the same at every zoom — only the selection changes. Hence:
//!
//! * the `class` column is **broken out** of the tag JSON, so the LOD filter is a cheap
//! dictionary-column read. `serde_json` in korp's `parse_tags` is **55.9 % of all
//! allocations** on a tile read (measured 2026-08-03); a network read must not repeat
//! that to answer "is this a motorway".
//! * the layout is **Hilbert-sorted rows + a GeoParquet 1.1 `covering.bbox` column**,
//! written by [`crate::spatial::pack_batches`] — the same writer, and therefore the
//! same reader seam, korp's bbox pruning already uses. A second read path would be a
//! second thing to keep true.
//!
//! # What rides in every row, and why
//!
//! [`network_fields`] is the ONE declaration; [`NetworkRecipe`] is derived from it, so a
//! column added, removed or retyped moves the recipe token with no edit anywhere else.
//!
//! * **`id`** — non-null. The tiler shipped without it and it cost a day: `MapFacet::way(id)`,
//! label placement and building height all collapsed onto way #0.
//! * **`class`** — non-null, the LOD axis (see above).
//! * **`geometry`** — WKB, **unsimplified**. A network is stored once for every zoom, so
//! there is no zoom to pick a Douglas-Peucker epsilon for.
//! * **`tags`** — the full JSON, for everything class does not answer (name, ref, surface,
//! oneway, bridge, tunnel).
//! * **`node_refs`** — the topology, and it is kept because it MEASURES as load-bearing:
//! the 2 252 829 highway ways have 28 291 212 node refs over 3 385 818 distinct
//! endpoints, of which **811 134 (24.0 %) are shared by two or more highway ways**.
//! That shared set is the junction set — the thing that makes the road network a graph
//! rather than a heap of lines — and it is an exact `i64` join, not a float-coordinate
//! guess. Cost, from the source table's own encoded size (237.1 MB compressed for
//! 123 475 514 refs ⇒ 1.92 B/ref): **≈54 MB for the road network, ~11 % on top of its
//! geometry**. Kept.
#![allow(clippy::float_arithmetic)]
use std::borrow::Cow;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::Context as _;
use arrow::array::{
Array, ArrayRef, BinaryArray, Int64Array, Int64Builder, ListArray, StringArray,
};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
use crate::metadata::{self, BBOX_COLUMN, GeoMeta};
// The ONE spelling of an Arrow type in a recipe token (LAW 5) — see its doc for why the
// private copy that used to live in this file was a correctness hole, not a style one.
use crate::shared::schema::short_type;
use crate::spatial::{PackOptions, PackStats, pack_batches};
/// The parquet key-value-metadata key carrying the [`NetworkRecipe`] in every network
/// table. Deliberately the same shape as [`crate::tiles::TILE_RECIPE_META_KEY`]: one
/// probe pattern for "was this object written by the current writer", whichever asset it
/// is looking at.
pub const NETWORK_RECIPE_META_KEY: &str = "osm-katana:network-recipe";
/// Name of the manifest a network directory carries.
pub const NETWORK_MANIFEST: &str = "networks.json";
// ── the columns ─────────────────────────────────────────────────────────────────
/// **The columns every network table carries**, and the ONE place they are declared
/// (LAW 5): [`build_batch`] builds its arrays from this and [`NetworkRecipe`] describes
/// itself from this, so a schema change moves the recipe automatically. The
/// `covering.bbox` column is appended by [`crate::spatial::pack_batches`] and is named in
/// the recipe separately, because it is that writer's contribution and not this one's.
pub fn network_fields() -> Vec<Field> {
let mut f = vec![
Field::new("id", DataType::Int64, false),
// **Which network this row belongs to, as a COLUMN.** The identity is the file
// name only until a reader merges the three into one layer, which is exactly what
// the map panel's per-network toggles need it to do: one query, and turning rail
// off must not cost a second one. Deriving the answer from `tags` in the client
// would rebuild the read path's worst hotspot — `serde_json` in korp's
// `parse_tags` is 55.9 % of all allocations on a tile read (measured 2026-08-03).
// One distinct value per file, so the dictionary makes it free.
Field::new("network", DataType::Utf8, false),
Field::new("class", DataType::Utf8, false),
// **The first rung this row may draw at** (DELTA 1). Declared by
// `shared::schema::min_zoom_field` and VALUED by `shared::class::min_zoom_of_tags`
// — the one tag → rung decision in the crate, taken from the row's own `tags`
// and therefore identical for a WAY and a NODE carrying the same tags. It lands
// in `network_fields`, so it is in `NetworkRecipe`'s column list automatically
// and a table cut before the column existed reads Stale.
crate::shared::schema::min_zoom_field(),
crate::shared::schema::geometry_field(),
Field::new("tags", DataType::Utf8, true),
Field::new(
"node_refs",
DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
true,
),
];
// **The class block** (row 18) — `class_id` / `kind` / `layer` / `bridge` / `tunnel`
// / `oneway`, declared by `shared::schema::class_fields` and valued by
// `shared::class::class_of_tags`, the same one pass the base tables run. It goes
// LAST for the reason `min_zoom` went last on the base tables: every positional read
// that existed before keeps its index.
//
// `min_zoom` stays where it is, above `geometry`, because moving a live column to
// join its block would be a schema churn that buys nothing — the block is a
// construction of the WRITER (`ClassColumns`), not of the column order. Being in
// `network_fields` at all is what matters: these ride `NetworkRecipe`'s column list
// automatically, so every network table cut before row 18 reads Stale and re-cuts.
f.extend(crate::shared::schema::class_fields());
f
}
fn network_schema(source: Source) -> Arc<Schema> {
let geo = GeoMeta::new(&[source.geometry_type()]).to_json();
let mut meta = HashMap::new();
meta.insert(String::from(metadata::GEO_KEY), geo);
Arc::new(Schema::new_with_metadata(network_fields(), meta))
}
/// Which table of the region extract a network is cut from.
///
/// **One schema for both** — a node layer writes a NULL `node_refs`, not a second column
/// list — so a reader that can read `roads.parquet` can read `rail_nodes.parquet` with no
/// second code path. The asymmetry is real and is exactly the join: a WAY's `node_refs`
/// name the nodes it runs through, and a NODE's own `id` is what those refs point AT.
/// `rail_nodes.id ∈ roads.node_refs ∩ rails.node_refs` is the level crossing.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Source {
Ways,
Nodes,
}
impl Source {
pub fn file(self) -> &'static str {
match self {
Source::Ways => "ways.parquet",
Source::Nodes => "nodes.parquet",
}
}
fn geometry_type(self) -> &'static str {
match self {
Source::Ways => "LineString",
Source::Nodes => "Point",
}
}
fn spell(self) -> &'static str {
match self {
Source::Ways => "ways",
Source::Nodes => "nodes",
}
}
}
// ── the selection, expressed as DATA ────────────────────────────────────────────
/// One admission rule. A way joins the network on the FIRST rule that matches, and that
/// rule decides its `class`.
///
/// The rules are data rather than a `match` arm on purpose: [`NetworkSpec::selector`]
/// renders them into the recipe token, so changing what a network *contains* re-cuts it
/// automatically. A hand-maintained version string next to a hand-written filter is
/// exactly one forgotten edit away from serving the wrong data under the right name.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Rule {
/// In the network iff tag `key` is present; `class` = that tag's own value.
ValueOf { key: &'static str },
/// In the network iff tag `key` holds one of `values`; `class` = `name`.
KeyIn {
key: &'static str,
values: &'static [&'static str],
name: &'static str,
},
/// In the network iff tag `key` is present (any value); `class` = `name`.
HasKey {
key: &'static str,
name: &'static str,
},
}
impl Rule {
/// The class this rule gives `tags`, or `None` when it does not match.
fn class<'a>(&self, tags: &'a str) -> Option<Cow<'a, str>> {
match *self {
Rule::ValueOf { key } => tag_value(tags, key),
Rule::KeyIn { key, values, name } => {
let v = tag_value(tags, key)?;
values.contains(&v.as_ref()).then(|| Cow::Borrowed(name))
}
Rule::HasKey { key, name } => tag_value(tags, key).map(|_| Cow::Borrowed(name)),
}
}
/// The rule's canonical spelling for the recipe token.
fn spell(&self) -> String {
match *self {
Rule::ValueOf { key } => format!("{key}=*"),
Rule::KeyIn { key, values, name } => format!("{key}∈[{}]>{name}", values.join("|")),
Rule::HasKey { key, name } => format!("{key}?>{name}"),
}
}
}
/// **Classes that mean "there is no such thing on the ground".** Removed, dismantled,
/// planned or not-yet-built. Refused at WRITE time for every network, from one list
/// (LAW 5), and named in the recipe token so flipping the decision re-cuts the data
/// instead of silently serving the old choice.
///
/// The choice, made with the numbers on the table (measured 2026-08-03): on `railway=*`
/// this drops **6 754 of 36 321 ways (18.6 %)** — `abandoned` 3 908, `disused` 1 662,
/// `razed` 547, `construction` 506, `proposed` 74, `dismantled` 53, `historic_path` 3,
/// `historic` 1. Torn-up track is not a railway network. On `highway=*` the same list
/// drops **1 706 of 2 252 829 (0.076 %)** — `construction` 1 341, `proposed` 353,
/// `planned` 7, `abandoned` 3, `razed` 2. Physically-present rail infrastructure that is
/// simply not running track (`platform`, `turntable`, `roundhouse`, `signal_box`) is
/// KEPT: it exists, it renders, and `class` lets a reader drop it for free.
pub const NOT_ON_THE_GROUND: &[&str] = &[
"abandoned",
"construction",
"demolished",
"dismantled",
"disused",
"historic",
"historic_path",
"planned",
"proposed",
"razed",
"removed",
];
/// A network: a name, an ordered rule list, and the refusal list.
#[derive(Clone, Copy, Debug)]
pub struct NetworkSpec {
/// File stem and manifest key (`roads` ⇒ `roads.parquet`).
pub name: &'static str,
/// Which extract table the rows come from.
pub source: Source,
pub rules: &'static [Rule],
pub refuse: &'static [&'static str],
}
impl NetworkSpec {
/// The class of `tags` in this network, or `None` when the way is not in it (either
/// no rule matched, or the matched class is refused).
pub fn class<'a>(&self, tags: &'a str) -> Option<Cow<'a, str>> {
let c = self.rules.iter().find_map(|r| r.class(tags))?;
if c.is_empty() || self.refuse.contains(&c.as_ref()) {
return None;
}
Some(c)
}
/// The canonical spelling of this network's selection, for the recipe token.
pub fn selector(&self) -> String {
format!(
"{}:{};!{}",
self.source.spell(),
self.rules
.iter()
.map(Rule::spell)
.collect::<Vec<_>>()
.join(","),
self.refuse.join("|")
)
}
}
/// `highway=*` — the road network. 2 252 829 ways / 472.8 MB WKB (measured 2026-08-03).
pub const ROADS: NetworkSpec = NetworkSpec {
source: Source::Ways,
name: "roads",
rules: &[Rule::ValueOf { key: "highway" }],
refuse: NOT_ON_THE_GROUND,
};
/// `railway=*` — the rail network. 36 321 ways / 7.6 MB WKB before refusal (measured
/// 2026-08-03); see [`NOT_ON_THE_GROUND`] for what the refusal costs and why.
pub const RAILS: NetworkSpec = NetworkSpec {
source: Source::Ways,
name: "rails",
rules: &[Rule::ValueOf { key: "railway" }],
refuse: NOT_ON_THE_GROUND,
};
/// **Trails — the part that exists as a WAY.**
///
/// A trail is usually a *relation* (`route=hiking|foot|bicycle|mtb|ski`) binding together
/// ways that are individually just `highway=path`. This spec deliberately covers only
/// what a way says about ITSELF; the relation half is a separate dataset, built by
/// resolving relation members, and the two are meant to be unioned on `id`.
///
/// MEASURED 2026-08-03 on `ways.parquet`, all 32 row groups: **75 566 ways (0.82 %)
/// carry a trail tag of their own** — `mtb:scale` 17 716, `trail_visibility` 17 590,
/// `snowmobile` 13 640, `piste:type` 9 456, `sac_scale` 9 233, `ski` 2 501 — while only
/// **62 ways** carry a `network=lwn|rwn|nwn|lcn|rcn` of their own and only **1 387**
/// carry a `route=` of their own. The rest of the trail network is relation-only.
///
/// Classes are kept in ONE namespace (`hiking` / `cycling` / `mtb` / `ski` /
/// `snowmobile`) rather than echoing each source key, so a reader's LOD filter is a
/// single `class IN (…)` no matter which tag admitted the way. The source detail stays
/// in `tags`.
pub const TRAILS: NetworkSpec = NetworkSpec {
source: Source::Ways,
name: "trails",
rules: &[
Rule::KeyIn {
key: "route",
values: &["hiking", "foot"],
name: "hiking",
},
Rule::KeyIn {
key: "route",
values: &["bicycle"],
name: "cycling",
},
Rule::KeyIn {
key: "route",
values: &["mtb"],
name: "mtb",
},
Rule::KeyIn {
key: "route",
values: &["ski", "piste"],
name: "ski",
},
Rule::KeyIn {
key: "network",
values: &["lwn", "rwn", "nwn", "iwn"],
name: "hiking",
},
Rule::KeyIn {
key: "network",
values: &["lcn", "rcn", "ncn", "icn"],
name: "cycling",
},
Rule::HasKey {
key: "sac_scale",
name: "hiking",
},
Rule::HasKey {
key: "trail_visibility",
name: "hiking",
},
Rule::HasKey {
key: "mtb:scale",
name: "mtb",
},
Rule::HasKey {
key: "piste:type",
name: "ski",
},
Rule::KeyIn {
key: "ski",
values: &["yes", "designated", "permissive"],
name: "ski",
},
Rule::KeyIn {
key: "snowmobile",
values: &["yes", "designated", "permissive"],
name: "snowmobile",
},
],
refuse: NOT_ON_THE_GROUND,
};
/// **`railway=*` on a NODE — the rail network's infrastructure, and the joint between
/// the road network and the rail one.**
///
/// MEASURED on `~/.korp/work/sweden/nodes.parquet`, all 214 row groups / 102 518 979
/// nodes (2026-08-03): `signal` 18 820, `milestone` 16 168, **`level_crossing` 9 148**,
/// `switch` 8 544, `crossing` 3 659, `buffer_stop` 2 818, `stop` 1 208, `station` 666,
/// `tram_stop` 531, `railway_crossing` 319.
///
/// # Why this is not just "points to draw"
///
/// A `railway=level_crossing` is **one node shared by a `highway` way and a `railway`
/// way**. It is a topological joint between the two networks, and it survives only
/// because both networks keep `node_refs` and this layer keeps the node's own `id`:
///
/// ```text
/// level crossings = rail_nodes[class="level_crossing"].id
/// ∩ flatten(roads.node_refs) ∩ flatten(rails.node_refs)
/// ```
///
/// Flatten everything to geometry and that question stops being askable — which is the
/// whole reason the ids are here. `railway=crossing` is the pedestrian-only variant and
/// `railway=railway_crossing` is rail-on-rail; all three are kept as distinct classes
/// rather than merged, because they are different joints.
pub const RAIL_NODES: NetworkSpec = NetworkSpec {
source: Source::Nodes,
name: "rail_nodes",
rules: &[Rule::ValueOf { key: "railway" }],
refuse: NOT_ON_THE_GROUND,
};
/// The networks this module cuts, in manifest order.
pub const ALL_NETWORKS: &[NetworkSpec] = &[ROADS, RAILS, TRAILS, RAIL_NODES];
/// **Is this way already carried by a network table, and therefore NOT the tiler's to
/// duplicate?**
///
/// The ONE place the tiler and the extractor agree about what "a road" is (LAW 5). A
/// `tiles.rs` that spelled the predicate out again would be a second copy of a filter
/// whose disagreement is invisible in code and total in data: a class the tiler dropped
/// but no network claimed would simply vanish from the map, and nothing would say so.
///
/// **Refused classes are deliberately NOT excluded here.** `highway=construction` is
/// refused by [`NOT_ON_THE_GROUND`], so no network table holds it — which makes the tile
/// pyramid its only home, and dropping it from there too would delete it from the map
/// entirely. The predicate answers "is it in a network", not "does it match a rule".
pub fn carried_by_a_network(source: Source, tags: &str) -> bool {
TILED_OUT
.iter()
.any(|n| n.source == source && n.class(tags).is_some())
}
/// The networks the tiler hands over to a network table instead of duplicating.
///
/// **Ways only, and the source match in [`carried_by_a_network`] is load-bearing.**
/// MEASURED on the `sweden` extract 2026-08-03: a source-blind predicate — one that asks
/// "do these tags match ROADS or RAILS" without asking what the row IS — dropped
/// **567 179 nodes** from the pyramid (102 518 979 → 101 951 800). `railway=*` and
/// `highway=*` are common NODE tags too (`highway=crossing`, `highway=bus_stop`,
/// `railway=signal`), and [`ROADS`]/[`RAILS`] read `ways.parquet`, so nothing would have
/// carried them: **~503 371 points would have vanished from the map with nothing saying
/// so.** Caught by reading the scan counts of a real run, not by a test.
///
/// [`RAIL_NODES`] is deliberately NOT in this list. It is an ADDITIONAL layer for the
/// topology join, not a replacement for drawing those points, and the pyramid keeps them.
///
/// **[`TRAILS`] is in it, and was not until `granska-5`.** A trail that says
/// `highway=path` about itself was already handed over by [`ROADS`]; a trail whose only
/// own tags are `snowmobile=yes`, `piste:type=*` or `route=hiking` matches neither
/// `highway` nor `railway` and so stayed in the pyramid **while `trails.parquet` carried
/// it too**. MEASURED on the real swedish `trails.parquet` (75 460 rows, 2026-08-04):
/// **24 895 rows = 32.99 %** were in both — `snowmobile` 19 834, `piste:type` 3 850,
/// `piste:type`+`route` 670, `piste:type`+`snowmobile` 320. korp draws both copies
/// (`map_networks.rs`: the pyramid as `Network::BASE`, the table as `Network::Trails`)
/// and `Network::tiers` LODs only one of them, so the duplicate is the one that never
/// thins out. The token was the worse half: [`tiled_out_selector`] stamps `excl=` into
/// every tile and it SAID the trails were handed over.
pub const TILED_OUT: &[NetworkSpec] = &[ROADS, RAILS, TRAILS];
/// The canonical spelling of [`carried_by_a_network`], for a tile recipe.
///
/// **Derived from the same specs the predicate consults**, so adding a network to
/// [`TILED_OUT`] moves every tile recipe with it and every existing pyramid reads as
/// STALE — which is the only thing that stops a server concluding "already tiled" over a
/// pyramid that still carries roads.
pub fn tiled_out_selector() -> String {
TILED_OUT
.iter()
.map(|n| n.selector())
.collect::<Vec<_>>()
.join("&")
}
// ── the recipe ──────────────────────────────────────────────────────────────────
/// **Everything about a network cut that a reader must re-cut for when it changes.**
///
/// Same defect, same shape, same reasoning as [`crate::tiles::TileRecipe`] — see its doc
/// for the incident (`collauto-6`: a manifest whose `format` string could not tell a
/// current pyramid from a two-schema-old one). Derived from [`network_fields`] and from
/// [`NetworkSpec::selector`], so **both** "what columns are in here" and "what rows are in
/// here" move the token with no edit in this struct.
///
/// Readable descriptor, not a hash: it lands in a manifest an operator greps, and it must
/// not be a `DefaultHasher` digest, which is stable only within one toolchain.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NetworkRecipe {
/// The network's name (`roads` / `rails` / `trails`).
pub network: String,
/// `name:Type` per column, in writer order, including the packer's `bbox`.
pub columns: Vec<String>,
/// [`NetworkSpec::selector`] — which rows are in.
pub selector: String,
/// Rows per row group — the pruning granularity, so re-tuning it re-cuts.
pub row_group_rows: usize,
/// **The `min_zoom` ladder that VALUED the column** —
/// [`crate::shared::class::rules_spelling`] — or `None` for a table that has no
/// `min_zoom` column to value.
///
/// The column list above already moves when the column is added or removed, but it
/// cannot see a rule EDIT: moving `place=city` from District to Country changes every
/// `min_zoom` cell in the file and not one column name. Without this term a re-tiered
/// ladder would serve the old rungs under a Current verdict — the same shape as the
/// stale-recipe defect [`crate::tiles::TileRecipe`]'s `geoms` term exists for.
///
/// `None` is not a shortcut. `graph_edges` / `graph_nodes` carry no `min_zoom`
/// (their rows are chosen by the network selectors and valued by the junction rule),
/// and until 2026-09-01 this field was stamped into their tokens anyway, so every
/// ladder edit re-cut 8.4 M graph rows for a change that could not appear in them.
/// An axis belongs in a token only when the table CARRIES it; `for_table` decides
/// that from the field list, the same place the column axis is decided.
pub min_zoom_ladder: Option<String>,
}
impl NetworkRecipe {
/// The recipe [`extract_networks`] implements right now for `spec`.
pub fn current(spec: &NetworkSpec, row_group_rows: usize) -> Self {
Self::for_table(
spec.name,
&network_fields(),
&spec.selector(),
row_group_rows,
&crate::spatial::generalised_rungs_for(&network_schema(spec.source), true),
)
}
/// **The recipe of ANY table in the network family**, from its own field list and its
/// own row selection.
///
/// [`Self::current`] and [`crate::graph`]'s two tables route through this one
/// constructor (LAW 5), so the `bbox` column the packer contributes, the token layout
/// and the staleness probe cannot drift between a network table and a graph table —
/// which is the whole reason both stamp the same [`NETWORK_RECIPE_META_KEY`].
/// `rungs` is what [`crate::spatial::generalised_rungs_for`] says the packer will
/// write for this table — `[]` for a point table like `rail_nodes` and for the graph
/// tables, which opt out. It comes from the caller rather than being decided here so
/// there is ONE decision (LAW 5): a token that claimed `geometry_z12` on a table the
/// packer gave none would judge every cut table stale, forever.
pub fn for_table(
name: &str,
fields: &[Field],
selector: &str,
row_group_rows: usize,
rungs: &[u8],
) -> Self {
let mut columns: Vec<String> = fields
.iter()
.map(|f| format!("{}:{}", f.name(), short_type(f.data_type())))
.collect();
// Appended by `spatial::pack_batches`, so it is in the object and must be in the
// token — but it is spelled here, where the rest of the column list is, rather
// than being silently absent because a different module contributes it. Spelled
// from the packer's OWN field (not a literal `Struct`), so a retype of a leaf is
// a moved token: `short_type` recurses into the children since 2026-09-01.
let bbox = crate::spatial::bbox_field();
columns.push(format!("{}:{}", bbox.name(), short_type(bbox.data_type())));
// The ladder axis is carried only by a table that carries the column it values.
let mz = crate::shared::schema::min_zoom_field();
// Row 18: the KINDS ride the same term, appended after the rules. `class_id` is
// an index into the rule table and `kind` is a column on disk, so a rule whose
// test and rung are unchanged but whose kind moved must still re-cut — and
// `rules_spelling` alone cannot see that. It is appended rather than folded into
// `rules_spelling` because that string is facett's checked-in copy (see
// `class::rules_spelling`).
let min_zoom_ladder = fields.iter().any(|f| f.name() == mz.name()).then(|| {
format!(
"{}|kinds={}",
crate::shared::class::rules_spelling(),
crate::shared::class::kinds_spelling()
)
});
// Same argument, for DELTA 2's generalisation columns. **Adding them MOVES this
// token**, so every cut network table reads as Stale and is re-cut — measured at
// 21.3 s for 10.8 M rows, which is why it is written down rather than avoided.
for &z in rungs {
columns.push(format!(
"{}:Binary",
crate::shared::schema::generalised_geometry_name(z)
));
}
Self {
network: String::from(name),
columns,
selector: String::from(selector),
row_group_rows,
min_zoom_ladder,
}
}
/// The stamped form. The `minzoom=` term is present exactly when the table carries
/// a `min_zoom` column; a table without one has no term rather than an empty one, so
/// the token says what the file holds and nothing about what it does not.
pub fn token(&self) -> String {
let mut t = format!(
"v1;net={};cols={};sel={};rg={}",
self.network,
self.columns.join(","),
self.selector,
self.row_group_rows,
);
if let Some(ladder) = &self.min_zoom_ladder {
t.push_str(";minzoom=");
t.push_str(ladder);
}
t
}
/// The recipe a network parquet's footer declares, or `None` when it declares none.
/// Reads the footer only; no row group is decoded.
pub fn of_table(path: &Path) -> anyhow::Result<Option<String>> {
Self::of_table_bytes(bytes::Bytes::from(std::fs::read(path)?))
}
/// [`Self::of_table`] for a table that was **fetched, not opened** — the object-store
/// shape. One footer lookup for both (LAW 5), so disk and store cannot disagree about
/// what a table claims.
pub fn of_table_bytes(b: bytes::Bytes) -> anyhow::Result<Option<String>> {
let r = ParquetRecordBatchReaderBuilder::try_new(b)?;
Ok(r.metadata()
.file_metadata()
.key_value_metadata()
.and_then(|kv| kv.iter().find(|k| k.key == NETWORK_RECIPE_META_KEY))
.and_then(|k| k.value.clone())
.filter(|v| !v.is_empty()))
}
}
// The spelling of an Arrow type in a recipe token USED TO LIVE HERE, as a private copy
// of `tiles::short_type` kept in step by a doc comment saying the two must agree. They
// had already drifted — this copy had an `Int8` arm for the graph's `oneway` and that one
// did not — and both still ended in the `Other` catch-all that could not tell
// `min_zoom: UInt8` from `levels: UInt8` from a retype of either. One writer now, in
// `shared::schema::short_type`, whose doc carries the reasoning and the measured cost.
// ── the flat-tag scanner ────────────────────────────────────────────────────────
/// **The value of `key` in a `tags` JSON object, without a JSON parser.**
///
/// `crate::tags::serialize_tags` emits a flat, BTreeMap-sorted `{"k":"v",…}` of strings
/// only, so a full `serde_json::Value` parse — which allocates a `Map` and a `String` per
/// pair — is pure waste when the question is "does this way have a `highway` tag". korp's
/// `parse_tags` doing exactly that is **55.9 % of all allocations** on a tile read
/// (measured 2026-08-03); this is the extraction side refusing to inherit it.
///
/// Scans at OBJECT level — key, `:`, value, `,`/`}` — so a *value* that happens to read
/// `"highway"` can never be mistaken for the key. Returns a borrowed slice when the value
/// has no escape (the overwhelming case: class values are bare words) and only allocates
/// to unescape when it must.
pub fn tag_value<'a>(json: &'a str, key: &str) -> Option<Cow<'a, str>> {
let b = json.as_bytes();
let mut i = 0usize;
// opening brace
while i < b.len() && b[i].is_ascii_whitespace() {
i += 1;
}
if i >= b.len() || b[i] != b'{' {
return None;
}
i += 1;
loop {
while i < b.len() && (b[i].is_ascii_whitespace() || b[i] == b',') {
i += 1;
}
if i >= b.len() || b[i] == b'}' {
return None;
}
if b[i] != b'"' {
return None; // not the flat object we write
}
let (k, next) = scan_string(b, i)?;
i = next;
while i < b.len() && b[i].is_ascii_whitespace() {
i += 1;
}
if i >= b.len() || b[i] != b':' {
return None;
}
i += 1;
while i < b.len() && b[i].is_ascii_whitespace() {
i += 1;
}
if i >= b.len() {
return None;
}
if b[i] != b'"' {
// A non-string value: skip it conservatively to the next top-level `,`/`}`.
let start = i;
let mut depth = 0i32;
while i < b.len() {
match b[i] {
b'[' | b'{' => depth += 1,
b']' | b'}' if depth > 0 => depth -= 1,
b'}' if depth == 0 => break,
b',' if depth == 0 => break,
_ => {}
}
i += 1;
}
debug_assert!(i > start);
continue;
}
let (v, next) = scan_string(b, i)?;
i = next;
let matched = json.get(k.0..k.1).is_some_and(|s| s == key);
if matched {
let raw = json.get(v.0..v.1)?;
return Some(if raw.as_bytes().contains(&b'\\') {
Cow::Owned(unescape(raw))
} else {
Cow::Borrowed(raw)
});
}
}
}
/// Scan the JSON string starting at `b[at] == '"'`. Returns the byte range of its
/// CONTENTS (escapes still in) and the index just past the closing quote.
fn scan_string(b: &[u8], at: usize) -> Option<((usize, usize), usize)> {
let mut i = at + 1;
let start = i;
while i < b.len() {
match b[i] {
b'\\' => i += 2,
b'"' => return Some(((start, i), i + 1)),
_ => i += 1,
}
}
None
}
/// Minimal JSON string unescape — reached only for values that contain a backslash.
fn unescape(s: &str) -> String {
let mut out = String::with_capacity(s.len());
let mut it = s.chars();
while let Some(c) = it.next() {
if c != '\\' {
out.push(c);
continue;
}
match it.next() {
Some('n') => out.push('\n'),
Some('t') => out.push('\t'),
Some('r') => out.push('\r'),
Some('b') => out.push('\u{8}'),
Some('f') => out.push('\u{c}'),
Some('u') => {
let hex: String = it.by_ref().take(4).collect();
match u32::from_str_radix(&hex, 16).ok().and_then(char::from_u32) {
Some(ch) => out.push(ch),
None => out.push('\u{fffd}'),
}
}
Some(other) => out.push(other),
None => {}
}
}
out
}
// ── extraction ──────────────────────────────────────────────────────────────────
/// Knobs for [`extract_networks`].
#[derive(Clone, Debug)]
pub struct NetworkOptions {
pub networks: Vec<NetworkSpec>,
pub compression: String,
/// Rows per row group — the bbox-pruning granularity. Defaults to
/// [`crate::spatial::PACK_ROW_GROUP_ROWS`], which carries the measurement. A
/// literal here would be a second writer of a tuning axis (LAW 5).
pub row_group_rows: usize,
/// Refuse to hold more than this many selected rows in memory across all networks.
/// A HARD stop, not a warning: an `ldeflate` test binary reached **474.9 GiB RSS on
/// this box on 2026-08-03** and took it down in a global OOM. A country's road
/// network is ~2.3 M rows, so the default leaves 4× headroom and still cannot eat a
/// planet by accident.
pub max_rows: usize,
/// Also cut the ADJACENCY — `graph_edges.parquet` + `graph_nodes.parquet` — out of the
/// same rows, in the same pass, through the same packer (LAW 5). See [`crate::graph`].
/// On by default: a network that cannot answer "what meets here" is a table of lines.
pub graph: bool,
}
impl Default for NetworkOptions {
fn default() -> Self {
Self {
networks: ALL_NETWORKS.to_vec(),
compression: String::from("zstd"),
row_group_rows: crate::spatial::PACK_ROW_GROUP_ROWS,
max_rows: 40_000_000,
graph: true,
}
}
}
/// What one network cut produced.
#[derive(Clone, Debug)]
pub struct NetworkStats {
pub name: String,
pub path: PathBuf,
pub rows: usize,
/// Rows a rule matched but [`NetworkSpec::refuse`] then dropped — reported, never
/// silent: a refusal that nobody counts is indistinguishable from a filter that
/// never fired.
pub refused: usize,
pub rows_no_geometry: usize,
pub row_groups: usize,
pub bytes: u64,
/// `class` → row count, so the LOD filter can be sized without opening the file.
pub classes: Vec<(String, usize)>,
pub recipe: String,
/// Set on the `graph_edges` entry only: everything the adjacency build MEASURED — how
/// many nodes the ways touch, how many of those are junctions, how many level
/// crossings are real joints, how many ways could not be sliced. Reported rather than
/// inferred, because "the table has rows" is not evidence that the graph is a graph.
pub graph: Option<crate::graph::GraphStats>,
}
/// One selected row, held until the whole scan is done (the Hilbert sort is global).
struct Rows {
id: Vec<i64>,
class: Vec<String>,
geom: Vec<Option<Vec<u8>>>,
tags: Vec<String>,
refs: Vec<Vec<i64>>,
refused: usize,
}
impl Rows {
fn new() -> Self {
Self {
id: Vec::new(),
class: Vec::new(),
geom: Vec::new(),
tags: Vec::new(),
refs: Vec::new(),
refused: 0,
}
}
fn len(&self) -> usize {
self.id.len()
}
fn absorb(&mut self, o: Rows) {
self.id.extend(o.id);
self.class.extend(o.class);
self.geom.extend(o.geom);
self.tags.extend(o.tags);
self.refs.extend(o.refs);
self.refused += o.refused;
}
}
/// **Cut every network in `opts` out of `<extract>/ways.parquet` in ONE pass.**
///
/// One reader, one classification of each way against every spec, three outputs — rather
/// than three near-identical extractors reading the same 852 MB three times (LAW 5). The
/// scan fans out on gatling, one unit per row group, LPT-scheduled heaviest-first by
/// compressed size, exactly as [`crate::verify`] does (LAW 3 — gatling, never rayon).
///
/// Rows are collected in row-group order and written through
/// [`crate::spatial::pack_batches`], so the layout — Hilbert order, outward-rounded
/// `covering.bbox`, `BYTE_STREAM_SPLIT` on the bbox leaves — is the one korp's pruning
/// already reads, with no second implementation to keep in step.
pub fn extract_networks(
extract: &Path,
out_dir: &Path,
opts: &NetworkOptions,
) -> anyhow::Result<Vec<NetworkStats>> {
let specs = opts.networks.clone();
// One pass per SOURCE table, and every network cut from that table is classified in
// that one pass — never one read of ways.parquet per network (LAW 5, and 852 MB).
let mut sources: Vec<Source> = Vec::new();
for sp in &specs {
if !sources.contains(&sp.source) {
sources.push(sp.source);
}
}
let mut acc: Vec<Rows> = specs.iter().map(|_| Rows::new()).collect();
for source in sources {
let table = extract.join(source.file());
anyhow::ensure!(table.is_file(), "no {} under {extract:?}", source.file());
let here: Vec<usize> = specs
.iter()
.enumerate()
.filter(|(_, sp)| sp.source == source)
.map(|(i, _)| i)
.collect();
let here_specs: Vec<NetworkSpec> = here.iter().map(|&i| specs[i]).collect();
let meta = {
let f = std::fs::File::open(&table).with_context(|| format!("open {table:?}"))?;
ArrowReaderMetadata::load(&f, ArrowReaderOptions::new())?
};
let pq_meta = meta.metadata().clone();
let n_rg = pq_meta.num_row_groups();
// One unit per row group, LPT-scheduled heaviest-first. Each opens its own reader
// over its single group, projected to only the columns a network needs.
let per_group = gatling::gatling_forkjoin::gatling_map_balanced(
&(0..n_rg).collect::<Vec<usize>>(),
0,
1,
|rg: &usize| pq_meta.row_group(*rg).compressed_size().max(0) as u64,
|_i, rg: &usize| scan_row_group(&table, &meta, *rg, &here_specs),
);
for r in per_group {
let groups = r?;
for (&si, g) in here.iter().zip(groups) {
acc[si].absorb(g);
}
}
// Charge the ceiling after EVERY source, not only at the end: nodes.parquet is
// 102 518 979 rows and a mis-specified rule there must stop before it has eaten
// the box, not after.
let held: usize = acc.iter().map(Rows::len).sum();
anyhow::ensure!(
held <= opts.max_rows,
"network extraction selected {held} rows, over the {} row ceiling — refusing \
to build the batches (raise NetworkOptions::max_rows deliberately)",
opts.max_rows
);
}
std::fs::create_dir_all(out_dir)?;
// ── the adjacency, from the rows this pass is already holding ────────────────
// Built BEFORE `acc` is consumed by the writers below, so there is no second read of
// ways.parquet and no second classification (LAW 5).
let graph_out: Vec<NetworkStats> = if opts.graph {
build_and_write_graph(&specs, &acc, out_dir, opts)?
} else {
Vec::new()
};
let mut out = Vec::with_capacity(specs.len());
for (spec, rows) in specs.iter().zip(acc) {
let schema = network_schema(spec.source);
let recipe = NetworkRecipe::current(spec, opts.row_group_rows);
let dst = out_dir.join(format!("{}.parquet", spec.name));
let classes = class_histogram(&rows.class);
let refused = rows.refused;
let batches = build_batches(&schema, spec.name, rows)?;
let pack = PackOptions {
compression: opts.compression.clone(),
row_group_rows: opts.row_group_rows,
covering: true,
generalise: true,
// The network tables KEEP their node_refs: `rail_nodes.id in roads.node_refs
// intersect rails.node_refs` is the level crossing, and korp asks it.
node_refs_sidecar: false,
};
let stats: PackStats = pack_batches(
batches,
&schema,
&dst,
&pack,
&[(String::from(NETWORK_RECIPE_META_KEY), recipe.token())],
)?;
out.push(NetworkStats {
name: String::from(spec.name),
path: dst,
rows: stats.rows,
refused,
rows_no_geometry: stats.rows_no_geometry,
row_groups: stats.row_groups,
bytes: stats.bytes,
classes,
recipe: recipe.token(),
graph: None,
});
}
out.extend(graph_out);
std::fs::write(out_dir.join(NETWORK_MANIFEST), manifest_json(&out))?;
Ok(out)
}
/// **Cut the adjacency out of the rows the network pass already holds, and write it
/// through the same packer.**
///
/// The ONE place `graph_edges` / `graph_nodes` are produced (LAW 5): they are a view of
/// the same selection, so a second extractor reading `ways.parquet` again could disagree
/// with the network tables about which ways are in — and nothing would say so.
///
/// The graph is cut from [`crate::graph::GRAPHED_NETWORKS`] only, and the recipe names
/// which of them were actually present: a graph cut from `--only roads` must NOT read as
/// the same object as one cut from roads and rails, because the level-crossing joints are
/// exactly what the second one has and the first cannot.
fn build_and_write_graph(
specs: &[NetworkSpec],
acc: &[Rows],
out_dir: &Path,
opts: &NetworkOptions,
) -> anyhow::Result<Vec<NetworkStats>> {
use crate::graph;
let mut ways: Vec<graph::WayIn<'_>> = Vec::new();
let mut nodes: Vec<graph::NodeIn<'_>> = Vec::new();
let mut fed: Vec<&str> = Vec::new();
for (spec, rows) in specs.iter().zip(acc) {
if graph::GRAPHED_NETWORKS.contains(&spec.name) {
fed.push(spec.name);
for i in 0..rows.len() {
ways.push(graph::WayIn {
id: rows.id[i],
network: spec.name,
class: &rows.class[i],
tags: &rows.tags[i],
geometry: rows.geom[i].as_deref(),
node_refs: &rows.refs[i],
});
}
} else if spec.name == graph::GRAPH_NODE_SOURCE {
fed.push(spec.name);
for i in 0..rows.len() {
nodes.push(graph::NodeIn {
id: rows.id[i],
class: &rows.class[i],
tags: &rows.tags[i],
});
}
}
}
if ways.is_empty() {
return Ok(Vec::new());
}
let built = graph::build_graph(&ways, &nodes)?;
let selector = format!("from:{};{}", fed.join("+"), graph::junction_rule());
let pack = PackOptions {
compression: opts.compression.clone(),
row_group_rows: opts.row_group_rows,
covering: true,
// **No generalisation columns on the GRAPH tables.** `graph_edges` /
// `graph_nodes` answer routing questions (`--question nearest-crossing`), and a
// router reads an edge's geometry for its length and its endpoints. A version of
// that geometry with vertices removed is a different length, so a coarse column
// here would be an invitation to compute a wrong answer quickly. Nothing draws
// these tables; the display tables beside them carry the rungs.
generalise: false,
node_refs_sidecar: false,
};
let mut out = Vec::with_capacity(2);
for (name, fields, batches, classes) in [
(
graph::GRAPH_EDGES,
graph::edge_fields(),
built.edges,
built.stats.edge_classes.clone(),
),
(
graph::GRAPH_NODES,
graph::node_fields(),
built.nodes,
built.stats.node_kinds.clone(),
),
] {
let schema = if name == graph::GRAPH_EDGES {
graph::edge_schema()
} else {
graph::node_schema()
};
// `pack.generalise` is false for the graph tables (see the `PackOptions` above),
// so this list is empty and the token does not claim columns the file lacks.
let recipe = NetworkRecipe::for_table(
name,
&fields,
&selector,
opts.row_group_rows,
&crate::spatial::generalised_rungs_for(&schema, pack.generalise),
);
let dst = out_dir.join(format!("{name}.parquet"));
let stats = pack_batches(
batches,
&schema,
&dst,
&pack,
&[(String::from(NETWORK_RECIPE_META_KEY), recipe.token())],
)?;
out.push(NetworkStats {
name: String::from(name),
path: dst,
rows: stats.rows,
refused: 0,
rows_no_geometry: stats.rows_no_geometry,
row_groups: stats.row_groups,
bytes: stats.bytes,
classes,
recipe: recipe.token(),
graph: None,
});
}
out[0].graph = Some(built.stats);
Ok(out)
}
/// Classify one row group's ways against every spec. Returns one [`Rows`] per spec, in
/// spec order.
fn scan_row_group(
path: &Path,
meta: &ArrowReaderMetadata,
rg: usize,
specs: &[NetworkSpec],
) -> anyhow::Result<Vec<Rows>> {
let f = std::fs::File::open(path)?;
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(f, meta.clone());
// `node_refs` exists on ways.parquet and not on nodes.parquet; naming a column the
// file does not have is an error, not an empty projection, so ask the schema.
let has_refs = (0..builder.parquet_schema().num_columns()).any(|i| {
builder
.parquet_schema()
.column(i)
.path()
.string()
.starts_with("node_refs")
});
let mut want: Vec<&str> = vec!["id", "geometry", "tags"];
if has_refs {
want.push("node_refs");
}
let mask = ProjectionMask::columns(builder.parquet_schema(), want);
let reader = builder
.with_row_groups(vec![rg])
.with_projection(mask)
.build()?;
let mut out: Vec<Rows> = specs.iter().map(|_| Rows::new()).collect();
for batch in reader {
let b = batch?;
let ids = col::<Int64Array>(&b, "id")?;
let geoms = col::<BinaryArray>(&b, "geometry")?;
let tags = col::<StringArray>(&b, "tags")?;
let refs = b
.column_by_name("node_refs")
.and_then(|c| c.as_any().downcast_ref::<ListArray>());
for r in 0..b.num_rows() {
let t = if tags.is_null(r) { "" } else { tags.value(r) };
if t.is_empty() {
continue;
}
for (si, spec) in specs.iter().enumerate() {
// The refusal is counted separately from the miss: `class` folds them
// together (both are `None`) so ask the rules directly for the count.
let matched = spec.rules.iter().find_map(|rl| rl.class(t));
let Some(c) = matched else { continue };
if c.is_empty() {
continue;
}
if spec.refuse.contains(&c.as_ref()) {
out[si].refused += 1;
continue;
}
let dst = &mut out[si];
dst.id.push(ids.value(r));
dst.class.push(c.into_owned());
dst.geom
.push((!geoms.is_null(r)).then(|| geoms.value(r).to_vec()));
dst.tags.push(String::from(t));
dst.refs.push(match refs {
Some(l) if l.is_valid(r) => l
.value(r)
.as_any()
.downcast_ref::<Int64Array>()
.map(|a| a.iter().flatten().collect())
.unwrap_or_default(),
_ => Vec::new(),
});
}
}
}
Ok(out)
}
fn col<'a, T: 'static>(b: &'a RecordBatch, name: &str) -> anyhow::Result<&'a T> {
b.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<T>())
.ok_or_else(|| anyhow::anyhow!("ways.parquet column `{name}` missing or wrong type"))
}
/// Build the network record batches. Chunked so no single Arrow array has to hold a
/// whole country's geometry contiguously.
fn build_batches(
schema: &Arc<Schema>,
network: &str,
rows: Rows,
) -> anyhow::Result<Vec<RecordBatch>> {
const CHUNK: usize = 65_536;
let n = rows.len();
let Rows {
id,
class,
geom,
tags,
refs,
..
} = rows;
let mut out = Vec::with_capacity(n.div_ceil(CHUNK).max(1));
let mut at = 0usize;
while at < n {
let end = (at + CHUNK).min(n);
let ids = Int64Array::from(id[at..end].to_vec());
let net = StringArray::from_iter_values((at..end).map(|_| network));
let cls = StringArray::from_iter_values(class[at..end].iter().map(String::as_str));
let gm = BinaryArray::from_opt_vec(
geom[at..end]
.iter()
.map(|g| g.as_deref())
.collect::<Vec<_>>(),
);
let tg = StringArray::from_iter_values(tags[at..end].iter().map(String::as_str));
// DELTA 1. One call, one map, both geometries: the rung is a function of the
// row's `tags` and of nothing else — not of the network it landed in, not of
// whether the row came from `ways` or `nodes`. That is the construction that
// makes "a way and a point with the same tags land on the same rung" true rather
// than watched (`korp-map-pt-1`, 2026-08-31).
let mz = arrow::array::UInt8Array::from_iter_values(
tags[at..end]
.iter()
.map(|t| crate::shared::class::min_zoom_of_tags(t)),
);
// A NODE row has no refs of its own — its `id` is what ways refer TO — so its
// cell is NULL, not an empty list. Null and empty are different claims: empty
// says "this way passes through no nodes", which is never true.
let mut lb = arrow::array::ListBuilder::new(Int64Builder::new());
for r in &refs[at..end] {
if r.is_empty() {
lb.append(false);
continue;
}
for v in r {
lb.values().append_value(*v);
}
lb.append(true);
}
// Row 18's class block, from the same `tags` and the same one pass
// `class_of_tags` makes for a base table — so a way in `roads.parquet` and the
// same way in `ways.parquet` carry identical class columns. `min_zoom` keeps its
// own position above; the six new columns are appended last, exactly as
// `network_fields` declares them.
let mut cls_cols = crate::shared::schema::ClassColumns::new();
for t in &tags[at..end] {
cls_cols.push(t);
}
let mut cols: Vec<ArrayRef> = vec![
Arc::new(ids),
Arc::new(net),
Arc::new(cls),
Arc::new(mz),
Arc::new(gm),
Arc::new(tg),
Arc::new(lb.finish()),
];
cols.extend(cls_cols.finish_without_min_zoom());
out.push(RecordBatch::try_new(schema.clone(), cols)?);
at = end;
}
if out.is_empty() {
out.push(RecordBatch::new_empty(schema.clone()));
}
Ok(out)
}
fn class_histogram(classes: &[String]) -> Vec<(String, usize)> {
let mut m: HashMap<&str, usize> = HashMap::new();
for c in classes {
*m.entry(c.as_str()).or_insert(0) += 1;
}
let mut v: Vec<(String, usize)> = m.into_iter().map(|(k, n)| (String::from(k), n)).collect();
v.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
v
}
/// The sidecar an operator (and korp) reads to learn what is here, at which recipe.
/// Deliberately carries the recipe token per network for the same reason
/// `tiles::manifest_json` does: a `format` string that cannot change when the data does
/// is how a two-schema-old asset gets served as current.
pub fn manifest_json(stats: &[NetworkStats]) -> String {
let nets = stats
.iter()
.map(|s| {
let classes = s
.classes
.iter()
.map(|(c, n)| format!(r#"{{"class":{},"rows":{n}}}"#, json_str(c)))
.collect::<Vec<_>>()
.join(",");
format!(
r#"{{"name":{},"rows":{},"refused":{},"row_groups":{},"bytes":{},"recipe":{},"classes":[{classes}]}}"#,
json_str(&s.name),
s.rows,
s.refused,
s.row_groups,
s.bytes,
json_str(&s.recipe)
)
})
.collect::<Vec<_>>()
.join(",");
format!(r#"{{"format":"geoparquet-1.1-covering-networks","networks":[{nets}]}}"#)
}
fn json_str(s: &str) -> String {
serde_json::to_string(s).unwrap_or_else(|_| String::from("\"\""))
}
#[cfg(test)]
mod tests {
use super::*;
// ── the tag scanner ─────────────────────────────────────────────────────────
/// ★ **RED-when-broken: a scanner that matched anywhere in the text would read a
/// VALUE as a key.** This is the whole reason the scanner walks the object instead
/// of calling `contains("\"highway\"")` — and a `contains` implementation passes
/// every other test in this file.
#[test]
fn a_value_that_reads_like_a_key_is_not_a_key() {
let t = r#"{"barrier":"highway","name":"Not A Road"}"#;
assert_eq!(
tag_value(t, "highway"),
None,
"`highway` appears only as a VALUE here; reading it as a key admits a barrier \
into the road network"
);
assert_eq!(tag_value(t, "barrier").as_deref(), Some("highway"));
// and the network built on it must agree
assert_eq!(ROADS.class(t), None);
}
#[test]
fn the_scanner_reads_every_position_in_the_object() {
let t = r#"{"a":"1","highway":"motorway","z":"9"}"#;
assert_eq!(tag_value(t, "a").as_deref(), Some("1"));
assert_eq!(tag_value(t, "highway").as_deref(), Some("motorway"));
assert_eq!(tag_value(t, "z").as_deref(), Some("9"));
assert_eq!(tag_value(t, "missing"), None);
assert_eq!(tag_value("{}", "highway"), None);
assert_eq!(tag_value("", "highway"), None);
assert_eq!(tag_value("not json", "highway"), None);
}
/// A quote inside a value must not end the value early — otherwise the scanner
/// desynchronises and every later key on the row is read wrong.
#[test]
fn escapes_do_not_desynchronise_the_scan() {
let t = r#"{"name":"He said \"hi\"","highway":"path"}"#;
assert_eq!(tag_value(t, "name").as_deref(), Some(r#"He said "hi""#));
assert_eq!(
tag_value(t, "highway").as_deref(),
Some("path"),
"the key AFTER an escaped value must still be found"
);
}
/// The scanner must agree with a real JSON parser on real tag strings, including the
/// ones the writer emits for Swedish names.
#[test]
fn the_scanner_agrees_with_serde_json() {
for t in [
r#"{"highway":"residential","name":"Sveavägen","surface":"asphalt"}"#,
r#"{"name":"Öäå","railway":"rail"}"#,
r#"{"highway":"path","sac_scale":"hiking","trail_visibility":"good"}"#,
r#"{"note":"a, b: c {d}","highway":"track"}"#,
] {
let v: serde_json::Value = serde_json::from_str(t).unwrap();
for k in ["highway", "railway", "name", "surface", "sac_scale", "nope"] {
let want = v.get(k).and_then(serde_json::Value::as_str);
assert_eq!(
tag_value(t, k).as_deref(),
want,
"scanner disagrees with serde_json on key `{k}` of {t}"
);
}
}
}
// ── the specs ───────────────────────────────────────────────────────────────
#[test]
fn roads_take_the_highway_value_as_the_class() {
assert_eq!(
ROADS.class(r#"{"highway":"motorway"}"#).as_deref(),
Some("motorway")
);
assert_eq!(
ROADS.class(r#"{"highway":"service"}"#).as_deref(),
Some("service")
);
assert_eq!(ROADS.class(r#"{"railway":"rail"}"#), None);
}
/// ★ **RED-when-broken: the refusal list must actually refuse.** A `refuse` that is
/// consulted but empty, or consulted after the class is lowercased/renamed, reads as
/// a working filter and admits torn-up track onto the map. Both arms asserted, so
/// "everything is refused" cannot pass either.
#[test]
fn torn_up_and_not_yet_built_are_refused_but_real_infrastructure_is_not() {
for gone in [
"abandoned",
"disused",
"razed",
"dismantled",
"proposed",
"construction",
] {
let t = format!(r#"{{"railway":"{gone}"}}"#);
assert_eq!(
RAILS.class(&t),
None,
"railway={gone} is not track on the ground and must not be in the rail network"
);
}
for there in [
"rail",
"tram",
"subway",
"light_rail",
"narrow_gauge",
"platform",
] {
let t = format!(r#"{{"railway":"{there}"}}"#);
assert_eq!(
RAILS.class(&t).as_deref(),
Some(there),
"railway={there} exists on the ground and must survive the refusal list"
);
}
// the same one list governs roads
assert_eq!(ROADS.class(r#"{"highway":"construction"}"#), None);
assert_eq!(
ROADS.class(r#"{"highway":"motorway"}"#).as_deref(),
Some("motorway")
);
}
/// Trails collapse many source keys onto ONE class namespace — that is the property
/// a reader's `class IN (…)` LOD filter depends on.
#[test]
fn trail_classes_live_in_one_namespace() {
let cases = [
(
r#"{"highway":"path","sac_scale":"mountain_hiking"}"#,
"hiking",
),
(r#"{"highway":"path","trail_visibility":"bad"}"#, "hiking"),
(r#"{"highway":"path","network":"rwn"}"#, "hiking"),
(r#"{"highway":"path","network":"rcn"}"#, "cycling"),
(r#"{"highway":"path","mtb:scale":"2"}"#, "mtb"),
(r#"{"piste:type":"nordic"}"#, "ski"),
(r#"{"snowmobile":"designated"}"#, "snowmobile"),
(r#"{"route":"hiking"}"#, "hiking"),
(r#"{"route":"bicycle"}"#, "cycling"),
];
for (t, want) in cases {
assert_eq!(TRAILS.class(t).as_deref(), Some(want), "trail class of {t}");
}
// A plain path with no trail signal of its own is NOT a trail — it is only one
// if a relation says so, and that is a different dataset.
assert_eq!(TRAILS.class(r#"{"highway":"path"}"#), None);
assert_eq!(TRAILS.class(r#"{"snowmobile":"no"}"#), None);
}
/// One way can legitimately be in two networks (a piste along a track). The
/// extractor must not treat the networks as a partition.
#[test]
fn networks_overlap_and_that_is_allowed() {
let t = r#"{"highway":"track","piste:type":"nordic"}"#;
assert_eq!(ROADS.class(t).as_deref(), Some("track"));
assert_eq!(TRAILS.class(t).as_deref(), Some("ski"));
}
// ── the recipe ──────────────────────────────────────────────────────────────
/// ★ **RED-when-broken: the recipe must MOVE when what the table holds changes**, in
/// all three of its axes — columns, selection, row-group size — and a table written
/// before the recipe existed must read as STALE rather than as current.
///
/// This is the `collauto-6` defect ported: a `format` string that says the same thing
/// across a schema change lets a server conclude "already built" over data cut by a
/// two-versions-old writer. Asserted against the CURRENT field list, so a column added
/// to `network_fields` moves the token with nobody editing a guard.
#[test]
fn the_recipe_moves_with_the_columns_the_selection_and_the_granularity() {
let base = NetworkRecipe::current(&ROADS, 8_192);
// columns axis
let mut fewer = base.clone();
fewer.columns.retain(|c| !c.starts_with("id:"));
assert_ne!(
fewer.token(),
base.token(),
"dropping `id` — the exact column whose absence cost a day — must change the recipe"
);
// selection axis: a different refusal list is different DATA at the same schema
let looser = NetworkSpec {
source: Source::Ways,
name: "rails",
rules: RAILS.rules,
refuse: &[],
};
assert_ne!(
NetworkRecipe::current(&looser, 8_192).token(),
NetworkRecipe::current(&RAILS, 8_192).token(),
"admitting razed track must not read as the same cut as refusing it"
);
// granularity axis
assert_ne!(
NetworkRecipe::current(&ROADS, 65_536).token(),
base.token(),
"a different row-group size prunes differently and is a different cut"
);
// networks are distinguishable from each other
assert_ne!(base.token(), NetworkRecipe::current(&RAILS, 8_192).token());
// the TYPE axis. `min_zoom` is `UInt8`, and until 2026-08-31 `short_type` had no
// `UInt8` arm, so this whole axis was invisible: the three `UInt8` columns in the
// network family (`min_zoom`, and buildings' `levels` and `height_source`) all
// spelled `Other`, and `Other` cannot distinguish two types.
assert!(
base.token().contains("min_zoom:UInt8"),
"the recipe must spell `min_zoom`'s TYPE, not `Other`: {}",
base.token()
);
// and the token is readable, not a hash
assert!(base.token().contains("net=roads"), "{}", base.token());
assert!(base.token().contains("highway=*"), "{}", base.token());
assert!(base.token().contains("id:Int64"), "{}", base.token());
assert!(
base.token().contains(&format!("{BBOX_COLUMN}:Struct")),
"{}",
base.token()
);
}
/// ★ **RED-when-broken: a RETYPED column must make a table on disk read Stale.**
///
/// The applied-output half of the `UInt8` fix, and the one that matters: not "two
/// strings differ" but "the footer a real file declares no longer matches the recipe
/// the code implements". A table is written with today's schema and today's token,
/// then the SAME table is re-described with `min_zoom` as `UInt16` — one type changed,
/// nothing else — and the judge must say Stale.
///
/// # The red arm, seen
///
/// **Measured 2026-08-31** by restoring `_ => "Other"` in
/// `shared::schema::short_type` and running this test:
///
/// ```text
/// on disk: …,min_zoom:Other,geometry:Binary,…
/// UInt16 : …,min_zoom:Other,geometry:Binary,…
/// a UInt8 → UInt16 retype of `min_zoom` did not move the token …
/// ```
///
/// Byte-identical. Every network table on every box would have been judged **Current**
/// while its `min_zoom` column was a different width — the `collauto-6` defect exactly,
/// and the cousin `ladder-1` already paid for (korp `ebf46d1`).
///
/// # Why the two sides are not built from one function
///
/// The trap this codebase keeps hitting is a guard comparing a value to a copy of
/// itself (`korp_tile_options().zooms == DEFAULT_TILE_ZOOMS`, where the function
/// returns that constant). Here the two field lists are built from ONE list and then
/// **one entry is edited by hand**, so the inputs genuinely differ; and one side is
/// read back off a file rather than recomputed.
#[test]
fn retyping_min_zoom_makes_a_table_on_disk_read_stale() {
let fields = network_fields();
let mz = fields
.iter()
.position(|f| f.name() == "min_zoom")
.expect("network_fields carries min_zoom");
assert_eq!(
fields[mz].data_type(),
&DataType::UInt8,
"min_zoom stopped being UInt8 — this guard is now pointed at nothing"
);
let sel = ROADS.selector();
let rungs = crate::spatial::generalised_rungs_for(&network_schema(Source::Ways), true);
let now = NetworkRecipe::for_table("roads", &fields, &sel, 8_192, &rungs);
// The same table, one column retyped by hand. Nothing else moves.
let mut retyped = fields.clone();
retyped[mz] = Field::new("min_zoom", DataType::UInt16, false);
let after = NetworkRecipe::for_table("roads", &retyped, &sel, 8_192, &rungs);
// Write a REAL parquet stamped with the recipe of the day, and judge it off its
// own footer — not off the struct we just built.
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("roads.parquet");
let sch = Arc::new(Schema::new(fields.clone()));
{
// The footer key, written the way `write_tile_parquet` writes it — the
// EVIDENCE copy `of_table` reads.
let props = parquet::file::properties::WriterProperties::builder()
.set_key_value_metadata(Some(vec![parquet::file::metadata::KeyValue::new(
String::from(NETWORK_RECIPE_META_KEY),
now.token(),
)]))
.build();
let f = std::fs::File::create(&path).unwrap();
let mut w =
parquet::arrow::ArrowWriter::try_new(f, sch.clone(), Some(props)).unwrap();
w.write(&RecordBatch::new_empty(sch)).unwrap();
w.close().unwrap();
}
let on_disk = NetworkRecipe::of_table(&path)
.unwrap()
.expect("the table declares a recipe");
println!("on disk: {on_disk}");
println!("UInt16 : {}", after.token());
assert_eq!(
on_disk,
now.token(),
"the file declares a different recipe than the code stamped into it"
);
assert_ne!(
on_disk,
after.token(),
"a UInt8 → UInt16 retype of `min_zoom` did not move the token, so this table \
reads CURRENT with a column that is no longer the width the recipe implies"
);
assert!(
on_disk.contains("min_zoom:UInt8") && after.token().contains("min_zoom:UInt16"),
"the two tokens differ but not in the type of min_zoom:\n {on_disk}\n {}",
after.token()
);
// And the same for a SWAP of the two UInt8 columns the buildings table carries —
// the shape lane HEIGHT named. Under `Other` both spelled `levels:Other` /
// `height_source:Other` and swapping their types was invisible.
let a = vec![
Field::new("levels", DataType::UInt8, true),
Field::new("height_source", DataType::UInt16, false),
];
let b = vec![
Field::new("levels", DataType::UInt16, true),
Field::new("height_source", DataType::UInt8, false),
];
let ta = NetworkRecipe::for_table("buildings", &a, "x", 8_192, &[]).token();
let tb = NetworkRecipe::for_table("buildings", &b, "x", 8_192, &[]).token();
println!("swap A: {ta}\nswap B: {tb}");
assert_ne!(
ta, tb,
"swapping the types of two UInt8-family columns must move the token"
);
}
/// ★ **RED-when-broken: a ladder edit cannot reach a table that carries no
/// `min_zoom`, so it must not re-cut one.**
///
/// Recorded as "observed, not fixed" on 2026-08-31: `for_table` stamped
/// `minzoom=<rules_spelling>` into EVERY token, `graph_edges`/`graph_nodes`
/// included, and those two tables have no `min_zoom` column — their rows are
/// selected by the network selectors and valued by the junction rule, never by the
/// ladder. So every rule edit re-cut 8.4 M graph rows for a change that could not
/// appear in them. A recipe axis belongs in a token only when the table CARRIES it.
///
/// Two arms: the token as built, and the token as READ BACK off a real cut, because
/// the constructor and the stamp site could disagree. Seen RED 2026-09-01:
/// `graph_edges token carries a minzoom= term for a column it does not have`.
#[test]
fn a_table_without_min_zoom_carries_no_ladder_in_its_recipe() {
let edges = crate::graph::edge_fields();
let nodes = crate::graph::node_fields();
for f in edges.iter().chain(nodes.iter()) {
assert_ne!(
f.name(),
"min_zoom",
"the graph tables grew a min_zoom column — this guard is now pointed at \
nothing, and the ladder term belongs in their token after all"
);
}
let ge = NetworkRecipe::for_table(crate::graph::GRAPH_EDGES, &edges, "x", 8_192, &[]).token();
let gn = NetworkRecipe::for_table(crate::graph::GRAPH_NODES, &nodes, "x", 8_192, &[]).token();
println!("graph_edges: {ge}\ngraph_nodes: {gn}");
assert!(
!ge.contains("minzoom="),
"graph_edges token carries a minzoom= term for a column it does not have: {ge}"
);
assert!(
!gn.contains("minzoom="),
"graph_nodes token carries a minzoom= term for a column it does not have: {gn}"
);
// The control: a table that DOES carry min_zoom still carries the ladder, and
// the ladder is the real spelling — not an empty term.
let roads = NetworkRecipe::current(&ROADS, 8_192).token();
let ladder = crate::shared::class::rules_spelling();
assert!(
roads.contains(&format!("minzoom={ladder}")),
"roads must keep the ladder term, or a rule edit cannot re-cut it: {roads}"
);
// Applied output: the tokens a real cut stamps into the graph footers.
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
fixture_ways(&src).unwrap();
fixture_nodes(&src).unwrap();
let dst = tmp.path().join("networks");
let all = extract_networks(&src, &dst, &NetworkOptions::default()).unwrap();
for name in [crate::graph::GRAPH_EDGES, crate::graph::GRAPH_NODES] {
let s = all.iter().find(|s| s.name == name).unwrap();
let tok = NetworkRecipe::of_table(&s.path).unwrap().expect("stamped");
assert!(
!tok.contains("minzoom="),
"{name} on disk declares a ladder it cannot carry: {tok}"
);
}
let roads = all.iter().find(|s| s.name == "roads").unwrap();
let tok = NetworkRecipe::of_table(&roads.path).unwrap().expect("stamped");
assert!(tok.contains("minzoom="), "roads on disk lost its ladder: {tok}");
}
// ── the written object ──────────────────────────────────────────────────────
fn fixture_ways(dir: &Path) -> anyhow::Result<()> {
use arrow::array::ListBuilder;
let fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new(
"node_refs",
DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
true,
),
Field::new("version", DataType::Int32, true),
];
let mut meta = HashMap::new();
meta.insert(
String::from(metadata::GEO_KEY),
GeoMeta::new(&["LineString"]).to_json(),
);
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
// Five ways. Note node 2: it is in the motorway's refs AND in the rail's refs,
// and `fixture_nodes` tags it `railway=level_crossing`. That triple is the join
// the whole rail-node layer exists to keep askable.
let rows: Vec<(i64, &str, [f64; 2], Vec<i64>)> = vec![
(
10,
r#"{"highway":"motorway","name":"E4"}"#,
[17.0, 59.0],
vec![1, 2],
),
(11, r#"{"railway":"rail"}"#, [18.0, 59.5], vec![2, 3]),
(12, r#"{"railway":"razed"}"#, [12.0, 57.0], vec![4, 5]),
(
13,
r#"{"highway":"path","sac_scale":"hiking"}"#,
[20.0, 64.0],
vec![5, 6],
),
(14, r#"{"building":"yes"}"#, [11.0, 56.0], vec![7, 8]),
];
let ids = Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>());
let geoms: Vec<Vec<u8>> = rows.iter().map(|r| wkb_line(r.2)).collect();
let ga = BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice));
let ta = StringArray::from_iter_values(rows.iter().map(|r| r.1));
let mut lb = ListBuilder::new(Int64Builder::new());
for r in &rows {
for v in &r.3 {
lb.values().append_value(*v);
}
lb.append(true);
}
let va = arrow::array::Int32Array::from(vec![1i32; rows.len()]);
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(ids),
Arc::new(ga),
Arc::new(ta),
Arc::new(lb.finish()),
Arc::new(va),
],
)?;
std::fs::create_dir_all(dir)?;
let f = std::fs::File::create(dir.join("ways.parquet"))?;
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema, None)?;
w.write(&batch)?;
w.close()?;
Ok(())
}
/// A two-point LineString WKB starting at `(x, y)`.
fn wkb_line([x, y]: [f64; 2]) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&2u32.to_le_bytes());
v.extend_from_slice(&2u32.to_le_bytes());
for (a, b) in [(x, y), (x + 0.01, y + 0.01)] {
v.extend_from_slice(&a.to_le_bytes());
v.extend_from_slice(&b.to_le_bytes());
}
v
}
/// ★ **RED-when-broken: the whole extraction, asserted on the WRITTEN OBJECT** —
/// the rows that landed, the class column's values, the ids, the recipe in the
/// footer, and the covering bbox that makes it prunable. Not a round trip through
/// the structs: a state round trip would pass with a writer that never wrote a file.
#[test]
fn a_cut_network_states_its_rows_its_ids_its_classes_and_its_recipe_from_the_file() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
let dst = tmp.path().join("networks");
fixture_ways(&src).unwrap();
fixture_nodes(&src).unwrap();
let stats = extract_networks(&src, &dst, &NetworkOptions::default()).unwrap();
assert_eq!(
stats.iter().map(|s| s.name.as_str()).collect::<Vec<_>>(),
vec![
"roads",
"rails",
"trails",
"rail_nodes",
"graph_edges",
"graph_nodes"
],
"the four networks AND the two graph tables come out of the one pass"
);
let roads = &stats[0];
assert_eq!(roads.name, "roads");
assert_eq!(roads.rows, 2, "motorway + path are both highway=*");
let rails = &stats[1];
assert_eq!(rails.rows, 1, "rail is in, razed is not");
assert_eq!(rails.refused, 1, "and the refusal is COUNTED, not silent");
let trails = &stats[2];
assert_eq!(trails.rows, 1, "only the sac_scale path");
assert_eq!(trails.classes, vec![(String::from("hiking"), 1)]);
// ── read the object back ────────────────────────────────────────────────
let f = std::fs::File::open(&roads.path).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let schema = b.schema().clone();
for want in [
"id",
"network",
"class",
"geometry",
"tags",
"node_refs",
BBOX_COLUMN,
] {
assert!(
schema.index_of(want).is_ok(),
"the written road network is missing column `{want}`; schema is {schema:?}"
);
}
let mut got: Vec<(i64, String)> = Vec::new();
let mut refs_total = 0usize;
for batch in b.build().unwrap() {
let batch = batch.unwrap();
let ids = col::<Int64Array>(&batch, "id").unwrap();
let cls = col::<StringArray>(&batch, "class").unwrap();
let nr = batch
.column_by_name("node_refs")
.unwrap()
.as_any()
.downcast_ref::<ListArray>()
.unwrap();
for i in 0..batch.num_rows() {
got.push((ids.value(i), String::from(cls.value(i))));
refs_total += nr.value(i).len();
}
}
got.sort();
assert_eq!(
got,
vec![(10, String::from("motorway")), (13, String::from("path"))],
"the REAL osm ids and the REAL class values must be in the file — the tiler's \
missing-id defect collapsed every object onto way #0"
);
assert_eq!(
refs_total, 4,
"node_refs must survive the cut — 2 ways × 2 refs"
);
// the recipe, read from the footer alone
assert_eq!(
NetworkRecipe::of_table(&roads.path).unwrap().as_deref(),
Some(NetworkRecipe::current(&ROADS, 8_192).token().as_str())
);
// ── THE RED HALF ────────────────────────────────────────────────────────
// A table written WITHOUT the recipe key — the shape of every pre-recipe cut —
// must not look current.
let bare = tmp.path().join("bare.parquet");
{
let sch = network_schema(Source::Ways);
let f = std::fs::File::create(&bare).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(f, sch.clone(), None).unwrap();
w.write(&RecordBatch::new_empty(sch)).unwrap();
w.close().unwrap();
}
assert_eq!(
NetworkRecipe::of_table(&bare).unwrap(),
None,
"a table cut before the recipe existed must read as STALE, not as current"
);
}
/// ★ **RED-when-broken: the covering bbox must actually PRUNE.** A `covering.bbox`
/// column whose row-group statistics span the whole extent is present, valid and
/// completely inert — precisely the failure `spatial.rs` exists to document. So the
/// assertion is on the STATISTICS a reader prunes with, not on the column's presence.
///
/// # Why the control is input order and not a threshold
///
/// The first version of this guard asserted "no group may span more than a quarter of
/// the file's longitude extent" and it went **red on correct output**: the widest
/// group spanned 16.79° of 20°. That number is real, and it is not a defect — a
/// Hilbert run is compact in AREA, not in either axis alone; the offending group was
/// 16.79° × 1.25° = 21 deg², i.e. 10.5 % of the fixture's 200 deg², which is exactly
/// its 10 % row share. A per-axis threshold measures the wrong quantity, and picking a
/// looser one would only have hidden that.
///
/// The control is therefore the same rows in **input order** — the layout convert
/// emits, where every group's bounds are the whole file's bounds and nothing can ever
/// be pruned. Remove the Hilbert sort and the two arms converge, which is what makes
/// this guard capable of red.
#[test]
fn the_covering_bbox_prunes_and_input_order_would_not() {
use parquet::file::statistics::Statistics;
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
let dst = tmp.path().join("networks");
let n = 40_000usize;
let rg = 4_096usize;
wide_ways(&src, n).unwrap();
let opts = NetworkOptions {
row_group_rows: rg,
..NetworkOptions::default()
};
let stats = extract_networks(&src, &dst, &opts).unwrap();
let roads = stats.iter().find(|s| s.name == "roads").unwrap();
assert!(
roads.row_groups >= 4,
"need several row groups to prune between"
);
// ── the packed file, read through the statistics a reader prunes with ────
let f = std::fs::File::open(&roads.path).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let md = b.metadata();
let leaf = |name: &str| {
(0..md.file_metadata().schema_descr().num_columns())
.find(|i| md.file_metadata().schema_descr().column(*i).path().string() == name)
.unwrap_or_else(|| panic!("no {name} leaf in the written network"))
};
let (lx, ly) = (leaf("bbox.xmin"), leaf("bbox.ymin"));
let span = |g: usize, c: usize| match md.row_group(g).column(c).statistics() {
Some(Statistics::Float(s)) => f64::from(*s.max_opt().unwrap() - *s.min_opt().unwrap()),
other => panic!("row group {g} column {c} has no float statistics: {other:?}"),
};
let packed: f64 = (0..md.num_row_groups())
.map(|g| span(g, lx) * span(g, ly))
.sum::<f64>()
/ md.num_row_groups() as f64;
// ── the control: the SAME rows in the order convert emitted them ─────────
let sf = std::fs::File::open(src.join("ways.parquet")).unwrap();
let sb = ParquetRecordBatchReaderBuilder::try_new(sf)
.unwrap()
.build()
.unwrap();
let mut boxes: Vec<crate::spatial::Bbox> = Vec::with_capacity(n);
for batch in sb {
let batch = batch.unwrap();
let g = col::<BinaryArray>(&batch, "geometry").unwrap();
for i in 0..batch.num_rows() {
boxes.push(crate::spatial::wkb_bbox(g.value(i)).unwrap());
}
}
assert_eq!(boxes.len(), n);
let unsorted: f64 = boxes
.chunks(rg)
.map(|c| {
let x0 = c.iter().map(|b| b.xmin).fold(f64::MAX, f64::min);
let x1 = c.iter().map(|b| b.xmax).fold(f64::MIN, f64::max);
let y0 = c.iter().map(|b| b.ymin).fold(f64::MAX, f64::min);
let y1 = c.iter().map(|b| b.ymax).fold(f64::MIN, f64::max);
(x1 - x0) * (y1 - y0)
})
.sum::<f64>()
/ boxes.chunks(rg).count() as f64;
assert!(
packed * 4.0 < unsorted,
"the Hilbert-packed layout must shrink the mean row-group bbox area by more \
than 4x against input order, or the covering column prunes nothing a reader \
can use: packed {packed:.2} deg² vs input-order {unsorted:.2} deg² over {} \
groups",
md.num_row_groups()
);
}
/// The node table beside `fixture_ways`. Node 2 is the level crossing shared by the
/// motorway (way 10) and the rail (way 11); node 9 is a signal on nothing; node 7 is
/// an untagged node that must not be in the layer.
fn fixture_nodes(dir: &Path) -> anyhow::Result<()> {
let fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new("version", DataType::Int32, true),
];
let mut meta = HashMap::new();
meta.insert(
String::from(metadata::GEO_KEY),
GeoMeta::new(&["Point"]).to_json(),
);
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
let rows: Vec<(i64, &str, [f64; 2])> = vec![
(2, r#"{"railway":"level_crossing"}"#, [17.5, 59.2]),
(3, r#"{"railway":"switch"}"#, [18.1, 59.6]),
(9, r#"{"railway":"signal"}"#, [18.2, 59.7]),
(4, r#"{"railway":"razed"}"#, [12.0, 57.0]),
(7, r#"{"amenity":"bench"}"#, [11.0, 56.0]),
];
let ids = Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>());
let geoms: Vec<Vec<u8>> = rows.iter().map(|r| wkb_point(r.2)).collect();
let ga = BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice));
let ta = StringArray::from_iter_values(rows.iter().map(|r| r.1));
let va = arrow::array::Int32Array::from(vec![1i32; rows.len()]);
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(ids), Arc::new(ga), Arc::new(ta), Arc::new(va)],
)?;
std::fs::create_dir_all(dir)?;
let f = std::fs::File::create(dir.join("nodes.parquet"))?;
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema, None)?;
w.write(&batch)?;
w.close()?;
Ok(())
}
/// A Point WKB.
fn wkb_point([x, y]: [f64; 2]) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&1u32.to_le_bytes());
v.extend_from_slice(&x.to_le_bytes());
v.extend_from_slice(&y.to_le_bytes());
v
}
/// Every `(id, class, node_refs)` a written network table holds.
fn read_back(p: &Path) -> Vec<(i64, String, Vec<i64>)> {
let f = std::fs::File::open(p).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f)
.unwrap()
.build()
.unwrap();
let mut out = Vec::new();
for batch in b {
let batch = batch.unwrap();
let ids = col::<Int64Array>(&batch, "id").unwrap();
let cls = col::<StringArray>(&batch, "class").unwrap();
let nr = batch
.column_by_name("node_refs")
.unwrap()
.as_any()
.downcast_ref::<ListArray>()
.unwrap();
for i in 0..batch.num_rows() {
let refs: Vec<i64> = if nr.is_null(i) {
Vec::new()
} else {
nr.value(i)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.iter()
.flatten()
.collect()
};
out.push((ids.value(i), String::from(cls.value(i)), refs));
}
}
out.sort();
out
}
/// ★ **RED-when-broken: the level crossing is still ASKABLE from the written files.**
///
/// This is the whole argument for keeping `node_refs` in the ways and `id` on the
/// nodes, and it is asserted the way it would actually be used — an id intersection
/// across three separate parquet files, read back off disk. Flatten any one of the
/// three to geometry and this test cannot pass; that is the point.
///
/// It has a real negative arm too: `railway=switch` (node 3) is on the rail and NOT
/// on any road, so it must NOT come out of the intersection. A join that returned
/// every rail node would satisfy the positive arm alone.
#[test]
fn a_level_crossing_is_still_a_joint_between_the_two_networks_after_the_cut() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
let dst = tmp.path().join("networks");
fixture_ways(&src).unwrap();
fixture_nodes(&src).unwrap();
extract_networks(&src, &dst, &NetworkOptions::default()).unwrap();
let roads = read_back(&dst.join("roads.parquet"));
let rails = read_back(&dst.join("rails.parquet"));
let nodes = read_back(&dst.join("rail_nodes.parquet"));
// A node layer carries no refs of its own — its id is what the ways point at.
assert!(
nodes.iter().all(|(_, _, r)| r.is_empty()),
"a node row must have a NULL node_refs, not a fabricated one: {nodes:?}"
);
assert_eq!(
nodes
.iter()
.map(|(i, c, _)| (*i, c.as_str()))
.collect::<Vec<_>>(),
vec![(2, "level_crossing"), (3, "switch"), (9, "signal")],
"the rail-node layer must hold the tagged nodes with their REAL osm ids, must \
refuse the razed one, and must not admit the bench"
);
// THE JOIN, done from the files alone.
let on_road: std::collections::HashSet<i64> = roads
.iter()
.flat_map(|(_, _, r)| r.iter().copied())
.collect();
let on_rail: std::collections::HashSet<i64> = rails
.iter()
.flat_map(|(_, _, r)| r.iter().copied())
.collect();
let crossings: Vec<i64> = nodes
.iter()
.filter(|(i, c, _)| c == "level_crossing" && on_road.contains(i) && on_rail.contains(i))
.map(|(i, _, _)| *i)
.collect();
assert_eq!(
crossings,
vec![2],
"node 2 is tagged level_crossing AND is in the motorway's refs AND in the \
rail's refs — the joint must survive the cut"
);
// THE NEGATIVE ARM: the switch is on the rail only, so the intersection drops it.
assert!(
on_rail.contains(&3) && !on_road.contains(&3),
"node 3 must be on the rail and not on any road, else the join proves nothing"
);
}
/// ★ **RED-when-broken: the graph comes out of the SAME pass, carries a recipe, and a
/// graph cut from roads alone must not read as one cut from roads AND rails.**
///
/// The partial cut is the dangerous one: it produces a `graph_edges.parquet` that is
/// valid, prunable, routable — and has no level crossings in it at all, because a
/// crossing is a joint only when both networks are counted together. If its recipe
/// matched the full cut's, a server would serve the crippled graph as the current one.
#[test]
fn the_graph_rides_the_network_pass_and_a_partial_cut_reads_as_a_different_object() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
fixture_ways(&src).unwrap();
fixture_nodes(&src).unwrap();
let full = tmp.path().join("full");
let all = extract_networks(&src, &full, &NetworkOptions::default()).unwrap();
let ge = all.iter().find(|s| s.name == "graph_edges").unwrap();
let gn = all.iter().find(|s| s.name == "graph_nodes").unwrap();
assert!(
ge.path.is_file() && gn.path.is_file(),
"both graph tables must be written"
);
// the recipe is in the FOOTER, read back through the same probe a network uses
let tok = NetworkRecipe::of_table(&ge.path)
.unwrap()
.expect("graph must be stamped");
assert!(tok.contains("net=graph_edges"), "{tok}");
assert!(tok.contains("from:roads+rails+rail_nodes"), "{tok}");
assert!(
tok.contains("level_crossing"),
"the junction rule must be in the token: {tok}"
);
assert!(
tok.contains("oneway:Int8"),
"the direction column must be in the token: {tok}"
);
let g = ge
.graph
.as_ref()
.expect("the edge table carries the build's measurements");
assert_eq!(
g.joints, 1,
"node 2 is on the motorway AND on the rail — one joint"
);
assert_eq!(g.whole_way_fallback, 0);
// ── THE RED HALF: roads only ────────────────────────────────────────────
let part = tmp.path().join("roads-only");
let opts = NetworkOptions {
networks: vec![ROADS],
..NetworkOptions::default()
};
let only = extract_networks(&src, &part, &opts).unwrap();
let pe = only.iter().find(|s| s.name == "graph_edges").unwrap();
let ptok = NetworkRecipe::of_table(&pe.path).unwrap().unwrap();
assert_ne!(
ptok, tok,
"a graph with no rail in it must NOT read as the same cut as one with rail — \
that is how a crippled graph gets served as current"
);
assert_eq!(
pe.graph.as_ref().unwrap().joints,
0,
"and it really is crippled: with rails left out there is no joint to find"
);
}
/// Many ways spread over a wide area, so row groups can be spatially distinct.
fn wide_ways(dir: &Path, n: usize) -> anyhow::Result<()> {
use arrow::array::ListBuilder;
let fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new(
"node_refs",
DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
true,
),
];
let mut meta = HashMap::new();
meta.insert(
String::from(metadata::GEO_KEY),
GeoMeta::new(&["LineString"]).to_json(),
);
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
let ids = Int64Array::from((0..n as i64).collect::<Vec<_>>());
let geoms: Vec<Vec<u8>> = (0..n)
.map(|i| {
// A pseudo-random scatter over 20° × 10°, so input order is NOT spatial
// order — the Hilbert sort has to be what makes the groups compact.
let h = (i as u64).wrapping_mul(2_654_435_761);
let x = 5.0 + ((h >> 8) % 20_000) as f64 / 1000.0;
let y = 55.0 + ((h >> 24) % 10_000) as f64 / 1000.0;
wkb_line([x, y])
})
.collect();
let ga = BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice));
let ta = StringArray::from_iter_values((0..n).map(|_| r#"{"highway":"residential"}"#));
let mut lb = ListBuilder::new(Int64Builder::new());
for _ in 0..n {
lb.values().append_value(1);
lb.append(true);
}
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(ids),
Arc::new(ga),
Arc::new(ta),
Arc::new(lb.finish()),
],
)?;
std::fs::create_dir_all(dir)?;
let f = std::fs::File::create(dir.join("ways.parquet"))?;
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema, None)?;
w.write(&batch)?;
w.close()?;
fixture_nodes(dir)?;
Ok(())
}
/// The row ceiling is a real stop, not a log line.
#[test]
fn the_row_ceiling_refuses_instead_of_growing() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
fixture_ways(&src).unwrap();
fixture_nodes(&src).unwrap();
let opts = NetworkOptions {
max_rows: 1,
..NetworkOptions::default()
};
let e = extract_networks(&src, &tmp.path().join("out"), &opts).unwrap_err();
assert!(
e.to_string().contains("row ceiling"),
"the ceiling must be the error, got: {e}"
);
}
#[test]
fn the_manifest_names_every_network_and_its_recipe() {
let s = vec![NetworkStats {
name: String::from("roads"),
path: PathBuf::from("/x/roads.parquet"),
rows: 3,
refused: 1,
rows_no_geometry: 0,
row_groups: 1,
bytes: 99,
classes: vec![(String::from("motorway"), 3)],
recipe: NetworkRecipe::current(&ROADS, 8_192).token(),
graph: None,
}];
let j = manifest_json(&s);
let v: serde_json::Value = serde_json::from_str(&j).expect("manifest must be valid JSON");
assert_eq!(v["networks"][0]["name"], "roads");
assert_eq!(v["networks"][0]["refused"], 1);
assert_eq!(v["networks"][0]["classes"][0]["class"], "motorway");
assert!(
v["networks"][0]["recipe"]
.as_str()
.unwrap()
.contains("net=roads")
);
}
}