crates/knut-pipelines/src/demo.rs
//! **The `nordisk_ingest` demo pipeline** — a ready-to-run Spark Declarative
//! Pipeline (SDP) so a pipeline viewer shows a *live, non-empty* dataflow graph
//! out of the box, with no hand-authoring.
//!
//! ## The story it tells
//! Seven Swedish authorities (**m1–m7**) drop case exports as CSV into their own
//! object-store bucket (`s3a://m{n}/…csv`). For **each** authority the pipeline runs
//! the same four-stage lane:
//!
//! ```text
//! s3a://m{n}/*.csv ──parse──▶ m{n}_parsed ──land──▶ m{n}_iceberg ──dedupe──▶ m{n}_dedup ──raw──▶ m{n}_raw
//! ```
//!
//! and then a final [`union`](DataflowGraph) fans all seven `m{n}_raw` into one
//! published `myndigheter_raw` table — so the rendered DAG is seven parallel lanes
//! converging into one node (visually obvious that it's real multi-source ingest).
//!
//! * **parse** — read the raw CSV from the bucket and give it typed columns
//! (a `TemporaryView`, not published).
//! * **land** — write the parsed rows into an Iceberg staging table (`Table`,
//! `format = iceberg`) — the untouched landing zone.
//! * **dedupe** — the demo's point: authorities re-send overlapping batches, so the
//! landing table has duplicate `arende_id`s. A `ROW_NUMBER()` window keeps the
//! latest row per `arende_id` (a published `MaterializedView`), visibly dropping
//! rows.
//! * **raw** — the clean, deduplicated raw layer per authority (`Table`, iceberg).
//!
//! The CSV the demo drops is generated by [`nordisk_ingest_demo_csv`] with a
//! deliberate ~20 % duplicate rate so the dedupe stage measurably removes rows.
//!
//! This is pure model construction over knut's public [`DataflowGraph`]/[`Dataset`]/
//! [`Flow`] types — no `protoc`, no network, no extra deps. A driver
//! ([`crate::client`], feature `grpc`) feeds this graph to Spark over Spark Connect
//! (`define_graph` → `start_run`) exactly like any other SDP; the flows' `query`
//! strings are the SQL Spark actually runs.
use crate::model::{DataflowGraph, Dataset, Flow, OutputType};
/// The seven authorities the demo ingests from: `(bucket code, human name)`.
/// The `code` is the object-store bucket + the dataset-name prefix (`m1`…`m7`).
pub const AGENCIES: [(&str, &str); 7] = [
("m1", "Skatteverket"),
("m2", "Försäkringskassan"),
("m3", "Polismyndigheten"),
("m4", "Migrationsverket"),
("m5", "Kronofogden"),
("m6", "Transportstyrelsen"),
("m7", "Bolagsverket"),
];
/// The published, unioned output every authority's raw layer fans into.
pub const UNION_TABLE: &str = "myndigheter_raw";
/// The CSV header the demo drops into each `s3a://m{n}` bucket (and the columns the
/// `parse` stage reads). One row per authority case record.
pub const CSV_HEADER: &str = "arende_id,personnr,myndighet,arende_typ,datum,belopp,status";
/// Build the `nordisk_ingest` SDP [`DataflowGraph`]: seven `parse→land→dedupe→raw`
/// lanes (one per authority) converging into the published [`UNION_TABLE`].
///
/// Every flow carries the SQL that defines it; the reads (edges) are recovered by
/// [`DataflowGraph::populate_reads`] (called here) so the viewer draws the DAG
/// without replaying the definitions against a server. The result [`validate`]s.
///
/// [`validate`]: DataflowGraph::validate
pub fn nordisk_ingest_demo() -> DataflowGraph {
let mut g = DataflowGraph::new();
g.default_catalog = Some("lake".to_string());
g.default_database = Some("nordisk".to_string());
for (code, name) in AGENCIES {
let parsed = format!("{code}_parsed");
let iceberg = format!("{code}_iceberg");
let dedup = format!("{code}_dedup");
let raw = format!("{code}_raw");
// parse: read the authority's CSV drop from its bucket → typed columns.
g = g
.with_dataset(
Dataset::new(&parsed, OutputType::TemporaryView)
.with_comment(format!("{name} ({code}) — parsed CSV from s3a://{code}")),
)
.with_flow(
Flow::new(format!("parse_{code}"), &parsed, Vec::<String>::new()).with_query(
format!(
"SELECT CAST(arende_id AS BIGINT) AS arende_id, personnr, \
'{name}' AS myndighet, arende_typ, CAST(datum AS DATE) AS datum, \
CAST(belopp AS DOUBLE) AS belopp, status \
FROM read_files('s3a://{code}/', format => 'csv', header => true)"
),
),
);
// land: parsed → Iceberg staging (untouched landing zone).
g = g
.with_dataset(iceberg_table(
&iceberg,
format!("{name} ({code}) — Iceberg landing (raw, with dupes)"),
))
.with_flow(
Flow::new(format!("land_{code}"), &iceberg, Vec::<String>::new())
.with_query(format!("SELECT * FROM {parsed}")),
);
// dedupe: keep the latest row per arende_id (drops the re-sent duplicates).
g = g
.with_dataset(
Dataset::new(&dedup, OutputType::MaterializedView).with_comment(format!(
"{name} ({code}) — deduplicated by arende_id (latest wins)"
)),
)
.with_flow(
Flow::new(format!("dedupe_{code}"), &dedup, Vec::<String>::new()).with_query(
format!(
"SELECT arende_id, personnr, myndighet, arende_typ, datum, belopp, status \
FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY arende_id \
ORDER BY datum DESC) AS _rn FROM {iceberg}) WHERE _rn = 1"
),
),
);
// raw: the clean per-authority raw layer, published to the catalog.
g = g
.with_dataset(iceberg_table(
&raw,
format!("{name} ({code}) — clean raw layer"),
))
.with_flow(
Flow::new(format!("raw_{code}"), &raw, Vec::<String>::new())
.with_query(format!("SELECT * FROM {dedup}")),
);
}
// union: fan all seven raw layers into one published table.
let union_sql = AGENCIES
.iter()
.map(|(code, _)| format!("SELECT * FROM {code}_raw"))
.collect::<Vec<_>>()
.join(" UNION ALL ");
g = g
.with_dataset(iceberg_table(
UNION_TABLE,
"All authorities' clean raw layers, unioned".to_string(),
))
.with_flow(Flow::new("union_all", UNION_TABLE, Vec::<String>::new()).with_query(union_sql));
g.populate_reads();
g
}
/// A `format = iceberg` [`OutputType::Table`] with a comment.
fn iceberg_table(name: &str, comment: String) -> Dataset {
let mut d = Dataset::new(name, OutputType::Table);
d.format = Some("iceberg".to_string());
d.comment = Some(comment);
d
}
/// Generate one authority's CSV drop: `rows` distinct case records for `agency_code`
/// plus a ~20 % tail of exact-duplicate `arende_id` rows (re-sent batches) so the
/// pipeline's dedupe stage measurably removes rows. Deterministic in `(agency_code,
/// batch_seed)` — the same seed yields the same CSV (no RNG crate, no wall clock), so
/// a "generate + ingest" button varies output purely by bumping `batch_seed`.
///
/// The columns match [`CSV_HEADER`]; the first line IS the header.
pub fn nordisk_ingest_demo_csv(agency_code: &str, batch_seed: u64, rows: u32) -> String {
// A tiny deterministic splitmix64-ish hash → pseudo-random field values.
fn mix(mut x: u64) -> u64 {
x = x.wrapping_add(0x9E3779B97F4A7C15);
let mut z = x;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
z ^ (z >> 31)
}
let base = mix(batch_seed) ^ mix(agency_code.bytes().fold(0u64, |a, b| mix(a ^ b as u64)));
let types = [
"ansokan",
"beslut",
"overklagan",
"utbetalning",
"granskning",
];
let statuses = [
"inkommen",
"under_handlaggning",
"beslutad",
"avslutad",
"avvisad",
];
let mut out = String::from(CSV_HEADER);
out.push('\n');
// Distinct records: arende_id is unique within this batch.
for i in 0..rows {
let h = mix(base ^ (i as u64).wrapping_mul(0x100000001B3));
let arende_id = 100_000 + (base % 900_000) + i as u64;
let personnr = 19_000_000_000_000u64 + (h % 800_000_000_000);
let arende_typ = types[(h >> 8) as usize % types.len()];
// A date in 2026, day-of-year from the hash.
let doy = 1 + (h >> 16) % 365;
let month = 1 + doy / 31;
let day = 1 + doy % 28;
let belopp = (h >> 24) % 250_000;
let status = statuses[(h >> 32) as usize % statuses.len()];
out.push_str(&format!(
"{arende_id},{personnr:013},{agency_code},{arende_typ},2026-{month:02}-{day:02},{belopp}.00,{status}\n"
));
}
// ~20 % re-sent duplicates: exact copies of earlier arende_ids (what dedupe removes).
let dupes = (rows / 5).max(1);
for j in 0..dupes {
let src = (mix(base ^ 0xDEAD ^ j as u64) % rows as u64) as u32;
let h = mix(base ^ (src as u64).wrapping_mul(0x100000001B3));
let arende_id = 100_000 + (base % 900_000) + src as u64;
let personnr = 19_000_000_000_000u64 + (h % 800_000_000_000);
let arende_typ = types[(h >> 8) as usize % types.len()];
let doy = 1 + (h >> 16) % 365;
let month = 1 + doy / 31;
let day = 1 + doy % 28;
let belopp = (h >> 24) % 250_000;
let status = statuses[(h >> 32) as usize % statuses.len()];
out.push_str(&format!(
"{arende_id},{personnr:013},{agency_code},{arende_typ},2026-{month:02}-{day:02},{belopp}.00,{status}\n"
));
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn nordisk_ingest_demo_is_a_valid_seven_lane_graph() {
let g = nordisk_ingest_demo();
g.validate()
.expect("the demo pipeline must be internally consistent");
// 7 authorities × 4 datasets + 1 union = 29 nodes; same count of flows.
assert_eq!(g.datasets.len(), 7 * 4 + 1);
assert_eq!(g.flows.len(), 7 * 4 + 1);
// The union reads all seven raw layers (edges recovered from its SQL).
let union = g.flows.iter().find(|f| f.target == UNION_TABLE).unwrap();
assert_eq!(union.reads.len(), 7, "union must fan in all 7 raw layers");
// Each lane's edges were recovered: dedupe reads its iceberg table.
let dd = g.flows.iter().find(|f| f.name == "dedupe_m1").unwrap();
assert_eq!(dd.reads, vec!["m1_iceberg".to_string()]);
}
#[test]
fn generated_csv_has_header_and_duplicate_tail() {
let csv = nordisk_ingest_demo_csv("m1", 42, 50);
let lines: Vec<&str> = csv.lines().collect();
assert_eq!(lines[0], CSV_HEADER);
// 50 distinct + 10 dupes + header = 61 lines.
assert_eq!(lines.len(), 1 + 50 + 10);
// Deterministic: same seed → identical bytes.
assert_eq!(csv, nordisk_ingest_demo_csv("m1", 42, 50));
// Different seed → different bytes (so a re-press varies the drop).
assert_ne!(csv, nordisk_ingest_demo_csv("m1", 43, 50));
// There ARE duplicate arende_ids (what the pipeline dedupes).
let ids: Vec<&str> = csv
.lines()
.skip(1)
.map(|l| l.split(',').next().unwrap())
.collect();
let distinct: std::collections::HashSet<_> = ids.iter().collect();
assert!(
distinct.len() < ids.len(),
"must contain duplicate arende_ids"
);
}
}