crates/knut-pipelines/src/progress.rs
//! Live Spark **job** progress, and the unified viewer state.
//!
//! Two things live here, both dependency-light (no gRPC, no `protoc`):
//!
//! * [`ExecutionProgress`] / [`StageProgress`] — a mirror of Spark Connect's
//! `ExecutePlanResponse.ExecutionProgress` (base.proto). Every Spark Connect
//! query — a plain `spark.sql(...)`, *not* only a pipeline — streams these
//! as it runs: per-stage task counts and bytes read. This is what makes the
//! viewer a *Spark jobs* viewer and not just a pipeline-DAG viewer.
//! * [`ViewerState`] — the single model the CLI text view and the egui/snarl
//! GUI both render. It folds the static [`DataflowGraph`] together with the
//! live stream (events + progress + per-node run status) so a frontend is a
//! pure function of this struct. Keeping it here (not in the GUI) is what
//! guarantees CLI parity: anything the GUI can show, [`ViewerState`] can
//! describe and the CLI can print.
use serde::{Deserialize, Serialize};
use crate::model::{DataflowGraph, EventLevel, PipelineRunEvent};
/// Progress of one Spark stage within a job.
///
/// Mirrors `ExecutionProgress.StageInfo`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct StageProgress {
/// Spark stage id.
pub stage_id: i64,
/// Total tasks planned for the stage.
pub num_tasks: i64,
/// Tasks completed so far.
pub num_completed_tasks: i64,
/// Input bytes read by the stage so far.
pub input_bytes_read: i64,
/// Whether the stage has finished.
pub done: bool,
}
impl StageProgress {
/// Fraction of tasks completed in `[0.0, 1.0]`. A stage with zero tasks
/// counts as complete (`1.0`) so empty stages don't stall a progress bar.
pub fn fraction(&self) -> f32 {
if self.num_tasks <= 0 {
return 1.0;
}
(self.num_completed_tasks as f32 / self.num_tasks as f32).clamp(0.0, 1.0)
}
}
/// A snapshot of a Spark job's progress: every stage plus the count of tasks
/// currently in flight.
///
/// Mirrors `ExecutePlanResponse.ExecutionProgress`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutionProgress {
/// Per-stage progress.
pub stages: Vec<StageProgress>,
/// Tasks running right now across all stages.
pub num_inflight_tasks: i64,
}
impl ExecutionProgress {
/// Total tasks across every stage.
pub fn total_tasks(&self) -> i64 {
self.stages.iter().map(|s| s.num_tasks).sum()
}
/// Total completed tasks across every stage.
pub fn completed_tasks(&self) -> i64 {
self.stages.iter().map(|s| s.num_completed_tasks).sum()
}
/// Total input bytes read across every stage.
pub fn bytes_read(&self) -> i64 {
self.stages.iter().map(|s| s.input_bytes_read).sum()
}
/// Overall completed fraction in `[0.0, 1.0]`, task-weighted across stages.
pub fn fraction(&self) -> f32 {
let total = self.total_tasks();
if total <= 0 {
return if self.stages.iter().all(|s| s.done) && !self.stages.is_empty() {
1.0
} else {
0.0
};
}
(self.completed_tasks() as f32 / total as f32).clamp(0.0, 1.0)
}
/// `true` when every stage reports done.
pub fn all_done(&self) -> bool {
!self.stages.is_empty() && self.stages.iter().all(|s| s.done)
}
/// One-line human summary, e.g. `7/12 tasks · 3 inflight · 1.2 MiB read`.
pub fn summary(&self) -> String {
format!(
"{}/{} tasks · {} inflight · {} read",
self.completed_tasks(),
self.total_tasks(),
self.num_inflight_tasks,
human_bytes(self.bytes_read()),
)
}
}
/// Format a byte count with binary units. Small, dependency-free.
pub fn human_bytes(n: i64) -> String {
const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];
if n < 1024 {
return format!("{n} B");
}
let mut v = n as f64;
let mut u = 0;
while v >= 1024.0 && u < UNITS.len() - 1 {
v /= 1024.0;
u += 1;
}
format!("{v:.1} {}", UNITS[u])
}
/// Run status of a single dataflow node (dataset), derived from the event
/// stream. Drives node colour in the GUI and the status column in the CLI.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NodeStatus {
/// Not yet started / no event seen.
#[default]
Idle,
/// A flow into this dataset is running.
Running,
/// Completed successfully.
Done,
/// A flow into this dataset failed.
Failed,
}
impl NodeStatus {
/// A short, stable label.
pub fn label(self) -> &'static str {
match self {
NodeStatus::Idle => "idle",
NodeStatus::Running => "running",
NodeStatus::Done => "done",
NodeStatus::Failed => "failed",
}
}
}
/// Lazily-computed, cached derived views over a [`ViewerState`]. Recomputed
/// only when the underlying state changes (the `dirty` flag), so the per-frame
/// GUI/CLI render doesn't re-scan the graph every frame. Skipped by serde and
/// rebuilt on demand after a clone/deserialize.
#[derive(Debug)]
struct DerivedCache {
/// `true` when `status_counts`/`topo_order` must be recomputed.
dirty: bool,
/// Cached `(idle, running, done, failed)` node-status counts.
status_counts: (usize, usize, usize, usize),
/// Cached topological order (`None` when the graph has a cycle).
topo_order: Option<Vec<String>>,
}
impl Default for DerivedCache {
fn default() -> Self {
// Start dirty so the first access recomputes against the real graph,
// regardless of how the owning state was built (new/default/deserialize).
DerivedCache {
dirty: true,
status_counts: (0, 0, 0, 0),
topo_order: None,
}
}
}
impl Clone for DerivedCache {
fn clone(&self) -> Self {
// A clone starts dirty: recompute on first access rather than copy a
// possibly-stale cache.
DerivedCache::default()
}
}
/// The complete viewer model: the static graph plus everything learned from a
/// live (or replayed) run. Both the CLI and the egui/snarl GUI render purely
/// from this — the single source of truth that keeps the two in lockstep.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ViewerState {
/// The dataflow DAG (may be empty before one is fetched/defined).
pub graph: DataflowGraph,
/// Per-dataset run status, indexed by dataset name.
pub node_status: std::collections::BTreeMap<String, NodeStatus>,
/// The event log, in arrival order.
pub events: Vec<PipelineRunEvent>,
/// Rows written per dataset, parsed from structured flow-progress events
/// (the latest reported count wins). Empty until a "wrote N rows"-style
/// event is attributed to a node.
#[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty", default)]
pub rows_written: std::collections::BTreeMap<String, u64>,
/// The most recent Spark job progress snapshot, if any.
#[serde(skip_serializing_if = "Option::is_none", default)]
pub progress: Option<ExecutionProgress>,
/// Dataset names in graph order, cached when the graph is set so event
/// attribution can skip the message scan when an event already names its
/// element (H11). Graph order is preserved so the "first matching dataset"
/// tie-break stays identical to scanning `graph.datasets` directly.
#[serde(skip)]
dataset_names: Vec<String>,
/// Lazily-recomputed derived views (status counts, topo order) guarded by a
/// dirty flag so they aren't rebuilt every frame (M5).
#[serde(skip)]
derived: std::cell::RefCell<DerivedCache>,
}
impl ViewerState {
/// A viewer over a static graph, with no run data yet.
pub fn new(graph: DataflowGraph) -> Self {
let node_status = graph
.datasets
.iter()
.map(|d| (d.name.clone(), NodeStatus::Idle))
.collect();
let dataset_names = graph.datasets.iter().map(|d| d.name.clone()).collect();
ViewerState {
graph,
node_status,
events: Vec::new(),
rows_written: std::collections::BTreeMap::new(),
progress: None,
dataset_names,
derived: std::cell::RefCell::new(DerivedCache::default()),
}
}
/// Status of one dataset (defaults to [`NodeStatus::Idle`]).
pub fn status_of(&self, dataset: &str) -> NodeStatus {
self.node_status.get(dataset).copied().unwrap_or_default()
}
/// Apply one streamed update, mutating the state the frontend renders.
///
/// Returns `true` when the update changed the rendered model (a node's
/// status, the progress snapshot, or the graph) so callers — notably the
/// GUI — can rebuild derived state and repaint only when something actually
/// moved, rather than every frame.
pub fn apply(&mut self, update: ViewerUpdate) -> bool {
match update {
ViewerUpdate::Event(ev) => {
// Parse the structured details out of the message once (flow
// status keyword + rows-written count), then attribute to a node.
let parsed = parse_event_message(&ev.message);
if let Some(node) = self.attribute(&ev) {
let status = match ev.level {
EventLevel::Error => NodeStatus::Failed,
// A structured status keyword (STARTING/COMPLETED/FAILED
// …) is authoritative; else fall back to the completion
// heuristic (message already lower-cased once, L).
_ => parsed.status.unwrap_or_else(|| {
if is_completion(&ev.message.to_ascii_lowercase()) {
NodeStatus::Done
} else {
NodeStatus::Running
}
}),
};
self.node_status.insert(node.clone(), status);
// Attribute a parsed row count to the same node (latest wins).
if let Some(rows) = parsed.rows_written {
self.rows_written.insert(node, rows);
}
}
self.events.push(ev);
// An event always touches the log and may touch node status;
// mark derived views (counts) stale.
self.derived.borrow_mut().dirty = true;
}
ViewerUpdate::Progress(p) => self.progress = Some(p),
ViewerUpdate::GraphReplaced(g) => *self = ViewerState::new(g),
}
true
}
/// Best-effort: which **dataset** an event is about, so its status/rows land
/// on the right node. Resolution order:
///
/// 1. an explicit `element` on the event (resolved flow→target if it names
/// a flow rather than a dataset — no scan);
/// 2. a quoted name parsed from the message (`Flow 'silver' …`), likewise
/// resolved flow→target;
/// 3. a scan of the message for any known dataset name (graph order).
///
/// The dataset-name set is cached on the state (H11) so steps 1–2 skip the
/// scan whenever the event already names its element.
fn attribute(&self, ev: &PipelineRunEvent) -> Option<String> {
if let Some(e) = &ev.element {
return Some(self.resolve_to_dataset(e));
}
// A quoted element parsed from the message (flow or dataset name).
if let Some(e) = parse_event_message(&ev.message).element {
if let Some(ds) = self.known_dataset(&e) {
return Some(ds);
}
}
// Scan only when nothing named the element. The cache is skipped by
// serde; rebuild from the graph if it was lost (e.g. after a
// deserialize) so attribution stays identical.
if self.dataset_names.is_empty() && !self.graph.datasets.is_empty() {
return self
.graph
.datasets
.iter()
.find(|d| ev.message.contains(&d.name))
.map(|d| d.name.clone());
}
self.dataset_names
.iter()
.find(|name| ev.message.contains(name.as_str()))
.cloned()
}
/// Resolve a name that may be a dataset **or** a flow into the dataset it
/// belongs to: a dataset name maps to itself; a flow name maps to its
/// target dataset; anything else is returned unchanged (best effort).
fn resolve_to_dataset(&self, name: &str) -> String {
if self.graph.dataset(name).is_some() {
return name.to_string();
}
if let Some(f) = self.graph.flows.iter().find(|f| f.name == name) {
return f.target.clone();
}
name.to_string()
}
/// If `name` names a known dataset (→ itself) or flow (→ its target),
/// return that dataset; else `None` so the caller can fall back to a scan.
fn known_dataset(&self, name: &str) -> Option<String> {
if self.graph.dataset(name).is_some() {
return Some(name.to_string());
}
self.graph
.flows
.iter()
.find(|f| f.name == name)
.map(|f| f.target.clone())
}
/// Recompute the cached derived views (status counts, topo order) when the
/// dirty flag is set; a no-op otherwise.
fn refresh_derived(&self) {
let mut d = self.derived.borrow_mut();
if !d.dirty {
return;
}
let (mut idle, mut running, mut done, mut failed) = (0, 0, 0, 0);
for ds in &self.graph.datasets {
match self.status_of(&ds.name) {
NodeStatus::Idle => idle += 1,
NodeStatus::Running => running += 1,
NodeStatus::Done => done += 1,
NodeStatus::Failed => failed += 1,
}
}
d.status_counts = (idle, running, done, failed);
d.topo_order = self.graph.topo_order();
d.dirty = false;
}
/// Count of nodes in each status, for a compact CLI/GUI summary line.
/// Cached behind a dirty flag (M5) so it isn't recomputed every frame.
pub fn status_counts(&self) -> (usize, usize, usize, usize) {
self.refresh_derived();
self.derived.borrow().status_counts
}
/// Topological dataset order, or `None` on a cycle. Cached behind the same
/// dirty flag as [`status_counts`](Self::status_counts) (M5).
pub fn topo_order(&self) -> Option<Vec<String>> {
self.refresh_derived();
self.derived.borrow().topo_order.clone()
}
/// A compact, machine-readable snapshot of everything the viewer is showing
/// — the **introspection surface for headless robot-testing**: assert what's
/// on screen without a display or a screenshot. The GUI dumps this to
/// `$KNUT_VIEWER_STATE` every frame; a test builds a `ViewerState`, applies
/// updates, and asserts on this directly (no `eframe::run`).
///
/// **Rule:** any new visible list / status / panel the viewer renders MUST
/// be reflected here — introspection coverage is part of the UI. `placeholders`
/// lists the empty-state strings currently shown, so a test can assert it's
/// empty for a state that should have data.
pub fn state_json(&self) -> serde_json::Value {
use serde_json::json;
let (idle, running, done, failed) = self.status_counts();
let datasets: Vec<&str> = self
.graph
.datasets
.iter()
.map(|d| d.name.as_str())
.collect();
let node_status: serde_json::Map<String, serde_json::Value> = self
.node_status
.iter()
.map(|(k, v)| (k.clone(), json!(v.label())))
.collect();
let mut placeholders: Vec<&str> = Vec::new();
if self.graph.datasets.is_empty() {
placeholders.push("no pipeline loaded");
}
if self.events.is_empty() {
placeholders.push("no run events yet");
}
json!({
"graph_id": self.graph.graph_id,
"datasets": datasets,
"dataset_count": datasets.len(),
"flow_count": self.graph.flows.len(),
"node_status": node_status,
"status_counts": { "idle": idle, "running": running, "done": done, "failed": failed },
"topo_order": self.topo_order(),
"events": self.events.len(),
"rows_written": self.rows_written,
"last_event": self.events.last().map(|e| e.message.as_str()),
"job_progress": self.progress.as_ref().map(|p| json!({
"fraction": p.fraction(),
"completed_tasks": p.completed_tasks(),
"total_tasks": p.total_tasks(),
"inflight_tasks": p.num_inflight_tasks,
"bytes_read": p.bytes_read(),
"stages": p.stages.len(),
"all_done": p.all_done(),
})),
"placeholders": placeholders,
})
}
}
/// Heuristic: does a message text indicate a flow/dataset finished?
///
/// Takes the **already lower-cased** message so the caller can lower-case once
/// per event and reuse it (L).
fn is_completion(lowered: &str) -> bool {
lowered.contains("completed")
|| lowered.contains("finished")
|| lowered.contains("materialized")
}
/// The structured details parsed out of a [`PipelineRunEvent`]'s free-text
/// message. The Spark Connect `PipelineEvent` proto still carries only a
/// timestamp + a human message (no typed fields), so a viewer recovers the
/// structure the DAG needs from the text — the element the line is about, the
/// flow's run status, and any row count it reports. As Spark enriches
/// `PipelineEvent` with typed fields, this becomes the fallback.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ParsedEvent {
/// A quoted element name the message is about (a flow or dataset), if one
/// is present (`Flow 'silver' has COMPLETED` → `silver`).
pub element: Option<String>,
/// The run status the message implies, if a status keyword is present.
pub status: Option<NodeStatus>,
/// A row count the message reports (`… wrote 1234 rows` → `1234`).
pub rows_written: Option<u64>,
}
/// Parse the structured details out of a pipeline event message.
///
/// Recognises the shapes Spark's SDP flow-progress events use: a quoted
/// element (`'name'` or `` `name` ``), a status keyword (STARTING / RUNNING /
/// QUEUED → running, COMPLETED / FINISHED / MATERIALIZED → done, FAILED /
/// ERROR → failed), and a "wrote N rows" / "N rows written" count. All
/// dependency-free (no regex).
pub fn parse_event_message(message: &str) -> ParsedEvent {
let element = quoted_name(message);
let lower = message.to_ascii_lowercase();
let status = if lower.contains("fail") || lower.contains("error") {
Some(NodeStatus::Failed)
} else if lower.contains("complet")
|| lower.contains("finished")
|| lower.contains("materialized")
{
Some(NodeStatus::Done)
} else if lower.contains("start")
|| lower.contains("running")
|| lower.contains("queued")
|| lower.contains("flowing")
{
Some(NodeStatus::Running)
} else {
None
};
ParsedEvent {
element,
status,
rows_written: rows_written(&lower),
}
}
/// The first single- or back-quoted name in `s`, if any.
fn quoted_name(s: &str) -> Option<String> {
for quote in ['\'', '`'] {
if let Some(start) = s.find(quote) {
if let Some(len) = s[start + 1..].find(quote) {
let name = &s[start + 1..start + 1 + len];
if !name.is_empty() {
return Some(name.to_string());
}
}
}
}
None
}
/// Parse a row count from an already-lower-cased message: the integer token
/// immediately preceding a `row`/`rows` word (`wrote 1,234 rows` → `1234`).
fn rows_written(lower: &str) -> Option<u64> {
let mut prev: Option<&str> = None;
for word in lower.split(|c: char| c.is_whitespace()) {
if word.starts_with("row") {
if let Some(p) = prev {
let digits: String = p.chars().filter(|c| c.is_ascii_digit()).collect();
if !digits.is_empty() {
if let Ok(n) = digits.parse::<u64>() {
return Some(n);
}
}
}
}
prev = Some(word);
}
None
}
/// One item off the unified live stream the viewer consumes. A run yields a
/// mix of pipeline events, Spark job-progress snapshots, and (when fetching a
/// fresh graph) a graph replacement.
#[derive(Debug, Clone)]
pub enum ViewerUpdate {
/// A pipeline run event (log line).
Event(PipelineRunEvent),
/// A Spark job-progress snapshot.
Progress(ExecutionProgress),
/// The static graph was (re)fetched; replace and reset run state.
GraphReplaced(DataflowGraph),
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{Dataset, Flow, OutputType, PipelineRunEvent};
fn graph() -> DataflowGraph {
DataflowGraph::new()
.with_dataset(Dataset::new("bronze", OutputType::Table))
.with_dataset(Dataset::new("silver", OutputType::MaterializedView))
.with_flow(Flow::new("ingest", "bronze", Vec::<String>::new()))
.with_flow(Flow::new("clean", "silver", ["bronze"]))
}
#[test]
fn stage_fraction_handles_zero_tasks() {
let s = StageProgress {
stage_id: 1,
num_tasks: 0,
num_completed_tasks: 0,
input_bytes_read: 0,
done: true,
};
assert_eq!(s.fraction(), 1.0);
let s = StageProgress {
stage_id: 1,
num_tasks: 4,
num_completed_tasks: 1,
input_bytes_read: 0,
done: false,
};
assert_eq!(s.fraction(), 0.25);
}
#[test]
fn execution_progress_aggregates() {
let p = ExecutionProgress {
stages: vec![
StageProgress {
stage_id: 0,
num_tasks: 4,
num_completed_tasks: 4,
input_bytes_read: 1024,
done: true,
},
StageProgress {
stage_id: 1,
num_tasks: 4,
num_completed_tasks: 2,
input_bytes_read: 2048,
done: false,
},
],
num_inflight_tasks: 2,
};
assert_eq!(p.total_tasks(), 8);
assert_eq!(p.completed_tasks(), 6);
assert_eq!(p.bytes_read(), 3072);
assert_eq!(p.fraction(), 0.75);
assert!(!p.all_done());
}
#[test]
fn human_bytes_units() {
assert_eq!(human_bytes(512), "512 B");
assert_eq!(human_bytes(1024), "1.0 KiB");
assert_eq!(human_bytes(1536), "1.5 KiB");
assert_eq!(human_bytes(1024 * 1024), "1.0 MiB");
}
#[test]
fn state_json_reflects_visible_state() {
let mut vs = ViewerState::new(graph());
// Before any run: datasets + flows populated, all idle, an empty-events
// placeholder, no job progress yet.
let s = vs.state_json();
assert_eq!(s["dataset_count"], 2);
assert_eq!(s["datasets"], serde_json::json!(["bronze", "silver"]));
assert_eq!(s["flow_count"], 2);
assert_eq!(s["status_counts"]["idle"], 2);
assert_eq!(s["events"], 0);
assert!(s["job_progress"].is_null());
let ph: Vec<&str> = s["placeholders"]
.as_array()
.unwrap()
.iter()
.map(|p| p.as_str().unwrap())
.collect();
assert!(ph.contains(&"no run events yet"));
assert!(!ph.contains(&"no pipeline loaded"));
// A completion event flips bronze to done and clears the events placeholder.
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow bronze has COMPLETED".into(),
element: Some("bronze".into()),
}));
let s = vs.state_json();
assert_eq!(s["node_status"]["bronze"], "done");
assert_eq!(s["status_counts"]["done"], 1);
assert_eq!(s["events"], 1);
assert_eq!(s["last_event"], "Flow bronze has COMPLETED");
let ph: Vec<&str> = s["placeholders"]
.as_array()
.unwrap()
.iter()
.map(|p| p.as_str().unwrap())
.collect();
assert!(!ph.contains(&"no run events yet"));
}
#[test]
fn state_json_empty_graph_flags_placeholder() {
let s = ViewerState::new(DataflowGraph::new()).state_json();
assert_eq!(s["dataset_count"], 0);
let ph: Vec<&str> = s["placeholders"]
.as_array()
.unwrap()
.iter()
.map(|p| p.as_str().unwrap())
.collect();
assert!(ph.contains(&"no pipeline loaded"));
}
#[test]
fn viewer_state_tracks_node_status_from_events() {
let mut vs = ViewerState::new(graph());
assert_eq!(vs.status_of("bronze"), NodeStatus::Idle);
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow ingest writing to bronze".into(),
element: Some("bronze".into()),
}));
assert_eq!(vs.status_of("bronze"), NodeStatus::Running);
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "bronze materialized".into(),
element: Some("bronze".into()),
}));
assert_eq!(vs.status_of("bronze"), NodeStatus::Done);
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: "silver failed: boom".into(),
element: None, // attributed by message scan
}));
assert_eq!(vs.status_of("silver"), NodeStatus::Failed);
let (idle, running, done, failed) = vs.status_counts();
assert_eq!((idle, running, done, failed), (0, 0, 1, 1));
assert_eq!(vs.events.len(), 3);
}
#[test]
fn viewer_state_records_progress() {
let mut vs = ViewerState::new(graph());
vs.apply(ViewerUpdate::Progress(ExecutionProgress {
stages: vec![StageProgress {
stage_id: 0,
num_tasks: 10,
num_completed_tasks: 5,
input_bytes_read: 4096,
done: false,
}],
num_inflight_tasks: 5,
}));
let p = vs.progress.as_ref().unwrap();
assert_eq!(p.fraction(), 0.5);
assert!(p.summary().contains("5/10 tasks"));
}
#[test]
fn graph_replace_resets_run_state() {
let mut vs = ViewerState::new(graph());
vs.apply(ViewerUpdate::Event(PipelineRunEvent::info("hi")));
vs.apply(ViewerUpdate::GraphReplaced(graph()));
assert!(vs.events.is_empty());
assert_eq!(vs.status_of("bronze"), NodeStatus::Idle);
}
// --- H11: cached dataset-name attribution -------------------------------
#[test]
fn attribute_explicit_element_skips_scan() {
let vs = ViewerState::new(graph());
// An explicit element wins and is returned verbatim, even if the
// message names a *different* dataset.
let got = vs.attribute(&PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "bronze materialized".into(),
element: Some("silver".into()),
});
assert_eq!(got.as_deref(), Some("silver"));
}
#[test]
fn attribute_scan_returns_first_dataset_in_graph_order() {
// Message mentions both datasets; the scan must return the first in
// graph order ("bronze"), identical to scanning `graph.datasets`.
let vs = ViewerState::new(graph());
let got = vs.attribute(&PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "joined bronze into silver".into(),
element: None,
});
assert_eq!(got.as_deref(), Some("bronze"));
}
#[test]
fn attribute_scan_rebuilds_cache_after_deserialize() {
// A round-trip drops the (serde-skipped) name cache; attribution must
// still work by rebuilding from the graph.
let json = serde_json::to_string(&ViewerState::new(graph())).unwrap();
let vs: ViewerState = serde_json::from_str(&json).unwrap();
assert!(vs.dataset_names.is_empty()); // cache not serialized
let got = vs.attribute(&PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "silver failed".into(),
element: None,
});
assert_eq!(got.as_deref(), Some("silver"));
}
// --- M5: cached status_counts / topo_order ------------------------------
#[test]
fn derived_views_cache_and_invalidate() {
let mut vs = ViewerState::new(graph());
// Empty-run baseline: both datasets idle, topo order bronze -> silver.
assert_eq!(vs.status_counts(), (2, 0, 0, 0));
assert_eq!(
vs.topo_order(),
Some(vec!["bronze".into(), "silver".into()])
);
// A second call hits the cache (dirty cleared) and returns the same.
assert_eq!(vs.status_counts(), (2, 0, 0, 0));
// Applying an event marks the cache dirty; the next read recomputes.
let changed = vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "bronze materialized".into(),
element: Some("bronze".into()),
}));
assert!(changed);
assert_eq!(vs.status_counts(), (1, 0, 1, 0));
}
// --- item 5: structured event parsing + flow-name attribution ----------
#[test]
fn parse_event_message_extracts_element_status_rows() {
let p = parse_event_message("Flow 'silver' has COMPLETED and wrote 1,234 rows.");
assert_eq!(p.element.as_deref(), Some("silver"));
assert_eq!(p.status, Some(NodeStatus::Done));
assert_eq!(p.rows_written, Some(1234));
let p = parse_event_message("Flow `bronze` is STARTING.");
assert_eq!(p.element.as_deref(), Some("bronze"));
assert_eq!(p.status, Some(NodeStatus::Running));
assert_eq!(p.rows_written, None);
let p = parse_event_message("Flow 'gold' FAILED: division by zero");
assert_eq!(p.element.as_deref(), Some("gold"));
assert_eq!(p.status, Some(NodeStatus::Failed));
// No quotes, no keyword, no rows.
assert_eq!(
parse_event_message("initializing run"),
ParsedEvent::default()
);
}
#[test]
fn events_attributed_by_flow_name_land_on_target_dataset() {
// The event names the *flow* ("clean"), not the dataset; it must be
// attributed to the flow's target ("silver").
let mut vs = ViewerState::new(graph());
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: "Flow 'clean' has COMPLETED and wrote 42 rows".into(),
element: None,
}));
assert_eq!(vs.status_of("silver"), NodeStatus::Done);
assert_eq!(vs.rows_written.get("silver"), Some(&42));
}
#[test]
fn replayed_run_drives_final_state_without_a_server() {
// A synthetic replay of the events a real SDP run would emit, driven
// through ViewerState end to end — no gRPC, no Spark server.
let mut vs = ViewerState::new(graph());
let stream = [
"Run started for dataflow graph.",
"Flow 'ingest' is STARTING.",
"Flow 'ingest' has COMPLETED and wrote 1000 rows.",
"Flow 'clean' is RUNNING.",
"Flow 'clean' has COMPLETED and wrote 500 rows.",
];
for msg in stream {
vs.apply(ViewerUpdate::Event(PipelineRunEvent::info(msg)));
}
// ingest -> bronze, clean -> silver: both done, row counts attributed.
assert_eq!(vs.status_of("bronze"), NodeStatus::Done);
assert_eq!(vs.status_of("silver"), NodeStatus::Done);
assert_eq!(vs.rows_written.get("bronze"), Some(&1000));
assert_eq!(vs.rows_written.get("silver"), Some(&500));
assert_eq!(vs.status_counts(), (0, 0, 2, 0));
let s = vs.state_json();
assert_eq!(s["rows_written"]["bronze"], 1000);
assert_eq!(s["rows_written"]["silver"], 500);
}
#[test]
fn apply_reports_change() {
let mut vs = ViewerState::new(graph());
assert!(vs.apply(ViewerUpdate::Progress(ExecutionProgress::default())));
assert!(vs.apply(ViewerUpdate::Event(PipelineRunEvent::info("hi"))));
assert!(vs.apply(ViewerUpdate::GraphReplaced(graph())));
}
}