crates/graphar-falkordb/tests/robot_falkor_graph.rs
//! **ROBOT — the live FalkorDB graph, asserted on what is IN it.**
//!
//! The existing live tier (`integration_falkordb`) reads back `count(n)` and
//! `count(r)`. That is a real read-back, and it is blind to the two failures
//! that actually happen to this loader:
//!
//! 1. **The empty MATCH.** [`FalkorDbLoader::insert_edges_keyed`] returns the
//! number of rows it *sent*, not the number of relationships FalkorDB
//! *created* (`total += 1` per input row, `loader.rs`). Its `UNWIND … MATCH
//! (a:L {k: …}) MATCH (b:M {k: …}) CREATE (a)-[…]->(b)` silently creates
//! nothing when the keys match nothing — and then reports `Ok(n)` and a GREEN
//! `functional_status` row saying "n edges". This is the shape of the live
//! `korp_demo` regression already documented in `create_index`'s doc comment:
//! *121200 nodes, 0 edges*.
//! 2. **The cartesian blow-up.** Drop the key predicate from either MATCH and
//! every src joins every dst. With 6 people and 3 companies that is 18
//! relationships instead of 4 — and `count(r) > 0` is perfectly happy.
//!
//! So this suite asserts the **applied output**: the exact set of node ids that
//! came back, the exact `(src, dst, property)` endpoint triples of every
//! relationship, and the per-node degree distribution. Every fixture value is
//! chosen off the identity: ids are 5001.. and 9001.. (never 0/1, which is also
//! what `_gar_id` would be if the key were ignored), the degree distribution is
//! 2/1/1 (never uniform, which cannot tell a keyed join from a cartesian one),
//! two people have NO edge at all, and `batch_size` is 2 against 6 rows so the
//! insert genuinely spans three windows.
//!
//! **Live tier** — needs a FalkorDB. Follows the repo convention (`#[ignore]`,
//! `$KNUT_FALKORDB_URL`, default `redis://127.0.0.1:6379`):
//!
//! ```bash
//! podman run -d --name knut-robotui-falkor -p 6399:6379 \
//! localhost/knut-falkordb-valkey9:latest
//! KNUT_FALKORDB_URL=redis://127.0.0.1:6399 \
//! cargo test -p graphar-falkordb --test robot_falkor_graph -- --ignored --nocapture
//! ```
use std::sync::Arc;
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use graphar_falkordb::{FalkorDbLoader, FalkorError, reply};
fn url() -> String {
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into())
}
/// Six people. `pnr` is the join key and is deliberately NOT `_gar_id`-shaped:
/// if the loader ever fell back to the row index the keys would be 0..5 and
/// every endpoint assertion below would move.
fn people() -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("pnr", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![
5001i64, 5002, 5003, 5004, 5005, 5006,
])),
Arc::new(StringArray::from(vec![
"alice", "bob", "carol", "dan", "eve", "frank",
])),
],
)
.unwrap()
}
/// Three companies, keyed on `orgnr`.
fn companies() -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("orgnr", DataType::Int64, false),
Field::new("cname", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![9001i64, 9002, 9003])),
Arc::new(StringArray::from(vec!["Ymer", "Bure", "Kvasir"])),
],
)
.unwrap()
}
/// The employment edges. Non-uniform on purpose:
/// * 9001 gets TWO employees, 9002 and 9003 one each → in-degree 2/1/1;
/// * 5004 and 5005 get NO edge → a cartesian join is impossible to mistake for
/// this;
/// * `since` is a distinct property per edge, so a mis-paired endpoint shows up
/// as a wrong triple and not just a wrong count.
///
/// `matching = false` rewrites the `pnr` carrier to ids that exist in NO Person
/// node — the "must produce NONE" half of the anchor.
fn works_at(matching: bool) -> RecordBatch {
let pnr: Vec<i64> = if matching {
vec![5001, 5002, 5003, 5006]
} else {
vec![7001, 7002, 7003, 7006]
};
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("pnr", DataType::Int64, false),
Field::new("orgnr", DataType::Int64, false),
Field::new("since", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(pnr)),
Arc::new(Int64Array::from(vec![9001i64, 9001, 9002, 9003])),
Arc::new(Int64Array::from(vec![2011i64, 2014, 2019, 2023])),
],
)
.unwrap()
}
/// The endpoint triples the keyed join MUST produce, in `p.pnr` order.
const EXPECTED_EDGES: [[&str; 3]; 4] = [
["5001", "9001", "2011"],
["5002", "9001", "2014"],
["5003", "9002", "2019"],
["5006", "9003", "2023"],
];
/// Fresh loader on a wiped graph, with a deliberately small window so the
/// 6-row/4-row inserts span several `UNWIND` batches (batch_size == row count
/// would never exercise the windowing).
async fn fresh(graph: &str) -> FalkorDbLoader {
let mut l = FalkorDbLoader::connect(&url(), graph)
.await
.unwrap_or_else(|e| panic!("connect to {} — {e}", url()));
// `GRAPH.DELETE` on a graph that does not exist is an error; ignore it.
let _ = l.run_query("MATCH (n) DETACH DELETE n").await;
l.with_batch_size(2).expect("batch size 2")
}
/// **The keyed join, asserted on its endpoints.**
///
/// Loads 6 Person + 3 Company + 4 keyed WORKS_AT edges into a live FalkorDB and
/// reads back the exact id sets, the exact endpoint triples and the exact
/// in-degree per company. A `count(r) == 4` assertion would pass on four WRONG
/// pairings; these do not.
#[tokio::test]
#[ignore = "live tier: needs a FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn robot_keyed_edges_land_on_the_right_endpoints() {
let mut l = fresh("robot_falkor_endpoints").await;
let np = l.insert_vertices("Person", &[people()]).await.unwrap();
let nc = l.insert_vertices("Company", &[companies()]).await.unwrap();
assert_eq!((np, nc), (6, 3), "the loader reports the rows it sent");
// Indexes on the join keys — the same order the population loader uses.
l.create_index("Person", "pnr")
.await
.expect("index Person.pnr");
l.create_index("Company", "orgnr")
.await
.expect("index Company.orgnr");
let sent = l
.insert_edges_keyed(
"Person",
"WORKS_AT",
"Company",
&["pnr"],
&["orgnr"],
&[works_at(true)],
)
.await
.unwrap();
assert_eq!(sent, 4, "four edge rows were sent");
// ── APPLIED OUTPUT 1: the exact node ids in the graph, not a count. ──
let got_pnr = reply::column0(
&l.run_query("MATCH (p:Person) RETURN p.pnr ORDER BY p.pnr")
.await
.unwrap(),
);
assert_eq!(
got_pnr,
vec!["5001", "5002", "5003", "5004", "5005", "5006"],
"every Person id must come back — a lost window drops ids, not just count"
);
let got_org = reply::column0(
&l.run_query("MATCH (c:Company) RETURN c.orgnr ORDER BY c.orgnr")
.await
.unwrap(),
);
assert_eq!(got_org, vec!["9001", "9002", "9003"]);
// ── APPLIED OUTPUT 2: the exact endpoint triples. ──
let got_edges = reply::rows(
&l.run_query(
"MATCH (p:Person)-[r:WORKS_AT]->(c:Company) \
RETURN p.pnr, c.orgnr, r.since ORDER BY p.pnr",
)
.await
.unwrap(),
);
let want: Vec<Vec<String>> = EXPECTED_EDGES
.iter()
.map(|t| t.iter().map(|s| s.to_string()).collect())
.collect();
assert_eq!(
got_edges, want,
"the keyed join must produce exactly these (person, company, since) triples"
);
// ── APPLIED OUTPUT 3: the degree distribution, 2/1/1 and not uniform. ──
let deg = reply::rows(
&l.run_query(
"MATCH (c:Company)<-[r:WORKS_AT]-() \
RETURN c.orgnr, count(r) ORDER BY c.orgnr",
)
.await
.unwrap(),
);
assert_eq!(
deg,
vec![vec!["9001", "2"], vec!["9002", "1"], vec!["9003", "1"],],
"in-degree must be 2/1/1 — a uniform distribution cannot distinguish a \
keyed join from a cartesian one"
);
// ── The anti-cartesian bound: 4, never 6×3 = 18. ──
let total = reply::scalar_i64(
&l.run_query("MATCH ()-[r:WORKS_AT]->() RETURN count(r)")
.await
.unwrap(),
);
assert_eq!(
total,
Some(4),
"exactly four relationships — 18 would mean the key predicate was dropped"
);
// ── And the two people with no employer really have none. ──
let unemployed = reply::column0(
&l.run_query("MATCH (p:Person) WHERE NOT (p)-[:WORKS_AT]->() RETURN p.pnr ORDER BY p.pnr")
.await
.unwrap(),
);
assert_eq!(
unemployed,
vec!["5004", "5005"],
"a graph where everyone is connected proves nothing about the join"
);
}
/// **The non-vacuous anchor, and the finding it pins — now the CLOSED contract.**
///
/// The SAME edge batch with `pnr` values that match no Person must create **zero**
/// relationships, and the loader must SAY SO.
///
/// # History, because the assertion below inverted
///
/// As filed (STINKY-19 R16, 2026-08-01) this test pinned the DEFECT: `MATCH … MATCH
/// … CREATE` succeeds whether or not the endpoints matched, and `insert_edges_keyed`
/// returned `Ok(4)` — one per input ROW — for a graph holding **0** relationships,
/// with a GREEN `functional_status` row to match. That is the live `korp_demo`
/// shape: *121 200 nodes, 0 edges, everything green.*
///
/// The design call the finding deferred has since been made in production code:
/// `loader.rs` now reads the server's own `Relationships created:` statistics
/// (`created_from_reply`) and `verdict()` turns "a non-empty batch created nothing"
/// into `FalkorError::NothingCreated`. So the return value no longer lies — and this
/// test asserted that it did, which left it **RED on knut `main`'s live tier**,
/// unnoticed because the whole tier is `#[ignore]`d (measured 2026-08-22 against a
/// live FalkorDB at `redis://127.0.0.1:6379`, valkey 9.1.0:
/// `called Result::unwrap() on an Err value: NothingCreated { what: "relationships",
/// rows: 4 }`).
///
/// It now pins the contract that replaced it, and still asserts BOTH halves and that
/// they DIFFER, so a loader that silently creates nothing can never be green.
#[tokio::test]
#[ignore = "live tier: needs a FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn robot_unmatched_keys_create_nothing_and_the_return_value_lies() {
// ── The half that MUST produce output. ──
let mut hit = fresh("robot_falkor_anchor_hit").await;
hit.insert_vertices("Person", &[people()]).await.unwrap();
hit.insert_vertices("Company", &[companies()])
.await
.unwrap();
let sent_hit = hit
.insert_edges_keyed(
"Person",
"WORKS_AT",
"Company",
&["pnr"],
&["orgnr"],
&[works_at(true)],
)
.await
.unwrap();
let made_hit = reply::scalar_i64(
&hit.run_query("MATCH ()-[r:WORKS_AT]->() RETURN count(r)")
.await
.unwrap(),
)
.expect("count(r) returned a row");
// ── The half that MUST produce NONE. Identical call, unmatchable keys. ──
let mut miss = fresh("robot_falkor_anchor_miss").await;
miss.insert_vertices("Person", &[people()]).await.unwrap();
miss.insert_vertices("Company", &[companies()])
.await
.unwrap();
// The loader must REFUSE this by name, not answer with a row count.
let miss_result = miss
.insert_edges_keyed(
"Person",
"WORKS_AT",
"Company",
&["pnr"],
&["orgnr"],
&[works_at(false)],
)
.await;
let made_miss = reply::scalar_i64(
&miss
.run_query("MATCH ()-[r:WORKS_AT]->() RETURN count(r)")
.await
.unwrap(),
)
.expect("count(r) returned a row");
// The vertices DID land in both graphs — so `made_miss == 0` is about the
// join, not about an empty database.
let miss_people = reply::scalar_i64(
&miss
.run_query("MATCH (p:Person) RETURN count(p)")
.await
.unwrap(),
);
assert_eq!(
miss_people,
Some(6),
"the miss graph is populated, only the join failed"
);
assert_eq!(made_hit, 4, "the matching keys built the graph");
assert_eq!(made_miss, 0, "the unmatchable keys built NOTHING");
assert_ne!(
made_hit, made_miss,
"the two cases must differ — if they did not, this whole suite would be vacuous"
);
// ── THE CONTRACT, pinned on both sides. ──
//
// The matching call reports what the SERVER created, not what was sent. The two
// happen to be equal here (4 rows in, 4 relationships out) and that is exactly
// why the miss half below has to exist: on this half alone, "rows sent" and
// "relationships created" are indistinguishable.
assert_eq!(
sent_hit as i64, made_hit,
"insert_edges_keyed must report the relationships the server CREATED \
(`Relationships created:` in the GRAPH.QUERY statistics), and the store \
says {made_hit}"
);
// The unmatchable call must FAIL BY NAME. `Ok(4)` here is the original R16
// defect verbatim — a caller (knut-popviz's `run_load` progress counters, the
// GREEN `functional_status` row) would read 4 edges into a graph with none,
// which is the korp_demo failure mode: nodes loaded, zero edges, all green.
match miss_result {
Err(FalkorError::NothingCreated { what, rows }) => {
assert_eq!(what, "relationships", "the refusal names what was not created");
assert_eq!(rows, 4, "…and how many rows were offered, so the message is actionable");
}
Err(other) => panic!(
"an unmatchable edge batch must be refused as NothingCreated, got a \
different error: {other}"
),
Ok(n) => panic!(
"insert_edges_keyed returned Ok({n}) for a batch that created ZERO \
relationships — that is STINKY-19 R16 back: the return value is rows \
SENT, not relationships CREATED, and the store holds {made_miss} edges."
),
}
}
/// **`_gar_id` across window boundaries.**
///
/// The legacy [`FalkorDbLoader::insert_edges`] path matches endpoints on
/// `_gar_id`, the *absolute* row index across all batches. With `batch_size` 2
/// and six rows split over two RecordBatches, a per-window or per-batch counter
/// would restart and the edges would join the wrong nodes — while still
/// producing the right *count*. Asserted on the name pairs, which move when the
/// ids do.
#[tokio::test]
#[ignore = "live tier: needs a FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn robot_gar_ids_stay_absolute_across_batches_and_windows() {
let mut l = fresh("robot_falkor_garid").await;
// Six people in TWO batches of 4 + 2, inserted through a window of 2.
let b1 = people().slice(0, 4);
let b2 = people().slice(4, 2);
let n = l.insert_vertices("Person", &[b1, b2]).await.unwrap();
assert_eq!(n, 6);
// _gar_id must be the ABSOLUTE index: alice=0 … frank=5.
let ids = reply::rows(
&l.run_query("MATCH (p:Person) RETURN p.name, p._gar_id ORDER BY p._gar_id")
.await
.unwrap(),
);
assert_eq!(
ids,
vec![
vec!["alice", "0"],
vec!["bob", "1"],
vec!["carol", "2"],
vec!["dan", "3"],
vec!["eve", "4"],
vec!["frank", "5"],
],
"_gar_id is the absolute row index across batches — a per-batch counter \
would repeat 0..3 then 0..1 and every edge below would land wrong"
);
// Edges that deliberately CROSS the batch boundary: 0→5 and 3→4.
let edges = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![0i64, 3])),
Arc::new(Int64Array::from(vec![5i64, 4])),
],
)
.unwrap();
l.insert_edges("Person", "KNOWS", "Person", &[edges])
.await
.unwrap();
let pairs = reply::rows(
&l.run_query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name, b.name ORDER BY a.name")
.await
.unwrap(),
);
assert_eq!(
pairs,
vec![vec!["alice", "frank"], vec!["dan", "eve"]],
"the cross-batch endpoints must be the named pairs, not merely two edges"
);
}