crates/graphar-falkordb/tests/integration_falkordb.rs
//! Live FalkorDB integration: bulk-load Arrow batches into a real FalkorDB
//! container via Podman (or Docker) and read them back with Cypher.
//!
//! ## Prerequisites
//!
//! - Podman socket: `systemctl --user start podman.socket`
//! - `export DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock`
//! (auto-detected below when unset)
//!
//! ## Image matrix
//!
//! `KNUT_FALKORDB_IMAGE` picks the server (default `localhost/knut-falkordb-valkey9:latest` — valkey-9, never stock redis).
//! For FalkorDB compiled against **Valkey 9** build the local image first:
//!
//! ```bash
//! podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .
//! KNUT_FALKORDB_IMAGE=localhost/knut-falkordb-valkey9:latest \
//! cargo test -p graphar-falkordb --features autostart --test integration_falkordb -- --ignored --nocapture
//! ```
//!
//! The container bring-up (`ensure_docker_host` + image resolution + `start`)
//! lives in `graphar_falkordb::falkordb_container` — the SAME helper the
//! `knut-popviz` binary uses — so this test needs `--features autostart`.
#![cfg(feature = "autostart")]
use std::sync::Arc;
use std::time::Instant;
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use graphar_falkordb::falkordb_container::start_falkordb;
use graphar_falkordb::{BulkEdgeGroup, BulkNodeGroup, FalkorDbLoader};
fn people_batch() -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("name", DataType::Utf8, false),
Field::new("age", DataType::Int64, false),
])),
vec![
Arc::new(StringArray::from(vec!["alice", "bob", "carol"])),
Arc::new(Int64Array::from(vec![30, 41, 25])),
],
)
.unwrap()
}
fn knows_batch() -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("since", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![0, 1])),
Arc::new(Int64Array::from(vec![1, 2])),
Arc::new(Int64Array::from(vec![2020, 2023])),
],
)
.unwrap()
}
/// Count scalar out of a GRAPH.QUERY reply (result set → first row → first cell).
fn scalar_i64(value: &redis::Value) -> Option<i64> {
fn walk(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(walk),
redis::Value::BulkString(b) => String::from_utf8_lossy(b).parse().ok(),
_ => None,
}
}
walk(value)
}
#[tokio::test]
#[ignore = "needs podman; run with --ignored"]
async fn load_and_query_live_falkordb() {
// Shared bring-up: probes docker host, resolves KNUT_FALKORDB_IMAGE, starts
// the container, and waits until the module answers — then hands back the
// guard + mapped URL. Keep `container` alive for the whole test.
let container = start_falkordb().await.expect("start FalkorDB container");
let url = container.url();
let mut loader = FalkorDbLoader::connect(url, "itest")
.await
.expect("connect to autostarted FalkorDB");
// GRAPH.QUERY must exist (proves the module is loaded — important for
// the Valkey 9 image, where we compiled it ourselves).
loader
.run_query("RETURN 1")
.await
.expect("GRAPH.QUERY available");
let inserted = loader
.insert_vertices("Person", &[people_batch()])
.await
.unwrap();
assert_eq!(inserted, 3);
let edges = loader
.insert_edges("Person", "KNOWS", "Person", &[knows_batch()])
.await
.unwrap();
assert_eq!(edges, 2);
let nodes = loader
.run_query("MATCH (n:Person) RETURN count(n)")
.await
.unwrap();
assert_eq!(scalar_i64(&nodes), Some(3), "raw reply: {nodes:?}");
let rels = loader
.run_query("MATCH (:Person)-[r:KNOWS]->(:Person) RETURN count(r)")
.await
.unwrap();
assert_eq!(scalar_i64(&rels), Some(2), "raw reply: {rels:?}");
// Property fidelity through the Arrow → Cypher path.
let age = loader
.run_query("MATCH (n:Person {name: 'bob'}) RETURN n.age")
.await
.unwrap();
assert_eq!(scalar_i64(&age), Some(41), "raw reply: {age:?}");
}
/// The parallel bulk path (`insert_vertices_par` / `insert_edges_par`) must
/// produce the SAME graph as the serial path — same returned counts AND the same
/// node/edge counts read back. RED-when-broken: if the concurrent dispatch
/// dropped, duplicated, or reordered-into-corruption any window, the counts
/// diverge and this fails. Loads the identical batches into two graphs (serial
/// vs parallel) on the one container and compares.
#[tokio::test]
#[ignore = "needs podman; run with --ignored"]
async fn parallel_bulk_matches_serial_counts() {
let container = start_falkordb().await.expect("start FalkorDB container");
let url = container.url();
// Serial reference load.
let mut serial = FalkorDbLoader::connect(url, "serial")
.await
.expect("connect serial");
let sv = serial
.insert_vertices("Person", &[people_batch()])
.await
.unwrap();
let se = serial
.insert_edges("Person", "KNOWS", "Person", &[knows_batch()])
.await
.unwrap();
// Parallel load of the SAME batches into a separate graph. `&self` — the
// par methods clone the connection per in-flight task. Vertices are a
// BARRIER before edges (edges MATCH endpoints that must already exist).
let parallel = FalkorDbLoader::connect(url, "parallel")
.await
.expect("connect parallel");
let pv = parallel
.insert_vertices_par("Person", &[people_batch()], 8)
.await
.unwrap();
let pe = parallel
.insert_edges_par("Person", "KNOWS", "Person", &[knows_batch()], 8)
.await
.unwrap();
assert_eq!(sv, pv, "parallel vertex count == serial");
assert_eq!(se, pe, "parallel edge count == serial");
// Read the parallel graph back — it must hold exactly the serial graph.
let mut check = FalkorDbLoader::connect(url, "parallel")
.await
.expect("reconnect parallel");
let nodes = check
.run_query("MATCH (n:Person) RETURN count(n)")
.await
.unwrap();
assert_eq!(scalar_i64(&nodes), Some(3), "parallel nodes: {nodes:?}");
let rels = check
.run_query("MATCH (:Person)-[r:KNOWS]->(:Person) RETURN count(r)")
.await
.unwrap();
assert_eq!(scalar_i64(&rels), Some(2), "parallel rels: {rels:?}");
}
/// **The `_gar_id` idempotency regression** (live, from korp): a graph that is
/// populated a SECOND time must not abort on the `_gar_id` index. Before the
/// fix, `create_index` re-issued `CREATE INDEX` on the already-indexed graph and
/// FalkorDB replied `Attribute: '_gar_id' is already indexed`, which propagated
/// out and killed the pipeline BEFORE the edge phase — leaving nodes but **zero
/// edges** (`korp_demo`: 121200 nodes, 0 edges).
///
/// This drives the real pipeline order — vertices → index → edges — TWICE over
/// the same graph and asserts:
/// 1. the second `create_index` returns `Ok` (no "already indexed" abort), and
/// 2. after the second pass the graph has `count(r) > 0` — the edge phase was
/// actually reached and edges loaded.
///
/// RED-when-broken: revert `create_index` to the bare non-idempotent `CREATE
/// INDEX` and pass 2's index step errors, the `?`/`unwrap` fails the test before
/// the edge assert — exactly the live failure.
#[tokio::test]
#[ignore = "needs podman; run with --ignored"]
async fn repopulate_is_idempotent_and_still_loads_edges() {
let container = start_falkordb().await.expect("start FalkorDB container");
let url = container.url();
let graph = "reindex_regression";
// One full pipeline pass (vertices → _gar_id index → edges) over `graph`.
// No `--drop`, so pass 2 hits a graph that already carries the index — the
// exact reuse that triggered the live abort.
async fn populate_once(url: &str, graph: &str) {
let mut loader = FalkorDbLoader::connect(url, graph)
.await
.expect("connect for populate");
loader
.insert_vertices("Person", &[people_batch()])
.await
.expect("vertices");
// The step under test: idempotent even when the index already exists.
loader
.create_index("Person", "_gar_id")
.await
.expect("index step must not abort on an already-indexed graph");
loader
.insert_edges("Person", "KNOWS", "Person", &[knows_batch()])
.await
.expect("edges must load — the phase after the index step");
}
populate_once(url, graph).await; // pass 1: creates the index
populate_once(url, graph).await; // pass 2: index already exists → must NOT abort
// The edge phase was reached on the reused graph: relationships exist.
let mut check = FalkorDbLoader::connect(url, graph)
.await
.expect("reconnect to verify");
let rels = check
.run_query("MATCH ()-[r]->() RETURN count(r)")
.await
.unwrap();
let edge_count = scalar_i64(&rels).unwrap_or(0);
assert!(
edge_count > 0,
"re-populate must still yield edges (got {edge_count}); raw reply: {rels:?}"
);
// And the index is present exactly once — a second create was a real no-op,
// not a duplicate. `db.indexes()` lists the Person/_gar_id index.
let idx = check.run_query("CALL db.indexes()").await.unwrap();
let idx_dump = format!("{idx:?}");
assert!(
idx_dump.contains("_gar_id"),
"the _gar_id index survives re-populate: {idx_dump}"
);
}
/// The **result rows** of a GRAPH.QUERY reply — element `[1]` of the
/// `[header, rows, stats]` triple. The `stats` element carries a
/// non-deterministic "Query internal execution time" line, so the parity gate
/// compares the deterministic row data only.
fn result_rows(reply: &redis::Value) -> &redis::Value {
match reply {
redis::Value::Array(items) | redis::Value::Set(items) if items.len() >= 2 => &items[1],
other => other,
}
}
/// Canonical, order-stable dump of a graph's node rows and edge rows (with the
/// `since` edge property) — the thing the parity gate compares between paths.
async fn dump_graph(loader: &mut FalkorDbLoader) -> (redis::Value, redis::Value) {
let nodes = loader
.run_query("MATCH (n:Person) RETURN n._gar_id, n.name, n.age ORDER BY n._gar_id")
.await
.unwrap();
let edges = loader
.run_query(
"MATCH (a:Person)-[r:KNOWS]->(b:Person) \
RETURN a._gar_id, b._gar_id, r.since ORDER BY a._gar_id, b._gar_id",
)
.await
.unwrap();
(result_rows(&nodes).clone(), result_rows(&edges).clone())
}
/// **The GRAPH.BULK parity gate.** The native `GRAPH.BULK` fast lane must
/// produce a BYTE-IDENTICAL graph — same nodes, same `_gar_id`, same properties,
/// same relationships (incl. the `since` edge prop) — as the `UNWIND … CREATE`
/// path, loading the SAME Arrow batches. RED-when-broken: any encoder drift (a
/// wrong type byte, endianness, a lost/duplicated node, a mis-translated edge
/// endpoint) changes the read-back dump and the ordered projections diverge.
#[tokio::test]
#[ignore = "needs podman; run with --ignored"]
async fn bulk_matches_unwind_byte_identical() {
let container = start_falkordb().await.expect("start FalkorDB container");
let url = container.url();
// Reference: the shipped UNWIND path (vertices → index → edges).
let mut unwind = FalkorDbLoader::connect(url, "unwind_ref")
.await
.expect("connect unwind");
unwind
.insert_vertices("Person", &[people_batch()])
.await
.unwrap();
unwind.create_index("Person", "_gar_id").await.unwrap();
unwind
.insert_edges("Person", "KNOWS", "Person", &[knows_batch()])
.await
.unwrap();
// Fast lane: the native GRAPH.BULK encoder over the SAME batches.
let mut bulk = FalkorDbLoader::connect(url, "bulk_fast")
.await
.expect("connect bulk");
assert!(
bulk.graph_is_empty().await,
"fresh graph is empty → bulk is chosen"
);
let people = people_batch();
let knows = knows_batch();
let (n, e) = bulk
.bulk_insert(
&[BulkNodeGroup {
label: "Person",
batches: std::slice::from_ref(&people),
}],
&[BulkEdgeGroup {
src_label: "Person",
edge_label: "KNOWS",
dst_label: "Person",
batches: std::slice::from_ref(&knows),
}],
)
.await
.expect("GRAPH.BULK load");
assert_eq!((n, e), (3, 2), "bulk reports 3 nodes + 2 relations");
let (un_nodes, un_edges) = dump_graph(&mut unwind).await;
let (bl_nodes, bl_edges) = dump_graph(&mut bulk).await;
assert_eq!(
bl_nodes, un_nodes,
"bulk nodes must byte-match the UNWIND nodes"
);
assert_eq!(
bl_edges, un_edges,
"bulk edges (incl. since) must byte-match UNWIND"
);
// The streamed/appendable sender must land the same graph too.
let mut streamed = FalkorDbLoader::connect(url, "bulk_streamed")
.await
.expect("connect streamed");
streamed
.bulk_insert_streamed(
&[BulkNodeGroup {
label: "Person",
batches: std::slice::from_ref(&people),
}],
&[BulkEdgeGroup {
src_label: "Person",
edge_label: "KNOWS",
dst_label: "Person",
batches: std::slice::from_ref(&knows),
}],
)
.await
.expect("streamed GRAPH.BULK load");
let (st_nodes, st_edges) = dump_graph(&mut streamed).await;
assert_eq!(st_nodes, un_nodes, "streamed bulk nodes byte-match UNWIND");
assert_eq!(st_edges, un_edges, "streamed bulk edges byte-match UNWIND");
}
/// Container-gated **bench arm**: native `GRAPH.BULK` vs the parallel-`UNWIND`
/// path at ~`SCALE` nodes, printing nodes/s for each so the Iceberg→FalkorDB
/// stage-4 win over the ~25k nodes/s single-connection baseline is measurable on
/// a live container. Not an assert-heavy test — it prints the numbers (run with
/// `--nocapture`) and only sanity-checks both loaded the same node count.
#[tokio::test]
#[ignore = "needs podman; bench — run with --ignored --nocapture"]
async fn bench_bulk_vs_unwind() {
const SCALE: i64 = 200_000;
let container = start_falkordb().await.expect("start FalkorDB container");
let url = container.url();
// One wide Person batch of SCALE rows (name + age).
let names: Vec<String> = (0..SCALE).map(|i| format!("n{i}")).collect();
let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
let ages: Vec<i64> = (0..SCALE).map(|i| 20 + (i % 60)).collect();
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("name", DataType::Utf8, false),
Field::new("age", DataType::Int64, false),
])),
vec![
Arc::new(StringArray::from(name_refs)),
Arc::new(Int64Array::from(ages)),
],
)
.unwrap();
// Parallel-UNWIND baseline.
let par = FalkorDbLoader::connect(url, "bench_unwind")
.await
.expect("connect unwind");
let t = Instant::now();
let un = par
.insert_vertices_par("Person", &[batch.clone()], 12)
.await
.unwrap();
let un_s = t.elapsed().as_secs_f64();
// Native GRAPH.BULK fast lane.
let mut bulk = FalkorDbLoader::connect(url, "bench_bulk")
.await
.expect("connect bulk");
let t = Instant::now();
let (bn, _) = bulk
.bulk_insert(
&[BulkNodeGroup {
label: "Person",
batches: std::slice::from_ref(&batch),
}],
&[],
)
.await
.unwrap();
let bulk_s = t.elapsed().as_secs_f64();
eprintln!(
"\n=== knut.falkordb_ingest @ {SCALE} nodes ===\n\
parallel-UNWIND: {un} nodes in {un_s:.3}s = {:.0} nodes/s\n\
native GRAPH.BULK: {bn} nodes in {bulk_s:.3}s = {:.0} nodes/s\n\
speedup: {:.1}x\n",
un as f64 / un_s,
bn as f64 / bulk_s,
un_s / bulk_s,
);
assert_eq!(un, bn, "both paths loaded the same node count");
}