feat(osm): Store each feature's country in the OSM index

docs/TECHNICAL_DESIGN.md

@@ -370,6 +370,10 @@ fetch away.
   spatial index (`OsmFeatureIndex`), queryable by S2 cell range
   without decoding every candidate (`index.rs`,
   [`src/tables/feature_index.rs`](../src/tables/feature_index.rs)).
+  The index also keeps the country of each feature’s centroid, one
+  byte per feature, so that later steps needn’t decode geometry to
+  find it; looking them up adds about 30 seconds on the planet. An
+  index in an older format is built again.
   **~4h48m** for the whole step on the #665 bare-VM run, back when the
   planet download itself went over BitTorrent and dominated that
   figure. On the newer, HTTPS-based cpx42 run, the download itself

src/pipeline/conflate/deletion.rs

@@ -508,13 +508,15 @@ fn swiss_uics(value: &str) -> impl Iterator<Item = &str> {
     split_values(value).filter(|code| code.starts_with("85"))
 }
 
-/// Why the unmatched OSM feature with `tags` is a candidate for
-/// deletion, or `None` if nothing suggests so. `location` gives the
-/// feature's (longitude, latitude); it is only called when needed,
-/// and at most once, since it means decoding the geometry.
+/// Why the unmatched OSM feature with `tags` in `country` is a
+/// candidate for deletion, or `None` if nothing suggests so.
+/// `location` gives the feature's (longitude, latitude); it is only
+/// called when needed, and at most once, since it means decoding the
+/// geometry.
 pub(super) fn deletion_reason<'a>(
     tags: impl Iterator<Item = (&'a str, &'a str)> + Clone,
     atp: &AtpCoverage,
