osm-katana/tests/min_zoom_one_ladder.rs
//! **DELTA 1's two load-bearing claims, on APPLIED OUTPUT.**
//!
//! Not on the classifier's return value — on the `min_zoom` cells that are actually in a
//! written parquet file, read back through the parquet reader. A test that calls
//! `min_zoom_of_tags` and compares it with `min_zoom_of_tags` is a value compared with
//! itself; these open the file.
//!
//! | test | the defect it is red for |
//! |---|---|
//! | [`a_way_and_a_node_with_the_same_tags_land_on_the_same_rung`] | `korp-map-pt-1`: points and ways through different tag → class maps |
//! | [`band_first_order_prunes_and_hilbert_only_does_not`] | the claim DELTA 1 is NOT allowed to make on its own |
//!
//! Run: `cargo test -p osm-katana --test min_zoom_one_ladder -- --nocapture`
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use arrow::array::{
Array, BinaryArray, Int32Array, Int64Array, RecordBatch, StringArray, UInt8Array,
};
use arrow::datatypes::{DataType, Field, Schema};
use osm_katana::metadata::{self, GeoMeta};
use osm_katana::network::{NetworkOptions, extract_networks};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::file::statistics::Statistics;
// ── GATE 2 ──────────────────────────────────────────────────────────────────────
/// **The tag sets that appear on BOTH a way and a node in the fixture.**
///
/// `railway=*` is the one selector this crate runs over both source tables — `RAILS`
/// reads `ways.parquet`, `RAIL_NODES` reads `nodes.parquet`, and their rule lists are
/// identical (`Rule::ValueOf { key: "railway" }`). So the same three tag strings can be
/// carried into two different output files by two different code paths, which is exactly
/// the shape needed to catch a second ladder.
///
/// The three are chosen to sit on three DIFFERENT rungs (6 · 9 · 12). Three rows on one
/// rung would pass this test on a ladder that had collapsed to a constant.
const BOTH: [&str; 3] = [
r#"{"railway":"rail"}"#,
r#"{"railway":"tram"}"#,
r#"{"railway":"signal"}"#,
];
/// ★ **RED-when-broken: a way and a node with the same tags on two different rungs.**
///
/// This is the `korp-map-pt-1` defect (2026-08-31) posed as a guard. There, points and
/// ways went through different tag → class maps and a sub-pixel lake became an
/// off-by-default grey dot; here, the same three `railway=*` tag strings are written into
/// `rails.parquet` (from `ways.parquet`) and `rail_nodes.parquet` (from `nodes.parquet`)
/// by two separate passes, and their `min_zoom` cells are compared **as read back off
/// disk**.
///
/// **RED ARM:** in `network.rs`'s `build_batches`, make the `min_zoom` array depend on
/// anything but the row's tags — `if network == "rail_nodes" { 12 } else { ... }` is the
/// one-line version — and this fails naming the tag string and the two rungs.
///
/// Non-vacuity: the three tag strings must resolve to three DIFFERENT rungs, or a
/// collapsed ladder that answered 12 to everything would satisfy "they agree".
#[test]
fn a_way_and_a_node_with_the_same_tags_land_on_the_same_rung() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("extract");
let dst = tmp.path().join("net");
write_ways(&src);
write_nodes(&src);
extract_networks(&src, &dst, &NetworkOptions::default()).unwrap();
let ways = min_zoom_by_tags(&dst.join("rails.parquet"));
let nodes = min_zoom_by_tags(&dst.join("rail_nodes.parquet"));
// The scan must have reached both files, or "they agree" is a statement about
// nothing. `rails` also carries `railway=razed`-free rows only, so the counts are
// exactly the three shared strings plus whatever each table holds alone.
assert!(
ways.len() >= BOTH.len() && nodes.len() >= BOTH.len(),
"the two tables hold {} way tag-sets and {} node tag-sets — fewer than the {} \
shared ones, so nothing was compared",
ways.len(),
nodes.len(),
BOTH.len()
);
let mut rungs = Vec::new();
for t in BOTH {
let w = *ways
.get(t)
.unwrap_or_else(|| panic!("`{t}` is missing from rails.parquet"));
let n = *nodes
.get(t)
.unwrap_or_else(|| panic!("`{t}` is missing from rail_nodes.parquet"));
assert_eq!(
w, n,
"`{t}` was written at rung {w} as a WAY and rung {n} as a NODE. Two tag → \
rung maps: that is `korp-map-pt-1` again, where a sub-pixel lake became an \
off-by-default grey dot because points and ways were classified apart."
);
rungs.push(w);
}
// NON-VACUITY. Every assertion above holds trivially on a ladder that returns one
// constant, so the fixture must be shown to span rungs.
rungs.sort_unstable();
rungs.dedup();
assert_eq!(
rungs,
vec![6u8, 9, 12],
"the shared fixture landed on {rungs:?}, not three distinct rungs — a ladder \
collapsed to a constant would pass every equality above"
);
println!("gate 2: {} shared tag sets, rungs {rungs:?}, ways == nodes", BOTH.len());
}
// ── the ordering claim DELTA 1 may NOT make on its own ──────────────────────────
/// ★ **The measurement, so nobody has to take the claim on faith.**
///
/// The advice everybody gives is "add a `min_zoom` column and push the predicate down".
/// On a Hilbert-sorted table that buys **nothing**, because every row group holds every
/// rung and so every row group's `min_zoom` statistics straddle the whole ladder. This
/// test writes the SAME 65 536 rows twice — once in Hilbert order, once ordered
/// `(min_zoom, hilbert)` — and counts, from the parquet footer statistics alone, how many
/// row groups a `min_zoom <= 6` reader may skip.
///
/// Both arms share one row set and one row-group size, so the difference is the ORDER and
/// nothing else. The Hilbert arm is the control: if it ever started pruning, the
/// comparison below would be measuring the fixture rather than the layout.
///
/// This is DELTA 3's gate, measured here early because DELTA 1 is otherwise free to
/// claim it. The number it prints is the honest statement of what the column does and
/// does not buy today.
#[test]
fn band_first_order_prunes_and_hilbert_only_does_not() {
const N: usize = 65_536;
const RG: usize = 8_192;
// A realistic rung mix. Shares are the shape a road network has — a small skeleton of
// arterials, a long tail of service roads and footpaths — not a uniform draw, which
// would flatter the band-first arm.
let mix: [(u8, usize); 4] = [(6, 5), (9, 10), (12, 45), (15, 40)];
let mut rung = Vec::with_capacity(N);
for (r, pct) in mix {
rung.extend(std::iter::repeat_n(r, N * pct / 100));
}
while rung.len() < N {
rung.push(15);
}
// A deterministic spatial key standing in for the Hilbert index, and a deterministic
// shuffle so the rungs are spatially interleaved exactly as they are in reality: a
// motorway and a footpath run through the same square kilometre.
let mut rows: Vec<(u64, u8)> = rung
.iter()
.enumerate()
.map(|(i, &r)| (splitmix(i as u64), r))
.collect();
// ARM A — Hilbert only (what is shipped today).
rows.sort_unstable_by_key(|&(h, _)| h);
let a = tmpfile("hilbert_only");
write_rungs(&a, &rows, RG);
let (a_hit, a_total) = groups_a_z6_reader_must_open(&a);
// ARM B — band first, Hilbert within the band (DELTA 3).
rows.sort_unstable_by_key(|&(h, r)| (r, h));
let b = tmpfile("band_first");
write_rungs(&b, &rows, RG);
let (b_hit, b_total) = groups_a_z6_reader_must_open(&b);
assert_eq!(a_total, b_total, "the two arms must hold the same row groups");
println!(
"min_zoom <= 6 over {N} rows, {a_total} row groups of {RG}:\n \
hilbert-only {a_hit}/{a_total} groups opened ({:.1} %)\n \
(min_zoom,hilbert) {b_hit}/{b_total} groups opened ({:.1} %)\n \
prune {:.1}x",
100.0 * a_hit as f64 / a_total as f64,
100.0 * b_hit as f64 / b_total as f64,
a_hit as f64 / b_hit.max(1) as f64
);
// THE CONTROL ARM, and it is the point of the test: on the shipped layout the column
// prunes NOTHING. If this ever passes with `a_hit < a_total`, the claim below is
// measuring the fixture, not the ordering.
assert_eq!(
a_hit, a_total,
"the Hilbert-only arm pruned {} of {a_total} row groups. It must prune NONE — \
every row group holds every rung — or this test is not the control it claims \
to be and DELTA 1 could take credit for DELTA 3's win.",
a_total - a_hit
);
// And the claim: ordering by band is what makes the predicate bite.
assert!(
b_hit * 4 <= b_total,
"band-first order opened {b_hit} of {b_total} row groups — korp's phase-1 gate is \
a z6 read touching at most 20 %, and this does not reach it"
);
}
// ── plumbing ────────────────────────────────────────────────────────────────────
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)
}
fn tmpfile(stem: &str) -> std::path::PathBuf {
let d = std::env::temp_dir().join(format!("osm-katana-minzoom-{}", std::process::id()));
std::fs::create_dir_all(&d).unwrap();
d.join(format!("{stem}.parquet"))
}
/// A one-column `min_zoom` table plus its spatial key, written at a fixed row-group size.
fn write_rungs(path: &Path, rows: &[(u64, u8)], rg: usize) {
let schema = Arc::new(Schema::new(vec![
Field::new("hilbert", DataType::Int64, false),
osm_katana::shared::schema::min_zoom_field(),
]));
let props = parquet::file::properties::WriterProperties::builder()
.set_max_row_group_row_count(Some(rg))
.build();
let f = std::fs::File::create(path).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema.clone(), Some(props)).unwrap();
let h = Int64Array::from(rows.iter().map(|r| r.0 as i64).collect::<Vec<_>>());
let m = UInt8Array::from(rows.iter().map(|r| r.1).collect::<Vec<_>>());
w.write(&RecordBatch::try_new(schema, vec![Arc::new(h), Arc::new(m)]).unwrap())
.unwrap();
w.close().unwrap();
}
/// How many row groups a reader asking `min_zoom <= 6` must open, from the footer
/// statistics alone — and how many there are. No rows are decoded, which is the whole
/// mechanism under test.
fn groups_a_z6_reader_must_open(path: &Path) -> (usize, usize) {
let f = std::fs::File::open(path).unwrap();
let r = SerializedFileReader::new(f).unwrap();
let md = r.metadata();
let mut hit = 0usize;
for i in 0..md.num_row_groups() {
let rg = md.row_group(i);
let col = (0..rg.num_columns())
.map(|c| rg.column(c))
.find(|c| c.column_descr().name() == "min_zoom")
.expect("min_zoom column must carry statistics");
let lo = match col.statistics() {
Some(Statistics::Int32(s)) => *s.min_opt().expect("min statistic") as u8,
Some(other) => panic!("min_zoom statistics came back as {other:?}"),
None => panic!("no statistics on min_zoom — nothing can prune"),
};
if lo <= 6 {
hit += 1;
}
}
assert!(md.num_row_groups() > 4, "too few row groups to measure a prune");
(hit, md.num_row_groups())
}
/// `tags` → `min_zoom`, read back out of a written network table.
fn min_zoom_by_tags(path: &Path) -> HashMap<String, u8> {
let f = std::fs::File::open(path).unwrap_or_else(|e| panic!("{path:?}: {e}"));
let rdr = ParquetRecordBatchReaderBuilder::try_new(f).unwrap().build().unwrap();
let mut out = HashMap::new();
for b in rdr {
let b = b.unwrap();
let tags = b
.column_by_name("tags")
.expect("tags column")
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let mz = b
.column_by_name("min_zoom")
.expect("min_zoom column — DELTA 1 did not reach this table")
.as_any()
.downcast_ref::<UInt8Array>()
.unwrap();
for i in 0..b.num_rows() {
out.insert(tags.value(i).to_string(), mz.value(i));
}
}
out
}
fn wkb_line([x, y]: [f64; 2]) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&2u32.to_le_bytes());
v.extend_from_slice(&2u32.to_le_bytes());
for k in 0..2 {
v.extend_from_slice(&(x + f64::from(k) * 0.01).to_le_bytes());
v.extend_from_slice(&(y + f64::from(k) * 0.01).to_le_bytes());
}
v
}
fn wkb_point([x, y]: [f64; 2]) -> Vec<u8> {
let mut v = vec![1u8];
v.extend_from_slice(&1u32.to_le_bytes());
v.extend_from_slice(&x.to_le_bytes());
v.extend_from_slice(&y.to_le_bytes());
v
}
fn geo_meta(kind: &str) -> HashMap<String, String> {
let mut m = HashMap::new();
m.insert(String::from(metadata::GEO_KEY), GeoMeta::new(&[kind]).to_json());
m
}
fn write_ways(dir: &Path) {
use arrow::array::{Int64Builder, ListBuilder};
let schema = Arc::new(Schema::new_with_metadata(
vec![
Field::new("id", DataType::Int64, false),
Field::new("geometry", DataType::Binary, true),
Field::new("tags", DataType::Utf8, true),
Field::new(
"node_refs",
DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
true,
),
Field::new("version", DataType::Int32, true),
],
geo_meta("LineString"),
));
// The three shared tag sets, plus one way-only row so the tables are not identical.
let rows: Vec<(i64, &str, [f64; 2])> = BOTH
.iter()
.enumerate()
.map(|(i, t)| (10 + i as i64, *t, [17.0 + i as f64, 59.0]))
.chain(std::iter::once((99, r#"{"highway":"motorway"}"#, [16.0, 58.0])))
.collect();
let ids = Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>());
let geoms: Vec<Vec<u8>> = rows.iter().map(|r| wkb_line(r.2)).collect();
let ga = BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice));
let ta = StringArray::from_iter_values(rows.iter().map(|r| r.1));
let mut lb = ListBuilder::new(Int64Builder::new());
for (i, _) in rows.iter().enumerate() {
lb.values().append_value(i as i64 + 1);
lb.values().append_value(i as i64 + 2);
lb.append(true);
}
let va = Int32Array::from(vec![1i32; rows.len()]);
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(ids),
Arc::new(ga),
Arc::new(ta),
Arc::new(lb.finish()),
Arc::new(va),
],
)
.unwrap();
std::fs::create_dir_all(dir).unwrap();
let f = std::fs::File::create(dir.join("ways.parquet")).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
fn write_nodes(dir: &Path) {
let schema = Arc::new(Schema::new_with_metadata(
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),
],
geo_meta("Point"),
));
let rows: Vec<(i64, &str, [f64; 2])> = BOTH
.iter()
.enumerate()
.map(|(i, t)| (50 + i as i64, *t, [17.0 + i as f64, 59.0]))
.collect();
let ids = Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>());
let geoms: Vec<Vec<u8>> = rows.iter().map(|r| wkb_point(r.2)).collect();
let ga = BinaryArray::from_iter_values(geoms.iter().map(Vec::as_slice));
let ta = StringArray::from_iter_values(rows.iter().map(|r| r.1));
let va = Int32Array::from(vec![1i32; rows.len()]);
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(ids), Arc::new(ga), Arc::new(ta), Arc::new(va)],
)
.unwrap();
std::fs::create_dir_all(dir).unwrap();
let f = std::fs::File::create(dir.join("nodes.parquet")).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(f, schema, None).unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}