crates/knut-bifrost/tests/edge_changes_falkordb.rs
//! **A changed connection, when the changelog carries only the old row's key**
//! — against a real FalkorDB.
//!
//! Spark writes an UPDATE as the full old row plus the new one, and the old row
//! names the old edge. A key-only writer (an equality delete on the key, then
//! the new row — Flink-style upserts) does not: the old row's foreign key is
//! gone. Two rules cover it, and this test holds the sink to both:
//!
//! * an `fk_edges` step is ONE edge per source node — the new row is the truth,
//! so its source's other outgoing edges of the type are replaced (and all of
//! them go when the foreign key is cleared);
//! * an edge table's edge deleted by its key alone is deleted by that key.
//!
//! ```bash
//! KNUT_FALKORDB_IMAGE=localhost/nordisk-falkordb-valkey:v4.20.1 \
//! cargo test -p knut-bifrost --features skade --test edge_changes_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, Float64Array, Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use testcontainers::core::ContainerPort;
use testcontainers::{GenericImage, ImageExt, runners::AsyncRunner};
const MAPPING: &str = r#"
version: 1
tables:
- table: s.companies
mode: cdc
pipeline:
- vertex: { label: Company, id_column: id, properties: [name] }
- table: s.users
mode: cdc
pipeline:
- vertex: { label: User, id_column: user_id, properties: [name] }
- fk_edges:
type: WORKS_AT
src: { label: User, column: user_id }
dst: { label: Company, column: company_id, id_column: id }
- table: s.knows
mode: cdc
pipeline:
- edge:
type: KNOWS
src: { label: User, column: from_id, id_column: user_id }
dst: { label: User, column: to_id, id_column: user_id }
key: [id]
properties: [weight]
"#;
fn companies(rows: &[(i64, &str)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false), Field::new("name", DataType::Utf8, 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<_>>())),
],
)
.unwrap()
}
fn users(rows: &[(i64, &str, Option<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()
}
fn knows(rows: &[(i64, i64, i64, f64)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("from_id", DataType::Int64, false),
Field::new("to_id", DataType::Int64, false),
Field::new("weight", DataType::Float64, true),
])),
vec![
Arc::new(Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>())) as ArrayRef,
Arc::new(Int64Array::from(rows.iter().map(|r| r.1).collect::<Vec<_>>())),
Arc::new(Int64Array::from(rows.iter().map(|r| r.2).collect::<Vec<_>>())),
Arc::new(Float64Array::from(rows.iter().map(|r| r.3).collect::<Vec<_>>())),
],
)
.unwrap()
}
fn key(name: &str, ids: &[i64]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(name, DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(ids.to_vec())) as ArrayRef],
)
.unwrap()
}
async fn append(dir: &std::path::Path, table: &str, batch: RecordBatch) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let mut t = wh.table_or_create(table, &batch.schema()).await.unwrap();
t.append(&[batch]).await.unwrap();
}
/// A key-only delete: an equality delete on one column.
async fn delete_by_key(dir: &std::path::Path, table: &str, key: RecordBatch) {
let wh = skade::Warehouse::open(dir).await.unwrap();
let mut t = wh.table(table).await.unwrap();
let col = key.schema().field(0).name().clone();
t.delete_equality(&key, &[col.as_str()]).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, 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("g").arg(q).query_async(conn).await.unwrap();
walk(&v).unwrap_or(-1)
}
#[tokio::test]
#[ignore = "needs podman and a FalkorDB image; run with --ignored"]
async fn a_key_only_changelog_still_moves_and_removes_edges() {
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();
append(dir, "s.companies", companies(&[(10, "acme"), (20, "globex")])).await;
append(dir, "s.users", users(&[(1, "a", Some(10)), (2, "b", Some(10)), (3, "c", Some(20))])).await;
append(dir, "s.knows", knows(&[(100, 1, 2, 0.5), (101, 2, 3, 0.7)])).await;
let state = tempfile::tempdir().unwrap();
let mut l = Listener::new(
MappingSpec::from_yaml(MAPPING).unwrap(),
TransformRegistry::new(),
IcebergSource::new(dir),
FalkorDbSink::connect(&url, "g").await.unwrap(),
SyncState::open(state.path().join("st.json")).unwrap(),
);
let r = l.sync_once().await.unwrap();
assert_eq!(r.failed, vec![]);
assert_eq!(count(&mut conn, "MATCH ()-[e:WORKS_AT]->() RETURN count(e)").await, 3);
assert_eq!(count(&mut conn, "MATCH ()-[e:KNOWS]->() RETURN count(e)").await, 2);
// ── the key-only writer's changes, one batch of commits ─────────────────
// user 1 moves to globex; user 2's company is cleared — each an equality
// delete of the key, then the new row.
delete_by_key(dir, "s.users", key("user_id", &[1, 2])).await;
append(dir, "s.users", users(&[(1, "a", Some(20)), (2, "b", None)])).await;
// edge 100 deleted by its id alone.
delete_by_key(dir, "s.knows", key("id", &[100])).await;
let r = l.sync_once().await.unwrap();
assert_eq!(r.failed, vec![]);
assert_eq!(count(&mut conn, "MATCH (:User {user_id: 1})-[e:WORKS_AT]->(:Company {id: 10}) RETURN count(e)").await, 0, "the old edge went");
assert_eq!(count(&mut conn, "MATCH (:User {user_id: 1})-[e:WORKS_AT]->(:Company {id: 20}) RETURN count(e)").await, 1, "the new one is there");
assert_eq!(count(&mut conn, "MATCH (:User {user_id: 2})-[e:WORKS_AT]->() RETURN count(e)").await, 0, "a cleared key has no edge");
assert_eq!(count(&mut conn, "MATCH ()-[e:WORKS_AT]->() RETURN count(e)").await, 2);
assert_eq!(count(&mut conn, "MATCH (n:User) RETURN count(n)").await, 3, "the users were updated, not deleted");
assert_eq!(count(&mut conn, "MATCH ()-[e:KNOWS {id: 100}]->() RETURN count(e)").await, 0, "deleted by its key");
assert_eq!(count(&mut conn, "MATCH ()-[e:KNOWS {id: 101}]->() RETURN count(e)").await, 1);
}