crates/knut-bifrost/tests/integration_spark41_streaming.rs
//! TOP GUN, blessed pair: the full streaming bifröst pipeline against the
//! **blessed GA** container pair —
//!
//! Spark **4.1.2** Connect + `iceberg-spark-runtime-4.1_2.13:1.11.0`
//! ──stream (sync round → delta round)──► FalkorDB
//!
//! This is the sibling of `integration_spark_falkordb.rs` pinned to the
//! constellation's *blessed* Spark/Iceberg pair (Spark 4.1.2 GA + Iceberg
//! 1.11.0), and it brings BOTH containers up **from Rust** (LAW 7) by reusing
//! the shared helpers instead of re-implementing container boot:
//!
//! - Spark Connect ← [`knut_pipelines::server::SparkServer`] (pure-Rust bring-up
//! over podman's REST API through draupnir; its default image is the blessed
//! 4.1.2 GA tag). We point it at the Iceberg-enabled build of that pair
//! (`knut-spark-iceberg`, based on `apache/spark:4.1.2-…`) so the `lake`
//! Iceberg catalog is live; override with `KNUT_SPARK_IMAGE`.
//! - FalkorDB ← [`graphar_falkordb::falkordb_container::start_falkordb`] (the
//! shared testcontainers→Podman autostart used by the graphar-falkordb
//! integration tests and `knut-popviz`).
//!
//! Rows inserted into an Iceberg table through Spark SQL come out as nodes and
//! relationships in FalkorDB, and the **streaming** contract is asserted: an
//! initial full sync, an idle round that ships nothing, then an incremental
//! round that ships only the appended delta.
//!
//! ## Prerequisites
//!
//! - Podman socket: `systemctl --user enable --now podman.socket`
//! - The blessed Iceberg image built once (pairs Spark 4.1.2 GA + Iceberg 1.11.0):
//! ```bash
//! podman build -t knut-spark-iceberg -f containers/spark-iceberg/Containerfile .
//! ```
//! (FalkorDB is pulled/started automatically by the autostart helper.)
//!
//! ## Run
//!
//! ```bash
//! cargo test -p knut-bifrost --features spark41 \
//! --test integration_spark41_streaming -- --ignored --nocapture
//! ```
#![cfg(feature = "spark41")]
use std::time::Duration;
use knut_bifrost::{
Listener, MappingSpec, SyncState, TransformRegistry, sink::FalkorDbSink,
source::SparkConnectSource,
};
use knut_pipelines::server::SparkServer;
/// The blessed pair's Iceberg-enabled image. The stock `apache/spark:4.1.2-…`
/// tag ([`SparkServer`]'s default) has no Iceberg runtime jar, so the streaming
/// *Iceberg* round-trip needs the baked image. Override with `KNUT_SPARK_IMAGE`.
const BLESSED_ICEBERG_IMAGE: &str = "localhost/knut-spark-iceberg:latest";
const MAPPING_YAML: &str = r#"
version: 1
tables:
- table: lake.itest.users
mode: incremental
pipeline:
- vertex:
label: Person
id_column: user_id
properties: [name]
- fk_edges:
type: WORKS_AT
src: { label: Person, column: user_id }
dst: { label: Company, column: company_id }
- table: lake.itest.companies
mode: full
pipeline:
- vertex:
label: Company
id_column: company_id
properties: [company_name]
"#;
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)
}
/// Connect a [`SparkConnectSource`] once Spark Connect is answering a real
/// query — the gRPC channel connects before the JVM is ready, so probe with SQL.
async fn connect_when_ready(url: &str) -> SparkConnectSource {
let mut attempt = 0;
loop {
let probe = match SparkConnectSource::connect(url).await {
Ok(s) => matches!(s.sql("SELECT 1").await, Ok(_)).then_some(s),
Err(_) => None,
};
match probe {
Some(s) => break s,
None if attempt < 60 => {
attempt += 1;
tokio::time::sleep(Duration::from_secs(3)).await;
}
None => panic!("Spark Connect never became ready after 180s"),
}
}
}
#[tokio::test]
#[ignore = "needs podman + the blessed knut-spark-iceberg image; run with --ignored"]
async fn blessed_pair_iceberg_streams_into_falkordb() {
// ── FalkorDB via the shared graphar-falkordb autostart (LAW 7) ──────────
graphar_falkordb::falkordb_container::ensure_docker_host();
let falkor = graphar_falkordb::falkordb_container::start_falkordb()
.await
.expect("autostart FalkorDB (needs podman.socket)");
let falkor_url = falkor.url().to_string();
// ── Spark Connect (blessed 4.1.2 GA + Iceberg 1.11.0) via SparkServer ───
// SparkServer's default is the stock 4.1.2 GA tag (no Iceberg jar); point it
// at the baked Iceberg image so the `lake` catalog is live. KNUT_SPARK_IMAGE
// still wins if set.
let image =
std::env::var("KNUT_SPARK_IMAGE").unwrap_or_else(|_| BLESSED_ICEBERG_IMAGE.to_string());
let spark = SparkServer::connect()
.expect("podman API socket (enable with: systemctl --user enable --now podman.socket)")
.with_name("knut-bifrost-spark41")
.with_image(image);
let sc_url = spark.up().await.expect("bring up blessed Spark Connect");
// ── Iceberg fixtures through Spark SQL ──────────────────────────────────
let source = connect_when_ready(&sc_url).await;
source
.sql("CREATE NAMESPACE IF NOT EXISTS lake.itest")
.await
.unwrap();
source
.sql("CREATE TABLE lake.itest.users (user_id BIGINT, name STRING, company_id BIGINT) USING iceberg")
.await
.unwrap();
source
.sql("CREATE TABLE lake.itest.companies (company_id BIGINT, company_name STRING) USING iceberg")
.await
.unwrap();
source
.sql("INSERT INTO lake.itest.companies VALUES (10, 'acme'), (11, 'globex')")
.await
.unwrap();
source
.sql("INSERT INTO lake.itest.users VALUES (1, 'alice', 10), (2, 'bob', 10), (3, 'carol', 11)")
.await
.unwrap();
// ── bifröst sync round 1: initial full materialisation ──────────────────
let state_dir = tempfile::tempdir().unwrap();
let sink = FalkorDbSink::connect(&falkor_url, "itest").await.unwrap();
let mut listener = Listener::new(
MappingSpec::from_yaml(MAPPING_YAML).unwrap(),
TransformRegistry::new(),
source,
sink,
SyncState::open(state_dir.path().join("state.json")).unwrap(),
);
let round = listener.sync_once().await.unwrap();
eprintln!("round 1: {round:?}");
assert_eq!(round.failed, vec![], "round 1 failures");
assert_eq!(round.synced.len(), 2, "both tables synced");
let client = redis::Client::open(falkor_url.clone()).unwrap();
let mut conn = client.get_multiplexed_async_connection().await.unwrap();
assert_eq!(
count(&mut conn, "itest", "MATCH (n:Person) RETURN count(n)").await,
3
);
assert_eq!(
count(&mut conn, "itest", "MATCH (n:Company) RETURN count(n)").await,
2
);
assert_eq!(
count(
&mut conn,
"itest",
"MATCH (:Person)-[r:WORKS_AT]->(:Company) RETURN count(r)"
)
.await,
3
);
// ── idle round: nothing changed, nothing re-shipped ─────────────────────
let round = listener.sync_once().await.unwrap();
assert!(round.synced.is_empty(), "no snapshot change → no sync");
// ── incremental round: one appended row streams through as a delta ──────
// (the listener owns the source now; reuse raw SQL through a 2nd session)
let admin = SparkConnectSource::connect(&sc_url).await.unwrap();
admin
.sql("INSERT INTO lake.itest.users VALUES (4, 'dave', 11)")
.await
.unwrap();
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "round 3 failures");
let users_round: Vec<_> = round
.synced
.iter()
.filter(|(t, _, _)| t == "lake.itest.users")
.collect();
assert_eq!(users_round.len(), 1);
assert_eq!(users_round[0].1, 1, "incremental read ships only the delta");
assert_eq!(
count(&mut conn, "itest", "MATCH (n:Person) RETURN count(n)").await,
4
);
assert_eq!(
count(
&mut conn,
"itest",
"MATCH (:Person)-[r:WORKS_AT]->(:Company) RETURN count(r)"
)
.await,
4
);
// ── teardown: stop the managed Spark container (FalkorDB drops with `falkor`)
let _ = spark.stop().await;
drop(falkor);
}