crates/graphar-falkordb/tests/live_keyed_edges.rs
//! **Live keyed-edge lane** — the ~2.7k edges/s boundary (benching doc,
//! 2026-10-19): endpoint-index coverage, the literal↔parameterized window
//! oracle, and the edges/s bench.
//!
//! Live tier — needs a FalkorDB. Repo convention (`#[ignore]`,
//! `$KNUT_FALKORDB_URL`, default `redis://127.0.0.1:6379`):
//!
//! ```bash
//! KNUT_FALKORDB_URL=redis://127.0.0.1:6399 \
//! cargo test -p graphar-falkordb --test live_keyed_edges -- --ignored --nocapture
//! ```
use std::sync::Arc;
use std::time::Instant;
use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use graphar_falkordb::FalkorDbLoader;
fn url() -> String {
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into())
}
/// Fresh loader on a wiped graph (same pattern as `robot_falkor_graph`).
async fn fresh(graph: &str) -> FalkorDbLoader {
let mut l = FalkorDbLoader::connect(&url(), graph)
.await
.unwrap_or_else(|e| panic!("connect to {} — {e}", url()));
let _ = l.run_query("MATCH (n) DETACH DELETE n").await;
l
}
fn node_batch(n: i64) -> RecordBatch {
let names: Vec<String> = (0..n).map(|i| format!("n{i}")).collect();
let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from((0..n).collect::<Vec<i64>>())),
Arc::new(StringArray::from(name_refs)),
],
)
.unwrap()
}
/// Chain edges `i -> i+1` over `_gar_id`, with a varied property set so the
/// oracle sees every renderer arm: an escaped string, a fractional and an
/// **integral** float (the "2.0" DOUBLE pin), and a null.
fn chain_edges(n: i64) -> RecordBatch {
let tags: Vec<Option<String>> = (0..n - 1)
.map(|i| {
if i % 7 == 0 {
None
} else {
Some(format!("t'{i}\\x"))
}
})
.collect();
let tag_refs: Vec<Option<&str>> = tags.iter().map(|t| t.as_deref()).collect();
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("w", DataType::Float64, false),
Field::new("tag", DataType::Utf8, true),
])),
vec![
Arc::new(Int64Array::from((0..n - 1).collect::<Vec<i64>>())),
Arc::new(Int64Array::from((1..n).collect::<Vec<i64>>())),
Arc::new(Float64Array::from(
(0..n - 1).map(|i| if i % 2 == 0 { 2.0 } else { 1.5 }).collect::<Vec<f64>>(),
)),
Arc::new(StringArray::from(tag_refs)),
],
)
.unwrap()
}
/// Debug-render of a reply's header+rows (stats carry timings; drop them).
fn result_rows(reply: &redis::Value) -> String {
match reply {
redis::Value::Array(items) if items.len() >= 2 => {
format!("{:?} {:?}", items[0], items[1])
}
other => format!("{other:?}"),
}
}
/// **RED-FIRST seam: the edge load must itself index its endpoint keys.**
///
/// The measured ~2.7k edges/s lane plans `Node By Label Scan` per endpoint when
/// nobody remembered `create_index` — O(N) per row. The loader knows its own
/// endpoint keys, so `insert_edges*` must guarantee the index, not hope the
/// caller did. RED before the fix: a fresh graph loaded without any manual
/// `create_index` call lists no `_gar_id` index.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn edge_load_creates_endpoint_indexes_itself() {
// Single legacy key.
let mut l = fresh("lane_autoidx").await;
l.insert_vertices("P", &[node_batch(50)]).await.expect("vertices");
l.insert_edges("P", "NEXT", "P", &[chain_edges(50)])
.await
.expect("edges");
let idx = l
.run_query("CALL db.indexes() YIELD label, properties RETURN label, properties")
.await
.expect("db.indexes()");
let rendered = format!("{idx:?}");
assert!(
rendered.contains("_gar_id") && rendered.contains('P'),
"edge load must have created the (P, _gar_id) endpoint index; db.indexes() = {rendered}"
);
// Composite key: both endpoint key properties must be indexed on the label.
let mut l = fresh("lane_autoidx_comp").await;
let people = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("country", DataType::Utf8, false),
Field::new("code", DataType::Int64, false),
])),
vec![
Arc::new(StringArray::from(vec!["SE", "SE", "NO"])),
Arc::new(Int64Array::from(vec![1i64, 2, 1])),
],
)
.unwrap();
l.insert_vertices("City", &[people]).await.expect("cities");
let routes = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("country", DataType::Utf8, false),
Field::new("code", DataType::Int64, false),
])),
vec![
Arc::new(StringArray::from(vec!["SE"])),
Arc::new(Int64Array::from(vec![1i64])),
],
)
.unwrap();
// Self-loop fixture: src == dst key columns, zero extra props.
l.insert_edges_keyed("City", "ROUTE", "City", &["country", "code"], &["country", "code"], &[routes])
.await
.expect("keyed edges");
let idx = l
.run_query("CALL db.indexes() YIELD label, properties RETURN label, properties")
.await
.expect("db.indexes()");
let rendered = format!("{idx:?}");
assert!(
rendered.contains("City") && rendered.contains("country") && rendered.contains("code"),
"composite edge load must have indexed BOTH (City, country) and (City, code); db.indexes() = {rendered}"
);
}
/// **ZERO-LOSS oracle: the loader's edge lane stores the same graph as the
/// legacy rendered-literal query.** Loads identical fixtures into two graphs —
/// one through raw `unwind_edge_query_keyed` literal windows, one through
/// `insert_edges` — and asserts the full endpoint+property read-back is equal,
/// row for row, byte for byte. Pins the literal→parameterized window swap.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn param_windows_store_identical_graph() {
const N: i64 = 300;
let nodes = node_batch(N);
let edges = chain_edges(N);
// Graph A: the legacy literal path, driven by hand through run_query so the
// oracle keeps measuring the historical bytes even after the loader moves on.
let mut a = fresh("lane_oracle_lit").await;
a.insert_vertices("P", &[nodes.clone()]).await.expect("A vertices");
a.create_index("P", "_gar_id").await.expect("A index");
{
use graphar_falkordb::cypher::{EdgeRow, arrow_value_to_cypher, unwind_edge_query_keyed};
let mut rows = Vec::new();
for r in 0..edges.num_rows() {
rows.push(EdgeRow {
src_id: vec![arrow_value_to_cypher(edges.column(0).as_ref(), r).unwrap()],
dst_id: vec![arrow_value_to_cypher(edges.column(1).as_ref(), r).unwrap()],
props: vec![
("w".into(), arrow_value_to_cypher(edges.column(2).as_ref(), r).unwrap()),
("tag".into(), arrow_value_to_cypher(edges.column(3).as_ref(), r).unwrap()),
],
});
}
for window in rows.chunks(64) {
let q = unwind_edge_query_keyed("P", "NEXT", "P", &["_gar_id"], &["_gar_id"], window);
a.run_query(&q).await.expect("A literal window");
}
}
// Graph B: the loader's production lane.
let mut b = fresh("lane_oracle_param").await;
b.insert_vertices("P", &[nodes]).await.expect("B vertices");
b.create_index("P", "_gar_id").await.expect("B index");
let created = b
.insert_edges("P", "NEXT", "P", &[edges])
.await
.expect("B edges");
assert_eq!(created, (N - 1) as usize, "loader must create every edge");
// Full read-back: endpoints + every property, deterministically ordered.
let read = "MATCH (x:P)-[r:NEXT]->(y:P) \
RETURN x._gar_id, y._gar_id, r.w, r.tag, toString(r.w) \
ORDER BY x._gar_id";
let ra = a.run_query(read).await.expect("A read");
let rb = b.run_query(read).await.expect("B read");
assert_eq!(
result_rows(&ra),
result_rows(&rb),
"literal and loader lanes must store identical graphs"
);
// And the stored node set is untouched/equal too.
let nread = "MATCH (n:P) RETURN n._gar_id, n.id, n.name ORDER BY n._gar_id";
let na = a.run_query(nread).await.expect("A nodes");
let nb = b.run_query(nread).await.expect("B nodes");
assert_eq!(result_rows(&na), result_rows(&nb), "node read-back must be identical");
}
/// **Bench — keyed edges/s** (print-only; run with `--nocapture`, alone:
/// `--exact bench_keyed_edges` so the other live tests don't share the timing).
///
/// Arm `lane_bench_main`: 100k nodes, 99,999 chain `_gar_id` edges with the
/// endpoint index stated explicitly first — the exact perf-scratch 2026-10-19
/// shape that measured ~2.7k edges/s. Arm `lane_bench_auto`: NO manual
/// `create_index`, small scale — pre-fix the loader planned a per-row label
/// scan here (measured live ~17 edges/s; 100k would be a ~70 h arm), post-fix
/// the lane indexes itself and must match the main arm's rate.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; bench — run with --ignored --nocapture"]
async fn bench_keyed_edges() {
for (graph, manual_index, n) in [
("lane_bench_main", true, 100_000i64),
("lane_bench_auto", false, 1_500i64),
] {
let nodes = node_batch(n);
let edges = chain_edges(n);
let mut l = fresh(graph).await;
let t = Instant::now();
let nv = l
.insert_vertices_par("P", std::slice::from_ref(&nodes), 12)
.await
.expect("vertices");
let nodes_s = t.elapsed().as_secs_f64();
if manual_index {
l.create_index("P", "_gar_id").await.expect("index");
}
let t = Instant::now();
let ne = l
.insert_edges("P", "NEXT", "P", std::slice::from_ref(&edges))
.await
.expect("edges");
let edges_s = t.elapsed().as_secs_f64();
assert_eq!(ne, (n - 1) as usize);
println!(
"BENCH {graph}: {nv} nodes in {nodes_s:.3}s = {:.0} nodes/s | {ne} edges in {edges_s:.3}s = {:.0} edges/s (manual_index={manual_index})",
nv as f64 / nodes_s,
ne as f64 / edges_s,
);
}
}