# knut-pipelines — design
> **Class:** `design` · **Function axis:** `f0` one-model-many-frontends · `f1` DAG model · `f2` run state + job progress · `f3` Spark Connect connector · `f4` renderers · `f5` GUI · `f6` introspection · `f7` facett adapter.
> Strict: this file states *only the shape we run*. Every *why*, every alternative, lives in [`pipelines-reasoning.md`](pipelines-reasoning.md).
>
> **Reuse:** `pipelines.uses = []` — standalone (default build is pure model + CLI). The facett adapter (`f7`) is an *optional* outbound edge to [facett-pipeline](../../facett/.nornir/facett-pipeline.md).
## f0 — one model, many frontends (the whole)
knut-pipelines is a viewer for [Spark Declarative Pipelines](https://issues.apache.org/jira/browse/SPARK-51727)
(SDP, Spark 4.x) **and** plain Spark Connect jobs. It ships the `knutpipe`
binary. The whole design is one invariant:
**`ViewerState` is the single source of truth; the CLI text/JSON view, the egui
GUI, and the headless run API are all pure functions of it.** A Spark Connect
server streams three things — the compiled dataflow DAG, pipeline run events, and
per-stage job progress — and `PipelinesClient::run_updates` merges them into one
`ViewerUpdate` stream that `ViewerState::apply` folds into that one state
(`progress.rs:240`). Every frontend renders the same state; none holds state of
its own.
Two laws hold the loop:
| law | statement | enforced by |
|---|---|---|
| **CLI parity** | every capability the GUI shows is reachable from the command line | `viewer::dump_text`/`dump_json`/`dump_state` (`viewer.rs:28`,`:18`,`:88`) mirror the GUI |
| **one merged stream** | pipeline events *and* job progress fold through one `apply` | `run_updates → UpdateStream` (`client.rs:449`), `ViewerState::apply` (`progress.rs:240`) |
**Heavy stacks are opt-in.** The default build is pure model + CLI `show`.
`grpc` (tonic 0.14 + vendored Spark Connect proto) and `gui` (egui/winit/glow)
are cargo features; `live = gui + grpc`. egui/eframe are pinned to **0.28** for
`egui-snarl` 0.4 compatibility.
## f1 — the DAG model (`src/model.rs`)
A pipeline is a **dataflow graph**: datasets (nodes) + flows (edges).
| item | shape | where |
|---|---|---|
| `DataflowGraph` | `{ datasets, flows, default_catalog, default_database }`; builder `new`/`with_dataset`/`with_flow` | `model.rs:186`,`:204` |
| `Dataset` | a node: `name` + `OutputType` (`MaterializedView`/`StreamingTable`/`View`/`Sink`) | `model.rs:88`,`:36` |
| `Flow` | an edge: `name`, `target`, `reads: Vec<String>`, optional `query` (`with_query`) — a flow with a `query` is *runnable* | `model.rs:133`,`:174` |
| `validate` / `topo_order` | acyclicity + a topological layout order for the **viewer** (not for execution — Spark schedules) | `model.rs:235`,`:272` |
| `PipelineRun` | `{ graph_id, full_refresh_all, storage, … }` — the run request | `model.rs:317` |
| `PipelineRunEvent` / `EventLevel` | one lifecycle event; `Info`/`Warn`/`Error` | `model.rs:359`,`:345` |
## f2 — run state + job progress (`src/progress.rs`)
`ViewerState` (`progress.rs:188`) is the folded truth: the `DataflowGraph`, a
per-dataset `NodeStatus`, the event log, and the latest `ExecutionProgress`.
| item | shape | where |
|---|---|---|
| `NodeStatus` | `Idle`/`Running`/`Done`/`Failed` per dataset (`label`) | `progress.rs:126`,`:140` |
| `ExecutionProgress` | a job snapshot: `StageProgress[]`, `fraction`, `total_tasks`/`completed_tasks`, `bytes_read`, `all_done`, `summary` | `progress.rs:54`,`:78` |
| `StageProgress` | one stage's task counts + `fraction` | `progress.rs:25`,`:41` |
| `ViewerUpdate` | the merge alphabet folded by `apply` (event \| progress \| …) | `progress.rs:397` |
| `ViewerState::apply` | fold one update; returns `true` on real change | `progress.rs:240` |
| `status_counts` / `topo_order` / `status_of` | derived views over the state | `progress.rs:313`,`:320`,`:230` |
## f3 — the Spark Connect connector (`src/client.rs` [feature `grpc`])
A thin hand-shaped driver over the **Spark Connect** gRPC service — not JDBC, not
the REST API, not Arrow Flight SQL. The one RPC that matters is server-streaming
`SparkConnectService.ExecutePlan`. See `assets/spark_connect_flow.svg`.
**Everything is an `ExecutePlan`.** There is no bespoke pipelines RPC: every
pipeline operation is a `Command` in a `Plan` on an `ExecutePlanRequest`, every
answer arrives on the returned server stream (`pipeline_request`, `client.rs:234`).
| direction | proto field | knut |
|---|---|---|
| request: pipelines command | `Command.pipeline_command` (commands.proto field 19) | `pipeline_request` wraps a `PipelineCommand` |
| response: command result | `ExecutePlanResponse.pipeline_command_result` (field 22) | graph id read back (`client.rs:278`) |
| response: pipeline event | `ExecutePlanResponse.pipeline_event_result` (field 21) | `event_from_proto → PipelineRunEvent` (`client.rs:74`) |
| response: job progress | `ExecutePlanResponse.execution_progress` | `progress_from_proto → ExecutionProgress` (`client.rs:102`) |
**The four-command dance.** `define_graph` (`client.rs:348`) drives, in order:
`CreateDataflowGraph` → returns a `dataflow_graph_id` (`create_dataflow_graph`,
`client.rs:254`); `DefineOutput` once per `Dataset` (`define_output`,
`client.rs:295`); `DefineFlow` once per `Flow` — a flow with a `query` becomes a
runnable `WriteRelationFlowDetails` (`define_flow`, `client.rs:320`); then
`start_run` against the graph id opens the event stream (`client.rs:367`).
**One merged stream.** `run_updates` (`client.rs:449`) returns an `UpdateStream`
(`type UpdateStream = Pin<Box<dyn Stream<Item = Result<ViewerUpdate>> + Send>>`,
`client.rs:156`); `response_to_update` (`client.rs:122`) keeps only events +
progress and drops arrow batches / sql results. The progress channel is
orthogonal: `execute_sql` (`client.rs:475`) streams `ExecutionProgress` for *any*
Spark SQL query (a `Plan` with a root relation, no `Command`), so the viewer is a
live Spark **job** monitor as well as a pipeline-DAG viewer.
**URL + session.** `connect` (`client.rs:205`) accepts `sc://host:port[/;params]`
(translated by `sc_url_to_http`, `client.rs:519`) or plain `http(s)://`, opens a
tonic `Channel`, and mints a per-connection `session_id` (`new_uuid`,
`client.rs:539`) that rides every request.
**Feature isolation.** The generated bindings (`proto::`, `client.rs:32`) carry
their own **tonic 0.14** tree via the crate `build.rs` (tonic-prost-build, gated),
so it never collides with the workspace's tonic 0.13 — the same trick
`knut-bifrost` uses for `spark-connect`.
### f3b — the Spark runtime + Structured-Streaming VIEW (`src/spark_view.rs`)
The **backend API korp's Spark pane calls to *see* a live runtime** — its version,
the Structured Streaming queries running on it, each query's status + recent
progress, and (the one control verb Connect exposes) to STOP one. Same
`ExecutePlan` transport; different `Command` variants:
| verb | proto command | folds to |
|---|---|---|
| `runtime_info` | `AnalyzePlan(SparkVersion)` (a typed `version` string — no arrow decode) | `SparkRuntimeInfo { version, connect_url }` |
| `list_active_streams` | `StreamingQueryManagerCommand.active` → `ActiveResult` | `Vec<StreamingQueryInfo { id, run_id, name }>` |
| `stream_status` | `StreamingQueryCommand.status` → `StatusResult` | `StreamingQueryStatus { message, is_active, is_data_available, is_trigger_active }` |
| `stream_recent_progress` | `StreamingQueryCommand.recent_progress` → `RecentProgressResult` | `Vec<String>` (the per-batch progress JSON, left undecoded to stay arrow-free) |
| `stop_stream` (control) | `StreamingQueryCommand.stop` | drains the confirmation |
The **model types** (`StreamingQueryInfo` / `StreamingQueryStatus` /
`SparkRuntimeInfo`, `spark_view.rs`) are **neutral and always compiled** — korp
depends on them without pulling gRPC in; the live methods that fill them live on
`PipelinesClient` behind `grpc`. **Honest scope:** cluster executor/worker
topology is *not* on the Connect wire (it lives in Spark's separate REST UI), so
this view deliberately covers only what Connect can answer and does not invent
fields the protocol cannot fill. The live methods need a reachable Spark Connect
server (no way to fake a cluster); the neutral fold is unit-tested
(`spark_view::tests`) and the gRPC wiring is compile-verified against the vendored
proto.
## f4 — renderers (`src/viewer.rs`)
Pure functions of the model/state; the CLI-parity surface.
| fn | renders | where |
|---|---|---|
| `dump_json(&DataflowGraph)` | the graph as JSON (the spec format) | `viewer.rs:18` |
| `dump_text(&DataflowGraph)` | the static text view | `viewer.rs:28` |
| `dump_state(&ViewerState, tail)` | the live folded state (status + progress + last `tail` events) | `viewer.rs:88` |
| `PipelineEventSink` / `StdoutEventSink` | the run-event sink `start_run` writes into; `format_line` | `viewer.rs:171`,`:188`,`:222` |
## f5 — the egui / snarl GUI (`src/gui.rs` [feature `gui`])
`launch(state, connect)` (`gui.rs:493`) / `launch_live_pipeline` (`gui.rs:517`)
run an eframe app; `DagViewer` impls `egui_snarl::SnarlViewer<NodeData>`
(`gui.rs:59`) to draw each dataset as a node coloured by `NodeStatus`, with a
Spark job-progress panel and the event log. The background stream feeds through
`spawn_live` (`gui.rs:688`) / `spawn_pipeline` (`gui.rs:620`), which consume the
same `UpdateStream` as the CLI. `KNUT_SCREENSHOT=<path>` captures one frame and
exits (`gui.rs:496`) — the headless GUI smoke path.
## f6 — introspection (headless robot-testing)
The viewer's entire UI state is observable from outside — no display, no
screenshot. Two always-on channels (mirrors nornir viz's `urdr-threads`):
1. **State dump.** `ViewerState::state_json()` (`progress.rs:335`) — `datasets`,
`node_status`, `status_counts`, `job_progress` (fraction/tasks/stages),
`events`+`last_event`, `topo_order`, `connect_status`/`source`/`error`, and
`placeholders` (empty-state strings). The GUI writes it every ~300 ms to
`$KNUT_VIEWER_STATE` (default `<tmp>/knut_viewer_state.json`, `gui.rs:243`),
overwritten in place. It lives in the **default build**, so the data-layer
matrix asserts on it with no `eframe::run` (`progress.rs` tests `state_json_*`).
2. **Activity trail.** one stderr line per real state change.
**Rule:** any new visible list/status/panel MUST be added to `state_json()`.
## f7 — the facett adapter (`src/facett_adapter.rs` [feature `facett-adapter`])
knut owns the *data + connector* side; the reusable **visual** component lives in
facett as an engine-agnostic viewer. The seam is one feature-gated file
(`facett-adapter = ["dep:facett-pipeline"]`) — the **Spark/SDP adapter**, the only
place that knows both knut's Spark types and facett's neutral model. It maps
knut's `ViewerState` onto facett's `PipelineModel`. "Show a Spark pipeline" is
exactly one adapter; a future Flink / Beam / skade-streaming viewer is *another
adapter*, not another viewer. See `assets/facett_adapter_flow.svg`.
**The mapping (precise):**
| knut input | facett output | how |
|---|---|---|
| each `Dataset` | a `Stage` | `Stage::new(name, status)`; name = both id and display name |
| `NodeStatus` | neutral `StageStatus` (`Idle`/`Running`/`Done`/`Failed`) | `map_status` — 1:1, no Spark vocabulary leaks |
| `Dataset.output_type` | stage `kind` sublabel | `with_kind(output_type.label())` |
| a failed node's matching error event | stage `error` | `error_for`: last `EventLevel::Error` whose `element == dataset`; only when `Failed` |
| each flow `(read → target)` | a `PipelineEdge` | one `PipelineEdge::new(read, target)` per read |
| Spark `ExecutionProgress` | `PipelineMetrics` | `fraction`/`tasks_done`/`tasks_total`/`inflight`/`bytes_read` straight through; `rows` = **real** sum of `ViewerState::rows_written` (parsed "wrote N rows" counts); `elapsed_s` = **measured** span between first/last timestamped event. Both `0` only when the run reported nothing — no `tasks_done × 1000` / `5.0` stand-in |
| `ViewerState::rows_written[name]` | per-stage `StageMetrics.rows` | each `Stage` carries its own real row count (`with_rows`) when the run attributed one to that node; `None` otherwise. Throughput/lag stay `None` (no per-node timer on the Spark stream) |
**Three entry points, one impl** (`to_pipeline_model`, `lib.rs:98`): the free
`to_pipeline_model(&ViewerState)`; `SparkPipelineAdapter<'a>(&'a ViewerState)`
impls `PipelineAdapter`; and `ViewerState` itself impls `PipelineAdapter`, so the
ergonomic form is `PipelineView::from_adapter(&viewer_state)`. A test pins that
all three agree.
**Back-compat law.** The default build is UNCHANGED. facett is pulled in *only*
under `facett-adapter` (`optional = true`), its model types **never leak into
knut's public API** (adapter takes knut types in, hands a `PipelineModel` out),
and facett's egui-0.34 tree stays off the default and `gui` (egui-0.28) builds.
## Architecture map
| Area | Where |
|---|---|
| DAG model | `src/model.rs` (`DataflowGraph`, `Dataset`, `Flow`, `OutputType`, `validate`, `topo_order`) |
| Run state + job progress | `src/progress.rs` (`ViewerState`, `ExecutionProgress`, `StageProgress`, `NodeStatus`, `ViewerUpdate`) |
| Renderers | `src/viewer.rs` (`dump_text`, `dump_json`, `dump_state`, `PipelineEventSink`, `StdoutEventSink`) |
| Spark Connect client | `src/client.rs` [`grpc`] (`PipelinesClient`, `run_updates`, `execute_sql`, `define_graph`, `UpdateStream`) |
| egui / snarl GUI | `src/gui.rs` [`gui`] (`launch`, `DagViewer: SnarlViewer`) |
| facett adapter | `src/facett_adapter.rs` [`facett-adapter`] (`SparkPipelineAdapter`, `to_pipeline_model`) |
| Binary | `src/bin/knutpipe.rs` (`show` / `watch` / `gui` / `pipeline`) |
| Vendored proto | `proto/spark/connect/*.proto` + `build.rs` (tonic-prost-build, gated) |