diff --git a/docs/TECHNICAL_DESIGN.md b/docs/TECHNICAL_DESIGN.md index ec064d5..52222ab 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 0a99d3e..84c51b0 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 0000000..9afbdf8 --- /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 5883d0a..cd07387 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 405bf0c..743008f 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 39acd30..2e247c7 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 dc650bd..58a41f7 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 31d2b9f..096cdcd 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 257a32f..b1232af 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 35eec22..75eb84f 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 0000000..b988fb6 --- /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 a3f2ebe..20fdeae 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 8644f55..8dce78d 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 +