26595150437

Committed 28 May 2026 06:47PM UTC coverage: 92.836% (-1.1%) from 93.977%

Build # 26595150437

Build Type

push

github

Committed by

web-flow

Commit Message

Suggest edits to OpenStreetMap features (#272)

On a MacBook Air 2026 with 10 Apple M5 CPU cores, it takes
42 seconds to generate 600260 edit suggestions for OpenStreetMap.

Coverage Stats

96 of 151 new or added lines in 4 files covered. (63.58%)

1 existing line in 1 file now uncovered.

3447 of 3713 relevant lines covered (92.84%)

215.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.78

/src/diff_places.rs

use crate::places::{Place, PlaceIndex, create_matcher};
use crate::s2_util::MergedCellRanges;
use crate::{make_progress_bar, match_distance};
use anyhow::Result;
use ext_sort::{ExternalSorter, ExternalSorterBuilder, buffer::mem::MemoryLimitedBufferBuilder};
use indicatif::{MultiProgress, ProgressBar};
use rayon::prelude::*;
use s2::{cap::Cap, cell::Cell, cellid::CellID, region::RegionCoverer};
use std::fs::{File, rename};
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
use std::thread;

pub fn suggest_edits(
    _coverage: &Path,
    atp: &Path,
    osm: &Path,
    progress: &MultiProgress,
    workdir: &Path,
) -> Result<PathBuf> {
    assert!(workdir.exists());

    let out_path = workdir.join("suggested-edits.jsonl");
    if out_path.exists() {
        return Ok(out_path);
    }

    let atp = PlaceIndex::open(atp, 1)?;
    let num_atp_places = atp.total_rows() as u64;
    let progress_bar = make_progress_bar(progress, "sugg-edit", num_atp_places, "ATP features");
    let osm = PlaceIndex::open(osm, 32)?; // TODO: More but smaller row groups for OSM.

    let mut producer_result = Ok(());
    let mut num_edits = Ok(0);
    thread::scope(|s| {
        let (tx, rx) = sync_channel::<Place>(8192);
        s.spawn(|| producer_result = produce_edits(atp.clone(), osm.clone(), &progress_bar, tx));
        s.spawn(|| num_edits = write_edits(rx, &out_path, workdir));
    });
    producer_result?;
    progress_bar.finish_with_message(format!("ATP features → {} suggested OSM edits", num_edits?));

    let cache_stats = osm.cache_stats();
    println!(
        "  cache hits: {} misses: {} hit rate: {:.1}%",
        cache_stats.hits,
        cache_stats.misses,
        cache_stats.hit_rate().unwrap_or(0.0) * 100.0
    );

    Ok(out_path)
}

fn produce_edits(
    atp: Arc<PlaceIndex>,
    osm: Arc<PlaceIndex>,
    progress_bar: &ProgressBar,
    out: SyncSender<Place>,
) -> Result<()> {
    let coverer = RegionCoverer {
        max_cells: 16,
        min_level: 12,
        max_level: s2::cellid::MAX_LEVEL as u8,
        level_mod: 1,
    };

    let num_atp_features = AtomicU64::new(0);
    let num_candidates = AtomicU64::new(0);
    let num_matches = AtomicU64::new(0);

    for group in atp.scan_row_groups() {
        // Each group is processed by the Rayon thread pool in parallel,
        // but the outer loop is sequential — so nearby places (within a
        // group) always go to nearby workers, preserving spatial locality.
        group?.par_iter().try_for_each(|place| {
            progress_bar.inc(1);
            if let Some(matcher) = create_matcher(place) {
                num_atp_features.fetch_add(1, Ordering::Relaxed);
                let s2_cell = Cell::from(CellID(place.s2_cell_id));
                let center = s2_cell.center();
                let radius = match_distance(&place.mask);
                let cap = Cap::from_center_chordangle(&center, &radius);
                let covering = coverer.covering(&cap);
                let mut best_candidate: Option<Place> = None;
                let mut best_score: f64 = 0.0;
                for (lo, hi) in MergedCellRanges::new(covering) {
                    let mut iter = osm.query(lo..=hi, place.mask)?;
                    let mut bc: Option<&Place> = None;
                    for candidate in &mut iter {
                        num_candidates.fetch_add(1, Ordering::Relaxed);
                        let candidate = candidate?;
                        let score = matcher.score(candidate);
                        if score > best_score {
                            bc = Some(candidate);
                            best_score = score;
                        }
                    }
                    if let Some(b) = bc {
                        best_candidate = Some(b.deep_clone());
                    }
                }
                if let Some(best_candidate) = best_candidate
                    && best_score > 0.0
                {
                    num_matches.fetch_add(1, Ordering::Relaxed);
                    if let Some(edit) = matcher.suggest_edit(&best_candidate) {
                        if false {
                            println!(
                                "score={} place={:?} best_candidate={:?} edit={:?}",
                                best_score, place, best_candidate, edit
                            );
                        }
                        out.send(edit)?;
                    }
                }
            };
            Ok::<(), anyhow::Error>(())
        })?;
    }

    Ok(())
}

fn write_edits(edits: Receiver<Place>, path: &Path, workdir: &Path) -> Result<u64> {
    let mut tmp_path = PathBuf::from(&path);
    tmp_path.add_extension("tmp");
    let mut writer = BufWriter::with_capacity(32768, File::create(&tmp_path)?);

    let sorter: ExternalSorter<Place, std::io::Error, MemoryLimitedBufferBuilder> =
        ExternalSorterBuilder::new()
            .with_tmp_dir(workdir)
            .with_buffer(MemoryLimitedBufferBuilder::new(150_000_000))
            .build()?;

    let num_edits = AtomicU64::new(0);
    let sorted = sorter.sort(edits.iter().map(|x| {
        num_edits.fetch_add(1, Ordering::Relaxed);
        std::io::Result::Ok(x)
    }))?;
    let mut last_osm_id = 0;
    for edit in sorted {
        let edit = edit?;
        // Only emit one single edit per OSM ID.
        if edit.osm_id == last_osm_id {
            continue;
        }
        last_osm_id = edit.osm_id;
        let mut line = edit.to_geojson().to_string();
        line.push('\n');
        writer.write_all(line.as_ref())?;
    }
    writer.flush()?;
    rename(&tmp_path, path)?;

    Ok(num_edits.load(Ordering::SeqCst))
}

1	use crate::places::{Place, PlaceIndex, create_matcher};
2	use crate::s2_util::MergedCellRanges;
3	use crate::{make_progress_bar, match_distance};
4	use anyhow::Result;
5	use ext_sort::{ExternalSorter, ExternalSorterBuilder, buffer::mem::MemoryLimitedBufferBuilder};
6	use indicatif::{MultiProgress, ProgressBar};
7	use rayon::prelude::*;
8	use s2::{cap::Cap, cell::Cell, cellid::CellID, region::RegionCoverer};
9	use std::fs::{File, rename};
10	use std::io::{BufWriter, Write};
11	use std::path::{Path, PathBuf};
12	use std::sync::Arc;
13	use std::sync::atomic::{AtomicU64, Ordering};
14	use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
15	use std::thread;
16
17	pub fn suggest_edits(	1✔
18	_coverage: &Path,	1✔
19	atp: &Path,	1✔
20	osm: &Path,	1✔
21	progress: &MultiProgress,	1✔
22	workdir: &Path,	1✔
23	) -> Result<PathBuf> {	1✔
24	assert!(workdir.exists());	1✔
25
26	let out_path = workdir.join("suggested-edits.jsonl");	1✔
27	if out_path.exists() {	1✔
28	return Ok(out_path);	×
29	}	1✔
30
31	let atp = PlaceIndex::open(atp, 1)?;	1✔
32	let num_atp_places = atp.total_rows() as u64;	1✔
33	let progress_bar = make_progress_bar(progress, "sugg-edit", num_atp_places, "ATP features");	1✔
34	let osm = PlaceIndex::open(osm, 32)?; // TODO: More but smaller row groups for OSM.	1✔
35
36	let mut producer_result = Ok(());	1✔
37	let mut num_edits = Ok(0);	1✔
38	thread::scope(\|s\| {	1✔
39	let (tx, rx) = sync_channel::<Place>(8192);	1✔
40	s.spawn(\|\| producer_result = produce_edits(atp.clone(), osm.clone(), &progress_bar, tx));	1✔
41	s.spawn(\|\| num_edits = write_edits(rx, &out_path, workdir));	1✔
42	});	1✔
43	producer_result?;	1✔
44	progress_bar.finish_with_message(format!("ATP features → {} suggested OSM edits", num_edits?));	1✔
45
46	let cache_stats = osm.cache_stats();	1✔
47	println!(	1✔
48	" cache hits: {} misses: {} hit rate: {:.1}%",
49	cache_stats.hits,
50	cache_stats.misses,
51	cache_stats.hit_rate().unwrap_or(0.0) * 100.0	1✔
52	);
53
54	Ok(out_path)	1✔
55	}	1✔
56
57	fn produce_edits(	1✔
58	atp: Arc<PlaceIndex>,	1✔
59	osm: Arc<PlaceIndex>,	1✔
60	progress_bar: &ProgressBar,	1✔
61	out: SyncSender<Place>,	1✔
62	) -> Result<()> {	1✔
63	let coverer = RegionCoverer {	1✔
64	max_cells: 16,	1✔
65	min_level: 12,	1✔
66	max_level: s2::cellid::MAX_LEVEL as u8,	1✔
67	level_mod: 1,	1✔
68	};	1✔
69
70	let num_atp_features = AtomicU64::new(0);	1✔
71	let num_candidates = AtomicU64::new(0);	1✔
72	let num_matches = AtomicU64::new(0);	1✔
73
74	for group in atp.scan_row_groups() {	1✔
75	// Each group is processed by the Rayon thread pool in parallel,
76	// but the outer loop is sequential — so nearby places (within a
77	// group) always go to nearby workers, preserving spatial locality.
78	group?.par_iter().try_for_each(\|place\| {	7✔
79	progress_bar.inc(1);	7✔
80	if let Some(matcher) = create_matcher(place) {	7✔
81	num_atp_features.fetch_add(1, Ordering::Relaxed);	3✔
82	let s2_cell = Cell::from(CellID(place.s2_cell_id));	3✔
83	let center = s2_cell.center();	3✔
84	let radius = match_distance(&place.mask);	3✔
85	let cap = Cap::from_center_chordangle(&center, &radius);	3✔
86	let covering = coverer.covering(&cap);	3✔
87	let mut best_candidate: Option<Place> = None;	3✔
88	let mut best_score: f64 = 0.0;	3✔
89	for (lo, hi) in MergedCellRanges::new(covering) {	42✔
90	let mut iter = osm.query(lo..=hi, place.mask)?;	42✔
91	let mut bc: Option<&Place> = None;	42✔
92	for candidate in &mut iter {	51✔
93	num_candidates.fetch_add(1, Ordering::Relaxed);	51✔
94	let candidate = candidate?;	51✔
95	let score = matcher.score(candidate);	51✔
96	if score > best_score {	51✔
97	bc = Some(candidate);	×
98	best_score = score;	×
99	}	51✔
100	}
101	if let Some(b) = bc {	42✔
102	best_candidate = Some(b.deep_clone());	×
103	}	42✔
104	}
105	if let Some(best_candidate) = best_candidate	3✔
106	&& best_score > 0.0	×
107	{
108	num_matches.fetch_add(1, Ordering::Relaxed);	×
NEW 109	if let Some(edit) = matcher.suggest_edit(&best_candidate) {	×
NEW 110	if false {	×
NEW 111	println!(	×
NEW 112	"score={} place={:?} best_candidate={:?} edit={:?}",	×
NEW 113	best_score, place, best_candidate, edit	×
NEW 114	);	×
NEW 115	}	×
NEW 116	out.send(edit)?;	×
UNCOV 117	}	×
118	}	3✔
119	};	4✔
120	Ok::<(), anyhow::Error>(())	7✔
121	})?;	7✔
122	}
123
124	Ok(())	1✔
125	}	1✔
126
127	fn write_edits(edits: Receiver<Place>, path: &Path, workdir: &Path) -> Result<u64> {	1✔
128	let mut tmp_path = PathBuf::from(&path);	1✔
129	tmp_path.add_extension("tmp");	1✔
130	let mut writer = BufWriter::with_capacity(32768, File::create(&tmp_path)?);	1✔
131
132	let sorter: ExternalSorter<Place, std::io::Error, MemoryLimitedBufferBuilder> =	1✔
133	ExternalSorterBuilder::new()	1✔
134	.with_tmp_dir(workdir)	1✔
135	.with_buffer(MemoryLimitedBufferBuilder::new(150_000_000))	1✔
136	.build()?;	1✔
137
138	let num_edits = AtomicU64::new(0);	1✔
139	let sorted = sorter.sort(edits.iter().map(\|x\| {	1✔
NEW 140	num_edits.fetch_add(1, Ordering::Relaxed);	×
NEW 141	std::io::Result::Ok(x)	×
NEW 142	}))?;	×
143	let mut last_osm_id = 0;	1✔
144	for edit in sorted {	1✔
NEW 145	let edit = edit?;	×
146	// Only emit one single edit per OSM ID.
NEW 147	if edit.osm_id == last_osm_id {	×
NEW 148	continue;	×
NEW 149	}	×
NEW 150	last_osm_id = edit.osm_id;	×
NEW 151	let mut line = edit.to_geojson().to_string();	×
NEW 152	line.push('\n');	×
NEW 153	writer.write_all(line.as_ref())?;	×
154	}
155	writer.flush()?;	1✔
156	rename(&tmp_path, path)?;	1✔
157
158	Ok(num_edits.load(Ordering::SeqCst))	1✔
159	}	1✔

diffed-places / pipeline / 26595150437

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous