crates/knut-bifrost/src/spark_demo.rs
//! **The knut ▸ Spark 4.1 + Iceberg 1.11.0 LOAD demo — one shared source of truth.**
//!
//! knut is a headless library workspace: it has no Tools menu of its own, so the
//! *selectable* demo lives in korp's shared 🧰 Tools menu (`korp::tools`) and drives
//! **this** module. Everything that is knut's business — which Iceberg namespace and
//! table the demo lands in, the exact SQL it issues, and how many rows it writes —
//! lives HERE, so the GUI entry, the headless CLI path, and the tests all read the
//! SAME plan instead of three copies of a SQL string.
//!
//! ## The two halves
//! * [`iceberg_demo_plan`] is **pure** and **feature-free**: it builds the ordered SQL
//! program (namespace → table → insert) plus the verification query. It compiles and
//! is unit-tested in the DEFAULT build — no Spark, no container, no network — so the
//! wiring is provable on any box.
//! * [`run_iceberg_demo`] (feature `spark`) EXECUTES that plan against a live Spark
//! Connect 4.x endpoint through the unchanged [`SparkConnectSource`] client — the
//! exact same client `tests/integration_spark41_streaming.rs` and korp's
//! `integration_spark41_iceberg` use. It needs a real container host.
//!
//! ## Blessed vintage
//! The demo targets Spark **4.1.2 GA** + `iceberg-spark-runtime-4.1_2.13:1.11.0` (the
//! pair Iceberg 1.11.0 actually ships a runtime for), reached over `spark-connect`
//! 0.2.2's arrow-59 fork so the batches that come back ARE workspace
//! `arrow_array::RecordBatch`es — no bridge, no IPC round-trip.
/// The default Iceberg **catalog** the blessed 4.1 + Iceberg image bakes into its
/// `spark-defaults.conf`. Everything the demo writes is `lake.<namespace>.<table>`.
pub const DEMO_CATALOG: &str = "lake";
/// The default namespace the demo lands in — scoped to knut so it can never collide
/// with a customer namespace in the same catalog.
pub const DEMO_NAMESPACE: &str = "knut_demo";
/// The default table the demo writes.
pub const DEMO_TABLE: &str = "people";
/// The default row count — small ON PURPOSE. This is a *wiring* demo (does the
/// Connect 4.x handshake work, is the Iceberg 1.11.0 runtime live, does a snapshot
/// commit); volume benchmarking is a bench-box job, not a Tools-menu click.
pub const DEMO_ROWS: usize = 8;
/// The ordered SQL program the knut ▸ Spark 4.1 + Iceberg demo runs, plus the
/// queries that VERIFY it landed. Pure data — build it anywhere, assert it in a
/// container-free test, execute it with [`run_iceberg_demo`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IcebergDemoPlan {
/// The Spark Connect endpoint the plan is executed against (`sc://host:port`).
pub url: String,
/// Fully-qualified namespace (`lake.knut_demo`).
pub namespace: String,
/// Fully-qualified table (`lake.knut_demo.people`).
pub table: String,
/// How many rows the INSERT writes.
pub rows: usize,
/// The statements to run IN ORDER: drop → create namespace → create table →
/// insert. Idempotent: re-running the demo re-creates the table from scratch.
pub statements: Vec<String>,
/// `SELECT count(*)` over the landed table — must return [`Self::rows`].
pub verify_count: String,
/// The Iceberg `.snapshots` metadata query — the exact path
/// [`SparkConnectSource::fetch_new`](crate::source::SparkConnectSource) polls.
/// A row here proves the Iceberg extension really loaded (not just Spark SQL).
pub verify_snapshots: String,
}
impl IcebergDemoPlan {
/// Every statement + verification query, in execution order — the flat program a
/// runner walks and a test asserts over.
pub fn program(&self) -> Vec<&str> {
self.statements
.iter()
.map(String::as_str)
.chain([self.verify_count.as_str(), self.verify_snapshots.as_str()])
.collect()
}
}
/// Build the demo plan for `url`, writing [`DEMO_ROWS`] rows into
/// `lake.knut_demo.people`. Pure — no I/O, no feature gate, no container.
pub fn iceberg_demo_plan(url: &str) -> IcebergDemoPlan {
iceberg_demo_plan_for(url, DEMO_CATALOG, DEMO_NAMESPACE, DEMO_TABLE, DEMO_ROWS)
}
/// [`iceberg_demo_plan`] with every knob explicit (a second zone/catalog can reuse the
/// same program without a second SQL copy).
pub fn iceberg_demo_plan_for(
url: &str,
catalog: &str,
namespace: &str,
table: &str,
rows: usize,
) -> IcebergDemoPlan {
let ns = format!("{catalog}.{namespace}");
let tbl = format!("{ns}.{table}");
// A deterministic, network-free row set — the demo must produce the SAME data on
// every box so the count assertion is exact, not "roughly some rows".
let values = (0..rows)
.map(|i| format!("({i}, 'person_{i}', {})", 20 + i))
.collect::<Vec<_>>()
.join(", ");
IcebergDemoPlan {
url: url.to_string(),
namespace: ns.clone(),
table: tbl.clone(),
rows,
statements: vec![
// Idempotent re-run: drop first so a second click is not an append.
format!("DROP TABLE IF EXISTS {tbl}"),
format!("CREATE NAMESPACE IF NOT EXISTS {ns}"),
format!("CREATE TABLE {tbl} (id BIGINT, name STRING, age BIGINT) USING iceberg"),
format!("INSERT INTO {tbl} VALUES {values}"),
],
verify_count: format!("SELECT count(*) AS n FROM {tbl}"),
verify_snapshots: format!("SELECT snapshot_id FROM {tbl}.snapshots"),
}
}
/// The cargo feature a caller must enable for [`run_iceberg_demo`] to have a Spark
/// Connect client to run through. knut's own is `spark41` (which turns on `spark` plus
/// the container autostart the integration tests link); korp spells its passthrough
/// `spark41` too, so ONE name is correct in both mouths.
pub const DEMO_FEATURE: &str = "spark41";
/// **Is the Spark Connect client actually linked into this build?**
///
/// A UI must be able to ask BEFORE it offers the button, and a test must be able to
/// assert which half of [`run_iceberg_demo`] it is exercising. `false` means the demo
/// can only fail — see [`feature_missing_error`].
pub const fn spark_client_linked() -> bool {
cfg!(feature = "spark")
}
/// The ONE wording for "this build cannot do the load", naming the missing feature AND
/// the target table so the operator can act on it.
///
/// It lives HERE, in knut, because knut owns the demo (L5). It used to exist only as a
/// `#[cfg]` twin invented inside korp: without the feature, `run_iceberg_demo` did not
/// exist AT ALL, so every consumer had to write its own stub and its own message — and a
/// consumer that forgot would either not compile or, worse, quietly route the click
/// somewhere that reported success. Now the symbol exists in EVERY build and the
/// feature-off build returns this.
pub fn feature_missing_error(table: &str) -> String {
format!(
"the knut -> Spark 4.1 + Iceberg load needs --features {DEMO_FEATURE} \
(target {table}); this build linked no Spark Connect client, so it cannot \
run the load at all"
)
}
/// Execute an [`IcebergDemoPlan`] against a live Spark Connect 4.x endpoint through the
/// UNCHANGED [`SparkConnectSource`](crate::source::SparkConnectSource) client, and
/// return the landed row count.
///
/// Needs a running Spark **4.1.2 GA + Iceberg 1.11.0** Connect server — in the
/// constellation that container is brought up THROUGH jera (korp's
/// `infra::spark_iceberg_spec` → `start_via_jera` → draupnir's one OCI engine); this
/// function never starts a container itself.
///
/// Fails LOUD: a missing Iceberg runtime surfaces as the CREATE TABLE error, a wrong
/// row count as an explicit mismatch — never a silent "probably fine".
#[cfg(feature = "spark")]
pub async fn run_iceberg_demo(plan: &IcebergDemoPlan) -> crate::error::Result<i64> {
use arrow_array::{Array, Int64Array};
use crate::error::BifrostError;
use crate::source::SparkConnectSource;
let source = SparkConnectSource::connect(&plan.url).await?;
for stmt in &plan.statements {
source.sql(stmt).await?;
}
let batches = source.sql(&plan.verify_count).await?;
let count = 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))
})
.ok_or_else(|| {
BifrostError::Source(format!("no count came back from {}", plan.verify_count))
})?;
if count != plan.rows as i64 {
return Err(BifrostError::Source(format!(
"{} landed {count} rows, expected {}",
plan.table, plan.rows
)));
}
// The Iceberg `.snapshots` metadata table must exist — proof the Iceberg 1.11.0
// extension is live on the 4.1 line, not just plain Spark SQL.
let snaps = source.sql(&plan.verify_snapshots).await?;
if !snaps.iter().any(|b| b.num_rows() > 0) {
return Err(BifrostError::Source(format!(
"{} committed no Iceberg snapshot — is the Iceberg runtime loaded?",
plan.table
)));
}
Ok(count)
}
/// **The feature-off half of [`run_iceberg_demo`] — same signature, honest answer.**
///
/// Without the `spark` client there is nothing to run the plan through, so this returns
/// [`feature_missing_error`] instead of not existing. That is the whole point: a
/// consumer (korp's 🧰 Tools entry, the `bifrost iceberg-demo` subcommand) writes ONE
/// call site, compiles in every build, and a lean build fails LOUD at the click with the
/// feature name and `lake.knut_demo.people` in the message — rather than the button
/// being silently absent, or a hand-written twin stub drifting from the real one.
#[cfg(not(feature = "spark"))]
pub async fn run_iceberg_demo(plan: &IcebergDemoPlan) -> crate::error::Result<i64> {
Err(crate::error::BifrostError::Source(feature_missing_error(
&plan.table,
)))
}
#[cfg(test)]
mod tests {
use super::*;
/// **RED-when-broken: the demo plan targets the blessed `lake` Iceberg catalog with
/// a knut-scoped namespace, and its program is the ordered drop→namespace→table→
/// insert.** Container-free: this is the wiring proof that runs on any box.
#[test]
fn demo_plan_is_the_ordered_iceberg_program() {
let plan = iceberg_demo_plan("sc://localhost:15012");
assert_eq!(plan.url, "sc://localhost:15012");
assert_eq!(plan.namespace, "lake.knut_demo");
assert_eq!(plan.table, "lake.knut_demo.people");
assert_eq!(plan.rows, DEMO_ROWS);
// Ordered program: a CREATE before the namespace exists, or an INSERT before the
// CREATE, is a broken demo.
assert_eq!(plan.statements.len(), 4);
assert!(
plan.statements[0].starts_with("DROP TABLE IF EXISTS"),
"idempotent re-run"
);
assert!(plan.statements[1].starts_with("CREATE NAMESPACE IF NOT EXISTS"));
assert!(
plan.statements[2].contains("USING iceberg"),
"the table is ICEBERG, not parquet"
);
assert!(plan.statements[3].starts_with(&format!("INSERT INTO {}", plan.table)));
// The verification half polls the exact `.snapshots` path the streaming source
// uses, so a green demo also proves the listener's poll path is live.
assert_eq!(
plan.verify_snapshots,
format!("SELECT snapshot_id FROM {}.snapshots", plan.table)
);
assert!(plan.verify_count.contains("count(*)"));
assert_eq!(plan.program().len(), 6, "4 statements + 2 verifications");
}
/// **RED-when-broken: the INSERT is deterministic and writes exactly `rows` tuples.**
/// A demo whose row count drifts makes the count assertion meaningless.
#[test]
fn demo_insert_is_deterministic_and_exact() {
let plan = iceberg_demo_plan("sc://localhost:15012");
let insert = &plan.statements[3];
assert_eq!(insert.matches("person_").count(), DEMO_ROWS);
assert!(insert.contains("(0, 'person_0', 20)"));
let last = DEMO_ROWS - 1;
assert!(insert.contains(&format!("({last}, 'person_{last}', {})", 20 + last)));
// Byte-identical on a re-build — no timestamps, no rng, no hostname.
assert_eq!(iceberg_demo_plan("sc://localhost:15012"), plan);
}
/// **RED-when-broken: the "you didn't enable the feature" message names BOTH the
/// feature and the target table.**
///
/// This is the exact string an operator reads when a lean build's demo button is
/// clicked. If it stops naming `--features spark41` the operator cannot act on it;
/// if it stops naming `lake.knut_demo.people` they cannot tell WHICH load failed.
#[test]
fn feature_missing_error_names_the_feature_and_the_table() {
let plan = iceberg_demo_plan("sc://localhost:15012");
let msg = feature_missing_error(&plan.table);
assert!(
msg.contains("--features spark41"),
"names the feature: {msg}"
);
assert!(
msg.contains("lake.knut_demo.people"),
"names the target table: {msg}"
);
assert_eq!(DEMO_FEATURE, "spark41");
}
/// **RED-when-broken: `run_iceberg_demo` EXISTS in every build, and the build that
/// linked no Spark client fails LOUD instead of reporting success.**
///
/// Both halves are asserted from the one place that knows which is compiled:
/// - no `spark` → the call resolves, and returns `Err` carrying the feature name.
/// (A stub that returned `Ok(0)` — "did nothing, reported success" — is exactly
/// what this forbids.)
/// - with `spark` → the symbol is the real one; it is not called here (it needs a
/// live container), `tests/integration_spark41_iceberg_demo.rs` runs it.
///
/// The `spark_client_linked()` flag must agree with which half compiled, so a UI
/// that greys the button by it can never disagree with what the button would do.
#[test]
fn run_iceberg_demo_exists_in_every_build_and_fails_loud_without_the_client() {
assert_eq!(spark_client_linked(), cfg!(feature = "spark"));
#[cfg(not(feature = "spark"))]
{
let plan = iceberg_demo_plan("sc://localhost:15012");
let err = tokio::runtime::Builder::new_current_thread()
.build()
.expect("current-thread runtime")
.block_on(run_iceberg_demo(&plan))
.expect_err("a build with no Spark client CANNOT have landed rows");
let msg = err.to_string();
assert!(
msg.contains("--features spark41"),
"loud about the feature: {msg}"
);
assert!(
msg.contains("lake.knut_demo.people"),
"loud about the table: {msg}"
);
}
}
/// **RED-when-broken: the knobs really scope the plan** — a second catalog/namespace
/// reuses the SAME program instead of a copied SQL string.
#[test]
fn demo_plan_knobs_scope_catalog_namespace_and_rows() {
let plan = iceberg_demo_plan_for("sc://box:9999", "lake", "zone_test", "boats", 3);
assert_eq!(plan.table, "lake.zone_test.boats");
assert_eq!(plan.rows, 3);
assert_eq!(plan.statements[3].matches("person_").count(), 3);
assert_ne!(plan.table, iceberg_demo_plan("sc://box:9999").table);
}
}