crates/knut-pipelines/src/facett_adapter.rs
//! **The Spark/SDP pipeline adapter** — knut's `ViewerState` → facett's
//! engine-agnostic [`PipelineModel`](facett_pipeline::PipelineModel).
//!
//! This is knut's implementation of facett's [`PipelineAdapter`] seam: the one
//! place that knows *both* knut's Spark types and facett's neutral model. It maps
//! the compiled SDP dataflow graph + live run state into the normalized shape the
//! `facett-pipeline` viewer renders — so "show a Spark pipeline" is exactly *one
//! adapter*, and a future Flink/Beam/skade-streaming viewer is *another adapter*,
//! not another viewer (see `facett_pipeline::model` docs + the design note).
//!
//! **Feature-gated (`facett-adapter`).** It is the only spot knut depends on
//! facett, and facett's model types never leak into knut's own public API — the
//! adapter takes knut types in and hands a `PipelineModel` out.
//!
//! ## The mapping
//!
//! * each [`Dataset`](crate::Dataset) → a [`Stage`](facett_pipeline::Stage):
//! the dataset name is both id and display name, the run
//! [`NodeStatus`](crate::NodeStatus) maps to the neutral
//! [`StageStatus`](facett_pipeline::StageStatus), the output type becomes the
//! stage `kind` sublabel, and — for a failed node — the matching error event's
//! message is surfaced as the stage `error`.
//! * each flow's `(read → target)` pair → a
//! [`PipelineEdge`](facett_pipeline::PipelineEdge) (source → consumer).
//! * the Spark job [`ExecutionProgress`](crate::ExecutionProgress) →
//! [`PipelineMetrics`](facett_pipeline::PipelineMetrics) (fraction, task
//! counts, bytes, inflight are genuine passthroughs). `rows` is the **real**
//! sum of the per-dataset row counts knut parsed from the run's "wrote N
//! rows" events; `elapsed_s` is the **measured** wall-clock span between the
//! first and last timestamped event. Both are `0` only when the run reported
//! nothing to measure — never a fabricated stand-in.
//! * per-dataset [`StageMetrics`](facett_pipeline::StageMetrics): each stage
//! carries its own real `rows` (from `ViewerState::rows_written`) when the
//! run attributed a row count to that node; `None` otherwise. Throughput/lag
//! stay `None` — Spark's stream carries no per-node timer to derive them from.
use facett_pipeline::{
PipelineAdapter, PipelineEdge, PipelineMetrics, PipelineModel, Stage, StageStatus,
};
use crate::{EventLevel, NodeStatus, ViewerState};
/// The pipeline-wide row total: the sum of the per-dataset row counts knut
/// attributed from the run's "wrote N rows" events. `0` when the run reported
/// none — a real measurement, never a `tasks_done * 1000` stand-in.
fn total_rows(vs: &ViewerState) -> u64 {
vs.rows_written.values().copied().sum()
}
/// The measured wall-clock span of the run, in seconds: the gap between the
/// earliest and latest event carrying a parseable RFC3339 timestamp. Returns
/// `0.0` when fewer than two events are timestamped (nothing honest to measure)
/// — never a `5.0` constant. Timestamps come from Spark's `PipelineEvent`
/// (`client::event_from_proto`); a replayed/synthetic run without them reads 0.
fn run_elapsed_s(vs: &ViewerState) -> f32 {
let times: Vec<chrono::DateTime<chrono::FixedOffset>> = vs
.events
.iter()
.filter_map(|e| e.timestamp.as_deref())
.filter_map(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
.collect();
match (times.iter().min(), times.iter().max()) {
(Some(min), Some(max)) if times.len() >= 2 => {
(*max - *min).num_milliseconds() as f32 / 1000.0
}
_ => 0.0,
}
}
/// Map knut's [`NodeStatus`] onto facett's neutral [`StageStatus`].
fn map_status(s: NodeStatus) -> StageStatus {
match s {
NodeStatus::Idle => StageStatus::Idle,
NodeStatus::Running => StageStatus::Running,
NodeStatus::Done => StageStatus::Done,
NodeStatus::Failed => StageStatus::Failed,
}
}
/// The last error-level event message attributed to `dataset` — the text the
/// failed stage surfaces.
///
/// Attribution prefers the event's explicit `element`. Only if no error event
/// names the dataset that way do we fall back to a **whole-word** mention in
/// the message. A bare `message.contains(dataset)` substring test (the old
/// behaviour) is wrong for short dataset names: a stage called `l` or `r`
/// matches almost any error message (e.g. "AnalysisException", "column"),
/// surfacing an unrelated error on the stage.
fn error_for(vs: &ViewerState, dataset: &str) -> Option<String> {
// 1. Exact `element` attribution (most recent wins).
if let Some(e) = vs
.events
.iter()
.rev()
.find(|e| e.level == EventLevel::Error && e.element.as_deref() == Some(dataset))
{
return Some(e.message.clone());
}
// 2. Fall back to a whole-word mention in the message.
vs.events
.iter()
.rev()
.find(|e| e.level == EventLevel::Error && message_mentions_word(&e.message, dataset))
.map(|e| e.message.clone())
}
/// True if `message` contains `word` delimited by word boundaries (anything
/// that isn't alphanumeric or `_`), so `"in sink"` matches `sink` but `"column"`
/// does not match `l`. An empty `word` never matches.
fn message_mentions_word(message: &str, word: &str) -> bool {
if word.is_empty() {
return false;
}
let is_word_char = |c: char| c.is_alphanumeric() || c == '_';
message.match_indices(word).any(|(i, _)| {
let before_ok = message[..i]
.chars()
.next_back()
.is_none_or(|c| !is_word_char(c));
let after_ok = message[i + word.len()..]
.chars()
.next()
.is_none_or(|c| !is_word_char(c));
before_ok && after_ok
})
}
/// Build the normalized [`PipelineModel`] from a knut [`ViewerState`]. The free
/// function form — handy when you have a `&ViewerState` directly. Equivalent to
/// `SparkPipelineAdapter(vs).to_pipeline_model()`.
pub fn to_pipeline_model(vs: &ViewerState) -> PipelineModel {
let stages: Vec<Stage> = vs
.graph
.datasets
.iter()
.map(|d| {
let status = map_status(vs.status_of(&d.name));
let mut stage = Stage::new(d.name.clone(), status).with_kind(d.output_type.label());
// Real per-node rows: the count knut parsed from this dataset's
// "wrote N rows" event, if the run reported one. No event → no
// rows metric (None), never a fabricated stand-in.
if let Some(&rows) = vs.rows_written.get(&d.name) {
stage = stage.with_rows(rows);
}
if status == StageStatus::Failed {
if let Some(err) = error_for(vs, &d.name) {
stage = stage.with_error(err);
}
}
stage
})
.collect();
// Flows' (read → target) pairs become source→consumer edges, by stage id
// (the dataset name). Reads that aren't graph datasets (external sources) are
// left for the view to drop when resolving ids.
let mut edges = Vec::new();
for f in &vs.graph.flows {
for r in &f.reads {
edges.push(PipelineEdge::new(r.clone(), f.target.clone()));
}
}
let mut model = PipelineModel {
title: "pipeline".to_string(),
stages,
edges,
metrics: None,
};
if let Some(p) = &vs.progress {
let done = p.completed_tasks().max(0) as u64;
model = model.with_metrics(PipelineMetrics {
fraction: Some(p.fraction()),
tasks_done: done,
tasks_total: p.total_tasks().max(0) as u64,
inflight: p.num_inflight_tasks.max(0) as u64,
// REAL rows: the sum of the per-dataset counts knut parsed from the
// run's "wrote N rows" events (Spark's `ExecutionProgress` itself
// carries only task/byte counters). 0 means the run reported no row
// counts, never a `tasks_done * 1000` stand-in.
rows: total_rows(vs),
bytes_read: p.bytes_read().max(0) as u64,
// MEASURED wall-clock: the span between the first and last
// timestamped event of the run. 0.0 means fewer than two events
// were timestamped — never a `5.0` constant.
elapsed_s: run_elapsed_s(vs),
});
}
model
}
/// The Spark/SDP [`PipelineAdapter`]: a thin wrapper over a `&ViewerState` so the
/// viewer can take `&impl PipelineAdapter` uniformly (`PipelineView::from_adapter`).
pub struct SparkPipelineAdapter<'a>(pub &'a ViewerState);
impl PipelineAdapter for SparkPipelineAdapter<'_> {
fn to_pipeline_model(&self) -> PipelineModel {
to_pipeline_model(self.0)
}
}
/// `ViewerState` itself implements [`PipelineAdapter`] — the most ergonomic form:
/// `PipelineView::from_adapter(&viewer_state)`.
impl PipelineAdapter for ViewerState {
fn to_pipeline_model(&self) -> PipelineModel {
to_pipeline_model(self)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
DataflowGraph, Dataset, ExecutionProgress, Flow, OutputType, PipelineRunEvent,
StageProgress, ViewerUpdate,
};
use facett_pipeline::StageStatus;
fn diamond() -> DataflowGraph {
let ds = |n: &str| Dataset::new(n, OutputType::MaterializedView);
DataflowGraph::new()
.with_dataset(ds("src"))
.with_dataset(ds("l"))
.with_dataset(ds("r"))
.with_dataset(ds("sink"))
.with_flow(Flow::new("f_src", "src", Vec::<String>::new()))
.with_flow(Flow::new("f_l", "l", ["src"]))
.with_flow(Flow::new("f_r", "r", ["src"]))
.with_flow(Flow::new("f_sink", "sink", ["l", "r"]))
}
#[test]
fn maps_structure_and_kind() {
let vs = ViewerState::new(diamond());
let m = vs.to_pipeline_model();
assert_eq!(m.stages.len(), 4);
// src→l, src→r, l→sink, r→sink = 4 edges.
assert_eq!(m.edges.len(), 4);
// every stage carries the output-type kind sublabel.
assert!(
m.stages
.iter()
.all(|s| s.kind.as_deref() == Some("materialized_view"))
);
// all idle before any run.
assert_eq!(m.status_counts(), (4, 0, 0, 0));
// stage id == name so edges resolve.
assert_eq!(m.stage_index("sink"), Some(3));
}
#[test]
fn maps_run_status_and_failed_error() {
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "src materialized".into(),
element: Some("src".into()),
}));
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: "AnalysisException: column missing in sink".into(),
element: Some("sink".into()),
}));
let m = vs.to_pipeline_model();
let src = m.stages.iter().find(|s| s.id == "src").unwrap();
assert_eq!(src.status, StageStatus::Done);
let sink = m.stages.iter().find(|s| s.id == "sink").unwrap();
assert_eq!(sink.status, StageStatus::Failed);
assert_eq!(
sink.error.as_deref(),
Some("AnalysisException: column missing in sink"),
"the failed stage surfaces its error event message"
);
}
#[test]
fn short_dataset_name_is_not_substring_misattributed() {
// FAIL-ON-BUG: the only error event names `sink` (element + message),
// and its message contains the letters 'l'/'r' (in "column", "error").
// The old bare-substring fallback made `error_for(vs, "l")` surface the
// sink's error on the `l` stage. PASS-ON-FIX: only `sink` is attributed;
// `l` and `r` get nothing because they're not whole words in the text.
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: "AnalysisException: column error in sink".into(),
element: Some("sink".into()),
}));
assert_eq!(
error_for(&vs, "sink").as_deref(),
Some("AnalysisException: column error in sink"),
"the sink stage keeps its own error",
);
assert_eq!(
error_for(&vs, "l"),
None,
"a single-char stage must not steal the sink's error via substring",
);
assert_eq!(error_for(&vs, "r"), None);
}
#[test]
fn word_boundary_fallback_still_attributes_unflagged_mentions() {
// No `element` set, but the message names the stage as a whole word →
// the word-boundary fallback still attributes it.
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: "failed to materialize sink".into(),
element: None,
}));
assert_eq!(
error_for(&vs, "sink").as_deref(),
Some("failed to materialize sink"),
);
assert_eq!(error_for(&vs, "l"), None);
}
#[test]
fn maps_progress_to_pipeline_metrics() {
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![
StageProgress {
stage_id: 0,
num_tasks: 8,
num_completed_tasks: 8,
input_bytes_read: 2 * 1024 * 1024,
done: true,
},
StageProgress {
stage_id: 1,
num_tasks: 8,
num_completed_tasks: 2,
input_bytes_read: 1024 * 1024,
done: false,
},
],
num_inflight_tasks: 6,
}));
let m = vs.to_pipeline_model();
let pm = m.metrics.expect("progress maps to pipeline metrics");
assert_eq!(pm.tasks_total, 16);
assert_eq!(pm.tasks_done, 10);
assert_eq!(pm.inflight, 6);
assert_eq!(pm.bytes_read, 3 * 1024 * 1024);
assert!((pm.fraction.unwrap() - 0.625).abs() < 1e-4);
// This progress-only run reported no "wrote N rows" events and no
// timestamped events, so the REAL row total is 0 and the MEASURED
// elapsed is 0.0 — a genuine "nothing to measure", never a
// `tasks_done * 1000` / `5.0` stand-in.
assert_eq!(
pm.rows, 0,
"no row events → real total is 0, not tasks_done*1000"
);
assert_eq!(
pm.elapsed_s, 0.0,
"no timestamps → measured elapsed is 0, not 5.0"
);
}
#[test]
fn per_node_rows_are_real_not_fabricated() {
// FAIL-ON-REGRESSION: a run that attributes 1234 rows to node `l` must
// surface exactly 1234 on that stage's StageMetrics — the real parsed
// count, not a `done * 1000` stand-in. A node with no row event carries
// no rows metric (None), never a fabricated value.
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow 'l' has COMPLETED and wrote 1234 rows".into(),
element: Some("l".into()),
}));
let m = vs.to_pipeline_model();
let l = m.stages.iter().find(|s| s.id == "l").unwrap();
assert_eq!(
l.metrics.rows,
Some(1234),
"the stage surfaces its real, parsed row count"
);
// Throughput/lag have no honest per-node source → stay unset.
assert_eq!(l.metrics.throughput_rows_s, None);
assert_eq!(l.metrics.lag_s, None);
// A node with no row event: no fabricated rows.
let src = m.stages.iter().find(|s| s.id == "src").unwrap();
assert_eq!(
src.metrics.rows, None,
"a node with no row event must not fabricate one"
);
}
#[test]
fn pipeline_rows_summed_from_real_attributions_not_tasks_x1000() {
// FAIL-ON-REGRESSION: pipeline-level rows = the SUM of real per-node
// counts (300 + 200 = 500), never the old `tasks_done * 1000` (= 4000).
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow 'l' has COMPLETED and wrote 300 rows".into(),
element: Some("l".into()),
}));
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow 'r' has COMPLETED and wrote 200 rows".into(),
element: Some("r".into()),
}));
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![StageProgress {
stage_id: 0,
num_tasks: 4,
num_completed_tasks: 4,
input_bytes_read: 0,
done: true,
}],
num_inflight_tasks: 0,
}));
let pm = vs.to_pipeline_model().metrics.expect("progress → metrics");
assert_eq!(
pm.rows, 500,
"pipeline rows is the real sum of per-node counts"
);
assert_ne!(
pm.rows,
pm.tasks_done * 1000,
"must NOT be the old tasks_done*1000 stand-in"
);
}
#[test]
fn elapsed_is_measured_from_timestamps_not_a_constant() {
// FAIL-ON-REGRESSION: elapsed_s is the wall-clock span between the
// first (12:00:00) and last (12:00:07) timestamped event = 7.0s —
// MEASURED, never the old `5.0` constant.
let mut vs = ViewerState::new(diamond());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: Some("2026-06-10T12:00:00Z".into()),
level: EventLevel::Info,
message: "Flow 'src' is STARTING".into(),
element: Some("src".into()),
}));
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: Some("2026-06-10T12:00:07Z".into()),
level: EventLevel::Info,
message: "Flow 'sink' has COMPLETED and wrote 10 rows".into(),
element: Some("sink".into()),
}));
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![StageProgress {
stage_id: 0,
num_tasks: 1,
num_completed_tasks: 1,
input_bytes_read: 0,
done: true,
}],
num_inflight_tasks: 0,
}));
let pm = vs.to_pipeline_model().metrics.expect("progress → metrics");
assert!(
(pm.elapsed_s - 7.0).abs() < 1e-3,
"elapsed measured from event timestamps, got {}",
pm.elapsed_s
);
assert_ne!(pm.elapsed_s, 5.0, "must NOT be the old 5.0 constant");
}
#[test]
fn adapter_trait_and_wrapper_agree() {
let vs = ViewerState::new(diamond());
let a = vs.to_pipeline_model();
let b = SparkPipelineAdapter(&vs).to_pipeline_model();
assert_eq!(a.stages.len(), b.stages.len());
assert_eq!(a.edges.len(), b.edges.len());
}
}