crates/knut-bifrost/tests/live_falkordb_sink.rs
//! **Live FalkorDB sink proof.**
//!
//! Drives a `GraphMutation` — built by the REAL bifröst rel2graph engine
//! (`rel2graph::build_mutation` over an in-memory [`ArrowTables`] source, plus
//! one hand-built parallel-edge set) — through [`FalkorDbSink::apply`] into a
//! reachable FalkorDB, then reads the graph back LIVE and asserts every
//! relational pattern a real Swedish-myndighet schema needs landed correctly:
//!
//! * **node / edge counts** match the hand-computed oracle;
//! * **composite-key junction** (`Arende-[:HAR_BESLUT]->Beslut`, a junction
//! table with a composite PK and NO vertex) resolves both endpoints;
//! * **self-referential FK** (`Handlaggare-[:REPORTS_TO]->Handlaggare`) has NO
//! self-loop and the org root (null manager) is NOT an edge;
//! * **nullable FK = no edge** — an unassigned case mints neither a null-keyed
//! node nor a dangling edge;
//! * **parallel edges of different rel-types** between the same node pair
//! (`BESLUTAR` + `GRANSKAR`, same `(Handlaggare, Arende)`) both survive;
//! * **parallel edges of the SAME rel-type** kept distinct by an
//! `edge_key_props` merge key (`LOGGADE` twice, different `ts`);
//! * **typed property projection** — int / float / bool / string properties
//! round-trip with their Cypher types intact.
//!
//! `#[ignore]`-gated so CI (no DB) skips it; run it on a host with FalkorDB:
//!
//! ```bash
//! cargo test -p knut-bifrost --test live_falkordb_sink -- --ignored --nocapture
//! ```
//!
//! It uses a UNIQUE throwaway graph name and `GRAPH.DELETE`s it at the end, so
//! it never touches a seeded graph such as `korp_demo`. Point it elsewhere with
//! `KNUT_FALKORDB_URL` (default `redis://127.0.0.1:6379`).
use std::sync::Arc;
use arrow_array::{BooleanArray, Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use knut_bifrost::{
MappingSpec,
mutation::{EdgeSet, GraphMutation},
rel2graph::{ArrowTables, build_mutation},
sink::{FalkorDbSink, GraphSink},
};
// ── fixture: a compact Swedish-myndighet case schema ─────────────────────────
fn schema(fields: Vec<Field>) -> Arc<Schema> {
Arc::new(Schema::new(fields))
}
/// Person nodes (the applicants). Typed props: `namn` (str), `fodelsear` (int).
fn persons() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("person_id", DataType::Int64, false),
Field::new("namn", DataType::Utf8, false),
Field::new("fodelsear", DataType::Int64, false),
]),
vec![
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(StringArray::from(vec!["Anna", "Björn", "Cecilia"])),
Arc::new(Int64Array::from(vec![1980, 1975, 1990])),
],
)
.unwrap()
}
/// Caseworker nodes with a SELF-REFERENTIAL, NULLABLE manager FK (`chef_id`).
/// Handläggare 10 is the org root (null chef).
fn handlaggare() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("handlaggare_id", DataType::Int64, false),
Field::new("chef_id", DataType::Int64, true), // nullable self-ref FK
Field::new("namn", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![10, 11, 12])),
Arc::new(Int64Array::from(vec![None, Some(10), Some(10)])),
Arc::new(StringArray::from(vec!["Chef", "H1", "H2"])),
],
)
.unwrap()
}
/// Case nodes. FK `sokande_id` → Person (mandatory), FK `handlaggare_id` →
/// Handlaggare (NULLABLE — arende 102 is unassigned). Typed props: `prioritet`
/// (int), `oppen` (bool), `belopp` (float).
fn arenden() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("arende_id", DataType::Int64, false),
Field::new("sokande_id", DataType::Int64, false),
Field::new("handlaggare_id", DataType::Int64, true), // nullable FK
Field::new("prioritet", DataType::Int64, false),
Field::new("oppen", DataType::Boolean, false),
Field::new("belopp", DataType::Float64, false),
]),
vec![
Arc::new(Int64Array::from(vec![100, 101, 102])),
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(Int64Array::from(vec![Some(11), Some(11), None])),
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(BooleanArray::from(vec![true, false, true])),
Arc::new(Float64Array::from(vec![1234.5, 50.0, 9999.9])),
],
)
.unwrap()
}
/// Decision nodes.
fn beslut() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("beslut_id", DataType::Int64, false),
Field::new("utfall", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![200, 201])),
Arc::new(StringArray::from(vec!["bifall", "avslag"])),
],
)
.unwrap()
}
/// JUNCTION table with a COMPOSITE PK `(arende_id, beslut_id)` → the M:N
/// `Arende-[:HAR_BESLUT]->Beslut` edge. No vertex step: the junction must NOT
/// mint a node label.
fn arende_beslut() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("arende_id", DataType::Int64, false),
Field::new("beslut_id", DataType::Int64, false),
Field::new("datum", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![100, 100, 101])),
Arc::new(Int64Array::from(vec![200, 201, 200])),
Arc::new(StringArray::from(vec![
"2026-01-01",
"2026-02-01",
"2026-03-01",
])),
],
)
.unwrap()
}
/// A single caseworker↔case handling row that projects to TWO edge types
/// (`BESLUTAR` + `GRANSKAR`) between the SAME node pair — parallel edges of
/// different rel-types.
fn handlaggning() -> RecordBatch {
RecordBatch::try_new(
schema(vec![
Field::new("handlaggare_id", DataType::Int64, false),
Field::new("arende_id", DataType::Int64, false),
]),
vec![
Arc::new(Int64Array::from(vec![11])),
Arc::new(Int64Array::from(vec![100])),
],
)
.unwrap()
}
const MAPPING: &str = r#"
version: 1
tables:
- table: mnd.person
pipeline:
- vertex: { label: Person, id_column: person_id, properties: [namn, fodelsear] }
- table: mnd.handlaggare
pipeline:
- vertex: { label: Handlaggare, id_column: handlaggare_id, properties: [namn] }
- fk_edges:
type: REPORTS_TO
src: { label: Handlaggare, column: handlaggare_id }
dst: { label: Handlaggare, column: chef_id, id_column: handlaggare_id }
- table: mnd.arende
pipeline:
- vertex: { label: Arende, id_column: arende_id, properties: [prioritet, oppen, belopp] }
- fk_edges:
type: SOKT_AV
src: { label: Arende, column: arende_id }
dst: { label: Person, column: sokande_id, id_column: person_id }
- fk_edges:
type: HANDLAGGS_AV
src: { label: Arende, column: arende_id }
dst: { label: Handlaggare, column: handlaggare_id }
- table: mnd.beslut
pipeline:
- vertex: { label: Beslut, id_column: beslut_id, properties: [utfall] }
- table: mnd.arende_beslut
pipeline:
- fk_edges:
type: HAR_BESLUT
src: { label: Arende, column: arende_id }
dst: { label: Beslut, column: beslut_id }
properties: [datum]
- table: mnd.handlaggning
pipeline:
- fk_edges:
type: BESLUTAR
src: { label: Handlaggare, column: handlaggare_id }
dst: { label: Arende, column: arende_id }
- fk_edges:
type: GRANSKAR
src: { label: Handlaggare, column: handlaggare_id }
dst: { label: Arende, column: arende_id }
"#;
/// Build the full mutation: the declarative engine output PLUS one hand-built
/// `LOGGADE` edge set carrying an `edge_key_props` merge key, so two rows with
/// distinct `ts` survive as PARALLEL edges of the same type (a shape the
/// declarative pipeline deliberately never emits — this proves the sink seam).
fn full_mutation() -> GraphMutation {
let spec = MappingSpec::from_yaml(MAPPING).expect("mapping parses");
let src = ArrowTables::new()
.with_table("mnd.person", vec![persons()])
.with_table("mnd.handlaggare", vec![handlaggare()])
.with_table("mnd.arende", vec![arenden()])
.with_table("mnd.beslut", vec![beslut()])
.with_table("mnd.arende_beslut", vec![arende_beslut()])
.with_table("mnd.handlaggning", vec![handlaggning()]);
let mut mutation = build_mutation(&spec, &src).expect("engine builds mutation");
// Parallel edges of the SAME type, kept distinct by `ts` (edge_key_props).
let loggade = RecordBatch::try_new(
schema(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("ts", DataType::Utf8, false),
]),
vec![
Arc::new(Int64Array::from(vec![11, 11])),
Arc::new(Int64Array::from(vec![100, 100])),
Arc::new(StringArray::from(vec![
"2026-05-01T10:00:00",
"2026-05-01T11:00:00",
])),
],
)
.unwrap();
mutation.edges.push(EdgeSet {
src_label: "Handlaggare".into(),
edge_type: "LOGGADE".into(),
dst_label: "Arende".into(),
src_id_column: "handlaggare_id".into(),
dst_id_column: "arende_id".into(),
edge_key_props: vec!["ts".into()],
replace_outgoing: false,
batch: loggade,
});
mutation
}
// ── live-query helpers ───────────────────────────────────────────────────────
fn first_int(v: &redis::Value) -> Option<i64> {
match v {
redis::Value::Int(i) => Some(*i),
redis::Value::Array(items) | redis::Value::Set(items) => items.iter().find_map(first_int),
redis::Value::BulkString(b) => String::from_utf8_lossy(b).trim().parse().ok(),
_ => None,
}
}
async fn count(conn: &mut redis::aio::MultiplexedConnection, graph: &str, q: &str) -> i64 {
let v: redis::Value = redis::cmd("GRAPH.QUERY")
.arg(graph)
.arg(q)
.query_async(conn)
.await
.unwrap_or_else(|e| panic!("GRAPH.QUERY `{q}` failed: {e}"));
first_int(&v).unwrap_or(-1)
}
async fn graph_list(conn: &mut redis::aio::MultiplexedConnection) -> Vec<String> {
let v: redis::Value = redis::cmd("GRAPH.LIST").query_async(conn).await.unwrap();
fn walk(v: &redis::Value, out: &mut Vec<String>) {
match v {
redis::Value::BulkString(b) => out.push(String::from_utf8_lossy(b).to_string()),
redis::Value::SimpleString(s) => out.push(s.clone()),
redis::Value::Array(items) | redis::Value::Set(items) => {
items.iter().for_each(|it| walk(it, out))
}
_ => {}
}
}
let mut out = Vec::new();
walk(&v, &mut out);
out
}
#[tokio::test]
#[ignore = "needs a reachable FalkorDB (set KNUT_FALKORDB_URL; default redis://127.0.0.1:6379); run with --ignored"]
async fn live_mutation_lands_and_reads_back_every_pattern() {
let url =
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into());
let graph = format!(
"knut_live_sink_{}_{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
);
// ── apply the mutation through the real sink ─────────────────────────────
let mut sink = FalkorDbSink::connect(&url, graph.clone())
.await
.expect("connect FalkorDbSink");
let mutation = full_mutation();
let report = sink.apply(&mutation).await.expect("sink applies mutation");
eprintln!("live sink report: {report:?}");
// ── read the graph back over an independent connection ───────────────────
let client = redis::Client::open(url.clone()).unwrap();
let mut conn = client.get_multiplexed_async_connection().await.unwrap();
// Gather every fact BEFORE asserting, then DELETE, so cleanup always runs.
let listed = graph_list(&mut conn).await.contains(&graph);
let nodes = count(&mut conn, &graph, "MATCH (n) RETURN count(n)").await;
let edges = count(&mut conn, &graph, "MATCH ()-[r]->() RETURN count(r)").await;
let persons_n = count(&mut conn, &graph, "MATCH (n:Person) RETURN count(n)").await;
let handlaggare_n = count(&mut conn, &graph, "MATCH (n:Handlaggare) RETURN count(n)").await;
let arende_n = count(&mut conn, &graph, "MATCH (n:Arende) RETURN count(n)").await;
let beslut_n = count(&mut conn, &graph, "MATCH (n:Beslut) RETURN count(n)").await;
// composite-key junction
let har_beslut = count(
&mut conn,
&graph,
"MATCH (:Arende)-[r:HAR_BESLUT]->(:Beslut) RETURN count(r)",
)
.await;
let arende100_beslut = count(
&mut conn,
&graph,
"MATCH (:Arende {arende_id: 100})-[:HAR_BESLUT]->(b:Beslut) RETURN count(b)",
)
.await;
// self-ref: total, NO self-loop, NO null-keyed node
let reports_to = count(
&mut conn,
&graph,
"MATCH (:Handlaggare)-[r:REPORTS_TO]->(:Handlaggare) RETURN count(r)",
)
.await;
let self_loops = count(
&mut conn,
&graph,
"MATCH (h:Handlaggare)-[r:REPORTS_TO]->(h) RETURN count(r)",
)
.await;
let null_handlaggare = count(
&mut conn,
&graph,
"MATCH (h:Handlaggare) WHERE h.handlaggare_id IS NULL RETURN count(h)",
)
.await;
// nullable FK = no edge (arende 102 unassigned)
let handlaggs_av = count(
&mut conn,
&graph,
"MATCH (:Arende)-[r:HANDLAGGS_AV]->(:Handlaggare) RETURN count(r)",
)
.await;
let arende102_handlaggare = count(
&mut conn,
&graph,
"MATCH (:Arende {arende_id: 102})-[r:HANDLAGGS_AV]->() RETURN count(r)",
)
.await;
// parallel edges of DIFFERENT types between the same pair (11 → 100)
let besluter = count(
&mut conn,
&graph,
"MATCH (:Handlaggare {handlaggare_id: 11})-[r:BESLUTAR]->(:Arende {arende_id: 100}) RETURN count(r)",
)
.await;
let granskar = count(
&mut conn,
&graph,
"MATCH (:Handlaggare {handlaggare_id: 11})-[r:GRANSKAR]->(:Arende {arende_id: 100}) RETURN count(r)",
)
.await;
let distinct_types_11_100 = count(
&mut conn,
&graph,
"MATCH (:Handlaggare {handlaggare_id: 11})-[r]->(:Arende {arende_id: 100}) RETURN count(DISTINCT type(r))",
)
.await;
// parallel edges of the SAME type kept distinct by edge_key_props (ts)
let loggade = count(
&mut conn,
&graph,
"MATCH (:Handlaggare {handlaggare_id: 11})-[r:LOGGADE]->(:Arende {arende_id: 100}) RETURN count(r)",
)
.await;
// typed property projection (int / float / bool / string round-trip)
let typed_int = count(
&mut conn,
&graph,
"MATCH (p:Person {person_id: 1}) WHERE p.fodelsear = 1980 RETURN count(p)",
)
.await;
let typed_bool = count(
&mut conn,
&graph,
"MATCH (a:Arende {arende_id: 100}) WHERE a.oppen = true RETURN count(a)",
)
.await;
let typed_float = count(
&mut conn,
&graph,
"MATCH (a:Arende {arende_id: 100}) WHERE a.belopp > 1000.0 AND a.belopp < 2000.0 RETURN count(a)",
)
.await;
let typed_str = count(
&mut conn,
&graph,
"MATCH (p:Person {person_id: 1}) WHERE p.namn = 'Anna' RETURN count(p)",
)
.await;
// ── teardown FIRST (so a failed assert never leaves the throwaway graph) ─
let _: redis::Value = redis::cmd("GRAPH.DELETE")
.arg(&graph)
.query_async(&mut conn)
.await
.expect("GRAPH.DELETE throwaway graph");
let gone = !graph_list(&mut conn).await.contains(&graph);
// ── assertions ───────────────────────────────────────────────────────────
assert!(listed, "GRAPH.LIST includes the throwaway graph `{graph}`");
assert_eq!(
nodes, 11,
"total nodes (3 Person + 3 Handlaggare + 3 Arende + 2 Beslut)"
);
assert_eq!(edges, 14, "total edges (2+3+2+3+1+1+2)");
assert_eq!(persons_n, 3);
assert_eq!(
handlaggare_n, 3,
"junctions minted no extra node; null root is a real node, not null-keyed"
);
assert_eq!(arende_n, 3);
assert_eq!(beslut_n, 2);
assert_eq!(
har_beslut, 3,
"composite-key junction HAR_BESLUT resolves every (arende, beslut) row"
);
assert_eq!(
arende100_beslut, 2,
"arende 100 has both its decisions (composite key on both endpoints)"
);
assert_eq!(
reports_to, 2,
"self-ref REPORTS_TO = the two non-root caseworkers"
);
assert_eq!(
self_loops, 0,
"NO self-loop: no caseworker reports to itself"
);
assert_eq!(
null_handlaggare, 0,
"the null-manager root minted NO null-keyed node"
);
assert_eq!(
handlaggs_av, 2,
"nullable FK: only the 2 assigned cases have a HANDLAGGS_AV edge"
);
assert_eq!(
arende102_handlaggare, 0,
"the unassigned case (null FK) has no edge = no edge"
);
assert_eq!(besluter, 1, "parallel-different-types: BESLUTAR present");
assert_eq!(granskar, 1, "parallel-different-types: GRANSKAR present");
assert_eq!(
distinct_types_11_100, 3,
"three distinct rel-types between the same pair (BESLUTAR, GRANSKAR, LOGGADE)"
);
assert_eq!(
loggade, 2,
"parallel-same-type: two LOGGADE edges kept distinct by edge_key_props (ts)"
);
assert_eq!(
typed_int, 1,
"typed int property fodelsear compares as an integer"
);
assert_eq!(
typed_bool, 1,
"typed bool property oppen compares as a boolean"
);
assert_eq!(
typed_float, 1,
"typed float property belopp compares as a float"
);
assert_eq!(
typed_str, 1,
"typed string property namn compares as a string"
);
// the sink report agrees with the live edge count and the oracle
assert_eq!(report.vertices_written, 11, "report vertices == live nodes");
assert_eq!(
report.edges_written, 14,
"report edges == live edges (null-FK rows not counted)"
);
assert!(
gone,
"throwaway graph `{graph}` deleted — korp_demo untouched"
);
}