crates/knut-bifrost/tests/mode_semantics_falkordb.rs
//! **`incremental` vs `cdc` on an UPDATED table** — what each mode does to the
//! graph, against a real FalkorDB.
//!
//! One Iceberg table, two graphs: the same mapping in `mode: incremental` and in
//! `mode: cdc`, synced after every commit. The commits are what an updated table
//! sees: an append, a Spark-style copy-on-write UPDATE (a file rewritten) and
//! DELETE, a key-only update (an equality delete, then the new row), and another
//! append. The table of outcomes is printed; `cdc` is held to the right answer,
//! and `incremental` — which on an Iceberg source reads the same changelog — to
//! `cdc`.
//!
//! It exists because both used to keep a copy-on-write DELETE out of the graph:
//! the delete rows' tags were kept only when the window had equality-delete
//! FILES, so a rewrite's DELETE became an upsert of the deleted row.
//!
//! ```bash
//! KNUT_FALKORDB_IMAGE=localhost/nordisk-falkordb-valkey:v4.20.1 \
//! cargo test -p knut-bifrost --features skade --test mode_semantics_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::{ArrayRef, Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use testcontainers::core::ContainerPort;
use testcontainers::{GenericImage, ImageExt, runners::AsyncRunner};
fn mapping(mode: &str) -> String {
format!(
"version: 1\ntables:\n - table: s.users\n mode: {mode}\n pipeline:\n \
- vertex: {{ label: User, id_column: user_id, properties: [name] }}\n \
- fk_edges: {{ type: WORKS_AT, src: {{ label: User, column: user_id }}, dst: {{ label: Company, column: company_id, id_column: id }} }}\n"
)
}
fn users(rows: &[(i64, &str, i64)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("user_id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
Field::new("company_id", DataType::Int64, true),
])),
vec![
Arc::new(Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>())) as ArrayRef,
Arc::new(StringArray::from(rows.iter().map(|r| r.1).collect::<Vec<_>>())),
Arc::new(Int64Array::from(rows.iter().map(|r| r.2).collect::<Vec<_>>())),
],
)
.unwrap()
}
async fn table(dir: &std::path::Path) -> (skade::Warehouse, skade::Table) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let t = wh.table("s.users").await.unwrap();
(wh, t)
}
/// Spark copy-on-write: every live file replaced by `rows` in one overwrite.
async fn rewrite(dir: &std::path::Path, rows: &[(i64, &str, i64)]) {
let (_wh, mut t) = table(dir).await;
let live = t.live_data_files().await.unwrap();
t.overwrite(&live, &[users(rows)]).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 q(conn: &mut redis::aio::MultiplexedConnection, graph: &str, query: &str) -> String {
let v: redis::Value = redis::cmd("GRAPH.QUERY").arg(graph).arg(query).query_async(conn).await.unwrap();
fn rows(v: &redis::Value, out: &mut Vec<String>) {
match v {
redis::Value::Array(items) => items.iter().for_each(|i| rows(i, out)),
redis::Value::Int(i) => out.push(i.to_string()),
redis::Value::BulkString(b) => out.push(String::from_utf8_lossy(b).to_string()),
redis::Value::SimpleString(s) => out.push(s.clone()),
_ => {}
}
}
let mut out = Vec::new();
if let redis::Value::Array(parts) = &v {
if let Some(data) = parts.get(1) {
rows(data, &mut out);
}
}
out.join(",")
}
/// The graph as one line: users `id:name->company` sorted, and the edge count.
async fn snapshot(conn: &mut redis::aio::MultiplexedConnection, graph: &str) -> String {
let users = q(
conn,
graph,
"MATCH (u:User) OPTIONAL MATCH (u)-[:WORKS_AT]->(c:Company) \
RETURN u.user_id, u.name, collect(c.id) ORDER BY u.user_id",
)
.await;
let edges = q(conn, graph, "MATCH ()-[e:WORKS_AT]->() RETURN count(e)").await;
format!("{users} | edges {edges}")
}
#[tokio::test]
#[ignore = "needs podman and a FalkorDB image; run with --ignored"]
async fn incremental_and_cdc_on_an_updated_table() {
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());
let mut conn = redis::Client::open(url.clone()).unwrap().get_multiplexed_async_connection().await.unwrap();
let lake = tempfile::tempdir().unwrap();
let dir = lake.path();
let state = tempfile::tempdir().unwrap();
{
let wh = skade::Warehouse::open(dir).await.unwrap();
let first = users(&[(1, "a", 10), (2, "b", 10), (3, "c", 20)]);
let mut t = wh.table_or_create("s.users", &first.schema()).await.unwrap();
t.append(&[first]).await.unwrap();
}
let listener = |mode: &'static str| {
let (url, dir, st) = (url.clone(), dir.to_path_buf(), state.path().join(format!("{mode}.json")));
async move {
Listener::new(
MappingSpec::from_yaml(&mapping(mode)).unwrap(),
TransformRegistry::new(),
IcebergSource::new(dir),
FalkorDbSink::connect(&url, mode).await.unwrap(),
SyncState::open(st).unwrap(),
)
}
};
let mut inc = listener("incremental").await;
let mut cdc = listener("cdc").await;
let mut report: Vec<(String, String, String)> = Vec::new();
macro_rules! step {
($what:expr) => {{
let ri = inc.sync_once().await.unwrap();
let rc = cdc.sync_once().await.unwrap();
assert_eq!((ri.failed.len(), rc.failed.len()), (0, 0), "{}: {:?} {:?}", $what, ri.failed, rc.failed);
report.push(($what.to_string(), snapshot(&mut conn, "incremental").await, snapshot(&mut conn, "cdc").await));
}};
}
step!("append 1:a->10 2:b->10 3:c->20");
rewrite(dir, &[(1, "a2", 20), (2, "b", 10), (3, "c", 20)]).await;
step!("CoW UPDATE 1: name a2, company 20");
rewrite(dir, &[(1, "a2", 20), (2, "b", 10)]).await;
step!("CoW DELETE 3");
{
let (_wh, mut t) = table(dir).await;
let key = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("user_id", DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(vec![2])) as ArrayRef],
)
.unwrap();
t.delete_equality(&key, &["user_id"]).await.unwrap();
t.append(&[users(&[(2, "b2", 20)])]).await.unwrap();
}
step!("key-only UPDATE 2: b2, company 20");
{
let (_wh, mut t) = table(dir).await;
t.append(&[users(&[(4, "d", 10)])]).await.unwrap();
}
step!("append 4:d->10");
eprintln!("\n{:<36} | {:<52} | cdc", "commit", "incremental");
for (what, i, c) in &report {
eprintln!("{what:<36} | {i:<52} | {c}");
}
// cdc is the right answer at every step.
let want = [
"1,a,[10],2,b,[10],3,c,[20] | edges 3",
"1,a2,[20],2,b,[10],3,c,[20] | edges 3",
"1,a2,[20],2,b,[10] | edges 2",
"1,a2,[20],2,b2,[20] | edges 2",
"1,a2,[20],2,b2,[20],4,d,[10] | edges 3",
];
for ((what, _, c), w) in report.iter().zip(want) {
assert_eq!(c, w, "cdc after {what}");
}
// On an Iceberg source `incremental` reads the same per-commit changelog as
// `cdc`: the same graph at every step, deletes included.
for (what, i, c) in &report {
assert_eq!(i, c, "incremental = cdc after {what}");
}
}