knut-thund 0.2.0

//! # 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"),
        );
    }
}