crates/knut-popviz/tests/sim_iceberg_falkordb.rs
//! **Generate and permutate, through Iceberg and bifrost, into a real FalkorDB** — with
//! each of the three mapping formats, which must make the same graph.
//!
//! `knut-popload --target iceberg` writes the population as Iceberg tables; bifrost reads
//! them (IcebergSource, `mode: cdc`) with `mappings/population.{yml,json,sql}` into three
//! graphs. Then two permutate rounds — Spark-style copy-on-write commits that move people,
//! flip benefits, close loans and buy vehicles — and bifrost reads each round's changes.
//! After every step the three graphs must agree, and the graph must say exactly what the
//! round did: the changed persons carry its `round`, nobody lives at two addresses, closed
//! loans are gone, bought vehicles are there, nothing is duplicated.
//!
//! ```bash
//! KNUT_FALKORDB_IMAGE=localhost/nordisk-falkordb-valkey:v4.20.1 \
//! cargo test -p knut-popviz --features iceberg --test sim_iceberg_falkordb -- --ignored --nocapture
//! ```
//!
//! The same road through MinIO AIStor's REST catalog: `generate_and_permutate_through_the_aistor_catalog`
//! (feature `iceberg-rest`, `LIVE_*` env), see its docs.
#![cfg(feature = "iceberg")]
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use knut_bifrost::mapping::load_mapping;
#[cfg(feature = "iceberg-rest")]
use knut_bifrost::delivery::IcebergLake;
use knut_popviz::lake::Lake;
use knut_bifrost::{Listener, SyncMode, SyncState, TransformRegistry, sink::FalkorDbSink, source::IcebergSource};
use knut_popviz::{IngestMode, LABELS, LoadParams, PERSON_REL, Phase, PopvizState, Target, run_load};
use testcontainers::core::ContainerPort;
use testcontainers::{GenericImage, ImageExt, runners::AsyncRunner};
const FORMATS: [&str; 3] = ["yml", "json", "sql"];
fn mapping(ext: &str) -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("mappings").join(format!("population.{ext}"))
}
fn ensure_docker_host() {
if std::env::var("DOCKER_HOST").is_err() {
let uid = std::process::Command::new("id")
.arg("-u")
.output()
.ok()
.and_then(|o| String::from_utf8(o.stdout).ok())
.and_then(|s| s.trim().parse::<u32>().ok())
.unwrap_or(1000);
let sock = format!("/run/user/{uid}/podman/podman.sock");
if Path::new(&sock).exists() {
unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
}
}
}
async fn int(conn: &mut redis::aio::MultiplexedConnection, graph: &str, q: &str) -> 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,
}
}
let v: redis::Value = redis::cmd("GRAPH.QUERY").arg(graph).arg(q).query_async(conn).await.unwrap();
walk(&v).unwrap_or(-1)
}
/// The graph as numbers: per label (count, distinct ids, sum of round), per edge type
/// (count), and the two invariants (persons at more than one address, persons with one).
async fn shape(conn: &mut redis::aio::MultiplexedConnection, g: &str) -> Vec<(String, i64)> {
let mut out = Vec::new();
for (l, _) in LABELS.iter() {
out.push((format!("{l}"), int(conn, g, &format!("MATCH (n:{l}) RETURN count(n)")).await));
out.push((format!("{l} distinct"), int(conn, g, &format!("MATCH (n:{l}) RETURN count(DISTINCT n.id)")).await));
out.push((format!("{l} round sum"), int(conn, g, &format!("MATCH (n:{l}) RETURN sum(n.round)")).await));
}
for rel in PERSON_REL.iter().skip(1) {
out.push((format!("-{rel}->"), int(conn, g, &format!("MATCH ()-[e:{rel}]->() RETURN count(e)")).await));
}
out.push(("two addresses".into(), int(conn, g, "MATCH (p:Person)-[e:LIVES_AT]->() WITH p, count(e) AS c WHERE c > 1 RETURN count(p)").await));
out.push(("one address".into(), int(conn, g, "MATCH (p:Person)-[:LIVES_AT]->() RETURN count(DISTINCT p)").await));
out
}
fn get(s: &[(String, i64)], k: &str) -> i64 {
s.iter().find(|(n, _)| n == k).map(|(_, v)| *v).unwrap_or_else(|| panic!("no `{k}`"))
}
#[tokio::test(flavor = "multi_thread")]
#[ignore = "needs podman and a FalkorDB image; run with --ignored"]
async fn generate_and_permutate_through_iceberg_and_bifrost_with_three_mappings() {
let (_falkor, url) = falkordb().await;
let dir = tempfile::tempdir().unwrap();
let wh = dir.path().join("lake");
road(
&url,
Target::Iceberg { warehouse: wh.display().to_string() },
Lake::Dir(wh.clone()),
|| IcebergSource::new(&wh),
)
.await;
}
/// **The same road through MinIO AIStor's own Iceberg REST catalog** — generate and
/// permutate write through `/_iceberg` (SigV4 `s3tables`), bifrost reads through it.
/// Every run is a fresh namespace `knut_sim_<unix time>` in the warehouse, and nothing is
/// dropped: the lake keeps what it was given.
///
/// ```bash
/// LIVE_REST_URI=https://172.16.12.9:9000/_iceberg LIVE_REST_WAREHOUSE=red LIVE_SIGV4=s3tables \
/// LIVE_S3_ENDPOINT=https://172.16.12.9:9000 LIVE_S3_CA=ca-chain.crt \
/// MINIO_ACCESS_KEY=… MINIO_SECRET_KEY=… KNUT_FALKORDB_IMAGE=… \
/// cargo test -p knut-popviz --features iceberg-rest --test sim_iceberg_falkordb -- --ignored --nocapture aistor
/// ```
#[cfg(feature = "iceberg-rest")]
#[tokio::test(flavor = "multi_thread")]
#[ignore = "live: needs MinIO AIStor (LIVE_REST_URI …), podman and a FalkorDB image"]
async fn generate_and_permutate_through_the_aistor_catalog() {
let env = |k: &str| std::env::var(k).ok().filter(|v| !v.is_empty());
let Some(uri) = env("LIVE_REST_URI") else {
eprintln!("SKIP: LIVE_REST_URI is not set - no live AIStor configured");
return;
};
let namespace = format!("knut_sim_{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs());
let rest = knut_popviz::lake::RestLake {
uri,
warehouse: Some(env("LIVE_REST_WAREHOUSE").unwrap_or_else(|| "red".into())),
namespace: namespace.clone(),
s3_endpoint: env("LIVE_S3_ENDPOINT").expect("LIVE_S3_ENDPOINT"),
s3_region: env("LIVE_S3_REGION").unwrap_or_else(|| "us-east-1".into()),
s3_bucket: env("LIVE_S3_BUCKET"),
s3_ca: env("LIVE_S3_CA").map(PathBuf::from),
sigv4: env("LIVE_SIGV4"),
};
eprintln!("lake: {}", Lake::Rest(rest.clone()).describe());
// bifrost's side of the same catalog, read-only.
let s3 = skade::S3Config {
endpoint: rest.s3_endpoint.clone(),
region: rest.s3_region.clone(),
bucket: rest.s3_bucket.clone().or_else(|| rest.warehouse.clone()).unwrap(),
access_key_id: env("MINIO_ACCESS_KEY").expect("MINIO_ACCESS_KEY"),
secret_access_key: env("MINIO_SECRET_KEY").expect("MINIO_SECRET_KEY"),
path_style: true,
tls: rest.s3_ca.clone().map(|ca| skade::tls::ClientTls { ca_pem: Some(ca), ..Default::default() }),
};
let catalog = skade::connect::IcebergRest {
uri: rest.uri.clone(),
warehouse: rest.warehouse.clone(),
prefix_or_ref: None,
token: None,
tls: s3.tls.clone(),
s3,
read_only: true,
sigv4: rest.sigv4.clone().map(|service| skade::connect::SigV4 { service }),
..Default::default()
};
let (_falkor, url) = falkordb().await;
road(&url, Target::IcebergRest(rest.clone()), Lake::Rest(rest), || {
IcebergSource::from_lake(IcebergLake::Catalog { connector: std::sync::Arc::new(catalog.clone()), namespace: Some(namespace.clone()) })
})
.await;
}
/// A FalkorDB container (`KNUT_FALKORDB_IMAGE`) and its `redis://` URL.
async fn falkordb() -> (testcontainers::ContainerAsync<GenericImage>, String) {
ensure_docker_host();
let image = std::env::var("KNUT_FALKORDB_IMAGE").unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
let falkor = GenericImage::new(name, tag)
.with_exposed_port(ContainerPort::Tcp(6379))
.with_startup_timeout(Duration::from_secs(120))
.start()
.await
.expect("start falkordb (build the image first, or set KNUT_FALKORDB_IMAGE)");
let url = format!("redis://127.0.0.1:{}", falkor.get_host_port_ipv4(6379).await.unwrap());
(falkor, url)
}
/// Generate (`target`), a first bifrost load per mapping format, then two permutate rounds
/// on `lake`, each read as a delta — checking the graph after every step.
async fn road(url: &str, target: Target, lake: Lake, source: impl Fn() -> IcebergSource) {
let url = url.to_string();
let mut conn = redis::Client::open(url.clone()).unwrap().get_multiplexed_async_connection().await.unwrap();
// ── generate: the population as Iceberg tables ──────────────────────────────────────
let scale = 0.001;
let t = Instant::now();
let state = Arc::new(Mutex::new(PopvizState::new(scale)));
let params = LoadParams {
url: String::new(),
graph: String::new(),
scale,
drop: true,
target,
ingest: IngestMode::Auto,
tls: None,
};
let st = state.clone();
tokio::task::spawn_blocking(move || run_load(params, st, || {})).await.unwrap();
{
let s = state.lock().unwrap();
assert_eq!(s.phase, Phase::Done, "generate: {:?}", s.error);
}
let generate_s = t.elapsed().as_secs_f64();
let totals: Vec<usize> = state.lock().unwrap().labels.iter().map(|l| l.total).collect();
// ── bifrost, one listener per mapping format ─────────────────────────────────────────
let states = tempfile::tempdir().unwrap();
let mut listeners = Vec::new();
for ext in FORMATS {
listeners.push(Listener::new(
load_mapping(mapping(ext), SyncMode::Cdc).unwrap(),
TransformRegistry::new(),
source(),
FalkorDbSink::connect(&url, format!("pop_{ext}")).await.unwrap(),
SyncState::open(states.path().join(format!("{ext}.json"))).unwrap(),
));
}
let mut first_s = Vec::new();
for (ext, l) in FORMATS.iter().zip(listeners.iter_mut()) {
let t = Instant::now();
let r = l.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "{ext}: first load");
first_s.push(t.elapsed().as_secs_f64());
}
let base = shape(&mut conn, "pop_yml").await;
for ext in ["json", "sql"] {
assert_eq!(shape(&mut conn, &format!("pop_{ext}")).await, base, "{ext} = yml after the first load");
}
for (i, (l, _)) in LABELS.iter().enumerate() {
assert_eq!(get(&base, l) as usize, totals[i], "{l}: every row a node");
assert_eq!(get(&base, &format!("{l} distinct")), get(&base, l), "{l}: no duplicate");
assert_eq!(get(&base, &format!("{l} round sum")), 0, "{l}: round 0");
}
assert_eq!(get(&base, "two addresses"), 0);
// ── two permutate rounds, each read as a delta ───────────────────────────────────────
let mut before = base;
let mut timings = Vec::new();
for round in 1..=2i64 {
let t = Instant::now();
// Round 2 asks for "the next round" (0), as the bench's Permutate unit does.
let rep = knut_popviz::permutate::permutate_lake(&lake, 0.10, if round == 2 { 0 } else { round }).await.unwrap();
assert_eq!(rep.round, round, "the next round after {} is {round}", round - 1);
let permutate_s = t.elapsed().as_secs_f64();
assert!(rep.persons_changed > 0 && rep.addresses_moved > 0 && rep.loans_deleted > 0 && rep.vehicles_added > 0, "{rep:?}");
let mut delta_s = Vec::new();
for (ext, l) in FORMATS.iter().zip(listeners.iter_mut()) {
let t = Instant::now();
let r = l.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "{ext}: round {round}");
delta_s.push(t.elapsed().as_secs_f64());
}
let now = shape(&mut conn, "pop_yml").await;
for ext in ["json", "sql"] {
assert_eq!(shape(&mut conn, &format!("pop_{ext}")).await, now, "{ext} = yml after round {round}");
}
for g in FORMATS.map(|e| format!("pop_{e}")) {
assert_eq!(
int(&mut conn, &g, &format!("MATCH (p:Person) WHERE p.round = {round} RETURN count(p)")).await as usize,
rep.persons_changed,
"{g}: exactly the changed persons carry round {round}"
);
assert_eq!(
int(&mut conn, &g, &format!("MATCH (p:Person)-[e:LIVES_AT]->() WHERE e.round = {round} RETURN count(e)")).await as usize,
rep.addresses_moved,
"{g}: one new LIVES_AT per move"
);
}
assert_eq!(get(&now, "two addresses"), 0, "round {round}: a move leaves one address");
assert_eq!(get(&now, "one address"), get(&before, "one address"), "round {round}: nobody lost their address");
assert_eq!(get(&now, "Loan"), get(&before, "Loan") - rep.loans_deleted as i64, "round {round}: closed loans gone");
assert_eq!(get(&now, "-HOLDS->"), get(&before, "-HOLDS->") - rep.loans_deleted as i64);
assert_eq!(get(&now, "Vehicle"), get(&before, "Vehicle") + rep.vehicles_added as i64, "round {round}: bought vehicles there");
assert_eq!(get(&now, "-DRIVES->"), get(&before, "-DRIVES->") + rep.vehicles_added as i64);
assert_eq!(get(&now, "Person"), get(&before, "Person"), "round {round}: persons are updated, not added");
for (l, _) in LABELS.iter() {
assert_eq!(get(&now, &format!("{l} distinct")), get(&now, l), "round {round} {l}: no duplicate");
}
timings.push((round, permutate_s, delta_s, rep));
before = now;
}
eprintln!("\nscale {scale}: {} persons, {} nodes", totals[0], totals.iter().sum::<usize>());
eprintln!("generate ({}) {generate_s:.2} s", lake.describe());
for (ext, s) in FORMATS.iter().zip(&first_s) {
eprintln!("first bifrost load {ext:<4} {s:.2} s");
}
for (round, p, d, rep) in &timings {
eprintln!(
"round {round}: permutate {p:.2} s ({} persons: {} moved, {} flips, {} loans closed, {} vehicles) · delta yml/json/sql {:.2}/{:.2}/{:.2} s",
rep.persons_changed, rep.addresses_moved, rep.bistand_flipped, rep.loans_deleted, rep.vehicles_added, d[0], d[1], d[2]
);
}
}