osm-katana/tests/min_zoom_repack.rs
//! **DELTA 3 on APPLIED OUTPUT — a re-pack writes `min_zoom`, sorts band-first, and NULLs
//! the rung cells a camera may not draw.**
//!
//! Every assertion here reads a parquet file back off disk. Nothing compares the packer
//! with a copy of the packer.
//!
//! | test | the defect it is red for |
//! |---|---|
//! | [`a_repack_values_min_zoom_from_the_one_ladder_when_the_input_lacks_it`] | `min_zoom` written only at convert time; a re-pack cannot add it |
//! | [`the_packed_order_is_band_first_and_a_country_camera_opens_few_groups`] | Hilbert-global order — every row group holds every rung, nothing prunes |
//! | [`a_rung_column_is_null_exactly_where_the_feature_may_not_draw`] | rung columns inherit point degradation instead of withholding the row |
//! | [`a_hilbert_only_table_with_the_column_is_not_packed`] | `is_packed` blind to the ORDER, so a stale layout never upgrades |
//!
//! Run: `cargo test -p osm-katana --test min_zoom_repack -- --nocapture`
use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use arrow::array::{Array, BinaryArray, Int64Array, RecordBatch, StringArray, UInt8Array};
use arrow::datatypes::{DataType, Field, Schema};
use osm_katana::metadata::{self, GeoMeta};
use osm_katana::shared::class::{RUNGS, min_zoom_of_tags};
use osm_katana::shared::schema::{GENERALISED_RUNGS, generalised_geometry_name};
use osm_katana::spatial::{PackOptions, is_packed, spatial_pack};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::file::statistics::Statistics;
/// One tag string per rung of the ladder, so the fixture can be shown to span it.
/// The rung each one lands on is the LADDER's answer, read through the one writer —
/// the test only insists that the four answers are four DIFFERENT rungs.
const TAGS: [&str; 4] = [
r#"{"highway":"motorway"}"#,
r#"{"highway":"tertiary"}"#,
r#"{"highway":"residential"}"#,
r#"{"highway":"service"}"#,
];
/// The rung mix — a road network's shape: a thin skeleton of arterials, a long tail.
const MIX: [usize; 4] = [5, 10, 45, 40];
const N: usize = 65_536;
/// Fine groups so the band-first arm has enough of them to measure a prune on.
const RG: usize = 1_024;
fn scratch(stem: &str) -> PathBuf {
let d = PathBuf::from(std::env::var_os("HOME").unwrap_or_default())
.join("scratch/q-lane-test")
.join(format!("minzoom-repack-{}", std::process::id()));
std::fs::create_dir_all(&d).unwrap();
d.join(format!("{stem}.parquet"))
}
fn splitmix(mut x: u64) -> u64 {
x = x.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = x;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
/// A deterministic point over a Sweden-sized box, spatially UNCORRELATED with the row
/// index — so the rungs are interleaved in space exactly as a motorway and a footpath
/// are in a real square kilometre.
fn scatter(i: u64) -> (f64, f64) {
let h = splitmix(i);
let x = (h & 0xFFFF_FFFF) as f64 / 4_294_967_296.0;
let y = (h >> 32) as f64 / 4_294_967_296.0;
(11.0 + x * 13.0, 55.0 + y * 14.0)
}
fn wkb_line(x: f64, y: f64) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&2u32.to_le_bytes());
v.extend_from_slice(&2u32.to_le_bytes());
for k in 0..2 {
v.extend_from_slice(&(x + f64::from(k) * 0.001).to_le_bytes());
v.extend_from_slice(&(y + f64::from(k) * 0.001).to_le_bytes());
}
v
}
/// The tag string row `i` carries, drawn from [`MIX`] by a hash so the rungs are
/// interleaved in ROW order too (input order is convert order, which is not spatial).
fn tags_of(i: usize) -> &'static str {
let r = (splitmix(i as u64 ^ 0xABCD) % 100) as usize;
let mut acc = 0;
for (k, pct) in MIX.iter().enumerate() {
acc += pct;
if r < acc {
return TAGS[k];
}
}
TAGS[3]
}
/// **The PRE-DELTA-1 way table** — `id, geometry, tags, node_refs, version`, no
/// `min_zoom`. This is the schema `/home/korp/.korp/work/sweden/unpacked-backup/ways.parquet`
/// has, and what a re-pack must be able to upgrade. `with_min_zoom = Some(v)` appends a
/// `min_zoom` column holding the constant `v` on every row — a PRESENT column whose value
/// disagrees with the tags, so "kept when present" and "recomputed" are distinguishable.
fn write_source(path: &Path, with_min_zoom: Option<u8>) {
let mut fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new("version", DataType::Int32, true),
];
if with_min_zoom.is_some() {
fields.push(osm_katana::shared::schema::min_zoom_field());
}
let mut meta = HashMap::new();
meta.insert(
String::from(metadata::GEO_KEY),
GeoMeta::new(&["LineString"]).to_json(),
);
let schema = Arc::new(Schema::new_with_metadata(fields, meta));
let mut ids = Vec::with_capacity(N);
let mut geoms: Vec<Vec<u8>> = Vec::with_capacity(N);
let mut tags: Vec<&str> = Vec::with_capacity(N);
for i in 0..N {
let (x, y) = scatter(i as u64);
ids.push(i as i64);
geoms.push(wkb_line(x, y));
tags.push(tags_of(i));
}
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(Int64Array::from(ids)),
Arc::new(BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice))),
Arc::new(StringArray::from(tags)),
Arc::new(arrow::array::Int32Array::from(vec![1i32; N])),
];
if let Some(v) = with_min_zoom {
cols.push(Arc::new(UInt8Array::from(vec![v; N])));
}
let f = std::fs::File::create(path).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema.clone(), None).unwrap();
w.write(&RecordBatch::try_new(schema, cols).unwrap()).unwrap();
w.close().unwrap();
}
fn pack(src: &Path, dst: &Path) {
let opts = PackOptions { row_group_rows: RG, ..PackOptions::default() };
spatial_pack(src, dst, &opts).unwrap_or_else(|e| panic!("spatial_pack {src:?}: {e:#}"));
}
/// Every row of `path`: `(id, tags, min_zoom, per-rung null flag, geometry null)`.
struct Row {
tags: String,
min_zoom: u8,
rung_null: Vec<(u8, bool)>,
geometry_null: bool,
}
fn read_rows(path: &Path) -> Vec<Row> {
let f = std::fs::File::open(path).unwrap();
let rdr = ParquetRecordBatchReaderBuilder::try_new(f).unwrap().build().unwrap();
let mut out = Vec::new();
for b in rdr {
let b = b.unwrap();
let tags = b
.column_by_name("tags")
.expect("tags")
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let mz = b
.column_by_name("min_zoom")
.expect("the packed table carries no `min_zoom` column — a re-pack cannot add it")
.as_any()
.downcast_ref::<UInt8Array>()
.expect("min_zoom is UInt8");
let geom = b
.column_by_name("geometry")
.expect("geometry")
.as_any()
.downcast_ref::<BinaryArray>()
.unwrap();
let rungs: Vec<(u8, &BinaryArray)> = GENERALISED_RUNGS
.iter()
.map(|&z| {
let c = b
.column_by_name(&generalised_geometry_name(z))
.unwrap_or_else(|| panic!("no {}", generalised_geometry_name(z)))
.as_any()
.downcast_ref::<BinaryArray>()
.unwrap();
(z, c)
})
.collect();
for i in 0..b.num_rows() {
assert!(!mz.is_null(i), "min_zoom must be NOT NULL on every row");
out.push(Row {
tags: tags.value(i).to_string(),
min_zoom: mz.value(i),
rung_null: rungs.iter().map(|(z, c)| (*z, c.is_null(i))).collect(),
geometry_null: geom.is_null(i),
});
}
}
out
}
/// Per row group: `(min, max)` of the `min_zoom` statistics, and the covering box.
fn footer(path: &Path) -> Vec<((u8, u8), (f32, f32, f32, f32))> {
let r = SerializedFileReader::new(std::fs::File::open(path).unwrap()).unwrap();
let md = r.metadata();
let mut out = Vec::new();
for g in 0..md.num_row_groups() {
let rg = md.row_group(g);
let (mut lo, mut hi) = (u8::MAX, 0u8);
let (mut xmin, mut ymin, mut xmax, mut ymax) = (f32::NAN, f32::NAN, f32::NAN, f32::NAN);
for c in 0..rg.num_columns() {
let col = rg.column(c);
let Some(st) = col.statistics() else { continue };
match col.column_path().string().as_str() {
"min_zoom" => match st {
Statistics::Int32(s) => {
lo = *s.min_opt().expect("min") as u8;
hi = *s.max_opt().expect("max") as u8;
}
other => panic!("min_zoom statistics came back as {other:?}"),
},
"bbox.xmin" => xmin = f32_stat(st, true),
"bbox.ymin" => ymin = f32_stat(st, true),
"bbox.xmax" => xmax = f32_stat(st, false),
"bbox.ymax" => ymax = f32_stat(st, false),
_ => {}
}
}
assert!(lo <= hi, "row group {g} has no min_zoom statistics — nothing can prune");
out.push(((lo, hi), (xmin, ymin, xmax, ymax)));
}
out
}
fn f32_stat(st: &Statistics, min: bool) -> f32 {
match st {
Statistics::Float(s) => {
if min { *s.min_opt().unwrap() } else { *s.max_opt().unwrap() }
}
other => panic!("bbox leaf statistics came back as {other:?}"),
}
}
// ── 1. the column is ADDED by a re-pack, from the one ladder ────────────────────
/// ★ **RED-when-broken: a re-pack of a table with no `min_zoom` produces one.**
///
/// Written from `tags` through `class::min_zoom_of_tags` — the ONE tag → rung writer —
/// and compared row by row against that function's answer for the row's own tags, as
/// read back off disk. Non-vacuity: the values written must span every rung of
/// [`RUNGS`], or a packer that wrote a constant would agree with itself.
///
/// The second arm is the other half of the contract: a column that IS present is kept
/// verbatim, not recomputed. The input carries `15` on every row against tags that earn
/// 6/9/12/15, so a packer that recomputed would be caught on the rows that differ.
#[test]
fn a_repack_values_min_zoom_from_the_one_ladder_when_the_input_lacks_it() {
let src = scratch("src-no-minzoom");
let dst = scratch("dst-no-minzoom");
write_source(&src, None);
pack(&src, &dst);
let rows = read_rows(&dst);
assert_eq!(rows.len(), N, "the pack dropped rows");
let mut seen = BTreeSet::new();
let mut wrong = 0usize;
for r in &rows {
let want = min_zoom_of_tags(&r.tags);
if r.min_zoom != want {
wrong += 1;
}
seen.insert(r.min_zoom);
}
assert_eq!(
wrong, 0,
"{wrong} of {N} rows carry a min_zoom that is not what the ladder gives their tags"
);
assert_eq!(
seen.iter().copied().collect::<Vec<u8>>(),
RUNGS.to_vec(),
"the written rungs {seen:?} do not span the ladder {RUNGS:?} — a constant column \
would pass the row-by-row check above"
);
// ARM 2 — kept when present, even when wrong. The input says 15 everywhere.
let src2 = scratch("src-with-minzoom");
let dst2 = scratch("dst-with-minzoom");
write_source(&src2, Some(15));
pack(&src2, &dst2);
let rows2 = read_rows(&dst2);
let recomputed = rows2.iter().filter(|r| r.min_zoom != 15).count();
assert_eq!(
recomputed, 0,
"{recomputed} rows had their present min_zoom overwritten — a re-pack must keep \
the column it was given, or a convert-time value can never survive a pack"
);
println!("gate 1: {N} rows, rungs written {seen:?}; present column kept on {} rows", rows2.len());
}
// ── 2. the ORDER is (min_zoom, hilbert) ─────────────────────────────────────────
/// ★ **RED-when-broken: the row groups are band-first, and a z6 reader opens few.**
///
/// Read off the footer: consecutive row groups must be monotone in `min_zoom`
/// (`max(g) <= min(g + 1)`), which is what "band first" means at row-group level, and
/// the groups a `min_zoom <= 6` reader must open are ≤ 20 % of the file — korp's phase-1
/// gate. On the Hilbert-only order every group straddles the ladder and both fail.
///
/// The control inside the band: each group's covering box must still be COMPACT — the
/// mean group area in the largest band under 15 % of the extent. A sort on `min_zoom`
/// alone (no Hilbert inside) would leave every group spanning the whole extent, and a
/// z6 read would then prune by band and scan the whole band for one city.
#[test]
fn the_packed_order_is_band_first_and_a_country_camera_opens_few_groups() {
let src = scratch("src-order");
let dst = scratch("dst-order");
write_source(&src, None);
pack(&src, &dst);
let groups = footer(&dst);
let total = groups.len();
assert!(total >= 32, "only {total} row groups — too few to measure a prune");
let mut straddle = Vec::new();
for (g, w) in groups.windows(2).enumerate() {
let ((_, hi), _) = w[0];
let ((lo, _), _) = w[1];
if hi > lo {
straddle.push((g, hi, lo));
}
}
assert!(
straddle.is_empty(),
"{} of {total} row-group boundaries are NOT band-first (max of g > min of g+1), \
first: {:?}. Every group holds every rung — this is the Hilbert-only order and \
a min_zoom predicate prunes nothing on it.",
straddle.len(),
&straddle[..straddle.len().min(3)]
);
let z6_open = groups.iter().filter(|((lo, _), _)| *lo <= 6).count();
println!(
"gate 2: min_zoom <= 6 opens {z6_open}/{total} row groups ({:.1} %)",
100.0 * z6_open as f64 / total as f64
);
assert!(
z6_open * 5 <= total,
"a z6 reader opens {z6_open} of {total} groups — more than korp's 20 % gate"
);
// Hilbert INSIDE the band: the largest band's groups are compact.
let extent = 13.0 * 14.0;
let mut by_band: HashMap<u8, Vec<f64>> = HashMap::new();
for ((lo, hi), (x0, y0, x1, y1)) in &groups {
if lo != hi {
continue; // a boundary group; its box is honestly two bands' worth
}
let area = f64::from(x1 - x0) * f64::from(y1 - y0);
by_band.entry(*lo).or_default().push(area);
}
let (band, areas) = by_band
.iter()
.max_by_key(|(_, v)| v.len())
.expect("at least one pure band");
let mean = areas.iter().sum::<f64>() / areas.len() as f64;
println!(
"gate 2: band {band} has {} pure groups, mean covering area {:.3} % of the extent",
areas.len(),
100.0 * mean / extent
);
assert!(
mean < 0.15 * extent,
"band {band}'s groups cover {:.1} % of the extent on average — the rows inside a \
band are not Hilbert-ordered, so a viewport read scans the whole band",
100.0 * mean / extent
);
assert!(is_packed(&dst).unwrap(), "the packer's own output must read as packed");
}
// ── 3. NULL where the camera may not draw ───────────────────────────────────────
/// ★ **RED-when-broken: `geometry_z<z>` is NULL exactly when `min_zoom > z`.**
///
/// Both directions, every rung, every row, off disk. A row the camera may draw keeps its
/// cell (a sub-pixel feature still degrades to a point — that rule is untouched); a row
/// it may not draw has no cell, so a reader projecting the rung column never decodes a
/// feature the ladder withheld. `geometry` (rung 15) is never NULL: every row draws at 15.
#[test]
fn a_rung_column_is_null_exactly_where_the_feature_may_not_draw() {
let src = scratch("src-null");
let dst = scratch("dst-null");
write_source(&src, None);
pack(&src, &dst);
let rows = read_rows(&dst);
let mut nulls: HashMap<u8, usize> = HashMap::new();
let mut wrong: Vec<String> = Vec::new();
for r in &rows {
assert!(!r.geometry_null, "`geometry` went NULL — rung 15 admits every row");
for &(z, is_null) in &r.rung_null {
let must_be_null = r.min_zoom > z;
if is_null {
*nulls.entry(z).or_default() += 1;
}
if is_null != must_be_null && wrong.len() < 5 {
wrong.push(format!(
"min_zoom {} at geometry_z{z}: null={is_null}, wanted null={must_be_null} ({})",
r.min_zoom, r.tags
));
}
}
}
assert!(
wrong.is_empty(),
"rung cells disagree with min_zoom on {} rows (first few): {wrong:#?}",
wrong.len()
);
for &z in &GENERALISED_RUNGS {
let n = nulls.get(&z).copied().unwrap_or(0);
println!(
"gate 3: geometry_z{z} NULL on {n}/{N} rows ({:.2} %)",
100.0 * n as f64 / N as f64
);
}
let z6 = nulls.get(&6).copied().unwrap_or(0);
assert!(
z6 > 0 && z6 < N,
"geometry_z6 is NULL on {z6} of {N} rows — all or none, so the fixture spans one \
rung and the equivalence above was never exercised"
);
}
// ── 4. a Hilbert-only table with the column is NOT packed ───────────────────────
/// ★ **RED-when-broken: `is_packed` is blind to the ORDER.**
///
/// Take the packer's own output, re-sort it on the Hilbert key ALONE (the layout every
/// table packed before DELTA 3 has, `min_zoom` present or not), write it back with the
/// same schema and row groups, and ask. It must say `false`, or `pack_in_place` skips
/// every stale table on disk forever.
#[test]
fn a_hilbert_only_table_with_the_column_is_not_packed() {
let src = scratch("src-stale");
let dst = scratch("dst-stale");
let stale = scratch("stale-hilbert-only");
write_source(&src, None);
pack(&src, &dst);
// Re-order on the covering centre's Hilbert index only.
let f = std::fs::File::open(&dst).unwrap();
let b = ParquetRecordBatchReaderBuilder::try_new(f).unwrap();
let schema = b.schema().clone();
let batches: Vec<RecordBatch> = b.build().unwrap().map(Result::unwrap).collect();
let all = arrow::compute::concat_batches(&schema, &batches).unwrap();
let geom = all
.column_by_name("geometry")
.unwrap()
.as_any()
.downcast_ref::<BinaryArray>()
.unwrap();
let mut keys: Vec<(u64, u32)> = (0..all.num_rows())
.map(|i| {
let bb = osm_katana::spatial::wkb_bbox(geom.value(i)).unwrap();
let (cx, cy) = bb.centre();
(osm_katana::spatial::hilbert_index(cx, cy), i as u32)
})
.collect();
keys.sort_unstable();
let idx = arrow::array::UInt32Array::from(keys.iter().map(|k| k.1).collect::<Vec<_>>());
let cols: Vec<Arc<dyn Array>> = all
.columns()
.iter()
.map(|c| arrow::compute::take(c.as_ref(), &idx, None).unwrap())
.collect();
let reordered = RecordBatch::try_new(schema.clone(), cols).unwrap();
let props = parquet::file::properties::WriterProperties::builder()
.set_max_row_group_row_count(Some(RG))
.build();
let out = std::fs::File::create(&stale).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(out, schema, Some(props)).unwrap();
w.write(&reordered).unwrap();
w.close().unwrap();
// The control: the re-ordered file genuinely straddles, or the assertion below is
// about a file that happens to be band-first by luck.
let g = footer(&stale);
let straddling = g.iter().filter(|((lo, hi), _)| lo != hi).count();
assert!(
straddling * 2 > g.len(),
"only {straddling} of {} groups straddle rungs after a Hilbert-only re-sort — the \
control did not produce the stale layout",
g.len()
);
assert!(
!is_packed(&stale).unwrap(),
"a Hilbert-only table carrying `min_zoom` reads as PACKED — it will never be \
upgraded to the band-first order, and every z6 read on it scans every group"
);
assert!(is_packed(&dst).unwrap(), "…while the band-first original must read as packed");
}