nordisk/knut

Running a pipeline end-to-end — design

Class: design · Function axis: f0 one run, three frontends · f1 define→run→progress → the CLI/GUI/web viewers · f2 the sc:// session seam · f3 the Airflow AIP-82 asset sink · f4 the native CDC / SCD merge operator · f5 the skade-sourced changelog cutover. Strict: this file states only the shape we run. The crate-local why lives in pipelines-reasoning.md and thund-reasoning.md.

Scope: the cross-crate story added on autonom/knut-impl — how a Spark Declarative Pipeline is defined, run, and watched (terminal / egui / browser), how a session authenticates over sc://, how completions become Airflow asset events, and how a change-data-capture merge is applied natively over the single-sourced _change_type vocabulary. Builds on pipelines-design.md, thund-design.md and bifrost-design.md.

f0 — one run, three frontends (the whole)

A pipeline run is one merged stream folded into one ViewerState (progress.rs:240), and every frontend is a pure function of that state — the CLI text/JSON dump, the egui/snarl node graph, and now a browser page over Server-Sent Events. Nothing about the run changes when you change how you watch it; the web viewer can show nothing the CLI can't.

define_graph ─► start_run ─► PipelineRunEvent + ExecutionProgress
     (client.rs:354)  (client.rs:373)          │
                                                ▼
                          ViewerState::apply  (progress.rs:240)   ◄── one truth
                             │            │             │
                   dump_state│      DagViewer│     WebEventSink│
                   (viewer.rs)      (gui.rs)      (web.rs:56)
                    terminal         egui/snarl       browser / SSE

Two laws from pipelines-design.md still hold — CLI parity (show/dump_json is the floor) and one merged stream — and the web viewer is added under them, never beside them: it folds events through the same PipelineEventSink seam (web.rs:96) into the same ViewerState.

f1 — the web viewer (src/web.rs [feature web])

A third frontend: an axum server + SSE, serving facett's Rust facett-pipeline viewer compiled to wasm (the facet-wrapped bundle) and feeding it PipelineModel frames. knut draws nothing in JavaScript: the page is facet-wrapped's own loader, served as bytes from a run-time directory (FacetBundle; --facet-dist / FACET_WRAPPED_DIST), and every frame is to_pipeline_model encoded by the ONE codec, facett_pipeline::wire::to_json (asserted by index_serves_the_facet_wrapped_loader_not_a_drawing and events_stream_the_pipeline_model_wire_to_completion). Before lane SW row 6 (2026-09-05) this module hand-drew the DAG in SVG/JS; that is deleted, not flagged.

The event source is pluggable, the sink is shared. run_source (web.rs:195) drives one connection through a WebEventSink (web.rs:56) — a PipelineEventSink that folds each event into a ViewerState and broadcasts the resulting state_json over an mpsc channel. Two sources feed it:

• Synthetic replay (connect = None): synthetic_run (web.rs:268) walks the DAG in topological order emitting one STARTING + one COMPLETED event per flow, so the demo page animates with no Spark server. delay paces it (DEFAULT_REPLAY_DELAY = 450 ms, web.rs:128; Duration::ZERO in tests).
• Live run (connect = Some(url), feature grpc): drive_live (web.rs:220) opens a PipelinesClient and folds run_updates through drive_updates (web.rs:248) — the pure translation core, unit-tested offline with a canned update stream (drive_live_translates_event_stream_to_snapshots).

The page (web.rs:292): fetch('graph.json') → layout() does longest-path layering over read → target edges and draws one g.node per dataset; an EventSource('events') recolours each node by node_status (st-idle / st-running / st-done / st-failed) and updates #meta / #log. A {"done":true} frame closes the stream.

The serve_on bound-listener seam

serve (web.rs:131) binds an address then delegates to serve_on (web.rs:152), which takes an already-bound TcpListener. That seam lets a caller bind("127.0.0.1:0"), recover the OS-assigned port via local_addr(), and then hand the live listener over — no bind race. The browser robot test uses exactly this to point fantoccini at the real server (robot_web_viewer.rs:63).

