osm-katana/src/spatial.rs
// Spatial layout for GeoParquet output — Hilbert row ordering + a `covering.bbox`
// column. See `.nornir/geoparquet-spatial-layout-design.md`.
//! **Make the file its own spatial index.**
//!
//! A Parquet file is read in pieces: a reader parses the footer, looks at each
//! row group's min/max statistics, and fetches only the column chunks of the row
//! groups that can possibly match. For that to prune anything, two things must
//! hold, and osm-katana's streaming convert output satisfied *neither*:
//!
//! 1. **Spatially-near rows must be physically adjacent.** Convert emits rows in
//! PBF/parse order, so any given row group spans the whole extract and every
//! row group's bounds are the file's bounds — nothing can ever be pruned.
//! [`hilbert_index`] gives each feature a 1-D key that preserves 2-D locality,
//! and sorting on it packs each row group into a compact region.
//!
//! 2. **There must be something prunable to compare.** Min/max statistics on the
//! WKB `geometry` column are *useless* for space: WKB compares
//! lexicographically, and a WKB point is
//! `[0x01][01 00 00 00][8 LE bytes of x][8 LE bytes of y]`, so the ordering is
//! dominated by the LOW-order mantissa byte of x. Hilbert sorting alone
//! therefore buys nothing a reader can see. [`bbox_field`] adds the
//! GeoParquet 1.1 `covering.bbox` struct column — four plain `f64`s whose
//! min/max statistics Parquet records natively and every engine understands.
//!
//! Both are needed; either alone is inert. The pair is what turns
//! "read when close to the point" from a wish into four byte-range requests.
//!
//! The covering column is **derived** — a pure function of `geometry` — so
//! [`crate::digest`] excludes it from the content digest and a Hilbert-packed
//! table still compares equal (multiset) to the convert output it came from.
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::Context as _;
use arrow::array::{
Array, ArrayRef, BinaryArray, BooleanArray, Float32Builder, Int8Array, StringArray,
StructArray, UInt8Array, UInt16Array,
};
use arrow::datatypes::{DataType, Field, Fields, Schema};
use arrow::record_batch::RecordBatch;
use parquet::arrow::ArrowWriter;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::file::properties::WriterProperties;
use crate::metadata::{self, BBOX_COLUMN};
/// Bits per axis in the Hilbert curve. 31 keeps the index inside 62 bits, so it
/// is always a non-negative `i64` (sortable as a signed key) while resolving
/// 360° / 2^31 ≈ 1.7e-7° ≈ **1.9 cm** of longitude — far finer than OSM's own
/// 1e-7° coordinate grid, so the ordering never collapses distinct nodes.
pub const HILBERT_BITS: u32 = 31;
/// Map `(lon, lat)` in degrees onto the `HILBERT_BITS`-per-axis integer grid.
#[inline]
#[allow(clippy::float_arithmetic)]
fn quantize(lon: f64, lat: f64) -> (u32, u32) {
let side = f64::from(1u32 << (HILBERT_BITS - 1)) * 2.0; // 2^31 as f64
let max = (1u64 << HILBERT_BITS) - 1;
let nx = (((lon + 180.0) / 360.0).clamp(0.0, 1.0) * side) as u64;
let ny = (((lat + 90.0) / 180.0).clamp(0.0, 1.0) * side) as u64;
(nx.min(max) as u32, ny.min(max) as u32)
}
/// The Hilbert d-index of grid cell `(x, y)` on a `2^HILBERT_BITS` square.
///
/// The classic iterative rotation algorithm (Wikipedia's `xy2d`): walk the
/// quadrant size from half the side down to 1, accumulate the quadrant's
/// contribution, and rotate the remaining coordinates into the sub-square's
/// frame. `O(HILBERT_BITS)`, no allocation, no floating point.
#[inline]
pub fn hilbert_d(mut x: u32, mut y: u32) -> u64 {
let mut rx: u32;
let mut ry: u32;
let mut d: u64 = 0;
let mut s: u32 = 1 << (HILBERT_BITS - 1);
while s > 0 {
rx = u32::from((x & s) > 0);
ry = u32::from((y & s) > 0);
d = d.wrapping_add(u64::from(s) * u64::from(s) * u64::from((3 * rx) ^ ry));
// rotate
if ry == 0 {
if rx == 1 {
x = s.wrapping_sub(1).wrapping_sub(x);
y = s.wrapping_sub(1).wrapping_sub(y);
}
std::mem::swap(&mut x, &mut y);
}
s >>= 1;
}
d
}
/// The Hilbert index of a `(lon, lat)` point in degrees. Locality-preserving:
/// points close in index are close on the ground (the converse is not guaranteed
/// at curve seams, which is why the reader still checks the real bbox).
#[inline]
pub fn hilbert_index(lon: f64, lat: f64) -> u64 {
let (x, y) = quantize(lon, lat);
hilbert_d(x, y)
}
/// A feature's bounding box in `[xmin, ymin, xmax, ymax]` order — the covering.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct Bbox {
pub xmin: f64,
pub ymin: f64,
pub xmax: f64,
pub ymax: f64,
}
impl Bbox {
/// Midpoint of the box — the point the Hilbert key is taken at.
#[allow(clippy::float_arithmetic)]
pub fn centre(&self) -> (f64, f64) {
((self.xmin + self.xmax) / 2.0, (self.ymin + self.ymax) / 2.0)
}
}
/// Bounding box of a WKB geometry, or `None` for null / unparsable / empty WKB.
///
/// Handles the geometry types osm-katana writes (Point, LineString, Polygon) and
/// their Multi\* / GeometryCollection wrappers, in either byte order, by walking
/// the WKB structure rather than materialising coordinates — one pass, no
/// allocation.
pub fn wkb_bbox(wkb: &[u8]) -> Option<Bbox> {
let mut b = Bbox {
xmin: f64::MAX,
ymin: f64::MAX,
xmax: f64::MIN,
ymax: f64::MIN,
};
let mut pos = 0usize;
scan(wkb, &mut pos, &mut b, 0)?;
if b.xmin > b.xmax { None } else { Some(b) }
}
/// Recursive WKB walker. `depth` bounds pathological nesting.
fn scan(w: &[u8], pos: &mut usize, b: &mut Bbox, depth: u32) -> Option<()> {
if depth > 8 {
return None;
}
let le = *w.get(*pos)? == 1;
*pos += 1;
let ty = rd_u32(w, pos, le)? % 1000; // strip Z/M/ZM (1000/2000/3000) qualifiers
match ty {
1 => pt(w, pos, b, le),
2 => {
let n = rd_u32(w, pos, le)? as usize;
for _ in 0..n {
pt(w, pos, b, le)?;
}
Some(())
}
3 => {
let rings = rd_u32(w, pos, le)? as usize;
for _ in 0..rings {
let n = rd_u32(w, pos, le)? as usize;
for _ in 0..n {
pt(w, pos, b, le)?;
}
}
Some(())
}
4 | 5 | 6 | 7 => {
let n = rd_u32(w, pos, le)? as usize;
for _ in 0..n {
scan(w, pos, b, depth + 1)?;
}
Some(())
}
_ => None,
}
}
#[inline]
fn pt(w: &[u8], pos: &mut usize, b: &mut Bbox, le: bool) -> Option<()> {
let x = rd_f64(w, pos, le)?;
let y = rd_f64(w, pos, le)?;
if x < b.xmin {
b.xmin = x;
}
if y < b.ymin {
b.ymin = y;
}
if x > b.xmax {
b.xmax = x;
}
if y > b.ymax {
b.ymax = y;
}
Some(())
}
#[inline]
fn rd_u32(w: &[u8], pos: &mut usize, le: bool) -> Option<u32> {
let s: [u8; 4] = w.get(*pos..*pos + 4)?.try_into().ok()?;
*pos += 4;
Some(if le {
u32::from_le_bytes(s)
} else {
u32::from_be_bytes(s)
})
}
#[inline]
fn rd_f64(w: &[u8], pos: &mut usize, le: bool) -> Option<f64> {
let s: [u8; 8] = w.get(*pos..*pos + 8)?.try_into().ok()?;
*pos += 8;
Some(if le {
f64::from_le_bytes(s)
} else {
f64::from_be_bytes(s)
})
}
/// The GeoParquet 1.1 covering column: `struct<xmin,ymin,xmax,ymax: float>`.
///
/// The spec requires the children be named `xmin, ymin, xmax, ymax` **in that
/// order** and be "of Parquet type FLOAT or DOUBLE", all the same type.
///
/// **FLOAT, not DOUBLE — measured.** On Stockholm's 4.7 M nodes the four covering
/// columns cost, compressed:
///
/// ```text
/// f64, plain 132611232
/// f64, byte-stream-split 97321255
/// f32, plain 46296340
/// f32, byte-stream-split 28774887 ← 4.6× smaller than f64 plain
/// ```
///
/// A covering is a *conservative approximation* by definition — the reader
/// re-checks the real geometry — so spending 8 bytes to carry a coordinate the
/// WKB column already holds exactly is pure waste. What f32 costs is ~2 m of slop
/// at Stockholm's latitude, which is invisible at row-group granularity (a group
/// spans hundreds of metres at minimum).
///
/// Correctness is preserved by rounding **outwards** ([`down`] / [`up`]): the
/// stored box always CONTAINS the true box, so a row group is never pruned away
/// when it might have matched. Rounding to nearest would shrink some boxes and
/// could drop a real hit — see the `covering_is_conservative` test.
///
/// The four leaves are also the reason the covering is QUANTISED rather than
/// two-leaved: for a point `xmin == xmax` bit for bit, but GeoParquet 1.1 names
/// four leaf paths and the shipped reader reads all four by name, so the
/// duplication is paid for in [`COVERING_GRID_SHIFT`]'s coin instead — read that
/// constant for the measurement, including what dropping two leaves does to a
/// real prune.
pub fn bbox_field() -> Field {
Field::new(
BBOX_COLUMN,
DataType::Struct(Fields::from(vec![
Field::new("xmin", DataType::Float32, true),
Field::new("ymin", DataType::Float32, true),
Field::new("xmax", DataType::Float32, true),
Field::new("ymax", DataType::Float32, true),
])),
true,
)
}
/// **The covering's quantisation grid, as a negative power of two of a DEGREE.**
///
/// Every covering leaf is snapped OUTWARD onto multiples of `2^-12` ° =
/// 0.000 244 140 625 ° — about **27 m north-south**, and 12–14 m east-west at
/// Nordic latitudes. [`down`] floors onto it, [`up`] ceilings onto it, so the
/// stored box still CONTAINS the true one and the covering stays conservative.
///
/// # Why a covering is quantised at all — the 346.97 MB that had no answer
///
/// `BaseTablePack::growth_pct`'s census found that `nodes.parquet`'s covering
/// column costs 693.94 MB on the Sweden cut and that **half of it is a byte-exact
/// duplicate**: for a Point `xmin == xmax` and `ymin == ymax`, and MEASURED on the
/// real file they are equal *bit for bit* in 24 576 of 24 576 sampled rows —
/// because the node coordinates the WKB carries are already exactly `f32`, so
/// `down` and `up` land on the same value. The two obvious fixes both fail:
///
/// * **Emit two leaves.** GeoParquet 1.1 names four leaf paths and korp's shipped
/// `mapquery::row_group_intersects` reads all four BY NAME; with `bbox.xmax`
/// absent its `gxmax` stays `-inf`, `!gxmax.is_finite()` fires and it returns
/// `true` — scan. MEASURED against that real function on the real Sweden files:
/// the prune goes from 277 of 12 633 row groups to **12 633 of 12 633** on a
/// Stockholm viewport (51× the rows) and to 12 633 on a 1 km cell (**21 064×**),
/// including for a viewport over open ocean with zero hits. Safe — it never
/// wrongly prunes — and useless.
/// * **Let the encoding do it.** There is nothing left: the writer already sets
/// `BYTE_STREAM_SPLIT` with the dictionary OFF, and re-encoding the real column
/// through the same writer gives BSS **693.94 MB**, PLAIN 1 002.38 MB (+44.4 %)
/// and dictionary-on 1 389.83 MB (+100.3 %) — the shipped choice is already the
/// best of the three by a wide margin. Parquet has no cross-column encoding, so
/// two columns holding identical values still pay for them twice. (Two leaves
/// would cost 346.97 MB — the redundancy figure exactly, which is the arithmetic
/// check that the census was right.)
///
/// What is left is to stop storing bits the prune cannot use. BYTE_STREAM_SPLIT
/// already flattens the sign/exponent/high-mantissa planes into runs; the low
/// mantissa is the only expensive plane and it carries sub-metre precision that a
/// row-group bounding box has no use for. Quantising is what makes that plane
/// compressible — and it keeps all four leaves, so the shipped reader is untouched.
///
/// # Why `2^-12` and why an ABSOLUTE grid (MEASURED, Sweden 2026-08-31)
///
/// The alternative was truncating mantissa bits, which is cheaper to compute but
/// gives *relative* precision: keeping 16 mantissa bits costs 10 m of slop at
/// longitude 24 and 109 m at latitude 69 — and 217 m near longitude 180, which is
/// a planet cut. The absolute grid buys the same bytes at half the worst-case slop
/// and behaves identically everywhere on Earth. Covering-column bytes on the real
/// `nodes.parquet` (103 488 583 rows), re-encoded through the same writer:
///
/// ```text
/// shipped (no quantisation) 693.94 MB slop 0
/// 2^-15 (3.4 m) 540.17 MB −22.2 %
/// 2^-14 (6.8 m) 481.91 MB −30.6 %
/// 2^-13 (13.6 m) 415.55 MB −40.1 %
/// 2^-12 (27.2 m) 344.55 MB −50.3 % <- this
/// mantissa-16 (10…109 m) 343.40 MB −50.5 % (same bytes, twice the slop)
/// ```
///
/// # What it costs at read time — nothing the prune can see
///
/// Against korp's REAL `row_group_intersects` on the real Sweden tables, the
/// row-group prune at `2^-12` is **identical to the unquantised baseline at every
/// zoom measured** — Sweden-wide, a 25 km Stockholm viewport (277/12 633), a 1 km
/// z15 cell (4/12 633), a 150 m z18 cell (3/12 633), and an ocean viewport
/// (0/12 633) — on both `nodes.parquet` and `ways.parquet`. Only the per-row cull
/// admits more, and it admits, never drops: +2 439 of 2 027 647 rows (+0.12 %) on
/// the Stockholm viewport, +90 of 167 at z18 — against 24 576 rows the row-group
/// prune already had to scan for those 167. For a POINT the reader then re-checks
/// the exact coordinate (`if !q.contains(coords[0].0, coords[0].1) { continue }`),
/// so on the node table the served reply is **unchanged** — MEASURED end to end
/// through `mapquery::query_ways_from`: 4 913 points at a z15 cell, 167 at z18 and
/// 0 over ocean, before and after, from the same row groups.
pub const COVERING_GRID_SHIFT: u32 = 12;
/// `2^COVERING_GRID_SHIFT` — the multiplier that puts a degree on the grid.
#[inline]
fn covering_grid() -> f64 {
f64::from(1u32 << COVERING_GRID_SHIFT)
}
/// Largest grid multiple that is `<= v`, as `f32` — the outward rounding for a
/// bbox MINIMUM. See [`COVERING_GRID_SHIFT`] for why it is a grid and not a cast.
///
/// The `f32` step is kept as a safety net: a grid multiple below `2^40`° is exactly
/// representable in `f32`, so it never fires for a coordinate, but a covering must
/// be conservative for whatever it is handed, not only for well-behaved input.
#[inline]
#[allow(clippy::float_cmp, clippy::float_arithmetic)]
pub fn down(v: f64) -> f32 {
let g = covering_grid();
let r = ((v * g).floor() / g) as f32;
if f64::from(r) <= v {
r
} else {
next_toward_neg_inf(r)
}
}
/// Smallest grid multiple that is `>= v`, as `f32` — the outward rounding for a
/// bbox MAXIMUM. See [`COVERING_GRID_SHIFT`] and [`down`].
#[inline]
#[allow(clippy::float_cmp, clippy::float_arithmetic)]
pub fn up(v: f64) -> f32 {
let g = covering_grid();
let r = ((v * g).ceil() / g) as f32;
if f64::from(r) >= v {
r
} else {
next_toward_pos_inf(r)
}
}
/// Is `v` exactly a multiple of the [`COVERING_GRID_SHIFT`] grid? The predicate the
/// applied-output guard reads a packed file with — kept beside the rounding so the
/// guard cannot drift from the writer (LAW 5).
#[inline]
#[allow(clippy::float_cmp, clippy::float_arithmetic)]
pub fn on_covering_grid(v: f32) -> bool {
let s = f64::from(v) * covering_grid();
s.fract() == 0.0
}
#[inline]
fn next_toward_neg_inf(x: f32) -> f32 {
if x.is_nan() || x == f32::NEG_INFINITY {
return x;
}
let b = x.to_bits();
// IEEE-754 magnitude ordering: stepping the bit pattern moves one ULP.
f32::from_bits(if x > 0.0 {
b - 1
} else if x == 0.0 {
0x8000_0001
} else {
b + 1
})
}
#[inline]
fn next_toward_pos_inf(x: f32) -> f32 {
if x.is_nan() || x == f32::INFINITY {
return x;
}
let b = x.to_bits();
f32::from_bits(if x >= 0.0 { b + 1 } else { b - 1 })
}
/// **Rows per row group — the ONE number (LAW 5), and it is MEASURED.**
///
/// Every packed table in this crate — the networks, the resolved relations, and the
/// base tables — takes its row-group size from here. A second literal `8192` in a
/// caller is a second writer of a tuning axis, which is how two files that both claim
/// `covering.bbox` end up pruning differently.
///
/// # The measurement (2026-08-31, oden)
///
/// Simulated over the real Hilbert row order of `networks/roads.parquet`
/// (2 277 368 rows), central Stockholm ~25 km viewport, against a ground truth of
/// 138 229 genuinely-intersecting rows:
///
/// ```text
/// rows/group groups hit rows scanned vs ideal
/// 1 024 240 of 2224 245 760 1.78x
/// 4 096 60 of 556 245 760 1.78x
/// 8 192 30 of 278 245 760 1.78x <- shipped
/// 16 384 18 of 139 294 912 2.13x
/// 65 536 6 of 35 393 216 2.84x
/// 100 000 5 of 23 500 000 3.62x
/// ```
///
/// Re-measured the same way on the two base tables once they were packed, same
/// viewport, ground truth 387 521 rows for ways and 2 304 820 for nodes:
///
/// ```text
/// ways.parquet (9 325 142 rows) nodes.parquet (103 488 583 rows)
/// rows/group groups hit scanned vs ideal groups hit scanned vs ideal
/// 1 024 512 of 9107 523 862 1.35x 2500 of 101064 2 560 000 1.11x
/// 4 096 135 of 2277 551 510 1.42x 628 of 25266 2 572 288 1.12x
/// 8 192 72 of 1139 584 278 1.51x 314 of 12633 2 572 288 1.12x
/// 16 384 47 of 570 756 310 1.95x 162 of 6317 2 654 208 1.15x
/// 65 536 19 of 143 1198 678 3.09x 48 of 1580 3 145 728 1.36x
/// 100 000 16 of 94 1525 142 3.94x 37 of 1035 3 700 000 1.61x
/// ```
///
/// **Be precise about what generalises.** On `roads.parquet` and `nodes.parquet`,
/// 8 192 is the largest size that still sits on the floor — 4 096 and 1 024 buy
/// nothing. On `ways.parquet` it does NOT: 8 192 scans 5.9 % more rows than 4 096 and
/// 11.5 % more than 1 024, because `ways` carries long features (coastlines,
/// boundaries) whose extent widens every group they land in, so the curve keeps
/// improving past the point the other tables flatten. 8 192 is still the call — 5.9 %
/// of scan against 4x the row groups and their footer is not a trade worth taking —
/// but "8 192 reaches the floor" is a statement about a table, not a law.
///
/// What IS uniform across all three: the 100 000-row group that every GeoParquet
/// tuning guide recommends is far worse — 2.03x on roads, 3.94x/1.61x here.
/// `the_row_group_size_is_the_measured_one` re-measures that arm on a fixture and
/// reports the regression rather than asserting an inequality.
///
/// The parquet **page index buys nothing here** and is not the fallback: 8 192 f32
/// rows is 32 KB against a 1 MB page limit, so the page IS the row group.
pub const PACK_ROW_GROUP_ROWS: usize = 8_192;
/// Options for [`spatial_pack`].
#[derive(Clone, Debug)]
pub struct PackOptions {
/// Parquet compression name (`zstd` | `snappy` | `none`).
pub compression: String,
/// Rows per row group. This is the pruning/compression dial and the default
/// is **measured, not guessed**. Sweeping Stockholm (bytes read for one
/// city-block bbox, and total file size):
///
/// ```text
/// rows/group nodes read ways read packed total
/// 8192 0.69% 8.83% 178486633
/// 16384 0.86% 15.59% 179619338
/// 32768 2.73% 29.90% 177283892
/// 65536 4.15% 48.24% 171899868
/// 131072 8.40% 100.00% 167987505
/// ```
///
/// 8192 reads 6x fewer bytes than 65536 for 3.8% more file. At ~200 KB of
/// compressed node data per group it is also a sensible HTTP range-request
/// size, which is the delivery this layout exists to serve.
///
/// The default is [`PACK_ROW_GROUP_ROWS`], re-measured on `roads.parquet` in 2026-08.
pub row_group_rows: usize,
/// Emit the `covering.bbox` column. Off ⇒ Hilbert ordering only, which keeps
/// the schema **byte-identical** to convert's output (useful to isolate the
/// ordering effect from the covering effect when measuring).
pub covering: bool,
/// Emit the generalisation columns `geometry_z6` / `geometry_z9` / `geometry_z12`
/// (DELTA 2). Off is the measurement's RED arm: the same read then has only the
/// full-precision `geometry` column to project and the 125.5 ms decode comes back.
///
/// Ignored for a table whose `geo` document declares only point geometry — see
/// [`crate::shared::generalise::table_generalises`], which is why `nodes.parquet`
/// does not carry three byte-for-byte copies of its own geometry column.
pub generalise: bool,
/// **Move `node_refs` out of the served table into a sidecar beside it** (row 22).
///
/// MEASURED on korp's live Sweden cut, 2026-09-05, `osm-katana census`, footer only:
/// `node_refs.list.item` is **437 313 567 B compressed — 28.4 % of
/// `ways.parquet`**, more than every rung column put together, in the table korp
/// reads for every viewport. Nothing korp serves reads it: the only two consumers
/// are [`crate::relgeom`]'s ring assembly, which matches ring endpoints on node id,
/// and `osm-katana verify`.
///
/// # Why the PACK moves it and not the writers
///
/// `ways_schema(include_node_refs)` could already drop the column at write time, and
/// that turned out to be a footgun rather than a feature: nothing else held the ids,
/// so `--node-refs false` did not move them, it **deleted** them — and `relgeom` did
/// not notice, because `col_i64_list` returns `None`, `nodes` goes empty and
/// `nodes.len() != coords.len()` skips every member way in silence. A cut made with
/// a flag the CLI offers today resolved zero areas and reported success.
///
/// The pack is the right seam and it is nearly free here: it already reads every
/// row, already decides the output layout (it adds the covering and the rung
/// columns), and already holds the batches in memory to sort them. Splitting one
/// column out costs no extra pass over the table, and it leaves pass 2 — the hot
/// parallel encode, four writer paths — completely untouched.
///
/// The sidecar is written in the SAME packed row order, so a positional join works;
/// `relgeom` joins by id anyway, because it wants only the member ways.
pub node_refs_sidecar: bool,
}
/// The column the sidecar carries out of the served table.
const NODE_REFS_COLUMN: &str = "node_refs";
/// **The sidecar's path for a packed table** — `ways.parquet` ⇒ `ways.refs.parquet`.
///
/// One formatter, so the writer, the two readers and the tests cannot spell the file
/// three ways.
#[must_use]
pub fn node_refs_sidecar_path(dst: &Path) -> PathBuf {
let stem = dst.file_stem().map_or_else(
|| String::from("ways"),
|s| s.to_string_lossy().into_owned(),
);
dst.with_file_name(format!("{stem}.refs.parquet"))
}
/// The sidecar's schema — the way's `id` and its `node_refs`, and nothing else.
///
/// `id` non-null because it is the join key; the list nullable for the same reason
/// `ways_schema` has it nullable (a NODE row's cell is NULL, and null and empty are
/// different claims).
#[must_use]
pub fn node_refs_sidecar_schema() -> Schema {
Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new(
NODE_REFS_COLUMN,
DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
true,
),
])
}
impl Default for PackOptions {
fn default() -> Self {
Self {
compression: String::from("zstd"),
row_group_rows: PACK_ROW_GROUP_ROWS,
covering: true,
generalise: true,
node_refs_sidecar: false,
}
}
}
/// Result of packing one table.
#[derive(Clone, Copy, Debug, Default)]
pub struct PackStats {
pub rows: usize,
/// Rows whose geometry was null / unparsable — they sort last, keeping the
/// operation total (no row is ever dropped).
pub rows_no_geometry: usize,
pub row_groups: usize,
pub bytes: u64,
}
/// Read `src`, sort its rows on the Hilbert index of each feature's bounding-box
/// centre, and write `dst` with a GeoParquet 1.1 `covering.bbox` column.
///
/// **Content-preserving by construction**: every input row is carried across
/// unchanged; only the row ORDER changes and (optionally) a derived column is
/// appended. `osm-katana verify --digest` reports the same `set` digest before
/// and after — that is the equivalence gate.
///
/// Rows with no usable geometry keep their relative order and are placed after
/// every geometric row, so they occupy their own trailing row groups and never
/// widen a real group's bounds.
pub fn spatial_pack(src: &Path, dst: &Path, opts: &PackOptions) -> anyhow::Result<PackStats> {
let file = fs::File::open(src).with_context(|| format!("open {src:?}"))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
let in_schema = builder.schema().clone();
let reader = builder.with_batch_size(8192).build()?;
let mut batches: Vec<RecordBatch> = Vec::new();
for b in reader {
batches.push(b?);
}
pack_batches(batches, &in_schema, dst, opts, &[])
.with_context(|| format!("spatial_pack {src:?}"))
}
/// The tables `convert` and `relations` emit, in the order they are packed.
///
/// **The ONE list (LAW 5).** [`spatial_pack_dir`] and [`pack_base_tables`] both read
/// it, so the CLI's `spatial-pack <dir>` and the pipeline's own packing can never
/// disagree about which files are base tables. `relations.resolved.parquet` is in here
/// because it is the relation table anything downstream actually reads — the raw
/// `relations.parquet` carries no assembled geometry.
pub const BASE_TABLES: &[&str] = &[
"nodes.parquet",
"ways.parquet",
"relations.parquet",
"relations.resolved.parquet",
];
/// Is this table ALREADY packed — i.e. does it carry the `covering.bbox` column, the
/// rung columns its geometry earns, and (when it has `tags`) a `min_zoom` column with
/// the row groups in band-first order?
///
/// Footer-only: no row group is decoded. This is what makes packing **idempotent**,
/// and idempotence is not a nicety here. Convert packs its three tables; `relations`
/// then reads `ways.parquet` and packs its own output; a resumed or re-run stage
/// would otherwise hand [`pack_batches`] a schema that already has a `bbox` field and
/// get a file with **two** columns of that name — valid parquet that every reader
/// resolves by position, i.e. a silent layout fork.
pub fn is_packed(path: &Path) -> anyhow::Result<bool> {
let file = fs::File::open(path).with_context(|| format!("open {path:?}"))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
let schema = builder.schema();
if schema.index_of(BBOX_COLUMN).is_err() {
return Ok(false);
}
// **A table packed by an older build is not packed for THIS one.** `covering.bbox`
// alone was the whole layout until DELTA 2; a file that has it but lacks the
// generalisation columns still forces the reader onto full-precision WKB, which is
// precisely the 125.5 ms this delta exists to remove. Reporting it as packed would
// make the upgrade silently a no-op on every extract already on disk.
//
// Answering `false` here is only safe because [`pack_batches`] DROPS the derived
// columns it is about to rebuild (see `carry_columns`), so a second pack replaces
// them instead of appending a second `bbox` — the silent layout fork this function's
// idempotence used to be the only guard against.
for &z in &crate::shared::schema::GENERALISED_RUNGS {
if generalised_rungs_for(&schema, true).contains(&z)
&& schema
.index_of(&crate::shared::schema::generalised_geometry_name(z))
.is_err()
{
return Ok(false);
}
}
// **DELTA 3: the ORDER is part of the layout.** A table that carries `tags` gets a
// `min_zoom` column from the pack (see `ensure_min_zoom`) and is sorted band-first
// on it. A table with the column but Hilbert-only order — everything packed before
// DELTA 3, `min_zoom` present or not — prunes nothing on that column, and calling it
// packed would leave it that way forever. Both checks are footer-only.
if schema.index_of(TAGS_COLUMN).is_ok() {
if schema.index_of(MIN_ZOOM_COLUMN).is_err() {
return Ok(false);
}
if !is_band_first(builder.metadata()) {
return Ok(false);
}
}
Ok(true)
}
/// **The footer key a packed table declares its recipe under** (row 38).
///
/// Named like [`crate::network::NETWORK_RECIPE_META_KEY`] because it is the same idea
/// applied to the third kind of object this crate writes: a tile says what cut it, a
/// network table says what cut it, and now a packed base table does too.
pub const PACK_RECIPE_META_KEY: &str = "osm-katana:pack-recipe";
/// **What a packed table's contents are a function of — the axes a re-pack must
/// invalidate.**
///
/// # The defect this closes
///
/// [`pack_batches`] had an `extra_meta` parameter from the beginning and **every
/// production caller passed `&[]`**, so a packed table carried no statement of what
/// produced it. korp's `pipeline::packed_state` therefore judged a cut on STRUCTURE
/// alone — does `ways.parquet` carry a `covering.bbox`, does it carry rung columns —
/// and both of those survive every edit to the thing that decides what is IN those
/// columns.
///
/// That is not hypothetical. The min-area rule landed in
/// [`crate::shared::generalise::area_rung_floor`] and the live Sweden cut went on
/// serving the pre-min-area table under a `Complete` verdict until the files were
/// swapped by hand. A recipe token is the difference between a pipeline that re-cuts
/// what it invalidated and one that needs an operator to remember.
///
/// # The terms, and the edit each one alone can see
///
/// | term | what moves it that nothing else would |
/// |---|---|
/// | `cols` | a column added, dropped or **retyped** — [`crate::shared::schema::short_type`] recurses, so a `bbox` leaf narrowing from `Float64` to `Float32` moves it |
/// | `rg` | the row-group size: the granularity every prune happens at |
/// | `rungs` | a generalisation column added or removed |
/// | `eps` | [`crate::tiles::DEFAULT_TILE_RASTER_PX`] moving, i.e. every rung's Douglas–Peucker tolerance at once. It moved once already, 256 → 2048 |
/// | `areafloor` | the min-area rule's own constant — the edit that actually bit |
/// | `minzoom` | a `rung!` line edited: the rungs AND the kinds, from [`crate::shared::class`]. A ladder edit changes every `min_zoom` and `class_id` cell and not one column name |
///
/// # `Option` on the ladder is not a shortcut
///
/// A table with no `tags` and no `min_zoom` — the graph tables — cannot express a ladder
/// edit in its contents, and stamping one anyway is what made every ladder edit re-cut
/// 8.4 M graph rows before `NetworkRecipe::for_table` learned the same lesson on
/// 2026-09-01. An axis belongs in a token only when the table carries it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PackRecipe {
/// `name:Type` per column of the WRITTEN schema, in write order — including the
/// packer's own `bbox` and every rung column.
pub columns: Vec<String>,
/// Rows per row group.
pub row_group_rows: usize,
/// The generalisation rungs this table actually carries.
pub rungs: Vec<u8>,
/// The simplification basis, spelled — the raster the epsilon is derived from.
pub eps_basis: String,
/// The min-area rule, spelled.
pub area_floor: String,
/// [`crate::shared::class::rules_spelling`] + [`crate::shared::class::kinds_spelling`],
/// or `None` for a table that carries no `min_zoom` column to value.
pub min_zoom_ladder: Option<String>,
}
/// What a packed table on disk says about itself, judged against a recipe.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PackVerdict {
/// The file's own token is the recipe's token.
Current,
/// The file declares a recipe and it is a different one. Carries the difference.
Stale(String),
/// The file declares **no** recipe at all — every table packed before row 38.
///
/// A variant of its own, and deliberately not `Stale`: the two want different
/// operator decisions (a re-pack versus "this is an old cut, decide"), and neither
/// is `Current`. Silence is not agreement — that is the false-Current this whole
/// token exists to prevent.
Unstamped(String),
}
impl PackRecipe {
/// The recipe for a table this build would write with `opts` from `written_schema`.
///
/// `written_schema` is the schema the packer actually WROTE (covering and rung
/// columns included), not the input — so the `cols` term is applied output rather
/// than an intention.
#[must_use]
pub fn of_written(written_schema: &Schema, opts: &PackOptions, rungs: &[u8]) -> Self {
let columns: Vec<String> = written_schema
.fields()
.iter()
.map(|f| {
format!(
"{}:{}",
f.name(),
crate::shared::schema::short_type(f.data_type())
)
})
.collect();
// The ladder axis is carried only by a table that carries the column it values —
// the same rule `NetworkRecipe::for_table` applies, decided from the field list
// in the same place the column axis is decided.
let min_zoom_ladder = written_schema.index_of(MIN_ZOOM_COLUMN).is_ok().then(|| {
format!(
"{}|kinds={}",
crate::shared::class::rules_spelling(),
crate::shared::class::kinds_spelling()
)
});
Self {
columns,
row_group_rows: opts.row_group_rows,
rungs: rungs.to_vec(),
eps_basis: eps_basis_spelling(),
area_floor: area_floor_spelling(),
min_zoom_ladder,
}
}
/// The canonical spelling. Anything that moves a term moves this string, and a moved
/// string is *Stale*, which re-packs — the fail-safe direction.
#[must_use]
pub fn token(&self) -> String {
let mut t = format!(
"v1;cols={};rg={};rungs={};eps={};areafloor={}",
self.columns.join(","),
self.row_group_rows,
self.rungs
.iter()
.map(|z| format!("z{z}"))
.collect::<Vec<_>>()
.join("+"),
self.eps_basis,
self.area_floor,
);
if let Some(l) = &self.min_zoom_ladder {
t.push_str(";minzoom=");
t.push_str(l);
}
t
}
/// The recipe a packed table's footer declares, or `None` when it declares none.
/// 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 rather than opened — one footer
/// lookup for both (LAW 5), so disk and object 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 == PACK_RECIPE_META_KEY))
.and_then(|k| k.value.clone())
.filter(|v| !v.is_empty()))
}
/// **The recipe a RE-PACK of the table at `path` would stamp, with today's code.**
///
/// That question — *"would re-packing this table change it?"* — is the one korp's
/// `packed_state` actually needs answered, and it is computable without guessing:
/// strip what the packer ADDS (the covering, the rung columns) to recover the input
/// schema a re-pack would see, put back the columns
/// [`ensure_min_zoom`]/[`ensure_class_columns`] would derive, and run the packer's
/// own [`generalised_rungs_for`] and [`pack_schema`] over it.
///
/// It goes through those two functions rather than reimplementing what they do (LAW
/// 5), so the prediction cannot drift from the writer: a change to either moves this
/// answer in the same commit.
///
/// Two consequences worth stating, because both are the point:
///
/// * an old cut that lacks a column today's pack would add — row 18's class block,
/// DELTA 3's band — reads **Stale**, because the predicted `cols` carries it and
/// the file's does not;
/// * `rg` is [`PackOptions::default`]'s, **not** the row counts observed in the
/// file. A small table's last row group is short by construction, and reading the
/// achieved count back would make every table with fewer rows than one group read
/// Stale against the very options that wrote it.
pub fn current_for(path: &Path) -> anyhow::Result<Self> {
let r = ParquetRecordBatchReaderBuilder::try_new(std::fs::File::open(path)?)?;
let file_schema = r.schema().as_ref().clone();
// Recover the input a re-pack would see: everything the packer did not add.
let rung_names: Vec<String> = crate::shared::schema::GENERALISED_RUNGS
.iter()
.map(|z| crate::shared::schema::generalised_geometry_name(*z))
.collect();
let mut input_fields: Vec<Arc<Field>> = file_schema
.fields()
.iter()
.filter(|f| f.name() != BBOX_COLUMN && !rung_names.contains(f.name()))
.cloned()
.collect();
// …plus the columns the pack would DERIVE for it.
input_fields.extend(derived_class_fields(&file_schema).into_iter().map(Arc::new));
let input = Schema::new_with_metadata(
arrow::datatypes::Fields::from(input_fields),
file_schema.metadata().clone(),
);
let opts = PackOptions::default();
let rungs = generalised_rungs_for(&input, opts.generalise);
let written = pack_schema(&input, opts.covering, &rungs, opts.node_refs_sidecar);
Ok(Self::of_written(&written, &opts, &rungs))
}
/// [`Self::verdict_against`] with the recipe this build implements for that table.
pub fn verdict(path: &Path) -> anyhow::Result<PackVerdict> {
let want = Self::current_for(path)?;
Self::verdict_against(path, &want)
}
/// **Judge the table at `path` against `want`.**
///
/// `want` is an ARGUMENT and not a recomputation, so a guard can hand this a
/// deliberately different recipe and watch the answer move. A verdict that built
/// both sides from one function could only ever compare a value with itself — the
/// identity trap LAW 2 names, and the shape korp `ebf46d1` was written to kill.
pub fn verdict_against(path: &Path, want: &Self) -> anyhow::Result<PackVerdict> {
let Some(found) = Self::of_table(path)? else {
return Ok(PackVerdict::Unstamped(format!(
"{path:?} declares no pack recipe — it was packed before the recipe \
existed, so nothing in it says which rules produced its rungs, its band \
or its geometry. Re-pack it with `osm-katana spatial-pack`."
)));
};
let want_t = want.token();
if found == want_t {
return Ok(PackVerdict::Current);
}
Ok(PackVerdict::Stale(first_difference(&found, &want_t)))
}
}
/// The first term that differs between two tokens, named — so a `Stale` says WHICH axis
/// moved instead of printing two long strings and leaving the reader to diff them.
fn first_difference(found: &str, want: &str) -> String {
let f: Vec<&str> = found.split(';').collect();
let w: Vec<&str> = want.split(';').collect();
for (a, b) in f.iter().zip(w.iter()) {
if a != b {
let axis = a.split('=').next().unwrap_or(a);
return format!(
"the pack recipe moved on `{axis}`:\n file: {a}\n code: {b}"
);
}
}
format!(
"the pack recipe moved ({} terms on the file, {} in this build):\n file: {found}\n code: {want}",
f.len(),
w.len()
)
}
/// **The simplification basis, spelled.** One string, derived from the ONE constant, so
/// a raster change moves every packed table's token — it already moved once (256 →
/// 2048) and the rung epsilons moved eightfold with it.
fn eps_basis_spelling() -> String {
format!(
"dp/raster{}",
crate::tiles::DEFAULT_TILE_RASTER_PX
)
}
/// **The min-area rule, spelled** — the edit that actually bit the live cut.
fn area_floor_spelling() -> String {
format!(
"extent>={}px",
crate::shared::generalise::AREA_MIN_EXTENT_PIXELS
)
}
/// Which of [`crate::shared::schema::class_fields`] this schema does not already carry.
fn missing_class_fields(schema: &Schema) -> Vec<Field> {
crate::shared::schema::class_fields()
.into_iter()
.filter(|f| schema.index_of(f.name()).is_err())
.collect()
}
/// **The columns a pack would DERIVE for `schema` from its `tags`** — [`ensure_min_zoom`]'s
/// column and [`ensure_class_columns`]' block, in write order.
///
/// The ONE statement of "what does the pack add for you" (LAW 5), so
/// [`PackRecipe::current_for`]'s prediction and the writer's own behaviour cannot
/// disagree about whether an old cut is missing something.
///
/// Empty when the table has no `Utf8` `tags` column: there is nothing to classify from,
/// which is why the graph tables are not judged for a class block they could never have.
fn derived_class_fields(schema: &Schema) -> Vec<Field> {
let has_tags = schema
.index_of(TAGS_COLUMN)
.is_ok_and(|i| matches!(schema.field(i).data_type(), DataType::Utf8));
if !has_tags {
return Vec::new();
}
let mut out = Vec::new();
if schema.index_of(MIN_ZOOM_COLUMN).is_err() {
out.push(crate::shared::schema::min_zoom_field());
}
out.extend(missing_class_fields(schema));
out
}
/// The `tags` column every base table carries — the input `min_zoom` is derived from.
const TAGS_COLUMN: &str = "tags";
/// The `min_zoom` column's name. Pinned to [`crate::shared::schema::min_zoom_field`] by
/// `min_zoom_pack_law::the_column_name_is_the_schemas`, so the packer and the schema
/// cannot spell it two ways.
const MIN_ZOOM_COLUMN: &str = "min_zoom";
/// **Is this file's row-group sequence monotone in `min_zoom`?** — the footer-visible
/// form of "band first, Hilbert within the band".
///
/// Every row group's `min_zoom` chunk statistics are read; the file is band-first when
/// `max(g) <= min(g + 1)` for every consecutive pair. No row is decoded. A file with no
/// `min_zoom` statistics at all answers `false` — nothing can be pruned on it.
pub fn is_band_first(md: &parquet::file::metadata::ParquetMetaData) -> bool {
use parquet::file::statistics::Statistics;
let mut prev_max: Option<i32> = None;
let mut seen = 0usize;
for g in 0..md.num_row_groups() {
let rg = md.row_group(g);
let Some(col) = rg
.columns()
.iter()
.find(|c| c.column_path().string() == MIN_ZOOM_COLUMN)
else {
return false;
};
let Some(Statistics::Int32(s)) = col.statistics() else {
return false;
};
let (Some(&lo), Some(&hi)) = (s.min_opt(), s.max_opt()) else {
return false;
};
if prev_max.is_some_and(|p| p > lo) {
return false;
}
prev_max = Some(hi);
seen += 1;
}
seen > 0
}
/// The input column indices a pack CARRIES ACROSS, i.e. everything that is not a
/// derived column this pass is about to rebuild.
///
/// `covering.bbox` and `geometry_z*` are pure functions of `geometry`. Carrying them
/// through and then appending fresh ones gives a file with two columns of each name —
/// valid parquet that every reader resolves by position, i.e. a silent layout fork. So
/// they are dropped on the way in and written on the way out, which makes a re-pack an
/// UPGRADE rather than a corruption, and is what lets [`is_packed`] be strict about the
/// layout it wants.
fn carry_columns(input: &Schema, split_refs: bool) -> Vec<usize> {
let mut derived: Vec<String> = std::iter::once(String::from(BBOX_COLUMN))
.chain(
crate::shared::schema::GENERALISED_RUNGS
.iter()
.map(|&z| crate::shared::schema::generalised_geometry_name(z)),
)
.collect();
// Row 22: `node_refs` leaves the served table for a sidecar. It is excluded HERE,
// in the one list both the output schema and the gather loop read, so the two
// cannot disagree about whether the column is present.
if split_refs {
derived.push(String::from(NODE_REFS_COLUMN));
}
input
.fields()
.iter()
.enumerate()
.filter(|(_, f)| !derived.contains(f.name()))
.map(|(i, _)| i)
.collect()
}
/// **The columns whose values are near-unique, so a dictionary page is pure cost.**
///
/// A parquet dictionary stores every distinct value once plus one index per row. That
/// is a win on `tags` (repeated JSON), on `version` and on `min_zoom` (a five-value
/// ladder). It is a LOSS on a column whose values are almost all distinct: the
/// "dictionary" is then a copy of the column with an index array bolted on, and — worse
/// — dictionary pages are RLE-encoded per page, which pre-empts the plain byte runs
/// zstd would otherwise flatten. The covering leaves have had it off since they landed;
/// `id` and the WKB geometry columns did not, and nothing had measured them.
///
/// # MEASURED on the live Sweden cut, 2026-09-03 (oden)
///
/// Re-encoded through the same writer settings, 60 row groups spread across
/// `nodes.parquet`, 24 across `ways.parquet`, 12 across `relations.resolved.parquet`.
/// The `as shipped` arm reproduces the footer to within 0.3 % on nodes, which is what
/// says the harness is measuring the writer and not itself.
///
/// ```text
/// nodes.parquet (490 567 rows) total id geometry
/// as shipped (dictionary on) 6 505 281 1 535 907 3 118 459
/// `id` dictionary off 5 718 244 748 870 3 118 459 −11.8 %
/// + geometry dictionary off 4 928 399 748 870 2 328 614 −24.2 %
///
/// ways.parquet (191 062 rows) id geometry*
/// as shipped 27 364 649 764 942 20 156 006
/// + both off 26 577 516 458 647 19 675 168 −2.9 %
///
/// relations.resolved (93 982 rows) id geometry*
/// as shipped 122 193 966 362 663 116 377 414
/// + both off 121 904 547 212 246 116 238 412 −0.2 %
/// ```
///
/// It is a win on every table and a large one on the points, which is the table that
/// is 24 % `id` by weight. `relations.resolved` barely moves because 95 % of it is one
/// enormous geometry column whose blobs zstd already handles.
///
/// # The alternative that was measured and DECLINED
///
/// Re-sorting each row group's rows by ascending `id` (the group's contents and its
/// statistics are unchanged, so no prune moves) makes `id` delta-encodable and takes it
/// to **182 247** bytes on the nodes sample — another −8.6 % of the whole file. It is
/// declined because it wins on ONE table: the same change costs `ways.parquet` +1.3 %
/// and `relations.resolved` +0.3 %, because breaking Hilbert locality inside a group
/// costs the geometry and covering columns more than the `id` column gains wherever the
/// rows are lines rather than points. A pack order that is right for points and wrong
/// for ways is a per-table tuning knob, i.e. a second write order — and this crate has
/// one write order on purpose (LAW 5). The number is recorded so a future lane that
/// wants it knows what it is buying.
fn dictionary_is_waste(schema: &Schema) -> Vec<String> {
schema
.fields()
.iter()
.map(|f| f.name().clone())
.filter(|n| {
n == "id" || n == "geometry" || n.starts_with("geometry_z")
})
.collect()
}
/// Hilbert-pack `path` **over itself**, or report `None` if it is already packed.
///
/// The rewrite is not in-place in the dangerous sense: [`pack_batches`] builds the
/// packed table under a `.katana-partial` temp in the same directory and only renames
/// it onto `path` once the last row group and the footer are on disk (see
/// [`crate::writer::PendingOutput`]). The source is read to completion and its handle
/// dropped before that rename, so a pack that dies half-way leaves the ORIGINAL table
/// intact — the property a "sort the file you just wrote" step has to have or it is a
/// way to lose a 19-minute convert.
///
/// # Why this is a pass and not folded into the streaming write
///
/// A Hilbert order is GLOBAL: the first output row can depend on the last input row,
/// so a writer cannot emit it while streaming without buffering the whole table. The
/// streaming writers in [`crate::writer`] flush every
/// [`crate::writer::ROW_GROUP_SIZE`] rows; sorting only within a flush would order
/// rows inside a 524 288-row window that already spans the whole extract, which is
/// exactly the layout that prunes nothing.
///
/// So the buffer is unavoidable. What IS avoidable is paying for it **on top of**
/// convert's own working set — pass 2 holds the 1.66 GB mmap'd node store and every
/// PBF decode buffer. Run as its own pass, after every writer is closed, the peak is
/// one table's worth instead of convert's plus the whole extract's, and the tables are
/// packed one at a time so the peak is the LARGEST table, not their sum. MEASURED on
/// the Sweden cut (oden, 2026-08-31): 15.6 GB peak for `nodes.parquet`, 4.8 GB for
/// `ways.parquet`, 0.9 GB for `relations.resolved.parquet`.
///
/// The price of the pass is one extra read of a file that was written seconds ago and
/// is therefore in page cache. MEASURED: 78.3 s for all three Sweden tables
/// (113 087 931 rows), against a 56-minute cut — **2.3 %**.
pub fn pack_in_place(path: &Path, opts: &PackOptions) -> anyhow::Result<Option<PackStats>> {
if is_packed(path)? {
return Ok(None);
}
spatial_pack(path, path, opts).map(Some)
}
/// One packed table's report, for the phase log and the CLI.
#[derive(Clone, Debug)]
pub struct BaseTablePack {
pub name: String,
/// `None` ⇒ the table was already packed and was left alone.
pub stats: Option<PackStats>,
pub bytes_before: u64,
pub elapsed: std::time::Duration,
}
impl BaseTablePack {
/// How much bigger the packed table is, as a percentage. 0.0 when it was skipped.
///
/// Reported rather than hidden because it is NOT small and it is not the row-group
/// size that causes it. MEASURED per column on the Sweden cut (2026-08-31), input
/// order → Hilbert order, compressed bytes:
///
/// ```text
/// ways.parquet 902.13 MB → 1171.55 MB (+29.9 %)
/// node_refs 239.12 MB → 438.52 MB (+199.40 MB) <- 74 % of the growth
/// bbox (x4) — → 72.18 MB (+72.18 MB)
/// id 19.05 MB → 30.74 MB (+11.68 MB)
/// tags 47.17 MB → 33.94 MB (-13.24 MB) <- locality HELPS here
/// geometry 593.89 MB → 591.30 MB (-2.59 MB) <- unchanged
///
/// nodes.parquet 709.05 MB → 1726.95 MB (+143.6 %)
/// bbox (x4) — → 693.94 MB (+693.94 MB) <- 68 % of the growth
/// id 164.39 MB → 324.12 MB (+159.73 MB)
/// geometry 514.93 MB → 655.18 MB (+140.24 MB)
/// ```
///
/// Two things fall out of that and neither is the row-group size:
///
/// 1. **The growth is id-delta locality, not small row groups.** `geometry` is
/// *unchanged* on `ways.parquet` at 8 192 rows per group, so the 1 139 groups
/// cost essentially nothing. What costs is that `id` and `node_refs` are
/// near-sequential in OSM order and near-random in Hilbert order.
/// `ways_schema(include_node_refs: false)` — which a map-delivery cut already
/// wants — takes `ways.parquet` from +29.9 % to **+10.6 %**.
/// 2. **On a POINT table half the covering column is a byte-exact duplicate.**
/// `nodes.parquet`'s `bbox.xmin` and `bbox.xmax` chunks are the same size to
/// the byte (198 873 981 each), as are `ymin`/`ymax` (148 093 832) — because
/// for a point they hold the same values. Confirmed since at the VALUE level:
/// `xmin == xmax` and `ymin == ymax` bit for bit in 24 576 of 24 576 sampled
/// rows, because the node coordinates in the WKB are already exactly `f32`.
/// That is 346.97 MB of pure redundancy on Sweden. GeoParquet 1.1 requires all
/// four leaves and the shipped reader reads all four BY NAME, so the leaves
/// stay — and the bytes come back another way. **FIXED** by quantising the
/// covering onto a `2^-12`° grid ([`crate::spatial::COVERING_GRID_SHIFT`]),
/// which recovers 349.48 MB of the 693.94 with the row-group prune measured
/// bit-identical against korp's real reader. The numbers above are the
/// UNQUANTISED census (input order → Hilbert order); after the grid the same
/// Hilbert order gives:
///
/// ```text
/// ways.parquet 902.13 MB → 1141.66 MB (+26.6 %, was +29.9 %)
/// bbox (x4) — → 42.29 MB (was 72.18 MB, −29.89 MB, −41.4 %)
///
/// nodes.parquet 709.05 MB → 1377.56 MB (+94.3 %, was +143.6 %)
/// bbox (x4) — → 344.55 MB (was 693.94 MB, −349.39 MB, −50.3 %)
///
/// relations.resolved.parquet 192.78 → 192.10 MB
/// bbox (x4) — → 1.36 MB (was 2.04 MB, −0.68 MB)
///
/// all three base tables 1 801.85 → 2 711.33 MB (+50.5 %, was +71.6 %)
/// ```
///
/// Round-tripped through korp's whole shipped read path
/// (`mapquery::query_ways_from`) on both real tables before and after: the
/// row groups scanned are IDENTICAL at every viewport (10/12 633 Sweden,
/// 4/12 633 at a z15 cell, 3/12 633 at z18, **0/12 633** over open ocean) and
/// the node reply is identical to the point (4 913 / 167 / 0) — the reader
/// re-checks a point's exact coordinate, so quantising its covering changes
/// nothing it serves. `ways.parquet` returns a bounded few more: 1 364 rather
/// than 1 255 at a z15 cell, 97 rather than 64 at z18, all of them within one
/// grid step of the viewport edge, against a 160 000-way budget.
///
/// Which leaves `node_refs` as the ONE remaining unfixed line of the pack
/// cost, and `--node-refs false` already answers that one.
#[allow(clippy::float_arithmetic)]
pub fn growth_pct(&self) -> f64 {
match &self.stats {
None => 0.0,
Some(s) => {
100.0 * (s.bytes as f64 - self.bytes_before as f64)
/ (self.bytes_before.max(1) as f64)
}
}
}
}
/// **Pack every base table an extract directory holds, over itself.**
///
/// This is what makes `nodes.parquet` / `ways.parquet` / `relations.resolved.parquet`
/// come out of the pipeline with the layout `networks/roads.parquet` has had all
/// along — Hilbert order, `covering.bbox`, [`PACK_ROW_GROUP_ROWS`]-row row groups —
/// through [`pack_batches`], the one packer (LAW 5). Missing tables are skipped;
/// already-packed tables are skipped and reported as such.
///
/// # What it bought, MEASURED (Sweden, oden, 2026-08-31, warm page cache)
///
/// Central Stockholm viewport `(17.85, 59.22)–(18.30, 59.45)`, row-group prune on the
/// `bbox.*` statistics, geometry projected:
///
/// ```text
/// row groups bytes rows scanned wall
/// ways.parquet BEFORE 32 of 32 902.13 MB 9 325 142 3538.2 ms
/// ways.parquet AFTER 71 of 1139 27.08 MB 581 632 144.0 ms
/// nodes.parquet AFTER 314 of 12633 30.70 MB 2 572 288 189.8 ms
/// ```
///
/// 33.3x fewer bytes and 24.6x less wall on `ways.parquet`, for +29.9 % on disk.
pub fn pack_base_tables(dir: &Path, opts: &PackOptions) -> anyhow::Result<Vec<BaseTablePack>> {
let mut out = Vec::new();
for name in BASE_TABLES {
let p = dir.join(name);
if !p.is_file() {
continue;
}
let bytes_before = fs::metadata(&p).map(|m| m.len()).unwrap_or(0);
let t = std::time::Instant::now();
let stats = pack_in_place(&p, opts).with_context(|| format!("pack {p:?}"))?;
out.push(BaseTablePack {
name: String::from(*name),
stats,
bytes_before,
elapsed: t.elapsed(),
});
}
Ok(out)
}
/// **The Hilbert sort + covering-bbox writer — the ONE of it (LAW 5).**
///
/// [`spatial_pack`] is this function with a parquet read in front; anything else that
/// produces its own in-memory rows (see [`crate::network`]) hands them here instead of
/// growing a second, near-identical writer. A twinned layout writer is the defect class
/// that is invisible in code and total in data: two files that both *say*
/// `covering.bbox` but disagree about outward rounding, row-group size or key metadata
/// would prune differently and only the served map would ever show it.
///
/// `batches` must all share `in_schema` and it must carry a `geometry` (WKB) column.
/// `extra_meta` is appended to the parquet footer's key-value metadata AND to the
/// Arrow schema metadata, so a caller can stamp its own recipe token into the object —
/// the same evidence-in-the-object pattern [`crate::tiles::TileRecipe`] uses.
pub fn pack_batches(
batches: Vec<RecordBatch>,
in_schema: &Arc<Schema>,
dst: &Path,
opts: &PackOptions,
extra_meta: &[(String, String)],
) -> anyhow::Result<PackStats> {
// **DELTA 3, step 0: every row has a rung before anything is ordered.** A table
// written before DELTA 1 (the live Sweden cut of 2026-08-31) has `tags` and no
// `min_zoom`; the column is derived here, through the ONE tag → rung writer, and
// appended LAST — the position `ways_schema` gives it — so the output of a re-pack
// and the output of convert-then-pack have one layout. A present column is kept.
let (batches, in_schema) = ensure_min_zoom(batches, in_schema.clone())?;
// **Step 0a: the CLASS block, on the same terms** (row 18). A table written before
// 2026-09-05 has `tags` and no `class_id`/`kind`/`layer`/`bridge`/`tunnel`/`oneway`;
// they are derived here from the same one pass over the same `tags`, so a re-packed
// old cut and a fresh convert-then-pack produce one layout. Present columns are kept.
let (batches, in_schema) = ensure_class_columns(batches, in_schema)?;
// **Step 0b: an AREA's rung also answers to its SIZE (the min-area rule).** The tag
// ladder alone put every `natural=water` islet on rung 6, where Douglas–Peucker
// degrades it to a bare `Point` — 407 of 454 water ways in the Gävle-coast box of
// The owner's 2026-09-01 country-zoom screen, painted as a solid dot-blob where the
// archipelago should be. See `shared::generalise::area_rung_floor`.
let batches = deepen_area_min_zoom(batches, &in_schema)?;
let total: usize = batches.iter().map(RecordBatch::num_rows).sum();
let geom_idx = in_schema
.index_of("geometry")
.map_err(|_| anyhow::anyhow!("no `geometry` column — not a GeoParquet table"))?;
let min_zoom_idx = in_schema.index_of(MIN_ZOOM_COLUMN).ok();
// ── key extraction ────────────────────────────────────────────────────────
// One (min_zoom, hilbert, batch, row) tuple per row. Batches are independent, so
// the per-batch WKB walk fans out on gatling (ROOT LAW #0 — no rayon, no
// hand-rolled pool); results come back in batch order so the key vector is
// reproducible.
let per_batch: Vec<Vec<(u8, u64, Option<Bbox>)>> =
gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| {
let g = b.column(geom_idx).as_any().downcast_ref::<BinaryArray>();
let mz = min_zoom_idx
.and_then(|i| b.column(i).as_any().downcast_ref::<UInt8Array>());
(0..b.num_rows())
.map(|r| {
// A table with no rung column (no `tags` to derive one from)
// sorts on one band, i.e. Hilbert-only — the order it had.
let band = mz.map_or(0, |a| a.value(r));
match g {
Some(a) if !a.is_null(r) => match wkb_bbox(a.value(r)) {
Some(bb) => {
let (cx, cy) = bb.centre();
(band, hilbert_index(cx, cy), Some(bb))
}
None => (band, u64::MAX, None),
},
_ => (band, u64::MAX, None),
}
})
.collect()
},
);
let mut order: Vec<PackKey> = Vec::with_capacity(total);
let mut boxes: Vec<Vec<Option<Bbox>>> = Vec::with_capacity(batches.len());
let mut no_geom = 0usize;
for (bi, keys) in per_batch.into_iter().enumerate() {
let mut bb = Vec::with_capacity(keys.len());
for (ri, (z, h, b)) in keys.into_iter().enumerate() {
if b.is_none() {
no_geom += 1;
}
order.push((z, h, bi as u32, ri as u32));
bb.push(b);
}
boxes.push(bb);
}
// **Band first, Hilbert within the band (DELTA 3).** The leading key is the rung,
// so a row group's `min_zoom` statistics are one rung wide except at a band edge,
// and a reader asking `min_zoom <= camera` skips whole groups off the footer. Inside
// a band the Hilbert key keeps every group spatially compact, so the covering prune
// is unchanged. Stable by construction: the tuple carries (batch, row), so equal
// keys — and the u64::MAX bucket of geometry-less rows — retain input order.
gatling::gatling_sort::gatling_sort_unstable_by(&mut order, |a, b| a.cmp(b));
// ── write ─────────────────────────────────────────────────────────────────
// The generalisation rungs THIS table gets (DELTA 2). Empty for a point-only
// table, and empty when the caller turned them off — the same list drives the
// schema, the `geo` document and the per-row-group column build, so a rung
// cannot be declared and not written, or written and not declared.
let rungs: Vec<u8> = generalised_rungs_for(&in_schema, opts.generalise);
let mut out_schema = pack_schema(&in_schema, opts.covering, &rungs, opts.node_refs_sidecar);
let geo_json = pack_geo_json(&in_schema, opts.covering, &rungs);
// **Row 38: the PACK stamps its own recipe, and the caller cannot forget to.**
//
// `extra_meta` existed from the beginning and every production caller passed `&[]`,
// so no packed table said what produced it and korp's `packed_state` judged a cut on
// structure alone. That is why the live cut served the pre-min-area geometry under a
// Complete verdict until somebody swapped the files by hand. The recipe is derived
// HERE, from the schema this function is about to write and the options it was
// handed, so it is applied output rather than an intention — and a caller that
// stamps nothing still gets one.
//
// A caller's own `extra_meta` is applied AFTER, so it can add keys but the recipe is
// not something it has to remember.
let recipe = PackRecipe::of_written(&out_schema, opts, &rungs).token();
{
// The stamp rides in BOTH places, for the same reason the tile writer does it:
// the Arrow schema copy survives an Arrow-only round trip, the parquet footer
// copy is what a footer-only probe (no row group decoded) can read.
let mut m = out_schema.metadata().clone();
m.insert(String::from(PACK_RECIPE_META_KEY), recipe.clone());
for (k, v) in extra_meta {
m.insert(k.clone(), v.clone());
}
out_schema = Schema::new_with_metadata(out_schema.fields().clone(), m);
}
let out_schema = Arc::new(out_schema);
let mut kv = vec![
parquet::file::metadata::KeyValue::new(String::from(metadata::GEO_KEY), geo_json),
parquet::file::metadata::KeyValue::new(String::from(PACK_RECIPE_META_KEY), recipe),
];
for (k, v) in extra_meta {
kv.push(parquet::file::metadata::KeyValue::new(k.clone(), v.clone()));
}
let mut pb = WriterProperties::builder()
.set_compression(crate::writer::codec(&opts.compression))
.set_max_row_group_row_count(Some(opts.row_group_rows.max(1)))
.set_key_value_metadata(Some(kv));
if opts.covering {
// BYTE_STREAM_SPLIT transposes the four bytes of each f32 into four
// planes. After Hilbert sorting, neighbouring rows share their sign +
// exponent + high mantissa bytes almost exactly, so three of the four
// planes become long runs that zstd flattens. Measured on Stockholm's
// covering columns: 46296340 → 28774887 bytes (−38%). Dictionary encoding
// is disabled on these leaves — the values are near-unique, so a
// dictionary is pure overhead and would pre-empt the split encoding.
for leaf in ["xmin", "ymin", "xmax", "ymax"] {
let path = parquet::schema::types::ColumnPath::new(vec![
String::from(BBOX_COLUMN),
String::from(leaf),
]);
pb = pb
.set_column_encoding(path.clone(), parquet::basic::Encoding::BYTE_STREAM_SPLIT)
.set_column_dictionary_enabled(path, false);
}
}
// **The dictionary comes OFF the near-unique columns** — the same reasoning the
// covering leaves above already state, applied to the two columns that were still
// paying for it. See [`dictionary_is_waste`] for the list and the measurement.
for name in dictionary_is_waste(&out_schema) {
pb = pb.set_column_dictionary_enabled(
parquet::schema::types::ColumnPath::new(vec![name]),
false,
);
}
let props = pb.build();
// Same contract as every writer in `crate::writer`: the packed table is built
// under a `.katana-partial` temp in `dst`'s own directory and only takes the
// output name once the last row group is on disk. A pack that dies half-way
// leaves nothing a directory sweep can mistake for a packed extract.
let (pending, out) = crate::writer::PendingOutput::parquet(dst)?;
let mut w = ArrowWriter::try_new(out, out_schema.clone(), Some(props))?;
// `geom_idx` indexes the INPUT schema; the write loop gathers only the carried
// columns, so the geometry column's position can shift when a derived column ahead
// of it is dropped. Resolved once, from the same list the loop iterates.
let carried_geom = carry_columns(&in_schema, opts.node_refs_sidecar)
.iter()
.position(|&c| c == geom_idx)
.ok_or_else(|| anyhow::anyhow!("`geometry` is not a carried column"))?;
let carried_min_zoom =
min_zoom_idx.and_then(|m| carry_columns(&in_schema, opts.node_refs_sidecar).iter().position(|&c| c == m));
// **Row 22: the `node_refs` sidecar.** Opened only when the column is actually
// leaving the served table — asking for the split on a table that has no
// `node_refs` (every table but `ways`) writes nothing rather than an empty file
// claiming the ids are elsewhere.
let refs_idx = opts
.node_refs_sidecar
.then(|| in_schema.index_of(NODE_REFS_COLUMN).ok())
.flatten();
let id_idx = in_schema.index_of("id").ok();
let mut refs_out = match (refs_idx, id_idx) {
(Some(r), Some(i)) => {
let sc = Arc::new(node_refs_sidecar_schema());
let path = node_refs_sidecar_path(dst);
let props = WriterProperties::builder()
.set_compression(crate::writer::codec(&opts.compression))
.set_max_row_group_row_count(Some(opts.row_group_rows.max(1)))
.build();
let (pend, f) = crate::writer::PendingOutput::parquet(&path)?;
Some((
ArrowWriter::try_new(f, sc.clone(), Some(props))?,
pend,
sc,
r,
i,
))
}
_ => {
anyhow::ensure!(
refs_idx.is_none(),
"`node_refs` is being split into a sidecar but the table has no `id` \
column to key it by — the sidecar would be unjoinable"
);
None
}
};
let mut groups = 0usize;
for chunk in order.chunks(opts.row_group_rows.max(1)) {
let idx = take_indices(chunk, &batches);
// The sidecar is gathered from the SAME `idx`, so it is in the same packed row
// order as the served table — a positional join works, and `relgeom`'s join by
// id works whatever the order.
if let Some((w, _, sc, r, i)) = refs_out.as_mut() {
let cols: Vec<ArrayRef> =
vec![gather(&batches, *i, &idx)?, gather(&batches, *r, &idx)?];
w.write(&RecordBatch::try_new(sc.clone(), cols)?)?;
}
let mut cols: Vec<ArrayRef> = Vec::with_capacity(out_schema.fields().len());
for c in carry_columns(&in_schema, opts.node_refs_sidecar) {
cols.push(gather(&batches, c, &idx)?);
}
if opts.covering {
cols.push(covering_array(chunk, &boxes));
}
// **Derived from the GATHERED geometry column, per row group.** Not from the
// input batches, and not held for the whole table: three simplified copies of
// 9.3 M ways retained until the last row group would be gigabytes of peak on
// top of the 4.8 GB the sort already holds. One row group is 8 192 rows.
//
// The gathered `min_zoom` column rides beside it: a row whose rung is deeper
// than `z` gets NULL in `geometry_z<z>`, not a collapsed point (DELTA 3).
if !rungs.is_empty() {
let base = cols[carried_geom].clone();
let bands = carried_min_zoom.map(|i| cols[i].clone());
for &z in &rungs {
cols.push(generalised_array(&base, bands.as_ref(), z)?);
}
}
w.write(&RecordBatch::try_new(out_schema.clone(), cols)?)?;
groups += 1;
}
w.close()?;
// The sidecar is committed BEFORE the served table takes its name. A reader that
// finds `ways.parquet` without a `node_refs` column must be able to rely on
// `ways.refs.parquet` already being there; the other order leaves a window in which
// the ids look deleted rather than moved.
if let Some((rw, pend, _, _, _)) = refs_out.take() {
rw.close()?;
pend.commit()?;
}
pending.commit()?;
Ok(PackStats {
rows: total,
rows_no_geometry: no_geom,
row_groups: groups,
bytes: fs::metadata(dst).map(|m| m.len()).unwrap_or(0),
})
}
/// Output schema = input schema (+ the covering column), carrying the `geo` key
/// in the Arrow metadata too so an Arrow-only round trip still sees it.
///
/// # The `geometry` column is RE-DECLARED, never carried
///
/// Every other column comes across as the input file described it. `geometry` does not,
/// and that asymmetry is the whole of `geo-strip-1`.
///
/// The `LogicalType::Geometry` declaration lives on the arrow field as the
/// `parquet_geospatial::WkbType` extension type — i.e. as field METADATA. A parquet
/// reader reconstructs the arrow schema from what the FILE says, so a table written
/// before `57aafb3` hands back a bare `Binary` field, and carrying it through gave an
/// output whose three rung columns (built fresh by
/// [`crate::shared::schema::generalised_geometry_field`]) were declared geometry while
/// the base column everything actually reads was not.
///
/// MEASURED on `~/scratch/reader-gen/ways.parquet`, packed 2026-08-31 15:56
/// by a build that HAD `57aafb3` from a `ways.parquet` cut at 06:13 that did not:
///
/// ```text
/// optional binary geometry; <- carried from the input
/// optional binary geometry_z6 (Geometry(crs=srid:0)) <- built fresh here
/// optional binary geometry_z9 (Geometry(crs=srid:0))
/// optional binary geometry_z12 (Geometry(crs=srid:0))
/// is_geo_stats_set(geometry) = False
/// ```
///
/// And it was unrecoverable short of a re-convert: [`is_packed`] answers `true` for that
/// file — it has `covering.bbox` and all three rungs — so no later pack would touch it.
///
/// So `geometry` goes through [`crate::shared::schema::geometry_field`] like every other
/// declaration of it in this crate (LAW 5), which is what makes a re-pack an UPGRADE of
/// the declaration and not just of the layout. `carry_columns`'s doc already claims
/// exactly that for the derived columns; this is the same claim honoured for the base one.
fn pack_schema(input: &Schema, covering: bool, rungs: &[u8], split_refs: bool) -> Schema {
let mut fields: Vec<Arc<Field>> = carry_columns(input, split_refs)
.into_iter()
.map(|i| {
let f = &input.fields()[i];
if f.name() == "geometry" {
Arc::new(crate::shared::schema::geometry_field())
} else {
f.clone()
}
})
.collect();
if covering {
fields.push(Arc::new(bbox_field()));
}
// Appended, never interleaved: every existing reader that resolves the base
// columns by POSITION keeps working, and the write loop can push them last.
for &z in rungs {
fields.push(Arc::new(crate::shared::schema::generalised_geometry_field(
z,
)));
}
let mut meta: HashMap<String, String> = input.metadata().clone();
meta.insert(
String::from(metadata::GEO_KEY),
pack_geo_json(input, covering, rungs),
);
Schema::new_with_metadata(fields, meta)
}
/// **Which generalisation rungs this table gets** — the ONE decision (LAW 5), read by
/// the schema, the `geo` document, the writer, and by [`crate::network::NetworkRecipe`],
/// which must name the columns in its staleness token or a table cut before DELTA 2 and
/// one cut after would carry the same token.
///
/// `[]` when the caller opted out, and `[]` for a table whose declared
/// `geometry_types` are all points: a point has no vertex Douglas–Peucker can drop, so
/// the three columns would be byte-for-byte copies of `geometry`. MEASURED on the
/// Sweden cut: that is **+1.97 GB** on `nodes.parquet` (655.18 MB of compressed point
/// geometry, three times) buying a reader nothing it could not get from the base
/// column. See [`crate::shared::generalise::table_generalises`].
pub fn generalised_rungs_for(input: &Schema, generalise: bool) -> Vec<u8> {
if !generalise {
return Vec::new();
}
let types = input
.metadata()
.get(metadata::GEO_KEY)
.and_then(|j| geometry_types_of(j))
.unwrap_or_default();
if crate::shared::generalise::table_generalises(&types) {
crate::shared::schema::GENERALISED_RUNGS.to_vec()
} else {
Vec::new()
}
}
/// **One generalisation column for one row group**, simplified out of the already
/// gathered `geometry` column at rung `z`'s epsilon.
///
/// Fans out on gatling (ROOT LAW #0 — no rayon, no hand-rolled pool) over contiguous
/// row spans, each worker holding its own Douglas–Peucker scratch for the whole span:
/// the keep-bitmap and index stack are what `tilemem-2` measured at 11.13 GiB of glibc
/// arena free list when they were allocated per call, and there are THREE rungs per row
/// here. Spans come back in order, so the column is reproducible.
///
/// A row whose geometry is null, or whose WKB does not decode, gets a NULL — never an
/// empty blob, so "I could not read this" and "this feature has no shape at this rung"
/// cannot wear the same cell.
fn generalised_array(
geometry: &ArrayRef,
min_zoom: Option<&ArrayRef>,
z: u8,
) -> anyhow::Result<ArrayRef> {
use arrow::array::BinaryBuilder;
let g = geometry
.as_any()
.downcast_ref::<BinaryArray>()
.ok_or_else(|| anyhow::anyhow!("the `geometry` column is not Binary"))?;
// **A feature the ladder withholds at this rung has NO cell here (DELTA 3).** The
// sub-pixel rule (a shape that simplifies below a ring degrades to a POINT, never a
// deletion) still holds for every row the rung ADMITS; this is the other question —
// whether the row may draw at `z` at all — answered by `min_zoom`, the one column
// that decides it. Without it a z6 camera reading `geometry_z6` decoded a point for
// every service road in Sweden: 5 609 616 of 9 323 619 rows as bare `Point`s
// (measured 2026-08-31), each one a feature no z6 painter draws.
let bands = min_zoom
.map(|a| {
a.as_any()
.downcast_ref::<UInt8Array>()
.ok_or_else(|| anyhow::anyhow!("the `min_zoom` column is not UInt8"))
})
.transpose()?;
let eps = crate::shared::generalise::generalisation_eps(z);
let n = g.len();
// ~256 rows per span: enough that a span's DP dominates its dispatch, small enough
// that a row group of 8 192 still fans out across the box.
const SPAN: usize = 256;
let spans: Vec<(usize, usize)> = (0..n)
.step_by(SPAN)
.map(|a| (a, (a + SPAN).min(n)))
.collect();
/// One span's contribution: values concatenated, per-row length, per-row validity.
struct Span {
vals: Vec<u8>,
lens: Vec<u32>,
valid: Vec<bool>,
}
let parts: Vec<Span> = gatling::gatling_forkjoin::gatling_map_balanced(
&spans,
0,
1,
|&(a, b): &(usize, usize)| (b - a) as u64,
|_i, &(a, b): &(usize, usize)| {
let mut scratch = crate::tiles::DpScratch::default();
let mut thin: Vec<(f64, f64)> = Vec::new();
let mut buf: Vec<u8> = Vec::new();
let mut out = Span {
vals: Vec::new(),
lens: Vec::with_capacity(b - a),
valid: Vec::with_capacity(b - a),
};
for r in a..b {
let admitted = bands.is_none_or(|m| m.value(r) <= z);
let ok = admitted
&& !g.is_null(r)
&& crate::shared::generalise::generalise_wkb_into(
g.value(r),
eps,
&mut scratch,
&mut thin,
&mut buf,
);
if ok {
out.vals.extend_from_slice(&buf);
#[allow(clippy::cast_possible_truncation)]
out.lens.push(buf.len() as u32);
} else {
out.lens.push(0);
}
out.valid.push(ok);
}
// The ring-safe simplifier's books, ONCE for this span (row 24) — the retry
// loop itself touches no shared cell.
crate::tiles::fold_selfx_books(scratch.books);
out
},
);
let total: usize = parts.iter().map(|p| p.vals.len()).sum();
let mut b = BinaryBuilder::with_capacity(n, total);
for p in &parts {
let mut off = 0usize;
for (i, &len) in p.lens.iter().enumerate() {
if p.valid[i] {
let len = len as usize;
b.append_value(&p.vals[off..off + len]);
off += len;
} else {
b.append_null();
}
}
}
Ok(Arc::new(b.finish()))
}
/// The `geo` document for the packed table: the input's geometry types (parsed
/// back out of its own `geo` key, so a repack never invents types) plus the
/// `covering` declaration.
fn pack_geo_json(input: &Schema, covering: bool, rungs: &[u8]) -> String {
let types = input
.metadata()
.get(metadata::GEO_KEY)
.and_then(|j| geometry_types_of(j))
.unwrap_or_else(|| vec![String::from("Point")]);
let refs: Vec<&str> = types.iter().map(String::as_str).collect();
let names: Vec<String> = rungs
.iter()
.map(|&z| crate::shared::schema::generalised_geometry_name(z))
.collect();
let mut m = metadata::GeoMeta::new(&refs);
if covering {
m = m.with_covering(BBOX_COLUMN);
}
// The generalisation columns are GeoParquet columns in their own right: a reader
// that trusts the `geo` document rather than the parquet logical type must find
// them there, or it will treat `geometry_z12` as opaque binary.
m = m.with_secondary_wkb_columns(
&names,
crate::shared::generalise::GENERALISED_GEOMETRY_TYPES,
);
m.to_json()
}
/// Pull `geometry_types` out of a `geo` JSON document without a JSON dependency
/// in the hot path (the array is a flat list of quoted strings).
fn geometry_types_of(geo_json: &str) -> Option<Vec<String>> {
let at = geo_json.find("\"geometry_types\"")?;
let rest = &geo_json[at..];
let open = rest.find('[')?;
let close = rest.find(']')?;
if close < open {
return None;
}
let inner = &rest[open + 1..close];
let out: Vec<String> = inner
.split(',')
.filter_map(|s| {
let s = s.trim();
s.strip_prefix('"')?.strip_suffix('"').map(String::from)
})
.collect();
if out.is_empty() { None } else { Some(out) }
}
/// One row's sort key: `(min_zoom, hilbert, batch, row)`. Tuple order IS the write
/// order — band first, Hilbert within the band, input order among equals.
type PackKey = (u8, u64, u32, u32);
/// Group a row-group's worth of keys into per-batch take lists.
fn take_indices(chunk: &[PackKey], batches: &[RecordBatch]) -> Vec<Vec<u32>> {
let mut idx: Vec<Vec<u32>> = vec![Vec::new(); batches.len()];
for &(_, _, b, r) in chunk {
idx[b as usize].push(r);
}
idx
}
/// **Give every batch a `min_zoom` column when the input has none** — from `tags`,
/// through [`crate::shared::class::min_zoom_of_tags`], the one tag → rung writer.
///
/// Three cases, and the output schema of each:
///
/// * `min_zoom` present — batches and schema returned untouched. The value convert wrote
/// is the value; a re-pack does not second-guess it.
/// * `min_zoom` absent, `tags` present — the column is computed per row (a NULL `tags`
/// cell is `{}`, the same spelling the writers give an untagged feature) and
/// appended LAST, where `ways_schema` puts it.
/// * neither — nothing to classify from; returned untouched. The pack then sorts on one
/// band and writes no rung NULLs, i.e. it is the DELTA 2 pack, and `is_packed` does
/// not demand the column of a table that has no `tags`.
fn ensure_min_zoom(
batches: Vec<RecordBatch>,
in_schema: Arc<Schema>,
) -> anyhow::Result<(Vec<RecordBatch>, Arc<Schema>)> {
if in_schema.index_of(MIN_ZOOM_COLUMN).is_ok() {
return Ok((batches, in_schema));
}
let Ok(tags_idx) = in_schema.index_of(TAGS_COLUMN) else {
return Ok((batches, in_schema));
};
if !matches!(in_schema.field(tags_idx).data_type(), DataType::Utf8) {
anyhow::bail!(
"`tags` is {:?}, not Utf8 — cannot derive `min_zoom` from it",
in_schema.field(tags_idx).data_type()
);
}
let mut fields: Vec<Arc<Field>> = in_schema.fields().iter().cloned().collect();
fields.push(Arc::new(crate::shared::schema::min_zoom_field()));
let schema = Arc::new(Schema::new_with_metadata(fields, in_schema.metadata().clone()));
// Per batch on gatling (ROOT LAW #0): 9.3 M tag scans on the Sweden way table.
let columns: Vec<ArrayRef> = gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| -> ArrayRef {
let tags = b.column(tags_idx).as_any().downcast_ref::<StringArray>();
let vals: Vec<u8> = (0..b.num_rows())
.map(|r| match tags {
Some(t) if !t.is_null(r) => crate::shared::class::min_zoom_of_tags(t.value(r)),
_ => crate::shared::class::min_zoom_of_tags("{}"),
})
.collect();
Arc::new(UInt8Array::from(vals))
},
);
let mut out = Vec::with_capacity(batches.len());
for (b, mz) in batches.into_iter().zip(columns) {
let mut cols: Vec<ArrayRef> = b.columns().to_vec();
cols.push(mz);
out.push(RecordBatch::try_new(schema.clone(), cols)?);
}
Ok((out, schema))
}
/// **Give every batch the CLASS BLOCK when the input has none** — from `tags`, through
/// [`crate::shared::class::class_of_tags`], the one tag → class writer (row 18).
///
/// [`ensure_min_zoom`]'s shape, and deliberately its shape: a table cut before the
/// columns existed (every cut before 2026-09-05, including the live Sweden one) carries
/// `tags` and nothing else, so a re-pack can recover the whole block without re-cutting
/// the planet. Columns already present are kept — the value convert wrote is the value,
/// and a re-pack does not second-guess it.
///
/// The six are appended in [`crate::shared::schema::class_fields`] order, which is where
/// the base-table schemas put them, so a re-pack and a convert-then-pack agree on layout.
/// A table with no `tags` at all (the graph tables) is returned untouched: there is
/// nothing to classify from, and a block of invented values would be worse than an
/// absent one.
fn ensure_class_columns(
batches: Vec<RecordBatch>,
in_schema: Arc<Schema>,
) -> anyhow::Result<(Vec<RecordBatch>, Arc<Schema>)> {
let missing: Vec<Field> = missing_class_fields(&in_schema);
if missing.is_empty() {
return Ok((batches, in_schema));
}
let Ok(tags_idx) = in_schema.index_of(TAGS_COLUMN) else {
return Ok((batches, in_schema));
};
if !matches!(in_schema.field(tags_idx).data_type(), DataType::Utf8) {
anyhow::bail!(
"`tags` is {:?}, not Utf8 — cannot derive the class columns from it",
in_schema.field(tags_idx).data_type()
);
}
let mut fields: Vec<Arc<Field>> = in_schema.fields().iter().cloned().collect();
fields.extend(missing.iter().cloned().map(Arc::new));
let schema = Arc::new(Schema::new_with_metadata(fields, in_schema.metadata().clone()));
let names: Vec<String> = missing.iter().map(|f| f.name().clone()).collect();
// Per batch on gatling (ROOT LAW #0), one pass over `tags` for all six columns —
// the same scan `class_of_tags` makes for a fresh cut.
let per_batch: Vec<Vec<ArrayRef>> = gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| -> Vec<ArrayRef> {
let tags = b.column(tags_idx).as_any().downcast_ref::<StringArray>();
let rows: Vec<crate::shared::class::RowClass> = (0..b.num_rows())
.map(|r| match tags {
Some(t) if !t.is_null(r) => crate::shared::class::class_of_tags(t.value(r)),
_ => crate::shared::class::class_of_tags("{}"),
})
.collect();
names
.iter()
.map(|n| -> ArrayRef {
match n.as_str() {
"class_id" => {
Arc::new(UInt16Array::from_iter_values(rows.iter().map(|c| c.class_id)))
}
"kind" => {
Arc::new(UInt8Array::from_iter_values(rows.iter().map(|c| c.kind)))
}
"layer" => {
Arc::new(Int8Array::from_iter_values(rows.iter().map(|c| c.layer)))
}
"bridge" => {
Arc::new(BooleanArray::from_iter(rows.iter().map(|c| Some(c.bridge))))
}
"tunnel" => {
Arc::new(BooleanArray::from_iter(rows.iter().map(|c| Some(c.tunnel))))
}
"oneway" => {
Arc::new(Int8Array::from_iter_values(rows.iter().map(|c| c.oneway)))
}
// Unreachable by construction: `names` is filtered from
// `class_fields()`, and every field it declares has an arm above.
other => unreachable!("no builder for class column `{other}`"),
}
})
.collect()
},
);
let mut out = Vec::with_capacity(batches.len());
for (b, added) in batches.into_iter().zip(per_batch) {
let mut cols: Vec<ArrayRef> = b.columns().to_vec();
cols.extend(added);
out.push(RecordBatch::try_new(schema.clone(), cols)?);
}
Ok((out, schema))
}
/// **Deepen `min_zoom` for sub-pixel AREAS** — the size half of the rung decision,
/// beside [`ensure_min_zoom`]'s tag half, and running on a PRESENT column too (a
/// re-pack of an already-banded table must still learn the rule).
///
/// Only Polygon/MultiPolygon rows move, only DEEPER (`max`, never a promotion), and
/// the floor comes from [`crate::shared::generalise::area_rung_floor`] over the
/// feature's own bbox extent — the one epsilon basis. Lines are untouched: a long thin
/// road simplifies fine at any rung, and a degraded LINE is a real path a coarse view
/// legitimately shows as a run.
fn deepen_area_min_zoom(
batches: Vec<RecordBatch>,
schema: &Arc<Schema>,
) -> anyhow::Result<Vec<RecordBatch>> {
let Ok(mz_idx) = schema.index_of(MIN_ZOOM_COLUMN) else {
return Ok(batches);
};
let Ok(geom_idx) = schema.index_of("geometry") else {
return Ok(batches);
};
let out: Vec<RecordBatch> = gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| -> RecordBatch {
let g = b.column(geom_idx).as_any().downcast_ref::<BinaryArray>();
let mz = b.column(mz_idx).as_any().downcast_ref::<UInt8Array>();
let (Some(g), Some(mz)) = (g, mz) else {
return b.clone();
};
let vals: Vec<u8> = (0..b.num_rows())
.map(|r| {
let z = if mz.is_null(r) { 0 } else { mz.value(r) };
if g.is_null(r) {
return z;
}
let w = g.value(r);
// The WKB type, honouring the byte-order byte; %1000 strips Z/M.
let ty = w.first().and_then(|&bo| {
let b4 = w.get(1..5)?;
let raw = if bo == 1 {
u32::from_le_bytes([b4[0], b4[1], b4[2], b4[3]])
} else {
u32::from_be_bytes([b4[0], b4[1], b4[2], b4[3]])
};
Some(raw % 1000)
});
if !matches!(ty, Some(3 | 6)) {
return z;
}
match wkb_bbox(w) {
Some(bb) => z.max(crate::shared::generalise::area_rung_floor(
(bb.xmax - bb.xmin).max(bb.ymax - bb.ymin),
)),
None => z,
}
})
.collect();
let mut cols = b.columns().to_vec();
cols[mz_idx] = Arc::new(UInt8Array::from(vals)) as ArrayRef;
RecordBatch::try_new(b.schema(), cols).expect("same shape, one column swapped")
},
);
Ok(out)
}
/// Gather column `col` across batches in the row-group's order.
///
/// Rows are taken per source batch with Arrow's `take` (vectorised) and the
/// per-batch pieces concatenated, then re-ordered — the concatenation order is
/// the same interleave `covering_array` uses, so the derived column stays aligned
/// with the data columns.
fn gather(batches: &[RecordBatch], col: usize, idx: &[Vec<u32>]) -> anyhow::Result<ArrayRef> {
let mut parts: Vec<ArrayRef> = Vec::new();
for (b, rows) in batches.iter().zip(idx) {
if rows.is_empty() {
continue;
}
let take = arrow::array::UInt32Array::from(rows.clone());
parts.push(arrow::compute::take(b.column(col).as_ref(), &take, None)?);
}
let refs: Vec<&dyn Array> = parts.iter().map(|a| a.as_ref()).collect();
Ok(arrow::compute::concat(&refs)?)
}
/// The covering column for one row group, in the SAME interleave order [`gather`]
/// produces (all of batch 0's picks, then batch 1's, …).
fn covering_array(chunk: &[PackKey], boxes: &[Vec<Option<Bbox>>]) -> ArrayRef {
let mut by_batch: Vec<Vec<u32>> = vec![Vec::new(); boxes.len()];
for &(_, _, b, r) in chunk {
by_batch[b as usize].push(r);
}
let n = chunk.len();
let (mut xmin, mut ymin) = (
Float32Builder::with_capacity(n),
Float32Builder::with_capacity(n),
);
let (mut xmax, mut ymax) = (
Float32Builder::with_capacity(n),
Float32Builder::with_capacity(n),
);
for (bi, rows) in by_batch.iter().enumerate() {
for &r in rows {
match boxes[bi][r as usize] {
Some(b) => {
// Outward rounding — the stored box must CONTAIN the true box.
xmin.append_value(down(b.xmin));
ymin.append_value(down(b.ymin));
xmax.append_value(up(b.xmax));
ymax.append_value(up(b.ymax));
}
None => {
xmin.append_null();
ymin.append_null();
xmax.append_null();
ymax.append_null();
}
}
}
}
let DataType::Struct(fields) = bbox_field().data_type().clone() else {
unreachable!("bbox_field is a Struct")
};
Arc::new(StructArray::new(
fields,
vec![
Arc::new(xmin.finish()) as ArrayRef,
Arc::new(ymin.finish()),
Arc::new(xmax.finish()),
Arc::new(ymax.finish()),
],
None,
))
}
/// Pack every GeoParquet table in a convert output directory.
pub fn spatial_pack_dir(
src: &Path,
dst: &Path,
opts: &PackOptions,
) -> anyhow::Result<Vec<(String, PackStats)>> {
fs::create_dir_all(dst)?;
let mut out = Vec::new();
for name in BASE_TABLES {
let p = src.join(name);
if !p.exists() {
continue;
}
let stats = spatial_pack(&p, &dst.join(name), opts)?;
out.push((String::from(*name), stats));
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use arrow::array::{BinaryBuilder, Int64Builder};
/// Locality is the whole point: two points a few metres apart must land near
/// each other in Hilbert index, and a point on another continent must not.
/// RED if the curve is ever replaced by something non-locality-preserving
/// (e.g. a raw interleave of the two coordinates' high bits, or plain
/// longitude order).
#[test]
fn hilbert_preserves_locality() {
let stockholm = hilbert_index(18.0686, 59.3293);
let next_block = hilbert_index(18.0696, 59.3298); // ~ 70 m away
let gothenburg = hilbert_index(11.9746, 57.7089); // ~ 400 km
let sydney = hilbert_index(151.2093, -33.8688); // other hemisphere
let d_near = stockholm.abs_diff(next_block);
let d_far = stockholm.abs_diff(gothenburg);
let d_world = stockholm.abs_diff(sydney);
assert!(
d_near < d_far,
"a neighbouring block must be closer in index than another city"
);
assert!(
d_far < d_world,
"another city must be closer than another continent"
);
// And the ordering must be a bijection on the grid — distinct cells,
// distinct indices.
assert_ne!(stockholm, next_block);
assert_eq!(hilbert_index(18.0686, 59.3293), stockholm, "deterministic");
}
/// The curve must cover the grid exactly once — a permutation, not a hash.
#[test]
fn hilbert_is_a_bijection_on_a_small_grid() {
// Exercise the real 31-bit curve but only over the coarsest 4x4 quadrants
// (shift the cell into the top bits), which is where the rotation logic
// lives.
let sh = HILBERT_BITS - 2;
let mut seen: Vec<u64> = Vec::new();
for y in 0..4u32 {
for x in 0..4u32 {
seen.push(hilbert_d(x << sh, y << sh));
}
}
let mut sorted = seen.clone();
sorted.sort_unstable();
sorted.dedup();
assert_eq!(
sorted.len(),
16,
"16 distinct quadrants must give 16 distinct indices"
);
// Consecutive Hilbert cells are grid-adjacent — the defining property.
let mut cells: Vec<(u64, u32, u32)> = Vec::new();
for y in 0..4u32 {
for x in 0..4u32 {
cells.push((hilbert_d(x << sh, y << sh), x, y));
}
}
cells.sort_unstable();
for w in cells.windows(2) {
let (_, x0, y0) = w[0];
let (_, x1, y1) = w[1];
let manhattan = x0.abs_diff(x1) + y0.abs_diff(y1);
assert_eq!(
manhattan, 1,
"consecutive Hilbert cells must be grid neighbours"
);
}
}
/// WKB bbox must handle every shape osm-katana writes, in both byte orders.
#[test]
fn wkb_bbox_covers_the_written_geometry_types() {
let p = crate::geometry::encode_point_inline(18.0686, 59.3293);
let bb = wkb_bbox(&p).expect("point must parse");
assert_eq!(bb.xmin, 18.0686);
assert_eq!(bb.xmax, 18.0686);
assert_eq!(bb.ymin, 59.3293);
assert_eq!(bb.ymax, 59.3293);
// LineString with three points → the enclosing box.
let mut ls: Vec<u8> = vec![1];
ls.extend_from_slice(&2u32.to_le_bytes());
ls.extend_from_slice(&3u32.to_le_bytes());
for (x, y) in [(1.0f64, 5.0f64), (3.0, 2.0), (-1.0, 4.0)] {
ls.extend_from_slice(&x.to_le_bytes());
ls.extend_from_slice(&y.to_le_bytes());
}
let bb = wkb_bbox(&ls).unwrap();
assert_eq!((bb.xmin, bb.ymin, bb.xmax, bb.ymax), (-1.0, 2.0, 3.0, 5.0));
// Big-endian point — the byte-order flag must actually be honoured.
let mut be: Vec<u8> = vec![0];
be.extend_from_slice(&1u32.to_be_bytes());
be.extend_from_slice(&7.5f64.to_be_bytes());
be.extend_from_slice(&(-2.5f64).to_be_bytes());
let bb = wkb_bbox(&be).unwrap();
assert_eq!((bb.xmin, bb.ymin), (7.5, -2.5));
// RED-when-broken: junk must be rejected, not silently boxed at 0,0
// (a 0,0 bbox would poison every row group's statistics).
assert!(wkb_bbox(&[]).is_none());
assert!(
wkb_bbox(&[1, 9, 9, 9, 9]).is_none(),
"unknown WKB type must not parse"
);
assert!(
wkb_bbox(&p[..10]).is_none(),
"truncated point must not parse"
);
}
/// A covering must CONTAIN the geometry it covers. `f32` cannot represent an
/// OSM 1e-7° coordinate exactly, so the rounding direction is load-bearing:
/// round-to-nearest would shrink roughly half of all boxes and a reader could
/// then prune away a row group holding a real hit.
///
/// RED-when-broken: replace `down`/`up` with `as f32` and this fails on the
/// first coordinate whose nearest f32 lands on the wrong side.
#[test]
fn covering_is_conservative() {
let mut shrunk = 0usize;
// Real OSM-grid coordinates across Stockholm, at 1e-7° resolution.
for k in 0..20_000i64 {
let lon = 17.5 + (k as f64) * 1e-7;
let lat = 59.1 + (k as f64) * 1e-7;
assert!(
f64::from(down(lon)) <= lon,
"xmin must round DOWN past {lon}"
);
assert!(f64::from(up(lon)) >= lon, "xmax must round UP past {lon}");
assert!(
f64::from(down(lat)) <= lat,
"ymin must round DOWN past {lat}"
);
assert!(f64::from(up(lat)) >= lat, "ymax must round UP past {lat}");
if (lon as f32) as f64 > lon {
shrunk += 1; // naive cast would have shrunk this box
}
}
assert!(
shrunk > 1000,
"the fixture must actually exercise the case a naive cast gets wrong \
(only {shrunk} of 20000 did)"
);
// Negative coordinates and zero must round outwards too.
for v in [-18.0685_f64, -0.000_000_3, 0.0, 179.999_999_9] {
assert!(f64::from(down(v)) <= v, "down({v})");
assert!(f64::from(up(v)) >= v, "up({v})");
}
}
/// **THE quantisation guard, on APPLIED OUTPUT** — the leaves a packed file
/// actually holds, not what `down`/`up` return in isolation.
///
/// Three things at once, because any one of them alone can pass while the
/// writer is broken:
///
/// 1. every emitted leaf lands EXACTLY on the [`COVERING_GRID_SHIFT`] grid —
/// this is what buys the 349.48 MB, and it is the half a size check cannot
/// see (a file can be small for a dozen reasons);
/// 2. every emitted box still CONTAINS its own point — the quantisation must be
/// outward, or the reader prunes away real hits;
/// 3. the box is no WIDER than one grid step per axis — a covering that silently
/// grew would also "land on the grid" and would also "contain the point",
/// and it would kill the prune this whole column exists for.
///
/// And the anti-blindness clause (LAW 2, the identity-value trap): the fixture
/// must actually be MOVED by the grid. Coordinates that already sit on it would
/// let a dead quantiser pass every assertion above.
///
/// RED-when-broken, proven: make [`down`]/[`up`] the plain `as f32` cast they
/// used to be and assertion 1 fires on the first row.
#[test]
#[allow(clippy::float_arithmetic, clippy::float_cmp)]
fn covering_leaves_land_on_the_grid_and_still_contain_the_point() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("in.parquet");
let dst = dir.path().join("out.parquet");
// 1e-7° OSM-grid coordinates, deliberately NOT multiples of 2^-12°.
let pts: Vec<(i64, f64, f64)> = (0..2_000i64)
.map(|k| {
(
k,
18.0 + (k as f64) * 1.7e-5,
59.0 + (k as f64) * 1.1e-5,
)
})
.collect();
tiny_table(&src, &pts);
let opts = PackOptions {
row_group_rows: 256,
compression: String::from("none"),
..Default::default()
};
spatial_pack(&src, &dst, &opts).unwrap();
let rdr = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(&dst).unwrap())
.unwrap()
.with_batch_size(1024)
.build()
.unwrap();
let step = 1.0 / f64::from(1u32 << COVERING_GRID_SHIFT);
let (mut rows, mut moved) = (0usize, 0usize);
for batch in rdr {
let batch = batch.unwrap();
let geom = batch
.column_by_name("geometry")
.unwrap()
.as_any()
.downcast_ref::<arrow::array::BinaryArray>()
.unwrap()
.clone();
let bb = batch
.column_by_name(BBOX_COLUMN)
.unwrap()
.as_any()
.downcast_ref::<arrow::array::StructArray>()
.unwrap()
.clone();
let leaf = |n: &str| {
bb.column_by_name(n)
.unwrap()
.as_any()
.downcast_ref::<arrow::array::Float32Array>()
.unwrap()
.clone()
};
let (xmin, ymin, xmax, ymax) =
(leaf("xmin"), leaf("ymin"), leaf("xmax"), leaf("ymax"));
for i in 0..geom.len() {
let truth = wkb_bbox(geom.value(i)).expect("point must parse");
let (lo_x, lo_y) = (f64::from(xmin.value(i)), f64::from(ymin.value(i)));
let (hi_x, hi_y) = (f64::from(xmax.value(i)), f64::from(ymax.value(i)));
// (1) on the grid, exactly.
for (name, v) in [
("xmin", xmin.value(i)),
("ymin", ymin.value(i)),
("xmax", xmax.value(i)),
("ymax", ymax.value(i)),
] {
assert!(
on_covering_grid(v),
"row {i} leaf bbox.{name} = {v} is not a multiple of 2^-{} — \
the covering is not quantised, so the low mantissa plane is \
still being paid for",
COVERING_GRID_SHIFT
);
}
// (2) still contains its own geometry.
assert!(
lo_x <= truth.xmin && hi_x >= truth.xmax,
"row {i}: [{lo_x}, {hi_x}] does not contain x {}",
truth.xmin
);
assert!(
lo_y <= truth.ymin && hi_y >= truth.ymax,
"row {i}: [{lo_y}, {hi_y}] does not contain y {}",
truth.ymin
);
// (3) and it grew by at most one grid step per axis.
assert!(
hi_x - lo_x <= step * 1.5 && hi_y - lo_y <= step * 1.5,
"row {i}: a POINT's covering spans {:e} x {:e}°, more than one \
{step:e}° grid step — a covering that widens silently prunes nothing",
hi_x - lo_x,
hi_y - lo_y
);
// The anti-blindness clause: the grid must have MOVED this point.
if lo_x != truth.xmin || hi_x != truth.xmax {
moved += 1;
}
rows += 1;
}
}
assert_eq!(rows, 2_000, "every row must have been read back");
assert!(
moved * 10 > rows * 9,
"only {moved} of {rows} coverings were moved by the grid — the fixture \
already sits on it, so this test cannot tell a working quantiser from a \
dead one"
);
}
fn tiny_table(path: &Path, pts: &[(i64, f64, f64)]) {
let schema = Arc::new(crate::shared::schema::nodes_schema());
let mut id = Int64Builder::new();
let mut g = BinaryBuilder::new();
for &(i, lon, lat) in pts {
id.append_value(i);
g.append_value(crate::geometry::encode_point_inline(lon, lat));
}
let n = pts.len();
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
arrow::array::new_null_array(&DataType::Utf8, n),
arrow::array::new_null_array(&DataType::Int32, n),
];
// DELTA 1 + row 18: the whole class block is non-null on every base table. These
// rows carry no tags at all, so they take the values the ladder names for "no
// rule matched" — through the writers' own builder, not a second opinion.
let mut cls = crate::shared::schema::ClassColumns::new();
for _ in 0..n {
cls.push("{}");
}
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let f = fs::File::create(path).unwrap();
let mut w = ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
/// THE deliverable, in miniature: after packing, each row group's `bbox.xmin`
/// / `bbox.xmax` statistics must describe a COMPACT region, not the whole
/// file — that is what makes pruning possible. Also asserts the covering
/// metadata is declared and no row is lost.
#[test]
fn pack_makes_row_groups_compact_and_declares_covering() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("in.parquet");
let dst = dir.path().join("out.parquet");
// Four well-separated clusters, INTERLEAVED on input — exactly the
// pathology convert produces (parse order ≠ spatial order).
let clusters = [
(18.06, 59.33),
(11.97, 57.71),
(13.00, 55.60),
(17.64, 59.85),
];
let mut pts = Vec::new();
for k in 0..64i64 {
let (lon, lat) = clusters[(k % 4) as usize];
let j = (k / 4) as f64 * 1e-4;
pts.push((k, lon + j, lat + j));
}
tiny_table(&src, &pts);
let opts = PackOptions {
row_group_rows: 16,
..Default::default()
};
let st = spatial_pack(&src, &dst, &opts).unwrap();
assert_eq!(st.rows, 64);
assert_eq!(st.rows_no_geometry, 0);
assert_eq!(st.row_groups, 4);
let f = fs::File::open(&dst).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let md = b.metadata().clone();
// The `covering` declaration must be in the FILE-level key-value metadata.
let kv = md
.file_metadata()
.key_value_metadata()
.expect("kv metadata");
let geo = kv
.iter()
.find(|k| k.key == "geo")
.and_then(|k| k.value.clone())
.expect("geo key");
assert!(
geo.contains("\"covering\""),
"packed file must declare covering.bbox"
);
// Locate the bbox.xmin / bbox.xmax leaves and read each group's stats.
let sch = md.file_metadata().schema_descr();
let (mut lo_i, mut hi_i) = (usize::MAX, usize::MAX);
for i in 0..sch.num_columns() {
match sch.column(i).path().string().as_str() {
"bbox.xmin" => lo_i = i,
"bbox.xmax" => hi_i = i,
_ => {}
}
}
assert_ne!(lo_i, usize::MAX, "bbox.xmin leaf must exist");
let mut widths = Vec::new();
for g in 0..md.num_row_groups() {
let lo = md.row_group(g).column(lo_i).statistics().unwrap();
let hi = md.row_group(g).column(hi_i).statistics().unwrap();
// FLOAT per `bbox_field` (4.6x smaller than DOUBLE, and the spec
// permits either). The reader side must therefore understand
// `Statistics::Float` — skade's `row_group_survives` gained that arm
// alongside this change; without it the covering prunes nothing.
let (
parquet::file::statistics::Statistics::Float(l),
parquet::file::statistics::Statistics::Float(h),
) = (lo, hi)
else {
panic!("covering statistics must be FLOAT (see bbox_field)");
};
widths.push(f64::from(*h.max_opt().unwrap()) - f64::from(*l.min_opt().unwrap()));
}
let file_width = 18.06 - 11.97;
for (g, w) in widths.iter().enumerate() {
assert!(
*w < file_width / 10.0,
"row group {g} spans {w}° — Hilbert packing must make it far tighter \
than the file's {file_width}°; unsorted input gives every group the full span"
);
}
// Content is preserved: same ids, same count.
let rdr = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(&dst).unwrap())
.unwrap()
.build()
.unwrap();
let mut ids: Vec<i64> = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let a = batch
.column(0)
.as_any()
.downcast_ref::<arrow::array::Int64Array>()
.unwrap();
ids.extend((0..a.len()).map(|i| a.value(i)));
}
assert_ne!(
ids,
(0..64).collect::<Vec<_>>(),
"the order MUST have changed (that is the point)"
);
ids.sort_unstable();
assert_eq!(
ids,
(0..64).collect::<Vec<_>>(),
"…but not one row may be lost or duplicated"
);
}
/// Rows with null/garbage geometry are kept, sorted last, and get a NULL
/// covering — never a fake 0,0 box that would wreck the statistics.
#[test]
fn geometryless_rows_survive_and_sort_last() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("in.parquet");
let dst = dir.path().join("out.parquet");
let schema = Arc::new(crate::shared::schema::nodes_schema());
let mut id = Int64Builder::new();
let mut g = BinaryBuilder::new();
for k in 0..10i64 {
id.append_value(k);
if k % 2 == 0 {
g.append_value(crate::geometry::encode_point_inline(
18.0 + k as f64 * 1e-3,
59.0,
));
} else {
g.append_null();
}
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
arrow::array::new_null_array(&DataType::Utf8, 10),
arrow::array::new_null_array(&DataType::Int32, 10),
];
// The class block, from the ONE writer — these rows carry no tags.
let mut cls = crate::shared::schema::ClassColumns::new();
for _ in 0..10 {
cls.push("{}");
}
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let f = fs::File::create(&src).unwrap();
let mut w = ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
let st = spatial_pack(&src, &dst, &PackOptions::default()).unwrap();
assert_eq!(st.rows, 10, "no row may be dropped");
assert_eq!(st.rows_no_geometry, 5);
let rdr = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(&dst).unwrap())
.unwrap()
.build()
.unwrap();
let mut ids: Vec<i64> = Vec::new();
let mut bbox_null: Vec<bool> = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let a = batch
.column(0)
.as_any()
.downcast_ref::<arrow::array::Int64Array>()
.unwrap();
ids.extend((0..a.len()).map(|i| a.value(i)));
let bb = batch
.column_by_name("bbox")
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap()
.column(0)
.clone();
bbox_null.extend((0..bb.len()).map(|i| bb.is_null(i)));
}
assert_eq!(
&ids[5..],
&[1, 3, 5, 7, 9],
"geometry-less rows sort last, in input order"
);
assert_eq!(
&bbox_null[5..],
&[true; 5],
"…with a NULL covering, not a fabricated 0,0 box"
);
assert_eq!(&bbox_null[..5], &[false; 5]);
}
}
/// **FAIL-ON-BUG for the base tables' layout** — the three files convert and
/// `relations` emit, held to the same standard `network.rs` already holds
/// `roads.parquet` to.
///
/// # Why this module exists at all
///
/// MEASURED on korp's live Sweden cut, 2026-08-31, before any of this landed:
/// `ways.parquet` was **9 325 142 rows in 32 row groups with no `bbox` column at
/// all**. Nothing about that file is invalid — it is well-formed GeoParquet, every
/// row is correct, and `verify` passed on it. It simply cannot be pruned, so the
/// central-Stockholm viewport was a **902.13 MB / 3 538.2 ms full scan**, 117× the
/// pyramid's z12. A layout defect is invisible in the code and total in the data,
/// which is why the assertions below are on the STATISTICS a reader prunes with and
/// never on whether a column is present.
#[cfg(test)]
mod base_table_pack_law {
use super::*;
use arrow::array::{BinaryBuilder, Int64Builder};
use parquet::file::statistics::Statistics;
/// A deterministic spatial scatter over a Sweden-sized box.
///
/// The point is that **id order is not spatial order**: OSM ids are temporal, so
/// consecutive rows in convert's output land anywhere. A fixture whose ids were
/// already spatially sorted would make every layout look good and every assertion
/// below vacuous.
pub(super) fn scatter(k: u64) -> (f64, f64) {
// A 64-bit mix (splitmix64 finaliser) — reproducible, and it destroys any
// correlation between id and position.
let mut z = k.wrapping_mul(0x9E37_79B9_7F4A_7C15).wrapping_add(0x1234_5678);
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^= z >> 31;
#[allow(clippy::float_arithmetic)]
let lon = 11.0 + ((z & 0xFFFF) as f64 / 65_535.0) * 13.0; // 11°..24°E
#[allow(clippy::float_arithmetic)]
let lat = 55.0 + (((z >> 16) & 0xFFFF) as f64 / 65_535.0) * 14.0; // 55°..69°N
(lon, lat)
}
/// The whole fixture's extent, in deg² — the area an UNSORTED row group covers.
const FIXTURE_AREA: f64 = 13.0 * 14.0;
/// Write a convert-order `nodes.parquet` + `ways.parquet` into `dir`.
///
/// Written through `ArrowWriter` directly, with NO covering column and one big
/// row group — i.e. byte-for-byte the shape `crate::writer`'s streaming writers
/// produce. That is the "before" this module is about.
fn convert_order_extract(dir: &Path, n: usize) -> Vec<Vec<(f64, f64)>> {
fs::create_dir_all(dir).unwrap();
let mut way_pts: Vec<Vec<(f64, f64)>> = Vec::with_capacity(n);
// nodes.parquet — points
{
let schema = Arc::new(crate::shared::schema::nodes_schema());
let mut id = Int64Builder::new();
let mut g = BinaryBuilder::new();
for k in 0..n {
let (lon, lat) = scatter(k as u64);
id.append_value(k as i64);
g.append_value(crate::geometry::encode_point_inline(lon, lat));
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
arrow::array::new_null_array(&DataType::Utf8, n),
arrow::array::new_null_array(&DataType::Int32, n),
];
// DELTA 1 + row 18: the whole class block is non-null on every base table,
// and these fixture rows carry no tags at all — so they take the ladder's
// own "no rule matched" values, built by the SAME `ClassColumns` the writers
// use rather than by a fixture's second opinion about what those are.
let mut cls = crate::shared::schema::ClassColumns::new();
for _ in 0..n {
cls.push("{}");
}
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let f = fs::File::create(dir.join("nodes.parquet")).unwrap();
let mut w = ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
// ways.parquet — short LineStrings, so the covering boxes have real AREA and
// a point table's degenerate `xmin == xmax` cannot be what makes this pass.
{
let schema = Arc::new(crate::shared::schema::ways_schema(false));
let mut id = Int64Builder::new();
let mut g = BinaryBuilder::new();
for k in 0..n {
let (lon, lat) = scatter(k as u64);
#[allow(clippy::float_arithmetic)]
let pts = vec![(lon, lat), (lon + 0.004, lat + 0.003), (lon + 0.007, lat)];
g.append_value(crate::shared::geometry::encode_wkb_line(&pts));
id.append_value(k as i64);
way_pts.push(pts);
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
arrow::array::new_null_array(&DataType::Utf8, n),
arrow::array::new_null_array(&DataType::Int32, n),
];
// DELTA 1 + row 18: the whole class block is non-null on every base table,
// and these fixture rows carry no tags at all — so they take the ladder's
// own "no rule matched" values, built by the SAME `ClassColumns` the writers
// use rather than by a fixture's second opinion about what those are.
let mut cls = crate::shared::schema::ClassColumns::new();
for _ in 0..n {
cls.push("{}");
}
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let f = fs::File::create(dir.join("ways.parquet")).unwrap();
let mut w = ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
way_pts
}
/// `nodes.parquet` alone, in convert order — the cheap fixture, for the guards
/// that need enough ROWS to make a 100 000-row group prunable. Returns each row's
/// point, indexed by id.
fn convert_order_nodes(dir: &Path, n: usize) -> Vec<(f64, f64)> {
fs::create_dir_all(dir).unwrap();
let schema = Arc::new(crate::shared::schema::nodes_schema());
let mut id = Int64Builder::new();
let mut g = BinaryBuilder::new();
let mut pts = Vec::with_capacity(n);
for k in 0..n {
let (lon, lat) = scatter(k as u64);
id.append_value(k as i64);
g.append_value(crate::geometry::encode_point_inline(lon, lat));
pts.push((lon, lat));
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
arrow::array::new_null_array(&DataType::Utf8, n),
arrow::array::new_null_array(&DataType::Int32, n),
];
// DELTA 1 + row 18: the whole class block is non-null on every base table. These
// rows carry no tags at all, so they take the values the ladder names for "no
// rule matched" — through the writers' own builder, not a second opinion.
let mut cls = crate::shared::schema::ClassColumns::new();
for _ in 0..n {
cls.push("{}");
}
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let f = fs::File::create(dir.join("nodes.parquet")).unwrap();
let mut w = ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
pts
}
/// Every row group's `(xmin, ymin, xmax, ymax)` as a reader reads it: from the
/// parquet column statistics of the four `bbox` leaves. Returns `None` for a file
/// with no covering column at all — which is the "before" state, and is why the
/// callers below treat that as a distinct outcome rather than a zero.
fn row_group_boxes(path: &Path) -> Option<Vec<(f64, f64, f64, f64)>> {
let f = fs::File::open(path).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let md = b.metadata().clone();
let sch = md.file_metadata().schema_descr();
let leaf = |name: &str| {
(0..sch.num_columns()).find(|i| sch.column(*i).path().string() == name)
};
let (xi, yi, xa, ya) = (
leaf("bbox.xmin")?,
leaf("bbox.ymin")?,
leaf("bbox.xmax")?,
leaf("bbox.ymax")?,
);
let get = |g: usize, c: usize| -> (f64, f64) {
match md.row_group(g).column(c).statistics() {
Some(Statistics::Float(s)) => (
f64::from(*s.min_opt().unwrap()),
f64::from(*s.max_opt().unwrap()),
),
other => panic!("{path:?} row group {g} leaf {c}: no FLOAT statistics ({other:?})"),
}
};
Some(
(0..md.num_row_groups())
.map(|g| {
let (x0, _) = get(g, xi);
let (y0, _) = get(g, yi);
let (_, x1) = get(g, xa);
let (_, y1) = get(g, ya);
(x0, y0, x1, y1)
})
.collect(),
)
}
#[allow(clippy::float_arithmetic)]
fn mean_area(boxes: &[(f64, f64, f64, f64)]) -> f64 {
if boxes.is_empty() {
return 0.0;
}
boxes.iter().map(|b| (b.2 - b.0) * (b.3 - b.1)).sum::<f64>() / boxes.len() as f64
}
/// The same quantity, computed from the RAW geometry in a given row order —
/// the control. `order` is a permutation of row indices.
#[allow(clippy::float_arithmetic)]
fn mean_area_of_order(pts: &[Vec<(f64, f64)>], order: &[usize], chunk: usize) -> f64 {
let mut areas = Vec::new();
for c in order.chunks(chunk) {
let (mut x0, mut y0, mut x1, mut y1) = (f64::MAX, f64::MAX, f64::MIN, f64::MIN);
for &i in c {
for &(x, y) in &pts[i] {
x0 = x0.min(x);
y0 = y0.min(y);
x1 = x1.max(x);
y1 = y1.max(y);
}
}
areas.push((x1 - x0) * (y1 - y0));
}
areas.iter().sum::<f64>() / areas.len() as f64
}
/// ★ **RED-when-broken: the base tables must PRUNE, and convert order must not.**
///
/// Two numbers, from two different places, over the same rows:
///
/// * `packed` — the mean row-group bbox area read out of the PACKED file's own
/// parquet statistics. This is literally the number a reader prunes with.
/// * `input` — the same mean, recomputed in the test from the raw WKB in the
/// order convert wrote it, chunked at the same row-group size.
///
/// The gate is the one `network.rs` already uses: the packed layout must shrink
/// that mean by more than **4×**. Take the Hilbert sort out of
/// [`pack_batches`] and the two arms converge onto the same number and this fails,
/// which is what makes it capable of red. VERIFIED red that way, 2026-08-31: with
/// `gatling_sort_unstable_by` commented out the packed mean went from
/// 0.0011 deg² to 178.3 deg² on `ways.parquet` and the assertion failed.
///
/// # The trap this guard is written around
///
/// A guard that compares a value to a copy of itself can never go red, and this
/// one is one line away from being exactly that — both means are "mean row-group
/// bbox area". So the control is pinned three ways before the 4× is trusted:
///
/// 1. the input-order mean must be a large fraction of the WHOLE fixture extent
/// (an unsorted group really does span everything — if the control were
/// accidentally computed over the sorted order it would be tiny and this fails);
/// 2. the same helper, run over the PACKED row order, must reproduce the number
/// read from the parquet statistics — which proves the two arms measure the
/// same quantity by two independent routes, so the gap between them is order
/// and not method;
/// 3. the unpacked file must have no covering column at all, so "before" is a
/// distinguishable state and not a small number.
#[test]
fn the_base_tables_prune_and_convert_order_would_not() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("extract");
let n = 40_000usize;
let rg = 1_024usize; // 39 groups — enough to average over in a unit test
let way_pts = convert_order_extract(&dir, n);
// (3) the "before" state is distinguishable: no covering column exists yet.
for t in ["nodes.parquet", "ways.parquet"] {
assert!(
!is_packed(&dir.join(t)).unwrap(),
"{t}: convert order must have NO covering column, or the pack below \
is a no-op and everything after it passes for the wrong reason"
);
assert!(
row_group_boxes(&dir.join(t)).is_none(),
"{t}: an unpacked table must expose no bbox statistics at all"
);
}
let opts = PackOptions {
row_group_rows: rg,
compression: String::from("none"),
..PackOptions::default()
};
let report = pack_base_tables(&dir, &opts).unwrap();
assert_eq!(report.len(), 2, "both base tables must have been packed");
for p in &report {
assert!(p.stats.is_some(), "{} was skipped, not packed", p.name);
}
// ── ways.parquet: real areas, not a point table's degenerate box ─────────
let boxes = row_group_boxes(&dir.join("ways.parquet"))
.expect("packed ways must expose bbox statistics");
assert!(boxes.len() >= 8, "need several row groups to prune between");
let packed = mean_area(&boxes);
let input_order: Vec<usize> = (0..n).collect();
let input = mean_area_of_order(&way_pts, &input_order, rg);
// (1) the control genuinely spans the fixture.
assert!(
input > FIXTURE_AREA * 0.5,
"the control is not a control: convert order gave a mean row-group area of \
{input:.2} deg² against a fixture extent of {FIXTURE_AREA:.2} deg². An \
unsorted group must span nearly everything; a small number here means the \
control was computed over the wrong order and the 4x below proves nothing"
);
// (2) the two arms measure the same quantity by independent routes.
let packed_order = read_ids(&dir.join("ways.parquet"));
let packed_recomputed = mean_area_of_order(&way_pts, &packed_order, rg);
let ratio = packed_recomputed / packed.max(f64::MIN_POSITIVE);
assert!(
(0.5..2.0).contains(&ratio),
"the parquet statistics ({packed:.6} deg²) and the raw-WKB recomputation \
over the SAME packed order ({packed_recomputed:.6} deg²) disagree by \
{ratio:.2}x — the two arms are not measuring the same thing, so their \
difference cannot be attributed to row order"
);
// ── THE GATE, the same one network.rs holds roads.parquet to ─────────────
assert!(
packed * 4.0 < input,
"the Hilbert-packed base table must shrink the mean row-group bbox area by \
more than 4x against convert order, or the covering column prunes nothing \
a reader can use: packed {packed:.4} deg² vs convert-order {input:.2} deg² \
over {} groups",
boxes.len()
);
// ── nodes.parquet gets the same treatment ────────────────────────────────
let nboxes = row_group_boxes(&dir.join("nodes.parquet"))
.expect("packed nodes must expose bbox statistics");
let npacked = mean_area(&nboxes);
let npts: Vec<Vec<(f64, f64)>> = (0..n).map(|k| vec![scatter(k as u64)]).collect();
let ninput = mean_area_of_order(&npts, &input_order, rg);
assert!(
ninput > FIXTURE_AREA * 0.5,
"nodes control is not a control: {ninput:.2} deg² of {FIXTURE_AREA:.2}"
);
assert!(
npacked * 4.0 < ninput,
"nodes.parquet: packed {npacked:.4} deg² vs convert-order {ninput:.2} deg²"
);
}
/// Read the `id` column back, which for these fixtures IS the input row index —
/// i.e. the permutation the packer chose.
fn read_ids(path: &Path) -> Vec<usize> {
let rdr = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(path).unwrap())
.unwrap()
.build()
.unwrap();
let mut out = Vec::new();
for batch in rdr {
let batch = batch.unwrap();
let a = batch
.column(0)
.as_any()
.downcast_ref::<arrow::array::Int64Array>()
.unwrap();
out.extend((0..a.len()).map(|i| a.value(i) as usize));
}
out
}
/// ★ **RED-when-broken: the row-group size is 8 192, and the alternative is
/// measured rather than asserted away.**
///
/// Two arms on the SAME rows and the SAME viewport:
///
/// * 8 192 — [`PACK_ROW_GROUP_ROWS`], what ships;
/// * 100 000 — the size every GeoParquet tuning guide recommends.
///
/// The assertion is not "8 192 < 100 000". It is that the 100 000-row layout makes
/// a reader **scan materially more rows for the same answer**, and the panic
/// message carries the measured ratio, so raising the default does not merely fail
/// a comparison — it prints its own price. On the live `roads.parquet` that price
/// is 2.03× (500 000 rows scanned against 245 760); this fixture is smaller and
/// spatially uniform, so it reproduces the direction, not the magnitude.
///
/// The pin on the constant is here too, because a lane that "tunes" it would
/// otherwise change every packed table in the tree — the networks included, since
/// `NetworkOptions::default` reads the same constant.
#[test]
fn the_row_group_size_is_the_measured_one() {
assert_eq!(
PACK_ROW_GROUP_ROWS, 8_192,
"the row-group size is MEASURED (see PACK_ROW_GROUP_ROWS): 8 192 is the \
largest size that still reaches the 1.78x scan floor on roads.parquet, and \
100 000 reads 2.03x more rows for the same answer. Changing it re-cuts \
every packed table in the tree; re-measure before you do"
);
assert_eq!(
PackOptions::default().row_group_rows,
PACK_ROW_GROUP_ROWS,
"PackOptions must take the size from the one constant, not a literal"
);
assert_eq!(
crate::network::NetworkOptions::default().row_group_rows,
PACK_ROW_GROUP_ROWS,
"the networks must take the size from the same constant, or the base tables \
and roads.parquet can drift apart on the one axis that decides pruning"
);
// ── the measurement, by the method the live one used ─────────────────────
// Pack ONCE — the Hilbert ORDER is identical for both candidate sizes, only
// the chunking differs — then simulate each chunking over that real order,
// exactly as the roads.parquet sweep in `PACK_ROW_GROUP_ROWS` did.
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("extract");
let n = 1_200_000usize; // 12 groups at 100 000 — enough that BOTH arms prune
let pts = convert_order_nodes(&dir, n);
let opts = PackOptions {
row_group_rows: PACK_ROW_GROUP_ROWS,
compression: String::from("none"),
..PackOptions::default()
};
pack_base_tables(&dir, &opts).unwrap();
let table = dir.join("nodes.parquet");
// The written file really is in 8 192-row groups — asserted on disk, not on
// the option that asked for it.
{
let f = fs::File::open(&table).unwrap();
let md = ParquetRecordBatchReaderBuilder::try_new(f)
.unwrap()
.metadata()
.clone();
let full: Vec<i64> = (0..md.num_row_groups() - 1)
.map(|g| md.row_group(g).num_rows())
.collect();
assert!(
full.iter().all(|&r| r == PACK_ROW_GROUP_ROWS as i64),
"every row group but the last must hold exactly {PACK_ROW_GROUP_ROWS} \
rows; got {:?}…",
&full[..full.len().min(4)]
);
}
// A viewport ~3.4 % of the fixture extent — small enough that the layout
// decides the answer, big enough that both arms hit more than one group.
let q = (14.0_f64, 60.0_f64, 16.5_f64, 62.5_f64);
let order = read_ids(&table);
assert_eq!(order.len(), n, "the packed table must carry every row");
// Ground truth: the rows that genuinely intersect. The ratios below are
// against THIS, not against each other, so neither arm can define the target.
let ideal = order
.iter()
.filter(|&&i| {
let (x, y) = pts[i];
x >= q.0 && x <= q.2 && y >= q.1 && y <= q.3
})
.count();
assert!(ideal > 10_000, "ground truth is only {ideal} rows — too thin to judge");
let sweep = |chunk: usize| -> (usize, usize, usize) {
let mut hit = 0usize;
let mut rows = 0usize;
let mut total = 0usize;
for c in order.chunks(chunk) {
total += 1;
let (mut x0, mut y0, mut x1, mut y1) = (f64::MAX, f64::MAX, f64::MIN, f64::MIN);
for &i in c {
let (x, y) = pts[i];
x0 = x0.min(x);
y0 = y0.min(y);
x1 = x1.max(x);
y1 = y1.max(y);
}
if x0 <= q.2 && x1 >= q.0 && y0 <= q.3 && y1 >= q.1 {
hit += 1;
rows += c.len();
}
}
(hit, total, rows)
};
let (hit_8k, tot_8k, rows_8k) = sweep(PACK_ROW_GROUP_ROWS);
let (hit_100k, tot_100k, rows_100k) = sweep(100_000);
// Anti-trap: both arms must actually prune something, or "fewer rows" is being
// read off two full scans and means nothing.
assert!(
hit_8k < tot_8k && hit_100k < tot_100k,
"neither layout pruned ({hit_8k}/{tot_8k} and {hit_100k}/{tot_100k}) — the \
comparison below would be between two full scans"
);
// …and the sweep must be reading a SORTED order. Over convert order every
// group spans everything and both arms would be full scans that happen to
// differ in group count, which is not what is being measured.
let (hit_raw, tot_raw) = {
let mut hit = 0usize;
let mut total = 0usize;
for c in (0..n).collect::<Vec<usize>>().chunks(PACK_ROW_GROUP_ROWS) {
total += 1;
let (mut x0, mut y0, mut x1, mut y1) = (f64::MAX, f64::MAX, f64::MIN, f64::MIN);
for &i in c {
let (x, y) = pts[i];
x0 = x0.min(x);
y0 = y0.min(y);
x1 = x1.max(x);
y1 = y1.max(y);
}
if x0 <= q.2 && x1 >= q.0 && y0 <= q.3 && y1 >= q.1 {
hit += 1;
}
}
(hit, total)
};
assert_eq!(
hit_raw, tot_raw,
"convert order must hit EVERY row group ({hit_raw} of {tot_raw}) — if it \
does not, the fixture's ids are already spatially sorted and the packed \
arm is not being compared against anything"
);
#[allow(clippy::float_arithmetic)]
let (r8, r100) = (
rows_8k as f64 / ideal as f64,
rows_100k as f64 / ideal as f64,
);
#[allow(clippy::float_arithmetic)]
let regression = rows_100k as f64 / rows_8k.max(1) as f64;
assert!(
rows_100k > rows_8k,
"the 100 000-row layout scanned {rows_100k} rows and 8 192 scanned {rows_8k} \
— if the big groups are not worse on this fixture the fixture is not \
discriminating and this guard is asleep"
);
assert!(
regression >= 1.2,
"100 000 rows/group cost only {regression:.2}x more scanned rows here \
({rows_100k} scanned / {ideal} genuinely hit = {r100:.2}x ideal, over \
{hit_100k} of {tot_100k} groups; against 8 192's {rows_8k} / {r8:.2}x over \
{hit_8k} of {tot_8k}). The live measurement on roads.parquet is 2.03x, so a \
ratio this flat means the fixture stopped discriminating between the sizes"
);
}
/// ★ **RED-when-broken: packing twice must not fork the layout.**
///
/// [`pack_batches`] APPENDS the covering column. Handed a table that already has
/// one it would produce a file with **two** `bbox` columns — valid parquet, which
/// every reader resolves positionally, i.e. a layout fork that no row count and no
/// digest would show. This is not hypothetical: convert packs its tables, then
/// `osm-katana relations` reads `ways.parquet` and packs its own output, and any
/// re-run or resume of either stage meets an already-packed file.
#[test]
fn packing_twice_leaves_exactly_one_covering_column() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path().join("extract");
convert_order_extract(&dir, 4_000);
let opts = PackOptions {
row_group_rows: 512,
compression: String::from("none"),
..PackOptions::default()
};
let first = pack_base_tables(&dir, &opts).unwrap();
assert!(
first.iter().all(|p| p.stats.is_some()),
"the first pack must actually pack, or the second is not a second"
);
let after_first = fs::read(dir.join("ways.parquet")).unwrap();
let second = pack_base_tables(&dir, &opts).unwrap();
assert!(
second.iter().all(|p| p.stats.is_none()),
"an already-packed table must be SKIPPED, not packed again: {:?}",
second.iter().map(|p| (&p.name, p.stats)).collect::<Vec<_>>()
);
assert_eq!(
fs::read(dir.join("ways.parquet")).unwrap(),
after_first,
"a skipped table must be left byte-identical — a rewrite that happens to \
be equivalent is still a rewrite of a file a server may be serving"
);
// And the thing the skip exists to prevent, stated directly.
let f = fs::File::open(dir.join("ways.parquet")).unwrap();
let sch = ParquetRecordBatchReaderBuilder::try_new(f).unwrap().schema().clone();
let bboxes = sch.fields().iter().filter(|f| f.name() == BBOX_COLUMN).count();
assert_eq!(
bboxes, 1,
"exactly one `{BBOX_COLUMN}` column, not {bboxes} — a second one is a \
silent layout fork"
);
}
/// ★ **RED-when-broken: `relations.resolved.parquet` is a base table.**
///
/// It was not in the list `spatial_pack_dir` walked, which is how the table korp
/// actually reads relations from ended up as the one unpacked file in a packed
/// directory. Named here so the list cannot quietly lose it again.
#[test]
fn the_resolved_relation_table_is_in_the_base_table_list() {
assert!(
BASE_TABLES.contains(&"relations.resolved.parquet"),
"relations.resolved.parquet is the relation table with assembled geometry — \
the raw relations.parquet carries none — so leaving it out of BASE_TABLES \
packs everything except the file a reader uses"
);
for t in ["nodes.parquet", "ways.parquet"] {
assert!(BASE_TABLES.contains(&t), "{t} missing from BASE_TABLES");
}
}
}
/// **DELTA 2 — the generalisation columns, through the WRITER.**
///
/// [`crate::shared::generalise`]'s own tests pin the simplifier at the function level.
/// These pin what comes out of `pack_batches`: the columns exist, they are DECLARED
/// geometry, they are measurably smaller in the FILE, a lake's island survives the whole
/// way into the written column, and a point table gets none of it.
///
/// Every assertion here has its control in the same test, on the same bytes.
#[cfg(test)]
mod min_zoom_pack_law {
use super::*;
/// The packer's spelling of the column and the schema's are one string. A literal
/// here that drifted from `min_zoom_field()` would make `ensure_min_zoom` append a
/// second column beside a present one, and `is_packed` demand a column no writer
/// declares.
#[test]
fn the_column_name_is_the_schemas() {
assert_eq!(
crate::shared::schema::min_zoom_field().name(),
MIN_ZOOM_COLUMN,
"spatial.rs spells the min_zoom column differently from shared/schema.rs"
);
}
/// ★ **The min-area rule reaches the packed table** (`waterblob-1`): a sub-pixel
/// `natural=water` islet is DEEPENED off rung 6 (tags alone put it there) and its
/// `geometry_z6` cell is NULL — no bare-`Point` confetti on a country view — while
/// a real lake keeps rung 6 and its z6 ring. Lines are untouched: a water LINE
/// (a stream) stays on its tag rung whatever its extent.
///
/// RED before `deepen_area_min_zoom` was wired into `pack_batches`: both areas
/// read back `min_zoom = 6` and the islet's z6 cell held a WKB `Point` — driven,
/// not described (comment the `deepen_area_min_zoom` call out to see it again).
#[test]
fn a_sub_pixel_area_is_deepened_off_the_coarse_rungs() {
use arrow::array::{BinaryBuilder, Int32Builder, Int64Builder, StringBuilder};
let geo = crate::metadata::geo_metadata(crate::metadata::WAY_GEOMETRY_TYPES, None);
let schema = Arc::new(crate::writer::schema_with_geo(
crate::shared::schema::ways_schema(false),
geo,
));
let ring = |cx: f64, cy: f64, r: f64| -> Vec<(f64, f64)> {
let mut v: Vec<(f64, f64)> = (0..16)
.map(|i| {
#[allow(clippy::cast_precision_loss)]
let a = std::f64::consts::TAU * i as f64 / 16.0;
(cx + r * a.cos(), cy + r * a.sin())
})
.collect();
v.push(v[0]);
v
};
let (mut id, mut g, mut tags, mut ver, mut cls) = (
Int64Builder::new(),
BinaryBuilder::new(),
StringBuilder::new(),
Int32Builder::new(),
crate::shared::schema::ClassColumns::new(),
);
let water = r#"{"natural":"water"}"#;
// Row 0: the median Gävle-coast islet — extent ~0.002°, sub-pixel at z6.
id.append_value(0);
g.append_value(crate::shared::geometry::encode_wkb_multipolygon(&[(
ring(17.2, 60.6, 0.001),
vec![],
)]));
tags.append_value(water);
ver.append_value(1);
cls.push(water);
// Row 1: a real lake — extent 0.4°, a shape at every rung.
id.append_value(1);
g.append_value(crate::shared::geometry::encode_wkb_multipolygon(&[(
ring(14.5, 58.9, 0.2),
vec![],
)]));
tags.append_value(water);
ver.append_value(1);
cls.push(water);
// Row 2: a tiny water LINE (a stream) — extent under the area floor, and it
// must NOT move: the rule is for areas only.
id.append_value(2);
let stream = r#"{"waterway":"stream"}"#;
g.append_value(crate::shared::geometry::encode_wkb_line(&[
(12.0, 57.0),
(12.0005, 57.0005),
]));
tags.append_value(stream);
ver.append_value(1);
cls.push(stream);
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
Arc::new(tags.finish()),
Arc::new(ver.finish()),
];
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let line_rung = crate::shared::class::min_zoom_of_tags(stream);
let dir = std::env::temp_dir().join(format!("katana-minarea-{}", std::process::id()));
fs::create_dir_all(&dir).unwrap();
let dst = dir.join("ways.parquet");
pack_batches(vec![batch], &schema, &dst, &PackOptions::default(), &[]).unwrap();
let rd = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(&dst).unwrap())
.unwrap()
.build()
.unwrap();
let mut by_id: std::collections::HashMap<i64, (u8, bool)> = std::collections::HashMap::new();
for b in rd {
let b = b.unwrap();
let ids = b.column_by_name("id").unwrap().as_any().downcast_ref::<arrow::array::Int64Array>().unwrap();
let mzs = b
.column_by_name(MIN_ZOOM_COLUMN)
.unwrap()
.as_any()
.downcast_ref::<UInt8Array>()
.unwrap();
let z6 = b
.column_by_name("geometry_z6")
.expect("the packed table carries the z6 rung column")
.as_any()
.downcast_ref::<BinaryArray>()
.unwrap();
for r in 0..b.num_rows() {
by_id.insert(ids.value(r), (mzs.value(r), z6.is_null(r)));
}
}
let (islet_mz, islet_z6_null) = by_id[&0];
let (lake_mz, lake_z6_null) = by_id[&1];
let (stream_mz, _) = by_id[&2];
assert_eq!(islet_mz, 9, "the islet is deepened to the rung where it is a shape");
assert!(islet_z6_null, "a deepened area has NO z6 cell — no Point confetti");
assert_eq!(lake_mz, 6, "a real lake keeps the tag ladder's rung");
assert!(!lake_z6_null, "…and its z6 ring is written");
assert_eq!(stream_mz, line_rung, "a LINE is never moved by the area rule");
}
/// `is_band_first` on a footer with no `min_zoom` statistics is `false`, and on
/// the statistics of a monotone sequence `true` — driven through a real file, not a
/// hand-built metadata object, because the point is what parquet writes.
#[test]
fn band_first_reads_the_footer() {
use arrow::array::UInt8Builder;
let dir = std::env::temp_dir().join(format!("katana-bandfirst-{}", std::process::id()));
fs::create_dir_all(&dir).unwrap();
let write = |name: &str, vals: &[u8]| -> std::path::PathBuf {
let schema = Arc::new(Schema::new(vec![crate::shared::schema::min_zoom_field()]));
let mut b = UInt8Builder::new();
b.append_slice(vals);
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(2))
.build();
let p = dir.join(name);
let mut w = ArrowWriter::try_new(fs::File::create(&p).unwrap(), schema.clone(), Some(props)).unwrap();
w.write(&RecordBatch::try_new(schema, vec![Arc::new(b.finish())]).unwrap()).unwrap();
w.close().unwrap();
p
};
let md = |p: &Path| {
ParquetRecordBatchReaderBuilder::try_new(fs::File::open(p).unwrap())
.unwrap()
.metadata()
.clone()
};
let sorted = write("sorted.parquet", &[6, 6, 9, 12, 12, 15]);
let mixed = write("mixed.parquet", &[6, 15, 9, 6, 12, 6]);
assert!(is_band_first(&md(&sorted)), "a band-first file read as not band-first");
assert!(!is_band_first(&md(&mixed)), "a straddling file read as band-first");
}
}
#[cfg(test)]
mod generalisation_law {
use super::*;
use crate::shared::schema::{GENERALISED_RUNGS, generalised_geometry_name};
use arrow::array::{BinaryBuilder, Int32Builder, Int64Builder, StringBuilder};
use parquet::basic::LogicalType;
use parquet::file::reader::{FileReader, SerializedFileReader};
/// A closed ring of `n` vertices around `(cx, cy)` at radius `r` degrees.
fn ring(cx: f64, cy: f64, r: f64, n: usize) -> Vec<(f64, f64)> {
let mut v: Vec<(f64, f64)> = (0..n)
.map(|i| {
#[allow(clippy::float_arithmetic, clippy::cast_precision_loss)]
let a = std::f64::consts::TAU * i as f64 / n as f64;
#[allow(clippy::float_arithmetic)]
(cx + r * a.cos(), cy + r * a.sin())
})
.collect();
v.push(v[0]);
v
}
/// `n` ways over Sweden: a dense wiggling coastline run each, plus every 16th row a
/// LAKE WITH AN ISLAND. Wiggle amplitude ~0.002° (~120 m), so z12 resolves it and z6
/// does not — a fixture that were uniformly smooth would make the byte assertions
/// pass on geometry rather than on the simplifier.
fn way_table(n: usize) -> (Arc<Schema>, Vec<RecordBatch>, Vec<usize>) {
let geo = crate::metadata::geo_metadata(crate::metadata::WAY_GEOMETRY_TYPES, None);
let schema = Arc::new(crate::writer::schema_with_geo(
crate::shared::schema::ways_schema(false),
geo,
));
let (mut id, mut g, mut tags, mut ver) = (
Int64Builder::new(),
BinaryBuilder::new(),
StringBuilder::new(),
Int32Builder::new(),
);
// `ways_schema` carries the whole class block (NOT NULL) since the rung columns
// and row 18 landed, so the fixture must fill it — and it fills it through
// `ClassColumns`, the ONE writer (LAW 5), not through literals.
let mut cls = crate::shared::schema::ClassColumns::new();
let mut lakes = Vec::new();
for k in 0..n {
let (lon, lat) = super::base_table_pack_law::scatter(k as u64);
id.append_value(k as i64);
// The rows are tagged as what they are — a lake and a coastline — because
// since DELTA 3 a rung column has a cell only for rows the ladder admits at
// that rung. Untagged rows sit on the unclassified rung (12) and would leave
// `geometry_z6` / `geometry_z9` entirely NULL, which is correct for a table
// of nothing and useless for measuring a simplifier. Both tag sets land on
// rung 6, so every rung column here has every row.
let t = if k % 16 == 0 {
lakes.push(k);
// A 0.2° lake with a 0.05° island: both far above z6's ~2.7e-3° epsilon.
g.append_value(crate::shared::geometry::encode_wkb_multipolygon(&[(
ring(lon, lat, 0.2, 64),
vec![ring(lon, lat, 0.05, 32)],
)]));
r#"{"natural":"water"}"#
} else {
let pts: Vec<(f64, f64)> = (0..256)
.map(|i| {
#[allow(clippy::float_arithmetic, clippy::cast_precision_loss)]
let t = i as f64 / 255.0;
#[allow(clippy::float_arithmetic)]
(lon + t * 0.3, lat + t * 0.2 + 0.002 * (t * 220.0).sin())
})
.collect();
g.append_value(crate::shared::geometry::encode_wkb_line(&pts));
r#"{"natural":"coastline"}"#
};
tags.append_value(t);
ver.append_value(1);
cls.push(t);
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
Arc::new(tags.finish()),
Arc::new(ver.finish()),
];
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
assert_eq!(
batch.num_columns(),
schema.fields().len(),
"the fixture does not build every column the way schema declares"
);
(schema, vec![batch], lakes)
}
/// Compressed bytes per leaf column, summed over every row group.
fn column_bytes(path: &Path) -> std::collections::BTreeMap<String, u64> {
let r = SerializedFileReader::new(fs::File::open(path).unwrap()).unwrap();
let md = r.metadata();
let mut out = std::collections::BTreeMap::new();
for gi in 0..md.num_row_groups() {
let rg = md.row_group(gi);
for c in 0..rg.num_columns() {
let col = rg.column(c);
#[allow(clippy::cast_sign_loss)]
let bytes = col.compressed_size().max(0) as u64;
*out.entry(col.column_path().string()).or_insert(0u64) += bytes;
}
}
out
}
/// ★ **RED-when-broken: the rung columns are written, DECLARED geometry, and
/// measurably smaller — with the un-generalised pack as the control.**
///
/// Both arms pack the SAME batches through the SAME writer; only `generalise`
/// differs. So a green cannot be the fixture being agreeable:
///
/// * the control arm must carry NO rung column — otherwise the flag does nothing and
/// the size comparison below is a file against itself;
/// * every rung column must be strictly smaller than `geometry`, and the ladder must
/// be monotone — a DP that drops everything and one that drops nothing would both
/// otherwise sail through;
/// * `geometry` itself must be byte-identical across the two arms, which is what
/// proves the generalisation is a NEW column and not a lossy rewrite of the old
/// one. That is the assertion a "simplify in place" regression trips first.
#[test]
fn the_rung_columns_are_written_smaller_and_the_plain_pack_is_the_control() {
let dir = tempfile::tempdir().unwrap();
let (schema, batches, _) = way_table(4_096);
let plain = dir.path().join("plain.parquet");
let genp = dir.path().join("genp.parquet");
let base = PackOptions {
compression: String::from("zstd"),
row_group_rows: 1_024,
covering: true,
generalise: false,
node_refs_sidecar: false,
};
pack_batches(batches.clone(), &schema, &plain, &base, &[]).unwrap();
pack_batches(
batches,
&schema,
&genp,
&PackOptions {
generalise: true,
..base
},
&[],
)
.unwrap();
let (pb, gb) = (column_bytes(&plain), column_bytes(&genp));
// The control: without the flag there is nothing to measure.
for &z in &GENERALISED_RUNGS {
let n = generalised_geometry_name(z);
assert!(
!pb.contains_key(&n),
"`generalise: false` still wrote {n} — the flag does nothing, so the \
comparison below is a file against itself"
);
assert!(gb.contains_key(&n), "`generalise: true` did not write {n}");
}
let full = gb["geometry"];
assert_eq!(
full, pb["geometry"],
"the `geometry` column differs between the two arms ({} vs {}) — the \
generalisation rewrote the full-precision column instead of adding new \
ones, and every reader of `geometry` just silently lost detail",
full, pb["geometry"]
);
let mut prev = full;
for &z in GENERALISED_RUNGS.iter().rev() {
let b = gb[&generalised_geometry_name(z)];
assert!(
b < prev,
"geometry_z{z} is {b} bytes against {prev} for the next finer column — a \
coarser rung that is not smaller has not simplified anything"
);
assert!(b > 0, "geometry_z{z} is empty — an absent column, not a coarse one");
prev = b;
}
assert!(
prev * 5 < full,
"the coarsest rung is {prev} bytes against {full} for full precision — under \
5x, so the byte gate this delta exists for is not being met on a fixture \
built to be simplifiable"
);
}
/// ★ **RED-when-broken: a lake keeps its island all the way into the written FILE.**
///
/// The shape of `tiles.rs`'s `a_lake_keeps_its_island_all_the_way_into_the_written_
/// tile_file`, pointed at the column. Not a duplicate of the module-level test in
/// [`crate::shared::generalise`]: that one stops at the encoder, this one goes
/// through `pack_batches`, Arrow, zstd and the parquet reader, which is where the
/// predecessor of this path lost the hole — the file was written, the row count was
/// right, the bytes were valid WKB, and the island was gone.
///
/// **Red arm:** flatten the rings in `generalise_wkb_into` (DP over
/// `shape.vertices()`, re-encode as a `LineString`) and every read-back row comes
/// back WKB type 2 — the `Polys` match below fails on the first lake.
#[test]
fn a_lake_keeps_its_island_all_the_way_into_the_written_column() {
use crate::shared::geometry::{WkbShape, decode_wkb};
use arrow::array::BinaryArray;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("ways.parquet");
let (schema, batches, lakes) = way_table(1_024);
assert!(!lakes.is_empty(), "the fixture built no lakes at all");
pack_batches(
batches,
&schema,
&path,
&PackOptions {
row_group_rows: 256,
..PackOptions::default()
},
&[],
)
.unwrap();
for &z in &GENERALISED_RUNGS {
let name = generalised_geometry_name(z);
let f = fs::File::open(&path).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let ci = b.schema().index_of(&name).unwrap();
let mut holed = 0usize;
let mut areas = 0usize;
for batch in b.build().unwrap() {
let batch = batch.unwrap();
let g = batch
.column(ci)
.as_any()
.downcast_ref::<BinaryArray>()
.unwrap();
for i in 0..batch.num_rows() {
if g.is_null(i) {
continue;
}
if let Some(WkbShape::Polys(ps)) = decode_wkb(g.value(i)) {
areas += 1;
if ps.iter().any(|(_, h)| !h.is_empty()) {
holed += 1;
}
}
}
}
assert_eq!(
areas,
lakes.len(),
"{name}: {areas} of {} lakes came back as areas. A 0.2 deg lake is {:.0}x \
this rung's epsilon; anything less than all of them is the flat \
ring-concatenating simplifier, not the tolerance.",
lakes.len(),
0.2 / crate::shared::generalise::generalisation_eps(z)
);
assert_eq!(
holed,
lakes.len(),
"{name}: only {holed} of {} lakes still have an island. The hole is \
0.05 deg, {:.0}x this rung's epsilon.",
lakes.len(),
0.05 / crate::shared::generalise::generalisation_eps(z)
);
}
}
/// ★ **RED-when-broken: every rung column is DECLARED geometry, not opaque binary.**
///
/// The control is in the same file: `tags` is a real column of the same table that
/// must NOT come back as `LogicalType::Geometry`. Without it a reader that returned
/// `Geometry` for everything would pass.
///
/// ⚠ **Its INPUT schema is built in memory by `way_table`, so it could not see
/// `geo-strip-1`.** [`pack_schema`] carried the base `geometry` field over from the
/// input, and a fixture whose input field was already declared handed it a good one —
/// green here, bare on the real re-packed table. The arm that covers a STALE input is
/// `tests/geometry_type_on_disk.rs::a_repack_upgrades_a_stale_geometry_column`, which
/// writes a pre-`57aafb3` file and packs THAT.
#[test]
fn the_rung_columns_carry_the_geometry_logical_type() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("ways.parquet");
let (schema, batches, _) = way_table(256);
pack_batches(batches, &schema, &path, &PackOptions::default(), &[]).unwrap();
let r = SerializedFileReader::new(fs::File::open(&path).unwrap()).unwrap();
let rg = r.metadata().row_group(0);
let ty = |name: &str| -> Option<LogicalType> {
(0..rg.num_columns())
.find(|&c| rg.column(c).column_path().string() == name)
.and_then(|c| rg.column(c).column_descr().logical_type_ref().cloned())
};
assert!(
!matches!(ty("tags"), Some(LogicalType::Geometry(_))),
"the control column `tags` came back as Geometry — this reader says Geometry \
about everything and the assertions below prove nothing"
);
for name in std::iter::once(String::from("geometry")).chain(
GENERALISED_RUNGS
.iter()
.map(|&z| generalised_geometry_name(z)),
) {
assert!(
matches!(ty(&name), Some(LogicalType::Geometry(_))),
"{name} came back as {:?}, not LogicalType::Geometry — it did not go \
through `shared::schema::geometry_field` and is a twin",
ty(&name)
);
}
}
/// ★ **RED-when-broken: the derived columns must not move the content digest.**
///
/// `verify --digest` is THE equivalence gate for a pack: same rows, same values,
/// only the order and some derived columns changed. `geometry_z12` is a pure
/// function of `geometry` and the rung's epsilon, so hashing it would make a
/// generalised table incomparable with the plain pack it came from — i.e. it would
/// break the gate that proves packing preserved content, while catching nothing that
/// a change to `geometry` does not already catch.
///
/// The control is the third arm: a table with ONE geometry cell changed must give a
/// DIFFERENT digest, or the digest is inert and the equality above means nothing.
#[test]
fn the_derived_columns_do_not_move_the_digest() {
let dir = tempfile::tempdir().unwrap();
let (schema, batches, _) = way_table(512);
let base = PackOptions {
row_group_rows: 128,
..PackOptions::default()
};
let plain = dir.path().join("plain.parquet");
let genp = dir.path().join("genp.parquet");
pack_batches(
batches.clone(),
&schema,
&plain,
&PackOptions {
generalise: false,
..base.clone()
},
&[],
)
.unwrap();
pack_batches(batches.clone(), &schema, &genp, &base, &[]).unwrap();
let a = crate::digest::digest_file(&plain).unwrap();
let b = crate::digest::digest_file(&genp).unwrap();
assert_eq!(
(a.set, a.rows),
(b.set, b.rows),
"the generalisation columns moved the content digest — a packed table can no \
longer be proved equal to the convert output it came from"
);
// The control: change one geometry cell and the digest MUST move.
let (schema2, batches2, _) = way_table(511);
let other = dir.path().join("other.parquet");
pack_batches(batches2, &schema2, &other, &base, &[]).unwrap();
let c = crate::digest::digest_file(&other).unwrap();
assert_ne!(
(b.set, b.rows),
(c.set, c.rows),
"a table with different rows digests the same — the digest is inert and the \
equality asserted above is not evidence of anything"
);
}
/// ★ **RED-when-broken: a POINT table gets no generalisation columns.**
///
/// MEASURED on the Sweden cut: `nodes.parquet` carries 655.18 MB of compressed point
/// geometry, so three rung columns on it are **+1.97 GB** of byte-for-byte duplicate.
/// The way table in the same test is the control — without it this would pass for a
/// writer that never generalises anything.
#[test]
fn a_point_table_gets_no_rung_columns_and_the_way_table_does() {
use arrow::array::BinaryBuilder;
let dir = tempfile::tempdir().unwrap();
let geo = crate::metadata::geo_metadata(&["Point"], None);
let schema = Arc::new(crate::writer::schema_with_geo(
crate::shared::schema::nodes_schema(),
geo,
));
let (mut id, mut g, mut tags, mut ver) = (
Int64Builder::new(),
BinaryBuilder::new(),
StringBuilder::new(),
Int32Builder::new(),
);
// `nodes_schema` carries the whole class block (NOT NULL) too — same reason,
// same ONE writer.
let mut cls = crate::shared::schema::ClassColumns::new();
for k in 0..256u64 {
let (lon, lat) = super::base_table_pack_law::scatter(k);
#[allow(clippy::cast_possible_wrap)]
id.append_value(k as i64);
g.append_value(crate::shared::geometry::encode_wkb_point(lon, lat));
tags.append_value("{}");
ver.append_value(1);
cls.push("{}");
}
let mut cols: Vec<arrow::array::ArrayRef> = vec![
Arc::new(id.finish()),
Arc::new(g.finish()),
Arc::new(tags.finish()),
Arc::new(ver.finish()),
];
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols).unwrap();
let nodes = dir.path().join("nodes.parquet");
pack_batches(vec![batch], &schema, &nodes, &PackOptions::default(), &[]).unwrap();
let nb = column_bytes(&nodes);
for &z in &GENERALISED_RUNGS {
assert!(
!nb.contains_key(&generalised_geometry_name(z)),
"the POINT table carries geometry_z{z} — three byte-for-byte copies of a \
column no epsilon can change"
);
}
// The control, same run, same writer: the way table DOES get them.
let ways = dir.path().join("ways.parquet");
let (ws, wb, _) = way_table(256);
pack_batches(wb, &ws, &ways, &PackOptions::default(), &[]).unwrap();
let wbytes = column_bytes(&ways);
for &z in &GENERALISED_RUNGS {
assert!(
wbytes.contains_key(&generalised_geometry_name(z)),
"the WAY table has no geometry_z{z} either — the writer generalises \
nothing, so the point-table assertion above is vacuous"
);
}
}
/// ★ **RED-when-broken: a re-pack UPGRADES the layout instead of forking it.**
///
/// The trap this closes: `pack_batches` used to carry every input column across and
/// append its derived ones, so packing an already-packed table produced a file with
/// TWO columns called `bbox` — valid parquet that every reader resolves by position,
/// i.e. a silent layout fork. `is_packed` was the only guard, and it could not stay
/// the guard once "packed" grew to mean "and carries the rung columns".
///
/// So: pack, pack again, and assert the schema is IDENTICAL — no duplicate names —
/// and that a table packed WITHOUT the rungs reports `is_packed == false` so the
/// upgrade actually runs.
#[test]
fn a_repack_upgrades_the_layout_and_never_duplicates_a_column() {
let dir = tempfile::tempdir().unwrap();
let (schema, batches, _) = way_table(512);
let base = PackOptions {
row_group_rows: 128,
..PackOptions::default()
};
let old = dir.path().join("old.parquet");
pack_batches(
batches,
&schema,
&old,
&PackOptions {
generalise: false,
..base.clone()
},
&[],
)
.unwrap();
assert!(
!is_packed(&old).unwrap(),
"a table with `covering.bbox` but no rung columns reports as packed — every \
extract already on disk would silently skip the upgrade"
);
let up = pack_in_place(&old, &base).unwrap();
assert!(up.is_some(), "the upgrade did not run");
assert!(
is_packed(&old).unwrap(),
"the upgraded table still does not report as packed — this would re-pack on \
every run"
);
let names = |p: &Path| -> Vec<String> {
let f = fs::File::open(p).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
b.schema()
.fields()
.iter()
.map(|f| f.name().clone())
.collect()
};
let before = names(&old);
let mut dedup = before.clone();
dedup.sort();
dedup.dedup();
assert_eq!(
dedup.len(),
before.len(),
"the upgraded table has duplicate column names: {before:?} — this is the \
silent layout fork, and every reader resolving by position now reads the \
wrong column"
);
assert!(
pack_in_place(&old, &base).unwrap().is_none(),
"a second pack of an already-upgraded table ran again — packing is not \
idempotent and the fork can come back"
);
assert_eq!(before, names(&old), "the no-op pack still rewrote the schema");
}
}