benches/collectors_parse.rs
//! **Per-collector parse/normalize throughput** — criterion benches over the
//! offline-safe sample source. Each bench pulls its input from
//! `sample::*_try_get_real_sample_data()` (a short-timeout live fetch that FALLS
//! BACK to an embedded synthetic fixture) so `cargo bench` runs with **no
//! network**. Set `KORP_COLLECTORS_OFFLINE=1` for a deterministic offline run.
//!
//! Run: `cargo bench --bench collectors_parse` (add `--features osm` for the OSM
//! ways bench).
use criterion::{Criterion, Throughput, criterion_group, criterion_main};
use std::hint::black_box;
use korp_collectors::sample::{self, Sample};
/// Bench `parse` over the sample body, sized by bytes for a MB/s figure. The
/// sample is fetched ONCE (outside the timed loop) so network latency never
/// pollutes the parse measurement.
fn bench_parse<F>(c: &mut Criterion, id: &str, sample: Sample, parse: F)
where
F: Fn(&[u8]),
{
let mut group = c.benchmark_group("collectors_parse");
group.throughput(Throughput::Bytes(sample.len() as u64));
let tag = if sample.is_synthetic() {
"synthetic"
} else {
"real"
};
group.bench_function(format!("{id}[{tag}]"), |b| {
b.iter(|| parse(black_box(&sample.body)))
});
group.finish();
}
fn dcat(c: &mut Criterion) {
bench_parse(c, "dcat", sample::dcat_try_get_real_sample_data(), |b| {
let rows = korp_collectors::dcat::parse_dcat_rdfxml(b).expect("parse");
black_box(korp_collectors::dcat::registry_batch(&rows).expect("batch"));
});
}
fn pxweb(c: &mut Criterion) {
bench_parse(c, "pxweb", sample::pxweb_try_get_real_sample_data(), |b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
black_box(korp_collectors::pxweb::parse_jsonstat2(&v).expect("cube"));
});
}
fn eurostat_sdmx(c: &mut Criterion) {
bench_parse(
c,
"eurostat-sdmx",
sample::eurostat_try_get_real_sample_data(),
|b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
black_box(korp_collectors::sdmx::parse_sdmx_json(&v).expect("sdmx"));
},
);
}
fn entsoe(c: &mut Criterion) {
bench_parse(
c,
"entsoe",
sample::entsoe_try_get_real_sample_data(),
|b| {
let xml = String::from_utf8_lossy(b);
let pts = korp_collectors::entsoe::parse_entsoe_xml(&xml).expect("xml");
black_box(korp_collectors::entsoe::points_batch(&pts).expect("batch"));
},
);
}
/// A REST preset bench: parse JSON → Arrow via the preset's pointer + field map.
fn rest_bench(
c: &mut Criterion,
id: &str,
collector: korp_collectors::rest::RestCollector,
sample: Sample,
) {
let pointer = collector.records_pointer().to_string();
let fields = collector.fields().to_vec();
bench_parse(c, id, sample, move |b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
black_box(korp_collectors::rest::parse_records(&v, &pointer, &fields).expect("map"));
});
}
fn polisen(c: &mut Criterion) {
rest_bench(
c,
"rest-polisen",
korp_collectors::rest::RestCollector::polisen(),
sample::polisen_try_get_real_sample_data(),
);
}
fn kolada(c: &mut Criterion) {
rest_bench(
c,
"rest-kolada",
korp_collectors::rest::RestCollector::kolada(),
sample::kolada_try_get_real_sample_data(),
);
}
fn smhi(c: &mut Criterion) {
rest_bench(
c,
"rest-smhi",
korp_collectors::rest::RestCollector::smhi_metobs(),
sample::smhi_try_get_real_sample_data(),
);
}
fn ais(c: &mut Criterion) {
rest_bench(
c,
"rest-digitraffic-ais",
korp_collectors::rest::RestCollector::digitraffic_ais(),
sample::ais_try_get_real_sample_data(),
);
}
fn wfs(c: &mut Criterion) {
bench_parse(c, "wfs", sample::wfs_try_get_real_sample_data(), |b| {
let xml = String::from_utf8_lossy(b);
let feats = korp_collectors::wfs::parse_wfs_gml(&xml).expect("gml");
black_box(korp_collectors::wfs::features_batch(&feats).expect("batch"));
});
}
fn nordpool(c: &mut Criterion) {
bench_parse(
c,
"nordpool",
sample::nordpool_try_get_real_sample_data(),
|b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
let prices = korp_collectors::nordpool::parse_nordpool_json(&v).expect("prices");
black_box(korp_collectors::nordpool::prices_batch(&prices).expect("batch"));
},
);
}
fn energy_charts(c: &mut Criterion) {
bench_parse(
c,
"energy-charts",
sample::energy_charts_try_get_real_sample_data(),
|b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
let readings =
korp_collectors::energy_charts::parse_energy_charts_json(&v).expect("power");
black_box(korp_collectors::energy_charts::power_batch(&readings).expect("batch"));
},
);
}
fn gtfs_rt(c: &mut Criterion) {
bench_parse(
c,
"gtfs-rt",
sample::gtfs_rt_try_get_real_sample_data(),
|b| {
let rows = korp_collectors::gtfs_rt::parse_gtfs_rt(b).expect("gtfs-rt");
black_box(korp_collectors::gtfs_rt::vehicles_batch(&rows).expect("batch"));
},
);
}
fn usgs(c: &mut Criterion) {
bench_parse(c, "usgs", sample::usgs_try_get_real_sample_data(), |b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
let quakes = korp_collectors::usgs::parse_usgs_geojson(&v).expect("quakes");
black_box(korp_collectors::usgs::quakes_batch(&quakes).expect("batch"));
});
}
fn rss(c: &mut Criterion) {
bench_parse(c, "rss-atom", sample::rss_try_get_real_sample_data(), |b| {
let xml = String::from_utf8_lossy(b);
let items = korp_collectors::rss::parse_feed(&xml).expect("feed");
black_box(korp_collectors::rss::feed_batch(&items).expect("batch"));
});
}
fn bolagsverket(c: &mut Criterion) {
rest_bench(
c,
"rest-bolagsverket",
korp_collectors::rest::RestCollector::bolagsverket(),
sample::bolagsverket_try_get_real_sample_data(),
);
}
fn trafikverket(c: &mut Criterion) {
let collector = korp_collectors::trafikverket::TrafikverketCollector::train_positions();
let pointer = collector.records_pointer();
let fields = collector.fields().to_vec();
bench_parse(
c,
"trafikverket",
sample::trafikverket_try_get_real_sample_data(),
move |b| {
let v: serde_json::Value = serde_json::from_slice(b).expect("json");
black_box(korp_collectors::rest::parse_records(&v, &pointer, &fields).expect("map"));
},
);
}
fn osm_admin(c: &mut Criterion) {
bench_parse(
c,
"osm_admin",
sample::osm_admin_try_get_real_sample_data(),
|b| {
let features = sample::parse_admin_features(b);
let units = korp_collectors::osm_admin::extract_admin_units(features);
black_box(korp_collectors::osm_admin::admin_batch(&units).expect("batch"));
},
);
}
#[cfg(feature = "gtfs")]
fn gtfs(c: &mut Criterion) {
let collector = korp_collectors::gtfs::GtfsCollector::stops();
let cols = collector.cols().to_vec();
bench_parse(
c,
"gtfs-stops",
sample::gtfs_try_get_real_sample_data(),
move |b| {
black_box(korp_collectors::gtfs::parse_gtfs_zip(b, "stops.txt", &cols).expect("gtfs"));
},
);
}
#[cfg(feature = "osm")]
fn osm_ways(c: &mut Criterion) {
// osm-katana reads a path — stage the sample bytes once, bench the parse.
let sample = sample::osm_ways_try_get_real_sample_data();
let dir = std::env::temp_dir();
let path = dir.join(format!("korp_bench_ways_{}.osm.pbf", std::process::id()));
std::fs::write(&path, &sample.body).unwrap();
let mut group = c.benchmark_group("collectors_parse");
group.throughput(Throughput::Bytes(sample.len() as u64));
let tag = if sample.is_synthetic() {
"synthetic"
} else {
"real"
};
let p = path.to_str().unwrap().to_string();
group.bench_function(format!("osm[{tag}]"), |b| {
b.iter(|| {
let ways = korp_collectors::osm::load_ways_pbf(black_box(&p)).expect("parse");
black_box(korp_collectors::osm::ways_to_batch(&ways).expect("batch"));
})
});
group.finish();
let _ = std::fs::remove_file(&path);
}
// The always-present core parsers (light, no heavy geo / no extra feature dep).
/// **The whole-registry sample sweep** — resolve the offline-safe sample for
/// EVERY collector in `registry()` through `sample::for_collector`, sized by the
/// total bytes handed back.
///
/// This is the seam every test and every other bench arm in this file now pulls
/// from, and it is the only surface in the crate whose cost scales with the
/// collector count: in offline mode it BUILDS each fixture (a DEFLATE GTFS feed
/// zip, an IBM-037 EBCDIC encode, the synthetic GRIB2/NetCDF containers), it does
/// not just hand back a `&'static str`. Adding a collector adds work here, so the
/// arm tracks the discovery surface rather than a frozen list.
///
/// Deterministic: forces `KORP_COLLECTORS_OFFLINE=1` so no live fetch ever lands
/// inside the timed loop.
fn registry_sample_sweep(c: &mut Criterion) {
// SAFETY: benches are single-threaded at setup; this runs before any sample
// resolution and only ever sets the offline flag.
unsafe { std::env::set_var("KORP_COLLECTORS_OFFLINE", "1") };
let names: Vec<String> = korp_collectors::registry()
.iter()
.map(|c| c.meta().name.clone())
.collect();
let total_bytes: u64 = names
.iter()
.filter_map(|n| sample::for_collector(n))
.map(|s| s.len() as u64)
.sum();
let mut group = c.benchmark_group("collectors_registry");
group.throughput(Throughput::Bytes(total_bytes));
group.bench_function(format!("sample_sweep[{}_collectors]", names.len()), |b| {
b.iter(|| {
let mut bytes = 0usize;
for n in &names {
if let Some(s) = sample::for_collector(black_box(n)) {
bytes += s.len();
}
}
black_box(bytes)
})
});
group.finish();
}
criterion_group!(
core,
registry_sample_sweep,
dcat,
pxweb,
eurostat_sdmx,
entsoe,
polisen,
kolada,
smhi,
ais,
wfs,
nordpool,
energy_charts,
usgs,
gtfs_rt,
rss,
bolagsverket,
trafikverket,
osm_admin
);
// Optional feature groups — each defined only when its feature is on, and folded
// into `criterion_main!` per feature combination (keeps the core list stable).
#[cfg(feature = "osm")]
criterion_group!(osm_group, osm_ways);
#[cfg(feature = "gtfs")]
criterion_group!(gtfs_group, gtfs);
#[cfg(all(feature = "osm", feature = "gtfs"))]
criterion_main!(core, osm_group, gtfs_group);
#[cfg(all(feature = "osm", not(feature = "gtfs")))]
criterion_main!(core, osm_group);
#[cfg(all(not(feature = "osm"), feature = "gtfs"))]
criterion_main!(core, gtfs_group);
#[cfg(all(not(feature = "osm"), not(feature = "gtfs")))]
criterion_main!(core);