crates/knut-bifrost/tests/watch_lake_s3.rs
//! **`watch` off S3 with no catalog, into a real FalkorDB.**
//!
//! The lake `bifrost watch --s3-root` reads: Iceberg tables on MinIO, found by
//! their root (`version-hint.text`, else the newest `*.metadata.json`), no
//! catalog in between. skade writes the table (a local redb catalog, data and
//! metadata on MinIO); the listener reads it through [`IcebergLake::S3`], which
//! never sees that catalog.
//!
//! Community MinIO and FalkorDB, both through testcontainers. The REST half
//! (MinIO AIStor's `/_iceberg`) needs AIStor, which community MinIO is not —
//! `live_iceberg_doris.rs` covers that locator against red.
//!
//! ```bash
//! KNUT_FALKORDB_IMAGE=localhost/nordisk-falkordb-valkey:v4.20.1 \
//! cargo test -p knut-bifrost --features skade-s3 --test watch_lake_s3 -- --ignored --nocapture
//! ```
#![cfg(feature = "skade-s3")]
use std::sync::Arc;
use std::time::Duration;
use knut_bifrost::delivery::IcebergLake;
use knut_bifrost::{Listener, MappingSpec, SyncState, TransformRegistry, sink::FalkorDbSink, source::IcebergSource};
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use testcontainers::core::{ContainerPort, WaitFor};
use testcontainers::{GenericImage, ImageExt, runners::AsyncRunner};
const MINIO_IMAGE: (&str, &str) = ("quay.io/minio/minio", "RELEASE.2025-04-08T15-41-24Z");
const USER: &str = "knutwatch";
const PASSWORD: &str = "knutwatch-secret";
const BUCKET: &str = "lake";
fn people(ids: std::ops::Range<i64>) -> RecordBatch {
let ids: Vec<i64> = ids.collect();
let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
RecordBatch::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)), Arc::new(StringArray::from(names))],
)
.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, a MinIO image and a FalkorDB image; run with --ignored"]
async fn watch_streams_iceberg_tables_off_s3_with_no_catalog_into_falkordb() {
ensure_docker_host();
// ── MinIO, and the bucket ───────────────────────────────────────────────
let minio = GenericImage::new(MINIO_IMAGE.0, MINIO_IMAGE.1)
.with_exposed_port(ContainerPort::Tcp(9000))
.with_wait_for(WaitFor::message_on_either_std("API:"))
.with_env_var("MINIO_ROOT_USER", USER)
.with_env_var("MINIO_ROOT_PASSWORD", PASSWORD)
.with_cmd(["server", "/data"])
.with_startup_timeout(Duration::from_secs(120))
.start()
.await
.expect("start minio");
let endpoint = format!("http://127.0.0.1:{}", minio.get_host_port_ipv4(9000).await.unwrap());
let ep = endpoint.clone();
tokio::task::spawn_blocking(move || aegir::Client::new(ep, "us-east-1", USER, PASSWORD).create_bucket(BUCKET))
.await
.unwrap()
.expect("create the bucket");
let s3 = skade::S3Config {
endpoint: endpoint.clone(),
region: "us-east-1".into(),
bucket: BUCKET.into(),
access_key_id: USER.into(),
secret_access_key: PASSWORD.into(),
path_style: true,
tls: None,
};
// ── FalkorDB ────────────────────────────────────────────────────────────
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 graph = "lake";
let mut conn = redis::Client::open(url.clone()).unwrap().get_multiplexed_async_connection().await.unwrap();
// ── the lake: two commits, written by skade through its own catalog ────
let root = format!("s3://{BUCKET}/wh");
let catalog = tempfile::tempdir().unwrap();
let append = |ids: std::ops::Range<i64>| {
let (db, root, s3) = (catalog.path().join("skade-catalog"), root.clone(), s3.clone());
async move {
let wh = skade::Warehouse::open_with_object_store_config(db, root, skade::ObjectStoreConfig::Aegir(s3))
.await
.expect("open the skade warehouse on MinIO");
let batch = people(ids);
let mut t = wh.table_or_create("people", &batch.schema()).await.unwrap();
t.append(&[batch]).await.unwrap();
t.inner().metadata().location().to_string()
}
};
append(0..500).await;
let location = append(500..1000).await;
eprintln!("table root {location}");
// The mapping names the table the way an operator would: `ns.name` under
// `--s3-root`. The layout skade wrote must be exactly that.
let rel = location.strip_prefix(&format!("{root}/")).expect("the table lives under the lake root");
let table = rel.replace('/', ".");
assert_eq!(table.split('.').count(), 2, "a table root is <root>/<ns>/<name>, got {location}");
let mapping = format!(
"version: 1\ntables:\n - table: {table}\n mode: incremental\n pipeline:\n \
- vertex: {{ label: Person, id_column: id, properties: [name] }}\n"
);
let state = tempfile::tempdir().unwrap();
let mut listener = Listener::new(
MappingSpec::from_yaml(&mapping).unwrap(),
TransformRegistry::new(),
IcebergSource::from_lake(IcebergLake::S3 { s3: s3.clone(), root: root.clone() }),
FalkorDbSink::connect(&url, graph).await.unwrap(),
SyncState::open(state.path().join("state.json")).unwrap(),
);
// ── first round: both commits ───────────────────────────────────────────
let r = listener.sync_once().await.unwrap();
assert_eq!(r.failed, vec![], "first round failures");
assert_eq!(r.synced.len(), 1);
assert_eq!(r.synced[0].1, 1000, "both commits shipped");
assert_eq!(count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await, 1000);
// ── idle: nothing new on S3 → nothing shipped ───────────────────────────
let r = listener.sync_once().await.unwrap();
assert!(r.synced.is_empty(), "no new snapshot, nothing shipped: {:?}", r.synced);
// ── a third commit: only its rows ───────────────────────────────────────
append(1000..1250).await;
let r = listener.sync_once().await.unwrap();
assert_eq!(r.failed, vec![]);
assert_eq!(r.synced[0].1, 250, "the delta, not a reload");
assert_eq!(count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await, 1250);
assert_eq!(count(&mut conn, graph, "MATCH (n:Person) WHERE n.id = 1249 AND n.name = 'p1249' RETURN count(n)").await, 1);
// ── the same table by its s3:// URL, verbatim ──────────────────────────
let by_url = IcebergLake::S3 { s3, root: "s3://elsewhere/ignored".into() }.locate(&location);
assert_eq!(by_url.source_id(), format!("s3:{}", location.trim_end_matches('/')));
// ── the operator's door: `bifrost watch --s3-root`, keys from the env ──
#[cfg(feature = "cli")]
{
let dir = tempfile::tempdir().unwrap();
let m = dir.path().join("m.yml");
std::fs::write(&m, &mapping).unwrap();
let out = std::process::Command::new(env!("CARGO_BIN_EXE_bifrost"))
.args(["watch", "--s3-root", &root, "--s3-endpoint", &endpoint, "--falkor", &url, "--graph", "lake_cli"])
.arg("--mapping")
.arg(&m)
.arg("--state")
.arg(dir.path().join("state.json"))
.arg("--once")
.env("MINIO_ACCESS_KEY", USER)
.env("MINIO_SECRET_KEY", PASSWORD)
.output()
.unwrap();
let log = String::from_utf8_lossy(&out.stdout).to_string() + &String::from_utf8_lossy(&out.stderr);
assert!(out.status.success(), "bifrost watch failed:\n{log}");
assert!(!log.contains(PASSWORD), "a secret reached the log");
assert_eq!(count(&mut conn, "lake_cli", "MATCH (n:Person) RETURN count(n)").await, 1250, "{log}");
// No keys in the environment: refused, naming the variable.
let out = std::process::Command::new(env!("CARGO_BIN_EXE_bifrost"))
.args(["watch", "--s3-root", &root, "--s3-endpoint", &endpoint, "--dry-run", "--once"])
.arg("--mapping")
.arg(&m)
.arg("--state")
.arg(dir.path().join("state2.json"))
.env_remove("MINIO_ACCESS_KEY")
.env_remove("MINIO_SECRET_KEY")
.output()
.unwrap();
let log = String::from_utf8_lossy(&out.stdout).to_string() + &String::from_utf8_lossy(&out.stderr);
assert!(!out.status.success());
assert!(log.contains("MINIO_ACCESS_KEY is not set"), "{log}");
}
}