crates/knut-bifrost/tests/integration_spark41_iceberg_demo.rs
//! **The knut ▸ Spark 4.1 + Iceberg 1.11.0 LOAD demo, EXECUTED.**
//!
//! `spark_demo::run_iceberg_demo` had unit tests over the *plan* and none over the
//! *load*: nothing in the workspace ever ran it. So "the demo works" rested on a SQL
//! string being shaped right — the plan could be perfect while the load was dead, and
//! nothing would have gone red. This test runs it for real against a live Spark
//! **4.1.2 GA** + `iceberg-spark-runtime-4.1_2.13:1.11.0` Connect endpoint and then
//! re-reads the table through an INDEPENDENT session, so the assertion never rests on
//! the loader's own report of itself.
//!
//! Target: **`lake.knut_demo.people`** — the exact table korp's 🧰 Tools entry and the
//! `bifrost iceberg-demo` subcommand write, from the one shared plan.
//!
//! ## Prerequisites (LAW 7 — the container comes up from Rust)
//! - Podman socket: `systemctl --user enable --now podman.socket`
//! - The blessed image, e.g. `localhost/nordisk-spark-iceberg:4.1.2`, published on
//! `15012`. korp brings it up with `korp::infra::bring_up_spark_iceberg`; knut's own
//! `integration_spark41_streaming` brings up its equivalent through `SparkServer`.
//! Point this test at whatever is live with `$KNUT_SPARK_ICEBERG_URL`.
//!
//! ## Run
//! ```bash
//! cargo test -p knut-bifrost --features spark41 \
//! --test integration_spark41_iceberg_demo -- --ignored --nocapture
//! ```
#![cfg(feature = "spark41")]
use std::time::Duration;
use arrow_array::{Array, Int64Array, StringArray};
use knut_bifrost::source::SparkConnectSource;
use knut_bifrost::spark_demo::{self, DEMO_ROWS};
/// The endpoint the demo runs against: `$KNUT_SPARK_ICEBERG_URL`, else korp's published
/// Spark 4.1 + Iceberg port (`sc://localhost:15012`).
fn demo_url() -> String {
std::env::var("KNUT_SPARK_ICEBERG_URL").unwrap_or_else(|_| "sc://localhost:15012".to_string())
}
/// Connect once Spark Connect answers a real query — the gRPC channel accepts long
/// before the JVM will plan SQL, so probing with `SELECT 1` is the only honest ready
/// signal.
async fn connect_when_ready(url: &str) -> SparkConnectSource {
for attempt in 0..60u32 {
if let Ok(s) = SparkConnectSource::connect(url).await {
if s.sql("SELECT 1").await.is_ok() {
return s;
}
}
let _ = attempt;
tokio::time::sleep(Duration::from_secs(3)).await;
}
panic!("Spark Connect at {url} never became ready after 180s");
}
/// **RED-when-broken: the demo LOADS. `lake.knut_demo.people` really holds
/// [`DEMO_ROWS`] Iceberg rows afterwards, and an Iceberg snapshot really committed.**
///
/// Three layers, each able to go red on its own:
/// 1. `run_iceberg_demo` returns the landed count (its own report).
/// 2. A **second, independent** session re-reads the table: the count, the id/name/age
/// of the first and last row. A loader that reported 8 while writing nothing, or
/// wrote to a different table, dies here.
/// 3. `.snapshots` carries a row — proof the Iceberg 1.11.0 extension is live, not
/// just Spark SQL over a session catalog.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "needs a live Spark 4.1 + Iceberg Connect endpoint; run with --ignored"]
async fn iceberg_demo_lands_rows_in_lake_knut_demo_people() {
let url = demo_url();
// Wait for the endpoint before the demo, so a slow JVM is not read as a failed load.
let probe = connect_when_ready(&url).await;
drop(probe);
let plan = spark_demo::iceberg_demo_plan(&url);
assert_eq!(
plan.table, "lake.knut_demo.people",
"the demo's blessed target"
);
assert!(
spark_demo::spark_client_linked(),
"this build linked the Spark client"
);
// ── 1) the load, reporting its own count ──────────────────────────────────
let landed = spark_demo::run_iceberg_demo(&plan)
.await
.unwrap_or_else(|e| panic!("knut -> Spark 4.1 + Iceberg load into {}: {e}", plan.table));
assert_eq!(
landed, DEMO_ROWS as i64,
"the loader reports every planned row landed"
);
// ── 2) an INDEPENDENT session re-reads the table ──────────────────────────
let check = SparkConnectSource::connect(&url)
.await
.expect("second Connect session");
let count = check
.sql(&plan.verify_count)
.await
.expect("count over the landed table");
assert_eq!(
scalar_i64(&count),
Some(DEMO_ROWS as i64),
"a session that did NOT write the rows still sees {DEMO_ROWS} in {}",
plan.table,
);
// The actual VALUES, not just the cardinality: `person_0`/20 first, `person_7`/27
// last. A truncating or mis-typed INSERT keeps the count and loses this.
let rows = check
.sql(&format!(
"SELECT id, name, age FROM {} ORDER BY id",
plan.table
))
.await
.expect("read the landed rows back");
let (ids, names, ages) = columns(&rows);
assert_eq!(ids.len(), DEMO_ROWS, "every planned row is readable");
assert_eq!((ids[0], names[0].as_str(), ages[0]), (0, "person_0", 20));
let last = DEMO_ROWS - 1;
assert_eq!(
(ids[last], names[last].as_str(), ages[last]),
(
last as i64,
format!("person_{last}").as_str(),
20 + last as i64
),
);
// ── 3) Iceberg really committed a snapshot ────────────────────────────────
let snaps = check
.sql(&plan.verify_snapshots)
.await
.expect("the `.snapshots` metadata table (Iceberg extension loaded)");
let snapshot_ids: Vec<i64> = snaps
.iter()
.filter_map(|b| b.column(0).as_any().downcast_ref::<Int64Array>())
.flat_map(|a| (0..a.len()).map(|i| a.value(i)).collect::<Vec<_>>())
.collect();
assert!(
!snapshot_ids.is_empty(),
"{} committed at least one Iceberg snapshot",
plan.table,
);
eprintln!(
"knut spark41 demo: {} rows in {} · snapshots {:?}",
landed, plan.table, snapshot_ids,
);
}
/// The single scalar of a one-cell result (`SELECT count(*)`).
fn scalar_i64(batches: &[arrow_array::RecordBatch]) -> Option<i64> {
batches
.iter()
.find(|b| b.num_rows() > 0 && b.num_columns() > 0)
.and_then(|b| {
b.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.map(|a| a.value(0))
})
}
/// Flatten `SELECT id, name, age` batches into three parallel column vectors.
fn columns(batches: &[arrow_array::RecordBatch]) -> (Vec<i64>, Vec<String>, Vec<i64>) {
let (mut ids, mut names, mut ages) = (Vec::new(), Vec::new(), Vec::new());
for b in batches {
if b.num_rows() == 0 {
continue;
}
let id = b
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.expect("id BIGINT");
let name = b
.column(1)
.as_any()
.downcast_ref::<StringArray>()
.expect("name STRING");
let age = b
.column(2)
.as_any()
.downcast_ref::<Int64Array>()
.expect("age BIGINT");
for i in 0..b.num_rows() {
ids.push(id.value(i));
names.push(name.value(i).to_string());
ages.push(age.value(i));
}
}
(ids, names, ages)
}