crates/knut-pipelines/src/lib.rs
//! # knut-pipelines
//!
//! A viewer for **Spark Declarative Pipelines** (SDP) — the `spark.pipelines`
//! feature added in Spark 4.x ([SPARK-51727]). It connects to a Spark Connect
//! server, fetches the compiled **dataflow DAG** (datasets = materialized
//! views / streaming tables / views / sinks; flows = the edges between them)
//! and the **run events**, and renders them — aiming toward a usable *live
//! stream* viewer where events arrive as a run progresses.
//!
//! ## Layout
//!
//! * [`model`] — the dependency-light DAG + event types ([`DataflowGraph`],
//! [`Dataset`], [`Flow`], [`PipelineRun`], [`PipelineRunEvent`]). These are
//! a faithful mirror of `spark/connect/pipelines.proto` but independent of
//! any generated code, so the default build needs no `protoc`.
//! * [`progress`] — Spark **job** progress ([`ExecutionProgress`] /
//! [`StageProgress`], a mirror of Spark Connect's `ExecutionProgress`) and
//! [`ViewerState`], the single model the CLI and GUI both render from.
//! * [`viewer`] — a text/JSON DAG dumper, a full-[`ViewerState`] text renderer
//! ([`dump_state`]), and the [`PipelineEventSink`] trait with a stdout
//! implementation.
//! * [`client`] — *(feature `grpc`)* the Spark Connect gRPC client that
//! connects to `sc://host:port`, defines/fetches a dataflow graph, starts a
//! run, and streams pipeline events **and** Spark job progress (also for
//! plain `execute_sql` queries).
//! * [`gui`] — *(feature `gui`)* an egui + `egui-snarl` desktop viewer: the
//! DAG as a live node graph, job-progress bars, and an event log.
//!
//! The `knutpipe` binary is the front door: `show` (static, default), `watch`
//! (live, `grpc`), `gui` (`gui`), `web` (axum + SSE, `web`).
//!
//! ## The `grpc` feature
//!
//! gRPC and proto codegen are **off by default**. Enabling `grpc` compiles
//! the vendored Spark Connect protos (`proto/`) with `tonic-build` + `protoc`
//! and pulls in a tonic 0.14 tree — isolated from the workspace's tonic 0.13
//! by the feature gate, exactly as the `spark-connect` crate's private tonic
//! tree is in `knut-bifrost`. The default build is pure model + viewer.
//!
//! ```
//! use knut_pipelines::{DataflowGraph, Dataset, Flow, OutputType, viewer};
//!
//! let graph = DataflowGraph::new()
//! .with_dataset(Dataset::new("bronze_users", OutputType::Table))
//! .with_dataset(Dataset::new("silver_users", OutputType::MaterializedView))
//! .with_flow(Flow::new("ingest", "bronze_users", Vec::<String>::new()))
//! .with_flow(Flow::new("clean", "silver_users", ["bronze_users"]));
//!
//! graph.validate().unwrap();
//! assert_eq!(
//! graph.topo_order().unwrap(),
//! vec!["bronze_users".to_string(), "silver_users".to_string()]
//! );
//! print!("{}", viewer::dump_text(&graph));
//! ```
//!
//! ## Roadmap
//!
//! See `docs/pipelines-viewer.md`. The bifröst Airflow AIP-82 ingestor
//! ([`knut_bifrost::airflow`]) is a downstream consumer of run events, now
//! wired through [`AirflowAssetSink`] (feature `airflow`): a
//! [`PipelineEventSink`] that folds a run and emits one asset event per
//! completed dataset via bifröst's `AssetEmitter`.
//!
//! [SPARK-51727]: https://issues.apache.org/jira/browse/SPARK-51727
//! [`knut_bifrost::airflow`]: https://docs.rs/knut-bifrost
/// The short git commit this crate was built from (`KNUT_GIT_HASH`, emitted by
/// `build.rs` via a pure-fs, zero-shell read of the repo's git metadata). `env!`
/// (not `option_env!`) because `build.rs` emits it on every build.
pub const GIT_HASH: &str = env!("KNUT_GIT_HASH");
/// `--version` line: `v<pkg-version> (<git-sha>)`. clap prints `<bin> <this>`,
/// so `--version` reads `<bin> v<ver> (<hash>)` — the suite-wide identity.
pub const VERSION_LINE: &str = concat!(
"v",
env!("CARGO_PKG_VERSION"),
" (",
env!("KNUT_GIT_HASH"),
")"
);
pub mod demo;
pub mod error;
pub mod model;
pub mod progress;
/// The Spark runtime + Structured-Streaming VIEW model (neutral, always
/// compiled). korp's Spark pane depends on these plain types; the live gRPC
/// methods that fill them live on [`PipelinesClient`] behind the `grpc` feature.
pub mod spark_view;
pub mod viewer;
/// The Spark/SDP → facett pipeline-viewer adapter (feature `facett-adapter`).
/// Maps knut's [`ViewerState`] into facett's engine-agnostic
/// `facett_pipeline::PipelineModel` so the one-viewer-many-adapters pipeline
/// viewer can render a Spark pipeline.
#[cfg(feature = "facett-adapter")]
pub mod facett_adapter;
#[cfg(feature = "grpc")]
pub mod client;
#[cfg(feature = "gui")]
pub mod gui;
/// Re-export of the exact `egui` the `gui` feature builds against (0.28, via
/// `eframe`). Lets a headless robot test in `tests/` name the same `egui` the
/// viewer uses — `egui::Context::run_ui` to drive real frames — without guessing
/// at an extern.
#[cfg(feature = "gui")]
pub use eframe::egui;
/// Airflow 3 (AIP-82) asset-event sink (feature `airflow`). A
/// [`PipelineEventSink`] that forwards each SDP output completing a run as an
/// Airflow asset event via `knut_bifrost::airflow::AssetEmitter`, so a
/// downstream Airflow DAG can schedule on a pipeline output.
#[cfg(feature = "airflow")]
pub mod airflow;
/// Web viewer (feature `web`, implies `facett-adapter`): an `axum` + SSE server
/// that serves facett's `facet-wrapped` wasm bundle — the Rust `facett-pipeline`
/// DAG viewer in the browser — and streams `PipelineModel` frames to it
/// (`to_pipeline_model` encoded by `facett_pipeline::wire`). knut draws nothing in
/// JavaScript. An additional frontend over the same model + [`PipelineEventSink`];
/// the CLI stays the parity floor. Demoable with no Spark server (synthetic replay).
#[cfg(feature = "web")]
pub mod web;
/// Spark Connect + SDP server bring-up (feature `server`). A pure-Rust,
/// zero-shell container manager ([`SparkServer`]) that pulls/creates/starts/
/// stops the Spark Connect container over podman's REST API — routed THROUGH
/// draupnir's ONE OCI engine (no second bollard engine) — and returns the `sc://…`
/// client URL. The `knut-spark-server` binary is a thin CLI over this; downstream
/// crates (korp's demo) use it to bring Spark up from Rust.
#[cfg(feature = "server")]
pub mod server;
pub use error::{PipelineError, Result};
pub use model::{
DataflowGraph, Dataset, EventLevel, Flow, OutputType, PipelineRun, PipelineRunEvent,
SourceLocation,
};
pub use progress::{
ExecutionProgress, NodeStatus, StageProgress, ViewerState, ViewerUpdate, human_bytes,
};
pub use spark_view::{SparkRuntimeInfo, StreamingQueryInfo, StreamingQueryStatus};
pub use viewer::{PipelineEventSink, StdoutEventSink, dump_json, dump_state, dump_text};
#[cfg(feature = "grpc")]
pub use client::PipelinesClient;
#[cfg(feature = "server")]
pub use server::{ServerStatus, SparkServer};
#[cfg(feature = "airflow")]
pub use airflow::AirflowAssetSink;
/// The facett handoff seam: [`SparkPipelineAdapter`] maps knut's `ViewerState`
/// onto facett's engine-agnostic `PipelineModel` (via [`to_pipeline_model`]).
/// This is an **external** seam — the sole in-repo consumer is the sanctioned
/// `tests/facett_handoff_contract.rs` (facett/wasm renders through it downstream),
/// so nothing in non-test knut source constructs a `PipelineModel` from these.
/// Keep the re-export: the contract test is what guards the mapping.
#[cfg(feature = "facett-adapter")]
pub use facett_adapter::{SparkPipelineAdapter, to_pipeline_model};
/// A small, self-contained demo pipeline — a bronze → silver → gold medallion
/// DAG — so `knutpipe show`/`gui`/`web` all do something useful with no
/// arguments and no Spark server, and the GUI robot test drives the *same*
/// built-in graph a user sees. Three datasets, three flows, acyclic.
pub fn demo_graph() -> DataflowGraph {
DataflowGraph::new()
.with_dataset(Dataset::new("bronze_events", OutputType::Table))
.with_dataset(Dataset::new("silver_events", OutputType::MaterializedView))
.with_dataset(Dataset::new("gold_daily", OutputType::MaterializedView))
.with_flow(Flow::new("ingest", "bronze_events", Vec::<String>::new()))
.with_flow(Flow::new("clean", "silver_events", ["bronze_events"]))
.with_flow(Flow::new("rollup", "gold_daily", ["silver_events"]))
}
/// **Shared test fixtures over the `ViewerState` model** (`testkit` feature) —
/// the synthetic progress snapshots knut's own contract test AND korp's pipeline
/// matrix inject. Owned here because they are struct literals over knut's PUBLIC
/// types: when `StageProgress`/`ExecutionProgress` gain a field, the type's owner
/// fixes the fixture once instead of every consumer repo discovering the break
/// on its own. Off by default; nothing rides in a release build.
#[cfg(any(test, feature = "testkit"))]
pub mod testkit;
#[cfg(test)]
mod tests;
/// **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). The CLI verbs and the event sink call this so `nornir test
/// --features testmatrix` SEES each surface's health.
#[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);
}
}