crates/knut-bifrost/tests/delta_iceberg_falkordb.rs
//! The **streaming/delta** Iceberg → FalkorDB path against a **real FalkorDB**.
//!
//! This is the LAW-7 container test for the incremental ingest: skade writes an
//! Iceberg snapshot, the `IcebergSource` + `Listener` ingest it, and we assert
//! the live FalkorDB node count via `GRAPH.QUERY` — then a SECOND skade snapshot
//! (more rows) is written and the **delta** ingest is run, asserting only the
//! new nodes were added (cursor advanced, no full reload).
//!
//! Pure Rust end to end: skade (Iceberg, arrow 59) → knut-bifrost delta scan
//! (manifest diff, arrow 59) → FalkorDbSink (Cypher MERGE) → FalkorDB. No Spark.
//!
//! ## Prerequisites
//! - Podman socket: `systemctl --user start podman.socket`
//! - FalkorDB image (override with `KNUT_FALKORDB_IMAGE`):
//! `podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .`
//! (or `KNUT_FALKORDB_IMAGE=localhost/knut-falkordb-valkey9:latest`)
//!
//! ## Run
//! ```bash
//! cargo test -p knut-bifrost --features skade --test delta_iceberg_falkordb \
//! -- --ignored --nocapture
//! ```
#![cfg(feature = "skade")]
use std::sync::Arc;
use std::time::Duration;
use knut_bifrost::{
Listener, MappingSpec, SyncState, TransformRegistry, sink::FalkorDbSink, source::IcebergSource,
};
use skade::arrow_array::{Int64Array, RecordBatch as SkadeBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use testcontainers::{GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner};
const MAPPING: &str = r#"
version: 1
tables:
- table: people
mode: incremental
pipeline:
- vertex:
label: Person
id_column: id
properties: [name]
"#;
fn people_batch(ids: &[i64]) -> SkadeBatch {
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
SkadeBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids.to_vec())),
Arc::new(StringArray::from(names)),
],
)
.unwrap()
}
/// Append `ids` as one new Iceberg snapshot via skade.
async fn append_iceberg(dir: &std::path::Path, ids: &[i64]) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let batch = people_batch(ids);
let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
table.append(&[batch]).await.unwrap();
// `wh`/`table` drop here → redb lock released for the IcebergSource.
}
/// Equality-delete `ids` as one new Iceberg `Delete` snapshot via skade.
async fn delete_iceberg(dir: &std::path::Path, ids: &[i64]) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let mut table = wh.table("people").await.unwrap();
let key = SkadeBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(ids.to_vec()))],
)
.unwrap();
table.delete_equality(&key, &["id"]).await.unwrap();
}
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 std::path::Path::new(&sock).exists() {
unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
}
}
}
async fn count(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)
}
// Throughput twin (rows/s for incremental-vs-full ingest) lives in the nornir
// bencher `knut.delta_ingest` (xtask/examples/nornir-bench.rs); this stays the
// downscaled fast correctness check (tiny N=3 / M=2 / one delete) of the same path.
#[tokio::test]
#[ignore = "needs podman + a FalkorDB image; run with --ignored"]
async fn delta_iceberg_streams_only_new_rows_into_falkordb() {
ensure_docker_host();
// ── setup: FalkorDB container ───────────────────────────────────────────
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 port = falkor.get_host_port_ipv4(6379).await.unwrap();
let url = format!("redis://127.0.0.1:{port}");
let graph = "delta";
let client = redis::Client::open(url.clone()).unwrap();
let mut conn = client.get_multiplexed_async_connection().await.unwrap();
// ── the Iceberg warehouse (skade) + the streaming listener ──────────────
let warehouse = tempfile::tempdir().unwrap();
let state = tempfile::tempdir().unwrap();
// ── snapshot A: N=3 rows → ingest → assert 3 nodes ──────────────────────
append_iceberg(warehouse.path(), &[1, 2, 3]).await;
let mut listener = Listener::new(
MappingSpec::from_yaml(MAPPING).unwrap(),
TransformRegistry::new(),
IcebergSource::new(warehouse.path()),
FalkorDbSink::connect(&url, graph).await.unwrap(),
SyncState::open(state.path().join("state.json")).unwrap(),
);
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "snapshot A round failures");
assert_eq!(round.synced.len(), 1, "table synced");
assert_eq!(round.synced[0].1, 3, "N=3 rows shipped from snapshot A");
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
3,
"N=3 Person nodes after snapshot A",
);
// ── idle: no new snapshot → nothing shipped, count unchanged ────────────
let idle = listener.sync_once().await.unwrap();
assert!(idle.synced.is_empty(), "no new snapshot → idle round");
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
3,
);
// ── snapshot B: M=2 MORE rows → DELTA ingest → assert only +2 (total 5) ──
append_iceberg(warehouse.path(), &[4, 5]).await;
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "snapshot B round failures");
assert_eq!(
round.synced[0].1, 2,
"DELTA shipped only M=2 new rows (not a full reload of 5)",
);
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
5,
"N+M = 5 Person nodes after the delta (cursor advanced, no reload)",
);
// ── the new ids 4 and 5 are present (the delta actually injected them) ───
assert_eq!(
count(
&mut conn,
graph,
"MATCH (n:Person) WHERE n.id IN [4,5] RETURN count(n)"
)
.await,
2,
"the two delta nodes (ids 4,5) are in the graph",
);
// ── snapshot C: a CDC equality-DELETE of id=2 → DELTA removes only it ────
delete_iceberg(warehouse.path(), &[2]).await;
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "snapshot C (delete) round failures");
assert_eq!(round.synced.len(), 1, "the delete is one synced round");
assert_eq!(
round.synced[0].2.deleted, 1,
"the CDC delta deleted exactly one node",
);
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
4,
"5 − 1 = 4 nodes after the equality-delete of id=2 (pure-Rust CDC delete)",
);
assert_eq!(
count(
&mut conn,
graph,
"MATCH (n:Person) WHERE n.id = 2 RETURN count(n)"
)
.await,
0,
"id=2 is gone; the other nodes are untouched",
);
// teardown: container + tempdirs drop here.
}