crates/knut-thund/src/lib.rs
//! # knut-thund (Þund) — a Rust-native, Arrow-centric streaming dataflow engine
//!
//! Þund is knut's **engine**: it authors, ingests, and runs **batch +
//! streaming** data pipelines, with a **pluggable execution backend** —
//! either a native Arrow/DataFusion engine (Arroyo-style, RecordBatch streams
//! end to end) or a lowering to **Spark Declarative Pipelines (SDP)** over
//! Spark Connect (Spark 4.1+/4.2). It is the "Airflow killer" half of the
//! pipelines story; `knut-pipelines` remains the *viewer*, which Þund reuses
//! (and which provides the Spark Connect client this engine lowers onto).
//!
//! ## The three layers
//!
//! | Layer | What | Module |
//! |---|---|---|
//! | **Authoring** | typed Rust builder + thin RON DSL + facett graph editor | [`authoring`] |
//! | **IR** | one Arrow-typed dataflow graph: datasets/flows/expectations, batch + streaming | [`ir`] |
//! | **Backends** | pluggable `ExecBackend`: native Arrow/DataFusion or Spark-Connect-SDP | [`backend`] |
//!
//! All three authoring surfaces produce the same [`ir::Pipeline`]; every
//! [`backend::ExecBackend`] consumes it. The IR is the contract; front-ends
//! and back-ends are interchangeable around it. See the design doc
//! `knut/.nornir/thund.md`.
//!
//! ## Why it kills Airflow
//!
//! Airflow's DAG encodes *task order* and is blind to the data it touches, so
//! it can only re-execute, never *reconcile*; data flows through the metadata
//! DB (XCom); the scheduler re-parses every DAG every loop; it is Python-only.
//! Þund takes Dagster's **asset model** (declare the artifact, infer the DAG
//! from declared inputs/outputs → recompute only what's stale), is **Arrow
//! end-to-end** like Dremio (one columnar format storage→exec→wire, Flight
//! transport, no XCom), runs **streaming first-class** (unlike any
//! orchestrator), and offers **CLI/Rust/DSL/visual parity** over one IR.
//!
//! ## Example
//!
//! ```
//! use knut_thund::ir::{Pipeline, Dataset, Flow, OutputType};
//!
//! let p = Pipeline::new("daily_sales")
//! .with_dataset(Dataset::new("clean_orders", OutputType::MaterializedView))
//! .with_flow(
//! Flow::batch("f_clean", "clean_orders", ["raw_orders"])
//! .with_query("SELECT * FROM raw_orders WHERE amount > 0"),
//! );
//! p.validate().unwrap();
//! assert_eq!(p.topo_order().unwrap(), vec!["clean_orders".to_string()]);
//!
//! // Lower to Spark Declarative Pipelines (works without a live Spark):
//! let sdp = p.to_sdp();
//! assert_eq!(sdp.datasets.len(), 1);
//! ```
#![deny(rustdoc::broken_intra_doc_links)]
pub mod authoring;
pub mod backend;
pub mod error;
pub mod ir;
#[cfg(feature = "rel2graph")]
pub mod rel2graph;
pub use backend::{
Capabilities, ExecBackend, RunEvent, RunHandle, RunPhase, SdpFlowDrop, SdpLoweringReport,
};
pub use error::{Result, ThundError};
pub use ir::Pipeline;
/// **Introspection / emit marker** — record one functional-status row for the
/// nornir test matrix. Wraps `nornir_testmatrix::functional_status` behind the
/// `testmatrix` feature (a compiled-out `#[inline]` no-op otherwise, with no
/// nornir dep).
///
/// Every surface calls this with honest health: the IR [`Pipeline::validate`]
/// gate, the IR→SDP lowering [`Pipeline::to_sdp`], and the RON DSL round-trip
/// ([`authoring::dsl`], feature `dsl`). The **execution** backends report their
/// real outcome too (see [`backend::native`] / [`backend::spark`]): with the
/// `native` feature the DataFusion backend's `execute` row is green when the
/// batch dataflow actually ran (red on a genuine execution/expectation
/// failure); with the `spark` feature the Spark-Connect backend's `live_run`
/// row is green when the run streamed to completion (red on a transport/run
/// error), and its IR→SDP `lower_to_sdp` sub-step, which always works, is
/// green. Built *without* those features, each backend's `run` records a red
/// row noting the feature (and hence the engine) is not compiled in.
#[inline]
pub fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(component, check, ok, detail);
#[cfg(not(feature = "testmatrix"))]
{
let _ = (component, check, ok, detail);
}
}
#[cfg(test)]
mod tests {
use super::ThundError;
use super::backend::{ExecBackend, native::NativeBackend, spark::SparkBackend};
use super::ir::*;
/// Build a small streaming pipeline: a Kafka source flow feeding a
/// streaming table, with a watermark and a data-quality expectation.
fn sample_streaming() -> Pipeline {
Pipeline::new("clicks")
.with_dataset(
Dataset::new("click_counts", OutputType::Table)
.incremental()
.with_schema(
DatasetSchema::new()
.field("window_end", "Timestamp(Microsecond, None)", false)
.field("n", "Int64", false),
),
)
.with_flow(
Flow::streaming(
"f_clicks",
"click_counts",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "clicks".into(),
format: "json".into(),
},
)
.with_watermark(Watermark {
event_time_column: "ts".into(),
allowed_lateness_ms: 5_000,
idle_timeout_ms: 30_000,
})
.with_query("SELECT window_end, count(*) n FROM clicks GROUP BY TUMBLE(ts, '10s')")
.expect(Expectation::new("n_positive", "n > 0").on(OnViolation::Drop)),
)
}
#[test]
fn validate_and_topo_order() {
let p = Pipeline::new("p")
.with_dataset(Dataset::new("a", OutputType::Table))
.with_dataset(Dataset::new("b", OutputType::MaterializedView))
.with_flow(Flow::batch("fa", "a", Vec::<String>::new()))
.with_flow(Flow::batch("fb", "b", ["a"]));
p.validate().expect("valid");
// a precedes b because fb reads a.
assert_eq!(
p.topo_order().unwrap(),
vec!["a".to_string(), "b".to_string()]
);
}
#[test]
fn dangling_flow_is_rejected() {
let p = Pipeline::new("p").with_flow(Flow::batch("f", "ghost", Vec::<String>::new()));
let err = p.validate().unwrap_err();
assert!(
matches!(err, ThundError::DanglingFlow { .. }),
"got {err:?}"
);
}
#[test]
fn cycle_is_rejected() {
let p = Pipeline::new("p")
.with_dataset(Dataset::new("a", OutputType::Table))
.with_dataset(Dataset::new("b", OutputType::Table))
.with_flow(Flow::batch("fa", "a", ["b"]))
.with_flow(Flow::batch("fb", "b", ["a"]));
assert!(matches!(p.validate().unwrap_err(), ThundError::Cyclic));
assert!(p.topo_order().is_none());
}
#[test]
fn streaming_pipeline_detected_and_lowers_to_sdp() {
let p = sample_streaming();
p.validate().expect("valid");
assert!(p.is_streaming(), "kafka flow makes it streaming");
// The graph SHAPE lowers faithfully to SDP even though SDP can't
// express the watermark/expectation knobs.
let sdp = p.to_sdp();
sdp.validate().expect("lowered SDP is valid");
assert_eq!(sdp.datasets.len(), 1);
assert_eq!(
sdp.datasets[0].output_type,
knut_pipelines::OutputType::Table
);
// Schema lowered to an SDP schema string.
assert_eq!(
sdp.datasets[0].schema.as_deref(),
Some("window_end Timestamp(Microsecond, None), n Int64")
);
assert_eq!(sdp.flows.len(), 1);
assert_eq!(
sdp.flows[0].query.as_deref(),
Some("SELECT window_end, count(*) n FROM clicks GROUP BY TUMBLE(ts, '10s')")
);
}
#[test]
fn spark_backend_rejects_expectations_via_capabilities() {
// SparkBackend reports expectations:false, so `check` must refuse a
// pipeline that carries one — BEFORE any network call. This is the
// SanityCheckPlan-style up-front rejection.
let p = sample_streaming(); // has an expectation
let spark = SparkBackend::new("sc://localhost:15002");
let err = spark.check(&p).unwrap_err();
assert!(
matches!(err, ThundError::Unsupported { ref what, .. } if what.contains("expectations")),
"got {err:?}"
);
}
#[test]
fn partition_transforms_are_refused_up_front_by_both_backends() {
use crate::ir::{Dataset, OutputType, Pipeline};
// Spark's SDP `PartitionHelper` accepts only IdentityTransform, and the
// native file sink maps `partition_cols` to literal Hive path segments.
// Neither can honour `months(event_ts)`, so BOTH must refuse it in
// `check` — locally, before a graph is created or a file is written.
let p = Pipeline::new("p").with_dataset(
Dataset::new("events", OutputType::MaterializedView)
.with_partition_cols(["months(event_ts)"]),
);
for err in [
SparkBackend::new("sc://localhost:15002")
.check(&p)
.unwrap_err(),
NativeBackend::new().check(&p).unwrap_err(),
] {
assert!(
matches!(err, ThundError::Unsupported { ref what, .. }
if what.contains("months(event_ts)")),
"got {err:?}"
);
}
// An identity column is fine on both.
let ok = Pipeline::new("p").with_dataset(
Dataset::new("events", OutputType::MaterializedView)
.with_partition_cols(["event_month"]),
);
SparkBackend::new("sc://localhost:15002")
.check(&ok)
.unwrap();
NativeBackend::new().check(&ok).unwrap();
}
#[test]
fn graph_sink_pipelines_are_refused_up_front_by_the_spark_backend() {
use crate::ir::{Dataset, OutputType, Pipeline};
// A rel2graph graph-sink dataset: a `falkordb-node` Sink carrying the
// `knut.node.label` target. SDP-over-Spark-Connect has no shape to carry
// the Cypher `UNWIND … MERGE` (an SDP flow is a SQL relation, an SDP
// output a table), so SparkBackend::check must refuse it LOUDLY — before
// any `DefineOutput` of an unresolvable `falkordb-node` format — and
// point the operator at NativeBackend or the PySpark projection.
let p = Pipeline::new("rel2graph").with_dataset(
Dataset::new("node__User", OutputType::Sink)
.with_format("falkordb-node")
.with_properties([("knut.node.label", "User")]),
);
let spark = SparkBackend::new("sc://localhost:15002");
assert!(!spark.capabilities().graph_sinks);
let err = spark.check(&p).unwrap_err();
assert!(
matches!(err, ThundError::Unsupported { backend, ref what }
if backend == "spark-connect-sdp"
&& what.contains("graph-sink")
&& what.contains("SparkBackend")
&& what.contains("NativeBackend")
&& what.contains("PySpark")),
"got {err:?}"
);
// The NativeBackend's answer tracks whether the executable graph-sink
// write is compiled in (`native + rel2graph`, i.e. the `falkordb`
// feature): with it, `check` accepts the graph load; without it, native
// honestly refuses a sink it has no code to run.
let native = NativeBackend::new();
if native.capabilities().graph_sinks {
native
.check(&p)
.expect("native (falkordb) accepts a graph-sink pipeline");
} else {
assert!(
matches!(
native.check(&p).unwrap_err(),
ThundError::Unsupported { .. }
),
"native without the falkordb feature refuses graph sinks"
);
}
// A plain Sink with an unrelated format is NOT a graph sink — the gate
// must not catch it (both conditions required, mirroring `graph_elem`).
let plain = Pipeline::new("p")
.with_dataset(Dataset::new("audit", OutputType::Sink).with_format("parquet"));
SparkBackend::new("sc://x:15002")
.check(&plain)
.expect("a non-graph Sink is not caught by the graph-sink gate");
}
#[test]
fn native_backend_accepts_everything_in_the_ir() {
// The native backend's capabilities cover the whole IR, so `check`
// passes the same pipeline Spark refused.
let p = sample_streaming();
let native = NativeBackend::new();
native
.check(&p)
.expect("native accepts streaming + expectations + watermark");
let caps = native.capabilities();
assert!(caps.streaming && caps.event_time && caps.cdc && caps.expectations);
}
#[test]
fn capabilities_differ_between_backends() {
let n = NativeBackend::new().capabilities();
let s = SparkBackend::new("sc://x:15002").capabilities();
assert!(
n.event_time && !s.event_time,
"native has event-time, spark does not"
);
assert_eq!(s.output_modes, vec!["append".to_string()]);
assert!(n.output_modes.contains(&"complete".to_string()));
}
#[cfg(feature = "dsl")]
#[test]
fn dsl_roundtrips_through_ir() {
let p = sample_streaming();
let ron = crate::authoring::dsl::to_ron(&p).expect("serialize");
let back = crate::authoring::dsl::from_ron(&ron).expect("parse");
assert_eq!(p, back, "RON DSL is a lossless view of the IR");
}
/// A **declarative file sink** — a `Dataset` that names its on-disk `path`
/// (and partitioning) in the IR itself — survives the RON round trip
/// losslessly, so a `.ron` spec can carry where its outputs land. RED-when-
/// broken: if `Dataset.path` were dropped on (de)serialize the equality
/// fails; the assert on the parsed field pins it explicitly.
#[cfg(feature = "dsl")]
#[test]
fn dsl_roundtrips_declarative_file_sink() {
let p = Pipeline::new("rollup_to_disk")
.with_dataset(
Dataset::new("by_region", OutputType::Sink)
.with_path("out/by_region")
.with_format("parquet")
.with_partition_cols(["region"]),
)
.with_flow(
Flow::batch("agg", "by_region", ["orders"])
.with_query("SELECT region, SUM(amount) AS total FROM orders GROUP BY region"),
);
let ron = crate::authoring::dsl::to_ron(&p).expect("serialize");
let back = crate::authoring::dsl::from_ron(&ron).expect("parse");
assert_eq!(
p, back,
"the declarative sink path/format/partitioning round-trips"
);
let d = back.dataset("by_region").expect("sink dataset present");
assert_eq!(
d.path.as_deref(),
Some("out/by_region"),
"the sink path survived"
);
assert_eq!(d.partition_cols, vec!["region".to_string()]);
}
/// A **typed projection/filter** flow — the introspectable pushdown node —
/// survives the RON round trip, so a `.ron` spec carries the structured
/// column list + predicate the engine prunes/pushes from (not an opaque SQL
/// string). RED-when-broken: if `Flow.projection`/`Flow.filter` were dropped
/// on (de)serialize the equality fails; the field asserts pin them.
#[cfg(feature = "dsl")]
#[test]
fn dsl_roundtrips_typed_projection_filter() {
let p = Pipeline::new("prune")
.with_dataset(Dataset::new("slim", OutputType::MaterializedView))
.with_flow(
Flow::batch("proj", "slim", ["orders"])
.with_projection(["customer", "amount"])
.with_filter("amount > 100"),
);
let ron = crate::authoring::dsl::to_ron(&p).expect("serialize");
let back = crate::authoring::dsl::from_ron(&ron).expect("parse");
assert_eq!(
p, back,
"the typed projection + filter round-trip losslessly"
);
let f = &back.flows[0];
assert_eq!(
f.projection,
vec!["customer".to_string(), "amount".to_string()]
);
assert_eq!(f.filter.as_deref(), Some("amount > 100"));
// And it lowers to the Spark SELECT (the pushdown node's Spark half).
assert_eq!(
f.to_sdp().query.as_deref(),
Some("SELECT customer, amount FROM orders WHERE amount > 100"),
);
}
/// A sink dataset's [`crate::ir::WriteMode`] survives the RON round trip, so a
/// `.ron` spec can declare `Overwrite` (clear-then-write) vs the default
/// `Append`. RED-when-broken: if `Dataset.write_mode` were dropped on
/// (de)serialize the equality fails and the mode assert catches it. Also pins
/// the L2-additive default: a dataset with no declared mode parses back as
/// `Append`.
#[cfg(feature = "dsl")]
#[test]
fn dsl_roundtrips_write_mode() {
use crate::ir::WriteMode;
let p = Pipeline::new("refresh")
.with_dataset(
Dataset::new("by_region", OutputType::Sink)
.with_path("out/by_region")
.with_partition_cols(["region"])
.overwrite(),
)
.with_dataset(Dataset::new("plain_sink", OutputType::Sink).with_path("out/plain_sink"))
.with_flow(
Flow::batch("agg", "by_region", ["orders"])
.with_query("SELECT region, SUM(amount) AS total FROM orders GROUP BY region"),
);
let ron = crate::authoring::dsl::to_ron(&p).expect("serialize");
let back = crate::authoring::dsl::from_ron(&ron).expect("parse");
assert_eq!(p, back, "the sink write_mode round-trips losslessly");
assert_eq!(
back.dataset("by_region").unwrap().write_mode,
WriteMode::Overwrite,
"the declared Overwrite mode survived",
);
// The dataset that declared no mode defaults to Append (additive default).
assert_eq!(
back.dataset("plain_sink").unwrap().write_mode,
WriteMode::Append,
"a sink with no declared write_mode defaults to Append (L2-additive)",
);
// And the default sink's mode is skipped on the wire (byte-identical form).
assert!(
!ron.contains("append"),
"the default Append mode is not serialized (skip_serializing_if) — wire stays additive",
);
}
/// **The exact korp surface: `from_ron(text).to_sdp()` → the `DataflowGraph`
/// korp renders.** The other DSL tests prove the RON round trip and the flows
/// lower one at a time; this one drives the *whole* korp path in one shot — a
/// multi-dataset, multi-flow medallion pipeline parsed from RON text and lowered
/// as a graph — and asserts the concrete SDP graph shape (dataset names +
/// output types, flow names + targets + SQL) korp's viewer consumes. RED-when-
/// broken: a dataset/flow dropped in `from_ron` or mis-lowered in `to_sdp` (a
/// renamed target, a lost query) fails a field assert here, not just an opaque
/// equality.
#[cfg(feature = "dsl")]
#[test]
fn from_ron_full_pipeline_lowers_to_the_dataflow_graph_korp_renders() {
// Author the medallion pipeline, render it to RON text, and parse it back —
// so the graph under test genuinely came through `from_ron`, not the builder.
let authored = Pipeline::new("medallion")
.with_dataset(Dataset::new("bronze", OutputType::Table))
.with_dataset(Dataset::new("silver", OutputType::MaterializedView))
.with_dataset(Dataset::new("gold", OutputType::MaterializedView))
.with_flow(
Flow::batch("refine", "silver", ["bronze"])
.with_query("SELECT * FROM bronze WHERE ok"),
)
.with_flow(
Flow::batch("rollup", "gold", ["silver"])
.with_query("SELECT region, SUM(amount) total FROM silver GROUP BY region"),
);
let text = crate::authoring::dsl::to_ron(&authored).expect("render RON");
let parsed = crate::authoring::dsl::from_ron(&text).expect("parse RON");
// The korp call: lower the parsed IR to the SDP DataflowGraph korp renders.
let g = parsed.to_sdp();
g.validate().expect("the lowered SDP graph is valid");
// Datasets: names + output types survived parse *and* lowering, in order.
assert_eq!(
g.datasets
.iter()
.map(|d| d.name.as_str())
.collect::<Vec<_>>(),
["bronze", "silver", "gold"],
);
assert_eq!(g.datasets[0].output_type, knut_pipelines::OutputType::Table);
assert_eq!(
g.datasets[1].output_type,
knut_pipelines::OutputType::MaterializedView
);
assert_eq!(
g.datasets[2].output_type,
knut_pipelines::OutputType::MaterializedView
);
// Flows: name → target → SQL all lowered faithfully.
assert_eq!(g.flows.len(), 2);
let refine = g
.flows
.iter()
.find(|f| f.name == "refine")
.expect("refine flow");
assert_eq!(refine.target, "silver");
assert_eq!(
refine.query.as_deref(),
Some("SELECT * FROM bronze WHERE ok")
);
let rollup = g
.flows
.iter()
.find(|f| f.name == "rollup")
.expect("rollup flow");
assert_eq!(rollup.target, "gold");
assert_eq!(
rollup.query.as_deref(),
Some("SELECT region, SUM(amount) total FROM silver GROUP BY region"),
);
}
}