crates/knut-bifrost/tests/mapping_formats_falkordb.rs
//! **The mapping formats against a real FalkorDB**: PuppyGraph's `schema.json`
//! and a `.sql` mapping, over the same Iceberg tables — PuppyGraph's own
//! "modern" graph — loaded, then changed by one batch of commits that deletes an
//! edge, deletes a person, adds an edge and adds a parallel one.
//!
//! What it holds the sink to:
//! * an edge-only table's `DELETE` removes its edge (and only it);
//! * a person's `DELETE` removes the person and its edges;
//! * an edge key (PuppyGraph's edge `id`, SQL's `_edge_id`) makes parallel edges;
//! * a SQL `WHERE` on a non-key column does not swallow a key-only `DELETE`.
//!
//! ```bash
//! KNUT_FALKORDB_IMAGE=localhost/nordisk-falkordb-valkey:v4.20.1 \
//! cargo test -p knut-bifrost --features sql --test mapping_formats_falkordb -- --ignored --nocapture
//! ```
#![cfg(all(feature = "skade", feature = "sql"))]
use std::sync::Arc;
use std::time::Duration;
use knut_bifrost::mapping::load_mapping;
use knut_bifrost::{Listener, SyncMode, 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 PUPPYGRAPH: &str = r#"{
"catalog": [ { "name": "lake", "type": "iceberg", "metastore": { "type": "rest", "uri": "http://unused" } } ],
"node": [
{ "label": "person",
"dataSourceGroup": { "externalDataSource": { "enabled": true, "catalog": "lake", "schema": "modern", "table": "person",
"mappedField": [ { "sourceFieldName": "id", "targetFieldName": "id" },
{ "sourceFieldName": "name", "targetFieldName": "name" },
{ "sourceFieldName": "age", "targetFieldName": "age" } ] } },
"id": [ { "name": "id", "type": "INT" } ],
"attribute": [ { "name": "name", "type": "STRING" }, { "name": "age", "type": "INT" } ] },
{ "label": "software",
"dataSourceGroup": { "externalDataSource": { "enabled": true, "catalog": "lake", "schema": "modern", "table": "software",
"mappedField": [ { "sourceFieldName": "id", "targetFieldName": "id" },
{ "sourceFieldName": "name", "targetFieldName": "name" },
{ "sourceFieldName": "lang", "targetFieldName": "lang" } ] } },
"id": [ { "name": "id", "type": "INT" } ],
"attribute": [ { "name": "name", "type": "STRING" }, { "name": "lang", "type": "STRING" } ] } ],
"edge": [
{ "label": "knows", "fromNodeLabel": "person", "toNodeLabel": "person",
"dataSourceGroup": { "externalDataSource": { "enabled": true, "catalog": "lake", "schema": "modern", "table": "knows",
"mappedField": [ { "sourceFieldName": "id", "targetFieldName": "id" },
{ "sourceFieldName": "from_id", "targetFieldName": "from_id" },
{ "sourceFieldName": "to_id", "targetFieldName": "to_id" },
{ "sourceFieldName": "weight", "targetFieldName": "weight" } ] } },
"id": [ { "name": "id", "type": "INT" } ],
"fromKey": [ { "name": "from_id", "type": "INT" } ], "toKey": [ { "name": "to_id", "type": "INT" } ],
"attribute": [ { "name": "weight", "type": "DOUBLE" } ] },
{ "label": "created", "fromNodeLabel": "person", "toNodeLabel": "software",
"dataSourceGroup": { "externalDataSource": { "enabled": true, "catalog": "lake", "schema": "modern", "table": "created",
"mappedField": [ { "sourceFieldName": "id", "targetFieldName": "id" },
{ "sourceFieldName": "from_id", "targetFieldName": "from_id" },
{ "sourceFieldName": "to_id", "targetFieldName": "to_id" },
{ "sourceFieldName": "weight", "targetFieldName": "weight" } ] } },
"id": [ { "name": "id", "type": "INT" } ],
"fromKey": [ { "name": "from_id", "type": "INT" } ], "toKey": [ { "name": "to_id", "type": "INT" } ],
"attribute": [ { "name": "weight", "type": "DOUBLE" } ] } ]
}"#;
const SQL: &str = "
SELECT 'Person' AS _label, id AS _id, upper(name) AS name, age FROM modern.person WHERE age > 28;
SELECT 'KNOWS' AS _type, 'Person' AS _from_label, from_id AS _from, 'Person' AS _to_label, to_id AS _to,
id AS _edge_id, weight FROM modern.knows;
";
fn people(rows: &[(i64, &str, i64)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
Field::new("age", 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 software(rows: &[(i64, &str, &str)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
Field::new("lang", 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<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.2).collect::<Vec<_>>())),
],
)
.unwrap()
}
/// `(id, from_id, to_id, weight)`.
fn links(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 ints(cols: &[(&str, Vec<i64>)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(cols.iter().map(|(n, _)| Field::new(*n, DataType::Int64, false)).collect::<Vec<_>>())),
cols.iter().map(|(_, v)| Arc::new(Int64Array::from(v.clone())) as ArrayRef).collect(),
)
.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();
}
async fn delete(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 cols: Vec<String> = key.schema().fields().iter().map(|f| f.name().clone()).collect();
let cols: Vec<&str> = cols.iter().map(String::as_str).collect();
t.delete_equality(&key, &cols).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)
}
#[tokio::test]
#[ignore = "needs podman and a FalkorDB image; run with --ignored"]
async fn puppygraph_and_sql_mappings_load_append_and_delete_in_falkordb() {
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();
// ── the lake: PuppyGraph's "modern" graph ──────────────────────────────
let lake = tempfile::tempdir().unwrap();
let dir = lake.path();
append(dir, "modern.person", people(&[(1, "marko", 29), (2, "vadas", 27), (4, "josh", 32), (6, "peter", 35)])).await;
append(dir, "modern.software", software(&[(3, "lop", "java"), (5, "ripple", "java")])).await;
append(dir, "modern.knows", links(&[(7, 1, 2, 0.5), (8, 1, 4, 1.0)])).await;
append(dir, "modern.created", links(&[(9, 1, 3, 0.4), (10, 4, 5, 1.0), (11, 4, 3, 0.4), (12, 6, 3, 0.2)])).await;
let files = tempfile::tempdir().unwrap();
let (pg_path, sql_path) = (files.path().join("schema.json"), files.path().join("graph.sql"));
std::fs::write(&pg_path, PUPPYGRAPH).unwrap();
std::fs::write(&sql_path, SQL).unwrap();
let listener = |path: std::path::PathBuf, graph: &'static str, state: &'static str| {
let (url, dir, files) = (url.clone(), dir.to_path_buf(), files.path().to_path_buf());
async move {
Listener::new(
load_mapping(&path, SyncMode::Cdc).unwrap(),
TransformRegistry::new(),
IcebergSource::new(dir),
FalkorDbSink::connect(&url, graph).await.unwrap(),
SyncState::open(files.join(state)).unwrap(),
)
}
};
let mut pg = listener(pg_path, "pg", "pg.json").await;
let mut sq = listener(sql_path, "sq", "sq.json").await;
// ── first load ─────────────────────────────────────────────────────────
let r = pg.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "puppygraph round 1");
assert_eq!(count(&mut conn, "pg", "MATCH (n:person) RETURN count(n)").await, 4);
assert_eq!(count(&mut conn, "pg", "MATCH (n:software) RETURN count(n)").await, 2);
assert_eq!(count(&mut conn, "pg", "MATCH ()-[e:knows]->() RETURN count(e)").await, 2);
assert_eq!(count(&mut conn, "pg", "MATCH ()-[e:created]->() RETURN count(e)").await, 4);
assert_eq!(count(&mut conn, "pg", "MATCH (:person {id: 1})-[e:knows {id: 8}]->(:person {id: 4}) WHERE e.weight = 1.0 RETURN count(e)").await, 1);
let r = sq.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "sql round 1");
// The WHERE keeps marko, josh, peter; vadas exists only as the placeholder
// end of the edge 1 -> 2 (an edge MERGEs its end nodes).
assert_eq!(count(&mut conn, "sq", "MATCH (n:Person) WHERE n.name IS NOT NULL RETURN count(n)").await, 3);
assert_eq!(count(&mut conn, "sq", "MATCH (n:Person {id: 1}) WHERE n.name = 'MARKO' RETURN count(n)").await, 1);
assert_eq!(count(&mut conn, "sq", "MATCH ()-[e:KNOWS]->() RETURN count(e)").await, 2);
// ── one batch of commits ───────────────────────────────────────────────
// An edge row deleted with its endpoints (as a copy-on-write DELETE carries them).
delete(dir, "modern.knows", ints(&[("id", vec![7]), ("from_id", vec![1]), ("to_id", vec![2])])).await;
// A person deleted by key alone: every other column null in the changelog.
delete(dir, "modern.person", ints(&[("id", vec![6])])).await;
// A new edge, and a parallel one: same endpoints as edge 8, another id.
append(dir, "modern.knows", links(&[(13, 4, 2, 0.3), (14, 1, 4, 0.9)])).await;
let r = pg.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "puppygraph round 2");
assert_eq!(count(&mut conn, "pg", "MATCH (n:person) RETURN count(n)").await, 3, "peter is gone");
assert_eq!(count(&mut conn, "pg", "MATCH ()-[e:created]->() RETURN count(e)").await, 3, "with his edge");
assert_eq!(count(&mut conn, "pg", "MATCH ()-[e:knows]->() RETURN count(e)").await, 3, "7 gone; 13 and 14 new");
assert_eq!(count(&mut conn, "pg", "MATCH ()-[e:knows {id: 7}]->() RETURN count(e)").await, 0);
assert_eq!(count(&mut conn, "pg", "MATCH (:person {id: 1})-[e:knows]->(:person {id: 4}) RETURN count(e)").await, 2, "parallel");
let r = sq.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "sql round 2");
assert_eq!(
count(&mut conn, "sq", "MATCH (n:Person {id: 6}) RETURN count(n)").await,
0,
"peter's key-only DELETE got past `WHERE age > 28`"
);
assert_eq!(count(&mut conn, "sq", "MATCH ()-[e:KNOWS]->() RETURN count(e)").await, 3);
assert_eq!(count(&mut conn, "sq", "MATCH (:Person {id: 1})-[e:KNOWS]->(:Person {id: 4}) RETURN count(e)").await, 2, "parallel");
}