//! **How a collector's landed table becomes typed nodes and real edges** — the
//! DECLARATION, and nothing that builds one.
//!
//! Lane EC O7. Until 2026-09-04 this knowledge lived in korp as a
//! `static SOURCES: [SourceMapping; 5]` — a hand-written array carrying Rust function
//! pointers (`prepare`, `mappings`) — and a hand-written `knut_plans()` roster beside
//! it. Two consequences, both structural:
//!
//! * **A collector could not describe its own graph.** The collector is the thing that
//! knows what its columns mean; korp had to be edited to say so, in a second repo,
//! in an array whose length is in its own type.
//! * **The roster was a list, so it could drift from the registry.** A plan named there
//! for a collector that no longer exists is a pipeline the UI offers and nothing can
//! run; a collector added without touching the array is a pipeline nobody can reach.
//!
//! So the declaration moves DOWN to the leaf and becomes DATA. This module holds no
//! knut type and no korp type — it cannot, korp-collectors is a leaf — and korp
//! compiles what it finds here into `knut_bifrost::TableMapping`s through ONE writer.
//!
//! ## Why the join step is declared rather than coded
//!
//! knut's rel2graph has **no join step**, so a name-to-code resolution has to happen
//! before the spec runs, and korp did it with a `fn` per source. Two of those five
//! functions were the same two behaviours: *resolve a NAME against Sweden's kommun/län
//! spine* (polisen's `location_name`) and *read a CODE whose granularity is the source's
//! own* (SCB's `region`: 4 characters is a kommun, 2 is a län). Both are now
//! [`SpineLookup`] values, and the third — synthesising a key column a table does not
//! carry — is [`DerivedId`].
//!
//! **An unresolved value yields NULL and the sink drops an edge whose endpoint is
//! NULL**, so "where a code is missing, do not invent an edge" stays true by
//! construction rather than by discipline. That property belongs to the sink and this
//! module does not weaken it.
use serde::{Deserialize, Serialize};
/// **One landed collector table, and the graph it honestly makes.**
///
/// Returned by [`Collector::graph_mapping`](crate::Collector::graph_mapping); `None` is
/// a real answer and the common one — a table whose rows declare no relationship to
/// anything (an AIS position: mmsi, lat, lon, sog — a position is not a place) must say
/// so rather than be given invented edges.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraphMapping {
/// The plan name when this source is run alone; also the graph's name.
pub plan: String,
/// The pane label.
pub title: String,
/// The landed table this maps.
pub table: String,
/// **What the mapping cannot honestly do, in one sentence.** Required, not
/// optional: every mapping has a limitation, and the ones that go unwritten are the
/// ones an operator discovers as a wrong number. It is stated in the run report and
/// in the pane.
pub limitation: String,
/// Key columns the table does not carry and the mapping needs. Computed before the
/// spec runs, in declaration order.
#[serde(default)]
pub derived_ids: Vec<DerivedId>,
/// How a column of this table resolves against Sweden's kommun/län spine, when it
/// does at all. `None` ⇒ no spine columns are added and no geographic edge can
/// exist — which is the honest answer for a catalogue row.
#[serde(default)]
pub spine: Option<SpineJoin>,
/// The node labels this mapping produces, with the column that identifies each.
pub vertices: Vec<VertexDecl>,
/// The relationships it produces, each from two columns of ONE row — the pattern
/// knut's engine is built for.
#[serde(default)]
pub edges: Vec<EdgeDecl>,
}
impl GraphMapping {
/// A mapping with no derived ids, no spine join and no edges — the smallest honest
/// declaration, and the shape a table of standalone objects has.
pub fn new(
plan: impl Into<String>,
title: impl Into<String>,
table: impl Into<String>,
limitation: impl Into<String>,
) -> Self {
GraphMapping {
plan: plan.into(),
title: title.into(),
table: table.into(),
limitation: limitation.into(),
derived_ids: Vec::new(),
spine: None,
vertices: Vec::new(),
edges: Vec::new(),
}
}
/// Add a vertex label keyed on `id_column`, carrying `properties`.
pub fn vertex(mut self, label: &str, id_column: &str, properties: &[&str]) -> Self {
self.vertices.push(VertexDecl {
label: label.into(),
id_column: id_column.into(),
properties: properties.iter().map(|s| (*s).to_string()).collect(),
});
self
}
/// Add an FK edge. `src`/`dst` are `(label, column, id_property)`; the id property
/// MUST equal the id column of whatever vertex owns that label, or the sink MERGEs
/// a second node for the same entity — see [`Self::dangling_endpoints`].
pub fn edge(self, rel: &str, src: (&str, &str, &str), dst: (&str, &str, &str)) -> Self {
self.edge_with(rel, src, dst, &[])
}
/// Add an FK edge carrying `properties` — columns of the same row placed on the
/// relationship. See [`EdgeDecl::properties`]; `edge` is this with none.
pub fn edge_with(
mut self,
rel: &str,
src: (&str, &str, &str),
dst: (&str, &str, &str),
properties: &[&str],
) -> Self {
self.edges.push(EdgeDecl {
rel: rel.into(),
src: Endpoint {
label: src.0.into(),
column: src.1.into(),
id_column: src.2.into(),
},
dst: Endpoint {
label: dst.0.into(),
column: dst.1.into(),
id_column: dst.2.into(),
},
properties: properties.iter().map(|s| (*s).to_string()).collect(),
});
self
}
/// Declare a key column the table does not carry.
pub fn derive_id(mut self, id: DerivedId) -> Self {
self.derived_ids.push(id);
self
}
/// Declare how a column resolves against the kommun/län spine.
pub fn with_spine(mut self, from_column: &str, lookup: SpineLookup) -> Self {
self.spine = Some(SpineJoin {
from_column: from_column.into(),
lookup,
});
self
}
/// Every node label this mapping can produce, in declaration order, deduplicated —
/// vertices first, then labels that appear only as an edge endpoint (the spine's
/// `Municipality` / `County`, which this mapping references but does not own).
pub fn labels(&self) -> Vec<String> {
let mut out: Vec<String> = Vec::new();
let mut push = |l: &str| {
if !out.iter().any(|x| x == l) {
out.push(l.to_string());
}
};
for v in &self.vertices {
push(&v.label);
}
for e in &self.edges {
push(&e.src.label);
push(&e.dst.label);
}
out
}
/// Every relationship type, in declaration order, deduplicated.
pub fn rels(&self) -> Vec<String> {
let mut out: Vec<String> = Vec::new();
for e in &self.edges {
if !out.iter().any(|x| x == &e.rel) {
out.push(e.rel.clone());
}
}
out
}
/// Does this mapping reference the kommun/län spine tables? True when it declares a
/// spine join OR names a spine label as an edge endpoint — either way the spine's
/// own vertices must be in the same run, or the edge lands on nothing.
pub fn uses_spine(&self) -> bool {
self.spine.is_some()
|| self
.edges
.iter()
.any(|e| is_spine_label(&e.src.label) || is_spine_label(&e.dst.label))
}
}
/// The spine's two labels and the id property each is keyed on. korp owns the spine
/// tables; a collector mapping only references them, so the key has to be stated
/// somewhere both sides agree on and this is the leaf-side half.
pub const SPINE_MUNICIPALITY: (&str, &str) = ("Municipality", "kommun_code");
/// See [`SPINE_MUNICIPALITY`].
pub const SPINE_COUNTY: (&str, &str) = ("County", "county_code");
/// Is `label` one of the spine's own?
pub fn is_spine_label(label: &str) -> bool {
label == SPINE_MUNICIPALITY.0 || label == SPINE_COUNTY.0
}
/// The id property a spine label is keyed on, if `label` is a spine label.
pub fn spine_key(label: &str) -> Option<&'static str> {
match label {
l if l == SPINE_MUNICIPALITY.0 => Some(SPINE_MUNICIPALITY.1),
l if l == SPINE_COUNTY.0 => Some(SPINE_COUNTY.1),
_ => None,
}
}
/// **The column that spine resolution reads, and how to read it.**
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpineJoin {
/// The column carrying the place.
pub from_column: String,
/// How its values are matched.
pub lookup: SpineLookup,
}
/// How a column's values match Sweden's kommun/län spine.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SpineLookup {
/// **The value is a NAME.** Try the kommun names, then the län names; a value
/// matching neither resolves to NULL and its edge is dropped. This is polisen's
/// `location_name`, which is free text an operator typed — the live feed on
/// 2026-09-02 was län-only, 21 distinct values.
ByName,
/// **The value is a CODE whose granularity is the source table's**, not a declared
/// field: 4 characters is a kommun (and its län follows), 2 is a län, `00` is Riket
/// and gets no node. SCB's `region` column.
ByCode,
}
/// **A key column the table does not carry.** Computed before the spec runs.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DerivedId {
/// The column to add.
pub column: String,
/// What it is built from.
pub from: DerivedFrom,
}
impl DerivedId {
/// A key from these columns, joined in order.
pub fn from_columns(column: &str, parts: &[&str]) -> Self {
DerivedId {
column: column.into(),
from: DerivedFrom::Columns(parts.iter().map(|s| (*s).to_string()).collect()),
}
}
/// A key from every column except these, in schema order — the CUBE key, whose
/// dimension set is message-driven and therefore not listable here.
pub fn from_all_except(column: &str, excluded: &[&str]) -> Self {
DerivedId {
column: column.into(),
from: DerivedFrom::AllColumnsExcept(
excluded.iter().map(|s| (*s).to_string()).collect(),
),
}
}
}
/// How a [`DerivedId`] is built. Values are joined with `|`.
///
/// **The two variants differ on NULL, deliberately**, and the difference is the whole
/// reason they are two variants:
///
/// * [`Columns`](DerivedFrom::Columns) — a NULL in ANY part makes the whole id NULL. A
/// train position with no `train_id` is not a fix of anything, so it has no identity,
/// and the sink drops an edge whose endpoint is NULL. Inventing a key like `|2026-…`
/// would MERGE every such row into one node.
/// * [`AllColumnsExcept`](DerivedFrom::AllColumnsExcept) — a NULL renders as the empty
/// string. A cube coordinate is POSITIONAL: the dimension is present and its value is
/// empty, which is a different fact from the dimension not existing, and the shape of
/// the key must not change with the data.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum DerivedFrom {
/// These columns, in this order. A NULL part ⇒ the id is NULL.
Columns(Vec<String>),
/// Every column of the batch except these, in SCHEMA order. A NULL renders empty.
AllColumnsExcept(Vec<String>),
}
/// A node label and the column that identifies it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct VertexDecl {
/// The label.
pub label: String,
/// The column whose value is this node's identity.
pub id_column: String,
/// Columns carried onto the node as properties.
#[serde(default)]
pub properties: Vec<String>,
}
/// An FK edge built from two columns of ONE row.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EdgeDecl {
/// The relationship type.
pub rel: String,
/// Where it starts.
pub src: Endpoint,
/// Where it ends.
pub dst: Endpoint,
/// **Columns of this row carried onto the relationship.**
///
/// An edge without them reaches the graph store as a bare arc — a road graph you
/// cannot colour by class, cost by speed or route by direction, which is most of
/// the reason to have one.
///
/// The capability was already in knut:
/// `knut_bifrost::transform::EdgeTransform` has carried `properties` all along.
/// This declaration simply never exposed it.
#[serde(default)]
pub properties: Vec<String>,
}
/// One end of an [`EdgeDecl`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Endpoint {
/// The node label.
pub label: String,
/// The column of THIS table carrying the endpoint's key value.
pub column: String,
/// The id PROPERTY on the node — must equal that label's vertex key.
pub id_column: String,
}
/// **Every label must be keyed ONE way, and every endpoint must agree with its label's
/// key** — checked across ALL mappings at once, because that is the scope the failure
/// lives at.
///
/// The sink keys a node on `(label, id property, value)`. So a label declared as a
/// vertex in one mapping and referenced as an endpoint in another with a different id
/// property makes the sink MERGE a SECOND node for the same entity: the graph silently
/// doubles and every count is wrong. `Organisation` is referenced by both the DCAT
/// mapping (`publisher`) and the SMHI one (`owner` → the `publisher` property) — one
/// entity, two collectors, and nothing but this check keeps them agreeing.
///
/// An endpoint label that NO mapping declares as a vertex is fine and common:
/// `EventType`, `Format`, `License` and `Organisation` are created by the edge itself
/// and carry no properties of their own. That is why this cannot be a per-mapping
/// check — a per-mapping one would have to flag them all.
///
/// Returns one line per conflict, empty when the whole set agrees.
pub fn key_conflicts(mappings: &[GraphMapping]) -> Vec<String> {
let mut keys: std::collections::BTreeMap<&str, (&str, &str)> = Default::default();
let mut out = Vec::new();
for (label, id) in SPINE_LABELS {
keys.insert(label, (id, "the kommun/län spine"));
}
for m in mappings {
for v in &m.vertices {
// The FIRST declaration wins the map, so one disagreement is reported once
// instead of once per endpoint that referenced the label afterwards.
match keys.get(v.label.as_str()) {
Some((prev, who)) if *prev != v.id_column => out.push(format!(
"label `{}` is keyed `{}` by `{}` and `{}` by `{}` — that is TWO \
nodes per entity",
v.label, prev, who, v.id_column, m.plan
)),
Some(_) => {}
None => {
keys.insert(&v.label, (&v.id_column, &m.plan));
}
}
}
}
for m in mappings {
for e in &m.edges {
for (side, ep) in [("src", &e.src), ("dst", &e.dst)] {
let Some((k, who)) = keys.get(ep.label.as_str()) else {
// An endpoint-only label: the edge creates it. Legitimate.
continue;
};
if *k != ep.id_column {
out.push(format!(
"`{}` in plan `{}`: {side} endpoint `{}` keys on `{}` but `{}` \
keys that label on `{}`",
e.rel, m.plan, ep.label, ep.id_column, who, k
));
}
}
}
}
out
}
/// The spine's labels and their keys, as one table — [`SPINE_MUNICIPALITY`] and
/// [`SPINE_COUNTY`] in the form [`key_conflicts`] folds over.
pub const SPINE_LABELS: [(&str, &str); 2] = [SPINE_MUNICIPALITY, SPINE_COUNTY];
#[cfg(test)]
mod tests {
use super::*;
/// **`key_conflicts` must catch the drift it exists for, and stay quiet on a
/// correct set.** Both directions, because a check that always answers "empty" is
/// the exact failure it is guarding against — and both KINDS, because a
/// vertex-vertex conflict and a vertex-endpoint conflict are different code paths.
#[test]
fn a_label_keyed_two_ways_is_caught_and_a_consistent_set_is_not() {
let a = GraphMapping::new("a", "A", "ta", "none")
.vertex("Event", "id", &["name"])
.vertex("Type", "type", &[])
.edge("OF_TYPE", ("Event", "id", "id"), ("Type", "type", "type"))
// an endpoint-only label the edge creates: legitimate, must NOT be flagged
.edge("BY", ("Event", "id", "id"), ("Organisation", "publisher", "publisher"))
// a SPINE endpoint, keyed the way korp keys the spine
.edge(
"IN_MUNICIPALITY",
("Event", "id", "id"),
("Municipality", "municipality_code", "kommun_code"),
);
assert_eq!(key_conflicts(&[a.clone()]), Vec::<String>::new());
// TWO mappings that key one label differently — the cross-collector failure,
// invisible to any per-mapping check.
let b = GraphMapping::new("b", "B", "tb", "none").vertex("Event", "event_uri", &[]);
let conflicts = key_conflicts(&[a.clone(), b]);
assert_eq!(conflicts.len(), 1, "{conflicts:?}");
assert!(conflicts[0].contains("Event"), "{conflicts:?}");
assert!(conflicts[0].contains("TWO"), "{conflicts:?}");
// An ENDPOINT that disagrees with its label's vertex key.
let c = GraphMapping::new("c", "C", "tc", "none")
.vertex("Event", "id", &[])
.vertex("Type", "type", &[])
.edge("OF_TYPE", ("Event", "id", "id"), ("Type", "type", "name"));
let conflicts = key_conflicts(&[c]);
assert_eq!(conflicts.len(), 1, "{conflicts:?}");
assert!(conflicts[0].contains("OF_TYPE"), "{conflicts:?}");
// A spine endpoint keyed on a property the spine is NOT keyed on — a mapping
// does not get to rename korp's spine.
let d = GraphMapping::new("d", "D", "td", "none").vertex("E", "id", &[]).edge(
"IN_COUNTY",
("E", "id", "id"),
("County", "county_code", "code"),
);
assert_eq!(key_conflicts(&[d]).len(), 1);
}
/// `labels`/`rels`/`uses_spine` are what a pane renders, so each is asserted on a
/// mapping where a wrong implementation would give a DIFFERENT answer: labels are
/// deduplicated across vertices and endpoints, and `uses_spine` is true from an
/// endpoint alone even with no declared join.
#[test]
fn the_pane_facts_are_derived_from_the_declaration() {
let m = GraphMapping::new("p", "T", "t", "none")
.vertex("Event", "id", &[])
.vertex("Type", "type", &[])
.edge("OF_TYPE", ("Event", "id", "id"), ("Type", "type", "type"))
.edge("ALSO", ("Event", "id", "id"), ("Type", "type2", "type"));
assert_eq!(m.labels(), vec!["Event", "Type"], "deduplicated");
assert_eq!(m.rels(), vec!["OF_TYPE", "ALSO"], "in declaration order");
assert!(!m.uses_spine(), "no spine join and no spine endpoint");
let geo = m.clone().edge(
"IN_COUNTY",
("Event", "id", "id"),
("County", "county_code", "county_code"),
);
assert!(geo.uses_spine(), "a spine ENDPOINT alone is enough");
assert_eq!(
geo.labels(),
vec!["Event", "Type", "County"],
"an endpoint-only label is still a label the plan produces"
);
let named = GraphMapping::new("p", "T", "t", "none")
.vertex("E", "id", &[])
.with_spine("location_name", SpineLookup::ByName);
assert!(named.uses_spine(), "a declared join alone is enough");
}
}