feat: add importance column to conflated.parquet

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

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

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;

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<PathBuf> {
     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::<ParquetRow>(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<Place>,
     osm: Option<Feature>,
-    // 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<ParquetRow>,
 ) -> 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()?;

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<UtcTimestamp>,
     atp_tags: Vec<(String, String)>,
     atp_shape_wkb: Vec<u8>,
+
+    /// 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<ParquetRow> {
         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,
         }
     }