src/sdmx.rs
//! **Eurostat SDMX-JSON 2.1** — the flagship EU statistics feed. Eurostat's SDMX
//! 2.1 REST endpoint (`format=jsondata`) serves an **SDMX-JSON** message whose
//! shape is genuinely distinct from the PxWeb / JSON-stat2 cube ([`pxweb`]): the
//! dimensions are split into a **series** set and an **observation** set, and the
//! data is a two-level sparse map —
//! `dataSets[0].series["s0:s1:…"].observations["o0:o1:…"] = [value, …attrs]` —
//! where each key is a colon-joined list of POSITIONS into that dimension's
//! `values` array. That series/observation split + colon-key decode is why this
//! needs its own flattener rather than the JSON-stat2 normaliser or the generic
//! REST field-map adapter.
//!
//! [`parse_sdmx_json`] is the offline contract: a canned SDMX-JSON message → an
//! Arrow `RecordBatch` with one Utf8 code column per dimension (series dims first,
//! then observation dims, each named by the dimension id) + a trailing nullable
//! Float64 `value` column. Rows are emitted in a deterministic
//! dimension-position order (independent of JSON map iteration order).
//!
//! [`pxweb`]: crate::pxweb
use anyhow::{Result, anyhow};
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Field, Schema};
use serde_json::Value;
use super::arrow_util::{f64_col, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// One decoded SDMX observation: the resolved `(dimension id, code)` pairs (series
/// dims then observation dims, in declared order) + the numeric value (nullable —
/// SDMX marks a missing observation with `null`/absent).
#[derive(Debug, Clone, PartialEq)]
pub struct SdmxObs {
/// `(dimension_id, code)` in canonical order (series dims, then obs dims).
pub dims: Vec<(String, String)>,
/// The observation value (first element of the SDMX value array).
pub value: Option<f64>,
}
/// A dimension descriptor read from the message structure: its `id` and the codes
/// of its `values` in position order (so position `p` → `codes[p]`).
struct Dim {
id: String,
codes: Vec<String>,
}
/// Read a dimension list (`structure.dimensions.series` / `.observation`) into
/// `(id, codes)` descriptors. A missing/empty list yields no dimensions.
fn read_dims(node: Option<&Value>) -> Vec<Dim> {
let Some(arr) = node.and_then(Value::as_array) else {
return Vec::new();
};
arr.iter()
.map(|d| {
let id = d
.get("id")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let codes = d
.get("values")
.and_then(Value::as_array)
.map(|vs| {
vs.iter()
.map(|v| {
v.get("id")
.and_then(Value::as_str)
.unwrap_or("")
.to_string()
})
.collect()
})
.unwrap_or_default();
Dim { id, codes }
})
.collect()
}
/// Parse a colon-joined SDMX key (`"0:1:2"`) into position indices. An empty key
/// (a lone-dimension observation or a dimensionless dataset) is `[]`.
fn parse_key(key: &str) -> Option<Vec<usize>> {
if key.is_empty() {
return Some(Vec::new());
}
key.split(':').map(|p| p.parse::<usize>().ok()).collect()
}
/// Coerce an SDMX observation value (JSON number, or occasionally a quoted number)
/// to `f64`; anything else (incl. `null`) is a missing value.
fn obs_value(arr: &Value) -> Option<f64> {
let first = arr.as_array().and_then(|a| a.first())?;
first
.as_f64()
.or_else(|| first.as_str().and_then(|s| s.trim().parse().ok()))
}
/// Locate the structure node (`data.structures[0]` for SDMX-JSON 2.1, or the
/// singular `structure` of SDMX-JSON 1.0) and the first dataSet.
fn structure_and_dataset(v: &Value) -> Result<(&Value, &Value)> {
let structure = v
.pointer("/data/structures/0")
.or_else(|| v.get("structure"))
.ok_or_else(|| anyhow!("SDMX-JSON: missing structure (data.structures[0] / structure)"))?;
let dataset = v
.pointer("/data/dataSets/0")
.or_else(|| v.get("dataSets").and_then(|d| d.get(0)))
.ok_or_else(|| anyhow!("SDMX-JSON: missing dataSets[0]"))?;
Ok((structure, dataset))
}
/// Flatten an SDMX-JSON message → decoded observations (series × observation
/// dimensions resolved to codes). Deterministic order: sorted by the full
/// dimension-position vector (series positions, then observation positions).
pub fn parse_sdmx_obs(v: &Value) -> Result<Vec<SdmxObs>> {
let (structure, dataset) = structure_and_dataset(v)?;
let dims = structure.get("dimensions");
let series_dims = read_dims(dims.and_then(|d| d.get("series")));
let obs_dims = read_dims(dims.and_then(|d| d.get("observation")));
// (full position vector, resolved obs) — sorted before the codes are read so
// the row order is independent of serde_json's map backend.
let mut rows: Vec<(Vec<usize>, SdmxObs)> = Vec::new();
let resolve = |sidx: &[usize], oidx: &[usize], value: Option<f64>| -> Option<SdmxObs> {
let mut pairs = Vec::with_capacity(series_dims.len() + obs_dims.len());
for (d, &p) in series_dims.iter().zip(sidx) {
pairs.push((d.id.clone(), d.codes.get(p).cloned().unwrap_or_default()));
}
for (d, &p) in obs_dims.iter().zip(oidx) {
pairs.push((d.id.clone(), d.codes.get(p).cloned().unwrap_or_default()));
}
// A key that names fewer positions than there are dimensions is malformed.
if sidx.len() != series_dims.len() || oidx.len() != obs_dims.len() {
return None;
}
Some(SdmxObs { dims: pairs, value })
};
if let Some(series) = dataset.get("series").and_then(Value::as_object) {
// Nested shape: series[skey].observations[okey] = [value, …].
for (skey, sval) in series {
let Some(sidx) = parse_key(skey) else {
continue;
};
let Some(obs) = sval.get("observations").and_then(Value::as_object) else {
continue;
};
for (okey, arr) in obs {
let Some(oidx) = parse_key(okey) else {
continue;
};
if let Some(row) = resolve(&sidx, &oidx, obs_value(arr)) {
let mut full = sidx.clone();
full.extend_from_slice(&oidx);
rows.push((full, row));
}
}
}
} else if let Some(obs) = dataset.get("observations").and_then(Value::as_object) {
// Flat shape: no series dims — observations sit directly on the dataSet.
for (okey, arr) in obs {
let Some(oidx) = parse_key(okey) else {
continue;
};
if let Some(row) = resolve(&[], &oidx, obs_value(arr)) {
rows.push((oidx.clone(), row));
}
}
}
rows.sort_by(|a, b| a.0.cmp(&b.0));
Ok(rows.into_iter().map(|(_, r)| r).collect())
}
/// The canonical column order (dimension ids, series then observation) for a
/// parsed message — the schema the batch carries, ahead of the `value` column.
fn dim_order(v: &Value) -> Result<Vec<String>> {
let (structure, _) = structure_and_dataset(v)?;
let dims = structure.get("dimensions");
let mut order: Vec<String> = read_dims(dims.and_then(|d| d.get("series")))
.into_iter()
.map(|d| d.id)
.collect();
order.extend(
read_dims(dims.and_then(|d| d.get("observation")))
.into_iter()
.map(|d| d.id),
);
Ok(order)
}
/// Normalise an SDMX-JSON message directly into one Arrow `RecordBatch`: a Utf8
/// code column per dimension (in [`dim_order`]) + a nullable Float64 `value`.
pub fn parse_sdmx_json(v: &Value) -> Result<RecordBatch> {
let order = dim_order(v)?;
let obs = parse_sdmx_obs(v)?;
sdmx_batch(&order, &obs)
}
/// Build the Arrow batch from a dimension order + decoded observations. Each
/// observation is expected to carry exactly the `order` dimensions (the parser
/// guarantees this); a missing dim resolves to an empty code.
pub fn sdmx_batch(order: &[String], obs: &[SdmxObs]) -> Result<RecordBatch> {
let mut fields: Vec<Field> = order
.iter()
.map(|id| Field::new(id, DataType::Utf8, false))
.collect();
fields.push(Field::new("value", DataType::Float64, true));
let schema = std::sync::Arc::new(Schema::new(fields));
let mut code_cols: Vec<Vec<Option<String>>> = vec![Vec::with_capacity(obs.len()); order.len()];
let mut values: Vec<Option<f64>> = Vec::with_capacity(obs.len());
for o in obs {
for (j, id) in order.iter().enumerate() {
let code = o.dims.iter().find(|(d, _)| d == id).map(|(_, c)| c.clone());
code_cols[j].push(Some(code.unwrap_or_default()));
}
values.push(o.value);
}
let mut cols: Vec<arrow_array::ArrayRef> = code_cols.iter().map(|c| str_col(c)).collect();
cols.push(f64_col(&values));
Ok(RecordBatch::try_new(schema, cols)?)
}
/// A provider descriptor for the generalised [`SdmxCollector`] — only the
/// branding + representative endpoint differ between SDMX providers; the
/// SDMX-JSON flattener ([`parse_sdmx_json`]) is shared. This is what lets ONE
/// collector cover Eurostat, OECD, ECB, IMF, INSEE and EEA with no new parsing.
#[derive(Debug, Clone)]
pub struct SdmxProvider {
/// Collector name (`"eurostat-sdmx"`, `"oecd-sdmx"`, …).
pub name: String,
/// Representative SDMX REST data endpoint (documentation + descriptor).
pub endpoint: String,
/// The warehouse table the observations land in.
pub table: String,
/// License string for the provider.
pub license: String,
/// Human description for the discovery record.
pub description: String,
/// The typed source endpoint (Eurostat has its own; others ride GenericRest).
pub source: SourceEndpoint,
/// Representative join keys for entity resolution.
pub join_keys: Vec<String>,
}
/// The **generalised SDMX-JSON statistics collector** — config-driven over any
/// SDMX 2.1 (`data.structures[0]`) or SDMX-JSON 1.0 (`structure`) provider. The
/// `base_url` carries the concrete REST data query
/// (`.../data/<flow>/<key>?format=jsondata`); the shared [`parse_sdmx_json`]
/// flattens the series/observation split to `(dims…, value)` rows with ZERO new
/// parsing per provider. Construct with a preset ([`eurostat`](Self::eurostat),
/// [`oecd`](Self::oecd), [`ecb`](Self::ecb)) or [`generic`](Self::generic).
pub struct SdmxCollector {
provider: SdmxProvider,
}
impl Default for SdmxCollector {
fn default() -> Self {
Self::eurostat()
}
}
impl SdmxCollector {
/// A generic dimension-column schema for `schema()` before a message is seen:
/// use `params.dims` when declared, else a representative cube shape.
fn cube_schema(cfg: &CollectorCfg) -> Schema {
let dims: Vec<String> = cfg
.params
.get("dims")
.and_then(|d| d.as_array())
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.unwrap_or_else(|| vec!["freq".into(), "geo".into(), "TIME_PERIOD".into()]);
let mut fields: Vec<Field> = dims
.iter()
.map(|d| Field::new(d, DataType::Utf8, false))
.collect();
fields.push(Field::new("value", DataType::Float64, true));
Schema::new(fields)
}
/// Construct from a raw provider descriptor.
pub fn new(provider: SdmxProvider) -> Self {
Self { provider }
}
/// **Eurostat** — the flagship EU statistics SDMX 2.1 endpoint.
pub fn eurostat() -> Self {
Self::new(SdmxProvider {
name: "eurostat-sdmx".into(),
endpoint: "https://ec.europa.eu/eurostat/api/dissemination/sdmx/2.1/data".into(),
table: "eurostat_sdmx".into(),
license: "open (Eurostat © European Union, free reuse with attribution)".into(),
description: "Eurostat SDMX-JSON 2.1 statistics — series/observation split \
flattened to one row per (series dims × observation dims) with the \
numeric value; EU-wide cubes keyed by NUTS geo + TIME_PERIOD."
.into(),
source: SourceEndpoint::Eurostat,
join_keys: vec!["geo".into(), "TIME_PERIOD".into()],
})
}
/// **OECD** — the OECD SDMX 2.1 public data API (economic/social indicators).
pub fn oecd() -> Self {
Self::new(SdmxProvider {
name: "oecd-sdmx".into(),
endpoint: "https://sdmx.oecd.org/public/rest/data".into(),
table: "oecd_sdmx".into(),
license: "OECD Terms (open, attribution)".into(),
description: "OECD SDMX 2.1 statistics — same series/observation flattener as \
Eurostat, pointed at the OECD public REST data API (economic + \
social indicators)."
.into(),
source: SourceEndpoint::GenericRest,
join_keys: vec!["REF_AREA".into(), "TIME_PERIOD".into()],
})
}
/// **ECB** — the European Central Bank SDMX data API (SDMX-JSON 1.0 shape).
pub fn ecb() -> Self {
Self::new(SdmxProvider {
name: "ecb-sdmx".into(),
endpoint: "https://data-api.ecb.europa.eu/service/data".into(),
table: "ecb_sdmx".into(),
license: "open (ECB, free reuse with attribution)".into(),
description: "ECB Statistical Data Warehouse SDMX — exchange rates, monetary + \
financial series, flattened by the shared SDMX-JSON parser (handles \
the ECB `structure` 1.0 shape as well as 2.1)."
.into(),
source: SourceEndpoint::GenericRest,
join_keys: vec!["CURRENCY".into(), "TIME_PERIOD".into()],
})
}
/// **INSEE** (France) — the Banque de Données Macro-économiques (BDM) SDMX 2.1
/// web service. The public `bdm.insee.fr/series/sdmx/data` endpoint serves the
/// same series/observation SDMX-JSON the shared flattener handles (zero new
/// parsing); French national accounts / prices / demography cubes keyed by a
/// BDM series id + `TIME_PERIOD`.
pub fn insee() -> Self {
Self::new(SdmxProvider {
name: "insee-sdmx".into(),
endpoint: "https://bdm.insee.fr/series/sdmx/data".into(),
table: "insee_sdmx".into(),
license: "open (INSEE / Licence Ouverte 2.0, free reuse with attribution)".into(),
description: "INSEE BDM SDMX 2.1 statistics (France) — same series/observation \
flattener as Eurostat, pointed at the INSEE macro-economic database \
(national accounts, prices, demography)."
.into(),
source: SourceEndpoint::Insee,
join_keys: vec!["REF_AREA".into(), "TIME_PERIOD".into()],
})
}
/// **ONS** (United Kingdom) — the Office for National Statistics Beta SDMX API.
/// Its SDMX 2.1 REST data endpoint serves the same SDMX-JSON the shared
/// flattener handles (zero new parsing); UK statistics cubes keyed by
/// geography + `TIME_PERIOD`.
pub fn ons() -> Self {
Self::new(SdmxProvider {
name: "ons-sdmx".into(),
endpoint: "https://api.beta.ons.gov.uk/v1/data".into(),
table: "ons_sdmx".into(),
license: "open (ONS / Open Government Licence v3.0)".into(),
description: "ONS Beta SDMX statistics (United Kingdom) — same \
series/observation flattener as Eurostat, pointed at the ONS Beta \
SDMX API."
.into(),
source: SourceEndpoint::Ons,
join_keys: vec!["Geography".into(), "TIME_PERIOD".into()],
})
}
/// **ISTAT** (Italy) — the Istituto Nazionale di Statistica SDMX 2.1 web
/// service (`esploradati.istat.it/SDMXWS/rest/data`). Serves the same SDMX-JSON
/// the shared flattener handles (zero new parsing); Italian statistics cubes
/// keyed by `REF_AREA` + `TIME_PERIOD`.
pub fn istat() -> Self {
Self::new(SdmxProvider {
name: "istat-sdmx".into(),
endpoint: "https://esploradati.istat.it/SDMXWS/rest/data".into(),
table: "istat_sdmx".into(),
license: "open (ISTAT / CC-BY 4.0)".into(),
description: "ISTAT SDMX 2.1 statistics (Italy) — same series/observation \
flattener as Eurostat, pointed at the ISTAT SDMX web service."
.into(),
source: SourceEndpoint::Istat,
join_keys: vec!["REF_AREA".into(), "TIME_PERIOD".into()],
})
}
/// **World Bank** — the World Bank SDMX REST data API
/// (`api.worldbank.org/v2/sdmx/rest/data`). The SDMX face of the World Bank
/// indicators (distinct from the plain-REST `worldbank-indicators` preset):
/// same series/observation flattener as Eurostat, zero new parsing.
pub fn worldbank() -> Self {
Self::new(SdmxProvider {
name: "worldbank-sdmx".into(),
endpoint: "https://api.worldbank.org/v2/sdmx/rest/data".into(),
table: "worldbank_sdmx".into(),
license: "CC-BY 4.0 (World Bank Open Data)".into(),
description: "World Bank SDMX statistics — the SDMX face of the World Bank \
indicators, flattened by the shared SDMX-JSON parser (REF_AREA × \
TIME_PERIOD)."
.into(),
source: SourceEndpoint::WorldBankSdmx,
join_keys: vec!["REF_AREA".into(), "TIME_PERIOD".into()],
})
}
/// A fully generic SDMX provider (the escape hatch: name + REST data endpoint).
pub fn generic(name: &str, endpoint: &str) -> Self {
Self::new(SdmxProvider {
name: name.into(),
endpoint: endpoint.into(),
table: name.replace('-', "_"),
license: "per-provider".into(),
description: format!("Generic SDMX-JSON statistics provider `{name}`."),
source: SourceEndpoint::GenericRest,
join_keys: vec!["TIME_PERIOD".into()],
})
}
}
/// Back-compat alias for the pre-generalisation type name — `SdmxCollector`
/// defaults to the Eurostat preset, so existing `EurostatSdmxCollector::default()`
/// / `EurostatSdmxCollector::eurostat()` call sites keep working.
pub type EurostatSdmxCollector = SdmxCollector;
#[async_trait::async_trait]
impl Collector for SdmxCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.provider.name.clone(),
source_kind: SourceKind::Sdmx,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["sdmx-json".into(), "json".into()],
license: self.provider.license.clone(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: self.provider.join_keys.clone(),
description: self.provider.description.clone(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
self.provider.source,
self.provider.endpoint.clone(),
self.provider.table.clone(),
)
}
/// **Seven authorities behind one transport.** [`parse_sdmx_json`] is the only
/// thing these presets share; nothing on this card is. Eurostat, the OECD, the
/// ECB, INSEE, the ONS, Istat and the World Bank are seven different bodies with
/// seven different licences, seven different support desks and — the part that
/// costs an operator a whole day — rate policies that are not even the same
/// ORDER of magnitude: Istat bans an IP for **one to two days** after 5 queries in
/// a minute, while the ONS answers 120 requests per 10 seconds and tells you in a
/// `Retry-After` header exactly when to come back. Giving them one shared card
/// would be the single most expensive convenience in this file.
///
/// So it dispatches by name — the same shape
/// [`max_per_sec`](crate::rest::RestCollector::max_per_sec) uses — and the generic
/// escape hatch ([`SdmxCollector::generic`]) answers `None` rather than inheriting
/// somebody else's terms, because a provider nobody has documented is exactly the
/// case `None` exists to name.
///
/// Provenance of the live figures below: each terms/docs URL was fetched on
/// 2026-09-04 and the quoted limits are the source's own published wording, except
/// the two marked *measured*, which are HTTP responses observed on that date.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit};
let d = match self.provider.name.as_str() {
// ── Eurostat ────────────────────────────────────────────────────────
"eurostat-sdmx" => HumanDoc::new(
"Eurostat — the statistical office of the European Union (European Commission, \
DG ESTAT)",
"Eurostat, compiling figures supplied by the national statistical institutes of \
the EU Member States",
"https://ec.europa.eu/eurostat/help/copyright-notice",
"https://ec.europa.eu/eurostat/web/user-guides/data-browser/api-data-access/\
api-introduction",
)
.with_contact(Contact::Form(
"https://ec.europa.eu/eurostat/web/main/contact-us/user-support".into(),
))
.with_rate_limit(RateLimit::Documented(
"no request-RATE cap is published; the API caps extraction SIZE instead — an \
over-large query is refused with `EXTRACTION_TOO_BIG`, and Eurostat routes big \
pulls to a separate asynchronous API"
.into(),
))
.with_quirk(
"`base_url` must already carry `format=jsondata`. This collector sends no \
`Accept` header, so an SDMX 2.1 URL without it comes back in the service's \
default XML representation and `serde_json` fails on the first byte.",
)
.with_quirk(
"An unfiltered whole-dataset query is REFUSED (`EXTRACTION_TOO_BIG`), not \
truncated — filter on `geo` / `TIME_PERIOD`. The asynchronous API Eurostat \
offers for such pulls is a different protocol this collector does not speak: \
`collect` does one synchronous GET.",
)
.with_quirk(
"Comext / PRODCOM datasets (the `DS_`-prefixed ones) live under a DIFFERENT base \
path — `https://ec.europa.eu/eurostat/api/comext/dissemination/` — and are \
unreachable from the dissemination endpoint this collector declares.",
)
.with_quirk(
"Only dimension CODES reach the batch: `geo` carries \"SE\", never \"Sweden\". \
The label maps in the message are read for neither columns nor values.",
)
.with_quirk(
"\"Free re-use with attribution\" is the headline, not the whole rule: the \
copyright notice carves out figures for non-EU countries (US, Japan, China, …) \
and some Swiss and Austrian trade data, which may not be re-used commercially \
without first being removed from the table.",
),
// ── OECD ────────────────────────────────────────────────────────────
"oecd-sdmx" => HumanDoc::new(
"OECD — Organisation for Economic Co-operation and Development",
"the OECD, but individual tables may be third-party owned: the OECD terms make \
it the RE-USER's job to check each table's `source` tab before republishing",
"https://www.oecd.org/en/about/terms-conditions.html",
"https://www.oecd.org/en/data/insights/data-explainers/2024/09/api.html",
)
.with_contact(Contact::Email("stat.contact@oecd.org".into()))
.with_quirk(
"*Measured 2026-09-04*: `www.oecd.org` sits behind Cloudflare bot management and \
answered HTTP 403 to a plain HTTP client on both URLs above, while \
`sdmx.oecd.org` answered 200. Read the documentation in a browser; the API \
itself is not the thing being blocked.",
)
.with_quirk(
"*Measured 2026-09-04*: a malformed dataflow key comes back as HTTP **422** with \
a `text/plain` body — not a JSON error document — so the failure surfaces as a \
`serde_json` parse error rather than as the API's own complaint.",
)
.with_quirk(
"`https://sdmx.oecd.org/public/rest/data` is the service ROOT. `base_url` must \
carry a full agency-qualified reference \
(`OECD.SDD.NAD,DSD_NAMAIN1@DF_QNA,1.1/<key>`) plus `format=jsondata`.",
)
.with_quirk(
"No rate limit was found published as of 2026-09-04 — but the OECD's own API \
best-practice page is behind the same 403, so this is \"not found\", not \
\"there is none\". Treat it gently.",
),
// ── ECB ─────────────────────────────────────────────────────────────
"ecb-sdmx" => HumanDoc::own(
"European Central Bank",
"https://www.ecb.europa.eu/services/disclaimer/html/index.en.html",
"https://data.ecb.europa.eu/help/api/overview",
)
.with_contact(Contact::Form(
"https://data.ecb.europa.eu/help/contact-us".into(),
))
.with_quirk(
"TWO hosts, and mixing them up is the first failure: the DOCS live on \
`data.ecb.europa.eu`, the DATA on `data-api.ecb.europa.eu`. A query URL copied \
out of the Data Portal in a browser will not answer this collector.",
)
.with_quirk(
"*Measured 2026-09-04*: `.../service/data/EXR/D.USD.EUR.SP00.A?\
lastNObservations=1` returns `application/vnd.sdmx.genericdata+xml;version=2.1`, \
and the SAME url with `format=jsondata` returns \
`application/vnd.sdmx.data+json;version=1.0.0-wd`. This collector sends no \
`Accept` header, so without that parameter it gets the XML.",
)
.with_quirk(
"That `version=1.0.0-wd` is the whole reason the ECB needed no new parsing: the \
ECB serves SDMX-JSON **1.0**, with the singular `structure` node rather than \
2.1's `data.structures[0]`. [`parse_sdmx_json`] accepts both shapes.",
)
.with_quirk(
"Attribution is a stated condition of re-use: the ECB disclaimer requires the \
ECB be cited as the source, that free availability be disclosed when the \
material is distributed commercially, and that any modification be noted.",
),
// ── INSEE (France) ──────────────────────────────────────────────────
"insee-sdmx" => HumanDoc::own(
"INSEE — Institut national de la statistique et des études économiques (France)",
"https://www.insee.fr/en/information/2409130",
"https://www.insee.fr/en/information/2868055",
)
.with_contact(Contact::Form(
"https://www.insee.fr/en/information/3313008".into(),
))
.with_rate_limit(RateLimit::Documented(
"no numeric cap is published; INSEE's legal notice permits automated extraction \
only \"to the extent that it does not impair the proper functioning of the \
site\" and reserves the right to limit the number of requests or temporarily \
block IP addresses"
.into(),
))
.with_quirk(
"Attribution is a licence CONDITION, not a courtesy. Licence Ouverte 2.0 \
requires the source be given as \"Source: Insee\" AND the date of the last \
update of the data when that is known.",
)
.with_quirk(
"INSEE explicitly steers bulk consumers off the website and onto its dedicated \
dissemination services (API, downloadable files, open-data bases). The BDM SDMX \
service this collector uses IS that sanctioned path — do not point `base_url` \
at `www.insee.fr` pages instead.",
)
.with_quirk(
"`https://bdm.insee.fr/series/sdmx/data` is the service ROOT: `base_url` must \
carry a real query — `/data/SERIES_BDM/<idbank>` for a single series, \
`/data/<dataflow>/<key>` for a thematic group — plus `format=jsondata`.",
)
.with_quirk(
"The real-or-fake sample never touches the network unless `INSEE_SDMX_URL` names \
a concrete query (see `crate::sample::insee_try_get_real_sample_data`), so a \
green bench measures the flattener, not INSEE's reachability.",
),
// ── ONS (United Kingdom) ────────────────────────────────────────────
"ons-sdmx" => HumanDoc::own(
"ONS — Office for National Statistics (United Kingdom)",
"https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/",
"https://developer.ons.gov.uk/",
)
.with_contact(Contact::Email("customise.my.data@ons.gov.uk".into()))
.with_rate_limit(RateLimit::Documented(
"120 requests / 10 seconds and 200 requests / minute across all site and API \
assets, and 15 requests / 10 seconds for high-demand assets, per the ONS guide \
to rate limiting and bot development"
.into(),
))
.with_quirk(
"Over the limit the ONS answers HTTP 429 with a `Retry-After` header, and says \
plainly that ignoring it \"may impose a block to our services for up to 1 \
hour\". This collector does one GET per run, so the cap only bites a tight \
scheduler — but the block, if earned, outlives the run.",
)
.with_quirk(
"The ONS asks bots to identify themselves in `User-Agent` \
(`botName/1.0.0 (organisation +https://…/bot.html)`) and asks that NO personal \
name or personal e-mail appear in that header. `CollectorCfg` carries no such \
header today — set one deliberately if you run this often.",
)
.with_quirk(
"The Developer Hub is still labelled Beta and ships a \"Guide to retirement of \
API endpoints\": the endpoint a config pins can be withdrawn under it.",
)
.with_quirk(
"This preset's join keys are `Geography` + `TIME_PERIOD`, NOT the `REF_AREA` the \
other national presets use — ONS dimension ids differ, so a cross-provider join \
has to map them rather than assume them.",
),
// ── Istat (Italy) ───────────────────────────────────────────────────
"istat-sdmx" => HumanDoc::own(
"Istat — Istituto Nazionale di Statistica (Italy)",
"https://www.istat.it/en/legal-notice/",
"https://www.istat.it/en/classifications-and-tools/sdmx-web-services/",
)
.with_contact(Contact::Form("https://www.istat.it/en/contact-us/".into()))
.with_rate_limit(RateLimit::Documented(
"5 queries per minute per IP — and exceeding it triggers an access block lasting \
between 1 and 2 DAYS, not a 429"
.into(),
))
.with_quirk(
"THE one that costs a working day. Istat's own words: \"To prevent SDMX web \
service overloading, a limit of 5 queries per minute has been set for each IP. \
Once this limit is exceeded, an access block lasting between 1 and 2 days is \
activated.\" There is no 429 to back off from — a retry loop locks the host out \
until tomorrow. Schedule this collector; never poll it.",
)
.with_quirk(
"*Measured 2026-09-04*: `GET https://esploradati.istat.it/SDMXWS/rest/dataflow` \
exceeded 10 MB (the fetch was aborted at that cap). Enumerate dataflows once \
and cache them — at five queries a minute you cannot afford to re-discover.",
)
.with_quirk(
"Istat's own published example queries ask for `format=structurespecificdata` \
(SDMX-ML). This collector reads SDMX-JSON only, so `base_url` must say \
`format=jsondata` instead.",
)
.with_quirk(
"Flow references are agency-qualified with COMMAS — `IT1,POP,1.0` — and the same \
`/SDMXWS` root also serves SOAP; only the `/SDMXWS/rest/…` paths are usable \
here.",
)
.with_quirk(
"CC BY 4.0 applies \"unless otherwise stated\", and Istat explicitly excludes \
third-party images, logos and trademarks carried on its pages.",
),
// ── World Bank ──────────────────────────────────────────────────────
"worldbank-sdmx" => HumanDoc::new(
"The World Bank",
"the World Bank, republishing indicators collected from national statistical \
systems and partner agencies",
"https://datacatalog.worldbank.org/public-licenses",
"https://datahelpdesk.worldbank.org/knowledgebase/articles/\
889392-about-the-indicators-api-documentation",
)
.with_contact(Contact::Form("https://datahelpdesk.worldbank.org/".into()))
.with_quirk(
"\"API keys and other authentication methods are no longer necessary to access \
the API\" — the World Bank's own words. Any guide telling an operator to \
register for a key is stale; `AuthKind::None` here is correct.",
)
.with_quirk(
"CC BY 4.0 is only the DEFAULT, for datasets the World Bank itself produces. The \
public-licences page also lists ODbL, the Microdata Research License and \
case-by-case custom terms, so `meta().license` is the common case rather than a \
guarantee — check the dataset before republishing.",
)
.with_quirk(
"Two collectors, one institution, two incompatible bodies: `worldbank-sdmx` \
(this one) speaks SDMX-JSON at `api.worldbank.org/v2/sdmx/rest/data`, while \
`worldbank-indicators` in [`crate::rest`] reads the plain \
`/v2/country/<iso>/indicator/<code>?format=json` array. Swapping the two \
`base_url`s fails in the parser, not in the fetch.",
)
.with_quirk(
"No rate limit is named in the Indicators API documentation as of 2026-09-04, \
and nothing else was found published — \"not found\", not \"there is none\".",
)
.with_quirk(
"Absent `WORLDBANK_SDMX_URL` the sample stays synthetic \
(`crate::sample::worldbank_sdmx_try_get_real_sample_data`), so bench numbers \
measure the flattener rather than the World Bank.",
),
// A provider nobody has documented gets `None`, not somebody else's card.
_ => return None,
};
Some(d)
}
/// **Every DIMENSION column of this cube** — which is knowable only from the cfg.
///
/// An SDMX observation is identified by its full coordinate: one code per
/// dimension. The derived key was `["REF_AREA", "TIME_PERIOD"]` (or the provider's
/// own spelling of the geo axis) — SDMX *concept* names, not the column names the
/// flattener emits, and short of the full coordinate anyway: a cube with `age` and
/// `sex` dimensions would have collapsed every age and sex into one row per
/// (region, period).
///
/// So the key is read from the SAME `params.dims` the schema is built from
/// ([`cube_schema`](Self::cube_schema)) — one writer, and a cube pointed at more
/// dimensions keys on more of them automatically.
fn business_key_for(&self, cfg: &CollectorCfg) -> crate::history::BusinessKey {
let dims: Vec<String> = Self::cube_schema(cfg)
.fields()
.iter()
.map(|f| f.name().clone())
.filter(|n| n != "value")
.collect();
if dims.is_empty() {
return crate::history::BusinessKey::append_only(
"the message declared no dimensions, so an observation has no coordinate",
);
}
let cols: Vec<&str> = dims.iter().map(String::as_str).collect();
crate::history::BusinessKey::of(
&cols,
format!(
"one observation per full cube coordinate ({}) — the dimension set is \
message-driven, so the key is read from the same `params.dims` the \
schema is",
dims.join(", ")
),
)
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
Ok(Self::cube_schema(cfg))
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(10);
let auth = cfg.auth.resolve_async().await?;
let body = http.get_text(&cfg.base_url, auth.as_ref()).await?;
let json: Value = serde_json::from_str(&body)?;
let rb = parse_sdmx_json(&json)?;
let n = rb.num_rows() as u64;
if n == 0 {
return Ok(CollectStats {
reason: Some(format!(
"{} SDMX message carried no observations",
self.provider.name
)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::Array;
/// A representative SDMX-JSON 2.1 message: 1 series dim (`geo`: SE, NO) × 1
/// observation dim (`TIME_PERIOD`: 2021, 2022) → 4 observations, flattened to
/// per-(geo, time) value rows. Offline DATA contract (no network) + a
/// testmatrix cell.
#[test]
fn sdmx_series_observations_flatten_to_dim_value_rows() {
let json: Value = serde_json::json!({
"data": {
"structures": [{
"dimensions": {
"series": [
{ "id": "geo", "name": "Geo",
"values": [ {"id":"SE","name":"Sweden"}, {"id":"NO","name":"Norway"} ] }
],
"observation": [
{ "id": "TIME_PERIOD", "name": "Period",
"values": [ {"id":"2021"}, {"id":"2022"} ] }
]
}
}],
"dataSets": [{
"series": {
"0": { "observations": { "0": [100.0], "1": [110.0] } },
"1": { "observations": { "0": [50.0], "1": [55.0] } }
}
}]
}
});
let obs = parse_sdmx_obs(&json).expect("parse sdmx");
assert_eq!(obs.len(), 4, "2 geo × 2 time = 4 observations");
// Deterministic order: geo pos then time pos → SE/2021, SE/2022, NO/2021, NO/2022.
assert_eq!(
obs[0].dims,
vec![
("geo".into(), "SE".into()),
("TIME_PERIOD".into(), "2021".into())
]
);
assert_eq!(obs[0].value, Some(100.0));
assert_eq!(
obs[3].dims,
vec![
("geo".into(), "NO".into()),
("TIME_PERIOD".into(), "2022".into())
]
);
assert_eq!(obs[3].value, Some(55.0));
let rb = parse_sdmx_json(&json).expect("sdmx batch");
assert_eq!(rb.num_rows(), 4, "4 value rows");
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
cols,
vec!["geo", "TIME_PERIOD", "value"],
"series dim, obs dim, value"
);
let geo = rb
.column_by_name("geo")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("geo is utf8");
let time = rb
.column_by_name("TIME_PERIOD")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("time is utf8");
let value = rb
.column_by_name("value")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("value is f64");
assert_eq!(geo.value(0), "SE");
assert_eq!(time.value(0), "2021");
assert!((value.value(0) - 100.0).abs() < 1e-9, "SE/2021 value");
assert_eq!(geo.value(2), "NO");
assert!((value.value(2) - 50.0).abs() < 1e-9, "NO/2021 value");
crate::functional_status(
"korp-collectors/eurostat-sdmx",
"series_obs_flatten",
true,
&format!(
"{} (geo, time) observation rows from SDMX-JSON",
rb.num_rows()
),
);
}
/// The flat shape (no series dims — observations directly on the dataSet) also
/// flattens, and a `null` observation lands as a nullable value (RAGNARÖK: a
/// missing cell is honestly null, not a dropped row).
#[test]
fn sdmx_flat_observations_and_null_value() {
let json: Value = serde_json::json!({
"structure": {
"dimensions": {
"observation": [
{ "id": "TIME_PERIOD", "values": [ {"id":"2020"}, {"id":"2021"}, {"id":"2022"} ] }
]
}
},
"dataSets": [{
"observations": { "0": [42.0], "1": [null], "2": [44.5] }
}]
});
let rb = parse_sdmx_json(&json).expect("flat sdmx");
assert_eq!(rb.num_rows(), 3, "three periods");
let value = rb
.column_by_name("value")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("value is f64");
assert!((value.value(0) - 42.0).abs() < 1e-9);
assert!(
value.is_null(1),
"the null observation is a nullable cell, not dropped"
);
assert!((value.value(2) - 44.5).abs() < 1e-9);
crate::functional_status(
"korp-collectors/eurostat-sdmx",
"flat_obs_null",
true,
"flat observations shape + nullable missing cell",
);
}
/// A message with an empty dataSet yields zero observations — the collector
/// maps that to a RAGNARÖK reason (RED-when-broken empty guard).
#[test]
fn sdmx_empty_dataset_yields_no_rows() {
let json: Value = serde_json::json!({
"data": {
"structures": [{ "dimensions": { "series": [], "observation": [] } }],
"dataSets": [{ "series": {} }]
}
});
assert!(parse_sdmx_obs(&json).expect("parse").is_empty());
let rb = parse_sdmx_json(&json).expect("batch");
assert_eq!(
rb.num_rows(),
0,
"no observations → zero rows (collector sets reason)"
);
}
/// The collector declares its Eurostat source → warehouse descriptor.
#[test]
fn eurostat_descriptor_is_declared() {
let st = SdmxCollector::eurostat().source_target();
assert_eq!(st.source, SourceEndpoint::Eurostat);
assert_eq!(st.target_table, "eurostat_sdmx");
assert!(
st.label().contains("eurostat"),
"label names the source: {}",
st.label()
);
assert_eq!(SdmxCollector::eurostat().meta().name, "eurostat-sdmx");
// The back-compat alias resolves to the Eurostat preset.
assert_eq!(
EurostatSdmxCollector::default().meta().name,
"eurostat-sdmx"
);
}
/// The SAME collector, generalised: OECD + ECB presets reuse the SDMX-JSON
/// parser with only branding/endpoint changed (zero new parsing) — and a
/// generic Eurostat-2.1 message parses identically through the OECD preset,
/// proving the flattener is provider-agnostic.
#[test]
fn sdmx_generalises_to_oecd_and_ecb_providers() {
let oecd = SdmxCollector::oecd();
assert_eq!(oecd.meta().name, "oecd-sdmx");
let st = oecd.source_target();
assert_eq!(st.target_table, "oecd_sdmx");
assert!(
st.source_endpoint.contains("oecd.org"),
"OECD endpoint: {}",
st.source_endpoint
);
let ecb = SdmxCollector::ecb();
assert_eq!(ecb.meta().name, "ecb-sdmx");
assert!(
ecb.source_target()
.source_endpoint
.contains("ecb.europa.eu")
);
// A generic SDMX-JSON message flattens the same way regardless of provider.
let json: Value = serde_json::json!({
"data": {
"structures": [{
"dimensions": {
"series": [ { "id": "REF_AREA",
"values": [ {"id":"USA"}, {"id":"DEU"} ] } ],
"observation": [ { "id": "TIME_PERIOD",
"values": [ {"id":"2021"}, {"id":"2022"} ] } ]
}
}],
"dataSets": [{
"series": {
"0": { "observations": { "0": [1.0], "1": [2.0] } },
"1": { "observations": { "0": [3.0], "1": [4.0] } }
}
}]
}
});
let rb = parse_sdmx_json(&json).expect("generic sdmx flattens");
assert_eq!(
rb.num_rows(),
4,
"2 area × 2 time observations, any provider"
);
let generic = SdmxCollector::generic("imf-sdmx", "https://sdmxcentral.imf.org/rest/data");
assert_eq!(generic.meta().name, "imf-sdmx");
assert_eq!(generic.source_target().target_table, "imf_sdmx");
crate::functional_status(
"korp-collectors/oecd-sdmx",
"sdmx_generalised",
true,
"OECD/ECB/IMF ride the shared SDMX-JSON parser (zero new parsing)",
);
}
/// The national-statistics presets (INSEE / ONS / ISTAT / World Bank) each
/// declare their own typed `source → target` (distinct endpoint + table +
/// SourceEndpoint variant) AND a representative national SDMX-JSON message
/// flattens to the RIGHT typed `(REF_AREA, TIME_PERIOD, value)` rows through
/// the SAME shared flattener (zero new parsing) — RED if any preset wiring or
/// the flatten breaks.
#[test]
fn sdmx_national_stats_presets_declare_and_flatten() {
use arrow_array::{Array, Float64Array, StringArray};
// Every national preset names a distinct endpoint, table + SourceEndpoint.
for (c, endpoint_frag, table, src) in [
(
SdmxCollector::insee(),
"insee.fr",
"insee_sdmx",
SourceEndpoint::Insee,
),
(
SdmxCollector::ons(),
"ons.gov.uk",
"ons_sdmx",
SourceEndpoint::Ons,
),
(
SdmxCollector::istat(),
"istat.it",
"istat_sdmx",
SourceEndpoint::Istat,
),
(
SdmxCollector::worldbank(),
"worldbank.org",
"worldbank_sdmx",
SourceEndpoint::WorldBankSdmx,
),
] {
let st = c.source_target();
assert_eq!(
st.source,
src,
"{} declares its SourceEndpoint",
c.meta().name
);
assert_eq!(st.target_table, table, "{} target table", c.meta().name);
assert!(
st.source_endpoint.contains(endpoint_frag),
"{} names its endpoint host ({}): {}",
c.meta().name,
endpoint_frag,
st.source_endpoint
);
}
// A representative national cube: REF_AREA (FR, IT) × TIME_PERIOD (2021,
// 2022) → 4 observations, flattened by the SHARED parser (any provider).
let json: Value = serde_json::json!({
"data": {
"structures": [{
"dimensions": {
"series": [ { "id": "REF_AREA",
"values": [ {"id":"FR"}, {"id":"IT"} ] } ],
"observation": [ { "id": "TIME_PERIOD",
"values": [ {"id":"2021"}, {"id":"2022"} ] } ]
}
}],
"dataSets": [{
"series": {
"0": { "observations": { "0": [67.5], "1": [67.8] } },
"1": { "observations": { "0": [59.1], "1": [58.9] } }
}
}]
}
});
let rb = parse_sdmx_json(&json).expect("national SDMX flattens");
assert_eq!(rb.num_rows(), 4, "2 REF_AREA × 2 TIME_PERIOD observations");
let area = rb
.column_by_name("REF_AREA")
.and_then(|a| a.as_any().downcast_ref::<StringArray>())
.expect("REF_AREA utf8");
let value = rb
.column_by_name("value")
.and_then(|a| a.as_any().downcast_ref::<Float64Array>())
.expect("value f64");
assert_eq!(area.value(0), "FR");
assert!((value.value(0) - 67.5).abs() < 1e-9, "FR/2021 value");
assert_eq!(area.value(2), "IT");
assert!((value.value(2) - 59.1).abs() < 1e-9, "IT/2021 value");
crate::functional_status(
"korp-collectors/insee-sdmx",
"national_sdmx_presets",
true,
"INSEE/ONS/ISTAT/World Bank presets ride the shared SDMX-JSON parser",
);
}
}