osm-katana/src/writer.rs
#![allow(clippy::arithmetic_side_effects)]
use std::{collections::HashMap, fs, path::Path, sync::Arc};
use anyhow::Context as _;
use arrow::{
array::{
ArrayRef, BinaryBuilder, Int32Builder, Int64Builder, ListBuilder, StringBuilder,
},
datatypes::Schema,
record_batch::RecordBatch,
};
use parquet::{
arrow::ArrowWriter,
arrow::arrow_writer::{ArrowColumnChunk, ArrowRowGroupWriterFactory, compute_leaves},
basic::Compression,
file::properties::WriterProperties,
file::writer::SerializedFileWriter,
};
use crate::shared::{schema, schema::ClassColumns};
use crate::{
metadata,
reader::{NodeRecord, RelationRecord, WayRecord},
};
pub(crate) const ROW_GROUP_SIZE: usize = 524_288; // 512 K rows → fewer zstd calls, better compression
/// Map the CLI compression name onto a parquet codec.
pub(crate) fn codec(compression: &str) -> Compression {
match compression {
"snappy" => Compression::SNAPPY,
"none" => Compression::UNCOMPRESSED,
_ => Compression::ZSTD(Default::default()),
}
}
/// Writer properties carrying the GeoParquet `geo` document as a **file-level
/// Parquet key-value metadata entry**.
///
/// This is the fix for the defect that made osm-katana's output *Parquet-with-WKB*
/// rather than GeoParquet. Putting `geo` only in the Arrow `Schema`'s metadata
/// (what `schema_with_geo` does, and all this crate used to do) buries it inside
/// the base64 `ARROW:schema` blob: arrow-rs and pyarrow decode that blob and so
/// "see" the key, but the spec requires `geo` in the Parquet metadata itself, and
/// GDAL/OGR, DuckDB-spatial and every non-Arrow reader look only there. Measured
/// on the shipped Stockholm pack before this change:
///
/// ```text
/// FILE-LEVEL parquet key_value_metadata keys: [b'ARROW:schema'] ← no 'geo'
/// arrow schema metadata: {b'geo': b'{"version":"1.0.0",…}'} ← hidden here
/// ```
///
/// We now write it in BOTH places: the Arrow-schema copy keeps the existing
/// round-trip behaviour (purely additive), the file-level copy is what makes the
/// file GeoParquet.
fn writer_props(compression: &str, geo_json: &str) -> WriterProperties {
use parquet::file::metadata::KeyValue;
WriterProperties::builder()
.set_compression(codec(compression))
.set_max_row_group_row_count(Some(ROW_GROUP_SIZE))
.set_key_value_metadata(Some(vec![KeyValue::new(
String::from(metadata::GEO_KEY),
String::from(geo_json),
)]))
.build()
}
pub(crate) fn schema_with_geo(schema: Schema, geo_json: String) -> Schema {
let mut meta = HashMap::new();
meta.insert(String::from(metadata::GEO_KEY), geo_json);
schema.with_metadata(meta)
}
// ── Atomic output — the file does not exist until it has been proven ─────────
/// The extension every in-flight parquet output carries.
///
/// Deliberately **not** `parquet`. `korp::pipeline::tile_files` keeps a directory
/// entry when `path.extension() == "parquet"`, and osm-katana's own tile sweeps
/// do `name.trim_end_matches(".parquet")`; a temp that matched either filter
/// would be a worse defect than the one this type exists to fix. The generated
/// name therefore contains no `.parquet` substring at all — see
/// [`PendingOutput::temp_path_for`].
const PARTIAL_EXT: &str = "katana-partial";
/// A parquet file written under a temporary name **in its final directory**, and
/// promoted to the real name only by [`Self::commit`].
///
/// # Why every writer in this module goes through it
///
/// [`RelationWriter::finish`] used to `close()` the `ArrowWriter` first — footer,
/// per-row-group statistics, the trailing `PAR1` magic, all of it — and *then*
/// run the `ensure!` that decides whether the `geo` document describes the rows.
/// A run that tripped the check returned `Err` and left behind a file that is
/// complete and valid to every reader on earth, whose content its own writer had
/// just rejected.
///
/// That is the `seam-24` family — four gates asking `Path::exists()` and
/// accepting a footless parquet from a SIGKILLed convert as a finished extract —
/// only worse: this file is not footless, it is *falsely complete*, so no amount
/// of footer-checking downstream can tell it from a good one.
///
/// The temp lives in the same directory as the output, so promotion is a
/// same-filesystem `rename(2)`: a concurrent reader sees either the previous file
/// or the new one, never a half-written one. A failed, rejected or panicking run
/// drops this guard, which unlinks the temp — so it leaves either nothing or a
/// name nothing will mistake for the output.
///
/// # Why `commit` fsyncs and re-reads the footer (`katana-footer-1`)
///
/// This comment used to say *"`rename` is not `fsync`, and that is on purpose …
/// a host that loses power mid-run can still leave a short file under the final
/// name"*. That reasoning covered POWER LOSS and missed the defect that actually
/// happened, which is not power loss at all: **a write error the process could
/// have seen, and did not.**
///
/// On oden, 2026-08-04 23:23:04, `/dev/md1` (xfs, `/home`) returned `ENOSPC` to a
/// tiling run. One minute later the next run cut a 1 606 570-tile `sweden`
/// pyramid, and at the next startup the pipeline found **34 tiles carrying no
/// parquet footer** — written by a binary that already routed every tile through
/// this type (`skriv-1`). The guard was not bypassed; it was not sufficient.
///
/// xfs and ext4 both do *delayed allocation*: `write(2)` returns success with the
/// bytes only in the page cache, and blocks are allocated at writeback. An
/// `ENOSPC` raised there is reported to `fsync`/`close` — never to the `write`
/// that "succeeded". So `write_all()?` said `Ok`, `drop(file)` discarded the
/// `close` error, and `commit` renamed a file whose tail had never reached the
/// disk into a name every reader treats as a finished tile. korp's upload walk
/// then `fs::read`s whatever carries `.parquet` and PUTs it, so all 34 were
/// served.
///
/// `commit` therefore does three things before the rename, and a failure at any
/// of them drops the guard and unlinks the temp:
///
/// 1. **`sync_all` on the guard's OWN descriptor**, error-checked. It must be the
/// descriptor opened by `create` — Linux reports a writeback error to a given
/// open file description once, measured against the point it was opened, so a
/// freshly-opened handle can sample *past* an error already recorded and
/// return a clean `Ok`. The guard holds its handle from `create` to `commit`
/// precisely so the error cannot escape between them.
/// 2. **The file is at least `PAR1 … PAR1`**, and
/// 3. **its last four bytes are `PAR1`** — read back from disk after the sync.
///
/// A type called `PendingOutput` promoting a non-parquet is the defect; refusing
/// to is LAW 5's "fix by construction". Every writer in this module, plus
/// `spatial::spatial_pack` and the tile loop in `tiles.rs`, promotes through this
/// one function, so the rule is stated once and cannot be forgotten at a new call
/// site. A guard that *detected* footerless tiles after the fact would be the
/// weaker fix; this one means they are never created.
///
/// **Cost, MEASURED on md1 (xfs) at the real z9 median tile size (7 238 B),
/// 20 000 files, one thread:** 29.6 µs/file without `fsync`, 149.2 µs with —
/// 5.04×. At the live pyramid's 1 589 860 tiles that is **+190 thread-seconds on
/// a write phase measured at 2 533.9 thread-seconds (+7.5 %)**, or ≈18 s of wall
/// on the ~12 min run, because the loop runs 31 threads. That is the price of not
/// serving a hole in the map, and it is worth paying. The 900 MB `ways.parquet`
/// the old comment worried about pays ONE fsync, not one per row group.
pub(crate) struct PendingOutput {
final_path: std::path::PathBuf,
temp_path: std::path::PathBuf,
/// The descriptor `create` opened, held until `commit` so `sync_all` reports a
/// writeback error against the point of open. See the type comment.
file: Option<fs::File>,
kind: OutputKind,
committed: bool,
}
/// **What "whole" means for the file this guard is promoting** (`skriv-2`).
///
/// The guard's value is not the `rename(2)` — it is that `commit` can REFUSE, and a
/// refusal needs a per-format answer to "did the writer finish?". Parquet had one
/// (`PAR1`); the three other final artefacts osm-katana writes — the wasm-embed
/// GeoParquet, both Arrow IPC legs of [`crate::geo2arrow`] and
/// `SearchIndex::save`'s JSON — went straight onto their output name with none, so a
/// killed run left a partial file under the name a reader accepts.
///
/// One writer for all four (LAW 5): the temp naming, the held descriptor, the
/// `sync_all`, the drop-unlink and the rename are stated once, and a format only
/// contributes the two facts that differ.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum OutputKind {
/// Apache Parquet: opens and closes with the four bytes `PAR1`.
Parquet,
/// Arrow IPC **file** format (`arrow::ipc::writer::FileWriter`): opens with
/// `ARROW1\0\0` and ends with the 6-byte `ARROW1` trailer that follows the
/// footer and its length. A stream writer's output has no trailer and is NOT
/// this kind.
ArrowIpc,
/// A JSON document (`serde_json::to_writer`). There is no magic, so the check is
/// the strongest structural fact available: the last non-whitespace byte of a
/// finished object or array is `}` or `]`. A truncated write ends mid-token.
Json,
}
impl OutputKind {
/// How short a file of this kind cannot possibly be and still be whole.
const fn min_len(self) -> u64 {
match self {
// Opening `PAR1` + closing `PAR1`.
OutputKind::Parquet => 8,
// `ARROW1\0\0` header + 4-byte footer length + `ARROW1` trailer.
OutputKind::ArrowIpc => 18,
// `{}` / `[]`.
OutputKind::Json => 2,
}
}
/// How many trailing bytes [`Self::tail_is_whole`] needs.
const fn tail_len(self) -> usize {
match self {
OutputKind::Parquet => 4,
OutputKind::ArrowIpc => 6,
// Enough that trailing whitespace from a pretty-printer cannot hide the
// closing brace.
OutputKind::Json => 16,
}
}
/// Does this tail belong to a file whose writer FINISHED?
fn tail_is_whole(self, tail: &[u8]) -> bool {
match self {
OutputKind::Parquet => tail.ends_with(b"PAR1"),
OutputKind::ArrowIpc => tail.ends_with(b"ARROW1"),
OutputKind::Json => matches!(
tail.iter().rev().find(|b| !b.is_ascii_whitespace()),
Some(b'}') | Some(b']')
),
}
}
/// What the refusal message says this file was supposed to end with.
const fn tail_description(self) -> &'static str {
match self {
OutputKind::Parquet => "the four bytes `PAR1` of a parquet footer",
OutputKind::ArrowIpc => "the six bytes `ARROW1` of an Arrow IPC file trailer",
OutputKind::Json => "the `}` or `]` that closes a JSON document",
}
}
}
impl PendingOutput {
/// Create the parent directory, open the temp file, and hand both back: the
/// `fs::File` goes to the `ArrowWriter`, the guard stays with the writer
/// until every check it owes its readers has passed.
/// A pending **parquet** output — the original and by far the busiest caller.
pub(crate) fn parquet(final_path: &Path) -> anyhow::Result<(Self, fs::File)> {
Self::create(final_path, OutputKind::Parquet)
}
/// A pending **Arrow IPC file** output ([`crate::geo2arrow`]).
pub(crate) fn arrow_ipc(final_path: &Path) -> anyhow::Result<(Self, fs::File)> {
Self::create(final_path, OutputKind::ArrowIpc)
}
/// A pending **JSON** output (`SearchIndex::save`).
pub(crate) fn json(final_path: &Path) -> anyhow::Result<(Self, fs::File)> {
Self::create(final_path, OutputKind::Json)
}
pub(crate) fn create(final_path: &Path, kind: OutputKind) -> anyhow::Result<(Self, fs::File)> {
if let Some(parent) = final_path.parent()
&& !parent.as_os_str().is_empty()
{
fs::create_dir_all(parent)
.with_context(|| format!("create parent of {final_path:?}"))?;
}
let temp_path = Self::temp_path_for(final_path)?;
let file = fs::File::create(&temp_path)
.with_context(|| format!("create {temp_path:?} (temp for {final_path:?})"))?;
// The guard keeps a duplicate of the SAME open file description (`dup(2)` via
// `try_clone`), not a re-open: a writeback error is reported once per open file
// description, measured from where it was opened, so only a handle that has been
// open since `create` is guaranteed to see one raised while the caller was
// writing. The caller writes through its copy; the guard never writes or seeks
// through its own, so the shared file offset is not a hazard.
let held = file
.try_clone()
.with_context(|| format!("retain a descriptor for {temp_path:?}"))?;
Ok((
Self {
final_path: final_path.to_path_buf(),
temp_path,
file: Some(held),
kind,
committed: false,
},
file,
))
}
/// `dir/.<stem>.<pid>.<n>.katana-partial`.
///
/// Hidden (leading dot), unique per process **and** per call within one, and
/// built from `file_stem` rather than `file_name` so the result never
/// contains the substring `.parquet` — an `extension() == "parquet"` filter
/// and a `trim_end_matches(".parquet")` one both skip it.
fn temp_path_for(final_path: &Path) -> anyhow::Result<std::path::PathBuf> {
use std::sync::atomic::{AtomicU64, Ordering};
static SEQ: AtomicU64 = AtomicU64::new(0);
let dir = match final_path.parent() {
Some(p) if !p.as_os_str().is_empty() => p,
_ => Path::new("."),
};
let stem = final_path
.file_stem()
.with_context(|| format!("{final_path:?} has no file name to write to"))?
.to_string_lossy()
.into_owned();
let n = SEQ.fetch_add(1, Ordering::Relaxed);
Ok(dir.join(format!(".{stem}.{}.{n}.{PARTIAL_EXT}", std::process::id())))
}
/// The name the file will carry **if** it is committed.
pub(crate) fn path(&self) -> &Path {
&self.final_path
}
/// Promote the temp to the output name — **only if it is a whole parquet that
/// has actually reached the disk.** Nothing downstream can see the file before
/// this returns `Ok`, and an `Err` leaves nothing behind at all (the drop
/// unlinks the temp).
///
/// See the type comment for `katana-footer-1` — the ENOSPC-at-writeback shape
/// this exists to stop, and what it costs.
pub(crate) fn commit(mut self) -> anyhow::Result<()> {
// 1. Force writeback and OBSERVE its error. This is the step whose absence
// let 34 footerless tiles onto the live map: `write_all` had already
// returned `Ok` with the tail still only in the page cache.
let file = self
.file
.take()
.context("PendingOutput::commit called twice — the descriptor is gone")?;
file.sync_all().with_context(|| {
format!(
"fsync {:?} before promoting it to {:?} — the bytes never reached the \
disk, so this file is NOT the table its writer thinks it wrote",
self.temp_path, self.final_path
)
})?;
drop(file);
// 2 + 3. Read the file back the way a reader will see it. `sync_all` above
// means this is the on-disk truth and not a page-cache echo of a
// write that never landed.
let len = fs::metadata(&self.temp_path)
.with_context(|| format!("stat {:?} before promoting it", self.temp_path))?
.len();
let min = self.kind.min_len();
anyhow::ensure!(
len >= min,
"REFUSING to promote {:?} → {:?}: the file is {len} B, and a {:?} output \
cannot be shorter than {min} B and still be whole. It is a truncated \
write, not an artefact",
self.temp_path,
self.final_path,
self.kind
);
let want = self.kind.tail_len() as u64;
let take = want.min(len) as usize;
let mut tail = vec![0u8; take];
{
use std::io::{Read, Seek, SeekFrom};
let mut f = fs::File::open(&self.temp_path)
.with_context(|| format!("re-open {:?} to check its tail", self.temp_path))?;
f.seek(SeekFrom::End(-(take as i64)))
.with_context(|| format!("seek to the tail of {:?}", self.temp_path))?;
f.read_exact(&mut tail)
.with_context(|| format!("read the tail of {:?}", self.temp_path))?;
}
anyhow::ensure!(
self.kind.tail_is_whole(&tail),
"REFUSING to promote {:?} → {:?}: it is {len} B and does NOT end in {} \
(last {take} bytes {tail:?}) — the writer never finished. Under the output \
name this is indistinguishable from a complete artefact to every reader, \
which is exactly how an interrupted run gets SERVED",
self.temp_path,
self.final_path,
self.kind.tail_description()
);
// 4. Only now does the name exist.
fs::rename(&self.temp_path, &self.final_path)
.with_context(|| format!("promote {:?} → {:?}", self.temp_path, self.final_path))?;
self.committed = true;
Ok(())
}
}
impl Drop for PendingOutput {
fn drop(&mut self) {
if !self.committed {
// Best effort: an error here can only mean the temp is already gone.
let _ = fs::remove_file(&self.temp_path);
}
}
}
// ── Nodes ─────────────────────────────────────────────────────────────────────
pub fn write_nodes(path: &Path, records: &[NodeRecord], compression: &str) -> anyhow::Result<()> {
let geo_json = metadata::geo_metadata(&["Point"], None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::nodes_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
let mut w = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
for chunk in records.chunks(ROW_GROUP_SIZE) {
let mut ids = Int64Builder::new();
let mut geoms = BinaryBuilder::new();
let mut tags = StringBuilder::new();
let mut versions = Int32Builder::new();
let mut cls = ClassColumns::new();
for r in chunk {
ids.append_value(r.id);
geoms.append_value(crate::geometry::encode_point_inline(
r.lon_lat.0,
r.lon_lat.1,
));
tags.append_option(Some(&r.tags_json));
versions.append_option(r.version);
cls.push(&r.tags_json);
}
let mut cols: Vec<ArrayRef> = vec![
Arc::new(ids.finish()),
Arc::new(geoms.finish()),
Arc::new(tags.finish()),
Arc::new(versions.finish()),
];
cols.extend(cls.finish());
let batch = RecordBatch::try_new(schema.clone(), cols)?;
w.write(&batch)?;
}
w.close()?;
pending.commit()
}
// ── Ways ──────────────────────────────────────────────────────────────────────
/// **The one place a `WayRecord` becomes Arrow columns.**
///
/// There used to be four: [`write_ways`] built its own builders in a `chunks()` loop,
/// [`WayWriter`] held a second set as struct fields, [`WayAccumulator`] a third, and
/// [`parallel_way_writer`] declared the schema a fourth time. All four repeated the
/// same five builder declarations, the same five `append` calls and the same five-way
/// `RecordBatch::try_new` — ~45 duplicated lines that had to agree, and that is why
/// `include_node_refs` sat unimplemented: a column you can turn off has to be turned
/// off in every writer or the extract's row groups disagree about their own schema.
///
/// They are three different LIFETIMES over one column set — a slice, a file-backed
/// stream, a detached buffer — so the lifetime stays with each of them and the columns
/// live here. Add a column once, and all four paths write it.
pub struct WayColumns {
schema: Arc<Schema>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
/// `None` ⇒ this extract's `ways.parquet` has no `node_refs` column at all
/// (see [`schema::ways_schema`]); the schema and the builders cannot disagree,
/// because the same flag makes both.
node_refs: Option<ListBuilder<Int64Builder>>,
vers: Int32Builder,
/// DELTA 1 + row 18. Valued by [`crate::shared::class::class_of_tags`] from this
/// row's own `tags` and nothing else — never from the table it is going into.
class_cols: ClassColumns,
count: usize,
}
impl WayColumns {
pub fn new(include_node_refs: bool) -> Self {
let geo_json = metadata::geo_metadata(metadata::WAY_GEOMETRY_TYPES, None);
let schema = Arc::new(schema_with_geo(
schema::ways_schema(include_node_refs),
geo_json,
));
Self {
schema,
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
node_refs: include_node_refs.then(|| ListBuilder::new(Int64Builder::new())),
vers: Int32Builder::new(),
class_cols: ClassColumns::new(),
count: 0,
}
}
/// The schema these columns produce — carry it to the `ArrowWriter`, never
/// rebuild it, or the writer and the batch can disagree about `node_refs`.
pub fn schema(&self) -> &Arc<Schema> {
&self.schema
}
/// Rows buffered since the last [`Self::take`].
pub fn len(&self) -> usize {
self.count
}
pub fn is_empty(&self) -> bool {
self.count == 0
}
pub fn push(&mut self, r: &WayRecord) {
self.ids.append_value(r.id);
match &r.geometry {
Some(g) => self.geoms.append_value(g),
None => self.geoms.append_null(),
}
self.tags.append_value(&r.tags_json);
if let Some(nr) = self.node_refs.as_mut() {
for &ref_id in &r.node_refs {
nr.values().append_value(ref_id);
}
nr.append(true);
}
self.vers.append_option(r.version);
self.class_cols.push(&r.tags_json);
self.count += 1;
}
/// Finish the buffered rows into one `RecordBatch` and reset. The column ORDER
/// is the schema's, built from the same flag, so it cannot drift.
pub fn take(&mut self) -> anyhow::Result<RecordBatch> {
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
];
if let Some(nr) = self.node_refs.as_mut() {
cols.push(Arc::new(nr.finish()));
}
cols.push(Arc::new(self.vers.finish()));
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.count = 0;
Ok(batch)
}
}
/// Whether a `ways.parquet` written by the plain (non-streaming) path keeps its
/// `node_refs` column. `true` is the historical shape and what every existing
/// consumer expects.
pub fn write_ways(path: &Path, records: &[WayRecord], compression: &str) -> anyhow::Result<()> {
write_ways_with(path, records, compression, true)
}
/// [`write_ways`] with the `node_refs` column made optional — see
/// [`schema::ways_schema`] for what dropping it costs and saves.
pub fn write_ways_with(
path: &Path,
records: &[WayRecord],
compression: &str,
include_node_refs: bool,
) -> anyhow::Result<()> {
let geo_json = metadata::geo_metadata(metadata::WAY_GEOMETRY_TYPES, None);
let props = writer_props(compression, &geo_json);
let mut cols = WayColumns::new(include_node_refs);
let schema = cols.schema().clone();
let (pending, file) = PendingOutput::parquet(path)?;
let mut w = ArrowWriter::try_new(file, schema, Some(props))?;
for chunk in records.chunks(ROW_GROUP_SIZE) {
for r in chunk {
cols.push(r);
}
let batch = cols.take()?;
w.write(&batch)?;
}
w.close()?;
pending.commit()
}
// ── NodeWriter — streaming sink for pass-1 node records ──────────────────────
/// Streams `NodeRecord`s to a Parquet file in batches of `ROW_GROUP_SIZE`.
/// Avoids accumulating all nodes in a `Vec` — pass 1 can push records directly.
///
/// Serial reference writer for the retained `RawParquetTypedSink` fallback; the
/// live raw/resolved paths use the parallel `ParallelNodeWriter` (worker-encode).
#[allow(dead_code)]
pub struct NodeWriter {
writer: ArrowWriter<fs::File>,
pending: PendingOutput,
schema: Arc<Schema>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
vers: Int32Builder,
class_cols: ClassColumns,
count: usize,
}
#[allow(dead_code)] // retained serial fallback (see NodeWriter doc)
impl NodeWriter {
pub fn new(path: &Path, compression: &str) -> anyhow::Result<Self> {
let geo_json = metadata::geo_metadata(&["Point"], None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::nodes_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
let writer = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
Ok(Self {
writer,
pending,
schema,
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
vers: Int32Builder::new(),
class_cols: ClassColumns::new(),
count: 0,
})
}
pub fn push(&mut self, r: &crate::reader::NodeRecord) -> anyhow::Result<()> {
self.ids.append_value(r.id);
self.geoms
.append_value(crate::geometry::encode_point_inline(
r.lon_lat.0,
r.lon_lat.1,
));
self.tags.append_option(Some(&r.tags_json));
self.vers.append_option(r.version);
self.class_cols.push(&r.tags_json);
self.count += 1;
if self.count >= ROW_GROUP_SIZE {
self.flush()?;
}
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.count == 0 {
return Ok(());
}
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
Arc::new(self.vers.finish()),
];
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.writer.write(&batch)?;
self.count = 0;
Ok(())
}
pub fn finish(mut self) -> anyhow::Result<()> {
self.flush()?;
self.writer.close()?;
self.pending.commit()
}
}
// ── WayWriter — streaming sink for pass-2 way records ────────────────────────
pub struct WayWriter {
pub(crate) writer: ArrowWriter<fs::File>,
pending: PendingOutput,
cols: WayColumns,
}
impl WayWriter {
pub fn new(path: &Path, compression: &str) -> anyhow::Result<Self> {
Self::with_node_refs(path, compression, true)
}
pub fn with_node_refs(
path: &Path,
compression: &str,
include_node_refs: bool,
) -> anyhow::Result<Self> {
let geo_json = metadata::geo_metadata(metadata::WAY_GEOMETRY_TYPES, None);
let props = writer_props(compression, &geo_json);
let cols = WayColumns::new(include_node_refs);
let (pending, file) = PendingOutput::parquet(path)?;
let writer = ArrowWriter::try_new(file, cols.schema().clone(), Some(props))?;
Ok(Self {
writer,
pending,
cols,
})
}
pub fn push(&mut self, r: &crate::reader::WayRecord) -> anyhow::Result<()> {
self.cols.push(r);
if self.cols.len() >= ROW_GROUP_SIZE {
self.flush()?;
}
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.cols.is_empty() {
return Ok(());
}
let batch = self.take_batch()?;
self.writer.write(&batch)?;
Ok(())
}
/// Build a RecordBatch from buffered rows and reset builders.
pub fn take_batch(&mut self) -> anyhow::Result<RecordBatch> {
self.cols.take()
}
pub fn finish(mut self) -> anyhow::Result<()> {
self.flush()?;
self.writer.close()?;
self.pending.commit()
}
}
pub struct RelWriter {
pub(crate) writer: ArrowWriter<fs::File>,
pending: PendingOutput,
schema: Arc<Schema>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
members: StringBuilder,
vers: Int32Builder,
class_cols: ClassColumns,
count: usize,
}
impl RelWriter {
pub fn new(path: &Path, compression: &str) -> anyhow::Result<Self> {
let geo_json = metadata::geo_metadata(metadata::RELATION_GEOMETRY_TYPES, None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::relations_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
let writer = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
Ok(Self {
writer,
pending,
schema,
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
members: StringBuilder::new(),
vers: Int32Builder::new(),
class_cols: ClassColumns::new(),
count: 0,
})
}
pub fn push(&mut self, r: &crate::reader::RelationRecord) -> anyhow::Result<()> {
self.ids.append_value(r.id);
self.geoms.append_null();
self.tags.append_value(&r.tags_json);
self.members.append_value(&r.members_json);
self.vers.append_option(r.version);
self.class_cols.push(&r.tags_json);
self.count += 1;
if self.count >= ROW_GROUP_SIZE {
self.flush()?;
}
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.count == 0 {
return Ok(());
}
let batch = self.take_batch()?;
self.writer.write(&batch)?;
Ok(())
}
/// Build a RecordBatch from buffered rows and reset builders.
pub fn take_batch(&mut self) -> anyhow::Result<RecordBatch> {
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
Arc::new(self.members.finish()),
Arc::new(self.vers.finish()),
];
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.count = 0;
Ok(batch)
}
pub fn finish(mut self) -> anyhow::Result<()> {
self.flush()?;
self.writer.close()?;
self.pending.commit()
}
}
// ── Relations ─────────────────────────────────────────────────────────────────
/// One row of `relations.parquet`, borrowed from whatever produced it.
///
/// `geometry` is `None` for the raw convert path (no member join has happened yet)
/// and `Some(wkb)` for [`crate::relgeom`], which assembles member ways into a
/// MultiPolygon. The two producers write through the SAME writer (LAW 5), so the
/// resolved file and the raw one can never drift in schema or in `geo` document.
pub(crate) struct RelationRow<'a> {
pub id: i64,
pub geometry: Option<&'a [u8]>,
pub tags: &'a str,
pub members: &'a str,
pub version: Option<i32>,
}
/// The ONE writer for `relations.parquet`, streaming, with the **`geo`-document
/// biconditional enforced by construction**.
///
/// # Why the declaration is checked here and not only in a test
///
/// `relations.parquet` shipped `geometry_types: ["MultiPolygon"]` over
/// 14 300 803 NULL geometries (measured, 2026-08-03 Sweden cut) and every
/// row-count check passed — the file was valid parquet and valid GeoParquet, and
/// the only thing wrong with it was the promise. `geo_declaration_tests` catches
/// that on the fixtures it happens to build; this catches it on **every file the
/// program ever writes**, including the ones no test constructs. [`Self::finish`]
/// fails when the declaration and the rows disagree in either direction.
pub(crate) struct RelationWriter {
writer: ArrowWriter<fs::File>,
/// The output name is not claimed until [`Self::finish`] has run the check —
/// see [`PendingOutput`] for why a rejected run used to leave a complete,
/// `PAR1`-stamped file behind.
pending: PendingOutput,
schema: Arc<Schema>,
declared: Vec<String>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
members: StringBuilder,
versions: Int32Builder,
class_cols: ClassColumns,
staged: usize,
pub rows: u64,
/// Rows whose geometry column is non-null — the fact the declaration must match.
pub with_geometry: u64,
}
impl RelationWriter {
pub(crate) fn create(
path: &Path,
compression: &str,
geometry_types: &[&str],
) -> anyhow::Result<Self> {
let geo_json = metadata::geo_metadata(geometry_types, None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::relations_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
let writer = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
Ok(Self {
writer,
pending,
schema,
declared: geometry_types.iter().map(|s| (*s).to_string()).collect(),
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
members: StringBuilder::new(),
versions: Int32Builder::new(),
class_cols: ClassColumns::new(),
staged: 0,
rows: 0,
with_geometry: 0,
})
}
pub(crate) fn push(&mut self, r: RelationRow<'_>) -> anyhow::Result<()> {
self.ids.append_value(r.id);
match r.geometry {
Some(g) if !g.is_empty() => {
self.geoms.append_value(g);
self.with_geometry += 1;
}
_ => self.geoms.append_null(),
}
self.tags.append_value(r.tags);
self.members.append_value(r.members);
self.versions.append_option(r.version);
self.class_cols.push(r.tags);
self.staged += 1;
self.rows += 1;
if self.staged >= ROW_GROUP_SIZE {
self.flush()?;
}
Ok(())
}
fn flush(&mut self) -> anyhow::Result<()> {
if self.staged == 0 {
return Ok(());
}
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
Arc::new(self.members.finish()),
Arc::new(self.versions.finish()),
];
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.writer.write(&batch)?;
self.staged = 0;
Ok(())
}
/// Close the temp file, refuse a `geo` document that does not describe the
/// rows, and only then claim the output name.
///
/// The order is the whole point. This used to `close()` — footer, statistics,
/// trailing `PAR1` — straight onto the output path and run the `ensure!`
/// afterwards, so a rejected run returned `Err` and still left a file that
/// every reader on earth calls complete. Now the footer lands on a
/// `.katana-partial` temp; the rename happens after the check, and a run that
/// fails the check drops [`PendingOutput`], which unlinks the temp. There is
/// no state in which the output name holds a table this writer rejected.
pub(crate) fn finish(mut self) -> anyhow::Result<(u64, u64)> {
self.flush()?;
self.writer.close()?;
anyhow::ensure!(
!self.declared.is_empty() == (self.with_geometry > 0),
"{:?}: `geo` declares geometry_types {:?} but {} of {} rows carry geometry — \
a GeoParquet `geo` document must describe the file, not a plan for it",
self.pending.path(),
self.declared,
self.with_geometry,
self.rows,
);
let (rows, with_geometry) = (self.rows, self.with_geometry);
self.pending.commit()?;
Ok((rows, with_geometry))
}
}
pub fn write_relations(
path: &Path,
records: &[RelationRecord],
compression: &str,
) -> anyhow::Result<()> {
let mut w = RelationWriter::create(path, compression, metadata::RELATION_GEOMETRY_TYPES)?;
for r in records {
w.push(RelationRow {
id: r.id,
// The convert path has done no member join, so it has no geometry to
// write. `osm-katana relations` is the stage that fills it in.
geometry: None,
tags: &r.tags_json,
members: &r.members_json,
version: r.version,
})?;
}
w.finish()?;
Ok(())
}
// ── Accumulators — build RecordBatches on parse side, no file I/O ─────────────
/// Accumulates node records into Arrow builders; emits a RecordBatch at ROW_GROUP_SIZE.
/// Pair with [`ParquetSink`] so the collector only does cheap builder appends
/// and the writer thread handles zstd + I/O off the hot path.
pub struct NodeAccumulator {
schema: Arc<Schema>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
vers: Int32Builder,
class_cols: ClassColumns,
count: usize,
}
impl NodeAccumulator {
pub fn new() -> Self {
let geo_json = metadata::geo_metadata(&["Point"], None);
let schema = Arc::new(schema_with_geo(schema::nodes_schema(), geo_json));
Self {
schema,
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
vers: Int32Builder::new(),
class_cols: ClassColumns::new(),
count: 0,
}
}
pub fn push(&mut self, r: &crate::reader::NodeRecord) {
self.ids.append_value(r.id);
self.geoms
.append_value(crate::geometry::encode_point_inline(
r.lon_lat.0,
r.lon_lat.1,
));
self.tags.append_option(Some(&r.tags_json));
self.vers.append_option(r.version);
self.class_cols.push(&r.tags_json);
self.count += 1;
}
pub fn take_if_full(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if self.count >= ROW_GROUP_SIZE {
Ok(Some(self.take()?))
} else {
Ok(None)
}
}
pub fn take_remaining(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if self.count > 0 {
Ok(Some(self.take()?))
} else {
Ok(None)
}
}
fn take(&mut self) -> anyhow::Result<RecordBatch> {
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
Arc::new(self.vers.finish()),
];
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.count = 0;
Ok(batch)
}
}
impl Default for NodeAccumulator {
fn default() -> Self {
Self::new()
}
}
/// Accumulates way records into Arrow builders; emits a RecordBatch at ROW_GROUP_SIZE.
pub struct WayAccumulator {
cols: WayColumns,
}
impl WayAccumulator {
pub fn new() -> Self {
Self::with_node_refs(true)
}
pub fn with_node_refs(include_node_refs: bool) -> Self {
Self {
cols: WayColumns::new(include_node_refs),
}
}
pub fn push(&mut self, r: &WayRecord) {
self.cols.push(r);
}
/// Returns a batch if we've hit ROW_GROUP_SIZE.
pub fn take_if_full(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if self.cols.len() >= ROW_GROUP_SIZE {
Ok(Some(self.cols.take()?))
} else {
Ok(None)
}
}
/// Drain remaining rows (call at end).
pub fn take_remaining(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if !self.cols.is_empty() {
Ok(Some(self.cols.take()?))
} else {
Ok(None)
}
}
}
/// Accumulates relation records into Arrow builders; emits a RecordBatch at ROW_GROUP_SIZE.
pub struct RelAccumulator {
schema: Arc<Schema>,
ids: Int64Builder,
geoms: BinaryBuilder,
tags: StringBuilder,
members: StringBuilder,
vers: Int32Builder,
class_cols: ClassColumns,
count: usize,
}
impl RelAccumulator {
pub fn new() -> Self {
let geo_json = metadata::geo_metadata(metadata::RELATION_GEOMETRY_TYPES, None);
let schema = Arc::new(schema_with_geo(schema::relations_schema(), geo_json));
Self {
schema,
ids: Int64Builder::new(),
geoms: BinaryBuilder::new(),
tags: StringBuilder::new(),
members: StringBuilder::new(),
vers: Int32Builder::new(),
class_cols: ClassColumns::new(),
count: 0,
}
}
pub fn push(&mut self, r: &RelationRecord) {
self.ids.append_value(r.id);
self.geoms.append_null();
self.tags.append_value(&r.tags_json);
self.members.append_value(&r.members_json);
self.vers.append_option(r.version);
self.class_cols.push(&r.tags_json);
self.count += 1;
}
pub fn take_if_full(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if self.count >= ROW_GROUP_SIZE {
Ok(Some(self.take()?))
} else {
Ok(None)
}
}
pub fn take_remaining(&mut self) -> anyhow::Result<Option<RecordBatch>> {
if self.count > 0 {
Ok(Some(self.take()?))
} else {
Ok(None)
}
}
fn take(&mut self) -> anyhow::Result<RecordBatch> {
let mut cols: Vec<ArrayRef> = vec![
Arc::new(self.ids.finish()),
Arc::new(self.geoms.finish()),
Arc::new(self.tags.finish()),
Arc::new(self.members.finish()),
Arc::new(self.vers.finish()),
];
cols.extend(self.class_cols.finish());
let batch = RecordBatch::try_new(self.schema.clone(), cols)?;
self.count = 0;
Ok(batch)
}
}
/// One fully-encoded, compressed Parquet row group: the column chunks for a
/// contiguous run of rows, ready to be stitched into the file without any
/// further compression. Produced on a Gatling worker, consumed (appended) by
/// the single in-order collector.
pub type EncodedRowGroup = Vec<ArrowColumnChunk>;
/// Encodes node `RecordBatch`es into compressed Parquet row groups.
///
/// Cheaply cloneable (`Arc` inside) and `Sync`, so every Gatling worker can
/// encode **its own** segment's row group in parallel: the zstd compression and
/// page encoding run on the worker thread, *not* on the single collector. This
/// is pure CPU on the calling thread — it spawns no threads and uses no rayon.
#[derive(Clone)]
pub struct NodeColumnEncoder {
factory: Arc<ArrowRowGroupWriterFactory>,
schema: Arc<Schema>,
}
impl NodeColumnEncoder {
/// **Does the file this encoder writes carry `name`?**
///
/// The encoder holds the schema the parquet footer will declare, so it is the ONE
/// place that knows whether an optional column exists (LAW 5). An accumulator that
/// decided that for itself could disagree with the writer — and did: `katana-6`'s
/// first cut wired `include_node_refs` into `parallel_way_writer_with` while
/// `pass2_accumulate`'s per-worker `WAY_ACC` still built its columns with
/// `WayAccumulator::new()` (always `true`), so every worker handed a 5-column batch
/// to a 4-column schema and 32 threads died inside arrow with `Field 'version' has
/// type Int32, array has type List(Int64)`. Asking the encoder removes the
/// disagreement instead of guarding against it.
pub fn has_column(&self, name: &str) -> bool {
self.schema.field_with_name(name).is_ok()
}
/// Encode one `RecordBatch` into a compressed row group (one column chunk
/// per leaf column, in schema order). Runs entirely on the caller's thread.
pub fn encode(&self, batch: &RecordBatch) -> anyhow::Result<EncodedRowGroup> {
// Say WHICH schemas disagree, here, rather than letting arrow report it as a
// type error about whichever column happened to line up wrong — the zip below
// pairs by POSITION, so a missing optional column shifts every field after it
// and the message names an innocent one (`katana-6`).
anyhow::ensure!(
self.schema.fields().len() == batch.num_columns(),
"row group encode: the writer's schema has {} column(s) {:?} but the batch \
has {} — an optional column was turned off on one side only",
self.schema.fields().len(),
self.schema
.fields()
.iter()
.map(|f| f.name().as_str())
.collect::<Vec<_>>(),
batch.num_columns(),
);
let mut writers = self.factory.create_column_writers(0)?;
let mut w = writers.iter_mut();
for (field, col) in self.schema.fields().iter().zip(batch.columns()) {
for leaf in compute_leaves(field.as_ref(), col)? {
w.next()
.ok_or_else(|| anyhow::anyhow!("column writer/leaf count mismatch"))?
.write(&leaf)?;
}
}
writers
.into_iter()
.map(|cw| cw.close().map_err(anyhow::Error::from))
.collect()
}
}
/// In-order Parquet file writer that only **stitches** pre-compressed row
/// groups produced by the workers. Runs on the single collector thread but does
/// no compression — it appends already-encoded column chunks and writes the
/// footer, so the collector is no longer the serial zstd bottleneck.
pub struct ParallelNodeWriter {
writer: SerializedFileWriter<fs::File>,
/// Same contract as the streaming writers: the stitched groups land on a
/// `.katana-partial` temp and take the output name only in [`Self::finish`],
/// so a collector that dies mid-run leaves nothing a directory sweep can
/// mistake for a finished table. See [`PendingOutput`].
pending: PendingOutput,
}
impl ParallelNodeWriter {
/// Append one pre-encoded row group to the file (pure I/O + metadata).
pub fn append(&mut self, rg: EncodedRowGroup) -> anyhow::Result<()> {
if rg.is_empty() {
return Ok(());
}
let mut rgw = self.writer.next_row_group()?;
for chunk in rg {
chunk.append_to_row_group(&mut rgw)?;
}
rgw.close()?;
Ok(())
}
pub fn finish(self) -> anyhow::Result<()> {
self.writer.close()?;
self.pending.commit()
}
}
/// Build a parallel node writer together with the matching column encoder.
///
/// The encoder (handed to the Gatling workers) and the writer (kept by the
/// collector) share the same Arrow/Parquet schema and the GeoParquet `geo`
/// metadata, so the stitched file is byte-for-byte a valid GeoParquet node
/// table — just encoded in parallel instead of on one thread.
pub fn parallel_node_writer(
path: &Path,
compression: &str,
) -> anyhow::Result<(NodeColumnEncoder, ParallelNodeWriter)> {
let geo_json = metadata::geo_metadata(&["Point"], None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::nodes_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
// ArrowWriter sets up the file metadata (geo + ARROW:schema); we then split
// it into the low-level SerializedFileWriter + the row-group factory so the
// actual encode can happen off-thread on the workers.
let aw = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
let (writer, factory) = aw.into_serialized_writer()?;
Ok((
NodeColumnEncoder {
factory: Arc::new(factory),
schema,
},
ParallelNodeWriter { writer, pending },
))
}
/// Schema-agnostic alias: the column encoder works for ANY Arrow schema (it
/// iterates `self.schema.fields()`), so the same type drives node, way and
/// relation row-group encoding on the workers. Named `NodeColumnEncoder` for
/// historical reasons; the way/rel pass-2 parallel path reuses it verbatim.
pub type ColumnEncoder = NodeColumnEncoder;
/// Schema-agnostic alias for the in-order row-group stitcher (see
/// [`ParallelNodeWriter`]). Reused by the pass-2 way/rel parallel path.
pub type ParallelRowGroupWriter = ParallelNodeWriter;
/// Build a parallel **way** writer + its column encoder. Mirror of
/// [`parallel_node_writer`] for the ways GeoParquet table (LineString/Polygon
/// geometry). The encoder is handed to the pass-2 Gatling workers so each
/// builds + zstd-encodes its own way row groups in parallel; the writer stays
/// on the collector and only stitches the pre-encoded groups in order.
pub fn parallel_way_writer(
path: &Path,
compression: &str,
) -> anyhow::Result<(ColumnEncoder, ParallelRowGroupWriter)> {
parallel_way_writer_with(path, compression, true)
}
/// [`parallel_way_writer`] with the optional `node_refs` column. The schema comes
/// from [`WayColumns`], the same object the row groups are built with, so this
/// writer and its batches cannot disagree about the column set.
pub fn parallel_way_writer_with(
path: &Path,
compression: &str,
include_node_refs: bool,
) -> anyhow::Result<(ColumnEncoder, ParallelRowGroupWriter)> {
let geo_json = metadata::geo_metadata(metadata::WAY_GEOMETRY_TYPES, None);
let props = writer_props(compression, &geo_json);
let schema = WayColumns::new(include_node_refs).schema().clone();
let (pending, file) = PendingOutput::parquet(path)?;
let aw = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
let (writer, factory) = aw.into_serialized_writer()?;
Ok((
NodeColumnEncoder {
factory: Arc::new(factory),
schema,
},
ParallelNodeWriter { writer, pending },
))
}
/// Build a parallel **relation** writer + its column encoder. Mirror of
/// [`parallel_node_writer`] for the relations table (no geometry).
pub fn parallel_rel_writer(
path: &Path,
compression: &str,
) -> anyhow::Result<(ColumnEncoder, ParallelRowGroupWriter)> {
let geo_json = metadata::geo_metadata(metadata::RELATION_GEOMETRY_TYPES, None);
let props = writer_props(compression, &geo_json);
let schema = Arc::new(schema_with_geo(schema::relations_schema(), geo_json));
let (pending, file) = PendingOutput::parquet(path)?;
let aw = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
let (writer, factory) = aw.into_serialized_writer()?;
Ok((
NodeColumnEncoder {
factory: Arc::new(factory),
schema,
},
ParallelNodeWriter { writer, pending },
))
}
// The former `ParquetSink` background-writer thread (single-thread node zstd +
// I/O, fed by a per-record `NodeAccumulator` on the collector) was removed once
// every convert path — raw single-pass and both resolved 2-pass entries — moved
// node encoding onto the Gatling workers via `NodeColumnEncoder` +
// `ParallelNodeWriter` (the collector only stitches pre-compressed row groups).
#[cfg(test)]
mod parallel_node_tests {
use super::*;
use crate::reader::NodeRecord;
use std::time::Instant;
fn make_batch(start: i64, n: i64) -> RecordBatch {
let mut acc = NodeAccumulator::new();
for i in start..start + n {
acc.push(&NodeRecord {
id: i,
lon_lat: (i as f64 * 1e-6, i as f64 * 2e-6),
tags_json: r#"{"amenity":"bench","note":"parallel-encode-bench"}"#.to_string(),
version: Some(1),
});
}
acc.take_remaining().unwrap().unwrap()
}
fn total_rows_in(path: &std::path::Path) -> (i64, usize, bool) {
let file = fs::File::open(path).unwrap();
let builder =
parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
let meta = builder.metadata();
let rows = meta.file_metadata().num_rows();
let rgs = meta.num_row_groups();
// GeoParquet `geo` metadata is carried inside the `ARROW:schema` kv (as
// arrow-rs does for the normal ArrowWriter path); the reader merges it
// back into the recovered Arrow schema's metadata.
let has_geo = builder.schema().metadata().contains_key("geo");
(rows, rgs, has_geo)
}
/// Regression for the 32 767-row-group parquet cap (planet SIGSEGV).
///
/// Before the fix, each worker-segment flushed its sub-full remainder as its
/// own row group (`take_remaining` per segment) → ~one group per segment →
/// >32 767 groups on planet → parquet `append` error → gatling UAF crash.
///
/// The fix persists one `NodeAccumulator` per worker across its segments and
/// only seals full `ROW_GROUP_SIZE` batches mid-stream (`take_if_full`),
/// flushing the remainder once at the end. This test mirrors that: many small
/// segments fed into a single persistent accumulator must coalesce into
/// `ceil(total / ROW_GROUP_SIZE)` groups — NOT one per segment — with every
/// group but the last exactly `ROW_GROUP_SIZE`.
#[test]
fn persistent_accumulator_packs_segments_into_full_row_groups() {
let seg_rows = 160_000usize; // ~one bz2 segment's worth, < ROW_GROUP_SIZE
let n_segments = 40usize; // would be 40 tiny groups under the old path
let total = seg_rows * n_segments;
let mut acc = NodeAccumulator::new();
let mut group_sizes: Vec<usize> = Vec::new();
let mut next_id = 0i64;
for _ in 0..n_segments {
for _ in 0..seg_rows {
acc.push(&NodeRecord {
id: next_id,
lon_lat: (0.0, 0.0),
tags_json: String::new(),
version: None,
});
next_id += 1;
// Seal full groups mid-segment, exactly like the worker does.
if let Some(b) = acc.take_if_full().unwrap() {
group_sizes.push(b.num_rows());
}
}
// NOTE: no per-segment take_remaining — the remainder carries forward.
}
// End-of-worker flush (finish_worker).
if let Some(b) = acc.take_remaining().unwrap() {
group_sizes.push(b.num_rows());
}
let expected_groups = total.div_ceil(ROW_GROUP_SIZE);
assert_eq!(
group_sizes.len(),
expected_groups,
"expected coalesced groups, got one-per-segment fragmentation"
);
assert!(
group_sizes.len() < n_segments,
"packing must produce fewer groups than segments"
);
for (i, &sz) in group_sizes.iter().enumerate() {
if i + 1 < group_sizes.len() {
assert_eq!(sz, ROW_GROUP_SIZE, "non-final group {i} must be full");
} else {
assert_eq!(
sz,
total - ROW_GROUP_SIZE * (expected_groups - 1),
"final group is the remainder"
);
}
}
assert_eq!(group_sizes.iter().sum::<usize>(), total, "no rows lost");
}
/// Each gatling unit encodes its own row group in parallel — exactly how the
/// Gatling pass-1 workers feed the collector.
/// Verifies the stitched file is valid GeoParquet with the right row count
/// and one row group per worker batch.
#[test]
fn parallel_encode_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("nodes.parquet");
let (enc, mut w) = parallel_node_writer(&path, "zstd").unwrap();
let k: i64 = 8;
let n: i64 = 20_000;
let batches: Vec<RecordBatch> = (0..k).map(|j| make_batch(j * n, n)).collect();
// Parallel encode, ROOT LAW #0: gatling fan-out (NOT a hand-rolled
// `thread::scope` pool), all units sharing the same (Sync) encoder — no
// rayon, no shared writer lock. Results come back in batch order, which
// is what the collector stitches.
let groups: Vec<EncodedRowGroup> = gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| enc.encode(b).unwrap(),
);
// Collector stitches the pre-compressed row groups in order.
for g in groups {
w.append(g).unwrap();
}
w.finish().unwrap();
let (rows, rgs, has_geo) = total_rows_in(&path);
assert_eq!(rows, k * n, "row count must match");
assert_eq!(rgs, k as usize, "one row group per worker batch");
assert!(has_geo, "GeoParquet `geo` metadata must be preserved");
}
/// Pass-2 saturation fix: way + relation row groups must be Arrow-built +
/// zstd-encoded ON THE WORKERS (parallel, shared `ColumnEncoder`) and merely
/// stitched in order by the collector — exactly like the pass-1 node path.
/// Injects real Way/RelationRecords across several "worker" threads, stitches
/// the pre-encoded groups, and asserts the round-tripped file has the right
/// row count, one group per worker batch, and preserved GeoParquet metadata.
#[test]
fn parallel_way_rel_encode_roundtrip() {
use crate::reader::{RelationRecord, WayRecord};
let dir = tempfile::tempdir().unwrap();
// ── Ways ──────────────────────────────────────────────────────────────
let way_path = dir.path().join("ways.parquet");
let (way_enc, mut way_w) = parallel_way_writer(&way_path, "zstd").unwrap();
let k = 6usize;
let n = 5_000i64;
let way_batches: Vec<RecordBatch> = (0..k as i64)
.map(|j| {
let mut acc = WayAccumulator::new();
for i in j * n..(j + 1) * n {
acc.push(&WayRecord {
id: i,
geometry: Some(vec![1u8, 2, 3, 4]), // dummy WKB
tags_json: r#"{"highway":"residential"}"#.to_string(),
node_refs: vec![i, i + 1, i + 2],
version: Some(3),
});
}
acc.take_remaining().unwrap().unwrap()
})
.collect();
// Parallel encode across gatling units sharing one Sync encoder (ROOT LAW
// #0 — no rayon, no hand-rolled `thread::scope` pool).
let way_groups: Vec<EncodedRowGroup> = gatling::gatling_forkjoin::gatling_map_balanced(
&way_batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| way_enc.encode(b).unwrap(),
);
for g in way_groups {
way_w.append(g).unwrap();
}
way_w.finish().unwrap();
let (rows, rgs, has_geo) = total_rows_in(&way_path);
assert_eq!(
rows,
k as i64 * n,
"way row count must match injected records"
);
assert_eq!(rgs, k, "one way row group per worker batch");
assert!(has_geo, "ways GeoParquet `geo` metadata must be preserved");
// ── Relations ───────────────────────────────────────────────────────────
let rel_path = dir.path().join("relations.parquet");
let (rel_enc, mut rel_w) = parallel_rel_writer(&rel_path, "zstd").unwrap();
let rel_batches: Vec<RecordBatch> = (0..k as i64)
.map(|j| {
let mut acc = RelAccumulator::new();
for i in j * n..(j + 1) * n {
acc.push(&RelationRecord {
id: i,
tags_json: r#"{"type":"multipolygon"}"#.to_string(),
members_json: r#"[{"type":"way","ref":1,"role":"outer"}]"#.to_string(),
version: Some(2),
});
}
acc.take_remaining().unwrap().unwrap()
})
.collect();
let rel_groups: Vec<EncodedRowGroup> = gatling::gatling_forkjoin::gatling_map_balanced(
&rel_batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| rel_enc.encode(b).unwrap(),
);
for g in rel_groups {
rel_w.append(g).unwrap();
}
rel_w.finish().unwrap();
let (rrows, rrgs, rgeo) = total_rows_in(&rel_path);
assert_eq!(
rrows,
k as i64 * n,
"relation row count must match injected records"
);
assert_eq!(rrgs, k, "one relation row group per worker batch");
assert!(
rgeo,
"relations GeoParquet `geo` metadata must be preserved"
);
}
/// Benchmark: parallel (per-worker-thread) encode vs serial single-thread
/// encode of the same row groups. Prints the speedup; asserts the parallel
/// path is not slower than serial within noise (so a regression that
/// accidentally serializes the encode is caught). Run with
/// `cargo test -p osm2geoparquet --release -- --nocapture parallel_encode_speedup`.
#[test]
fn parallel_encode_speedup() {
let threads = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
let k: i64 = threads as i64 * 2; // a couple of row groups per core
let n: i64 = 50_000;
let batches: Vec<RecordBatch> = (0..k).map(|j| make_batch(j * n, n)).collect();
let dir = tempfile::tempdir().unwrap();
let (enc, _w) = parallel_node_writer(&dir.path().join("bench.parquet"), "zstd").unwrap();
// Serial: encode every row group on one thread.
let t0 = Instant::now();
for b in &batches {
std::hint::black_box(enc.encode(b).unwrap());
}
let serial = t0.elapsed();
// Parallel: gatling fan-out over the row groups, shared encoder. Same
// engine the convert path uses, so this measures the real thing.
let t1 = Instant::now();
std::hint::black_box(gatling::gatling_forkjoin::gatling_map_balanced(
&batches,
0,
1,
|b: &RecordBatch| b.num_rows() as u64,
|_i, b: &RecordBatch| std::hint::black_box(enc.encode(b).unwrap()),
));
let parallel = t1.elapsed();
let speedup = serial.as_secs_f64() / parallel.as_secs_f64();
eprintln!(
"parallel node encode: {k} row groups × {n} rows on {threads} cores → \
serial={serial:?} parallel={parallel:?} speedup={speedup:.2}×"
);
// The encode must genuinely run in parallel: allow generous slack for a
// busy or low-core CI box, but a fully-serialized regression (speedup
// ≈ 1 or worse on a multi-core box) should fail here.
if threads >= 4 {
assert!(
speedup > 1.5,
"expected parallel encode speedup > 1.5× on {threads} cores, got {speedup:.2}×"
);
}
}
}
#[cfg(test)]
mod geo_declaration_tests {
//! FAIL-ON-BUG for **a `geo` document that promises geometry the file does
//! not hold.**
//!
//! `relations.parquet` declared `geometry_types: ["MultiPolygon"]` and wrote
//! 14 300 803 NULLs (measured on the 2026-08-03 Sweden cut). Nothing failed:
//! the file is valid parquet, valid GeoParquet, and every row-count check
//! passed. The only thing wrong with it was the promise — which is exactly
//! the class of defect no count-based test can see.
//!
//! So this asserts the biconditional over BOTH tables at once, from the
//! bytes on disk: a non-empty `geometry_types` **iff** at least one row's
//! geometry is non-null. `ways.parquet` is in the same test on purpose —
//! without an arm that legitimately DOES declare and DOES deliver, a guard
//! that only ever sees an empty declaration would pass on a producer that
//! stopped writing way geometry too.
use std::fs;
use crate::reader::{RelationRecord, WayRecord};
use crate::shared::gpu_backend::{CpuBackend, GpuBackend as _, RawWay};
/// The declared `geometry_types` list, read out of the file's own `geo`
/// key-value metadata. `None` when there is no `geo` document at all.
fn declared_types(path: &std::path::Path) -> Option<Vec<String>> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let b = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(path).unwrap()).unwrap();
let kv = b.metadata().file_metadata().key_value_metadata()?.clone();
let geo = kv.iter().find(|k| k.key == "geo")?.value.clone()?;
// Minimal extraction — the writer emits a fixed shape, and pulling in a
// JSON parser here would test the parser, not the declaration.
let i = geo.find("\"geometry_types\":[")? + "\"geometry_types\":[".len();
let rest = &geo[i..];
let end = rest.find(']')?;
let inner = &rest[..end];
Some(
inner
.split(',')
.map(|s| s.trim().trim_matches('"').to_string())
.filter(|s| !s.is_empty())
.collect(),
)
}
/// Assert the biconditional on one file.
fn assert_declaration_matches_rows(path: &std::path::Path, label: &str) {
let types =
declared_types(path).unwrap_or_else(|| panic!("{label}: no `geo` document at all"));
let c = crate::verify::verify(path).expect("verify");
assert!(
c.rows > 0,
"{label}: fixture wrote no rows — the guard would be vacuous"
);
assert!(c.has_geom, "{label}: no geometry column");
assert_eq!(
!types.is_empty(),
c.geom_set > 0,
"{label}: declared {types:?} but {} of {} rows carry geometry — \
a `geo` document must describe the file, not a plan for it",
c.geom_set,
c.rows,
);
}
#[test]
fn declared_geometry_types_match_the_rows_actually_written() {
let dir = tempfile::tempdir().unwrap();
// ── ways: DECLARES and DELIVERS ──────────────────────────────────────
let way_path = dir.path().join("ways.parquet");
// A real 2-point WKB LineString from the SAME encoder the converter
// uses, so `verify` counts these rows as geometry-set.
let wkb = CpuBackend
.encode_wkb(&[RawWay {
coords: vec![(17.0, 59.0), (17.1, 59.1)],
is_area: false,
}])
.remove(0);
assert!(wkb.is_some(), "fixture WKB encoder produced nothing");
let ways: Vec<WayRecord> = (1..=200i64)
.map(|i| WayRecord {
id: i,
// A real 2-point WKB LineString, so `verify` counts it as set.
geometry: wkb.clone(),
tags_json: r#"{"highway":"residential"}"#.to_string(),
node_refs: vec![i, i + 1],
version: Some(1),
})
.collect();
super::write_ways(&way_path, &ways, "zstd").unwrap();
assert_declaration_matches_rows(&way_path, "ways.parquet");
// ── relations: DECLARES NOTHING, delivers nothing ────────────────────
let rel_path = dir.path().join("relations.parquet");
let rels: Vec<RelationRecord> = (1..=200i64)
.map(|i| RelationRecord {
id: i,
tags_json: r#"{"type":"multipolygon","natural":"water"}"#.to_string(),
members_json: r#"[{"type":"way","ref":1,"role":"outer"}]"#.to_string(),
version: Some(1),
})
.collect();
super::write_relations(&rel_path, &rels, "zstd").unwrap();
assert_declaration_matches_rows(&rel_path, "relations.parquet");
}
}
// ── stinky katana: the way table had four column builders ────────────────────
#[cfg(test)]
mod way_column_law {
//! **One builder, four writers.** `write_ways`, [`super::WayWriter`],
//! [`super::WayAccumulator`] and `parallel_way_writer` each held their own copy
//! of the way table's five Arrow builders and their own `RecordBatch::try_new`.
//! Four copies that must agree is why `node_refs` could not be made optional: a
//! column turned off in three writers and on in the fourth gives one
//! `ways.parquet` whose row groups disagree with its own schema.
//!
//! These guards assert the two properties that consolidation is FOR — every
//! writer produces the same schema, and the same rows — and they are red the
//! moment any writer grows its own opinion again.
use super::*;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
fn ways(n: i64) -> Vec<WayRecord> {
(0..n)
.map(|i| WayRecord {
id: 1_000 + i,
geometry: Some(vec![1u8, 2, 0, 0, 0, (i % 251) as u8]),
tags_json: format!(r#"{{"k":"v{i}"}}"#),
node_refs: vec![i, i + 1, i + 2, i],
version: Some((i % 97) as i32),
})
.collect()
}
/// Column names of a written `ways.parquet`, in file order.
fn columns_of(path: &Path) -> Vec<String> {
let b = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(path).unwrap()).unwrap();
b.schema()
.fields()
.iter()
.map(|f| f.name().clone())
.collect()
}
/// Every row of a written `ways.parquet` as (id, version, tags), read back.
fn rows_of(path: &Path) -> Vec<(i64, i32, String)> {
use arrow::array::{Int32Array, Int64Array, StringArray};
let reader = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(path).unwrap())
.unwrap()
.build()
.unwrap();
let mut out = Vec::new();
for batch in reader {
let b = batch.unwrap();
let id = b
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let ver = b
.column_by_name("version")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
let tg = b
.column_by_name("tags")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
for i in 0..b.num_rows() {
out.push((id.value(i), ver.value(i), tg.value(i).to_string()));
}
}
out
}
/// **The three record-taking writers agree, byte-shape for byte-shape.** Same
/// records in, same column list and same rows out — whichever writer wrote them.
#[test]
fn every_way_writer_produces_the_same_table() {
let recs = ways(2_500);
let dir = tempfile::tempdir().unwrap();
let a = dir.path().join("a.parquet");
write_ways(&a, &recs, "zstd").unwrap();
let b = dir.path().join("b.parquet");
let mut w = WayWriter::new(&b, "zstd").unwrap();
for r in &recs {
w.push(r).unwrap();
}
w.finish().unwrap();
// The accumulator hands batches to a caller-owned ArrowWriter — the shape
// the streaming paths use.
let c = dir.path().join("c.parquet");
let mut acc = WayAccumulator::new();
let geo_json = metadata::geo_metadata(metadata::WAY_GEOMETRY_TYPES, None);
let schema = WayColumns::new(true).schema().clone();
let mut cw = ArrowWriter::try_new(
fs::File::create(&c).unwrap(),
schema,
Some(writer_props("zstd", &geo_json)),
)
.unwrap();
for r in &recs {
acc.push(r);
if let Some(batch) = acc.take_if_full().unwrap() {
cw.write(&batch).unwrap();
}
}
if let Some(batch) = acc.take_remaining().unwrap() {
cw.write(&batch).unwrap();
}
cw.close().unwrap();
let want_cols = vec![
"id",
"geometry",
"tags",
"node_refs",
"version",
"min_zoom",
"class_id",
"kind",
"layer",
"bridge",
"tunnel",
"oneway",
];
for (name, p) in [
("write_ways", &a),
("WayWriter", &b),
("WayAccumulator", &c),
] {
assert_eq!(
columns_of(p),
want_cols,
"{name} wrote a different column set"
);
assert_eq!(rows_of(p).len(), recs.len(), "{name} lost rows");
}
assert_eq!(
rows_of(&a),
rows_of(&b),
"write_ways and WayWriter disagree"
);
assert_eq!(
rows_of(&a),
rows_of(&c),
"write_ways and WayAccumulator disagree"
);
// The instrument is alive: the rows are not all identical to each other,
// so an equality that passed on a constant would be meaningless.
let r = rows_of(&a);
assert!(
r[0] != r[1] && r.len() > 2,
"fixture rows must differ from each other"
);
}
/// **`include_node_refs = false` drops the column and NOTHING else.**
///
/// Not a null-filled column, not empty lists — absent. Every other column, and
/// every row, is unchanged. MEASURED on korp's 2026-08-03 `sweden` cut, this is
/// 237.1 MB of an 893.3 MB `ways.parquet` (26.5 %) and 1 009.3 MB of
/// 3 648.3 MB uncompressed (27.7 %).
#[test]
fn dropping_node_refs_drops_that_column_and_only_that_column() {
let recs = ways(1_200);
let dir = tempfile::tempdir().unwrap();
let with = dir.path().join("with.parquet");
let without = dir.path().join("without.parquet");
write_ways_with(&with, &recs, "zstd", true).unwrap();
write_ways_with(&without, &recs, "zstd", false).unwrap();
// Written out, never derived from `ways_schema` — a column list compared with
// the schema it came from is the identity trap, and this test's whole job is to
// say what the file's columns ARE.
assert_eq!(
columns_of(&with),
vec![
"id", "geometry", "tags", "node_refs", "version", "min_zoom", "class_id",
"kind", "layer", "bridge", "tunnel", "oneway"
]
);
assert_eq!(
columns_of(&without),
vec![
"id", "geometry", "tags", "version", "min_zoom", "class_id", "kind",
"layer", "bridge", "tunnel", "oneway"
],
"node_refs must be ABSENT, not a column of empty lists — and neither \
`min_zoom` nor the row-18 class block is optional, so dropping refs must \
not drop them"
);
assert_eq!(
rows_of(&with),
rows_of(&without),
"dropping node_refs must not touch any other column or any row"
);
assert_eq!(rows_of(&with).len(), recs.len());
// And it is actually smaller on disk — the whole point.
let (bw, bo) = (
fs::metadata(&with).unwrap().len(),
fs::metadata(&without).unwrap().len(),
);
println!(
"ways.parquet with node_refs {bw} B, without {bo} B ({:.1} % smaller)",
100.0 * (bw - bo) as f64 / bw as f64
);
assert!(bo < bw, "the leaner table must be smaller: {bo} vs {bw}");
}
/// The streaming writers honour the same flag as the batch one — the shape
/// that made the four-copy version impossible to change safely.
#[test]
fn the_streaming_writers_honour_the_node_refs_flag_too() {
let recs = ways(600);
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("ww.parquet");
let mut w = WayWriter::with_node_refs(&p, "zstd", false).unwrap();
for r in &recs {
w.push(r).unwrap();
}
w.finish().unwrap();
assert_eq!(
columns_of(&p),
vec![
"id", "geometry", "tags", "version", "min_zoom", "class_id", "kind",
"layer", "bridge", "tunnel", "oneway"
]
);
assert_eq!(rows_of(&p).len(), recs.len());
let mut acc = WayAccumulator::with_node_refs(false);
for r in &recs {
acc.push(r);
}
let batch = acc.take_remaining().unwrap().expect("rows buffered");
assert_eq!(
batch
.schema()
.fields()
.iter()
.map(|f| f.name().as_str())
.collect::<Vec<_>>(),
vec![
"id", "geometry", "tags", "version", "min_zoom", "class_id", "kind",
"layer", "bridge", "tunnel", "oneway"
],
"WayAccumulator must drop the column too, or its batches will not match \
the file schema the writer was opened with"
);
assert_eq!(batch.num_rows(), recs.len());
}
}
#[cfg(test)]
mod relation_writer_law {
//! FAIL-ON-BUG for **a `geo` document that does not describe its own rows**,
//! enforced where the file is written rather than on the fixtures a test
//! happens to build.
//!
//! `geo_declaration_tests` (below) asserts the same biconditional from the
//! bytes on disk, and it is the reason we know the pre-2026-08-03
//! `relations.parquet` was lying. But it can only judge files a test made.
//! [`RelationWriter::finish`] judges every file the program ever writes, and
//! that is the difference between a guard and a property.
use super::*;
fn one_row(geometry: Option<&[u8]>) -> RelationRow<'_> {
RelationRow {
id: 7,
geometry,
tags: "{}",
members: "[]",
version: None,
}
}
/// Every name in `dir`, sorted — the view a directory sweep gets.
fn listing(dir: &Path) -> Vec<String> {
let mut v: Vec<String> = fs::read_dir(dir)
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
v.sort();
v
}
/// Does this file end in the parquet magic — i.e. would every reader on
/// earth call it a complete parquet file?
fn is_par1_stamped(p: &Path) -> bool {
let Ok(b) = fs::read(p) else { return false };
b.len() >= 8 && &b[b.len() - 4..] == b"PAR1"
}
/// ★ **RED-when-broken: promising MultiPolygon over NULLs FAILS the write.**
///
/// This is the exact shape of the shipped defect — `geometry_types:
/// ["MultiPolygon"]` over 14 300 803 NULL geometries, valid parquet, valid
/// GeoParquet, every row count correct. Both directions are asserted, because
/// a check that only catches over-promising would pass on a writer that
/// silently stopped declaring geometry it does deliver.
#[test]
fn a_geo_document_that_lies_about_its_rows_cannot_be_closed() {
let d = tempfile::tempdir().unwrap();
// Over-promise: declares MultiPolygon, writes NULL.
let p = d.path().join("over.parquet");
let mut w = RelationWriter::create(&p, "none", &["MultiPolygon"]).unwrap();
w.push(one_row(None)).unwrap();
let e = w
.finish()
.expect_err("declaring geometry over NULL rows must fail");
let msg = format!("{e:#}");
assert!(
msg.contains("0 of 1 rows carry geometry"),
"the error must name the counts that disagree, got: {msg}"
);
// Under-declare: writes real geometry, declares none.
let wkb = crate::shared::geometry::encode_wkb_multipolygon(&[(
vec![(0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 0.0)],
vec![],
)]);
let p = d.path().join("under.parquet");
let mut w = RelationWriter::create(&p, "none", &[]).unwrap();
w.push(one_row(Some(&wkb))).unwrap();
w.finish()
.expect_err("writing geometry under an empty declaration must fail");
// …and the two HONEST files close cleanly, so the guard is not simply
// "every write fails".
let p = d.path().join("null_ok.parquet");
let mut w = RelationWriter::create(&p, "none", &[]).unwrap();
w.push(one_row(None)).unwrap();
assert_eq!(w.finish().unwrap(), (1, 0));
let p = d.path().join("geom_ok.parquet");
let mut w = RelationWriter::create(&p, "none", &["MultiPolygon"]).unwrap();
w.push(one_row(Some(&wkb))).unwrap();
assert_eq!(w.finish().unwrap(), (1, 1));
}
/// ★ **RED-when-broken: a rejected run leaves NO `PAR1`-stamped output.**
///
/// The check above proves `finish` returns `Err`. It says nothing about what
/// is on disk afterwards, and for five commits the answer was: a complete
/// file. `finish` called `writer.close()` — footer, statistics, trailing
/// `PAR1` — onto the output path, and ran the `ensure!` after. Every reader
/// on earth would have opened `over.parquet` and found a valid GeoParquet
/// table with 1 row and `geometry_types: ["MultiPolygon"]` — the exact lie
/// the writer had just refused to sign. That is the `seam-24` family with the
/// evidence removed: not a footless file a gate can spot, a *falsely
/// complete* one it cannot.
///
/// So this asserts the applied output — the bytes in the directory — and not
/// the return value. Reverting `finish` to close-then-check makes it fail on
/// the first `assert!`.
#[test]
fn a_rejected_run_leaves_no_output_file_at_all() {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("over.parquet");
let mut w = RelationWriter::create(&p, "none", &["MultiPolygon"]).unwrap();
w.push(one_row(None)).unwrap();
// Mid-flight the output name is still unclaimed, and the temp in the
// directory is not something a `*.parquet` sweep would pick up.
assert!(
!p.exists(),
"the output name must not exist before finish() commits"
);
let mid = listing(d.path());
assert_eq!(mid.len(), 1, "expected exactly the temp, got {mid:?}");
assert!(
mid[0].starts_with('.')
&& mid[0].ends_with(PARTIAL_EXT)
&& !mid[0].contains(".parquet"),
"the temp must be hidden, carry the partial extension and be free of \
`.parquet` (korp's tile_files and the tile sweeps filter on it), got {mid:?}"
);
w.finish()
.expect_err("declaring geometry over NULL rows must fail");
assert!(
!p.exists(),
"a run the writer rejected left {p:?} behind — PAR1-stamped and \
complete to every reader alive"
);
assert!(
!is_par1_stamped(&p),
"{p:?} carries the parquet end magic after a rejected run"
);
assert!(
listing(d.path()).is_empty(),
"a rejected run must leave the directory as it found it, got {:?}",
listing(d.path())
);
}
/// The temp name has to survive **both** filters that actually run over these
/// directories: `korp::pipeline::tile_files` keeps entries whose
/// `extension() == "parquet"`, and osm-katana's tile sweeps parse
/// `name.trim_end_matches(".parquet")`. Neither may ever see a partial.
#[test]
fn the_temp_name_cannot_be_mistaken_for_output() {
for out in [
Path::new("/x/nodes.parquet"),
Path::new("/x/ways.parquet"),
Path::new("/x/relations.parquet"),
Path::new("/x/relations.resolved.parquet"),
Path::new("/x/tiles/z12/1130/590.parquet"),
] {
let t = PendingOutput::temp_path_for(out).unwrap();
assert_eq!(
t.parent(),
out.parent(),
"the temp must share the output's directory"
);
assert_ne!(t, out.to_path_buf());
let name = t.file_name().unwrap().to_string_lossy().into_owned();
assert!(name.starts_with('.'), "temp must be hidden: {name}");
assert_ne!(
t.extension()
.map(|e| e.to_string_lossy().into_owned())
.as_deref(),
Some("parquet"),
"an extension()==\"parquet\" sweep would swallow {name}"
);
assert!(
!name.contains(".parquet"),
"a trim_end_matches(\".parquet\") sweep would mangle {name}"
);
}
// Unique per call, or two concurrent writers to the same table would
// stomp each other's temp.
let a = PendingOutput::temp_path_for(Path::new("/x/nodes.parquet")).unwrap();
let b = PendingOutput::temp_path_for(Path::new("/x/nodes.parquet")).unwrap();
assert_ne!(a, b, "two temps for the same output must not collide");
}
/// ★ **RED PROBE for `katana-footer-1`: `commit` must REFUSE a footerless temp,
/// and the rename it used to do would have accepted it.**
///
/// This is the deliberately-truncated tile the fix is judged on. It reproduces
/// what oden's xfs did on 2026-08-04: bytes that pass `write_all` but are not a
/// whole parquet by the time the name goes live.
///
/// Both halves are asserted **in the same test, against the same bytes**, because
/// a fix is only shown to be a fix by the comparison:
///
/// * the OLD promotion (`fs::rename`, reproduced literally below) puts a
/// footerless file under the output name — the green that could never go red;
/// * the NEW one returns `Err`, names `PAR1` in the message, and leaves **no
/// file at the output name and no temp** behind.
///
/// Deleting the footer check from `commit` fails this test on
/// `expect_err`; deleting the `Drop` unlink fails it on the stray-temp sweep.
#[test]
fn commit_refuses_a_footerless_file_that_a_bare_rename_would_have_published() {
let d = tempfile::tempdir().unwrap();
// A tile whose tail never reached the disk: opens like a parquet, and stops.
// Byte-for-byte what a reader of the 34 offenders would have found.
let truncated: &[u8] = b"PAR1\x15\x04\x19\x0c\x00\x00truncated-mid-row-group";
// ── The OLD promotion, reproduced exactly: create temp, write, rename. ──
// Proof the previous code path could not have caught this.
let old_out = d.path().join("old_way.parquet");
{
let (pending, mut f) = PendingOutput::parquet(&old_out).unwrap();
use std::io::Write as _;
f.write_all(truncated).unwrap();
drop(f);
// What `commit` used to be, verbatim.
fs::rename(&pending.temp_path, &pending.final_path).unwrap();
std::mem::forget(pending); // don't let Drop unlink the renamed-away temp
}
assert!(
old_out.exists(),
"the OLD promotion was a bare rename, so it MUST publish the truncated file \
— if this fails the probe is not reproducing the defect it claims to"
);
let published = fs::read(&old_out).unwrap();
assert_ne!(
&published[published.len() - 4..],
b"PAR1",
"the file the old path published really does NOT end in PAR1 — this is the \
footerless tile, live under a name every reader trusts"
);
// ── The NEW promotion, same bytes. ──
let new_out = d.path().join("new_way.parquet");
let err = {
let (pending, mut f) = PendingOutput::parquet(&new_out).unwrap();
use std::io::Write as _;
f.write_all(truncated).unwrap();
drop(f);
pending.commit().expect_err(
"commit MUST refuse a file that does not end in PAR1 — this failing is \
the whole of katana-footer-1 regressing",
)
};
let msg = format!("{err:#}");
assert!(
msg.contains("PAR1"),
"the refusal must name the missing footer so an operator can act on it; got: {msg}"
);
assert!(
!new_out.exists(),
"a refused commit must leave NOTHING at the output name — a partial file \
under the real name is the entire defect"
);
// …and no temp either: the guard's Drop unlinks it, so a directory sweep
// cannot find a second thing to mistake for a tile.
let strays: Vec<_> = fs::read_dir(d.path())
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.file_name().to_string_lossy().into_owned())
.filter(|n| n.contains(PARTIAL_EXT))
.collect();
assert!(
strays.is_empty(),
"a refused commit left temps behind: {strays:?}"
);
}
/// The other half of LAW 2: the new guard is not simply "every commit fails".
/// A genuine parquet — written by a real `ArrowWriter` through the same type —
/// still lands under its own name.
#[test]
fn commit_still_promotes_a_whole_parquet() {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("nodes.parquet");
write_nodes(&p, &[], "none").expect("an honest empty table must still commit");
assert!(p.exists(), "the good path must still produce its output");
let bytes = fs::read(&p).unwrap();
assert_eq!(
&bytes[..4],
b"PAR1",
"and it is a real parquet, opening with PAR1"
);
assert_eq!(&bytes[bytes.len() - 4..], b"PAR1", "…and closing with it");
}
/// An empty temp — the `create`d-but-never-written case a SIGKILL between open
/// and first write leaves — is refused on LENGTH, before the footer read can
/// panic on a negative seek. Zero-byte files were 0 of the 34 offenders
/// measured on oden, but they are the same class and the seek would have
/// underflowed.
#[test]
fn commit_refuses_an_empty_temp_without_panicking() {
let d = tempfile::tempdir().unwrap();
let out = d.path().join("empty.parquet");
let (pending, f) = PendingOutput::parquet(&out).unwrap();
drop(f);
let err = pending
.commit()
.expect_err("a 0-byte file is not a parquet table");
assert!(
format!("{err:#}").contains("truncated write"),
"an empty temp must be refused on length, with a message that says so"
);
assert!(!out.exists());
}
/// A writer that is dropped without `finish` — a `?` early-return, a panic, a
/// cancelled convert — must not leave the temp lying around either.
#[test]
fn an_abandoned_writer_cleans_up_after_itself() {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("nodes.parquet");
{
let mut w = RelationWriter::create(&p, "none", &[]).unwrap();
w.push(one_row(None)).unwrap();
assert_eq!(
listing(d.path()).len(),
1,
"temp should exist while writing"
);
}
assert!(!p.exists());
assert!(
listing(d.path()).is_empty(),
"dropping a writer must unlink its temp, got {:?}",
listing(d.path())
);
}
/// The sibling writers go through the same door (LAW 5): none of them claims
/// its output name before its own `finish`.
#[test]
fn every_writer_claims_its_name_only_at_finish() {
let d = tempfile::tempdir().unwrap();
let n = d.path().join("nodes.parquet");
let mut nw = NodeWriter::new(&n, "none").unwrap();
nw.push(&crate::reader::NodeRecord {
id: 1,
lon_lat: (0.0, 0.0),
tags_json: String::from("{}"),
version: None,
})
.unwrap();
assert!(
!n.exists(),
"NodeWriter claimed nodes.parquet before finish"
);
nw.finish().unwrap();
assert!(
is_par1_stamped(&n),
"NodeWriter must commit a complete file"
);
let wp = d.path().join("ways.parquet");
let mut ww = WayWriter::new(&wp, "none").unwrap();
ww.push(&crate::reader::WayRecord {
id: 1,
geometry: Some(crate::geometry::encode_point_inline(0.0, 0.0).to_vec()),
tags_json: String::from("{}"),
node_refs: Vec::new(),
version: None,
})
.unwrap();
assert!(!wp.exists(), "WayWriter claimed ways.parquet before finish");
ww.finish().unwrap();
assert!(
is_par1_stamped(&wp),
"WayWriter must commit a complete file"
);
let rp = d.path().join("relations.parquet");
let mut rw = RelWriter::new(&rp, "none").unwrap();
rw.push(&crate::reader::RelationRecord {
id: 1,
tags_json: String::from("{}"),
members_json: String::from("[]"),
version: None,
})
.unwrap();
assert!(
!rp.exists(),
"RelWriter claimed relations.parquet before finish"
);
rw.finish().unwrap();
assert!(
is_par1_stamped(&rp),
"RelWriter must commit a complete file"
);
let pp = d.path().join("parallel.parquet");
let (enc, mut pw) = parallel_node_writer(&pp, "none").unwrap();
assert!(
!pp.exists(),
"ParallelNodeWriter claimed its name before finish"
);
// The class block comes from `ClassColumns` here too, so this guard's batch
// cannot fall out of step with the schema when a column is added.
let mut cols: Vec<ArrayRef> = vec![
Arc::new(Int64Builder::new().finish()) as ArrayRef,
Arc::new(BinaryBuilder::new().finish()),
Arc::new(StringBuilder::new().finish()),
Arc::new(Int32Builder::new().finish()),
];
cols.extend(ClassColumns::new().finish());
let batch = RecordBatch::try_new(enc.schema.clone(), cols).unwrap();
pw.append(enc.encode(&batch).unwrap()).unwrap();
assert!(
!pp.exists(),
"ParallelNodeWriter claimed its name after append"
);
pw.finish().unwrap();
assert!(
is_par1_stamped(&pp),
"ParallelNodeWriter must commit a complete file"
);
// Nothing partial survived any of the four.
assert_eq!(
listing(d.path()),
vec![
"nodes.parquet".to_string(),
"parallel.parquet".to_string(),
"relations.parquet".to_string(),
"ways.parquet".to_string(),
],
);
}
}
#[cfg(test)]
mod pending_output_kind_tests {
use super::*;
use std::io::Write as _;
/// ★ **RED-when-broken (`skriv-2`): a half-written artefact never gets the output
/// name — and the check is not the same check for all three formats.**
///
/// The three final artefacts this widened the guard to cover — the wasm-embed
/// GeoParquet ([`crate::optimize`]), both Arrow IPC legs of [`crate::geo2arrow`]
/// and `SearchIndex::save`'s JSON — used to be `File::create(final_name)` → write →
/// close, so a killed run left a partial file under the name the next reader
/// accepts. They now share the parquet writers' guard.
///
/// This drives **applied output**: the file system after `commit`, not a returned
/// flag. A refused promotion must leave the final name ABSENT — a guard that
/// returns `Err` and renames anyway is the defect wearing a hat.
#[test]
fn a_truncated_output_is_refused_and_a_whole_one_is_promoted() {
let dir = tempfile::tempdir().unwrap();
// (kind, whole bytes, truncated bytes) — the truncation is what a writer that
// never reached its own trailer actually leaves.
let cases: [(OutputKind, Vec<u8>, Vec<u8>); 3] = [
(
OutputKind::Parquet,
b"PAR1\x00\x00\x00\x00PAR1".to_vec(),
b"PAR1\x00\x00\x00\x00\x00\x00\x00\x00".to_vec(),
),
(
OutputKind::ArrowIpc,
b"ARROW1\x00\x00................\x08\x00\x00\x00ARROW1".to_vec(),
b"ARROW1\x00\x00................\x08\x00\x00\x00".to_vec(),
),
(
OutputKind::Json,
br#"{"features":[1,2,3]}"#.to_vec(),
br#"{"features":[1,2,"#.to_vec(),
),
];
for (kind, whole, truncated) in cases {
// THE RED HALF FIRST, so a `commit` that always said Ok cannot hide behind
// the green one below.
let bad = dir.path().join(format!("{kind:?}-truncated.out"));
let (pending, mut f) = PendingOutput::create(&bad, kind).unwrap();
f.write_all(&truncated).unwrap();
drop(f);
let err = pending
.commit()
.expect_err(&format!("{kind:?}: a truncated artefact was PROMOTED"));
let msg = format!("{err:#}");
assert!(
msg.contains("REFUSING to promote"),
"{kind:?}: refused for the wrong reason — {msg}"
);
assert!(
!bad.exists(),
"{kind:?}: commit refused and the output name EXISTS anyway — {bad:?}"
);
// And the guard is not simply refusing everything.
let good = dir.path().join(format!("{kind:?}-whole.out"));
let (pending, mut f) = PendingOutput::create(&good, kind).unwrap();
f.write_all(&whole).unwrap();
drop(f);
pending
.commit()
.unwrap_or_else(|e| panic!("{kind:?}: a WHOLE artefact was refused — {e:#}"));
assert_eq!(
std::fs::read(&good).unwrap(),
whole,
"{kind:?}: promoted, but not byte-for-byte what was written"
);
}
// No temp survives either outcome: the drop unlinks it.
let strays: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.file_name().to_string_lossy().into_owned())
.filter(|n| n.contains(PARTIAL_EXT))
.collect();
assert!(strays.is_empty(), "temp files left behind: {strays:?}");
}
/// The three kinds do not accept each other's tails — otherwise one lax check
/// would be silently doing the work of all three.
#[test]
fn each_kind_refuses_the_other_kinds_trailer() {
assert!(OutputKind::Parquet.tail_is_whole(b"PAR1"));
assert!(!OutputKind::Parquet.tail_is_whole(b"ARROW1"));
assert!(!OutputKind::Parquet.tail_is_whole(b"...}"));
assert!(OutputKind::ArrowIpc.tail_is_whole(b"ARROW1"));
assert!(!OutputKind::ArrowIpc.tail_is_whole(b"PAR1"));
assert!(OutputKind::Json.tail_is_whole(b"}"));
assert!(OutputKind::Json.tail_is_whole(b"]\n \t"));
assert!(!OutputKind::Json.tail_is_whole(b"PAR1"));
assert!(!OutputKind::Json.tail_is_whole(br#""na"#));
}
}
#[cfg(test)]
mod no_column_is_all_null {
use super::*;
/// ★ **RED-when-broken (`katana-13`): no column of a written table may be null in
/// every row.**
///
/// `nodes.parquet` declared `changeset` and `timestamp` and `append_null`ed both on
/// every row, unconditionally, because neither node writer ever read the PBF `Info`
/// block they come from. MEASURED on korp's 2026-08-03 `sweden` cut: 102 518 979
/// rows, 102 518 979 nulls in each — 205 037 958 cells no code path could fill. A
/// consumer joining on `changeset` got zero rows and nothing went wrong anywhere.
///
/// The guard is stated as a LAW over the whole schema rather than as two names, so
/// the next column somebody declares before wiring it is caught by the same test.
/// It reads the parquet's own row-group statistics — the applied output, what a
/// reader sees — not the builders.
///
/// Both node writers are driven: the batch path ([`write_nodes`]) and the streaming
/// one ([`NodeWriter`]), because the defect was duplicated across them.
#[test]
fn a_written_nodes_table_has_no_column_that_is_null_in_every_row() {
use parquet::file::reader::{FileReader, SerializedFileReader};
let dir = tempfile::tempdir().unwrap();
let recs: Vec<NodeRecord> = (0..64)
.map(|k| NodeRecord {
id: k + 1,
lon_lat: (18.0 + k as f64 * 1e-4, 59.0),
tags_json: String::from(r#"{"place":"city"}"#),
version: Some(1),
})
.collect();
let batch_path = dir.path().join("batch.parquet");
write_nodes(&batch_path, &recs, "zstd").unwrap();
let stream_path = dir.path().join("stream.parquet");
let mut w = NodeWriter::new(&stream_path, "zstd").unwrap();
for r in &recs {
w.push(r).unwrap();
}
w.finish().unwrap();
for path in [&batch_path, &stream_path] {
let rdr = SerializedFileReader::new(std::fs::File::open(path).unwrap()).unwrap();
let md = rdr.metadata();
let mut rows = 0i64;
let mut nulls: Vec<i64> = vec![0; md.file_metadata().schema_descr().num_columns()];
for rg in 0..md.num_row_groups() {
let g = md.row_group(rg);
rows += g.num_rows();
for (c, n) in nulls.iter_mut().enumerate() {
*n += g
.column(c)
.statistics()
.and_then(|s| s.null_count_opt())
.map(|v| v as i64)
.unwrap_or(0);
}
}
// NON-VACUITY: an empty table has no all-null column either, and that would
// make this pass for the wrong reason.
assert_eq!(rows, recs.len() as i64, "{path:?}: wrote no rows to inspect");
assert!(!nulls.is_empty(), "{path:?}: no columns to inspect");
let schema = md.file_metadata().schema_descr();
let dead: Vec<String> = nulls
.iter()
.enumerate()
.filter(|&(_, &n)| n == rows)
.map(|(c, _)| schema.column(c).name().to_string())
.collect();
assert!(
dead.is_empty(),
"{path:?}: {dead:?} are declared and NULL in all {rows} rows — a schema \
promising a fact no producer can deliver (`katana-13`). Fill the column \
or do not declare it."
);
}
}
}
// ── DELTA 1: the rung on the base tables, at the sites `convert` does not reach ──
#[cfg(test)]
mod min_zoom_law {
//! **`min_zoom` on `nodes` / `ways` / `relations`, read back OFF DISK.**
//!
//! `tests/min_zoom_base_tables.rs` covers the sites a real `convert()` runs
//! through — the accumulators, `WayWriter` and `RelWriter`. It cannot reach the
//! rest, because `writer` is `pub(crate)`. These are the rest:
//!
//! | site | who calls it |
//! |---|---|
//! | [`super::write_nodes`] | `lib.rs`, the in-memory convert path |
//! | [`super::write_relations`] → `RelationWriter` | `lib.rs`, and `relgeom` for the resolved rings |
//! | [`super::NodeWriter`] | **nobody** — `RawParquetTypedSink`, itself never constructed |
//!
//! The last one is listed as dead on purpose rather than quietly counted as
//! coverage. It is wired because the schema is shared and an unwired builder set
//! does not compile; it is tested because a retained fallback that is wrong is
//! worse than one that is absent.
use super::*;
use crate::reader::{NodeRecord, RelationRecord};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
/// Tag sets spanning four rungs, with the rung written out as a LITERAL. Not
/// `min_zoom_of_tags(t)` — a table of `f(x) == f(x)` is green on any `f`,
/// including one that returns a constant (LAW 2's identity-value trap).
const CORPUS: &[(&str, u8)] = &[
(r#"{"highway":"motorway"}"#, 6),
(r#"{"highway":"tertiary"}"#, 9),
(r#"{"highway":"residential"}"#, 12),
(r#"{"highway":"footway"}"#, 15),
];
/// `(tags → min_zoom)` from a written parquet. Panics on a missing or null column.
fn rungs_by_tags(path: &Path) -> HashMap<String, u8> {
use arrow::array::{Array, StringArray, UInt8Array};
let rdr = ParquetRecordBatchReaderBuilder::try_new(fs::File::open(path).unwrap())
.unwrap()
.build()
.unwrap();
let mut out = HashMap::new();
for b in rdr {
let b = b.unwrap();
let mz = b
.column_by_name("min_zoom")
.unwrap_or_else(|| panic!("{} has no min_zoom column", path.display()))
.as_any()
.downcast_ref::<UInt8Array>()
.expect("min_zoom is UInt8");
let tg = b
.column_by_name("tags")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
for i in 0..b.num_rows() {
assert!(!mz.is_null(i), "{} row {i}: min_zoom is NULL", path.display());
out.insert(tg.value(i).to_string(), mz.value(i));
}
}
out
}
/// ★ **RED-when-broken: an in-memory-path writer that does not ask the ladder.**
///
/// **RED ARM** (seen): `min_zooms.append_value(6)` in `write_nodes`'s chunk loop —
/// `nodes.parquet: {"highway":"tertiary"} is on disk at rung 6, the ladder says 9`.
#[test]
fn the_in_memory_writers_carry_the_rung_too() {
let d = tempfile::tempdir().unwrap();
let np = d.path().join("nodes.parquet");
let rp = d.path().join("relations.parquet");
let nodes: Vec<NodeRecord> = CORPUS
.iter()
.enumerate()
.map(|(i, (t, _))| NodeRecord {
id: i as i64,
lon_lat: (18.0 + i as f64 * 1e-3, 59.0),
tags_json: (*t).to_string(),
version: Some(1),
})
.collect();
write_nodes(&np, &nodes, "none").unwrap();
let rels: Vec<RelationRecord> = CORPUS
.iter()
.enumerate()
.map(|(i, (t, _))| RelationRecord {
id: i as i64,
tags_json: (*t).to_string(),
members_json: String::from("[]"),
version: Some(1),
})
.collect();
write_relations(&rp, &rels, "none").unwrap();
// The dead-but-retained serial node writer, so a revival cannot revive a hole.
let sp = d.path().join("serial_nodes.parquet");
let mut nw = NodeWriter::new(&sp, "none").unwrap();
for r in &nodes {
nw.push(r).unwrap();
}
nw.finish().unwrap();
let mut checked = 0usize;
for (what, p) in [("nodes", &np), ("relations", &rp), ("NodeWriter", &sp)] {
let got = rungs_by_tags(p);
for (t, want) in CORPUS {
let z = *got
.get(*t)
.unwrap_or_else(|| panic!("{what}: `{t}` is not in the table"));
assert_eq!(
z, *want,
"{what}: `{t}` is on disk at rung {z}, but the ladder puts it on \
{want} — this writer is not asking `min_zoom_of_tags`"
);
checked += 1;
}
}
assert_eq!(checked, CORPUS.len() * 3, "only {checked} cells were compared");
// Non-vacuity: the fixture must span rungs, or a collapsed ladder passes.
let spread: std::collections::BTreeSet<u8> =
rungs_by_tags(&np).into_values().collect();
assert_eq!(
spread.iter().copied().collect::<Vec<u8>>(),
crate::shared::class::RUNGS.to_vec(),
"the fixture landed on {spread:?}, not on every rung — every equality above \
would hold on a ladder collapsed to a constant"
);
}
/// ★ **RED-when-broken: `min_zoom` missing from a base table's schema.**
///
/// The declaration half, checked against the same `min_zoom_field()` the derived
/// tables use — one column, one name, one type, across all six tables that carry
/// it. Deleting the field from any one schema fails here naming that table.
#[test]
fn every_base_table_declares_the_same_min_zoom_column() {
let want = schema::min_zoom_field();
let mut checked = 0usize;
for (what, s) in [
("nodes", schema::nodes_schema()),
("ways", schema::ways_schema(true)),
("ways (no node_refs)", schema::ways_schema(false)),
("relations", schema::relations_schema()),
] {
let f = s
.field_with_name("min_zoom")
.unwrap_or_else(|_| panic!("{what} declares no min_zoom column"));
assert_eq!(f.data_type(), want.data_type(), "{what}: wrong type");
assert!(!f.is_nullable(), "{what}: min_zoom must be non-null");
checked += 1;
}
assert_eq!(checked, 4, "only {checked} schemas were examined");
}
}