src/osm.rs
//! **OSM → GeoParquet** connector — a first-class catalogue entry. The basemap +
//! road/building geometry layer the facett map facets render UNDER the transport
//! overlays. Per the mandate it **uses znippy-zoomies / `osm-katana`** (the fast
//! geo primitives) for the OSM-PBF unpack/convert: pass-1 builds a node store,
//! pass-2 resolves way node-refs and **WKB-encodes** the geometry. We land those
//! way records as `(way_id, geometry_wkb, tags)` into skade Iceberg (WKB lon/lat),
//! the same pure-Rust → Arrow → skade korp-lite path as every other collector.
//!
//! Behind `--features collectors-osm` (pulls `osm-katana`). Offline contract:
//! [`ways_to_batch`] maps way records → Arrow (asserted as DATA without needing a
//! live PBF); the live `collect` reads a PBF path from `params.path`.
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
/// Re-exported so consumers + tests can build/parse way records without a direct
/// `osm-katana` dependency (it is korp-collectors' own, behind the `osm` feature).
pub use osm_katana::{PbfWayRecord, load_ways_pbf};
use super::arrow_util::{batch, bin_col, i64_col, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
use super::osm_admin::{admin_batch, admin_schema, extract_admin_units};
/// The OSM way table schema: `way_id` (Int64), `geometry_wkb` (Binary, WKB
/// LineString/Polygon in lon/lat), `tags` (Utf8 JSON map).
pub fn osm_schema() -> Schema {
schema(&[
("way_id", DataType::Int64, false),
("geometry_wkb", DataType::Binary, true),
("tags", DataType::Utf8, true),
])
}
/// **The OSM container guard.** `osm-katana` is a fast, trusting reader: handed a
/// file that is not an OSM container it does not return an error, it **panics**
/// (inside its own blocking worker), which surfaces to a caller as an opaque
/// `JoinError` instead of an honest RAGNARÖK `reason`. znippy-zoomies is
/// hands-off, so the sniff belongs here, at the seam that decides what to hand it.
///
/// Returns `Ok(())` when `path` looks like an OSM container:
/// * **PBF** — a `BlobHeader` whose type string (`OSMHeader`/`OSMData`) appears in
/// the first 64 bytes, the fileformat's fixed preamble;
/// * **XML** — an `<?xml` / `<osm` document opener.
///
/// Otherwise `Err` naming what was found, so the collector turns it into a
/// reason. (Found by the registry RAGNARÖK sweep feeding every collector an empty
/// source: this one panicked instead of reporting.)
pub fn ensure_osm_container(path: &str) -> Result<()> {
use std::io::Read;
let mut file = std::fs::File::open(path)
.map_err(|e| anyhow::anyhow!("OSM source `{path}` is not readable: {e}"))?;
let mut head = [0u8; 64];
let n = file
.read(&mut head)
.map_err(|e| anyhow::anyhow!("OSM source `{path}` is not readable: {e}"))?;
let head = &head[..n];
if head.is_empty() {
return Err(anyhow::anyhow!("OSM source `{path}` is empty (0 bytes)"));
}
// PBF: the BlobHeader's `type` field is the literal "OSMHeader" (first blob)
// or "OSMData"; both sit within the first few bytes of a well-formed file.
let is_pbf = head
.windows(9)
.any(|w| w == b"OSMHeader" || w[..7] == *b"OSMData");
// XML: an `<?xml …` prologue or a bare `<osm` root.
let lead = String::from_utf8_lossy(&head[..head.len().min(16)]);
let is_xml = lead.trim_start().starts_with("<?xml") || lead.trim_start().starts_with("<osm");
if is_pbf || is_xml {
return Ok(());
}
Err(anyhow::anyhow!(
"OSM source `{path}` is not an OSM container: {n} byte(s) with neither a PBF \
BlobHeader (OSMHeader/OSMData) nor an XML opener"
))
}
/// osm-katana way records → one Arrow `RecordBatch` (WKB geometry preserved).
pub fn ways_to_batch(ways: &[PbfWayRecord]) -> Result<RecordBatch> {
let cols = vec![
i64_col(&ways.iter().map(|w| Some(w.id)).collect::<Vec<_>>()),
bin_col(&ways.iter().map(|w| w.geometry.clone()).collect::<Vec<_>>()),
str_col(
&ways
.iter()
.map(|w| Some(w.tags.clone()))
.collect::<Vec<_>>(),
),
];
batch(&osm_schema(), cols)
}
/// **The OpenStreetMap half of both human cards, written once.**
///
/// `osm-geoparquet` and `osm-admin-boundaries` read the SAME `.osm.pbf` and differ
/// only in which features they keep, so authority, data owner and terms page are
/// identical by construction rather than by two hand-written copies agreeing. Only
/// `docs_url` and the per-arm quirks differ, so only `docs_url` is an argument.
///
/// The authority/owner split is the real one and worth stating precisely: the OSMF
/// hosts the project and publishes the licence, but the data belongs to the
/// contributors who surveyed it, which is why the attribution string names *them*.
fn osm_card(docs_url: &str) -> crate::human_doc::HumanDoc {
crate::human_doc::HumanDoc::new(
"OpenStreetMap — the OpenStreetMap Foundation (OSMF) hosts the project and \
publishes the licence; it does not own the data",
"the OpenStreetMap contributors who surveyed it — attribution is \
\"© OpenStreetMap contributors\"",
"https://www.openstreetmap.org/copyright",
docs_url,
)
}
/// The OSM → GeoParquet collector.
#[derive(Default)]
pub struct OsmGeoCollector;
#[async_trait::async_trait]
impl Collector for OsmGeoCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "osm-geoparquet".into(),
source_kind: SourceKind::BulkFile,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["osm-pbf".into(), "osm-xml".into(), "geoparquet(wkb)".into()],
license: "ODbL (OpenStreetMap)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["coordinate".into(), "way_id".into()],
description: "OSM PBF/XML → WKB geometry via znippy-zoomies/osm-katana → skade \
(basemap + road/building layer for map facets)."
.into(),
}
}
/// **ODbL's share-alike is the operator-facing fact here, and it is not a licence
/// footnote — it is a condition on what you may do with the WAREHOUSE.**
///
/// Everything else about OSM is comfortable: no key, no account, no rate limit,
/// because this collector never makes a request at all — it reads a `.osm.pbf` off
/// the disk. The thing that bites arrives much later, when a derived table built
/// from these ways is published, and by then nobody remembers where the geometry
/// came from. So the card leads with it.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, RateLimit, Registration};
Some(osm_card("https://wiki.openstreetmap.org/wiki/Downloading_data")
.with_contact(Contact::Form(
"https://community.openstreetmap.org/".into(),
))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"none applies to this collector: it reads a local `.osm.pbf` from \
`params.path` and makes no HTTP request whatsoever. Downloading the extract \
is a SEPARATE step under the mirror's own policy — the declared source is a \
Geofabrik Sweden extract, and Geofabrik publishes its terms at \
download.geofabrik.de/technical.html."
.into(),
))
.with_quirk(
"ODbL is SHARE-ALIKE, and that reaches further than attribution: publish a \
Derivative Database built from these ways and you must offer it under ODbL \
too, with attribution to \"© OpenStreetMap contributors\". A warehouse table \
joined against `osm_ways` can be one. Decide this before the table is \
shared, not after.",
)
.with_quirk(
"`params.path` is a LOCAL FILE, not a URL — the collector never fetches. With \
neither `params.path` nor `base_url` set it reports `no OSM PBF path \
configured (params.path)` rather than downloading anything.",
)
.with_quirk(
"Handed a file that is not an OSM container, `osm-katana` PANICS inside its \
blocking worker and the caller sees an opaque `JoinError`. \
`ensure_osm_container` sniffs the first 64 bytes for a PBF `BlobHeader` \
(`OSMHeader`/`OSMData`) or an XML opener FIRST and turns a miss into an \
honest reason. Found by the registry sweep that fed every collector an empty \
source.",
)
.with_quirk(
"Ways only, and geometry is WKB in lon/lat order — nodes and relations are \
not landed by this arm (relations are `osm-admin-boundaries`' job). Tags \
arrive as one JSON string per way in `tags`, not as columns.",
))
}
fn source_target(&self) -> SourceTarget {
// The basemap ways layer facett's osm3d scene renders UNDER the overlays.
SourceTarget::map_scene(
SourceEndpoint::Osm,
"https://download.geofabrik.de/europe/sweden-latest.osm.pbf",
"osm_ways",
"osm3d",
)
}
/// **One row per OSM way.** The derived key named `coordinate`, which the way
/// table does not carry (its geometry is a WKB blob).
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["way_id"],
"one row per OSM way — `way_id` is OSM's own permanent id for it",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(osm_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
// `params.path` (or base_url as a local path) → the .osm.pbf to convert.
let path = cfg
.params
.get("path")
.and_then(|v| v.as_str())
.map(str::to_string)
.unwrap_or_else(|| cfg.base_url.clone());
if path.is_empty() {
return Ok(CollectStats {
reason: Some("no OSM PBF path configured (params.path)".into()),
..Default::default()
});
}
// Guard BEFORE handing the path to osm-katana: a non-OSM file makes it
// panic in the blocking worker, which would reach the caller as an opaque
// JoinError rather than an honest reason.
if let Err(e) = ensure_osm_container(&path) {
return Ok(CollectStats {
reason: Some(e.to_string()),
..Default::default()
});
}
// osm-katana is sync + CPU-heavy → run it off the async runtime.
let ways = tokio::task::spawn_blocking(move || osm_katana::load_ways_pbf(&path)).await??;
if ways.is_empty() {
return Ok(CollectStats {
reason: Some("OSM PBF held no resolvable ways".into()),
..Default::default()
});
}
let n = ways.len() as u64;
sink.push(ways_to_batch(&ways)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
/// The **administrative-boundaries** collector — filters OSM
/// `boundary=administrative` features → an `admin_units` border table (the
/// borders-only world facet). Authoritative admin areas are OSM *relations*
/// (multipolygons): `osm-katana::load_admin_boundaries_pbf` resolves their member
/// ways, stitches rings (+ holes), and emits a **WKB MultiPolygon** per relation,
/// so the `geometry_wkb` column now carries real border geometry — not just the
/// level/name/ISO registry + trie.
#[derive(Default)]
pub struct OsmAdminCollector;
#[async_trait::async_trait]
impl Collector for OsmAdminCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "osm-admin-boundaries".into(),
source_kind: SourceKind::BulkFile,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["osm-pbf".into(), "geoparquet(wkb)".into()],
license: "ODbL (OpenStreetMap)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["country".into(), "iso".into(), "admin_level".into()],
description: "OSM boundary=administrative → admin_units border table \
(world admin-division trie: Sweden kommuner/regioner, NO/FI/…)."
.into(),
}
}
/// **Same source, same licence, different bite.** The ODbL share-alike warning is
/// shared with `osm-geoparquet` via [`osm_card`]; what is specific here is that
/// administrative borders are CROWD-MAPPED — `admin_level` is a convention that
/// differs per country, and an official border register is a different product with
/// different terms. An operator rendering a clickable Sweden needs to know that
/// before someone asks whether a kommun boundary is authoritative.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, RateLimit, Registration};
Some(
osm_card("https://wiki.openstreetmap.org/wiki/Tag:boundary%3Dadministrative")
.with_contact(Contact::Form("https://community.openstreetmap.org/".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"none applies to this collector: it reads a local `.osm.pbf` from \
`params.path` and makes no HTTP request whatsoever. Downloading the \
extract is a SEPARATE step under the mirror's own policy — the \
declared source is a Geofabrik Sweden extract, and Geofabrik \
publishes its terms at download.geofabrik.de/technical.html."
.into(),
))
.with_quirk(
"ODbL is SHARE-ALIKE: an `admin_units` table published onward, or a \
database derived from it, carries the same obligation and must credit \
\"© OpenStreetMap contributors\". Borders feel like reference data \
and get copied around freely — this is the one that gets forgotten.",
)
.with_quirk(
"These borders are CROWD-MAPPED, not an official register. \
`admin_level` is an OSM convention whose meaning differs per country \
(Sweden's regioner and kommuner do not sit at the same level as \
another country's), so filter by level per country rather than \
globally, and do not treat a boundary as legally authoritative.",
)
.with_quirk(
"Admin areas are OSM RELATIONS, not ways: `osm-katana` resolves each \
relation's member ways, stitches rings and holes, and emits one WKB \
MultiPolygon. A relation whose ring does not close contributes no \
geometry, so a missing border is an upstream mapping gap rather than \
a parse failure.",
)
.with_quirk(
"`params.path` is a LOCAL FILE and the same non-OSM sniff applies as \
for `osm-geoparquet`: without `ensure_osm_container`, a file that is \
not an OSM container makes `osm-katana` panic inside its blocking \
worker and reaches the caller as an opaque `JoinError`.",
),
)
}
fn source_target(&self) -> SourceTarget {
// The admin_units border table the facett clickable-Sweden geomap renders.
SourceTarget::map_scene(
SourceEndpoint::OsmAdmin,
"https://download.geofabrik.de/europe/sweden-latest.osm.pbf",
"admin_units",
"clickable-sweden",
)
}
/// **One row per administrative unit** — `osm_id`, OSM's own permanent relation id.
///
/// The derived key was `["country", "iso", "admin_level"]`, which is not an
/// identity at all: every one of Sweden's 290 kommuner shares
/// `(Sverige, SE, 7)`, so an Upsert on it would have collapsed the whole country
/// into ONE row per admin level. This is the worst of the 29 keys the O3 sweep
/// found, and it names exactly why a partly-plausible key is more dangerous than
/// no key.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["osm_id"],
"one row per administrative unit — `osm_id` is OSM's own permanent relation \
id; `(country, iso, admin_level)` is shared by every kommun in the country",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(admin_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let path = cfg
.params
.get("path")
.and_then(|v| v.as_str())
.map(str::to_string)
.unwrap_or_else(|| cfg.base_url.clone());
if path.is_empty() {
return Ok(CollectStats {
reason: Some("no OSM PBF path configured (params.path)".into()),
..Default::default()
});
}
// Same guard as the ways collector: a non-OSM file must come back as a
// reason, not a panic inside osm-katana's blocking worker.
if let Err(e) = ensure_osm_container(&path) {
return Ok(CollectStats {
reason: Some(e.to_string()),
..Default::default()
});
}
// Admin areas are OSM relations (multipolygons): osm-katana resolves their
// member ways and assembles WKB MultiPolygon geometry — `geometry_wkb` now
// lands real borders, not an empty column.
let rels =
tokio::task::spawn_blocking(move || osm_katana::load_admin_boundaries_pbf(&path))
.await??;
let units = extract_admin_units(rels.into_iter().map(|r| (r.id, r.tags, r.geometry)));
if units.is_empty() {
return Ok(CollectStats {
reason: Some("no boundary=administrative relations in this PBF".into()),
..Default::default()
});
}
let n = units.len() as u64;
sink.push(admin_batch(&units)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}