From 1fff1acc3e94bf3c0a3f72759aa82b9de0af33d6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 12 Nov 2026 09:48:06 +0000 Subject: [PATCH] feat(spider-stats): keep ATP's dataset attributes per spider import_atp now keeps the dataset_attributes of each spider's FeatureCollection, such as license and attribution, in alltheplaces.spider-counts.tsv, and compute_spider_stats writes them to spider-stats.parquet as a MAP column. Downstream tools can join on atp.fetched.spider instead of repeating them on every row of conflated.parquet. Counts files from before are still read, without attributes, and import_atp counts again where the zip is there. Fixes #910. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ttfb2KDndfjnQEycP3Ay6G --- docs/outputs/SPIDER_STATS.md | 26 ++++++ src/pipeline/atp/mod.rs | 27 ++++++- src/pipeline/atp/spider_counts.rs | 129 ++++++++++++++++++++++++++++-- src/pipeline/spider_stats.rs | 57 ++++++++++++- tests/integration_test.rs | 30 +++++++ 5 files changed, 257 insertions(+), 12 deletions(-) diff --git a/docs/outputs/SPIDER_STATS.md b/docs/outputs/SPIDER_STATS.md index c051c93..cd72e63 100644 --- a/docs/outputs/SPIDER_STATS.md +++ b/docs/outputs/SPIDER_STATS.md @@ -42,6 +42,7 @@ invalid UINT64 nullable — left out: no usable geometry, or 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 +dataset_attributes MAP nullable — the spider's dataset attributes, such as its license 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" @@ -102,6 +103,22 @@ 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. +### Dataset attributes + +`dataset_attributes` holds the `dataset_attributes` that AllThePlaces +puts on the spider's GeoJSON FeatureCollection, such as `license`, +`license:wikidata`, `attribution:name`, `use:openstreetmap` and +`spider:lineage`; see AllThePlaces' +[documentation](https://github.com/alltheplaces/alltheplaces/blob/master/docs/SPIDER_LINEAGE.md). +They are there so that import documents and edits can cite the source, +license and attribution of a feature without repeating them on every +row of `conflated.parquet`: join on `atp.fetched.spider` there. The +entries are sorted by key. A value that isn't a string in the GeoJSON, +such as a number, is given as its JSON text. The map is empty for a +spider whose file in the dump is empty, and null for one without a +file. We keep the attributes of spiders whose features we leave out +too, since their license says why. + ### How they matched `rules` counts rows by `match.evidence.rule`; see @@ -215,6 +232,15 @@ ORDER BY missing_in_osm DESC LIMIT 10; ``` +The license of each spider whose features we left out: + +```sql +SELECT spider, not_for_osm, dataset_attributes['license'] AS license +FROM 'spider-stats.parquet' +WHERE not_for_osm > 0 +ORDER BY not_for_osm DESC; +``` + And the brands of one spider: ```sql diff --git a/src/pipeline/atp/mod.rs b/src/pipeline/atp/mod.rs index cd07387..e4fd855 100644 --- a/src/pipeline/atp/mod.rs +++ b/src/pipeline/atp/mod.rs @@ -116,7 +116,10 @@ pub fn import_atp( // out aren't in the parquet file; without it, they stay // missing. let counts = spider_counts::path(workdir); - let counts_missing = !counts.exists() || is_stale(&counts)?; + // Counts from a version without dataset attributes count + // as missing. + let counts_missing = + !counts.exists() || is_stale(&counts)? || !spider_counts::is_current(&counts)?; if !counts_missing || zip_modified.is_none() { if counts_missing { log::warn!("import_atp: no spider counts, and no zip to count from"); @@ -335,15 +338,19 @@ fn process_geojson( let mut json = first_line?; json.push_str("]}"); let parsed = json.parse::()?; - if let Some(spider) = parsed + let attrs = parsed .foreign_members .as_ref() - .and_then(|m| m.get("dataset_attributes")) + .and_then(|m| m.get("dataset_attributes")); + if let Some(spider) = attrs .and_then(|a| a.get("@spider")) .and_then(|s| s.as_str()) { counts.spider = spider.to_string(); } + if let Some(attrs) = attrs { + counts.dataset_attributes = spider_counts::attribute_pairs(attrs); + } if !is_usable_for_osm(&parsed) { // Ignore FeatureCollections that can't be conflated with OSM // according to the AllThePlaces metadata. @@ -968,9 +975,23 @@ mod tests { let licensed = file(r#""license:wikidata":"Q20007257""#); let c = process_geojson(Cursor::new(licensed), "output/x.geojson", None, None)?; assert_eq!((c.features, c.not_for_osm, c.written()), (3, 3, 0)); + // Kept even for a spider whose features we leave out, since + // the license says why. + assert_eq!( + c.dataset_attributes, + vec![ + ("@spider".to_string(), "s".to_string()), + ("license:wikidata".to_string(), "Q20007257".to_string()), + ( + "spider:collection_time".to_string(), + "2026-11-01T00:00:00".to_string() + ), + ] + ); let c = process_geojson(Cursor::new(""), "output/crashed.geojson", None, None)?; assert_eq!((c.spider.as_str(), c.features), ("crashed", 0)); + assert!(c.dataset_attributes.is_empty()); Ok(()) } diff --git a/src/pipeline/atp/spider_counts.rs b/src/pipeline/atp/spider_counts.rs index 2e247c7..3dbaaa9 100644 --- a/src/pipeline/atp/spider_counts.rs +++ b/src/pipeline/atp/spider_counts.rs @@ -13,7 +13,12 @@ use super::run_stats::SpiderRunStats; /// Filename, within `workdir`, of the counts that `import_atp` writes. const FILENAME: &str = "alltheplaces.spider-counts.tsv"; -const HEADER: &str = "spider\tfeatures\tnot_for_osm\tinvalid\tin_china\tdatum"; +const HEADER: &str = "spider\tfeatures\tnot_for_osm\tinvalid\tin_china\tdatum\tdataset_attributes"; + +/// The header before the file had `dataset_attributes`; [read] still +/// reads such files, with no attributes, and [is_current] tells +/// `import_atp` to count again where it can. +const OLD_HEADER: &str = "spider\tfeatures\tnot_for_osm\tinvalid\tin_china\tdatum"; /// Path of the counts that `import_atp` writes in `workdir`. pub(in crate::pipeline) fn path(workdir: &Path) -> PathBuf { @@ -44,6 +49,12 @@ pub(in crate::pipeline) struct SpiderCounts { /// look, as in tests without OSM data. The other two are converted /// to WGS-84. pub datum: String, + + /// The `dataset_attributes` of the spider's FeatureCollection, such + /// as `license` or `attribution:name`, sorted by key. Values that + /// aren't JSON strings are kept as JSON text. Empty for a spider + /// that crashed, since its file is empty. + pub dataset_attributes: Vec<(String, String)>, } impl SpiderCounts { @@ -63,10 +74,22 @@ pub(super) fn write(counts: &mut [SpiderCounts], out: &Path) -> Result<()> { ); writeln!(w, "{HEADER}")?; for c in counts.iter() { + // As a JSON object, which escapes any tabs and newlines. + let attrs: serde_json::Map = c + .dataset_attributes + .iter() + .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone()))) + .collect(); writeln!( w, - "{}\t{}\t{}\t{}\t{}\t{}", - c.spider, c.features, c.not_for_osm, c.invalid, c.in_china, c.datum + "{}\t{}\t{}\t{}\t{}\t{}\t{}", + c.spider, + c.features, + c.not_for_osm, + c.invalid, + c.in_china, + c.datum, + serde_json::Value::Object(attrs) )?; } w.into_inner()?.sync_all()?; @@ -78,15 +101,25 @@ pub(super) fn write(counts: &mut [SpiderCounts], out: &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) { - anyhow::bail!("{path:?} doesn't start with the expected header"); - } + let num_cols = match lines.next().transpose()?.as_deref() { + Some(HEADER) => 7, + Some(OLD_HEADER) => 6, + _ => anyhow::bail!("{path:?} doesn't start with the expected header"), + }; let mut counts = Vec::new(); for line in lines { let line = line?; let cols: Vec<&str> = line.split('\t').collect(); - let [spider, features, not_for_osm, invalid, in_china, datum] = cols[..] else { + if cols.len() != num_cols { anyhow::bail!("bad line in {path:?}: {line:?}"); + } + let [spider, features, not_for_osm, invalid, in_china, datum] = cols[..6] else { + unreachable!("checked the number of columns"); + }; + let dataset_attributes = match cols.get(6) { + Some(json) => parse_attributes(json) + .with_context(|| format!("bad dataset_attributes in {path:?}: {line:?}"))?, + None => Vec::new(), }; let num = |s: &str| -> Result { s.parse() @@ -99,11 +132,50 @@ pub(in crate::pipeline) fn read(path: &Path) -> Result> { invalid: num(invalid)?, in_china: num(in_china)?, datum: datum.to_string(), + dataset_attributes, }); } Ok(counts) } +/// Whether the counts at `path` were written by this version, with the +/// dataset attributes, rather than by an earlier one. +pub(super) fn is_current(path: &Path) -> Result { + let file = std::fs::File::open(path).with_context(|| format!("could not open {path:?}"))?; + let first = BufReader::new(file).lines().next().transpose()?; + Ok(first.as_deref() == Some(HEADER)) +} + +/// The dataset attributes of a spider's FeatureCollection, as string +/// key/value pairs sorted by key; see +/// [SpiderCounts::dataset_attributes]. +pub(super) fn attribute_pairs(attrs: &serde_json::Value) -> Vec<(String, String)> { + let Some(attrs) = attrs.as_object() else { + return Vec::new(); + }; + let mut pairs: Vec<(String, String)> = attrs + .iter() + .filter(|(_, v)| !v.is_null()) + .map(|(k, v)| { + let v = match v { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + (k.clone(), v) + }) + .collect(); + pairs.sort(); + pairs +} + +fn parse_attributes(json: &str) -> Result> { + let value: serde_json::Value = serde_json::from_str(json)?; + if !value.is_object() { + anyhow::bail!("not a JSON object"); + } + Ok(attribute_pairs(&value)) +} + /// Totals of `counts`, compared with what AllThePlaces says it wrote, /// for the log. #[derive(Debug, Default, Eq, PartialEq)] @@ -216,16 +288,59 @@ mod tests { invalid: 1, in_china: 4, datum: "gcj02".into(), + dataset_attributes: vec![ + ("attribution:name".into(), "Zara\tSA\nLine 2".into()), + ("license".into(), "Creative Commons Zero".into()), + ], }, counts("crashed", 0), ]; write(&mut written, &out)?; assert_eq!(written[0].spider, "crashed"); assert_eq!(read(&out)?, written); + assert!(is_current(&out)?); assert_eq!(written[1].written(), 9); Ok(()) } + #[test] + fn test_read_old_header() -> Result<()> { + let dir = tempfile::tempdir()?; + let out = path(dir.path()); + std::fs::write(&out, format!("{OLD_HEADER}\nzara\t10\t0\t1\t4\tgcj02\n"))?; + let c = read(&out)?; + assert_eq!((c[0].features, c[0].datum.as_str()), (10, "gcj02")); + assert!(c[0].dataset_attributes.is_empty()); + assert!(!is_current(&out)?); + Ok(()) + } + + #[test] + fn test_attribute_pairs() { + let attrs = serde_json::json!({ + "license": "CC0", + "@spider": "x", + "nsi_id": 42, + "use:openstreetmap": true, + "gone": null, + }); + let pairs = attribute_pairs(&attrs); + let pairs: Vec<(&str, &str)> = pairs + .iter() + .map(|(k, v)| (k.as_str(), v.as_str())) + .collect(); + assert_eq!( + pairs, + vec![ + ("@spider", "x"), + ("license", "CC0"), + ("nsi_id", "42"), + ("use:openstreetmap", "true"), + ] + ); + assert!(attribute_pairs(&serde_json::json!("x")).is_empty()); + } + #[test] fn test_summarize() { let mut china = counts("china", 5); diff --git a/src/pipeline/spider_stats.rs b/src/pipeline/spider_stats.rs index a4f28cf..f68e37c 100644 --- a/src/pipeline/spider_stats.rs +++ b/src/pipeline/spider_stats.rs @@ -25,8 +25,8 @@ 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, + Array, ArrayRef, Float32Array, Float32Builder, ListArray, ListBuilder, MapBuilder, + MapFieldNames, RecordBatch, StringArray, StringBuilder, StructBuilder, UInt64Builder, }; use arrow::datatypes::{DataType, Field, Fields, Schema}; use geo::Centroid; @@ -564,6 +564,24 @@ fn histogram_field(name: &str) -> Field { ) } +/// `MAP`, like the `tags` of `conflated.parquet`. The +/// entries are sorted by key, but like there, the type doesn't say so, +/// since `MapBuilder` can't. +fn attributes_type() -> DataType { + DataType::Map(Arc::new(attributes_entries()), /* sorted */ false) +} + +fn attributes_entries() -> Field { + Field::new( + "key_value", + DataType::Struct(Fields::from(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, false), + ])), + false, + ) +} + fn list_of(fields: Fields) -> DataType { DataType::List(Arc::new(Field::new( "item", @@ -583,6 +601,7 @@ pub(super) fn schema() -> Schema { Field::new("no_matcher", DataType::UInt64, true), Field::new("in_china", DataType::UInt64, true), Field::new("datum", DataType::Utf8, true), + Field::new("dataset_attributes", attributes_type(), true), u64_field("rows"), u64_field("same"), u64_field("located_in"), @@ -643,6 +662,16 @@ fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result) -> Result { + for (k, v) in &c.dataset_attributes { + attributes.keys().append_value(k); + attributes.values().append_value(v); + } + attributes.append(true)?; + } + None => attributes.append(false)?, + } num_rows.append_value(s.rows); same.append_value(s.same); located_in.append_value(s.located_in); @@ -822,6 +861,7 @@ fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result().unwrap(); let brand = brands.value(0); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 2068ce4..84774a1 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -229,6 +229,36 @@ fn assert_spider_stats(path: &Path) -> Result<()> { // Small spiders move all their features for chance pairs. assert_eq!(numbers("chance_sampled"), [1, 0, 1, 3, 1, 2]); + // The dataset attributes of each spider's FeatureCollection; none + // for the crashed spider, whose file is empty. + let attributes = column("dataset_attributes"); + let attributes = attributes + .as_any() + .downcast_ref::() + .unwrap(); + let attribute = |row: usize, key: &str| -> Option { + let entries = attributes.value(row); + let keys = entries.column(0).as_any().downcast_ref::()?; + let values = entries.column(1).as_any().downcast_ref::()?; + (0..keys.len()) + .find(|&i| keys.value(i) == key) + .map(|i| values.value(i).to_string()) + }; + assert_eq!(attributes.value(1).len(), 0); + assert_eq!(attribute(0, "@spider").as_deref(), Some("denner_ch")); + assert_eq!( + attribute(0, "spider:lineage").as_deref(), + Some("S_ATP_BRANDS") + ); + assert_eq!( + attribute(5, "license").as_deref(), + Some("Creative Commons Zero") + ); + assert_eq!( + attribute(5, "attribution:name:en").as_deref(), + Some("City of Winterthur") + ); + let offset = column("offset_east_m"); let offset = offset .as_any()