osm-katana/src/verify.rs
//! GeoParquet output verification — the ONE canonical implementation.
//!
//! Both the `osm-katana verify` CLI subcommand and any programmatic caller go
//! through [`verify`] so they can never report divergent numbers. It reads the
//! parquet footer once (rows + geometry null/set straight from column
//! statistics — no data decode) and, for `node_refs`, does a no-barrier,
//! column-projected, row-group-parallel count. If the footer lacks geometry
//! statistics it falls back to a (still column-projected, parallel) geometry
//! decode rather than silently reporting wrong numbers.
use std::path::Path;
use std::sync::Arc;
use parquet::file::metadata::ParquetMetaData;
/// Verified counts for a single parquet file — obtained from the footer (rows +
/// geometry null) for free, and from a no-barrier, column-projected,
/// row-group-parallel decode for `node_refs`.
pub struct VerifyCounts {
pub rows: usize,
pub has_geom: bool,
pub geom_set: usize,
pub geom_null: usize,
pub has_refs: bool,
pub refs_nonempty: usize,
pub refs_total: usize,
/// **The DELTA 3 layout, read off the footer — see [`Fault::UnbandedPacked`].**
///
/// `packed` is `covering.bbox` present; `banded` is a `min_zoom` column whose
/// row-group statistics are monotone across the file
/// ([`crate::spatial::is_band_first`], the packer's own predicate — LAW 5, so the
/// guard cannot drift from the writer). `taggable` is a `tags` column, i.e. a table
/// the one ladder CAN rung; a table with no tags is not judged on a column it can
/// never derive.
pub packed: bool,
pub banded: bool,
pub taggable: bool,
/// Row groups in the file — what makes the fault message say how big the scan it
/// prevented would have been.
pub row_groups: usize,
}
/// **The tables an extract is judged on, and the ONE place that list lives.**
///
/// `relations.resolved.parquet` is in here because it is the table
/// [`crate::tiles::tile_region`] *prefers* — it reads exactly three files, and the
/// relation one is the resolved table when it exists and the raw one otherwise. A
/// list that stopped at `relations.parquet` therefore never looked at the table the
/// cut actually consumes: MEASURED 2026-08-24, the `verify` verb's output on an
/// extract whose `relations.resolved.parquet` was absent, 0-row or all-null was
/// **byte-identical** to its output on a good one.
pub const EXTRACT_TABLES: [TableSpec; 4] = [
TableSpec::with_geometry("nodes.parquet"),
TableSpec::with_geometry("ways.parquet"),
// ⚠ RAW relations carry NO geometry, and that is CORRECT, not a fault. Geometry for
// relations is produced by `crate::relgeom` into `relations.resolved.parquet`; the
// raw table is the INPUT to that step and holds ids, tags and members only.
//
// MEASURED on the live Sweden extract 2026-08-24: `relations.parquet` reports
// 0 of 14 632 472 geometries set while `relations.resolved.parquet` reports
// 225 799 of 274 206 set. A blanket "geometry all null is a fault" rule fails a
// HEALTHY tree here — which is exactly the shape of guard this whole change exists
// to remove, so the exemption is declared rather than discovered.
TableSpec::without_geometry("relations.parquet"),
TableSpec::with_geometry("relations.resolved.parquet"),
];
/// One judged table: its filename, and whether an empty `geometry` column is a FAULT
/// for it. The flag is per-table because "has a geometry column" and "is supposed to
/// have geometry in it" are different questions — see the note on `relations.parquet`.
#[derive(Clone, Copy, Debug)]
pub struct TableSpec {
pub name: &'static str,
pub geometry_required: bool,
}
impl TableSpec {
pub const fn with_geometry(name: &'static str) -> Self {
Self { name, geometry_required: true }
}
pub const fn without_geometry(name: &'static str) -> Self {
Self { name, geometry_required: false }
}
}
/// **Why one table failed** — the reasons a `verify` is entitled to exit non-zero.
///
/// Each arm is a shape MEASURED on the live stack, not a hypothetical:
/// `GeometryAllNull` is `~/.korp/work/sweden/relations.parquet` on 2026-08-04 (0 of
/// 14 592 834 geometries set, which cut a whole Sweden pyramid with no lake, no
/// forest clearing and no building outline in it); `Absent` is the same day's
/// missing `relations.resolved.parquet`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Fault {
/// The named table is not in the directory at all.
Absent,
/// The table is there and holds no rows.
NoRows,
/// Every row's `geometry` is null — a table of ids and tags with no shapes.
GeometryAllNull { rows: usize },
/// **A PACKED table with no `min_zoom` band.** It has `covering.bbox`, so every
/// consumer treats it as the served layout and stops looking; it has `tags`, so the
/// one ladder (`crate::shared::class::min_zoom_of_tags`) could have rung every row;
/// and it has neither the column nor the band-first order, so the rung prune that
/// [`crate::spatial::pack_batches`] exists to enable does nothing on it and a
/// country camera falls back to the covering box alone.
///
/// This is not a hypothetical either. It is `~/.korp/work/sweden/nodes.parquet` as
/// it stood on 2026-09-03: packed 2026-09-01 from a convert that predated DELTA 1,
/// 12 633 row groups, 103 488 583 rows, `tags` present, `min_zoom` absent. Its two
/// siblings in the same directory carried the band. korp's parquet mirror measured
/// what that cost — **587–990 MB for one country viewport against 7.8–15.2 MB for
/// the banded tables** — and excluded the points table rather than the fault, which
/// is what a missing guard buys: the artefact stays wrong and the consumer routes
/// around it.
///
/// The repair is one command and it is idempotent — [`crate::spatial::is_packed`]
/// already answers `false` for this shape, so `osm-katana spatial-pack` re-cuts it
/// and a second run reports it already complete.
UnbandedPacked { rows: usize, row_groups: usize },
}
impl std::fmt::Display for Fault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Absent => write!(f, "ABSENT"),
Self::NoRows => write!(f, "0 rows"),
Self::GeometryAllNull { rows } => {
write!(f, "geometry ALL NULL — 0 of {rows} rows carry a shape")
}
Self::UnbandedPacked { rows, row_groups } => write!(
f,
"PACKED but UNBANDED — `covering.bbox` and `tags` are present and \
`min_zoom` is not, so the rung prune is dead and a country camera \
must consider all {row_groups} row groups ({rows} rows); re-run \
`osm-katana spatial-pack`"
),
}
}
}
impl VerifyCounts {
/// **The verdict on one present table, as ONE expression.**
///
/// This is the whole point of putting the rule here rather than in the binary:
/// the line `osm-katana verify` prints and the exit code it returns come from
/// this one call, so they can never say different things. Before this existed
/// the verb printed `0 set, 100 null` and exited **0**.
pub fn fault(&self, geometry_required: bool) -> Option<Fault> {
if self.rows == 0 {
return Some(Fault::NoRows);
}
// Two conditions, not one. A table must HAVE a geometry column (a future table
// without one must not be judged on a column it never carried) AND be one whose
// geometry is supposed to be populated at this stage — `relations.parquet` is
// the counter-example that made this flag necessary.
if geometry_required && self.has_geom && self.geom_set == 0 {
return Some(Fault::GeometryAllNull { rows: self.rows });
}
// Three conditions, and each one is load-bearing. PACKED, because an unpacked
// convert output is a legitimate intermediate that the pack pass is about to
// consume. TAGGABLE, because a table with no `tags` has nothing for the ladder
// to read and its absence of a rung is honest. NOT BANDED, which is the finding.
if self.packed && self.taggable && !self.banded {
return Some(Fault::UnbandedPacked {
rows: self.rows,
row_groups: self.row_groups,
});
}
None
}
}
/// One table's counts and verdict. `counts` is `None` exactly when the file is absent.
pub struct TableVerdict {
pub name: String,
pub counts: Option<VerifyCounts>,
pub fault: Option<Fault>,
}
impl TableVerdict {
pub fn ok(&self) -> bool {
self.fault.is_none()
}
}
/// **Verify every named table under `dir`.** Returns one [`TableVerdict`] per entry
/// of `tables` in the order given — reporting is the caller's job, but the JUDGEMENT
/// is not.
///
/// A read error is propagated rather than folded into a [`Fault`]: "this file is
/// corrupt enough that the parquet reader refused it" is not the same finding as
/// "this table is empty", and a verb that flattened the two would be back to
/// guessing.
pub fn verify_dir(dir: &Path, tables: &[TableSpec]) -> anyhow::Result<Vec<TableVerdict>> {
let mut out = Vec::with_capacity(tables.len());
for spec in tables {
let path = dir.join(spec.name);
if !path.exists() {
out.push(TableVerdict {
name: spec.name.to_string(),
counts: None,
fault: Some(Fault::Absent),
});
continue;
}
let counts = verify(&path)?;
let fault = counts.fault(spec.geometry_required);
out.push(TableVerdict {
name: spec.name.to_string(),
counts: Some(counts),
fault,
});
}
Ok(out)
}
/// Verify a single GeoParquet file, returning its [`VerifyCounts`].
pub fn verify(path: &Path) -> anyhow::Result<VerifyCounts> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use std::fs::File;
// One footer read. No data decode yet.
let builder = ParquetRecordBatchReaderBuilder::try_new(File::open(path)?)?;
let arrow_schema = builder.schema().clone();
let has_geom = arrow_schema.index_of("geometry").is_ok();
let has_refs = arrow_schema.index_of("node_refs").is_ok();
// The DELTA 3 layout facts, off the same footer read — no extra open, no decode.
// `is_band_first` is the PACKER's predicate, not a copy of it.
let packed = arrow_schema.index_of(crate::metadata::BBOX_COLUMN).is_ok();
let taggable = arrow_schema.index_of("tags").is_ok();
let meta = builder.metadata().clone();
let rows = meta.file_metadata().num_rows() as usize;
let n_rg = meta.num_row_groups();
// --- geometry null/set: metadata fast-path (footer statistics, no decode) ---
// Locate the geometry leaf column(s) in the parquet schema by root path part.
let mut geom_null: usize = 0;
let mut geom_stats_ok = true;
if has_geom {
for rg in meta.row_groups() {
// geometry is a single Binary leaf; sum any column chunk rooted at "geometry".
let mut found = false;
for col in rg.columns() {
let parts = col.column_path().parts();
if parts.first().map(|s| s.as_str()) == Some("geometry") {
found = true;
match col.statistics().and_then(|s| s.null_count_opt()) {
Some(n) => geom_null += n as usize,
None => {
geom_stats_ok = false;
}
}
}
}
if !found {
geom_stats_ok = false;
}
if !geom_stats_ok {
break;
}
}
}
// --- node_refs: needs real data. Parallelize over row groups, project ONLY
// node_refs, one reader per worker, no barrier — final sum is the only join.
//
// **Row 22: the ids may live BESIDE the table.** When the served table has no
// `node_refs` column, the same count is taken from `<name>.refs.parquet`, the
// sidecar the packer moved it into. Counting zero because the column is absent
// would report a table with no member ids as healthy, which is exactly the
// silent state row 22 exists to end.
let sidecar = crate::spatial::node_refs_sidecar_path(path);
let (refs_nonempty, refs_total) = if has_refs && n_rg > 0 {
count_node_refs(path, &meta, n_rg)?
} else if !has_refs && sidecar.exists() {
let sb = ParquetRecordBatchReaderBuilder::try_new(File::open(&sidecar)?)?;
let smeta = sb.metadata().clone();
let s_rg = smeta.num_row_groups();
if s_rg > 0 {
count_node_refs(&sidecar, &smeta, s_rg)?
} else {
(0, 0)
}
} else {
(0, 0)
};
// If the footer lacked geometry statistics, never report wrong numbers:
// fall back to a (still column-projected, parallel) decode of geometry only.
let geom_set;
if has_geom {
if geom_stats_ok {
geom_set = rows - geom_null;
} else {
let (g_null, g_set) = count_geometry(path, &meta, n_rg)?;
geom_null = g_null;
geom_set = g_set;
}
} else {
geom_set = 0;
geom_null = 0;
}
Ok(VerifyCounts {
rows,
has_geom,
geom_set,
geom_null,
has_refs,
refs_nonempty,
refs_total,
packed,
banded: crate::spatial::is_band_first(&meta),
taggable,
row_groups: n_rg,
})
}
/// Count `node_refs` across every row group, ROOT LAW #0 style: gatling fan-out,
/// one unit per ROW GROUP, LPT-scheduled heaviest-first by the group's compressed
/// byte size. Each unit opens its OWN reader projected to ONLY `node_refs` over
/// its single group and counts locally; the sum is the only join.
///
/// The old shape was a hand-rolled `std::thread::scope` pool over a **round-robin
/// `rg % n_workers` partition** — a static carve-up that could not rebalance: row
/// groups in an osm-katana table differ several-fold in compressed size (a group
/// full of long `node_refs` lists against a group of bare nodes), and whichever
/// worker drew the fat ones held the whole tail alone. One unit per group means a
/// worker that draws a cheap group immediately claims the next, and heaviest-first
/// keeps the fattest group from being the last thing started.
fn count_node_refs(
path: &Path,
meta: &Arc<ParquetMetaData>,
n_rg: usize,
) -> anyhow::Result<(usize, usize)> {
let results = gatling::gatling_forkjoin::gatling_for_each_balanced(
n_rg,
0,
1,
|rg| meta.row_group(rg).compressed_size().max(0) as u64,
|rg| count_refs_for_row_groups(path, meta, std::slice::from_ref(&rg)),
);
let mut nonempty = 0usize;
let mut total = 0usize;
for r in results {
let (ne, t) = r?;
nonempty += ne;
total += t;
}
Ok((nonempty, total))
}
fn count_refs_for_row_groups(
path: &Path,
meta: &Arc<ParquetMetaData>,
rgs: &[usize],
) -> anyhow::Result<(usize, usize)> {
use arrow::array::{Array, ListArray};
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
use std::fs::File;
// Reuse the already-parsed footer; project ONLY node_refs (never touch
// the 57 GB geometry/tags columns just to count refs).
let arm = ArrowReaderMetadata::try_new(meta.clone(), ArrowReaderOptions::new())?;
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(File::open(path)?, arm);
let mask = ProjectionMask::columns(builder.parquet_schema(), ["node_refs"]);
let reader = builder
.with_row_groups(rgs.to_vec())
.with_projection(mask)
.build()?;
let mut nonempty = 0usize;
let mut total = 0usize;
for batch in reader {
let b = batch?;
// node_refs is the only projected column → index 0 in the output schema.
if let Some(l) = b.column(0).as_any().downcast_ref::<ListArray>() {
let offsets = l.value_offsets();
for i in 0..l.len() {
if l.is_valid(i) {
let len = (offsets[i + 1] - offsets[i]) as usize;
total += len;
if len > 0 {
nonempty += 1;
}
}
}
}
}
Ok((nonempty, total))
}
/// Fallback (only when the footer lacks geometry statistics): row-group-parallel,
/// geometry-only projected decode. Never silently report wrong numbers.
///
/// Same gatling shape as [`count_node_refs`] — one unit per row group, LPT by the
/// group's compressed size, sum-reduced. (Was the same round-robin
/// `std::thread::scope` pool.)
fn count_geometry(
path: &Path,
meta: &Arc<ParquetMetaData>,
n_rg: usize,
) -> anyhow::Result<(usize, usize)> {
let results = gatling::gatling_forkjoin::gatling_for_each_balanced(
n_rg,
0,
1,
|rg| meta.row_group(rg).compressed_size().max(0) as u64,
|rg| count_geom_for_row_groups(path, meta, std::slice::from_ref(&rg)),
);
let mut null = 0usize;
let mut set = 0usize;
for r in results {
let (n, s) = r?;
null += n;
set += s;
}
Ok((null, set))
}
fn count_geom_for_row_groups(
path: &Path,
meta: &Arc<ParquetMetaData>,
rgs: &[usize],
) -> anyhow::Result<(usize, usize)> {
use arrow::array::Array;
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
use std::fs::File;
let arm = ArrowReaderMetadata::try_new(meta.clone(), ArrowReaderOptions::new())?;
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(File::open(path)?, arm);
let mask = ProjectionMask::columns(builder.parquet_schema(), ["geometry"]);
let reader = builder
.with_row_groups(rgs.to_vec())
.with_projection(mask)
.build()?;
let mut null = 0usize;
let mut set = 0usize;
for batch in reader {
let b = batch?;
let col = b.column(0);
let n = col.null_count();
null += n;
set += col.len() - n;
}
Ok((null, set))
}
// ── GEOMETRY QUALITY — the census `verify --geometry-quality` prints ─────────────
//
// The v2.1 map lane's question (2026-09-02): when the client draws a tangle, is the
// GEOMETRY bad at the source or is the reader wrong? The answer needs a census of the
// shapes actually in the tables — per geometry column, rung columns included, with
// books that close (the census.rs lesson: a ratio over an unbalanced population is the
// wrong-answer machine).
//
// What is counted, per column:
// * decode_fail — blobs `shared::geometry::decode_wkb` refuses. Any of these is a
// FAULT: the readers downstream will drop or mis-draw them.
// * points / lines / polys — the shape histogram.
// * unclosed_ring — a polygon ring (outer or hole) whose first and last vertex do
// not coincide. A FAULT: `is_closed_ring` gates fill, so an unclosed ring strokes
// as an outline instead of filling (half of the z12.3 water tangle).
// * ring_under_4 — a "ring" with fewer than 4 vertices: cannot fill, cannot extrude.
// * zero_area — a closed outer ring whose shoelace area is exactly 0 (a collapsed
// area: still costs its row and draws nothing).
// * self_intersecting — rings whose segments cross (bowtie). Checked exactly for
// rings of ≤ [`SELF_INTERSECT_MAX_SEGS`] segments (O(n²) with early exit), larger
// rings are counted in `big_ring_unchecked` instead of silently passing.
// Douglas–Peucker CAN legitimately introduce these at coarse rungs, so they are
// REPORTED rather than gating — the report is what says whether a rung's epsilon
// needs the area rule tightened.
//
// The books: null + decode_fail + points + lines + polys == rows, printed per column;
// an unbalanced census is itself a FAULT.
/// Above this many segments a ring's exact O(n²) self-intersection check is skipped
/// and the ring is counted `big_ring_unchecked` — never silently passed. 512 segments
/// covers >99 % of rung-column rings (they are DP output at a pixel epsilon).
pub const SELF_INTERSECT_MAX_SEGS: usize = 512;
/// One geometry column's quality census. See the section header for what each count
/// means and which ones are faults.
#[derive(Clone, Debug, Default)]
pub struct GeomQuality {
pub column: String,
pub rows: usize,
pub null: usize,
pub decode_fail: usize,
pub points: usize,
pub lines: usize,
pub polys: usize,
pub unclosed_ring: usize,
pub ring_under_4: usize,
pub zero_area: usize,
pub self_intersecting: usize,
pub big_ring_unchecked: usize,
}
impl GeomQuality {
/// The rows this census could not place — MUST be 0 or the census is wrong.
#[must_use]
pub fn unaccounted(&self) -> isize {
self.rows as isize
- (self.null + self.decode_fail + self.points + self.lines + self.polys) as isize
}
/// Is this column RED? Decode failures and unclosed rings break the readers;
/// unbalanced books break the census itself. Self-intersections and collapsed
/// areas are REPORTED (DP can legitimately produce them at coarse epsilons) —
/// they inform the pack rules rather than fail the verb.
#[must_use]
pub fn fault(&self) -> bool {
self.decode_fail > 0 || self.unclosed_ring > 0 || self.unaccounted() != 0
}
fn add_ring(&mut self, ring: &[(f64, f64)], is_outer: bool) {
let closed = ring.len() >= 2 && {
let (a, b) = (ring[0], ring[ring.len() - 1]);
(a.0 - b.0).abs() < 1e-9 && (a.1 - b.1).abs() < 1e-9
};
if !closed {
self.unclosed_ring += 1;
}
if ring.len() < 4 {
self.ring_under_4 += 1;
}
if is_outer && closed && ring.len() >= 4 && shoelace_area(ring) == 0.0 {
self.zero_area += 1;
}
let segs = ring.len().saturating_sub(1);
if segs > SELF_INTERSECT_MAX_SEGS {
self.big_ring_unchecked += 1;
} else if ring_self_intersects(ring) {
self.self_intersecting += 1;
}
}
}
/// Twice the signed shoelace area (sign irrelevant here — only ==0 is asked).
fn shoelace_area(ring: &[(f64, f64)]) -> f64 {
let mut s = 0.0;
for w in ring.windows(2) {
s += w[0].0 * w[1].1 - w[1].0 * w[0].1;
}
s
}
/// Do any two non-adjacent segments of `ring` properly cross? Exact O(n²) with early
/// exit — callers cap the ring size ([`SELF_INTERSECT_MAX_SEGS`]). Shared endpoints
/// (adjacent segments, and the closing seam) are not crossings.
///
/// The ONE writer of "does this ring cross itself" (LAW 5): the census counts on it,
/// `tiles::simplify_polys` retries a Douglas–Peucker pass on it, and the pack's own
/// bounded test ([`ring_self_intersects_bounded`]) is this predicate under a budget. A
/// second implementation is how any two of them would come to disagree about the same
/// ring — which is why it is `pub`: a TEST that needs to know whether its fixture is a
/// clean ring must ask this and not write its own.
#[must_use]
pub fn ring_self_intersects(ring: &[(f64, f64)]) -> bool {
let n = ring.len().saturating_sub(1); // segments
if n < 4 {
return false;
}
for i in 0..n {
// j starts past the neighbour; the (0, n-1) pair shares the closing vertex.
for j in (i + 2)..n {
if i == 0 && j == n - 1 {
continue;
}
if segments_properly_cross(ring, i, j) {
return true;
}
}
}
false
}
/// **Do segments `i` and `j` of `ring` properly cross?** The ONE orientation predicate
/// (LAW 5): the exact census walk above and the pack's bounded sweep below must not be
/// able to disagree about a pair, so there is one of these and both call it.
///
/// "Properly" — a shared endpoint is not a crossing, and neither is a collinear overlap:
/// both need a strict sign change on both sides, which a zero determinant denies.
#[inline]
fn segments_properly_cross(ring: &[(f64, f64)], i: usize, j: usize) -> bool {
let cross = |o: (f64, f64), a: (f64, f64), b: (f64, f64)| -> f64 {
(a.0 - o.0) * (b.1 - o.1) - (a.1 - o.1) * (b.0 - o.0)
};
let (a, b) = (ring[i], ring[i + 1]);
let (c, d) = (ring[j], ring[j + 1]);
let d1 = cross(a, b, c);
let d2 = cross(a, b, d);
let d3 = cross(c, d, a);
let d4 = cross(c, d, b);
((d1 > 0.0 && d2 < 0.0) || (d1 < 0.0 && d2 > 0.0))
&& ((d3 > 0.0 && d4 < 0.0) || (d3 < 0.0 && d4 > 0.0))
}
/// **The same question as [`ring_self_intersects`], under a DECLARED bound** — the pack's
/// entry point (row 24).
///
/// # The algorithm, and its bound
///
/// A left-to-right sweep on the segments' x-intervals. Two segments that do not overlap
/// in x cannot cross, so only the segments whose interval spans the current segment's
/// left edge are candidates: sort the segment indices by `min x` (in place — `sort_
/// unstable_by` allocates nothing), walk them, and hold the still-open ones in an active
/// list, retiring each as the sweep passes its `max x`. The candidate set is exact — no
/// crossing pair is ever skipped — so a `Clean` or `Crosses` verdict here is the same
/// verdict the exact walk gives, and `bounded_agrees_with_exact` pins that.
///
/// **The bound: `O(n log n)` comparisons for the sort, at most `n` retirement steps over
/// the whole sweep (each segment is retired once), and at most
/// `max(`[`SELFX_PACK_PAIR_FLOOR`]`, n *`[`SELFX_PACK_PAIRS_PER_SEG`]`)` orientation
/// tests** — after which it stops and answers [`SelfxVerdict::Unknown`]. Nothing here is
/// quadratic in the ring, and nothing about the input can make it so.
///
/// For a simple closed curve — every real shore ring — the active list is bounded by the
/// segments that span one longitude, and the whole thing is the sort. **The measured
/// per-segment cost and the budget it is compared against live on
/// [`SELFX_PACK_PAIRS_PER_SEG`], and are NOT repeated here.**
///
/// They were repeated here, and they had drifted: this line said *"2.0 candidate pairs
/// per segment against a budget of 32"* while the constant two hundred lines down said
/// **75.5** against **256**, both in this file, both claiming to be measured on the same
/// fixture. Neither reader can tell which is current, which is the whole defect —
/// `Step::named`, `short_type` and the four operator-facing roster strings are the same
/// lesson in this constellation, three times over. One writer; the link is the citation.
///
/// # Why `Unknown` is not a pass
///
/// The census's exact walk caps at [`SELF_INTERSECT_MAX_SEGS`] and counts what it skipped
/// (`big_ring_unchecked`) rather than passing it. This holds the same line by making
/// "I did not check" a value the type system will not let a caller ignore: the pack ships
/// the SOURCE ring on `Unknown` exactly as it does on `Crosses`.
///
/// `ring` is expected closed (first vertex repeated last), like every other ring here.
/// A NaN coordinate makes the sort order unspecified but cannot panic and cannot turn a
/// crossing into `Clean`: the orientation tests below reject NaN determinants anyway.
#[must_use]
pub fn ring_self_intersects_bounded(
ring: &[(f64, f64)],
s: &mut SelfxScratch,
) -> (SelfxVerdict, usize) {
let n = ring.len().saturating_sub(1); // segments
if n > SELFX_PACK_MAX_SEGS {
return (SelfxVerdict::Unknown, 0);
}
selfx_sweep(ring, s, SELFX_PACK_PAIR_FLOOR.max(n.saturating_mul(SELFX_PACK_PAIRS_PER_SEG)))
}
/// The sweep itself, with the budget as an argument and the pairs actually spent as a
/// second return — the ONE implementation (LAW 5). [`ring_self_intersects_bounded`] is
/// this at the shipped budget; `selfx_budget_survey` runs it at `usize::MAX` to MEASURE
/// what real rings cost, which is how [`SELFX_PACK_PAIRS_PER_SEG`] was chosen rather
/// than guessed.
fn selfx_sweep(
ring: &[(f64, f64)],
s: &mut SelfxScratch,
mut budget: usize,
) -> (SelfxVerdict, usize) {
let n = ring.len().saturating_sub(1); // segments
if n < 4 {
return (SelfxVerdict::Clean, 0);
}
let budget0 = budget;
let SelfxScratch { order, active, lo, hi } = s;
lo.clear();
hi.clear();
order.clear();
active.clear();
lo.reserve(n);
hi.reserve(n);
order.reserve(n);
for i in 0..n {
let (a, b) = (ring[i], ring[i + 1]);
lo.push(a.0.min(b.0));
hi.push(a.0.max(b.0));
#[allow(clippy::cast_possible_truncation)]
order.push(i as u32);
}
// In place: `sort_unstable_by` is pattern-defeating quicksort and allocates nothing,
// which `sort_by` (a merge sort with a scratch buffer) would.
order.sort_unstable_by(|&x, &y| {
lo[x as usize].partial_cmp(&lo[y as usize]).unwrap_or(std::cmp::Ordering::Equal)
});
for oi in 0..order.len() {
let si = order[oi];
let i = si as usize;
let x0 = lo[i];
// Retire and test in ONE pass. A retirement is not charged to the budget because
// a segment is retired at most once in the whole sweep — `n` steps in total, not
// `n` per segment; charging them would make the budget describe the wrong thing.
let mut w = 0usize;
for k in 0..active.len() {
let sj = active[k];
let j = sj as usize;
if hi[j] < x0 {
continue; // its x-interval closed before this one opened
}
active[w] = sj;
w += 1;
let d = i.abs_diff(j);
if d <= 1 || d == n - 1 {
continue; // adjacent, or the pair that shares the closing vertex
}
if budget == 0 {
return (SelfxVerdict::Unknown, budget0);
}
budget -= 1;
// The y half of the bbox reject. The sweep's active list already says the two
// x-intervals overlap; two vertical compares here throw out the pairs that
// are merely at the same longitude, which on a big ring is most of them — the
// segments along a shore's north and south flanks all span the same x. It is
// still charged to the budget: the budget bounds the WORK, and this is the
// work. It just makes each unit of it four multiplies cheaper.
let (c, d) = (ring[j], ring[j + 1]);
let (a, b) = (ring[i], ring[i + 1]);
if a.1.min(b.1) > c.1.max(d.1) || c.1.min(d.1) > a.1.max(b.1) {
continue;
}
if segments_properly_cross(ring, i, j) {
return (SelfxVerdict::Crosses, budget0 - budget);
}
}
active.truncate(w);
active.push(si);
}
(SelfxVerdict::Clean, budget0 - budget)
}
// ── The PACK's bounded crossing test (row 24) ────────────────────────────────────
//
// The census above walks rings that are already DP output at a pixel epsilon, so its
// exact O(n²) with a 512-segment cap is the right instrument there. The PACK is a
// different question: `tiles::simplify_polys` runs the same test on a ring it has just
// simplified, and a z12 Vänern ring is tens of thousands of vertices — exact O(n²) with
// no bound is one adversarial ring holding a worker for seconds, in the hot loop of
// every rung of every pack.
//
// So the pack gets its own entry point with a DECLARED bound, on caller-owned scratch,
// and a third answer: `Unknown`. See [`ring_self_intersects_bounded`].
/// Above this many segments the pack does not attempt the crossing test at all: the
/// simplification is not shipped either, the SOURCE ring goes instead.
///
/// 65 536 covers every real ring in the Sweden cut — the largest single ring in the
/// live relations table is Vänern's shore at ~34 000 vertices — so the cap is a guard
/// against pathological input, not a rule the pack meets daily. It also bounds the
/// scratch: four vectors of one word per segment, 2 MiB at the cap.
pub const SELFX_PACK_MAX_SEGS: usize = 65_536;
/// **The pair-test budget, per segment.** [`ring_self_intersects_bounded`] examines at
/// most `max(`[`SELFX_PACK_PAIR_FLOOR`]`, SELFX_PACK_PAIRS_PER_SEG * segs)` candidate
/// segment pairs before giving up with [`SelfxVerdict::Unknown`].
///
/// **MEASURED, not guessed** (`the_budget_is_generous_for_the_rings_the_pack_meets`, this
/// box): a 50 000-segment star-shaped ripple — the shape a big shore ring presents to the
/// sweep — costs **75.5 candidate pairs per segment**. Not the small constant a first
/// reading suggests: a vertical sweep line near a ring's east or west extreme meets the
/// whole near-horizontal flank at once, and every one of those segments spans that
/// longitude. 256 is 3.4× that measurement.
///
/// The worst case this buys is worth stating plainly: at the segment cap it is
/// 16 777 216 candidate pairs, and a candidate costs ~1.4 ns here (a y-interval reject,
/// four multiplies only when that passes) — about 24 ms for one check, 120 ms for a ring
/// that walks the whole retry ladder. The unbounded test it replaces was 2.1 × 10⁹ pairs
/// (~3 s) per attempt at the same size, and had no answer at all above it.
pub const SELFX_PACK_PAIRS_PER_SEG: usize = 256;
/// The budget floor. A ring of `n` segments has at most `n²/2` non-adjacent pairs, so
/// this floor answers **every ring up to 362 segments EXACTLY even in its worst case**
/// (362²/2 = 65 522) — the per-segment rate only starts to bind above 2 048 segments,
/// where a linear budget is what makes the whole test O(n log n).
pub const SELFX_PACK_PAIR_FLOOR: usize = 65_536;
/// What [`ring_self_intersects_bounded`] could establish.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SelfxVerdict {
/// No two non-adjacent segments cross. Proven, within the budget.
Clean,
/// Two non-adjacent segments cross. Proven — a witness was found.
Crosses,
/// The budget ran out first. NOT a pass: the caller must treat this exactly as it
/// treats `Crosses`, because an unchecked ring is a bowtie nobody can see.
Unknown,
}
/// **The sweep's working memory** — one per worker, inside `tiles::DpScratch`, reused
/// across every ring of every retry of every row (row 24, `tilemem-2`'s rule).
///
/// Four parallel arrays over the ring's segments, all overwritten each call: the x
/// interval of each segment (`lo`/`hi`), the segment indices in ascending `lo`
/// (`order`), and the sweep's active set (`active`).
#[derive(Default)]
pub struct SelfxScratch {
order: Vec<u32>,
active: Vec<u32>,
lo: Vec<f64>,
hi: Vec<f64>,
}
/// **Census every geometry column of one file** — `geometry` plus every
/// `geometry_z*` rung column present. Row-group-parallel (gatling, LPT by
/// compressed size — ROOT LAW #0), each unit decoding ONLY its column slice; the
/// sum is the only join.
pub fn geometry_quality(path: &Path) -> anyhow::Result<Vec<GeomQuality>> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use std::fs::File;
let builder = ParquetRecordBatchReaderBuilder::try_new(File::open(path)?)?;
let schema = builder.schema().clone();
let meta = builder.metadata().clone();
let n_rg = meta.num_row_groups();
let cols: Vec<String> = schema
.fields()
.iter()
.map(|f| f.name().clone())
.filter(|n| n == "geometry" || n.starts_with("geometry_z"))
.collect();
let mut out = Vec::with_capacity(cols.len());
for col in &cols {
let results = gatling::gatling_forkjoin::gatling_for_each_balanced(
n_rg.max(1),
0,
1,
|rg| meta.row_group(rg.min(n_rg.saturating_sub(1))).compressed_size().max(0) as u64,
|rg| quality_for_row_group(path, &meta, col, rg),
);
let mut q = GeomQuality { column: col.clone(), ..Default::default() };
for r in results {
let part = r?;
q.rows += part.rows;
q.null += part.null;
q.decode_fail += part.decode_fail;
q.points += part.points;
q.lines += part.lines;
q.polys += part.polys;
q.unclosed_ring += part.unclosed_ring;
q.ring_under_4 += part.ring_under_4;
q.zero_area += part.zero_area;
q.self_intersecting += part.self_intersecting;
q.big_ring_unchecked += part.big_ring_unchecked;
}
out.push(q);
}
Ok(out)
}
fn quality_for_row_group(
path: &Path,
meta: &Arc<ParquetMetaData>,
col: &str,
rg: usize,
) -> anyhow::Result<GeomQuality> {
use arrow::array::{Array, BinaryArray, LargeBinaryArray};
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
use std::fs::File;
let arm = ArrowReaderMetadata::try_new(meta.clone(), ArrowReaderOptions::new())?;
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(File::open(path)?, arm);
let mask = ProjectionMask::columns(builder.parquet_schema(), [col]);
let reader = builder.with_row_groups(vec![rg]).with_projection(mask).build()?;
let mut q = GeomQuality::default();
for batch in reader {
let b = batch?;
let c = b.column(0);
// One iterator, two Binary widths — the tables write plain Binary today, but a
// census that silently read zero rows off a LargeBinary rewrite would be the
// flat-copy bug again.
let rows: Box<dyn Iterator<Item = Option<&[u8]>> + '_> =
if let Some(a) = c.as_any().downcast_ref::<BinaryArray>() {
Box::new((0..a.len()).map(move |i| (!a.is_null(i)).then(|| a.value(i))))
} else if let Some(a) = c.as_any().downcast_ref::<LargeBinaryArray>() {
Box::new((0..a.len()).map(move |i| (!a.is_null(i)).then(|| a.value(i))))
} else {
anyhow::bail!("column `{col}` is not Binary — the census cannot read it");
};
for bytes in rows {
q.rows += 1;
let Some(bytes) = bytes else {
q.null += 1;
continue;
};
match crate::shared::geometry::decode_wkb(bytes) {
None => q.decode_fail += 1,
Some(crate::shared::geometry::WkbShape::Point(_)) => q.points += 1,
Some(crate::shared::geometry::WkbShape::Lines(_)) => q.lines += 1,
Some(crate::shared::geometry::WkbShape::Polys(ps)) => {
q.polys += 1;
for (outer, holes) in &ps {
q.add_ring(outer, true);
for h in holes {
q.add_ring(h, false);
}
}
}
}
}
}
Ok(q)
}
// ── THE GATE — the census, IN the chain (katana-data lane row 39) ────────────────
//
// The census above has been printable since lane V21. It has never been ASKABLE by the
// thing that produces the cut. korp-collectors' `osm_cut_verify::check_extract` — the
// chain's post-condition step — asserts foreign nodes, wholly-foreign ways, WKB shape
// and presence: four questions about WHERE a geometry is, none about whether it is a
// shape at all. The 11 510 bowties in the relation rungs were found by a human running
// the CLI, after a chain run that reported success.
//
// ## Why a BASELINE and not a zero
//
// `ways::geometry` on the live Sweden cut carries 99 unclosed rings, 599
// self-crossing rings and 28 zero-area outers (MEASURED 2026-09-02; see
// [`GeometryBaseline::RECORDED`]). A gate at zero fails that cut on its first run, and
// a gate that fails on today's real artefact is a gate somebody switches off — which is
// not a hypothetical here: it is exactly how the two dead gates this row also names
// (`refused_placements`, `tiles_per_zoom`) came to be a constant pass and a field that
// no longer exists. So the threshold is **no NEW defects against a record**: the record
// is in this file, with its provenance, and a count ABOVE it fails. The budget can only
// be spent down.
//
// ## What is never budgeted
//
// * `decode_fail` — a blob no reader can decode is a row that will be dropped or
// mis-drawn downstream, and there have never been any. Budget 0, everywhere.
// * `unaccounted` — books that do not close mean the census stopped counting, and a
// budget over a number that is itself wrong is worse than no gate. Refused under
// EVERY baseline, [`GeometryBaseline::ZERO`] and [`GeometryBaseline::RECORDED`] alike.
//
// ## What is REPORTED and not gated, on purpose
//
// `zero_area` and `ring_under_4` cost a row and draw nothing, but they mis-draw
// nothing either. They travel in [`GeometryVerdict::census`] so the numbers are in
// front of whoever reads a refusal, and they gate nothing.
/// **What a column is allowed to carry** — a recorded count, or an admission that the
/// count has never been recorded.
///
/// `Unmeasured` exists because the alternative is a lie in one direction or the other:
/// budgeting a column at 0 when nobody ever counted it makes the chain refuse a cut
/// over a number the record cannot support, and budgeting it at ∞ silently hides it.
/// It fails nothing and it says so, in the same UNASSERTED idiom korp-collectors'
/// `check_extract` already uses for `geography_asserted`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Budget {
/// Recorded: this many were counted, and one more is a NEW defect.
At(usize),
/// Never censused on a real cut. Reported, never a failure — and named as
/// unmeasured wherever the verdict is printed.
Unmeasured,
}
impl Budget {
/// The recorded count, or `None` when this column was never censused.
#[must_use]
pub fn recorded(self) -> Option<usize> {
match self {
Self::At(n) => Some(n),
Self::Unmeasured => None,
}
}
/// `Some(allowed)` when `counted` is ABOVE the record — i.e. a NEW defect.
#[must_use]
pub fn exceeded_by(self, counted: usize) -> Option<usize> {
match self {
Self::At(n) if counted > n => Some(n),
_ => None,
}
}
}
/// One `(table, column)` pair's recorded defect budget.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeometryAllowance {
pub table: &'static str,
pub column: &'static str,
pub unclosed_ring: Budget,
pub self_intersecting: Budget,
/// Always [`Budget::At`]`(0)` in every baseline shipped here — see the section
/// header. The field exists so a future record cannot express "tolerate a blob no
/// reader can decode" by accident; it has to be typed out.
pub decode_fail: Budget,
}
impl GeometryAllowance {
/// The budget a column the record does not name gets: nothing. An unrecorded column
/// carrying a defect is a NEW defect — that is what "no new defects" means, and it
/// is what makes a rung column arriving in a future pack get looked at.
const fn clean(table: &'static str, column: &'static str) -> Self {
Self {
table,
column,
unclosed_ring: Budget::At(0),
self_intersecting: Budget::At(0),
decode_fail: Budget::At(0),
}
}
const fn record(
table: &'static str,
column: &'static str,
unclosed_ring: usize,
self_intersecting: Budget,
) -> Self {
Self {
table,
column,
unclosed_ring: Budget::At(unclosed_ring),
self_intersecting,
decode_fail: Budget::At(0),
}
}
}
/// **A recorded defect budget for a whole cut, and where the record came from.**
///
/// The record is Rust, not a data file beside it, for the reason the workspace states
/// about build variants: the CODE is the source. A baseline in a `.toml` drifts away
/// from the census that produced it and from the gate that reads it; a `const` cannot
/// be edited without the diff landing next to the provenance that justifies it.
///
/// `PartialEq` because a consumer holds a `&'static GeometryBaseline` inside its own
/// declaration struct (korp-collectors' `CutExpectations`), and a declaration that
/// cannot be compared cannot be tested for "the config said what I think it said".
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeometryBaseline {
/// Named in every refusal, so a green under one baseline can never be read as a
/// green under another.
pub name: &'static str,
/// WHERE the numbers come from — the cut, the date, the command, the commit.
pub provenance: &'static str,
pub allow: &'static [GeometryAllowance],
}
impl GeometryBaseline {
/// **Nothing is tolerated.** What a fresh cut off the fixed encoder must pass, and
/// what the fixture cuts in this repo's tests are held to.
pub const ZERO: Self = Self {
name: "ZERO",
provenance: "no defect is budgeted — every count must be 0",
allow: &[],
};
/// **The live Sweden cut, censused 2026-09-02.**
///
/// Provenance: `osm-katana verify --geometry-quality` on
/// `/home/korp/.korp/work/sweden` (oden) after the ring-safe-DP re-pack was swapped
/// in at 13:10, zoomies `6d54ac0`; written up in korp `.nornir/map-v2-design.md`
/// §V2.1 and carried as the `rung ring quality` row of `korp-v2.md`. The base
/// column's numbers — 99 unclosed / 599 selfx / 28 zero-area in `ways::geometry` —
/// come from CONVERT, not from the pack, which is why a re-pack could not move them.
///
/// **This record is expected to shrink to [`ZERO`](Self::ZERO).** The encoder that
/// wrote the 99 open rings was fixed in zoomies `10e63b8`
/// ([`crate::shared::gpu_backend`]'s `wkb_polygon` closes the ring at encode now),
/// so the 99 die with the next re-CUT and this record must be re-taken and lowered
/// then. It is a debt with a number on it, not a permission.
///
/// MEASURED on t14s 2026-09-06 for contrast: a small real cut off the fixed
/// encoder (`~/scratch/korp-gg20/cut`, 192 ways / 64 nodes / 24
/// resolved relations, 9 geometry columns) censuses **0** on every counter — the
/// budget below is history, not the shape of new output.
pub const RECORDED: Self = Self {
name: "sweden-2026-09-02",
provenance: "osm-katana verify --geometry-quality on the live Sweden cut \
(oden /home/korp/.korp/work/sweden) after the ring-safe-DP re-pack \
swap 13:10, zoomies 6d54ac0; korp .nornir/map-v2-design.md §V2.1",
allow: &[
// The base column, straight out of convert. The one the census went red on.
GeometryAllowance::record("ways.parquet", "geometry", 99, Budget::At(599)),
// The rungs, AFTER the ring-safe DP re-pack (was 4 153 / 14 682 / 2 118).
GeometryAllowance::record("ways.parquet", "geometry_z6", 0, Budget::At(4)),
GeometryAllowance::record("ways.parquet", "geometry_z9", 0, Budget::At(19)),
GeometryAllowance::record("ways.parquet", "geometry_z12", 0, Budget::At(92)),
// Relations' rungs (was 8 360 / 11 510 / 496 — the wedge census).
GeometryAllowance::record(
"relations.resolved.parquet",
"geometry_z6",
0,
Budget::At(0),
),
GeometryAllowance::record(
"relations.resolved.parquet",
"geometry_z9",
0,
Budget::At(4),
),
GeometryAllowance::record(
"relations.resolved.parquet",
"geometry_z12",
0,
Budget::At(24),
),
// The relations BASE column: its `unclosed`/`decode_fail` are known 0 (the
// census FAILS on those, and it reported only `ways::geometry` red), but its
// selfx count was never written down. Recorded as unmeasured rather than
// guessed at 0 — a chain must not refuse a cut over a number nobody took.
GeometryAllowance::record(
"relations.resolved.parquet",
"geometry",
0,
Budget::Unmeasured,
),
// Points carry no rings; 0 is not a budget here, it is arithmetic.
GeometryAllowance::clean("nodes.parquet", "geometry"),
],
};
/// The budget for one column — [`GeometryAllowance::clean`] when the record does
/// not name it.
#[must_use]
pub fn allowance(&self, table: &str, column: &str) -> GeometryAllowance {
self.allow
.iter()
.find(|a| a.table == table && a.column == column)
.copied()
.unwrap_or(GeometryAllowance::clean("", ""))
}
/// **Every NEW defect this column carries.** Empty is the pass.
#[must_use]
pub fn findings(&self, table: &str, q: &GeomQuality) -> Vec<GeometryFinding> {
let a = self.allowance(table, &q.column);
let mut out = Vec::new();
let mut check = |defect: &'static str, budget: Budget, counted: usize| {
if let Some(allowed) = budget.exceeded_by(counted) {
out.push(GeometryFinding {
table: table.to_string(),
column: q.column.clone(),
defect,
counted,
allowed,
});
}
};
check("decode_fail", a.decode_fail, q.decode_fail);
check("unclosed_ring", a.unclosed_ring, q.unclosed_ring);
check("self_intersecting", a.self_intersecting, q.self_intersecting);
// Never budgeted, under any baseline: see the section header.
let un = q.unaccounted();
if un != 0 {
out.push(GeometryFinding {
table: table.to_string(),
column: q.column.clone(),
defect: "unaccounted",
counted: un.unsigned_abs(),
allowed: 0,
});
}
out
}
/// **What this baseline declined to assert about this column**, so an UNASSERTED
/// green is never printed as a measured one.
#[must_use]
pub fn notes(&self, table: &str, q: &GeomQuality) -> Vec<GeometryNote> {
[
("decode_fail", q.decode_fail),
("unclosed_ring", q.unclosed_ring),
("self_intersecting", q.self_intersecting),
]
.into_iter()
.zip({
let a = self.allowance(table, &q.column);
[a.decode_fail, a.unclosed_ring, a.self_intersecting]
})
.filter(|(_, b)| *b == Budget::Unmeasured)
.map(|((defect, counted), _)| GeometryNote {
table: table.to_string(),
column: q.column.clone(),
defect,
counted,
})
.collect()
}
}
/// One NEW defect: what, where, how many, and what the record allowed.
#[derive(Clone, Debug)]
pub struct GeometryFinding {
pub table: String,
pub column: String,
/// `decode_fail` | `unclosed_ring` | `self_intersecting` | `unaccounted`.
pub defect: &'static str,
pub counted: usize,
pub allowed: usize,
}
impl std::fmt::Display for GeometryFinding {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{}::{} {} {} (baseline allows {})",
self.table, self.column, self.defect, self.counted, self.allowed
)
}
}
/// A count the baseline declines to judge, carried so it is visible rather than absent.
#[derive(Clone, Debug)]
pub struct GeometryNote {
pub table: String,
pub column: String,
pub defect: &'static str,
pub counted: usize,
}
impl std::fmt::Display for GeometryNote {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{}::{} {} {} — UNASSERTED: this column's {} has never been censused on a \
real cut, so the baseline neither passes nor fails it",
self.table, self.column, self.defect, self.counted, self.defect
)
}
}
/// **The chain's geometry post-condition**: the census of every geometry column in a
/// cut, judged against a baseline.
#[derive(Debug)]
pub struct GeometryVerdict {
/// The baseline's [`GeometryBaseline::name`] — in every message.
pub baseline: &'static str,
/// `("<table>::<column>", census)` for every geometry column found, defect-free
/// ones included: a refusal must be readable next to the numbers that produced it.
pub census: Vec<(String, GeomQuality)>,
pub findings: Vec<GeometryFinding>,
pub notes: Vec<GeometryNote>,
pub scanned_ms: u128,
}
impl GeometryVerdict {
#[must_use]
pub fn ok(&self) -> bool {
self.findings.is_empty()
}
/// The refusal, with every finding's numbers in it — `None` exactly when
/// [`ok`](Self::ok).
#[must_use]
pub fn reason(&self) -> Option<String> {
if self.findings.is_empty() {
return None;
}
Some(format!(
"geometry quality: {} NEW defect(s) above baseline `{}` — {}",
self.findings.len(),
self.baseline,
self.findings
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("; ")
))
}
}
/// **Census every geometry column of every named table under `dir` and judge it.**
///
/// This is the entry point the cut chain calls — korp-collectors'
/// `osm_cut_verify::check_extract`. Absent tables are NOT reported here: that is
/// [`verify_dir`]'s [`Fault::Absent`], and a second writer of the same finding is how
/// two verdicts come to disagree (LAW 5).
pub fn check_cut_geometry(
dir: &Path,
tables: &[TableSpec],
baseline: &GeometryBaseline,
) -> anyhow::Result<GeometryVerdict> {
let started = std::time::Instant::now();
let mut v = GeometryVerdict {
baseline: baseline.name,
census: Vec::new(),
findings: Vec::new(),
notes: Vec::new(),
scanned_ms: 0,
};
for spec in tables {
let path = dir.join(spec.name);
if !path.exists() {
continue;
}
for q in geometry_quality(&path)? {
v.findings.extend(baseline.findings(spec.name, &q));
v.notes.extend(baseline.notes(spec.name, &q));
v.census.push((format!("{}::{}", spec.name, q.column), q));
}
}
v.scanned_ms = started.elapsed().as_millis();
Ok(v)
}
/// **Row 24 — the pack's bounded crossing test.**
///
/// | test | the defect it is red for |
/// |---|---|
/// | [`bounded_agrees_with_exact`] | a sweep that skips a candidate pair, so the pack passes a bowtie the census then counts |
/// | [`selfx_budget_survey`] | (measurement) the per-segment budget, chosen from what real rings cost rather than guessed |
#[cfg(test)]
mod selfx_bounded_tests {
use super::{
SELFX_PACK_PAIRS_PER_SEG, SelfxScratch, SelfxVerdict, ring_self_intersects,
ring_self_intersects_bounded, selfx_sweep,
};
/// A star-shaped ripple — simple by construction, and DP keeps its extrema, so it is
/// the shape a big shore ring presents to the sweep.
fn rippled(n: usize, r0: f64, amp: f64, m: f64) -> Vec<(f64, f64)> {
let mut v = Vec::with_capacity(n + 1);
for i in 0..n {
#[allow(clippy::cast_precision_loss)]
let t = (i as f64) / (n as f64) * std::f64::consts::TAU;
let r = amp.mul_add((m * t).sin(), r0);
v.push((r.mul_add(t.cos(), 13.2), r.mul_add(t.sin(), 58.9)));
}
v.push(v[0]);
v
}
/// A bowtie: a square whose two ends are swapped, so exactly one pair crosses.
fn bowtie() -> Vec<(f64, f64)> {
vec![(0.0, 0.0), (1.0, 1.0), (1.0, 0.0), (0.0, 1.0), (0.0, 0.0)]
}
/// **The bounded sweep and the exact walk answer the same question.** The sweep only
/// tests pairs whose x-intervals overlap; if that filter were ever wrong the pack
/// would start shipping rings the census counts as bowties, which is the whole defect
/// row 24 must not reintroduce. Run at an unlimited budget, so `Unknown` cannot hide
/// a disagreement.
#[test]
fn bounded_agrees_with_exact() {
let mut cases: Vec<Vec<(f64, f64)>> = vec![bowtie(), Vec::new(), vec![(0.0, 0.0)]];
for n in [4usize, 5, 8, 37, 200, 2_000] {
for m in [1.0_f64, 3.0, 7.0, 250.0] {
cases.push(rippled(n, 0.5, 1.0e-3, m));
cases.push(rippled(n, 0.5, 0.6, m)); // deep enough to bowtie itself
}
}
let mut s = SelfxScratch::default();
let mut crossing = 0;
for (k, r) in cases.iter().enumerate() {
let exact = ring_self_intersects(r);
let (v, _) = selfx_sweep(r, &mut s, usize::MAX);
crossing += usize::from(exact);
assert_eq!(
v,
if exact { SelfxVerdict::Crosses } else { SelfxVerdict::Clean },
"case {k} ({} vertices): the sweep and the exact walk disagree",
r.len(),
);
}
assert!(crossing > 0, "the corpus holds no crossing ring — it proves only half");
}
/// **`Unknown` is what an exhausted budget says, and the shipped budget answers the
/// rings the pack really meets.** Also a measurement: printed, so the constant can be
/// defended rather than asserted.
#[test]
fn the_budget_is_generous_for_the_rings_the_pack_meets() {
let mut s = SelfxScratch::default();
// A budget of one cannot settle a 2 000-segment ring.
let big = rippled(2_000, 0.5, 1.0e-3, 500.0);
assert_eq!(selfx_sweep(&big, &mut s, 1).0, SelfxVerdict::Unknown);
// At the shipped budget every one of these is answered.
for (tag, n, m) in [
("z12 Vänern-sized", 50_000usize, 12_500.0_f64),
("a big lake", 12_000, 3_000.0),
("a rung-column ring", 400, 100.0),
] {
let r = rippled(n, 0.5, 1.0e-3, m);
let (v, used) = selfx_sweep(&r, &mut s, usize::MAX);
let segs = r.len() - 1;
#[allow(clippy::cast_precision_loss)]
let per = used as f64 / segs as f64;
println!(
"row24 sweep {tag}: {segs} segs, {used} pair tests, {per:.2}/seg \
(budget {SELFX_PACK_PAIRS_PER_SEG}/seg), {v:?}",
);
assert_eq!(
ring_self_intersects_bounded(&r, &mut s).0,
v,
"{tag} ({segs} segs) needs {per:.2} pair tests per segment — over the \
shipped budget of {SELFX_PACK_PAIRS_PER_SEG}, so the pack would ship it \
unsimplified",
);
}
}
}
#[cfg(test)]
mod tests {
use super::verify;
use std::sync::Arc;
use arrow::array::{Array, BinaryBuilder, Int64Builder, ListBuilder};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
/// Build a small GeoParquet fixture with known geometry null/set counts and
/// known `node_refs` list contents, written across TWO row groups so the
/// row-group-parallel counter actually partitions work.
///
/// Layout (4 rows):
/// row0 geometry=b"AAA" node_refs=[1,2,3]
/// row1 geometry=NULL node_refs=[] (empty list, valid)
/// row2 geometry=b"BB" node_refs=[4,5]
/// row3 geometry=b"C" node_refs=NULL (null list)
///
/// ⇒ rows=4, geom_set=3, geom_null=1, refs_nonempty=2, refs_total=5.
fn write_fixture(path: &std::path::Path) {
let mut gb = BinaryBuilder::new();
gb.append_value(b"AAA");
gb.append_null();
gb.append_value(b"BB");
gb.append_value(b"C");
let geom = gb.finish();
let mut lb = ListBuilder::new(Int64Builder::new());
lb.values().append_value(1);
lb.values().append_value(2);
lb.values().append_value(3);
lb.append(true); // row0 [1,2,3]
lb.append(true); // row1 [] empty, valid
lb.values().append_value(4);
lb.values().append_value(5);
lb.append(true); // row2 [4,5]
lb.append(false); // row3 null list
let refs = lb.finish();
let schema = Schema::new(vec![
Field::new("geometry", DataType::Binary, true),
Field::new("node_refs", refs.data_type().clone(), true),
]);
let batch =
RecordBatch::try_new(Arc::new(schema), vec![Arc::new(geom), Arc::new(refs)]).unwrap();
// Force 2 row groups (2 rows each) so count_node_refs partitions.
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(2))
.build();
let file = std::fs::File::create(path).unwrap();
let mut w = ArrowWriter::try_new(file, batch.schema(), Some(props)).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
/// **RED-first (LAW 2): every quality counter fires on a fixture built to trip it,
/// and stays zero on a clean one.** Hand-built WKB, one defect per row, so a
/// counter that went blind names its row.
#[test]
fn the_geometry_quality_census_sees_each_defect_and_only_those() {
use super::{geometry_quality, GeomQuality};
fn poly(rings: &[&[(f64, f64)]], close: bool) -> Vec<u8> {
let mut b = vec![1u8];
b.extend(3u32.to_le_bytes());
b.extend((rings.len() as u32).to_le_bytes());
for ring in rings {
let mut r: Vec<(f64, f64)> = ring.to_vec();
if close {
if let (Some(&f), Some(&l)) = (r.first(), r.last()) {
if f != l {
r.push(f);
}
}
}
b.extend((r.len() as u32).to_le_bytes());
for (x, y) in &r {
b.extend(x.to_le_bytes());
b.extend(y.to_le_bytes());
}
}
b
}
let sq: &[(f64, f64)] = &[(0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)];
// A bowtie: closed, 4 segments, two properly crossing — and deliberately
// ASYMMETRIC so its shoelace area is non-zero: a symmetric bowtie has area 0
// and would double-count into `zero_area`, blurring the two counters.
let bowtie: &[(f64, f64)] = &[(0.0, 0.0), (1.0, 1.0), (1.0, 0.0), (0.0, 2.0)];
// A collapsed outer: closed, 4+ vertices, shoelace area exactly 0.
let flat: &[(f64, f64)] = &[(0.0, 0.0), (1.0, 0.0), (2.0, 0.0), (1.0, 0.0)];
let rows: Vec<Option<Vec<u8>>> = vec![
Some(poly(&[sq], true)), // clean polygon
Some(poly(&[sq], false)), // UNCLOSED outer ring
Some(b"\x01\x63\x00\x00\x00junk".to_vec()), // DECODE FAIL (type 99)
Some(poly(&[bowtie], true)), // SELF-INTERSECTING
Some(poly(&[flat], true)), // ZERO AREA
None, // null
Some(crate::shared::geometry::encode_wkb_point(15.0, 60.0)),
];
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("q.parquet");
{
let mut gb = BinaryBuilder::new();
for r in &rows {
match r {
Some(v) => gb.append_value(v),
None => gb.append_null(),
}
}
let geom = gb.finish();
let schema = Schema::new(vec![Field::new("geometry", DataType::Binary, true)]);
let batch =
RecordBatch::try_new(Arc::new(schema), vec![Arc::new(geom)]).unwrap();
let file = std::fs::File::create(&path).unwrap();
let mut w = ArrowWriter::try_new(file, batch.schema(), None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
let qs = geometry_quality(&path).unwrap();
assert_eq!(qs.len(), 1, "one geometry column censused");
let q: &GeomQuality = &qs[0];
assert_eq!(q.rows, 7);
assert_eq!(q.null, 1);
assert_eq!(q.decode_fail, 1, "the junk blob must be COUNTED, not skipped");
assert_eq!(q.points, 1);
assert_eq!(q.polys, 4);
assert_eq!(q.unclosed_ring, 1, "the open square must read UNCLOSED");
assert_eq!(q.self_intersecting, 1, "the bowtie must read SELF-INTERSECTING");
assert_eq!(q.zero_area, 1, "the collapsed outer must read AREA 0");
assert_eq!(q.unaccounted(), 0, "the books must close");
assert!(q.fault(), "decode_fail + unclosed make this column a FAULT");
// The clean twin: same writer, only the clean rows — every defect counter 0.
let clean = dir.path().join("clean.parquet");
{
let mut gb = BinaryBuilder::new();
gb.append_value(poly(&[sq], true));
gb.append_value(crate::shared::geometry::encode_wkb_point(15.0, 60.0));
let geom = gb.finish();
let schema = Schema::new(vec![Field::new("geometry", DataType::Binary, true)]);
let batch =
RecordBatch::try_new(Arc::new(schema), vec![Arc::new(geom)]).unwrap();
let file = std::fs::File::create(&clean).unwrap();
let mut w = ArrowWriter::try_new(file, batch.schema(), None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
let q = &geometry_quality(&clean).unwrap()[0];
assert!(
!q.fault()
&& q.unclosed_ring == 0
&& q.self_intersecting == 0
&& q.zero_area == 0
&& q.decode_fail == 0,
"a clean file must census clean: {q:?}",
);
}
#[test]
fn counts_geometry_and_node_refs_exactly() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("fixture.parquet");
write_fixture(&path);
let v = verify(&path).unwrap();
assert_eq!(v.rows, 4, "rows");
assert!(v.has_geom, "has_geom");
assert_eq!(v.geom_set, 3, "geom_set");
assert_eq!(v.geom_null, 1, "geom_null");
assert!(v.has_refs, "has_refs");
assert_eq!(v.refs_nonempty, 2, "refs_nonempty (rows with >0 refs)");
assert_eq!(v.refs_total, 5, "refs_total (sum of all list lengths)");
}
#[test]
fn absent_columns_report_zero() {
// A parquet with neither `geometry` nor `node_refs`.
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("bare.parquet");
let mut ib = Int64Builder::new();
ib.append_value(10);
ib.append_value(20);
let ids = ib.finish();
let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]);
let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(ids)]).unwrap();
let file = std::fs::File::create(&path).unwrap();
let mut w = ArrowWriter::try_new(file, batch.schema(), None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
let v = verify(&path).unwrap();
assert_eq!(v.rows, 2, "rows");
assert!(!v.has_geom, "has_geom must be false");
assert!(!v.has_refs, "has_refs must be false");
assert_eq!(v.geom_set, 0);
assert_eq!(v.geom_null, 0);
assert_eq!(v.refs_nonempty, 0);
assert_eq!(v.refs_total, 0);
}
}