knut-bifrost 0.3.0

//! **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);
    }
}