From 204ce65fbca67f73ac5f494e3f1de77c46307192 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Nov 2026 15:10:37 +0000 Subject: [PATCH] feat(pipeline): write and publish spider-stats.parquet A new step, compute_spider_stats, scans conflated.parquet once and writes one row per AllThePlaces spider: what became of its features, how they matched (by relation, rule and a distance histogram), how far their tags agree with OpenStreetMap, and per country and Wikidata ID, how its brands compare with OpenStreetMap's totals and deletion candidates. It adds the side files of import_atp and collect_wikidata_ids and AllThePlaces' own run statistics. The file is uploaded and listed in datapackage.json as a debugging aid, and documented in docs/outputs/SPIDER_STATS.md. Part of #924. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ttfb2LhDRExioXkdHbxLno --- docs/TECHNICAL_DESIGN.md | 21 +- docs/outputs/README.md | 7 +- docs/outputs/SPIDER_STATS.md | 166 ++++++ src/pipeline/atp/mod.rs | 4 +- src/pipeline/atp/run_stats.rs | 4 +- src/pipeline/atp/spider_counts.rs | 7 +- src/pipeline/brand_totals.rs | 87 +++ src/pipeline/conflate/evidence.rs | 39 ++ src/pipeline/conflate/mod.rs | 2 +- src/pipeline/mod.rs | 8 + src/pipeline/spider_stats.rs | 856 ++++++++++++++++++++++++++++++ src/pipeline/upload.rs | 21 + tests/integration_test.rs | 88 ++- 13 files changed, 1297 insertions(+), 13 deletions(-) create mode 100644 docs/outputs/SPIDER_STATS.md create mode 100644 src/pipeline/spider_stats.rs diff --git a/docs/TECHNICAL_DESIGN.md b/docs/TECHNICAL_DESIGN.md index ec064d5a..52222ab3 100644 --- a/docs/TECHNICAL_DESIGN.md +++ b/docs/TECHNICAL_DESIGN.md @@ -239,6 +239,14 @@ graph TD JOIN_CONFLATED_TILES --> CONFLATED_PMTILES[conflated.pmtiles] CONFLATED_PMTILES --> UPLOAD_CONFLATED_TILES(upload_conflated_tiles) --> S3C[("public S3")] + CONFLATED --> SPIDER_STATS(compute_spider_stats) + SPIDER_COUNTS --> SPIDER_STATS + ATP_STATS --> SPIDER_STATS + OSM_BRANDS --> SPIDER_STATS + BRAND_NAMES --> SPIDER_STATS + SPIDER_STATS --> SPIDER_STATS_PARQUET[spider-stats.parquet] + SPIDER_STATS_PARQUET --> UPLOAD_SPIDER_STATS(upload_spider_stats) --> S3A + CONFLATED --> SUGGEST_EDITS(suggest_edits) --> LAYERS["*.jsonl (for visualization)"] LAYERS --> RENDER_TILES("render_tiles
(tippecanoe, auto zoom)") --> PMTILES[diffed-places.pmtiles] PMTILES --> UPLOAD_TILES(upload_tiles) --> S3B[("public S3")] @@ -247,10 +255,11 @@ graph TD UPLOAD_CONFLATED --> UPLOAD_DATAPACKAGE UPLOAD_CONFLATED_TILES --> UPLOAD_DATAPACKAGE UPLOAD_TILES --> UPLOAD_DATAPACKAGE + UPLOAD_SPIDER_STATS --> UPLOAD_DATAPACKAGE UPLOAD_BOM --> UPLOAD_DATAPACKAGE(upload_datapackage
data/datapackage.json) --> S3A classDef process fill:#fce4ec,stroke:#ad1457,stroke-width:2px,color:#4a0e28,font-weight:bold; - class FETCH_INPUTS,IMPORT_ATP,OSMVIEWS_RANKS,COLLECT_WIKI,QRANK_QUANTILES,IMPORT_OSM,CONFLATE,UPLOAD_CONFLATED,EXTRACT_CONFLATED_LAYERS,RENDER_CONFLATED_OVERVIEW,RENDER_CONFLATED_DETAIL,JOIN_CONFLATED_TILES,UPLOAD_CONFLATED_TILES,SUGGEST_EDITS,RENDER_TILES,UPLOAD_TILES,UPLOAD_BOM,UPLOAD_DATAPACKAGE process; + class FETCH_INPUTS,IMPORT_ATP,OSMVIEWS_RANKS,COLLECT_WIKI,QRANK_QUANTILES,IMPORT_OSM,CONFLATE,UPLOAD_CONFLATED,EXTRACT_CONFLATED_LAYERS,RENDER_CONFLATED_OVERVIEW,RENDER_CONFLATED_DETAIL,JOIN_CONFLATED_TILES,UPLOAD_CONFLATED_TILES,SPIDER_STATS,UPLOAD_SPIDER_STATS,SUGGEST_EDITS,RENDER_TILES,UPLOAD_TILES,UPLOAD_BOM,UPLOAD_DATAPACKAGE process; ``` (Pink boxes are processing steps; plain rectangles are the files they @@ -520,6 +529,16 @@ fetch away. higher zoom levels” example). **~2m41s**, **2.16 GB peak RSS**, **~1.8 GB output** (z0–16) on the same run — total added cost over a single-pass `conflated.pmtiles` build: **~4m47s**, **+~1.05 GB**. +- **`compute_spider_stats`** + ([`src/pipeline/spider_stats.rs`](../src/pipeline/spider_stats.rs)) + — scans `conflated.parquet` once and writes `spider-stats.parquet`, + one row per AllThePlaces spider: what became of its features, how + they matched, how far their tags agree, and how its brands compare + with OpenStreetMap by country. Adds the counts that `import_atp` + and `collect_wikidata_ids` set aside, and AllThePlaces' own run + statistics. See + [`docs/outputs/SPIDER_STATS.md`](outputs/SPIDER_STATS.md) for the + schema. Uploaded by `upload_spider_stats`, like the files below. - **`suggest_edits`** ([`src/pipeline/edits.rs`](../src/pipeline/edits.rs)) — scans `conflated.parquet` for matched rows and asks an [`edit_suggesters`](../src/edit_suggesters/) implementation what diff --git a/docs/outputs/README.md b/docs/outputs/README.md index 0a99d3e1..84c51b0d 100644 --- a/docs/outputs/README.md +++ b/docs/outputs/README.md @@ -12,6 +12,9 @@ pipeline’s code, not for people using its output.) - [`CONFLATED_TILES.md`](CONFLATED_TILES.md) — `conflated.pmtiles`, visualizing `conflated.parquet`, matched or not, useful for debugging. +- [`SPIDER_STATS.md`](SPIDER_STATS.md) — `spider-stats.parquet`, what + became of each AllThePlaces spider's features, for a dashboard; + useful for debugging. ## Downloading the current output @@ -62,7 +65,9 @@ have `jq` installed there. The `conflated` resource is [`conflated.parquet`](CONFLATED_PARQUET.md); `bom` is its [CycloneDX](CONFLATED_PARQUET.md#data-provenance) -provenance BOM as a standalone file. The two `*-tiles` resources are +provenance BOM as a standalone file. `spider-stats` is +[`spider-stats.parquet`](SPIDER_STATS.md), a debugging aid like the +tiles. The two `*-tiles` resources are PMTiles archives you can open right in the browser via [`pmtiles.io`](https://pmtiles.io) (append `#url=&inspectFeatures=true`) — but they’re a debugging aid, diff --git a/docs/outputs/SPIDER_STATS.md b/docs/outputs/SPIDER_STATS.md new file mode 100644 index 00000000..9afbdf83 --- /dev/null +++ b/docs/outputs/SPIDER_STATS.md @@ -0,0 +1,166 @@ +# `spider-stats.parquet` + +`spider-stats.parquet` says, for each +[AllThePlaces](https://www.alltheplaces.xyz/) spider, what became of +its features: how many it wrote, how many we left out and why, how +they matched OpenStreetMap, how far their tags agree, and how its +brands compare with OpenStreetMap. It is meant for a dashboard that +shows where conflation goes badly and why (see +[brawer/osmdiffs#924](https://github.com/brawer/osmdiffs/issues/924)). + +This file is a debugging aid, not a data product. Its structure may +change, or it may stop being published, without notice. To build +something on top of this project, use +[`conflated.parquet`](CONFLATED_PARQUET.md) instead. + +## Layout + +There is one row per spider, sorted by spider name. Row groups hold 64 +spiders each, so that a web page can fetch one spider with HTTP range +requests, for example with +[hyparquet](https://github.com/hyparam/hyparquet), without +downloading the whole file. + +All numbers are counts, never shares (with one exception, +`opening_hours_top_share`). That way, you can add them up across +spiders, countries or brands before you divide. + +The list includes every spider that AllThePlaces lists in its run +statistics, has a file in its dump, or has a row in `conflated.parquet`. +A spider that AllThePlaces lists but that has no file in the dump has +nulls for the dump's counts. + +## Schema + +``` +spider UTF8 non-null +atp_features UINT64 nullable — features AllThePlaces says the spider wrote +atp_errors UINT64 nullable — errors AllThePlaces says the spider logged +dump_features UINT64 nullable — features in the spider's file of the dump +not_for_osm UINT64 nullable — left out: license doesn't allow OpenStreetMap +invalid UINT64 nullable — left out: no usable geometry, or no spider name +no_matcher UINT64 nullable — kept, but no matcher handles their kind +in_china UINT64 nullable — features in mainland China +datum UTF8 nullable — "wgs84", "gcj02" or "bd09", for in_china +rows UINT64 non-null — rows in conflated.parquet +same UINT64 non-null — rows with match.relation = "same" +located_in UINT64 non-null — rows with match.relation = "located_in" +missing_in_osm UINT64 non-null — rows with match.relation = "missing_in_osm" +shared_osm_features UINT64 non-null — same rows whose OSM feature another feature of the spider also matched +distance_m LIST non-null — histogram of match distances, see below +opening_hours_top_share FLOAT nullable — share of features with the most common opening_hours +rules LIST non-null + rule UTF8 non-null — match.evidence.rule + rows UINT64 non-null +tags LIST non-null + tag UTF8 non-null — "name", "phone", "opening_hours", ... + atp UINT64 non-null — same rows whose ATP feature has the tag + agreeing UINT64 non-null — ... and OSM agrees + differing UINT64 non-null — ... and OSM has a different value + missing_in_osm UINT64 non-null — ... and OSM has none +brands LIST non-null + country UTF8 nullable — ISO 3166-1 alpha-2 code + wikidata_id UTF8 nullable — "Q379911"; null for features without one + name UTF8 nullable — a display name for wikidata_id + atp_features UINT64 non-null + same UINT64 non-null + located_in UINT64 non-null + missing_in_osm UINT64 non-null + osm_features UINT64 nullable — OSM features with this ID in this country + deletion_candidates UINT64 nullable — deletion candidates with this ID in this country +``` + +### What became of the features + +The counts add up like this: + +``` +dump_features = not_for_osm + invalid + no_matcher + rows +rows = same + located_in + missing_in_osm +``` + +`atp_features` and `atp_errors` come from AllThePlaces' own statistics +for the run. `atp_features` normally equals `dump_features`. Where it +doesn't, something went wrong between the spider and the dump. A +spider that crashed often has errors and no features at all. + +`in_china` and `datum` concern features in mainland China. Many +Chinese sources give coordinates in the GCJ-02 or BD-09 datum, which +are hundreds of meters off. We find out which datum a spider uses by +comparing its features with OpenStreetMap, and convert them to WGS-84 +before matching. + +### How they matched + +`rules` counts rows by `match.evidence.rule`; see +[`CONFLATED_PARQUET.md`](CONFLATED_PARQUET.md#match) +for what each rule means. Rows without a rule, such as +`missing_in_osm`, are not listed. + +`distance_m` is a histogram of `match.evidence.distance_m` over the +`same` rows. Its bins are bounded by 1, 10, 25, 50, 100, 250 and 1000 +meters: the first counts distances below 1 m, the second from 1 m up +to 10 m, and the last 1000 m or more. The file's key-value metadata +`osmdiffs:distance_bins_m` lists the bounds, in case they change. A +spider whose matches pile up far away likely has wrong coordinates; +one whose matches are all at exactly the same distance may place every +feature at the center of its town. + +`shared_osm_features` counts `same` rows whose OpenStreetMap feature +also matched another feature of the same spider. A high count hints at +duplicates in the spider, or at OpenStreetMap having one feature where +there are several. + +`opening_hours_top_share` is the share of the spider's features with +`opening_hours` that have its most common value. Near 1 for a large +spider, it suggests that the spider writes the same hours everywhere, +such as those of its head office. + +### How far tags agree + +`tags` compares the tags that `match.evidence.agreeing` compares, for +the `same` rows. `atp` counts rows whose AllThePlaces feature has the +tag; these then split into `agreeing`, `differing` and `missing_in_osm`. +Tags no feature of the spider has are not listed. Phone numbers, email +addresses and websites count under `phone`, `email` and `website` +whichever key they're under, such as `contact:phone`. + +### Brands + +`brands` splits the spider's rows by the country of its features and +by their `brand:wikidata`, `operator:wikidata` and `network:wikidata`. +A feature with several IDs counts for each of them; a feature without +any counts under a null `wikidata_id`. So the brand counts of a spider +can add up to more than its rows. + +`name` is the most common `brand`, `operator` or `network` name for +the ID in OpenStreetMap and AllThePlaces, preferring one in Latin +script; it is there so that readers needn't look up Wikidata IDs. + +`osm_features` counts the OpenStreetMap features in the country with +the same ID, of the same kinds as the spider's features (such as +`shop=supermarket`), whether they matched or not. If a spider has 300 +supermarkets in a country and OpenStreetMap 500, either the spider +misses some or OpenStreetMap has some that closed. +`deletion_candidates` counts the `deletion_candidate` rows with the +same ID in the country. Both are null for features without an ID. + +## Example + +With [DuckDB](https://duckdb.org/), the spiders with the most features +that found no match: + +```sql +SELECT spider, missing_in_osm, rows +FROM 'spider-stats.parquet' +ORDER BY missing_in_osm DESC +LIMIT 10; +``` + +And the brands of one spider: + +```sql +SELECT UNNEST(brands, recursive := true) +FROM 'spider-stats.parquet' +WHERE spider = 'denner_ch'; +``` diff --git a/src/pipeline/atp/mod.rs b/src/pipeline/atp/mod.rs index 5883d0a5..cd07387b 100644 --- a/src/pipeline/atp/mod.rs +++ b/src/pipeline/atp/mod.rs @@ -27,8 +27,8 @@ use crate::{PROGRESS_BAR_STYLE, matchers::MatchMask}; mod datum; mod fetch; pub(super) mod nondescript; -mod run_stats; -mod spider_counts; +pub(super) mod run_stats; +pub(super) mod spider_counts; // Only these two re-exported crate-wide (rather than making all of // `fetch` pub(crate)): provenance needs them to assemble this diff --git a/src/pipeline/atp/run_stats.rs b/src/pipeline/atp/run_stats.rs index 405bf0c9..743008f1 100644 --- a/src/pipeline/atp/run_stats.rs +++ b/src/pipeline/atp/run_stats.rs @@ -10,7 +10,7 @@ use std::path::Path; /// What AllThePlaces reports for one spider. #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq)] -pub(super) struct SpiderRunStats { +pub(in crate::pipeline) struct SpiderRunStats { pub spider: String, /// Features the spider wrote. @@ -35,7 +35,7 @@ pub(super) fn parse(json: &[u8]) -> Result> { /// Reads the statistics that `fetch_atp` stored in `workdir`, by /// spider, or `None` if there are none. -pub(super) fn read(workdir: &Path) -> Result>> { +pub(in crate::pipeline) fn read(workdir: &Path) -> Result>> { let path = workdir.join(super::fetch::STATS_JSON_FILENAME); if !path.exists() { return Ok(None); diff --git a/src/pipeline/atp/spider_counts.rs b/src/pipeline/atp/spider_counts.rs index 39acd301..2e247c7a 100644 --- a/src/pipeline/atp/spider_counts.rs +++ b/src/pipeline/atp/spider_counts.rs @@ -16,12 +16,12 @@ const FILENAME: &str = "alltheplaces.spider-counts.tsv"; const HEADER: &str = "spider\tfeatures\tnot_for_osm\tinvalid\tin_china\tdatum"; /// Path of the counts that `import_atp` writes in `workdir`. -pub(super) fn path(workdir: &Path) -> PathBuf { +pub(in crate::pipeline) fn path(workdir: &Path) -> PathBuf { workdir.join(FILENAME) } #[derive(Clone, Debug, Default, Eq, PartialEq)] -pub(super) struct SpiderCounts { +pub(in crate::pipeline) struct SpiderCounts { pub spider: String, /// Features in the spider's file of the dump. A spider that crashed @@ -75,8 +75,7 @@ pub(super) fn write(counts: &mut [SpiderCounts], out: &Path) -> Result<()> { } /// Reads counts that [write] wrote. -#[allow(dead_code)] // Used by tests until the spider stats read it. -pub(super) fn read(path: &Path) -> Result> { +pub(in crate::pipeline) fn read(path: &Path) -> Result> { let file = std::fs::File::open(path).with_context(|| format!("could not open {path:?}"))?; let mut lines = BufReader::new(file).lines(); if lines.next().transpose()?.as_deref() != Some(HEADER) { diff --git a/src/pipeline/brand_totals.rs b/src/pipeline/brand_totals.rs index dc650bd1..58a41f7c 100644 --- a/src/pipeline/brand_totals.rs +++ b/src/pipeline/brand_totals.rs @@ -191,6 +191,72 @@ fn clean_name(name: &str) -> String { .collect() } +/// OSM features per Wikidata ID, country and main category. +pub(super) type OsmFeatureCounts = HashMap<(u64, Option, String), u64>; + +/// Reads [features_path]. +pub(super) fn read_features(workdir: &Path) -> Result { + let mut features = HashMap::new(); + for cols in read_tsv(&features_path(workdir), 4)? { + let n = cols[3] + .parse() + .with_context(|| format!("bad count {:?}", cols[3]))?; + features.insert( + ( + parse_id(&cols[0])?, + parse_country(&cols[1]), + cols[2].clone(), + ), + n, + ); + } + Ok(features) +} + +/// Reads [names_path] into one name per Wikidata ID: its `brand` name, +/// or else its `operator` name, or else its `network` name. +pub(super) fn read_names(workdir: &Path) -> Result> { + let rank = |key: &str| { + ["brand", "operator", "network"] + .iter() + .position(|k| *k == key) + }; + let mut best: HashMap = HashMap::new(); + for cols in read_tsv(&names_path(workdir), 5)? { + let Some(r) = rank(&cols[1]) else { continue }; + let id = parse_id(&cols[0])?; + if best.get(&id).is_none_or(|(old, _)| r < *old) { + best.insert(id, (r, cols[2].clone())); + } + } + Ok(best.into_iter().map(|(id, (_, name))| (id, name)).collect()) +} + +fn parse_id(id: &str) -> Result { + id.strip_prefix('Q') + .and_then(|n| n.parse().ok()) + .with_context(|| format!("bad Wikidata ID {id:?}")) +} + +fn parse_country(code: &str) -> Option { + <[u8; 2]>::try_from(code.as_bytes()) + .ok() + .and_then(Country::from_code) +} + +/// The lines of a TSV file after its header, split into `columns`. +fn read_tsv(path: &Path, columns: usize) -> Result>> { + let data = std::fs::read_to_string(path).with_context(|| format!("could not read {path:?}"))?; + data.lines() + .skip(1) + .map(|line| { + let cols: Vec = line.split('\t').map(String::from).collect(); + anyhow::ensure!(cols.len() == columns, "bad line in {path:?}: {line:?}"); + Ok(cols) + }) + .collect() +} + fn write_tsv(out: &Path, header: &str, lines: impl Iterator) -> Result<()> { let mut tmp = out.to_path_buf(); tmp.add_extension("tmp"); @@ -275,6 +341,27 @@ mod tests { Ok(()) } + #[test] + fn test_read() -> Result<()> { + let mut totals = BrandTotals::default(); + let tags = [ + ("shop", "supermarket"), + ("brand", "Migros"), + ("brand:wikidata", "Q680727"), + ("operator", "Migros Zürich"), + ("operator:wikidata", "Q680727"), + ]; + totals.add(Source::Osm, tags.iter().copied(), CH); + totals.add(Source::Osm, tags.iter().copied(), None); + let dir = tempfile::tempdir()?; + totals.write(dir.path())?; + let features = read_features(dir.path())?; + assert_eq!(features[&(680727, CH, "shop=supermarket".to_string())], 1); + assert_eq!(features[&(680727, None, "shop=supermarket".to_string())], 1); + assert_eq!(read_names(dir.path())?[&680727], "Migros"); + Ok(()) + } + #[test] fn test_display_name() { let shown = |list: &[(&str, u64)]| { diff --git a/src/pipeline/conflate/evidence.rs b/src/pipeline/conflate/evidence.rs index 31d2b9fa..096cdcd4 100644 --- a/src/pipeline/conflate/evidence.rs +++ b/src/pipeline/conflate/evidence.rs @@ -116,6 +116,30 @@ const AGREEING: &[(&str, Check)] = &[ ("uic_ref", Check::Text(&["uic_ref"], text)), ]; +/// The names of the tags that [agreeing] compares, in its order. +pub(in crate::pipeline) fn compared_tags() -> impl Iterator { + AGREEING.iter().map(|(name, _)| *name) +} + +/// Whether `tags` have a value for the tag that [agreeing] calls +/// `name`, such as any phone key for `phone`. +pub(in crate::pipeline) fn has_compared_tag<'a>( + name: &str, + mut tags: impl Iterator, +) -> bool { + let Some((_, check)) = AGREEING.iter().find(|(n, _)| *n == name) else { + return false; + }; + tags.any(|(k, v)| { + !v.trim().is_empty() + && match check { + Check::Text(keys, _) => keys.contains(&k), + Check::Whole(key, _) => k == *key, + Check::Contact(is_key) => is_key(k), + } + }) +} + fn text(value: &str) -> Option { let v = normalize_name(value); (!v.is_empty()).then_some(v) @@ -129,6 +153,21 @@ mod tests { Geometry::from(geo::Point::new(x, y)) } + #[test] + fn test_has_compared_tag() { + let tags = [ + ("contact:phone", "+41 44 000 00 00"), + ("name", " "), + ("brand", "X"), + ]; + assert!(has_compared_tag("phone", tags.iter().copied())); + assert!(has_compared_tag("brand", tags.iter().copied())); + assert!(!has_compared_tag("name", tags.iter().copied())); + assert!(!has_compared_tag("email", tags.iter().copied())); + assert!(!has_compared_tag("unknown", tags.iter().copied())); + assert_eq!(compared_tags().next(), Some("name")); + } + #[test] fn test_distance_meters() { // 0.001° of latitude is about 111 m. diff --git a/src/pipeline/conflate/mod.rs b/src/pipeline/conflate/mod.rs index 257a32fe..b1232af5 100644 --- a/src/pipeline/conflate/mod.rs +++ b/src/pipeline/conflate/mod.rs @@ -38,7 +38,7 @@ use std::{ const PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(30); mod deletion; -mod evidence; +pub(super) mod evidence; mod local_coverage; mod located_in; mod s2_util; diff --git a/src/pipeline/mod.rs b/src/pipeline/mod.rs index 35eec22a..75eb84f2 100644 --- a/src/pipeline/mod.rs +++ b/src/pipeline/mod.rs @@ -45,6 +45,7 @@ mod osm_views; mod packaged; mod provenance; mod qrank; +mod spider_stats; // Only these re-exported at the pipeline level (rather than making all // of `atp`/`osm` pub(crate)): `provenance` needs them to assemble this @@ -345,6 +346,13 @@ fn run_pipeline_steps( upload::upload_conflated_tiles(&conflated_tiles_out, &date, progress) })?); + let spider_stats = run_step("compute_spider_stats", || { + spider_stats::compute_spider_stats(&conflated, progress, workdir) + })?; + published.push(run_step("upload_spider_stats", || { + upload::upload_spider_stats(&spider_stats, &date, progress) + })?); + let edits = run_step("suggest_edits", || { edits::suggest_edits(&conflated, progress, workdir) })?; diff --git a/src/pipeline/spider_stats.rs b/src/pipeline/spider_stats.rs new file mode 100644 index 00000000..b988fb65 --- /dev/null +++ b/src/pipeline/spider_stats.rs @@ -0,0 +1,856 @@ +//! Writes `spider-stats.parquet`: for each AllThePlaces spider, what +//! became of its features, how they matched OpenStreetMap, and how far +//! their tags agree, so that a dashboard can show where conflation goes +//! badly and why; see brawer/osmdiffs#924 and +//! [docs/outputs/SPIDER_STATS.md](../../docs/outputs/SPIDER_STATS.md). +//! +//! One row per spider, sorted by spider, in small row groups, so that a +//! web page can fetch one spider's row with HTTP range requests. All +//! numbers are counts, so that a reader can add them up across spiders, +//! countries or brands before dividing. +//! +//! Reads `conflated.parquet` once, and the side files that earlier +//! steps wrote: `alltheplaces.spider-counts.tsv` and +//! `alltheplaces.stats.json` from `import_atp`, and `osm-brands.tsv` and +//! `brand-names.tsv` from `collect_wikidata_ids`. + +use super::atp::{run_stats, spider_counts}; +use super::brand_totals; +use super::conflate::evidence::{compared_tags, has_compared_tag}; +use super::wikidata_ids::WIKIDATA_KEYS; +use crate::make_progress_bar; +use crate::matchers::{contact::main_category, parse_wikidata_ids}; +use crate::utils::countries::{Country, country}; +use crate::utils::parquet::{get_binary, get_string, get_struct, get_tags, get_u64}; +use anyhow::{Context, Result}; +use arrow::array::{ + Array, ArrayRef, Float32Array, Float32Builder, ListArray, ListBuilder, RecordBatch, + StringArray, StringBuilder, StructBuilder, UInt64Builder, +}; +use arrow::datatypes::{DataType, Field, Fields, Schema}; +use geo::Centroid; +use geo_traits::to_geo::ToGeoGeometry; +use indicatif::MultiProgress; +use parquet::arrow::ArrowWriter; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::basic::{Compression, ZstdLevel}; +use parquet::file::metadata::KeyValue; +use parquet::file::properties::WriterProperties; +use rayon::prelude::*; +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::fs::File; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use wkb::reader::read_wkb; + +/// Filename, within `workdir`, of the file [compute_spider_stats] +/// writes. +const FILENAME: &str = "spider-stats.parquet"; + +/// Spiders per row group: few, so that a web page fetching one spider +/// with HTTP range requests reads little more than that spider. +const ROWS_PER_GROUP: usize = 64; + +/// Upper bounds, in meters, of the bins of the `distance_m` histogram; +/// the last bin has no upper bound. +pub(super) const DISTANCE_BINS_M: [f32; 7] = [1.0, 10.0, 25.0, 50.0, 100.0, 250.0, 1000.0]; + +/// Computes the stats of every spider from `conflated` and the side +/// files in `workdir`, and writes them to `workdir/spider-stats.parquet`, +/// unless that is already newer than `conflated`. +pub fn compute_spider_stats( + conflated: &Path, + progress: &MultiProgress, + workdir: &Path, +) -> Result { + let out = workdir.join(FILENAME); + if out.exists() && super::modified(&out)? >= super::modified(conflated)? { + return Ok(out); + } + + let counts: HashMap = + spider_counts::read(&spider_counts::path(workdir))? + .into_iter() + .map(|c| (c.spider.clone(), c)) + .collect(); + let run_stats = run_stats::read(workdir)?.unwrap_or_default(); + let osm_brands = brand_totals::read_features(workdir)?; + let names = brand_totals::read_names(workdir)?; + + let stats = scan_conflated(conflated, progress)?; + + let mut spiders: BTreeSet<&str> = counts.keys().map(String::as_str).collect(); + spiders.extend(run_stats.keys().map(String::as_str)); + spiders.extend(stats.spiders.keys().map(String::as_str)); + let empty = Spider::default(); + let rows: Vec = spiders + .into_iter() + .map(|spider| Row { + spider, + counts: counts.get(spider), + run_stats: run_stats.get(spider), + stats: stats.spiders.get(spider).unwrap_or(&empty), + }) + .collect(); + let lookups = Lookups { + osm_brands: &osm_brands, + names: &names, + deletions: &stats.deletions, + }; + write(&rows, &lookups, &out)?; + log::info!( + spiders = rows.len(), + conflated_rows = stats.rows, + deletion_candidates = stats.deletions.values().sum::(), + bytes = std::fs::metadata(&out)?.len(); + "compute_spider_stats: done" + ); + Ok(out) +} + +/// What one row of `conflated.parquet` contributes to the stats. +#[derive(Debug, Default)] +struct RowInfo { + /// `None` for a deletion candidate, which has no ATP side. + spider: Option, + relation: Relation, + rule: Option, + distance_m: Option, + /// Of the ATP feature, or for a deletion candidate, of the OSM one. + country: Option, + /// The brand, operator and network IDs of the ATP feature, or for a + /// deletion candidate, of the OSM one. + wikidata: Vec, + category: Option, + /// For each tag of [compared_tags] that the ATP feature has, its + /// position there and how it compares with OSM. + tags: Vec<(usize, TagState)>, + opening_hours: Option, + /// The OSM feature of a `same` row, as `id * 4 + type`. + osm_ref: Option, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Relation { + #[default] + Same, + LocatedIn, + MissingInOsm, + DeletionCandidate, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TagState { + Agreeing, + Differing, + MissingInOsm, +} + +/// What we count for one brand, operator or network of a spider in one +/// country. +#[derive(Debug, Default)] +struct Brand { + atp_features: u64, + same: u64, + located_in: u64, + missing_in_osm: u64, + /// The main categories of the ATP features, so that the OSM total + /// counts features of the same kinds only. + categories: HashSet, +} + +/// What we count for one spider from `conflated.parquet`. +#[derive(Debug, Default)] +struct Spider { + rows: u64, + same: u64, + located_in: u64, + missing_in_osm: u64, + rules: HashMap, + distance_m: [u64; DISTANCE_BINS_M.len() + 1], + /// Per tag of [compared_tags]: ATP features with the tag, and how + /// many of them agree, differ, or have no such tag in OSM. + tags: Vec<[u64; 4]>, + opening_hours: HashMap, + osm_refs: HashMap, + /// By country and Wikidata ID; 0 for features without one. + brands: HashMap<(Option, u64), Brand>, +} + +impl Spider { + fn add(&mut self, row: RowInfo) { + self.rows += 1; + match row.relation { + Relation::Same => self.same += 1, + Relation::LocatedIn => self.located_in += 1, + Relation::MissingInOsm => self.missing_in_osm += 1, + Relation::DeletionCandidate => {} + } + if let Some(rule) = row.rule { + *self.rules.entry(rule).or_default() += 1; + } + if row.relation == Relation::Same + && let Some(d) = row.distance_m + { + let bin = DISTANCE_BINS_M.partition_point(|&upper| upper <= d); + self.distance_m[bin] += 1; + } + if self.tags.is_empty() { + self.tags = vec![[0; 4]; compared_tags().count()]; + } + for (i, state) in row.tags { + let t = &mut self.tags[i]; + t[0] += 1; + match state { + TagState::Agreeing => t[1] += 1, + TagState::Differing => t[2] += 1, + TagState::MissingInOsm => t[3] += 1, + } + } + if let Some(oh) = row.opening_hours { + *self.opening_hours.entry(oh).or_default() += 1; + } + if let Some(r) = row.osm_ref { + *self.osm_refs.entry(r).or_default() += 1; + } + let ids = if row.wikidata.is_empty() { + vec![0] + } else { + row.wikidata + }; + for id in ids { + let brand = self.brands.entry((row.country, id)).or_default(); + brand.atp_features += 1; + match row.relation { + Relation::Same => brand.same += 1, + Relation::LocatedIn => brand.located_in += 1, + Relation::MissingInOsm => brand.missing_in_osm += 1, + Relation::DeletionCandidate => {} + } + if let Some(c) = &row.category + && !brand.categories.contains(c) + { + brand.categories.insert(c.clone()); + } + } + } + + /// ATP features whose OSM feature also matched another feature of + /// this spider. + fn shared_osm_features(&self) -> u64 { + self.osm_refs.values().filter(|&&n| n > 1).sum() + } + + /// The share of ATP features with `opening_hours` that have the + /// most common value, or `None` if none have any. + fn opening_hours_top_share(&self) -> Option { + let total: u64 = self.opening_hours.values().sum(); + let top = self.opening_hours.values().max()?; + Some(*top as f32 / total as f32) + } +} + +#[derive(Debug, Default)] +struct Scanned { + rows: u64, + spiders: HashMap, + /// Deletion candidates by country and Wikidata ID. + deletions: HashMap<(Option, u64), u64>, +} + +impl Scanned { + fn add(&mut self, row: RowInfo) { + self.rows += 1; + if row.relation == Relation::DeletionCandidate { + for &id in &row.wikidata { + *self.deletions.entry((row.country, id)).or_default() += 1; + } + return; + } + let Some(spider) = &row.spider else { return }; + if !self.spiders.contains_key(spider) { + self.spiders.insert(spider.clone(), Spider::default()); + } + self.spiders.get_mut(spider).expect("inserted").add(row); + } +} + +fn scan_conflated(conflated: &Path, progress: &MultiProgress) -> Result { + let builder = ParquetRecordBatchReaderBuilder::try_new(File::open(conflated)?)?; + let num_rows = builder.metadata().file_metadata().num_rows() as u64; + let bar = make_progress_bar(progress, "spider_stats", num_rows, "conflated rows"); + let mut scanned = Scanned::default(); + for batch in builder.build()? { + let batch = batch?; + let rows: Vec = (0..batch.num_rows()) + .into_par_iter() + .map(|i| row_info(&batch, i)) + .collect::>()?; + bar.inc(rows.len() as u64); + for row in rows { + scanned.add(row); + } + } + bar.finish(); + Ok(scanned) +} + +fn row_info(batch: &RecordBatch, row: usize) -> Result { + let m = get_struct(batch, "match")?; + let relation = match get_string(m, "relation", row)?.as_str() { + "same" => Relation::Same, + "located_in" => Relation::LocatedIn, + "missing_in_osm" => Relation::MissingInOsm, + "deletion_candidate" => Relation::DeletionCandidate, + // A relation added later; see CONFLATED_PARQUET.md. + _ => return Ok(RowInfo::default()), + }; + let evidence = get_struct(m, "evidence")?; + let rule = string_or_null(evidence, "rule", row)?; + let distance_m = f32_or_null(evidence, "distance_m", row)?; + let agreeing = string_list(evidence, "agreeing", row)?; + + let atp = get_struct(batch, "atp")?; + let osm = get_struct(batch, "osm")?; + let osm_tags = if osm.is_null(row) { + Vec::new() + } else { + get_tags(osm, "tags", row)? + }; + let osm_tag_refs = || osm_tags.iter().map(|(k, v)| (k.as_str(), v.as_str())); + + if relation == Relation::DeletionCandidate { + let geometry = get_binary(batch, "osm_geometry", row)?; + return Ok(RowInfo { + relation, + country: wkb_country(&geometry), + wikidata: wikidata_ids(osm_tag_refs()), + ..Default::default() + }); + } + if atp.is_null(row) { + return Ok(RowInfo::default()); + } + let atp_tags = get_tags(atp, "tags", row)?; + let atp_tag_refs = || atp_tags.iter().map(|(k, v)| (k.as_str(), v.as_str())); + let spider = get_string(get_struct(atp, "fetched")?, "spider", row)?; + let geometry = get_binary(batch, "atp_geometry", row)?; + + let mut tags = Vec::new(); + if relation == Relation::Same { + for (i, name) in compared_tags().enumerate() { + if !has_compared_tag(name, atp_tag_refs()) { + continue; + } + let state = if agreeing.iter().any(|a| a == name) { + TagState::Agreeing + } else if has_compared_tag(name, osm_tag_refs()) { + TagState::Differing + } else { + TagState::MissingInOsm + }; + tags.push((i, state)); + } + } + let osm_ref = if relation == Relation::Same { + let kind = match get_string(osm, "type", row)?.as_str() { + "node" => 1, + "way" => 2, + _ => 3, + }; + Some(get_u64(osm, "id", row)? * 4 + kind) + } else { + None + }; + Ok(RowInfo { + spider: Some(spider), + relation, + rule, + distance_m, + country: wkb_country(&geometry), + wikidata: wikidata_ids(atp_tag_refs()), + category: main_category(atp_tag_refs()), + tags, + opening_hours: atp_tag_refs() + .find(|(k, _)| *k == "opening_hours") + .map(|(_, v)| v.to_string()), + osm_ref, + }) +} + +/// The distinct brand, operator and network Wikidata IDs in `tags`. +fn wikidata_ids<'a>(tags: impl Iterator) -> Vec { + let mut ids: Vec = tags + .filter(|(k, _)| WIKIDATA_KEYS.contains(k)) + .flat_map(|(_, v)| parse_wikidata_ids(v)) + .collect(); + ids.sort_unstable(); + ids.dedup(); + ids +} + +/// The country at the centroid of a WKB geometry. +fn wkb_country(wkb: &[u8]) -> Option { + let c = read_wkb(wkb).ok()?.to_geometry().centroid()?; + country(c.x(), c.y()) +} + +fn string_or_null( + src: &arrow::array::StructArray, + name: &str, + row: usize, +) -> Result> { + let col = src + .column_by_name(name) + .with_context(|| format!("missing column '{name}'"))? + .as_any() + .downcast_ref::() + .with_context(|| format!("column '{name}' is not a string"))?; + Ok((!col.is_null(row)).then(|| col.value(row).to_string())) +} + +fn f32_or_null(src: &arrow::array::StructArray, name: &str, row: usize) -> Result> { + let col = src + .column_by_name(name) + .with_context(|| format!("missing column '{name}'"))? + .as_any() + .downcast_ref::() + .with_context(|| format!("column '{name}' is not Float32"))?; + Ok((!col.is_null(row)).then(|| col.value(row))) +} + +fn string_list(src: &arrow::array::StructArray, name: &str, row: usize) -> Result> { + let col = src + .column_by_name(name) + .with_context(|| format!("missing column '{name}'"))? + .as_any() + .downcast_ref::() + .with_context(|| format!("column '{name}' is not a list"))?; + let values = col.value(row); + let values = values + .as_any() + .downcast_ref::() + .with_context(|| format!("column '{name}' is not a list of strings"))?; + Ok((0..values.len()) + .map(|i| values.value(i).to_string()) + .collect()) +} + +/// One spider's row, from everything we know about it. +struct Row<'a> { + spider: &'a str, + counts: Option<&'a spider_counts::SpiderCounts>, + run_stats: Option<&'a run_stats::SpiderRunStats>, + stats: &'a Spider, +} + +/// What a spider's brands are compared with. +struct Lookups<'a> { + osm_brands: &'a brand_totals::OsmFeatureCounts, + names: &'a HashMap, + deletions: &'a HashMap<(Option, u64), u64>, +} + +fn u64_field(name: &str) -> Field { + Field::new(name, DataType::UInt64, false) +} + +fn brand_fields() -> Fields { + Fields::from(vec![ + Field::new("country", DataType::Utf8, true), + Field::new("wikidata_id", DataType::Utf8, true), + Field::new("name", DataType::Utf8, true), + u64_field("atp_features"), + u64_field("same"), + u64_field("located_in"), + u64_field("missing_in_osm"), + Field::new("osm_features", DataType::UInt64, true), + Field::new("deletion_candidates", DataType::UInt64, true), + ]) +} + +fn tag_fields() -> Fields { + Fields::from(vec![ + Field::new("tag", DataType::Utf8, false), + u64_field("atp"), + u64_field("agreeing"), + u64_field("differing"), + u64_field("missing_in_osm"), + ]) +} + +fn rule_fields() -> Fields { + Fields::from(vec![ + Field::new("rule", DataType::Utf8, false), + u64_field("rows"), + ]) +} + +fn list_of(fields: Fields) -> DataType { + DataType::List(Arc::new(Field::new( + "item", + DataType::Struct(fields), + false, + ))) +} + +pub(super) fn schema() -> Schema { + Schema::new(vec![ + Field::new("spider", DataType::Utf8, false), + Field::new("atp_features", DataType::UInt64, true), + Field::new("atp_errors", DataType::UInt64, true), + Field::new("dump_features", DataType::UInt64, true), + Field::new("not_for_osm", DataType::UInt64, true), + Field::new("invalid", DataType::UInt64, true), + Field::new("no_matcher", DataType::UInt64, true), + Field::new("in_china", DataType::UInt64, true), + Field::new("datum", DataType::Utf8, true), + u64_field("rows"), + u64_field("same"), + u64_field("located_in"), + u64_field("missing_in_osm"), + u64_field("shared_osm_features"), + Field::new( + "distance_m", + DataType::List(Arc::new(Field::new("item", DataType::UInt64, false))), + false, + ), + Field::new("opening_hours_top_share", DataType::Float32, true), + Field::new("rules", list_of(rule_fields()), false), + Field::new("tags", list_of(tag_fields()), false), + Field::new("brands", list_of(brand_fields()), false), + ]) +} + +fn write(rows: &[Row], lookups: &Lookups, out: &Path) -> Result<()> { + let schema = Arc::new(schema()); + let bins = DISTANCE_BINS_M + .iter() + .map(|b| b.to_string()) + .collect::>() + .join(","); + let properties = WriterProperties::builder() + .set_compression(Compression::ZSTD(ZstdLevel::try_new(9)?)) + .set_max_row_group_row_count(Some(ROWS_PER_GROUP)) + .set_key_value_metadata(Some(vec![KeyValue::new( + "osmdiffs:distance_bins_m".to_string(), + bins, + )])) + .build(); + let mut tmp = out.to_path_buf(); + tmp.add_extension("tmp"); + let mut writer = ArrowWriter::try_new(File::create(&tmp)?, schema.clone(), Some(properties))?; + for chunk in rows.chunks(ROWS_PER_GROUP) { + writer.write(&batch(chunk, lookups, &schema)?)?; + writer.flush()?; + } + writer.close()?; + std::fs::rename(&tmp, out)?; + Ok(()) +} + +fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result { + let tag_names: Vec<&str> = compared_tags().collect(); + let mut spider = StringBuilder::new(); + let mut atp_features = UInt64Builder::new(); + let mut atp_errors = UInt64Builder::new(); + let mut dump_features = UInt64Builder::new(); + let mut not_for_osm = UInt64Builder::new(); + let mut invalid = UInt64Builder::new(); + let mut no_matcher = UInt64Builder::new(); + let mut in_china = UInt64Builder::new(); + let mut datum = StringBuilder::new(); + let mut num_rows = UInt64Builder::new(); + let mut same = UInt64Builder::new(); + let mut located_in = UInt64Builder::new(); + let mut missing_in_osm = UInt64Builder::new(); + let mut shared = UInt64Builder::new(); + let mut distance = ListBuilder::new(UInt64Builder::new()).with_field(Arc::new(Field::new( + "item", + DataType::UInt64, + false, + ))); + let mut oh_share = Float32Builder::new(); + let struct_list = |fields: Fields| { + ListBuilder::new(StructBuilder::from_fields(fields.clone(), 0)).with_field(Arc::new( + Field::new("item", DataType::Struct(fields), false), + )) + }; + let mut rules = struct_list(rule_fields()); + let mut tags = struct_list(tag_fields()); + let mut brands = struct_list(brand_fields()); + + for row in rows { + let s = row.stats; + spider.append_value(row.spider); + atp_features.append_option(row.run_stats.map(|r| r.features)); + atp_errors.append_option(row.run_stats.map(|r| r.errors)); + dump_features.append_option(row.counts.map(|c| c.features)); + not_for_osm.append_option(row.counts.map(|c| c.not_for_osm)); + invalid.append_option(row.counts.map(|c| c.invalid)); + // Every feature that a matcher handles gets exactly one row. + no_matcher.append_option(row.counts.map(|c| c.written().saturating_sub(s.rows))); + in_china.append_option(row.counts.map(|c| c.in_china)); + datum.append_option( + row.counts + .map(|c| c.datum.as_str()) + .filter(|d| !d.is_empty()), + ); + num_rows.append_value(s.rows); + same.append_value(s.same); + located_in.append_value(s.located_in); + missing_in_osm.append_value(s.missing_in_osm); + shared.append_value(s.shared_osm_features()); + distance.values().append_slice(&s.distance_m); + distance.append(true); + oh_share.append_option(s.opening_hours_top_share()); + + let mut rule_counts: Vec<(&String, &u64)> = s.rules.iter().collect(); + rule_counts.sort(); + let b = rules.values(); + for (rule, n) in rule_counts { + b.field_builder::(0) + .unwrap() + .append_value(rule); + b.field_builder::(1) + .unwrap() + .append_value(*n); + b.append(true); + } + rules.append(true); + + let b = tags.values(); + for (i, t) in s.tags.iter().enumerate() { + if t[0] == 0 { + continue; + } + b.field_builder::(0) + .unwrap() + .append_value(tag_names[i]); + for (j, n) in t.iter().enumerate() { + b.field_builder::(j + 1) + .unwrap() + .append_value(*n); + } + b.append(true); + } + tags.append(true); + + let mut keys: Vec<&(Option, u64)> = s.brands.keys().collect(); + keys.sort_by_key(|(c, id)| (c.map(|c| c.code_bytes()), *id)); + let b = brands.values(); + for key in keys { + let (country, id) = *key; + let brand = &s.brands[key]; + b.field_builder::(0) + .unwrap() + .append_option(country.map(|c| c.code().to_string())); + b.field_builder::(1) + .unwrap() + .append_option((id != 0).then(|| format!("Q{id}"))); + b.field_builder::(2) + .unwrap() + .append_option(lookups.names.get(&id)); + let counts = [ + brand.atp_features, + brand.same, + brand.located_in, + brand.missing_in_osm, + ]; + for (j, n) in counts.iter().enumerate() { + b.field_builder::(3 + j) + .unwrap() + .append_value(*n); + } + // Only for a Wikidata ID: features without one have nothing + // to compare with. + let osm = (id != 0).then(|| { + brand + .categories + .iter() + .map(|c| { + lookups + .osm_brands + .get(&(id, country, c.clone())) + .copied() + .unwrap_or(0) + }) + .sum::() + }); + b.field_builder::(7) + .unwrap() + .append_option(osm); + let deletions = + (id != 0).then(|| lookups.deletions.get(&(country, id)).copied().unwrap_or(0)); + b.field_builder::(8) + .unwrap() + .append_option(deletions); + b.append(true); + } + brands.append(true); + } + + let columns: Vec = vec![ + Arc::new(spider.finish()), + Arc::new(atp_features.finish()), + Arc::new(atp_errors.finish()), + Arc::new(dump_features.finish()), + Arc::new(not_for_osm.finish()), + Arc::new(invalid.finish()), + Arc::new(no_matcher.finish()), + Arc::new(in_china.finish()), + Arc::new(datum.finish()), + Arc::new(num_rows.finish()), + Arc::new(same.finish()), + Arc::new(located_in.finish()), + Arc::new(missing_in_osm.finish()), + Arc::new(shared.finish()), + Arc::new(distance.finish()), + Arc::new(oh_share.finish()), + Arc::new(rules.finish()), + Arc::new(tags.finish()), + Arc::new(brands.finish()), + ]; + Ok(RecordBatch::try_new(schema.clone(), columns)?) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{StructArray, UInt64Array}; + + fn same(spider: &str, distance_m: f32, osm_ref: u64) -> RowInfo { + RowInfo { + spider: Some(spider.into()), + relation: Relation::Same, + rule: Some("brand_wikidata".into()), + distance_m: Some(distance_m), + country: Country::from_code(*b"CH"), + wikidata: vec![379911], + category: Some("shop=supermarket".into()), + tags: vec![(0, TagState::Agreeing), (1, TagState::Differing)], + opening_hours: Some("Mo-Sa 08:00-20:00".into()), + osm_ref: Some(osm_ref), + } + } + + #[test] + fn test_spider_add() { + let mut s = Spider::default(); + // A distance on a bin's upper bound goes into the next bin. + for (d, osm) in [(0.0, 1), (1.0, 2), (24.9, 3), (5000.0, 3)] { + s.add(same("x", d, osm)); + } + s.add(RowInfo { + spider: Some("x".into()), + relation: Relation::MissingInOsm, + opening_hours: Some("24/7".into()), + ..Default::default() + }); + assert_eq!((s.rows, s.same, s.missing_in_osm), (5, 4, 1)); + assert_eq!(s.distance_m, [1, 1, 1, 0, 0, 0, 0, 1]); + assert_eq!(s.rules["brand_wikidata"], 4); + assert_eq!(s.tags[0], [4, 4, 0, 0]); + assert_eq!(s.tags[1], [4, 0, 4, 0]); + assert_eq!(s.shared_osm_features(), 2); + assert_eq!(s.opening_hours_top_share(), Some(0.8)); + let ch = Country::from_code(*b"CH"); + assert_eq!(s.brands[&(ch, 379911)].same, 4); + assert_eq!(s.brands[&(None, 0)].missing_in_osm, 1); + assert_eq!(Spider::default().opening_hours_top_share(), None); + } + + #[test] + fn test_scanned_add() { + let mut scanned = Scanned::default(); + scanned.add(same("a", 3.0, 1)); + scanned.add(RowInfo { + relation: Relation::DeletionCandidate, + country: Country::from_code(*b"CH"), + wikidata: vec![1, 2], + ..Default::default() + }); + assert_eq!(scanned.rows, 2); + assert_eq!(scanned.spiders["a"].rows, 1); + assert_eq!(scanned.deletions.len(), 2); + } + + #[test] + fn test_write() -> Result<()> { + let dir = tempfile::tempdir()?; + let out = dir.path().join(FILENAME); + let ch = Country::from_code(*b"CH"); + let mut stats = Spider::default(); + stats.add(same("denner", 3.0, 1)); + let counts = spider_counts::SpiderCounts { + spider: "denner".into(), + features: 3, + invalid: 1, + ..Default::default() + }; + let empty = Spider::default(); + let rows = [ + Row { + spider: "denner", + counts: Some(&counts), + run_stats: None, + stats: &stats, + }, + Row { + spider: "unknown", + counts: None, + run_stats: None, + stats: &empty, + }, + ]; + let osm_brands = HashMap::from([ + ((379911, ch, "shop=supermarket".to_string()), 7), + ((379911, ch, "amenity=cafe".to_string()), 100), + ]); + let names = HashMap::from([(379911, "Denner".to_string())]); + let deletions = HashMap::from([((ch, 379911), 2)]); + let lookups = Lookups { + osm_brands: &osm_brands, + names: &names, + deletions: &deletions, + }; + write(&rows, &lookups, &out)?; + + let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(&out)?)?.build()?; + let batch = reader.into_iter().next().context("no batch")??; + assert_eq!(batch.schema().as_ref(), &schema()); + assert_eq!(batch.num_rows(), 2); + let u64s = |name: &str| { + batch + .column_by_name(name) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .clone() + }; + // 3 features, 1 invalid, 1 conflated row: 1 without a matcher. + assert_eq!(u64s("no_matcher").value(0), 1); + assert!(u64s("no_matcher").is_null(1)); + assert!(u64s("atp_features").is_null(0)); + + let brands = batch.column_by_name("brands").unwrap(); + let brands = brands.as_any().downcast_ref::().unwrap(); + let brand = brands.value(0); + let brand = brand.as_any().downcast_ref::().unwrap(); + let field = |name: &str| { + brand + .column_by_name(name) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .value(0) + }; + // Only the supermarkets count, since the spider has no cafes. + assert_eq!(field("osm_features"), 7); + assert_eq!(field("deletion_candidates"), 2); + assert_eq!(brands.value(1).len(), 0); + Ok(()) + } +} diff --git a/src/pipeline/upload.rs b/src/pipeline/upload.rs index a3f2ebec..20fdeae4 100644 --- a/src/pipeline/upload.rs +++ b/src/pipeline/upload.rs @@ -404,6 +404,19 @@ const EDITS_TILES: Publish = Publish { sha512: false, }; +const SPIDER_STATS: Publish = Publish { + name: "spider-stats", + stem: "spider-stats", + ext: "parquet", + title: "spider-stats.parquet \u{2014} per-spider conflation statistics", + description: "Debugging aid for the AllThePlaces dashboard. Not a data product; \ + structure may change, or production may stop, without notice.", + format: "parquet", + mediatype: "application/vnd.apache.parquet", + describes: None, + sha512: false, +}; + /// Hashes `local`, uploads it to `data/--.` on /// the public bucket, and returns its manifest entry. fn publish( @@ -449,6 +462,14 @@ pub fn upload_conflated( publish(&CONFLATED, conflated, date, progress) } +pub fn upload_spider_stats( + stats: &Path, + date: &str, + progress: &MultiProgress, +) -> Result { + publish(&SPIDER_STATS, stats, date, progress) +} + pub fn upload_conflated_tiles( tiles: &Path, date: &str, diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 8644f55d..8dce78d0 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -84,6 +84,7 @@ fn test_pipeline() -> Result<()> { ); } assert_publish_artifacts(workdir.path())?; + assert_spider_stats(&workdir.path().join("spider-stats.parquet"))?; Ok(()) } @@ -106,8 +107,8 @@ fn assert_publish_artifacts(workdir: &Path) -> Result<()> { assert_eq!(manifest["name"], "osmdiffs"); let resources = manifest["resources"].as_array().context("no resources")?; - // conflated + bom + conflated-tiles + edits-tiles - assert_eq!(resources.len(), 4, "manifest resources: {resources:#?}"); + // conflated + bom + conflated-tiles + spider-stats + edits-tiles + assert_eq!(resources.len(), 5, "manifest resources: {resources:#?}"); // Resource name -> the workdir file it was uploaded from (the local // copies keep their undated names; only the S3 key and the manifest // path carry the date+hash). @@ -115,6 +116,7 @@ fn assert_publish_artifacts(workdir: &Path) -> Result<()> { "conflated" => "conflated.parquet", "conflated-tiles" => "conflated.pmtiles", "edits-tiles" => "diffed-places.pmtiles", + "spider-stats" => "spider-stats.parquet", "bom" => "", // exists under its manifest path other => panic!("unexpected resource {other}"), }; @@ -159,6 +161,88 @@ fn assert_publish_artifacts(workdir: &Path) -> Result<()> { Ok(()) } +/// Checks `spider-stats.parquet` against the fixture: one row per +/// spider, in order, with the counts that `assert_conflated_parquet` +/// expects of `conflated.parquet`. +fn assert_spider_stats(path: &Path) -> Result<()> { + use arrow::array::{Array, ListArray, StringArray, StructArray, UInt64Array}; + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + + let builder = ParquetRecordBatchReaderBuilder::try_new(std::fs::File::open(path)?)?; + let kv = builder + .metadata() + .file_metadata() + .key_value_metadata() + .context("no key-value metadata")?; + assert!( + kv.iter().any(|e| e.key == "osmdiffs:distance_bins_m" + && e.value.as_deref() == Some("1,10,25,50,100,250,1000")), + "distance bins missing from {kv:?}" + ); + let batches = builder.build()?.collect::, _>>()?; + let batch = arrow::compute::concat_batches(&batches[0].schema(), &batches)?; + let column = |name: &str| batch.column_by_name(name).expect(name).clone(); + let strings = |name: &str| -> Vec { + let c = column(name); + let a = c.as_any().downcast_ref::().expect(name); + (0..a.len()).map(|i| a.value(i).to_string()).collect() + }; + let numbers = |name: &str| -> Vec { + let c = column(name); + let a = c.as_any().downcast_ref::().expect(name); + (0..a.len()).map(|i| a.value(i)).collect() + }; + + let spiders = strings("spider"); + assert_eq!( + spiders, + [ + "denner_ch", + "empty", + "mediamarkt", + "misenso_ch", + "tchibo", + "winterthur_ch" + ] + ); + // 8 conflated rows, 3 of them matched; see assert_conflated_parquet. + assert_eq!(numbers("rows"), [1, 0, 1, 3, 1, 2]); + assert_eq!(numbers("same"), [1, 0, 1, 0, 1, 0]); + assert_eq!(numbers("missing_in_osm"), [0, 0, 0, 3, 0, 2]); + assert_eq!(numbers("atp_features"), [1, 0, 1, 3, 1, 4]); + // Two Winterthur features have no matcher, so no conflated row. + assert_eq!(numbers("no_matcher"), [0, 0, 0, 0, 0, 2]); + + // denner_ch matched 13.8 m away: bin 2, from 10 to 25 meters. + let distance = column("distance_m"); + let distance = distance.as_any().downcast_ref::().unwrap(); + let denner = distance.value(0); + let denner = denner.as_any().downcast_ref::().unwrap(); + assert_eq!(denner.values().to_vec(), [0, 0, 1, 0, 0, 0, 0, 0]); + + let brands = column("brands"); + let brands = brands.as_any().downcast_ref::().unwrap(); + let denner = brands.value(0); + let denner = denner.as_any().downcast_ref::().unwrap(); + assert_eq!(denner.len(), 1); + let field = |name: &str| denner.column_by_name(name).expect(name).clone(); + let string_at = |name: &str| { + field(name) + .as_any() + .downcast_ref::() + .unwrap() + .value(0) + .to_string() + }; + assert_eq!(string_at("country"), "CH"); + assert_eq!(string_at("wikidata_id"), "Q379911"); + assert_eq!(string_at("name"), "Denner"); + let osm_features = field("osm_features"); + let osm_features = osm_features.as_any().downcast_ref::().unwrap(); + assert_eq!(osm_features.value(0), 1); + Ok(()) +} + /// Byte-identical across independent runs on the same inputs -- the /// guarantee a Kubernetes retry (fresh volume, or a crash that re-runs a /// step) depends on. Covers `conflated.parquet` (embedded BOM +