+    country: Option<Country>,
     location: impl FnOnce() -> Option<(f64, f64)>,
 ) -> Option<Candidate> {
     let mut location = Some(location);
@@ -551,7 +553,7 @@ pub(super) fn deletion_reason<'a>(
             // stops that the register has since given their own
             // country's codes, such as `uic_ref=8581987` in Nauders,
             // Austria. Neither Swiss signal is about such a stop.
-            if in_swiss_home(country(lng, lat), lng, lat) {
+            if in_swiss_home(country, lng, lat) {
                 if has_swiss_id {
                     return Some(Candidate::swiss(Reason::SwissTransitId));
                 }
@@ -569,7 +571,6 @@ pub(super) fn deletion_reason<'a>(
 
     let brand = brand_tags(tags)?;
     let local_kind = brand.local_kind();
-    let country = locate().and_then(|(lng, lat)| country(lng, lat));
     let keys: Vec<LocalKey> = brand
         .ids
         .iter()
@@ -792,7 +793,8 @@ mod tests {
     }
 
     fn reason(atp: &AtpCoverage, tags: &[(&str, &str)], at: (f64, f64)) -> Option<Reason> {
-        deletion_reason(tags.iter().copied(), atp, || Some(at)).map(|c| c.reason)
+        deletion_reason(tags.iter().copied(), atp, country(at.0, at.1), || Some(at))
+            .map(|c| c.reason)
     }
 
     #[test]
@@ -1081,7 +1083,10 @@ mod tests {
 
         // Without a location.
         let tags = [("highway", "bus_stop")];
-        assert_eq!(deletion_reason(tags.iter().copied(), &atp, || None), None);
+        assert_eq!(
+            deletion_reason(tags.iter().copied(), &atp, None, || None),
+            None
+        );
 
         // ATP lacks the Swiss register this week.
         assert_eq!(reason(&coverage(false), &tags, ZURICH_NORTH), None);
@@ -1190,7 +1195,7 @@ mod tests {
         assert_eq!(reason(&atp, &tags, SINGEN), Some(Reason::Wikidata));
         // Without a location, nothing says where the stop is.
         assert_eq!(
-            deletion_reason(old_code.iter().copied(), &atp, || None),
+            deletion_reason(old_code.iter().copied(), &atp, None, || None),
             None
         );
     }
@@ -1199,7 +1204,13 @@ mod tests {
     fn test_candidate_keys() {
         let atp = coverage(false);
         let tags = [("shop", "supermarket"), ("brand:wikidata", "Q9;Q1")];
-        let c = deletion_reason(tags.iter().copied(), &atp, || Some(SINGEN)).expect("candidate");
+        let c = deletion_reason(
+            tags.iter().copied(),
+            &atp,
+            country(SINGEN.0, SINGEN.1),
+            || Some(SINGEN),
+        )
+        .expect("candidate");
         assert_eq!(c.reason, Reason::Wikidata);
         // Only Q1 is covered by ATP, so only Q1 counts locally.
         assert_eq!(c.keys, vec![LocalKey::wikidata(1, None)]);

src/pipeline/conflate/mod.rs

@@ -542,7 +542,8 @@ fn send_deletion_candidates(
                 strings: &osm.strings,
             };
             let location = || centroid(&candidate);
-            let Some(found) = deletion_reason(candidate.tags(), coverage, location) else {
+            let country = osm.index.country(r);
+            let Some(found) = deletion_reason(candidate.tags(), coverage, country, location) else {
                 return Ok(None);
             };
             // A feature without a location can't be put on a map, nor

src/pipeline/inputs.rs

@@ -237,14 +237,12 @@ mod tests {
         ] {
             symlink(fixture(name), workdir.path().join(name))?;
         }
-        for name in [
-            "alltheplaces.parquet",
-            "osm-features.index",
-            "osm-features.index.inverted",
-            "osm-assemble.strings",
-        ] {
+        for name in ["alltheplaces.parquet", "osm-assemble.strings"] {
             std::fs::write(workdir.path().join(name), b"")?;
         }
+        crate::tables::OsmFeatureIndex::write_stub_for_test(
+            &workdir.path().join("osm-features.index"),
+        )?;
 
         let inputs = fetch_inputs(
             &test_client(),

src/pipeline/osm/fetch.rs

@@ -322,13 +322,10 @@ mod tests {
                 .path()
                 .join(format!("{PLANET_PBF_FILENAME}.meta.json")),
         )?;
-        for name in [
-            "osm-features.index",
-            "osm-features.index.inverted",
-            "osm-assemble.strings",
-        ] {
-            std::fs::write(workdir.path().join(name), b"")?;
-        }
+        std::fs::write(workdir.path().join("osm-assemble.strings"), b"")?;
+        crate::tables::OsmFeatureIndex::write_stub_for_test(
+            &workdir.path().join("osm-features.index"),
+        )?;
 
         let input = fetch_planet(
             UNREACHABLE_URL,

src/pipeline/wikidata_ids.rs

@@ -27,9 +27,7 @@ use crate::make_progress_bar;
 use crate::matchers::{OsmCandidate, parse_wikidata_ids};
 use crate::places::PlaceReader;
 use crate::tables::{OsmFeatures, U64Set};
-use crate::utils::countries::country;
 use anyhow::Result;
-use geo::Centroid;
 use indicatif::MultiProgress;
 use rayon::prelude::*;
 use std::path::{Path, PathBuf};
@@ -141,8 +139,8 @@ fn collect_from_atp(atp: &Path) -> Result<(Vec<u64>, BrandTotals)> {
 /// The Wikidata IDs in `osm`'s brand/operator/network tags, with
 /// duplicates, and their [BrandTotals]. Visits every feature in the
 /// index, in parallel; tag keys are compared as string pool indices, so
-/// only features with matching tags get their tags looked up as strings
-/// and their geometry decoded, for the country.
+/// only features with matching tags get their tags looked up as
+/// strings.
 fn collect_from_osm(
     osm: &OsmFeatures,
     progress: &MultiProgress,
@@ -185,12 +183,7 @@ fn collect_from_osm(
                         feature: &feature,
                         strings: &osm.strings,
                     };
-                    let country = candidate
-                        .geometry()
-                        .ok()
-                        .and_then(|g| g.centroid())
-                        .and_then(|c| country(c.x(), c.y()));
-                    totals.add(Source::Osm, candidate.tags(), country);
+                    totals.add(Source::Osm, candidate.tags(), osm.index.country(r));
                 }
                 Ok::<_, anyhow::Error>((ids, totals))
             },