crates/knut-rel2graph-bench/tests/correctness.rs
//! **Correctness bencher** — property-style assertions that hold at every scale.
//!
//! The graph is built by the real bifröst mapping fabric; its correctness is
//! judged against an INDEPENDENT oracle computed straight from the generated
//! relational data (distinct PKs, FK cardinalities). If the mapping added,
//! dropped, mislabelled, or mis-pointed a single element, one of these fails.
use std::collections::BTreeSet;
use knut_bifrost::{GraphPlan, codegen::CodegenOptions};
use knut_rel2graph_bench::{
DEFAULT_SEED, build_mutation, distinct_count, generate, mapping_spec, nonnull_count, schema,
summarize,
};
/// The 10 entity labels — the junction tables must contribute NONE of these,
/// and no label beyond them.
const EXPECTED_LABELS: &[&str] = &[
"Address", "Category", "Customer", "Employee", "Order", "Payment", "Product", "Review",
"Shipper", "Supplier",
];
/// The full set of correctness invariants, checked at one scale.
fn check_scale(scale: usize) {
let spec = mapping_spec();
let db = generate(scale, DEFAULT_SEED);
let mutation = build_mutation(&spec, &db);
let g = summarize(&mutation);
let model = schema();
// ── Rule: table → node label, PK → node key ──────────────────────────────
// Exactly the 10 entity labels appear — junctions produce NO vertex.
let got: BTreeSet<String> = g.labels.iter().cloned().collect();
let want: BTreeSet<String> = EXPECTED_LABELS.iter().map(|s| s.to_string()).collect();
assert_eq!(
got, want,
"scale {scale}: node labels are exactly the 10 entities (no junction vertex)"
);
// node count per label == distinct PKs of that table.
for t in &model {
if let Some(v) = &t.vertex {
let expected = distinct_count(&db, t.name, v.key);
let actual = g.node_counts.get(v.label).copied().unwrap_or(0);
assert_eq!(
actual, expected,
"scale {scale}: node `{}` count == distinct `{}` of {}",
v.label, v.key, t.name
);
assert_eq!(
expected,
t.rows_at(scale),
"scale {scale}: {} rows scale linearly",
t.name
);
}
}
// ── Rule: FK → edge, cardinality preserved ────────────────────────────────
for t in &model {
for e in t.edges {
let expected = nonnull_count(&db, t.name, e.dst_col);
let actual = g.edges_for(e.src_label, e.rel, e.dst_label);
assert_eq!(
actual, expected,
"scale {scale}: edge {}-[{}]->{} count == non-null `{}` in {}",
e.src_label, e.rel, e.dst_label, e.dst_col, t.name
);
}
}
// ── Rule: junction table → EDGE not a spurious vertex ─────────────────────
for t in &model {
if t.junction {
assert!(t.vertex.is_none());
// Its edges are present and non-empty…
for e in t.edges {
assert!(
g.edges_for(e.src_label, e.rel, e.dst_label) > 0,
"scale {scale}: junction {} projects edge {}-[{}]->{}",
t.name,
e.src_label,
e.rel,
e.dst_label
);
}
}
}
// …and no vertex label named after a junction ever leaked in.
for junk in [
"OrderItem",
"OrderItems",
"ProductSupplier",
"ProductSuppliers",
] {
assert!(
!g.labels.contains(junk),
"scale {scale}: no spurious `{junk}` vertex"
);
}
// ── Rule: no dangling endpoints ───────────────────────────────────────────
assert_eq!(
g.dangling, 0,
"scale {scale}: every non-null edge endpoint resolves to a node"
);
// ── Rule: self-referential FK maps correctly (Employee-REPORTS_TO->Employee)
let emp_rows = db.row_count("erp.hr.employees");
let reports = g.edges_for("Employee", "REPORTS_TO", "Employee");
let managers = nonnull_count(&db, "erp.hr.employees", "manager_id");
assert_eq!(
reports, managers,
"scale {scale}: REPORTS_TO == employees with a manager"
);
assert!(
reports < emp_rows,
"scale {scale}: org roots (null manager) are NOT edges"
);
// Self-loop: both endpoints share the Employee label / key set.
assert!(reports > 0, "scale {scale}: some REPORTS_TO edges exist");
// The null-manager roots were skipped as "no edge", not counted as dangling.
assert!(
g.null_endpoint_rows >= emp_rows - managers,
"scale {scale}: null FKs skipped, not dangled"
);
// ── Rule: composite-key junctions resolve on BOTH keys ────────────────────
// order_items → Order-CONTAINS->Product (keys order_id + product_id);
// product_suppliers → Product-SUPPLIED_BY->Supplier. Their edge counts equal
// the generated junction row counts (every composite row = one resolved edge).
assert_eq!(
g.edges_for("Order", "CONTAINS", "Product"),
db.row_count("erp.sales.order_items"),
"scale {scale}: every (order,product) composite row is one CONTAINS edge"
);
assert_eq!(
g.edges_for("Product", "SUPPLIED_BY", "Supplier"),
db.row_count("erp.catalog.product_suppliers"),
"scale {scale}: every (product,supplier) composite row is one SUPPLIED_BY edge"
);
}
#[test]
fn correctness_at_scale_1() {
check_scale(1);
}
#[test]
fn correctness_at_scale_10() {
check_scale(10);
}
/// The invariants are scale-free: node/edge counts scale exactly linearly with
/// the node tables (junctions grow with their parents, so within a band).
#[test]
fn totals_grow_with_scale() {
let spec = mapping_spec();
let g1 = summarize(&build_mutation(&spec, &generate(1, DEFAULT_SEED)));
let g10 = summarize(&build_mutation(&spec, &generate(10, DEFAULT_SEED)));
assert_eq!(
g1.total_nodes() * 10,
g10.total_nodes(),
"node total scales x10"
);
assert!(
g10.total_edges() > g1.total_edges() * 5,
"edge total grows with scale"
);
assert_eq!(g1.dangling, 0);
assert_eq!(g10.dangling, 0);
}
/// **The two projections of the ONE plan agree on shape** — and neither invents
/// a junction vertex. `GraphPlan::from_mapping` yields exactly 10 node plans (no
/// junction table among them) + 11 edge plans; the thund projection yields the
/// matching `node__*` / `edge__*` datasets.
#[test]
fn plan_and_thund_projection_have_no_junction_vertex() {
let spec = mapping_spec();
let plan = GraphPlan::from_mapping(&spec, &CodegenOptions::default()).expect("plan builds");
assert_eq!(plan.nodes.len(), 10, "one node plan per entity table");
assert_eq!(
plan.edges.len(),
11,
"one edge plan per FK / junction relationship"
);
for n in &plan.nodes {
assert!(
!n.table.contains("order_items") && !n.table.contains("product_suppliers"),
"no NODE plan is sourced from a junction table (got `{}`)",
n.table
);
}
// The junction relationships are present as EDGES in the plan.
let rels: BTreeSet<&str> = plan.edges.iter().map(|e| e.rel.as_str()).collect();
assert!(rels.contains("CONTAINS") && rels.contains("SUPPLIED_BY"));
assert!(rels.contains("REPORTS_TO"), "self-ref edge present in plan");
// Second projection: bifröst GraphPlan → thund Pipeline.
let pipeline = knut_thund::rel2graph::to_thund_pipeline(&plan).expect("thund projection");
let node_ds = pipeline
.datasets
.iter()
.filter(|d| d.name.starts_with("node__"))
.count();
let edge_ds = pipeline
.datasets
.iter()
.filter(|d| d.name.starts_with("edge__"))
.count();
assert_eq!(node_ds, 10, "thund pipeline: 10 node datasets");
assert_eq!(edge_ds, 11, "thund pipeline: 11 edge datasets");
for d in &pipeline.datasets {
assert!(
!d.name.contains("OrderItem") && !d.name.contains("ProductSupplier"),
"no junction node dataset (`{}`)",
d.name
);
}
}
/// **Full 12-table PROJECTION PARITY.** The one `GraphPlan` fans out to two
/// backend projections — the PySpark job (`GraphPlan::to_pyspark`, the Spark
/// path) and the thund `Pipeline` (`to_thund_pipeline`, the native path). This
/// asserts they describe the SAME graph over the realistic 12-table ERP schema,
/// down to endpoint KEYS — so the composite-key junction edges (CONTAINS,
/// SUPPLIED_BY), the self-referential REPORTS_TO, and every FK edge map
/// identically on both sides. Each side's facts are parsed INDEPENDENTLY (the
/// generated Python on one, the pipeline's `knut.*` properties on the other);
/// any divergence — a renamed label, a flipped edge, a wrong merge key, a
/// different batch size, or a leaked junction vertex — fails here.
///
/// This is the scaled-up sibling of `knut_thund::rel2graph::tests::
/// two_projections_agree` (a 2-table toy): the same invariant, but on the full
/// bench schema and INCLUDING the from/to endpoint keys the toy test omits.
#[test]
fn twelve_table_projections_agree_down_to_keys() {
let spec = mapping_spec();
let plan = GraphPlan::from_mapping(&spec, &CodegenOptions::default()).expect("plan builds");
let py = plan.to_pyspark();
let pipe = knut_thund::rel2graph::to_thund_pipeline(&plan).expect("thund projection");
// ── nodes: (label, merge-key) from BOTH sides, sorted, compared ───────────
let py_nodes = pyspark_nodes(&py);
let thund_nodes = thund_nodes(&pipe);
assert_eq!(
py_nodes, thund_nodes,
"node labels+merge-keys agree across the PySpark and thund projections"
);
assert_eq!(py_nodes.len(), 10, "exactly the 10 entity node labels");
// ── edges: (from, rel, to, from_col, to_col, from_key, to_key) both sides ─
// The full tuple — endpoint READ COLUMNS *and* MERGE KEYS — so composite-key
// junctions and the self-ref edge are pinned to map identically on both
// projections, VALUE column and MERGE property alike.
let py_edges = pyspark_edges(&py);
let thund_edges = thund_edges(&pipe);
assert_eq!(
py_edges, thund_edges,
"edge directions, read columns AND merge keys agree across the two projections"
);
assert_eq!(
py_edges.len(),
11,
"exactly the 11 FK / junction relationships"
);
// ── batch size agrees ─────────────────────────────────────────────────────
assert_eq!(
pyspark_batch(&py),
thund_batch(&pipe),
"batch_rows agrees across the two projections"
);
// ── the hard cases are present and mapped correctly on BOTH sides ─────────
// Composite-key junctions: CONTAINS reads (order_id, product_id) and MERGEs
// on the same; SUPPLIED_BY reads (product_id, supplier_id).
let contains = edge(
"Order",
"CONTAINS",
"Product",
"order_id",
"product_id",
"order_id",
"product_id",
);
assert!(
py_edges.contains(&contains),
"PySpark: composite-key CONTAINS reads+MERGEs (order_id, product_id)"
);
assert!(
thund_edges.contains(&contains),
"thund: composite-key CONTAINS reads+MERGEs (order_id, product_id)"
);
let supplied = edge(
"Product",
"SUPPLIED_BY",
"Supplier",
"product_id",
"supplier_id",
"product_id",
"supplier_id",
);
assert!(
py_edges.contains(&supplied),
"PySpark: SUPPLIED_BY reads+MERGEs (product_id, supplier_id)"
);
assert!(
thund_edges.contains(&supplied),
"thund: SUPPLIED_BY reads+MERGEs (product_id, supplier_id)"
);
// Self-referential edge — the regression this fixes. REPORTS_TO must READ the
// FK columns (employee_id → manager_id) while MERGEing both endpoints on the
// Employee node key (employee_id). Reading the merge key as the value (the
// old bug) would give `employee_id AS dst` — a self-loop of every employee to
// itself. Read column ≠ merge key here, so this pins the split.
let reports_to = edge(
"Employee",
"REPORTS_TO",
"Employee",
/* read */ "employee_id",
"manager_id",
/* merge */ "employee_id",
"employee_id",
);
assert!(
py_edges.contains(&reports_to),
"PySpark: self-ref REPORTS_TO reads employee_id→manager_id, MERGEs on employee_id (no self-loop)"
);
assert!(
thund_edges.contains(&reports_to),
"thund: self-ref REPORTS_TO reads employee_id→manager_id, MERGEs on employee_id (no self-loop)"
);
// Guard the regression explicitly: no edge reads the SAME column for both
// endpoints (which would be a self-loop) unless it is genuinely a same-column
// link — none in this schema, so `from_col != to_col` for every edge.
for (f, r, t, fc, tc, _, _) in &thund_edges {
assert_ne!(
fc, tc,
"edge {f}-[{r}]->{t} reads two DISTINCT endpoint columns (not a self-loop)"
);
}
// ── no junction vertex leaked into EITHER projection ──────────────────────
for (label, _) in &py_nodes {
assert!(
!label.contains("OrderItem") && !label.contains("ProductSupplier"),
"PySpark: no junction vertex `{label}`"
);
}
for (label, _) in &thund_nodes {
assert!(
!label.contains("OrderItem") && !label.contains("ProductSupplier"),
"thund: no junction vertex `{label}`"
);
}
}
// ── projection-fact extractors: parsed straight from each emitter's output, so
// the two sides are derived INDEPENDENTLY and only meet at `assert_eq!`. ────
type EdgeFacts = (String, String, String, String, String, String, String);
/// `(from, rel, to, from_col, to_col, from_key, to_key)` — the full edge shape:
/// direction, the two VALUE columns read from the edge table, and the two MERGE
/// key properties the endpoint nodes are matched by.
#[allow(clippy::too_many_arguments)]
fn edge(f: &str, r: &str, t: &str, fc: &str, tc: &str, fk: &str, tk: &str) -> EdgeFacts {
(
f.into(),
r.into(),
t.into(),
fc.into(),
tc.into(),
fk.into(),
tk.into(),
)
}
/// (label, key) node facts parsed from the PySpark job's startup index lines:
/// `g.query('CREATE INDEX FOR (n:`Label`) ON (n.`key`)')`.
fn pyspark_nodes(py: &str) -> Vec<(String, String)> {
let mut out = Vec::new();
for line in py.lines() {
if let Some(rest) = line.split_once("CREATE INDEX FOR (n:`") {
let (label, tail) = rest.1.split_once("`) ON (n.`").unwrap();
let key = tail.split_once("`)").unwrap().0;
out.push((label.to_string(), key.to_string()));
}
}
out.sort();
out
}
/// Full edge facts parsed from the PySpark job. The MERGE keys come from the
/// `_write_edges("From", "REL", "To", "from_key", "to_key", rows)` CALL sites;
/// the READ columns come from the immediately-preceding
/// `selectExpr("`from_col` AS src", "`to_col` AS dst", …)` line — so this pins
/// BOTH halves of the endpoint mapping the way Spark actually runs it.
fn pyspark_edges(py: &str) -> Vec<EdgeFacts> {
let mut out = Vec::new();
// The last-seen edge selectExpr's read columns, paired with the next
// `_write_edges` call (they are emitted as two consecutive lines in batch
// mode: the `df = spark.table(...).selectExpr(...)` then its foreachPartition).
let mut pending: Option<(String, String)> = None;
for line in py.lines() {
if line.trim_start().starts_with("def ") {
continue; // the writer DEFINITIONS, not call sites
}
if line.contains("AS src") && line.contains("AS dst") {
if let (Some(fc), Some(tc)) =
(col_before(line, "` AS src"), col_before(line, "` AS dst"))
{
pending = Some((fc, tc));
}
}
if let Some((_, rest)) = line.split_once("_write_edges(") {
let args: Vec<&str> = rest.split(", ").collect();
let unq = |s: &str| s.trim().trim_matches('"').to_string();
let (fc, tc) = pending
.take()
.expect("an edge selectExpr precedes its _write_edges");
out.push((
unq(args[0]),
unq(args[1]),
unq(args[2]),
fc,
tc,
unq(args[3]),
unq(args[4]),
));
}
}
out.sort();
out
}
/// Extract the backtick-quoted column immediately left of `marker`
/// (`"` AS src"` / `"` AS dst"`), i.e. the `col` in `` `col` AS src``.
fn col_before(line: &str, marker: &str) -> Option<String> {
let idx = line.find(marker)?;
let left = &line[..idx];
let open = left.rfind('`')?;
Some(left[open + 1..].to_string())
}
fn pyspark_batch(py: &str) -> String {
py.lines()
.find_map(|l| l.strip_prefix("BATCH_ROWS = "))
.unwrap()
.to_string()
}
/// (label, key) node facts read from the thund pipeline's `knut.node.*` props.
fn thund_nodes(p: &knut_thund::ir::Pipeline) -> Vec<(String, String)> {
use knut_thund::rel2graph::prop;
let mut out: Vec<(String, String)> = p
.datasets
.iter()
.filter_map(|d| {
Some((
d.properties.get(prop::NODE_LABEL)?.clone(),
d.properties.get(prop::NODE_KEY)?.clone(),
))
})
.collect();
out.sort();
out
}
/// Full edge facts (direction, read columns, merge keys) from the thund
/// `knut.edge.*` props.
fn thund_edges(p: &knut_thund::ir::Pipeline) -> Vec<EdgeFacts> {
use knut_thund::rel2graph::prop;
let mut out: Vec<EdgeFacts> = p
.datasets
.iter()
.filter_map(|d| {
Some((
d.properties.get(prop::EDGE_FROM)?.clone(),
d.properties.get(prop::EDGE_REL)?.clone(),
d.properties.get(prop::EDGE_TO)?.clone(),
d.properties.get(prop::EDGE_FROM_COL)?.clone(),
d.properties.get(prop::EDGE_TO_COL)?.clone(),
d.properties.get(prop::EDGE_FROM_KEY)?.clone(),
d.properties.get(prop::EDGE_TO_KEY)?.clone(),
))
})
.collect();
out.sort();
out
}
fn thund_batch(p: &knut_thund::ir::Pipeline) -> String {
use knut_thund::rel2graph::prop;
p.datasets
.iter()
.find_map(|d| d.properties.get(prop::BATCH_ROWS).cloned())
.unwrap()
}
/// **The dangling check bites.** Point an FK at a non-existent parent and the
/// summary must count it as dangling — proving the "no dangling" assertion above
/// is not vacuously true.
#[test]
fn dangling_detection_is_not_vacuous() {
use arrow_array::{Array, Int64Array, RecordBatch};
use std::sync::Arc;
let spec = mapping_spec();
let db = generate(1, DEFAULT_SEED);
// Rebuild the payments batch with order_id forced to a bogus id (9_999_999).
let mut mutation = build_mutation(&spec, &db);
let corrupt = mutation
.edges
.iter_mut()
.find(|e| e.edge_type == "PAYS")
.expect("PAYS edge exists");
let b = &corrupt.batch;
let n = b.num_rows();
let src = b.column(b.schema().index_of("src").unwrap()).clone();
let bogus: Int64Array = (0..n).map(|_| Some(9_999_999i64)).collect();
let schema = b.schema();
let cols: Vec<Arc<dyn Array>> = schema
.fields()
.iter()
.enumerate()
.map(|(i, f)| {
if f.name() == "dst" {
Arc::new(bogus.clone()) as _
} else if f.name() == "src" {
src.clone()
} else {
b.column(i).clone()
}
})
.collect();
corrupt.batch = RecordBatch::try_new(schema, cols).unwrap();
let g = summarize(&mutation);
assert!(
g.dangling >= n,
"corrupted FK is detected as dangling (got {})",
g.dangling
);
}