crates/knut-thund/examples/medallion_sql.rs
//! A **scalable** medallion pipeline authored entirely in SQL.
//!
//! Generates a bronze/silver/gold medallion graph over `N` source domains,
//! parses it with the [`sql`](knut_thund::authoring::sql) front-end, and
//! reports what each stage costs — so "does the SQL surface scale?" is a
//! measurement, not a claim.
//!
//! Per domain the script emits three datasets and a wide fan-in at the end:
//!
//! ```text
//! landing_<i> -> bronze_<i> -> silver_<i> -> gold_live_<i>
//! (MV, months(event_ts), (STREAMING TABLE,
//! EXPECT amount >= 0) FROM STREAM(silver_<i>))
//!
//! silver_0 .. silver_<n-1> -> gold_all (one UNION ALL over every domain)
//! ```
//!
//! `gold_all` is the interesting part for scaling: a single flow whose `reads`
//! is *every* silver dataset, which is what stresses `collect_reads` and the
//! topological sort rather than just the per-statement parse.
//!
//! # Run
//!
//! ```bash
//! cargo run -p knut-thund --features sql,dsl --release --example medallion_sql -- 500
//! ```
//!
//! `N` defaults to 200 (601 datasets, 601 flows). It is a pure front-end
//! benchmark: no backend runs, so it needs neither `native` nor a Spark server.
use std::time::Instant;
use knut_thund::authoring::sql;
use knut_thund::ir::OutputType;
/// Build the medallion SQL script for `n` source domains.
fn generate(n: usize) -> String {
// Pre-size roughly: ~500 bytes per domain keeps this to one allocation.
let mut s = String::with_capacity(n * 512 + 64 * n);
for i in 0..n {
s.push_str(&format!(
"CREATE MATERIALIZED VIEW bronze_{i} (
event_id BIGINT NOT NULL,
event_ts TIMESTAMP,
user_id STRING,
amount DOUBLE,
CONSTRAINT amount_non_negative_{i} EXPECT (amount >= 0) ON VIOLATION DROP ROW
)
PARTITIONED BY (months(event_ts))
COMMENT 'bronze domain {i}'
AS SELECT event_id, event_ts, user_id, amount FROM landing_{i};
CREATE MATERIALIZED VIEW silver_{i} AS
SELECT event_id, event_ts, user_id, amount
FROM bronze_{i}
WHERE amount > 0;
CREATE STREAMING TABLE gold_live_{i} AS
SELECT user_id, amount FROM STREAM(silver_{i});
"
));
}
// The wide fan-in: one dataset reading all `n` silver tables.
s.push_str("CREATE MATERIALIZED VIEW gold_all\nPARTITIONED BY (months(event_ts))\nAS\n");
for i in 0..n {
if i > 0 {
s.push_str("UNION ALL\n");
}
s.push_str(&format!(
"SELECT event_id, event_ts, user_id, amount FROM silver_{i}\n"
));
}
s.push_str(";\n");
s
}
fn main() {
let n: usize = std::env::args()
.nth(1)
.and_then(|a| a.parse().ok())
.unwrap_or(200);
println!("== Þund SQL front-end — medallion over {n} domains ==\n");
let t = Instant::now();
let script = generate(n);
let gen_ms = t.elapsed().as_secs_f64() * 1e3;
println!(
"generate {:>8.1} ms {} statements, {:.1} KiB of SQL",
gen_ms,
n * 3 + 1,
script.len() as f64 / 1024.0
);
let t = Instant::now();
let pipeline = match sql::from_sql("medallion", &script) {
Ok(p) => p,
Err(e) => {
eprintln!("parse failed: {e}");
std::process::exit(1);
}
};
let parse_ms = t.elapsed().as_secs_f64() * 1e3;
println!(
"SQL -> IR {:>8.1} ms {} datasets, {} flows ({:.0} stmt/s)",
parse_ms,
pipeline.datasets.len(),
pipeline.flows.len(),
(n * 3 + 1) as f64 / (parse_ms / 1e3),
);
// `from_sql` already validated; time the topological sort on its own since
// that is what the wide fan-in actually stresses.
let t = Instant::now();
let order = pipeline.topo_order().expect("medallion graph is acyclic");
let topo_ms = t.elapsed().as_secs_f64() * 1e3;
println!(
"topo_order {:>8.1} ms {} nodes ordered",
topo_ms,
order.len()
);
let t = Instant::now();
let rendered = sql::to_sql(&pipeline).expect("render");
let render_ms = t.elapsed().as_secs_f64() * 1e3;
println!(
"IR -> SQL {:>8.1} ms {:.1} KiB",
render_ms,
rendered.len() as f64 / 1024.0
);
let t = Instant::now();
let reparsed = sql::from_sql("medallion", &rendered).expect("re-parse");
let reparse_ms = t.elapsed().as_secs_f64() * 1e3;
println!("SQL -> IR (2) {:>8.1} ms round-trip", reparse_ms);
assert_eq!(
pipeline.datasets.len(),
reparsed.datasets.len(),
"round-trip lost datasets"
);
#[cfg(feature = "dsl")]
{
use knut_thund::authoring::dsl;
let t = Instant::now();
let ron = dsl::to_ron(&pipeline).expect("to_ron");
let ron_out_ms = t.elapsed().as_secs_f64() * 1e3;
let t = Instant::now();
let from = dsl::from_ron(&ron).expect("from_ron");
let ron_in_ms = t.elapsed().as_secs_f64() * 1e3;
println!(
"IR -> RON {:>8.1} ms {:.1} KiB",
ron_out_ms,
ron.len() as f64 / 1024.0
);
println!("RON -> IR {:>8.1} ms", ron_in_ms);
assert_eq!(pipeline, from, "RON round-trip must be lossless");
println!("\n SQL -> IR -> RON -> IR is bit-identical.");
}
// --- what the graph actually contains -------------------------------
let mvs = pipeline
.datasets
.iter()
.filter(|d| d.output_type == OutputType::MaterializedView)
.count();
let tables = pipeline
.datasets
.iter()
.filter(|d| d.output_type == OutputType::Table)
.count();
let streaming = pipeline
.flows
.iter()
.filter(|f| f.kind.is_unbounded())
.count();
let expectations: usize = pipeline.flows.iter().map(|f| f.expectations.len()).sum();
let transformed = pipeline
.datasets
.iter()
.filter(|d| d.partition_cols.iter().any(|c| c.contains('(')))
.count();
println!("\n materialized views {mvs}");
println!(" streaming tables {tables}");
println!(" streaming flows {streaming}");
println!(" expectations {expectations} (Spark SDP: rejects this syntax)");
println!(" transform-partitioned {transformed} (Spark SDP: identity only)");
let fan_in = pipeline
.flows
.iter()
.find(|f| f.target == "gold_all")
.map(|f| f.reads.len())
.unwrap_or(0);
println!(" widest fan-in {fan_in} reads on `gold_all`");
println!("\n pipeline.is_streaming() = {}", pipeline.is_streaming());
// --- the capability gate --------------------------------------------
// The IR carries `months(event_ts)` happily; no shipped backend can run it.
// `check` says so locally, before a graph is created or a file is written,
// instead of letting it fail as an opaque JVM error mid-run.
{
use knut_thund::ExecBackend;
use knut_thund::backend::{native::NativeBackend, spark::SparkBackend};
println!("\n ExecBackend::check on this pipeline:");
for (label, res) in [
(
"spark-connect-sdp",
SparkBackend::new("sc://localhost:15002").check(&pipeline),
),
("native", NativeBackend::new().check(&pipeline)),
] {
match res {
Ok(()) => println!(" {label:<20} accepted"),
Err(e) => println!(" {label:<20} refused: {e}"),
}
}
}
}