From f48f8fc7e9cadffb9f8ec9a0cbafdc2c9ca274f4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 01:20:47 +0000 Subject: [PATCH] feat: add importance column to conflated.parquet Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015Nw5ZNMD34EwRUnASRv4jy --- docs/TECHNICAL_DESIGN.md | 34 +++--- docs/outputs/CONFLATED_PARQUET.md | 39 +++++- src/matchers/mod.rs | 1 - src/pipeline/conflate/mod.rs | 62 +++++++++- src/pipeline/conflate/writer.rs | 32 ++++- src/pipeline/importance.rs | 191 ++++++++++++++++++++++++++++++ src/pipeline/mod.rs | 21 +++- src/pipeline/osm_views.rs | 7 +- src/pipeline/packaged.rs | 12 ++ src/pipeline/provenance.rs | 159 +++++++++++++++++++++---- src/pipeline/qrank/mod.rs | 102 ++++++++++++++-- src/pipeline/wikidata_ids.rs | 4 +- tests/integration_test.rs | 10 ++ 13 files changed, 604 insertions(+), 70 deletions(-) create mode 100644 src/pipeline/importance.rs diff --git a/docs/TECHNICAL_DESIGN.md b/docs/TECHNICAL_DESIGN.md index 45c96565..9c7a008b 100644 --- a/docs/TECHNICAL_DESIGN.md +++ b/docs/TECHNICAL_DESIGN.md @@ -211,6 +211,11 @@ graph TD ATP_PARQUET --> CONFLATE(conflate) OSM_INDEX --> CONFLATE + NONDESCRIPT_PHONES --> CONFLATE + NONDESCRIPT_EMAILS --> CONFLATE + OSMVIEWS_TIFF --> CONFLATE + OSMVIEWS_R --> CONFLATE + QRANK_Q --> CONFLATE CONFLATE --> CONFLATED[conflated.parquet] CONFLATED --> UPLOAD_CONFLATED(upload_conflated) --> S3A[("public S3")] @@ -239,13 +244,7 @@ graph TD ``` (Pink boxes are processing steps; plain rectangles are the files they -read or write. `qrank-quantiles`, `osmviews-ranks`, -`alltheplaces.nondescript-phones` and `alltheplaces.nondescript-emails` -have no outgoing edge above: nothing consumes them yet, they’re -generated on behalf of planned future work, see -[#713](https://github.com/brawer/osmdiffs/issues/713) for the first; -the other two are for using phone numbers and email addresses as -matching signals.) +read or write.) Every top-level step above is logged with its own wall-clock time and memory snapshot, regardless of success or failure — see @@ -356,7 +355,7 @@ fetch away. turned into its quantile among the places ATP covers. Raw ranks of inhabited places bunch up (in Switzerland, 80% lie between 0.40 and 0.61); quantiles spread them over 0 to 1 like the QRank quantiles. - Not consumed by anything yet. + Used by `conflate` for the `importance` column. - **`collect_wikidata_ids`** ([`src/pipeline/wikidata_ids.rs`](../src/pipeline/wikidata_ids.rs)) — collects every `brand:wikidata`, `operator:wikidata` and `network:wikidata` @@ -367,10 +366,12 @@ fetch away. rivers and the like. Visits every feature in `osm-features.index`. - **`compute_qrank_quantiles`** ([`src/pipeline/qrank/mod.rs`](../src/pipeline/qrank/mod.rs)) — gives - each ID in `wikidata-ids` its QRank quantile among those same IDs - (mid-rank for ties; IDs without pageviews share the lowest), so brand - importance can be combined with unrelated signals such as OSMViews. - Not consumed by anything yet. + each ID in `wikidata-ids` that has pageviews its QRank quantile among + those same IDs (mid-rank for ties), so brand importance can be + combined with unrelated signals such as OSMViews. IDs without + pageviews are left out and count as no brand: many brand items from + the Name Suggestion Index have no Wikipedia article. Used by + `conflate` for the `importance` column. - **`conflate`** ([`src/pipeline/conflate/mod.rs`](../src/pipeline/conflate/mod.rs)) — a single scan over AllThePlaces (the smaller of the two datasets), @@ -380,7 +381,12 @@ fetch away. the best. Opens the nondescript phone and email sets from `import_atp`, so that hotlines and central mailboxes never count as matching evidence. Writes one row per ATP feature — matched or not — to - `conflated.parquet`. See + `conflated.parquet`, each with an `importance` from 0 to 1 (see + [`src/pipeline/importance.rs`](../src/pipeline/importance.rs)): a + weighted mean of the location's OSMViews quantile, the brand's QRank + quantile and a coarse score for the kind of place + (`MatchMask::importance`), turned into a quantile among all rows + once every row is known. See [`docs/outputs/CONFLATED_PARQUET.md`](outputs/CONFLATED_PARQUET.md) for that file’s schema. **~8 minutes** (~5 min matching, ~3 min writing) for 3.8M ATP features against the full OpenStreetMap @@ -565,7 +571,7 @@ Two properties keep that safe: component’s `version`, the workflow’s `timeStart`/`timeEnd`) to `max(AllThePlaces run start, OSM planet replication)` — the freshness of the most-recently-updated input — rather than to - “now”. Its `serialNumber` is a version-5 UUID derived from the two + “now”. Its `serialNumber` is a version-5 UUID derived from the inputs’ SHA-256s, not a random v4 UUID. - The Parquet row order is a total order ([`ParquetRow::cmp`](../src/pipeline/conflate/writer.rs)): the S2 diff --git a/docs/outputs/CONFLATED_PARQUET.md b/docs/outputs/CONFLATED_PARQUET.md index 0ea26510..e2a9559e 100644 --- a/docs/outputs/CONFLATED_PARQUET.md +++ b/docs/outputs/CONFLATED_PARQUET.md @@ -40,6 +40,8 @@ osm STRUCT, nullable id UINT64 non-null role UTF8 nullable — null if OSM gives the member no role osm_geometry BINARY (WKB) nullable — null exactly when osm is null + +importance FLOAT non-null, 0.0 to 1.0 ``` A few things worth calling out that aren’t obvious from the above: @@ -78,6 +80,36 @@ A few things worth calling out that aren’t obvious from the above: each with its own type (node/way/relation), ID, and role (e.g. “outer”, “inner”, or `null` if OpenStreetMap doesn’t record one). +## Importance + +`importance` says how much a row’s place matters to map users, from +0.0 (least) to 1.0 (most), so you can work through the file in +priority order instead of in whatever order it happens to be in. It +is a *relative* position among the rows of this file: `importance >= +0.9` always selects the top tenth of rows, and the same place can get +a different value from one week to the next as other places come and +go. + +It combines three signals: + +- **Location**: how much the place’s location is looked at on + OpenStreetMap-based maps, according to + [OSMViews](https://osmviews.brawer.ch/). A shop on a busy downtown + street ranks above the same shop in a remote valley. +- **Brand**: how often the Wikipedia articles about the place’s brand, + operator or network (from its `brand:wikidata`, `operator:wikidata` + and `network:wikidata` tags) were viewed, according to + [QRank](https://qrank.toolforge.org/). A place without such a tag, or + whose brand has no Wikipedia article, is ranked on the other two + signals alone rather than pushed down. +- **Kind of place**: a coarse, hand-made score by category, so that a + defibrillator, pharmacy or hospital ranks above a bench or a tree. + +The exact weights and categories are an implementation detail that +may change; see +[`src/pipeline/importance.rs`](../../src/pipeline/importance.rs) and +[#713](https://github.com/brawer/osmdiffs/issues/713). + ## Geometry `atp_geometry`/`osm_geometry` are standard [OGC Simple @@ -147,9 +179,10 @@ embedded in its own metadata — CycloneDX being an established, [standardized](https://ecma-international.org/wp-content/uploads/ECMA-424_2nd_edition_december_2025.pdf) format for exactly this kind of “where did this data come from” record. It answers questions like which AllThePlaces run and which -OpenStreetMap snapshot went into this file, and which version of our -pipeline produced it. It also records licensing information for both -inputs and the output. +OpenStreetMap snapshot went into this file, which QRank and OSMViews +builds the `importance` column was computed from, and which version of +our pipeline produced it. It also records licensing information for +every input and the output. The same document is also published as a **standalone file** next to the Parquet — the `bom` resource in diff --git a/src/matchers/mod.rs b/src/matchers/mod.rs index a58e2b5a..df906c6a 100644 --- a/src/matchers/mod.rs +++ b/src/matchers/mod.rs @@ -116,7 +116,6 @@ impl MatchMask { /// signals for ranking features /// ([brawer/osmdiffs#713](https://github.com/brawer/osmdiffs/issues/713)), /// along with the location's and the brand's popularity. - #[allow(dead_code)] // Used by conflate in a follow-up change. pub fn importance(&self) -> f32 { if self.0 & Self::IMPORTANT.0 != 0 { return 1.0; diff --git a/src/pipeline/conflate/mod.rs b/src/pipeline/conflate/mod.rs index ce5c3764..269e940d 100644 --- a/src/pipeline/conflate/mod.rs +++ b/src/pipeline/conflate/mod.rs @@ -1,4 +1,5 @@ use super::atp::nondescript::Signal; +use super::importance::{Quantiles, Signals}; use super::last_modified; use crate::{ make_progress_bar, @@ -33,23 +34,44 @@ mod writer; use s2_util::MergedCellRanges; use writer::{ParquetRow, ParquetWriter}; +/// The files that the `importance` column is computed from; see +/// `pipeline::importance`. +pub struct ImportanceInputs<'a> { + pub osmviews: &'a Path, + pub osmviews_ranks: &'a Path, + pub qrank_quantiles: &'a Path, +} + pub fn conflate( atp: &Path, osm: &OsmFeatures, + importance: &ImportanceInputs, progress: &MultiProgress, workdir: &Path, pipeline_run_id: &str, ) -> Result { let nondescript_phones = Signal::Phone.path(workdir); let nondescript_emails = Signal::Email.path(workdir); - let input_modified = - last_modified(&[atp, &nondescript_phones, &nondescript_emails])?.max(osm.modified()?); + let input_modified = last_modified(&[ + atp, + &nondescript_phones, + &nondescript_emails, + importance.osmviews, + importance.osmviews_ranks, + importance.qrank_quantiles, + ])? + .max(osm.modified()?); let out_path = workdir.join("conflated.parquet"); if out_path.exists() && last_modified(&[&out_path])? >= input_modified { return Ok(out_path); } let nondescript = Nondescript::open(&nondescript_phones, &nondescript_emails)?; + let signals = Signals::open( + importance.osmviews, + importance.osmviews_ranks, + importance.qrank_quantiles, + )?; let atp_reader = PlaceReader::open(atp)?; let num_atp_features = atp_reader.total_rows() as u64; let producer_progress = @@ -61,7 +83,14 @@ pub fn conflate( thread::scope(|s| { let (tx, rx) = sync_channel::(8192); s.spawn(|| { - producer_result = produce_rows(&atp_reader, osm, &nondescript, producer_progress, tx); + producer_result = produce_rows( + &atp_reader, + osm, + &nondescript, + &signals, + producer_progress, + tx, + ); }); s.spawn(|| { writer_result = @@ -77,13 +106,15 @@ pub fn conflate( struct ConflatedFeature { atp: Option, osm: Option, - // TODO: Signals for ranking. + /// See `pipeline::importance::score`. + score: f32, } fn produce_rows( atp_reader: &PlaceReader, osm: &OsmFeatures, nondescript: &Nondescript, + signals: &Signals, progress_bar: ProgressBar, out: SyncSender, ) -> Result<()> { @@ -124,6 +155,7 @@ fn produce_rows( let mut conflated = ConflatedFeature { atp: Some(atp.deep_clone()), osm: None, + score: 0.0, }; let mut best_score: f64 = 0.0; let covering = { @@ -149,6 +181,16 @@ fn produce_rows( } } + conflated.score = match &conflated.osm { + Some(feature) => { + let candidate = OsmCandidate { + feature, + strings: &osm.strings, + }; + signals.score(atp, candidate.tags()) + } + None => signals.score(atp, std::iter::empty()), + }; let row = ParquetRow::from_conflated(conflated, &osm.strings)?; out.send(row)?; @@ -229,10 +271,18 @@ fn write_conflated( EXTERNAL_SORT_CHUNK_BYTES as u64, )) .build()?; + // Every row's importance score, collected on the way into the sort: + // `importance` is a score's quantile among all of them, which is + // only known once every row has been produced. + let scores = std::sync::Mutex::new(Vec::new()); let sorted = sorter.sort(cf.iter().map(|row| { progress.set_length(row_count.fetch_add(1, Ordering::SeqCst)); + scores.lock().expect("scores lock").push(row.score()); std::io::Result::Ok(row) }))?; + // `sort` has consumed every row by now. Taken rather than + // `into_inner`, since `sorted` still borrows `scores`. + let quantiles = Quantiles::new(std::mem::take(&mut *scores.lock().expect("scores lock"))); progress.set_length(row_count.load(Ordering::SeqCst)); let provenance_bom = crate::pipeline::provenance::build_bom_for_conflated_parquet( workdir, @@ -244,7 +294,9 @@ fn write_conflated( let mut writer = ParquetWriter::create(out, /* max_rows_per_group */ 200_000, &provenance_bom)?; for row in sorted { - writer.write(row?)?; + let row = row?; + let importance = quantiles.get(row.score()); + writer.write(row, importance)?; progress.inc(1); } writer.close()?; diff --git a/src/pipeline/conflate/writer.rs b/src/pipeline/conflate/writer.rs index 5b56d860..5fac0f8d 100644 --- a/src/pipeline/conflate/writer.rs +++ b/src/pipeline/conflate/writer.rs @@ -2,8 +2,8 @@ use anyhow::{Context, Ok, Result}; use arrow::array::{ ArrayRef, RecordBatch, StructArray, builder::{ - BinaryBuilder, ListBuilder, MapBuilder, MapFieldNames, StringBuilder, StructBuilder, - TimestampMillisecondBuilder, UInt32Builder, UInt64Builder, + BinaryBuilder, Float32Builder, ListBuilder, MapBuilder, MapFieldNames, StringBuilder, + StructBuilder, TimestampMillisecondBuilder, UInt32Builder, UInt64Builder, }, }; use arrow_buffer::builder::NullBufferBuilder; @@ -71,6 +71,8 @@ pub struct ParquetWriter { osm_geometries: BinaryBuilder, // See `atp_geometry_tally` above. osm_geometry_tally: GeometryTally, + + importance: Float32Builder, } /// A single row in the conflated parquet file. @@ -110,6 +112,11 @@ pub struct ParquetRow { atp_fetched: Option, atp_tags: Vec<(String, String)>, atp_shape_wkb: Vec, + + /// Internal, like `s2_cell_id`: the published `importance` is this + /// score's quantile among all rows, which the writer computes once + /// every row is known. See `pipeline::importance`. + score: f32, } /// Key under which the CycloneDX provenance BOM (see `pipeline::provenance`) @@ -220,6 +227,8 @@ impl ParquetWriter { /* data_capacity */ max_rows_per_group * 21, ), osm_geometry_tally: GeometryTally::default(), + + importance: Float32Builder::with_capacity(max_rows_per_group), }) } @@ -270,7 +279,9 @@ impl ParquetWriter { ) } - pub fn write(&mut self, row: ParquetRow) -> Result<()> { + /// Appends `row`, whose published `importance` (see + /// `pipeline::importance`) is `importance`. + pub fn write(&mut self, row: ParquetRow, importance: f32) -> Result<()> { let row_s2_cell_id = row.s2_cell_id.get(); assert!(row_s2_cell_id >= self.last_s2_cell_id); self.last_s2_cell_id = row_s2_cell_id; @@ -387,6 +398,7 @@ impl ParquetWriter { self.osm_geometries.append_null(); } + self.importance.append_value(importance); self.rows_in_group += 1; Ok(()) } @@ -557,6 +569,7 @@ impl ParquetWriter { Arc::new(atp_geometry) as ArrayRef, Arc::new(osm_struct) as ArrayRef, Arc::new(osm_geometry) as ArrayRef, + Arc::new(self.importance.finish()) as ArrayRef, ], )?; @@ -591,6 +604,7 @@ impl ParquetRow { ) -> Result { let atp = cf.atp; let osm = cf.osm; + let score = cf.score; // Internal sort key. Almost always available for free from `atp` // -- `produce_rows` only ever calls this with `atp: Some(..)`, @@ -717,8 +731,13 @@ impl ParquetRow { osm_way_members, osm_relation_members, osm_shape_wkb, + score, }) } + + pub(super) fn score(&self) -> f32 { + self.score + } } impl Ord for ParquetRow { @@ -818,7 +837,11 @@ mod schema { ); let osm_geometry = new_geo_field("osm_geometry", NULLABLE); - Schema::new(vec![atp, atp_geometry, osm, osm_geometry]) + // How much this feature matters to map users, from 0.0 to 1.0; + // see `pipeline::importance`. + let importance = Field::new("importance", DataType::Float32, NOT_NULLABLE); + + Schema::new(vec![atp, atp_geometry, osm, osm_geometry, importance]) } /// Fields of one `osm.relation_members` list entry: the member's own @@ -906,6 +929,7 @@ mod tests { ConflatedFeature { atp: Some(atp), osm: Some(osm), + score: 0.0, } } diff --git a/src/pipeline/importance.rs b/src/pipeline/importance.rs new file mode 100644 index 00000000..52795522 --- /dev/null +++ b/src/pipeline/importance.rs @@ -0,0 +1,191 @@ +//! How much a conflated feature matters to map users, from 0.0 to 1.0, +//! so that downstream users can work through conflated.parquet in +//! priority order +//! ([brawer/osmdiffs#713](https://github.com/brawer/osmdiffs/issues/713)). +//! +//! Three signals go in, each a unit-free value in `[0, 1]`: +//! +//! * **location**: the quantile of the place's OSMViews rank among the +//! locations of all AllThePlaces features (see [`super::osm_views`]); +//! * **brand**: the highest QRank quantile among the place's +//! `brand:`, `operator:` and `network:wikidata` IDs (see +//! [`super::qrank`]), from its AllThePlaces tags, else from the +//! matched OpenStreetMap feature's; and +//! * **category**: [`MatchMask::importance`], a coarse hand-made score +//! by kind of place, so that an AED ranks above a flower bed. +//! +//! [`score`] takes their weighted mean. A place without any brand drops +//! that term instead of being pushed down for it, since most places +//! that matter (a local hospital, an independent bakery) have no +//! Wikidata brand. The published `importance` is the score's quantile +//! among all conflated rows (see [`Quantiles`]), so `importance >= 0.9` +//! always selects the top tenth. + +use super::osm_views::OsmViewsQuantiles; +use super::qrank::QRankQuantiles; +use crate::matchers::MatchMask; +use crate::places::Place; +use anyhow::{Context, Result}; +use osmviews::OsmViews; +use std::path::Path; + +const LOCATION_WEIGHT: f32 = 0.4; +const BRAND_WEIGHT: f32 = 0.3; +const CATEGORY_WEIGHT: f32 = 0.3; + +/// Weighted mean of the signals, each in `[0, 1]`. `brand` is `None` +/// for places without any brand, operator or network; its weight then +/// drops out. +pub fn score(location: f32, brand: Option, category: f32) -> f32 { + let (brand_term, brand_weight) = match brand { + Some(b) => (BRAND_WEIGHT * b, BRAND_WEIGHT), + None => (0.0, 0.0), + }; + (LOCATION_WEIGHT * location + brand_term + CATEGORY_WEIGHT * category) + / (LOCATION_WEIGHT + brand_weight + CATEGORY_WEIGHT) +} + +/// The lookup tables [`Signals::score`] needs, opened once per +/// pipeline run and shared by all matching threads. +pub struct Signals<'a> { + osmviews: OsmViews, + locations: OsmViewsQuantiles, + brands: QRankQuantiles<'a>, +} + +impl Signals<'_> { + pub fn open(osmviews: &Path, osmviews_ranks: &Path, qrank_quantiles: &Path) -> Result { + Ok(Signals { + osmviews: OsmViews::open(osmviews) + .with_context(|| format!("Failed to open {}", osmviews.display()))?, + locations: OsmViewsQuantiles::open(osmviews_ranks)?, + brands: QRankQuantiles::open(qrank_quantiles)?, + }) + } + + /// The [`score`] of `atp`, matched to an OpenStreetMap feature + /// with `osm_tags` (empty if unmatched). + pub fn score<'t>( + &self, + atp: &Place, + osm_tags: impl IntoIterator, + ) -> f32 { + let ll = s2::latlng::LatLng::from(s2::cellid::CellID(atp.s2_cell_id)); + let rank = self.osmviews.rank(ll.lng.deg(), ll.lat.deg()); + let location = self.locations.quantile(rank) as f32; + let atp_tags = atp.tags.iter().map(|(k, v)| (k.as_str(), v.as_str())); + let brand = self + .brands + .brand_quantile(atp_tags) + .or_else(|| self.brands.brand_quantile(osm_tags)); + score(location, brand, category(atp.mask)) + } +} + +fn category(mask: MatchMask) -> f32 { + mask.importance() +} + +/// Turns scores into quantiles among all of them, for the published +/// `importance`. Mid-rank, so ties share a value: with `less` scores +/// below this one and `same` equal to it, out of `n`, the quantile is +/// `(less + same / 2) / n`. +pub struct Quantiles { + /// Ascending. + sorted: Vec, +} + +impl Quantiles { + pub fn new(mut scores: Vec) -> Self { + scores.sort_unstable_by(f32::total_cmp); + Quantiles { sorted: scores } + } + + /// The quantile of `score`, or 0.5 if there are no scores at all. + pub fn get(&self, score: f32) -> f32 { + if self.sorted.is_empty() { + return 0.5; + } + let less = self.sorted.partition_point(|&s| s < score); + let less_or_same = self.sorted.partition_point(|&s| s <= score); + let same = (less_or_same - less) as f64; + ((less as f64 + same / 2.0) / self.sorted.len() as f64) as f32 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn mask(tags: &[(&str, &str)]) -> MatchMask { + let mut mask = MatchMask::default(); + for (key, value) in tags { + mask.add_tag(key, value); + } + mask + } + + #[test] + fn test_score() { + assert_eq!(score(1.0, Some(1.0), 1.0), 1.0); + assert_eq!(score(0.0, Some(0.0), 0.0), 0.0); + assert_eq!(score(0.5, Some(1.0), 0.0), 0.5); + // Without a brand, location and category share the weight. + assert!((score(0.5, None, 1.0) - 0.714286).abs() < 1e-6); + assert_eq!(score(0.4, None, 0.4), 0.4); + } + + /// Pairs at the same location where the first should rank above + /// the second. Documents the intent, and catches regressions when + /// weights or category scores change. + #[test] + fn test_score_pairs() { + let at = |location: f32, brand: Option, tags: &[(&str, &str)]| { + score(location, brand, category(mask(tags))) + }; + let pairs: [(&str, f32, f32); 6] = [ + ( + "AED beats flower bed", + at(0.5, None, &[("emergency", "defibrillator")]), + at(0.5, None, &[("landuse", "flowerbed")]), + ), + ( + "hospital beats vending machine", + at(0.5, None, &[("amenity", "hospital")]), + at(0.5, None, &[("amenity", "vending_machine")]), + ), + ( + "supermarket beats bench", + at(0.5, None, &[("shop", "supermarket")]), + at(0.5, None, &[("amenity", "bench")]), + ), + ( + "famous brand beats obscure brand", + at(0.5, Some(0.99), &[("shop", "supermarket")]), + at(0.5, Some(0.12), &[("shop", "supermarket")]), + ), + ( + "busy street beats quiet one", + at(0.9, None, &[("shop", "bakery")]), + at(0.2, None, &[("shop", "bakery")]), + ), + ( + "AED on a quiet street beats bench on a busy one", + at(0.3, None, &[("emergency", "defibrillator")]), + at(0.7, None, &[("amenity", "bench")]), + ), + ]; + for (name, a, b) in pairs { + assert!(a > b, "{name}: {a} <= {b}"); + } + } + + #[test] + fn test_quantiles() { + let q = Quantiles::new(vec![0.7, 0.0, 0.5, 0.5, 0.5]); + assert_eq!(q.get(0.0), 0.1); + assert_eq!(q.get(0.5), 0.5); + assert_eq!(q.get(0.7), 0.9); + assert_eq!(Quantiles::new(Vec::new()).get(0.3), 0.5); + } +} diff --git a/src/pipeline/mod.rs b/src/pipeline/mod.rs index 648b3d7f..bf54749a 100644 --- a/src/pipeline/mod.rs +++ b/src/pipeline/mod.rs @@ -35,6 +35,7 @@ mod conflate; mod conflated_tiles; mod datapackage; mod edits; +mod importance; mod inputs; mod logging; mod memstats; @@ -230,8 +231,7 @@ fn run_pipeline_steps( let atp = run_step("import_atp", || { atp::import_atp(&inputs.atp, progress, workdir) })?; - // Not consumed by anything yet; see `osm_views`. - let _osmviews_ranks = run_step("compute_osmviews_ranks", || { + let osmviews_ranks = run_step("compute_osmviews_ranks", || { osm_views::compute_osmviews_ranks(&atp, &inputs.osmviews.path, workdir) })?; // `osm_features` is scoped to just these steps, not bound at @@ -250,12 +250,23 @@ fn run_pipeline_steps( let wikidata_ids = run_step("collect_wikidata_ids", || { wikidata_ids::collect_wikidata_ids(&atp, &osm_features, progress, workdir) })?; - // Not consumed by anything yet either; see `qrank`. - let _qrank_quantiles = run_step("compute_qrank_quantiles", || { + let qrank_quantiles = run_step("compute_qrank_quantiles", || { qrank::compute_qrank_quantiles(&inputs.qrank.path, &wikidata_ids, workdir) })?; + let importance = conflate::ImportanceInputs { + osmviews: &inputs.osmviews.path, + osmviews_ranks: &osmviews_ranks, + qrank_quantiles: &qrank_quantiles, + }; run_step("conflate", || { - conflate::conflate(&atp, &osm_features, progress, workdir, pipeline_run_id) + conflate::conflate( + &atp, + &osm_features, + &importance, + progress, + workdir, + pipeline_run_id, + ) })? }; diff --git a/src/pipeline/osm_views.rs b/src/pipeline/osm_views.rs index 2f390d51..b5345933 100644 --- a/src/pipeline/osm_views.rs +++ b/src/pipeline/osm_views.rs @@ -16,8 +16,8 @@ //! [`OsmViewsQuantiles`] maps any `rank()` value to its quantile among //! them, so the same scale works for OpenStreetMap features too. //! -//! Not consumed by anything yet; see -//! [brawer/osmdiffs#713](https://github.com/brawer/osmdiffs/issues/713). +//! Used for the `importance` column of conflated.parquet; see +//! [`super::importance`]. use super::packaged::PackagedResource; use crate::places::PlaceReader; @@ -106,14 +106,11 @@ fn write_ranks(ranks: &[f32], out: &Path) -> Result<()> { /// Maps an OSMViews `rank()` to its quantile among the locations of all /// AllThePlaces features, as stored by [`compute_osmviews_ranks`]. -// Used once the ranking signals are combined (#713). -#[allow(dead_code)] pub struct OsmViewsQuantiles { /// Ascending. sorted: Vec, } -#[allow(dead_code)] impl OsmViewsQuantiles { pub fn open(path: &Path) -> Result { let mut data = Vec::new(); diff --git a/src/pipeline/packaged.rs b/src/pipeline/packaged.rs index ed73f6d5..9e758d94 100644 --- a/src/pipeline/packaged.rs +++ b/src/pipeline/packaged.rs @@ -235,6 +235,18 @@ async fn write_meta_json(metadata: &PackagedMetadata, dest: &Path) -> Result<()> Ok(()) } +/// The metadata that [`fetch_packaged`] stored for `resource` in +/// `workdir`, for the provenance BOM. +pub(crate) fn read_cached_metadata( + resource: &PackagedResource, + workdir: &Path, +) -> Result { + let path = workdir.join(resource.meta_filename); + let data = std::fs::read_to_string(&path) + .with_context(|| format!("Failed to read {}", path.display()))?; + serde_json::from_str(&data).with_context(|| format!("Failed to parse {}", path.display())) +} + async fn read_meta_json(path: &Path) -> Result { let data = tokio::fs::read_to_string(path) .await diff --git a/src/pipeline/provenance.rs b/src/pipeline/provenance.rs index 559e85c5..e82cf595 100644 --- a/src/pipeline/provenance.rs +++ b/src/pipeline/provenance.rs @@ -13,6 +13,7 @@ //! values, so this stays decoupled from however those importers (and //! whatever indexing they feed into) end up wired together. +use crate::pipeline::packaged::{PackagedMetadata, PackagedResource, read_cached_metadata}; use crate::pipeline::{self, AtpMetadata, OsmMetadata}; use anyhow::{Context, Result}; use serde_json::{Value, json}; @@ -21,6 +22,14 @@ use time::UtcDateTime; use time::format_description::well_known::Rfc3339; use uuid::Uuid; +/// Supplier of QRank and OSMViews, which feed the `importance` column. +fn brawer_supplier() -> Value { + json!({ + "name": "Sascha Brawer", + "url": ["https://brawer.ch"] + }) +} + /// GitHub repository this pipeline is published from -- used to build /// `metadata.tools.components[0]`'s external references and purl. const REPO_URL: &str = "https://github.com/brawer/osmdiffs"; @@ -149,8 +158,17 @@ pub fn build_bom_for_conflated_parquet( let osm_metadata = pipeline::read_cached_metadata(workdir) .context("could not read OpenStreetMap provenance")?; + let qrank_metadata = read_cached_metadata(&pipeline::qrank::QRANK, workdir) + .context("could not read QRank provenance")?; + let osmviews_metadata = read_cached_metadata(&pipeline::osm_views::OSMVIEWS, workdir) + .context("could not read OSMViews provenance")?; + let anchor = format_rfc3339(anchor_timestamp(&atp_metadata, &osm_metadata)); - let serial_number = deterministic_serial_number(&atp_metadata, &osm_metadata)?; + let serial_number = deterministic_serial_number( + &atp_metadata, + &osm_metadata, + &[&qrank_metadata, &osmviews_metadata], + )?; Ok(json!({ "bomFormat": "CycloneDX", @@ -168,16 +186,18 @@ pub fn build_bom_for_conflated_parquet( "components": [ atp_component(&atp_metadata)?, osm_component(&osm_metadata)?, + packaged_component(&pipeline::qrank::QRANK, &qrank_metadata), + packaged_component(&pipeline::osm_views::OSMVIEWS, &osmviews_metadata), ], - // Declares conflated.parquet's two inputs as *its* dependencies, + // Declares conflated.parquet's inputs as *its* dependencies, // so the dependency graph has a single root (conflated.parquet) - // instead of three unlinked ones -- an NTIA-minimum-elements + // instead of several unlinked ones -- an NTIA-minimum-elements // validator flags components no other component depends on as // extra roots. Same pattern as scripts/sbom/pipeline.jq uses for // the container-image SBOM. "dependencies": [{ "ref": "conflated.parquet", - "dependsOn": ["alltheplaces.zip", pipeline::PLANET_PBF_FILENAME], + "dependsOn": input_refs(), }], "formulation": [formulation(pipeline_run_id, &anchor, &anchor)], })) @@ -194,6 +214,17 @@ pub(crate) fn anchor_timestamp(atp: &AtpMetadata, osm: &OsmMetadata) -> UtcDateT atp.start_time.max(osm.replication_timestamp) } +/// The `bom-ref`s of conflated.parquet's inputs, in the order the BOM +/// lists them. +fn input_refs() -> [&'static str; 4] { + [ + "alltheplaces.zip", + pipeline::PLANET_PBF_FILENAME, + pipeline::qrank::QRANK.filename, + pipeline::osm_views::OSMVIEWS.filename, + ] +} + /// Fixed, arbitrary namespace for the version-5 (SHA-1) UUID used as the /// BOM's `serialNumber` -- a constant so the derivation /// `UUIDv5(namespace, atp_sha256 ":" osm_sha256)` is entirely @@ -205,11 +236,17 @@ const SERIAL_NAMESPACE: Uuid = Uuid::from_bytes([ 0x18, 0x68, 0x7b, 0x2b, 0xa4, 0x9c, 0x56, 0x07, 0xb1, 0xe7, 0xf7, 0x40, 0xda, 0xee, 0x83, 0x2f, ]); -/// The BOM's `serialNumber`, derived from the two inputs' SHA-256s so an -/// identical input pair always yields an identically identified BOM -- +/// The BOM's `serialNumber`, derived from the inputs' SHA-256s so +/// identical inputs always yield an identically identified BOM -- /// rather than a fresh random v4 UUID per run, which made every BOM -/// (and, with it, `conflated.parquet`) non-reproducible. -fn deterministic_serial_number(atp: &AtpMetadata, osm: &OsmMetadata) -> Result { +/// (and, with it, `conflated.parquet`) non-reproducible. The packaged +/// inputs (QRank, OSMViews) are appended after ATP's and OSM's, so a +/// new QRank or OSMViews build also gets a new serial number. +fn deterministic_serial_number( + atp: &AtpMetadata, + osm: &OsmMetadata, + packaged: &[&PackagedMetadata], +) -> Result { let atp_sha = atp .sha256 .as_deref() @@ -217,11 +254,14 @@ fn deterministic_serial_number(atp: &AtpMetadata, osm: &OsmMetadata) -> Result Result { })) } -/// Ties `tool_component()`, the two input `*_component()`s, and +/// A weekly build of QRank or OSMViews, both CC0-licensed and +/// published as Frictionless Data Packages with their own CycloneDX +/// BOM, which this component links to. +fn packaged_component(resource: &PackagedResource, metadata: &PackagedMetadata) -> Value { + let mut external_references = vec![ + json!({"type": "distribution", "url": metadata.url}), + license_external_reference(CC0_URL), + ]; + if let Some(bom_url) = &metadata.bom_url { + external_references.push(json!({"type": "bom", "url": bom_url})); + } + json!({ + "bom-ref": resource.filename, + "type": "data", + "name": resource.filename, + "version": metadata.version, + "supplier": brawer_supplier(), + "licenses": license("CC0-1.0", CC0_URL), + "hashes": [{"alg": "SHA-256", "content": metadata.sha256}], + "purl": format!( + "pkg:generic/{}@{}?download_url={}&checksum=sha256:{}", + resource.name, metadata.version, metadata.url, metadata.sha256 + ), + "data": [{"type": "dataset", "contents": {"url": metadata.url}}], + "externalReferences": external_references, + }) +} + +/// Ties `tool_component()`, the input `*_component()`s, and /// `output_component()` together into one workflow: which tool consumed /// which inputs to produce which output. fn formulation(pipeline_run_id: &str, time_start: &str, time_end: &str) -> Value { @@ -374,10 +442,10 @@ fn formulation(pipeline_run_id: &str, time_start: &str, time_end: &str) -> Value "timeStart": time_start, "timeEnd": time_end, "resourceReferences": [{"ref": "tool-osmdiffs"}], - "inputs": [ - {"resource": {"ref": "alltheplaces.zip"}}, - {"resource": {"ref": pipeline::PLANET_PBF_FILENAME}}, - ], + "inputs": input_refs() + .iter() + .map(|r| json!({"resource": {"ref": r}})) + .collect::>(), "outputs": [{"resource": {"ref": "conflated.parquet"}}], }], }) @@ -425,6 +493,12 @@ mod tests { "sha256": "56e12b62871018c7a969c9924bcfb1bdce15676bee706156260f101db809b9e1" }"#, )?; + for name in ["qrank.meta.json", "osmviews.meta.json"] { + let mut src = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")); + src.push("tests/test_data"); + src.push(name); + std::os::unix::fs::symlink(&src, workdir.join(name))?; + } Ok(()) } @@ -467,7 +541,7 @@ mod tests { assert_eq!(output["copyright"], OSM_COPYRIGHT); let components = bom["components"].as_array().expect("components array"); - assert_eq!(components.len(), 2); + assert_eq!(components.len(), 4); let atp = &components[0]; assert_eq!(atp["bom-ref"], "alltheplaces.zip"); @@ -514,8 +588,43 @@ mod tests { 56e12b62871018c7a969c9924bcfb1bdce15676bee706156260f101db809b9e1" ); + let qrank = &components[2]; + assert_eq!(qrank["bom-ref"], "qrank.csv.gz"); + assert_eq!(qrank["version"], "2026-10-11"); + assert_eq!(qrank["supplier"]["name"], "Sascha Brawer"); + assert_eq!(qrank["licenses"][0]["license"]["id"], "CC0-1.0"); + assert_eq!( + qrank["hashes"][0]["content"], + "f478d9967d46b3cb7aa5bbf8c8625770b3de25d68e007ab42c4f9d5259709a0b" + ); + assert_eq!( + qrank["purl"], + "pkg:generic/qrank@2026-10-11\ + ?download_url=https://qrank.toolforge.org/download/qrank-20261011.csv.gz\ + &checksum=sha256:f478d9967d46b3cb7aa5bbf8c8625770b3de25d68e007ab42c4f9d5259709a0b" + ); + let qrank_bom = qrank["externalReferences"] + .as_array() + .expect("externalReferences") + .iter() + .find(|r| r["type"] == "bom") + .expect("a bom external reference"); + assert_eq!( + qrank_bom["url"], + "https://qrank.toolforge.org/download/qrank-20261011.cdx.json" + ); + + let osmviews = &components[3]; + assert_eq!(osmviews["bom-ref"], "osmviews.tiff"); + assert_eq!(osmviews["version"], "2026-10-18"); + assert_eq!(osmviews["licenses"][0]["license"]["id"], "CC0-1.0"); + assert_eq!( + osmviews["data"][0]["contents"]["url"], + "https://osmviews.brawer.ch/data/osmviews-20261018.tiff" + ); + // Single-rooted dependency graph: conflated.parquet depends on - // both inputs, so neither shows up as an orphan root. + // every input, so none shows up as an orphan root. assert_eq!(bom["dependencies"][0]["ref"], "conflated.parquet"); let depends_on = bom["dependencies"][0]["dependsOn"] .as_array() @@ -525,7 +634,12 @@ mod tests { .collect::>(); assert_eq!( depends_on, - vec!["alltheplaces.zip", pipeline::PLANET_PBF_FILENAME] + vec![ + "alltheplaces.zip", + pipeline::PLANET_PBF_FILENAME, + "qrank.csv.gz", + "osmviews.tiff" + ] ); // None of the deliberately-dropped ATP/OSM fields (run-health @@ -554,6 +668,8 @@ mod tests { "conflated.parquet", "alltheplaces.zip", pipeline::PLANET_PBF_FILENAME, + "qrank.csv.gz", + "osmviews.tiff", ]; let workflow = &bom["formulation"][0]["workflows"][0]; assert_eq!(workflow["uid"], "k8s-job-42"); @@ -564,6 +680,7 @@ mod tests { assert_eq!(workflow["timeStart"], "2026-03-04T15:16:17Z"); assert_eq!(workflow["timeEnd"], bom["metadata"]["timestamp"]); assert_eq!(workflow["resourceReferences"][0]["ref"], "tool-osmdiffs"); + assert_eq!(workflow["inputs"].as_array().expect("inputs").len(), 4); for input in workflow["inputs"].as_array().expect("inputs") { let r = input["resource"]["ref"].as_str().expect("ref"); assert!(known_refs.contains(&r), "unknown bom-ref {r}"); diff --git a/src/pipeline/qrank/mod.rs b/src/pipeline/qrank/mod.rs index 91e5c21e..1bbc3d94 100644 --- a/src/pipeline/qrank/mod.rs +++ b/src/pipeline/qrank/mod.rs @@ -14,10 +14,10 @@ //! so against all of Wikidata every brand would end up in the bottom //! quantiles. Each brand counts once, however many stores it has. //! -//! Not consumed by anything yet; see -//! [brawer/osmdiffs#713](https://github.com/brawer/osmdiffs/issues/713). +//! [`QRankQuantiles`] reads the result back, for the `importance` +//! column of conflated.parquet; see [`super::importance`]. -use crate::matchers::parse_wikidata_id; +use crate::matchers::{parse_wikidata_id, parse_wikidata_ids}; use crate::tables::{BlobTable, U64Set}; use anyhow::{Context, Result, bail}; use flate2::read::MultiGzDecoder; @@ -54,8 +54,12 @@ const QUANTILES_FILENAME: &str = "qrank-quantiles"; /// `workdir/qrank-quantiles`: a [BlobTable] keyed by numeric Wikidata ID /// (42 for Q42), each value a little-endian `f32` in `(0, 1)`. /// -/// Items that QRank doesn't list have no pageviews, and share the -/// lowest quantile. Ties get the middle of the range they share (the +/// Items that QRank doesn't list have no pageviews, and are left out +/// altogether: many brands that the Name Suggestion Index links to +/// have a Wikidata item but no Wikipedia article (Shell's fuel brand, +/// Q110716465, is one), so having no pageviews says nothing about how +/// well-known a brand is. Their features are ranked as if they had no +/// brand. Ties get the middle of the range they share (the /// "mid-rank" quantile), so the result doesn't depend on how ties /// happen to be ordered: with `less` IDs viewed less than this one, and /// `same` IDs (including this one) viewed exactly as much, out of `n`, @@ -73,7 +77,7 @@ pub fn compute_qrank_quantiles( let ids = U64Set::open(wikidata_ids)?; let (pageviews, rows) = read_pageviews(qrank, &ids)?; - let quantiles = quantiles(ids.iter().map(|id| (id, *pageviews.get(&id).unwrap_or(&0)))); + let quantiles = quantiles(pageviews.iter().map(|(&id, &views)| (id, views))); BlobTable::create( quantiles @@ -92,6 +96,43 @@ pub fn compute_qrank_quantiles( Ok(out) } +/// The table that [`compute_qrank_quantiles`] writes, opened for +/// lookups. +pub struct QRankQuantiles<'a> { + table: BlobTable<'a>, +} + +impl QRankQuantiles<'_> { + pub fn open(path: &Path) -> Result { + let table = BlobTable::open(path)?; + Ok(QRankQuantiles { table }) + } + + /// The quantile of Wikidata item `id`, or `None` if it isn't in + /// the reference population or has no pageviews. + pub fn get(&self, id: u64) -> Option { + let bytes = self.table.lookup(id)?; + Some(f32::from_le_bytes(bytes.try_into().ok()?)) + } + + /// The brand quantile of a feature with `tags`: the highest + /// quantile among the IDs in its `brand:wikidata`, + /// `operator:wikidata` and `network:wikidata` tags, or `None` if + /// it has none. The highest, because a franchise often carries both + /// a well-known brand and an obscure operator, and the brand is + /// what makes the place matter. + pub fn brand_quantile<'t>( + &self, + tags: impl IntoIterator, + ) -> Option { + tags.into_iter() + .filter(|(key, _)| super::wikidata_ids::is_wikidata_key(key)) + .flat_map(|(_, value)| parse_wikidata_ids(value)) + .filter_map(|id| self.get(id)) + .reduce(f32::max) + } +} + /// Reads the pageview count of each ID in `ids` from `qrank`, and the /// total number of data rows. IDs that QRank doesn't list are absent /// from the returned map. @@ -181,12 +222,13 @@ mod tests { // Q64 are not in our set of brands, so they must not count. write_qrank( &qrank, - "Entity,QRank\nQ64,900000\nQ42,500000\nQ37158,8000\nQ379911,300\nQ680727,300\n", + "Entity,QRank\nQ64,900000\nQ42,500000\nQ37158,8000\nQ379911,300\nQ680727,300\nQ2381223,100\n", )?; let ids_path = workdir.path().join("wikidata-ids"); - // Starbucks, Denner, Migros, and an item without pageviews. + // Starbucks, Denner, Migros, MediaMarkt, and an item without + // pageviews. U64Set::create( - [37158, 379911, 680727, 116151325].into_iter(), + [37158, 379911, 680727, 2381223, 116151325].into_iter(), workdir.path(), &ids_path, crate::pipeline::EXTERNAL_SORT_CHUNK_BYTES, @@ -195,12 +237,17 @@ mod tests { let out = compute_qrank_quantiles(&qrank, &ids_path, workdir.path())?; let table = BlobTable::open(&out)?; assert_eq!(table.len(), 4); - assert_eq!(quantile(&table, 116151325), Some(0.125)); + assert_eq!(quantile(&table, 116151325), None); + assert_eq!(quantile(&table, 2381223), Some(0.125)); assert_eq!(quantile(&table, 379911), Some(0.5)); assert_eq!(quantile(&table, 680727), Some(0.5)); assert_eq!(quantile(&table, 37158), Some(0.875)); assert_eq!(quantile(&table, 42), None); + let q = QRankQuantiles::open(&out)?; + assert_eq!(q.get(37158), Some(0.875)); + assert_eq!(q.get(42), None); + // A second call reuses the file. let modified = std::fs::metadata(&out)?.modified()?; compute_qrank_quantiles(&qrank, &ids_path, workdir.path())?; @@ -208,6 +255,41 @@ mod tests { Ok(()) } + #[test] + fn test_brand_quantile() -> Result<()> { + let workdir = TempDir::new()?; + let qrank = workdir.path().join("qrank.csv.gz"); + write_qrank(&qrank, "Entity,QRank\nQ1,30\nQ2,20\nQ3,10\nQ5,5\n")?; + let ids_path = workdir.path().join("wikidata-ids"); + U64Set::create( + [1, 2, 3, 4, 5].into_iter(), + workdir.path(), + &ids_path, + crate::pipeline::EXTERNAL_SORT_CHUNK_BYTES, + )?; + let q = QRankQuantiles::open(&compute_qrank_quantiles(&qrank, &ids_path, workdir.path())?)?; + // Q1 0.875, Q2 0.625, Q3 0.375, Q5 0.125; Q4 has no pageviews. + + // The highest of several IDs and keys wins. + let tags = [ + ("shop", "supermarket"), + ("operator:wikidata", "Q3"), + ("brand:wikidata", "Q4;Q2"), + ]; + assert_eq!(q.brand_quantile(tags), Some(0.625)); + assert_eq!(q.brand_quantile([("network:wikidata", "Q1")]), Some(0.875)); + assert_eq!(q.brand_quantile([("brand:wikidata", "Q5")]), Some(0.125)); + // Brands without pageviews count as no brand. + assert_eq!(q.brand_quantile([("brand:wikidata", "Q4")]), None); + + // Plain `wikidata` tags, unknown IDs and junk don't count. + assert_eq!(q.brand_quantile([("wikidata", "Q1")]), None); + assert_eq!(q.brand_quantile([("brand:wikidata", "Q99")]), None); + assert_eq!(q.brand_quantile([("brand:wikidata", "junk")]), None); + assert_eq!(q.brand_quantile([]), None); + Ok(()) + } + #[test] fn test_malformed_row() -> Result<()> { let workdir = TempDir::new()?; diff --git a/src/pipeline/wikidata_ids.rs b/src/pipeline/wikidata_ids.rs index d2ff879f..1c24635d 100644 --- a/src/pipeline/wikidata_ids.rs +++ b/src/pipeline/wikidata_ids.rs @@ -3,7 +3,7 @@ //! `brand:wikidata`, `operator:wikidata` and `network:wikidata` tags, //! and writes them to a [U64Set] on disk. //! -//! Not consumed by anything yet. This set is the reference population +//! This set is the reference population //! for ranking features by the importance of their brand //! ([brawer/osmdiffs#713](https://github.com/brawer/osmdiffs/issues/713)): //! a feature's brand gets the quantile of its QRank among the brands @@ -36,7 +36,7 @@ const WIKIDATA_KEYS: [&str; 3] = ["brand:wikidata", "network:wikidata", "operato /// True for tag keys whose value(s) are the Wikidata IDs of a brand, /// operator or network. -fn is_wikidata_key(key: &str) -> bool { +pub(super) fn is_wikidata_key(key: &str) -> bool { WIKIDATA_KEYS.contains(&key) } diff --git a/tests/integration_test.rs b/tests/integration_test.rs index f27101b3..97877e5c 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -292,6 +292,16 @@ fn assert_conflated_parquet(path: &Path) -> Result<()> { .as_any() .downcast_ref::() .context("'osm_geometry' is not binary")?; + let importance = batch + .column_by_name("importance") + .context("missing 'importance' column")? + .as_any() + .downcast_ref::() + .context("'importance' is not float32")?; + assert_eq!(importance.null_count(), 0); + for value in importance.values() { + assert!((0.0..=1.0).contains(value), "importance {value}"); + } for row in 0..batch.num_rows() { if osm.is_null(row) { continue;