The browser robot test (tests/robot_web_viewer.rs [feature web, #[ignore]])

The browser analogue of robot_snarl_gui.rs: serve the real page on an ephemeral port with a fast synthetic replay (Duration::from_millis(50)), drive a real browser through fantoccini (WebDriver), and gate on a red/green Row matrix. dag_animates_to_all_done (robot_web_viewer.rs:53) asserts the page drew one #nodes g.node per dataset, every node reaches class st-done, #meta shows "N done", and #log names the terminal rollup flow — ≥ 8 steps, zero reds. It is #[ignore] because it needs a WebDriver at 127.0.0.1:4444 (chromedriver / geckodriver), so the default build never links it and CI runs it only when a driver is present.

f2 — the sc:// session seam (src/client.rs [feature grpc])

connect (client.rs:206) now threads the Spark Connect URL's session params instead of dropping them. The sc:// grammar is sc://host:port/;user_id=<u>;session_id=<s>;token=<t>;use_ssl=<b>;…; parse_sc_url (client.rs:552) splits host:port from the ;k=v tail into an ScConnect (client.rs:539):

AuthInterceptor (client.rs:592) implements tonic::service::Interceptor; a None token is a pure pass-through so unauthenticated servers are unaffected (auth_interceptor_sets_bearer_only_when_token_present). A plain http(s):// URL carries no params — generated session, no auth. The client's inner type is now an InterceptedService<Channel, AuthInterceptor>.

The execute_sql progress finding

execute_sql (client.rs:491) streams ExecutionProgress for any Spark SQL query, making the viewer a live Spark job monitor. The correction pinned on this branch: progress is server-driven, not requested. Spark Connect pushes ExecutePlanResponse.execution_progress (base.proto field 18) unconditionally while a plan runs, paced by the server-side spark.connect.progress.reportInterval config — there is no per-request toggle. ExecutePlanRequest.request_options (field 5) is documented "currently unused" and only carries ReattachOptions / ResultChunkingOptions / an extension Any, none of which control progress. So the request leaves request_options at its default (empty) and simply consumes the frames the server volunteers. The live integration test knutpipe_live_path_streams_spark_job_progress (tests/integration_spark_connect.rs:83) asserts progress snapshots actually arrive.

f3 — the Airflow AIP-82 asset sink (src/airflow.rs [feature airflow])

AirflowAssetSink (airflow.rs:39) is a PipelineEventSink that turns each SDP output completing into an Airflow asset event, so a downstream Airflow 3 DAG can schedule on a pipeline output instead of a cron. It reuses bifröst's model — knut_bifrost::airflow::{AssetEvent, AssetEmitter} — and this crate's ViewerState for attribution + status + row parsing; it duplicates neither.

• Collection is synchronous (fits the sync sink seam): on_event (airflow.rs:113) folds the event into a ViewerState, then scan_completed (airflow.rs:87) appends one AssetEvent { uri, extra } the first time each dataset reaches NodeStatus::Done — URI {uri_prefix}{dataset}, extra carrying dataset / rows / output_type / source: "spark-sdp". Each output emits at most once (emitted: BTreeSet); a failed output advances no state and emits nothing (its asset is not current).
• Emission is a separate async step: emit (airflow.rs:75) forwards the batch through any AssetEmitter (e.g. bifröst's HTTP AirflowEmitter) — the same "derive events / POST them" split as bifröst's own emit_round. The AIP-82 wire body is bifröst's asset_event_body.

f4 — the native CDC / SCD merge operator (knut-thund/src/backend/native.rs [feature native])

Previously a FlowKind::Cdc flow was reported honestly as deferred. This branch implements it natively over DataFusion. Dispatch is at native.rs:476: a CDC flow routes to run_cdc_flow (native.rs:710), which resolves the changelog from an upstream reads source and folds it into the target as an idempotent, key-matched merge ordered by cdc.sequence_by.

The IR (knut-thund/src/ir/flow.rs): CdcSpec { keys, sequence_by, apply_as_deletes, scd_type } (flow.rs:30); ScdType::{Type1, Type2} (flow.rs:18); FlowKind::Cdc { cdc } (flow.rs:66).

The changelog contract — apply_cdc_merge (native.rs:846) — consumes skade / knut-bifrost's shared _change_type vocabulary (CHANGE_TYPE_COLUMN, native.rs:699):

The merge is a pure SQL rewrite over a registered changelog MemTable, with identifiers quoted via quote_ident (native.rs:828):

• SCD type 1 — ROW_NUMBER() OVER (PARTITION BY keys ORDER BY sequence_by DESC), keep rank 1 that is not a delete: the target is the latest non-deleted row per key (history overwritten). The _change_type column is stripped from the output so the target renders like an ordinary table.
• SCD type 2 — LEAD(sequence_by) OVER (PARTITION BY keys ORDER BY sequence_by) derives __start_at / __end_at validity bounds (__end_at = NULL for the current version); every version is kept, and a delete closes the open version without opening a new one.

Micro-batch vs bounded. When the changelog source was seeded as a micro-batch stream, run_cdc_flow accumulates the changelog seen so far and re-merges each trigger — apply-changes is a pure function of the ordered changelog, so recomputation is correct and deterministic. With no seeded stream it runs once over the bounded changelog. Backend capability: only NativeBackend runs CDC honestly; SparkBackend gates it behind spark_4_2() (see thund-design.md).

Tests (knut-thund/tests/streaming_microbatch.rs): cdc_scd_type1_upserts_and_deletes (:375), cdc_scd_type2_keeps_version_history (:405).

f5 — the skade-sourced changelog cutover (knut-bifrost/src/source.rs [feature skade])

Before this branch, knut-bifrost's IcebergSource kept a private DeltaPlan struct and re-derived the CDC changelog itself (its own INSERT/DELETE tagging, delete-identity widening, full-reload fallback). The cutover makes skade the single source of truth:

• The private DeltaPlan is deleted; bifröst now re-exports the canonical skade::DeltaPlan (source.rs:355).
• ingest_delta (source.rs:443) calls skade::Table::read_changelog(from, to) (source.rs:473) — skade walks the manifest lineage (read_delta), tags appended rows INSERT and equality-deleted identities DELETE (widening them to the table schema), and re-reads the whole to snapshot as an INSERT stream on an overwrite/position-delete window. The _change_type sentinel is skade::change_type::COLUMN.
• bifröst's contract is preserved by post-processing skade's ChangelogBatch (source.rs:370): a pure-append or full-reload window flows through untagged (strip_change_type, source.rs:545) straight to the upsert pipeline; only a window that carries equality deletes keeps skade's INSERT/DELETE tags for the listener's changelog split. skade clears delete_files on a full reload, so plan.delete_files.is_empty() is exactly "no CDC deletes to route" — the same branch bifröst took before the cutover.

The upshot: the _change_type changelog vocabulary is now defined once in skade and consumed identically by knut-bifrost (source side) and knut-thund (merge side, f4) — no divergent private copies.

Also on this branch

• graphar-flight nested-Cypher Utf8 fallback (graphar-flight/src/convert.rs): a non-scalar Cypher Value (map / node / list) has no scalar Arrow type, so cell_type falls back to Utf8 and the cell is rendered via its Debug form — stringified, never dropped. Pinned by auto_nested_cypher_value_falls_back_to_utf8 and cell_type_maps_nested_values_to_utf8.
• xtask build-wasm-guest (xtask/src/main.rs): a pure-Rust task that builds the wasm32-unknown-unknown guest under crates/knut-bifrost/wasm-guest/ (excluded from the native workspace), leaving the artifact where the #[ignore]d knut-bifrost tests/wasm_guest.rs expects it. Best-effort rustup target add, then a --target wasm32-unknown-unknown --release build, then a non-empty-artifact check. Closes TEST-GAP-2.

Feature-flag matrix

Dev-only: fantoccini (browser robot, web), async-trait + a minimal tokio (airflow-sink mock-emitter test). All default builds are unchanged — every surface above is off by default.

Architecture map