osm-katana/tests/triangle_index_cost.rs
//! ★ **Row 33, the arithmetic half: what a pre-tessellated triangle index column would
//! COST on the served cut, exactly.**
//!
//! Row 33 asks for a `tri_z{6,9,12}: List<UInt32>` written at pack time, so facett's
//! ear-clip count on a served viewport is 0. The saving is already measured and it is
//! **cold-only**: GG #14 replaced facett's ear-clip memo with a content-keyed two-
//! generation map, so a warm client pays nothing per frame; what remains is the FIRST
//! pass over each level, which that row priced at **16.5 ms (z11)** and **11.1 ms (z12)**
//! of ear clipping on the 890M against a `REBUILD 2D` budget of 15–19 ms.
//!
//! This file measures the other side of that trade, and it does so **exactly rather than
//! by estimate**, because the triangle count of a polygon is not an empirical question:
//! a simple polygon with `V` vertices and `H` holes triangulates to exactly
//! `V + 2H − 2` triangles, whatever the tessellator. So the index column's RAW size is
//! arithmetic over the WKB — no earcut needed, no sampling, no extrapolation — and it is
//! reported beside the on-disk COMPRESSED size of the very column those triangles would
//! index, read out of the same parquet's own column-chunk metadata.
//!
//! Run:
//! ```text
//! cargo test -p osm-katana --release --test triangle_index_cost -- --ignored --nocapture
//! ```
use std::path::{Path, PathBuf};
use osm_katana::shared::geometry::{WkbShape, decode_wkb};
/// The served cut. Read-only, always.
fn ways_parquet() -> PathBuf {
std::env::var("KATANA_LIVE_WAYS")
.map(PathBuf::from)
.unwrap_or_else(|_| PathBuf::from("/home/korp/.korp/work/sweden/ways.parquet"))
}
/// facett's own line between "the hand-rolled clip is fine" and "this needs earcut"
/// (`facett_map::tessellate::EARCUT_MIN_VERTICES`). A ring under it is a building
/// footprint and costs microseconds; a ring over it is the coastline that cost 100 s of
/// frozen UI before earcutr. Carried as a literal because osm-katana cannot depend on
/// facett; the number is quoted with its source so a reader can check it.
const EARCUT_MIN_VERTICES: usize = 256;
/// What one column would cost.
#[derive(Default, Debug)]
struct Cost {
rows: usize,
poly_rows: usize,
polys: usize,
rings: usize,
holes: usize,
verts: usize,
/// EXACT triangles: `Σ (V + 2H − 2)` over polygons, the identity every tessellator
/// obeys.
tris: usize,
/// … of which come from rings at or over [`EARCUT_MIN_VERTICES`] — the only ones
/// whose tessellation is measurably expensive.
tris_big: usize,
big_rings: usize,
}
impl Cost {
/// Raw bytes of a `List<UInt32>` holding the triples: 3 indices × 4 B per triangle,
/// plus the per-row overhead the column pays for having one cell per ROW of the
/// table rather than one per polygon.
///
/// **The overhead differs between the two variants, and getting that wrong once is
/// why it is spelled out here.** A DENSE column (every polygon carries triples)
/// really does pay an i32 offset per row — 9 325 142 × 4 B = 37.3 MB per column. A
/// SPARSE one (only the rings facett would find expensive) is null in 99.9 % of its
/// cells, and parquet spends a definition level per row, not an offset — call it a
/// bit, 1.17 MB per column. Charging the dense overhead to the sparse variant made
/// its first printed total **181.7 MB when the triangles in it are 32.5 MB**:
/// 149.2 MB of that was offsets for cells that are not there. Corrected here.
fn raw_bytes(&self) -> usize {
self.tris * 12 + self.rows * 4
}
/// The sparse variant: triangle bytes plus a one-bit-per-row null map.
fn raw_bytes_big(&self) -> usize {
self.tris_big * 12 + self.rows.div_ceil(8)
}
fn absorb(&mut self, o: &Cost) {
self.rows += o.rows;
self.poly_rows += o.poly_rows;
self.polys += o.polys;
self.rings += o.rings;
self.holes += o.holes;
self.verts += o.verts;
self.tris += o.tris;
self.tris_big += o.tris_big;
self.big_rings += o.big_rings;
}
}
fn measure(shape: &WkbShape, c: &mut Cost) {
let WkbShape::Polys(ps) = shape else { return };
c.poly_rows += 1;
for (outer, holes) in ps {
c.polys += 1;
c.rings += 1 + holes.len();
c.holes += holes.len();
let v: usize = outer.len() + holes.iter().map(Vec::len).sum::<usize>();
c.verts += v;
// `V + 2H − 2`, floored at 0 — a degenerate ring yields no triangles and must
// not subtract from the total.
let t = (v + 2 * holes.len()).saturating_sub(2);
c.tris += t;
if outer.len() >= EARCUT_MIN_VERTICES {
c.big_rings += 1;
c.tris_big += t;
}
}
}
/// On-disk COMPRESSED bytes of one column, summed over every row group, out of the
/// parquet's own metadata. The like-for-like denominator: the triangle column would sit
/// beside this one, compressed by the same codec.
fn column_bytes(md: &parquet::file::metadata::ParquetMetaData, name: &str) -> u64 {
let mut total = 0u64;
for rg in md.row_groups() {
for c in rg.columns() {
if c.column_path().string() == name {
total += c.compressed_size().max(0) as u64;
}
}
}
total
}
fn columns_present(md: &parquet::file::metadata::ParquetMetaData) -> Vec<String> {
md.file_metadata()
.schema_descr()
.columns()
.iter()
.map(|c| c.path().string())
.collect()
}
/// The columns row 33 is about, in the order a camera climbs them.
const GEOM_COLUMNS: [&str; 4] = ["geometry_z6", "geometry_z9", "geometry_z12", "geometry"];
/// **One column's cost, FANNED ACROSS THE BOX — one gatling unit per ROW GROUP.**
///
/// # This was serial, and the owner caught it
///
/// The first version of this file walked all 1 139 row groups of the served
/// `ways.parquet` on ONE thread, four times over (once per geometry column), and
/// published its wall figures without saying so — 0.1 + 0.4 + 2.4 + 3.5 s of
/// single-threaded WKB decode on a 32-core box, in the repo whose ROOT LAW #0 is that
/// concurrency is gatling. The *counts* it produced were never wrong (a triangle census
/// is `V + 2H − 2`, which no scheduler can move) but every SECOND it reported was
/// 1/32 of the machine.
///
/// So: one unit per row group, LPT-scheduled heaviest-first by the group's own
/// compressed footer size, each unit opening its own projected reader over its single
/// group — **the same shape `side_outputs::buildings::extract_buildings` already uses**,
/// deliberately, so there is one fan-out pattern over parquet in this crate and not two.
/// `Cost` is a plain sum type, so the fold is associative and the answer is identical to
/// the serial one whatever order the units land in — which
/// [`the_fan_out_agrees_with_the_serial_walk`] asserts rather than assumes.
fn cost_of_column(path: &Path, column: &str, view: Option<[f64; 4]>) -> anyhow::Result<Cost> {
use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
let f = std::fs::File::open(path)?;
let meta = ArrowReaderMetadata::load(&f, ArrowReaderOptions::new())?;
if !meta
.parquet_schema()
.columns()
.iter()
.any(|c| c.path().string() == column)
{
return Ok(Cost::default());
}
let pq = meta.metadata().clone();
let groups: Vec<usize> = (0..pq.num_row_groups()).collect();
let per_group = gatling::gatling_forkjoin::gatling_map_balanced(
&groups,
0,
1,
// Heaviest first: a row group's compressed size is what its decode costs, and
// the groups are not equal — the Hilbert pack puts dense city ground in some and
// empty forest in others.
|rg: &usize| pq.row_group(*rg).compressed_size().max(0) as u64,
|_i, rg: &usize| scan_one_group(path, &meta, *rg, column, view),
);
let mut c = Cost::default();
for r in per_group {
c.absorb(&r?);
}
Ok(c)
}
/// One row group's contribution — its own file handle, its own projected reader, no
/// shared mutable state, so the units cannot contend and the fold cannot race.
fn scan_one_group(
path: &Path,
meta: &parquet::arrow::arrow_reader::ArrowReaderMetadata,
rg: usize,
column: &str,
view: Option<[f64; 4]>,
) -> anyhow::Result<Cost> {
use arrow::array::{Array, BinaryArray};
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let builder =
ParquetRecordBatchReaderBuilder::new_with_metadata(std::fs::File::open(path)?, meta.clone());
let mask = ProjectionMask::columns(builder.parquet_schema(), [column]);
let reader = builder.with_row_groups(vec![rg]).with_projection(mask).build()?;
let mut c = Cost::default();
for batch in reader {
let b = batch?;
let g = b
.column_by_name(column)
.and_then(|a| a.as_any().downcast_ref::<BinaryArray>().cloned())
.ok_or_else(|| anyhow::anyhow!("`{column}` is not a Binary column"))?;
for i in 0..b.num_rows() {
c.rows += 1;
if g.is_null(i) {
continue;
}
let Some(shape) = decode_wkb(g.value(i)) else { continue };
if let Some(v) = view {
let Some((x0, y0, x1, y1)) = shape.bbox() else { continue };
if x1 < v[0] || x0 > v[2] || y1 < v[1] || y0 > v[3] {
continue;
}
}
measure(&shape, &mut c);
}
}
Ok(c)
}
fn report(label: &str, c: &Cost, on_disk: u64) {
println!(
" {label:<14} rows {:>9} · poly rows {:>9} · rings {:>10} (holes {:>7}) · verts {:>11}",
c.rows, c.poly_rows, c.rings, c.holes, c.verts
);
println!(
" TRIANGLES {:>11} → raw index column {:>13} B ({:.1} MB) | the column \
it indexes is {:>12} B ({:.1} MB) on disk ⇒ **{:.2}× the geometry it would ride beside**",
c.tris,
c.raw_bytes(),
c.raw_bytes() as f64 / 1e6,
on_disk,
on_disk as f64 / 1e6,
c.raw_bytes() as f64 / on_disk.max(1) as f64
);
println!(
" …restricted to rings ≥ {EARCUT_MIN_VERTICES} verts (facett's own \
earcut line): {:>8} rings, {:>10} triangles, {:>12} B ({:.1} MB, incl. a \
1-bit-per-row null map) ⇒ {:.3}× — {:.4} % of the rings carry {:.2} % of the \
triangles",
c.big_rings,
c.tris_big,
c.raw_bytes_big(),
c.raw_bytes_big() as f64 / 1e6,
c.raw_bytes_big() as f64 / on_disk.max(1) as f64,
c.big_rings as f64 * 100.0 / c.rings.max(1) as f64,
c.tris_big as f64 * 100.0 / c.tris.max(1) as f64,
);
}
/// ★ **The whole served table, column by column.**
#[test]
#[ignore = "decodes every WKB blob in the live 1.5 GB ways.parquet, four times"]
fn what_a_triangle_index_column_would_cost_on_the_served_cut() {
use parquet::file::reader::{FileReader, SerializedFileReader};
let path = ways_parquet();
assert!(path.is_file(), "{path:?} is not on this box — nothing to measure");
let md = SerializedFileReader::new(std::fs::File::open(&path).unwrap())
.unwrap()
.metadata()
.clone();
let present = columns_present(&md);
let file_bytes = std::fs::metadata(&path).unwrap().len();
println!(
"\n{path:?} — {} B ({:.1} MB), {} rows, {} row groups\ncolumns: {}\n",
file_bytes,
file_bytes as f64 / 1e6,
md.file_metadata().num_rows(),
md.num_row_groups(),
present.join(", ")
);
let mut grand_raw = 0usize;
let mut grand_raw_big = 0usize;
for col in GEOM_COLUMNS {
if !present.iter().any(|p| p == col) {
println!(" {col:<14} NOT IN THIS CUT — skipped");
continue;
}
let t0 = std::time::Instant::now();
let c = cost_of_column(&path, col, None).expect("scan the column");
let on_disk = column_bytes(&md, col);
println!("{col} ({:.1} s to decode)", t0.elapsed().as_secs_f64());
report(col, &c, on_disk);
grand_raw += c.raw_bytes();
grand_raw_big += c.raw_bytes_big();
assert!(c.rows > 0, "the column scan must actually read rows");
}
println!(
"\nTOTAL over the rungs + full precision: raw triangle indices {} B ({:.1} MB, \
{:+.1} % of the whole {:.1} MB table); ≥{EARCUT_MIN_VERTICES}-vertex rings only \
{} B ({:.1} MB, {:+.2} %)\n",
grand_raw,
grand_raw as f64 / 1e6,
grand_raw as f64 * 100.0 / file_bytes as f64,
file_bytes as f64 / 1e6,
grand_raw_big,
grand_raw_big as f64 / 1e6,
grand_raw_big as f64 * 100.0 / file_bytes as f64,
);
assert!(grand_raw > 0, "a cut with rungs has triangles to count");
}
/// ★ **One real served viewport — the WIRE cost, which is the half that is paid on every
/// query rather than once.**
///
/// Central Stockholm at z12, the camera GG #19 measured its tag-byte saving over.
#[test]
#[ignore = "decodes every WKB blob in the live 1.5 GB ways.parquet"]
fn what_it_would_cost_on_the_wire_for_one_viewport() {
use parquet::file::reader::{FileReader, SerializedFileReader};
let path = ways_parquet();
assert!(path.is_file(), "{path:?} is not on this box — nothing to measure");
let md = SerializedFileReader::new(std::fs::File::open(&path).unwrap())
.unwrap()
.metadata()
.clone();
// Central Stockholm, roughly the 900×700 pane GG #14 and GG #19 both used.
let view = [17.95_f64, 59.28, 18.16, 59.37];
println!("\nviewport {view:?} (central Stockholm, z12)");
for col in ["geometry_z12", "geometry"] {
if !columns_present(&md).iter().any(|p| p == col) {
continue;
}
let c = cost_of_column(&path, col, Some(view)).expect("scan");
println!(
" {col:<14} {} polygons in view · {} rings · {} triangles ⇒ **{} B ({:.2} MB) \
of index on the wire, per viewport**; ≥{EARCUT_MIN_VERTICES}-vertex rings \
only: {} rings, {} triangles, {} B ({:.3} MB)",
c.polys,
c.rings,
c.tris,
c.tris * 12,
c.tris as f64 * 12.0 / 1e6,
c.big_rings,
c.tris_big,
c.tris_big * 12,
c.tris_big as f64 * 12.0 / 1e6,
);
assert!(c.rows > 0, "the scan read rows");
}
}
/// ★ **RED-when-broken: the fan-out and a serial walk produce the SAME census.**
///
/// A parallel scan whose answer depends on scheduling is a scan that reports a different
/// number every run, and the whole value of this file is that its numbers are exact and
/// reproducible — that is the argument row 24 just made about replacing a clock with a
/// count, and it applies to the instrument as much as to the thing measured.
///
/// So this walks a real column BOTH ways and compares every field. It runs on the live
/// cut when there is one and is skipped-with-a-reason when there is not, because a guard
/// that silently passes on a box with no data is not a guard.
#[test]
#[ignore = "reads the live cut"]
fn the_fan_out_agrees_with_the_serial_walk() {
use arrow::array::{Array, BinaryArray};
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
let path = ways_parquet();
assert!(path.is_file(), "{path:?} is not on this box — nothing to compare");
// The coarsest rung: same shapes, ~93 830 polygons, and it walks in a second serially
// so the comparison itself is cheap.
let column = "geometry_z6";
let t0 = std::time::Instant::now();
let fanned = cost_of_column(&path, column, None).expect("the fanned scan");
let fan_ms = t0.elapsed().as_secs_f64() * 1e3;
// The serial walk, spelled out here rather than kept behind a flag: a second code
// path that only exists to be compared is one that rots, and this one is six lines.
let t1 = std::time::Instant::now();
let f = std::fs::File::open(&path).unwrap();
let meta = ArrowReaderMetadata::load(&f, ArrowReaderOptions::new()).unwrap();
let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(
std::fs::File::open(&path).unwrap(),
meta,
);
let mask = ProjectionMask::columns(builder.parquet_schema(), [column]);
let reader = builder.with_projection(mask).build().unwrap();
let mut serial = Cost::default();
for batch in reader {
let b = batch.unwrap();
let g = b.column_by_name(column).unwrap();
let g = g.as_any().downcast_ref::<BinaryArray>().unwrap();
for i in 0..b.num_rows() {
serial.rows += 1;
if g.is_null(i) {
continue;
}
if let Some(shape) = decode_wkb(g.value(i)) {
measure(&shape, &mut serial);
}
}
}
let ser_ms = t1.elapsed().as_secs_f64() * 1e3;
println!(
"{column}: fanned {fan_ms:.1} ms vs serial {ser_ms:.1} ms ({:.2}x) on {} cores — \
and the CENSUS is what must be identical, not the clock",
ser_ms / fan_ms.max(1e-9),
std::thread::available_parallelism().map_or(0, std::num::NonZeroUsize::get),
);
// NOT VACUOUS: two empty censuses are equal.
assert!(serial.rows > 1_000_000, "the serial walk read {} rows", serial.rows);
assert!(serial.tris > 0, "the serial walk found no triangles");
// Field by field, so a failure says WHICH count the fan-out moved.
for (name, a, b) in [
("rows", fanned.rows, serial.rows),
("poly_rows", fanned.poly_rows, serial.poly_rows),
("polys", fanned.polys, serial.polys),
("rings", fanned.rings, serial.rings),
("holes", fanned.holes, serial.holes),
("verts", fanned.verts, serial.verts),
("tris", fanned.tris, serial.tris),
("tris_big", fanned.tris_big, serial.tris_big),
("big_rings", fanned.big_rings, serial.big_rings),
] {
assert_eq!(a, b, "the fan-out changed `{name}`: {a} fanned vs {b} serial");
}
}