facett-sdp/src/lib.rs
//! **facett-sdp** — a Spark Declarative Pipelines (SDP) editor facet.
//!
//! An SDP pipeline is a set of **flows**; each flow has a `name`, an optional list
//! of upstream `deps` (flow/dataset names it reads from), and a Spark **SQL**
//! `query` that defines it. This component lets a user edit those flows — the SQL
//! in a multi-line box, plus name/deps — add/remove flows, and see a tiny inline
//! DAG preview built from the deps. It does **static** validation only (does every
//! dep reference a known flow?); it never parses SQL and never talks to Spark — the
//! host owns submission. That keeps this a pure, dependency-light editor.
//!
//! It exposes its edited state via [`SdpEditor::state_json`] so a host (korp) can
//! read the flows back, map each to a knut `DataflowGraph` node, and submit to
//! Spark. Mirrors the [`Facet`] shape (title / ui / state_json) like the sibling
//! `facett-markdown` / `facett-search` facets; pure egui, no async, no network.
//!
//! [`Facet`]: https://git.gunnar.rs/nordisk/facett
use std::collections::{HashMap, HashSet};
use egui::{Color32, Ui};
use serde::{Deserialize, Serialize};
/// Editor view mode: edit the flows, or preview the dependency DAG.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum Mode {
Edit,
Graph,
}
/// One flow in an SDP pipeline — the clean, portable representation a host reads
/// back from [`SdpEditor::flows`] / `state_json`. `deps` are the *names* of the
/// upstream flows/datasets this flow's SQL reads from; `query` is Spark SQL.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Flow {
pub name: String,
pub deps: Vec<String>,
pub query: String,
}
impl Flow {
pub fn new(name: impl Into<String>, deps: impl IntoIterator<Item = impl Into<String>>, query: impl Into<String>) -> Self {
Self {
name: name.into(),
deps: deps.into_iter().map(Into::into).collect(),
query: query.into(),
}
}
}
/// Result of the static (non-SQL) validation pass.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Validation {
pub valid: bool,
pub errors: Vec<String>,
}
/// An editable flow row. `deps` are edited as one comma/space-separated line
/// (`deps_raw`) so a single [`egui::TextEdit`] drives them; the parsed list is
/// re-derived on read. Kept private — hosts see the clean [`Flow`].
#[derive(Clone, Debug, Default)]
struct Row {
name: String,
deps_raw: String,
query: String,
}
impl Row {
fn from_flow(f: &Flow) -> Self {
Self { name: f.name.clone(), deps_raw: f.deps.join(", "), query: f.query.clone() }
}
fn to_flow(&self) -> Flow {
Flow { name: self.name.trim().to_string(), deps: parse_deps(&self.deps_raw), query: self.query.clone() }
}
}
const ACCENT: Color32 = Color32::from_rgb(247, 198, 0); // Tillsynia yellow
const OK: Color32 = Color32::from_rgb(120, 190, 140);
const ERR: Color32 = Color32::from_rgb(230, 120, 120);
const NODE: Color32 = Color32::from_rgb(58, 62, 74);
const EDGE: Color32 = Color32::from_rgb(150, 156, 168);
const TEXT: Color32 = Color32::from_rgb(214, 219, 228);
/// The SDP pipeline editor facet.
pub struct SdpEditor {
title: String,
rows: Vec<Row>,
mode: Mode,
}
impl SdpEditor {
pub fn new(title: impl Into<String>) -> Self {
Self { title: title.into(), rows: Vec::new(), mode: Mode::Edit }
}
/// Builder: seed with a flow.
pub fn with_flow(mut self, flow: Flow) -> Self {
self.rows.push(Row::from_flow(&flow));
self
}
/// Builder: seed with many flows.
pub fn with_flows(mut self, flows: impl IntoIterator<Item = Flow>) -> Self {
self.rows = flows.into_iter().map(|f| Row::from_flow(&f)).collect();
self
}
/// Append a flow.
pub fn add_flow(&mut self, flow: Flow) {
self.rows.push(Row::from_flow(&flow));
}
/// Append a blank flow (what the "＋ Add flow" button does).
pub fn add_empty_flow(&mut self) {
self.rows.push(Row::default());
}
/// Remove the flow at `index` (no-op if out of range).
pub fn remove_flow(&mut self, index: usize) {
if index < self.rows.len() {
self.rows.remove(index);
}
}
pub fn len(&self) -> usize {
self.rows.len()
}
pub fn is_empty(&self) -> bool {
self.rows.is_empty()
}
pub fn set_mode(&mut self, mode: Mode) {
self.mode = mode;
}
/// Set the SQL of the flow at `index` (no-op if out of range). Handy for hosts
/// and tests that drive an edit programmatically.
pub fn set_flow_query(&mut self, index: usize, query: impl Into<String>) {
if let Some(r) = self.rows.get_mut(index) {
r.query = query.into();
}
}
/// Set the deps of the flow at `index` (no-op if out of range).
pub fn set_flow_deps(&mut self, index: usize, deps: impl IntoIterator<Item = impl Into<String>>) {
if let Some(r) = self.rows.get_mut(index) {
r.deps_raw = deps.into_iter().map(Into::into).collect::<Vec<String>>().join(", ");
}
}
/// The edited pipeline as clean [`Flow`]s — the host reads this to build its
/// dataflow graph.
pub fn flows(&self) -> Vec<Flow> {
self.rows.iter().map(Row::to_flow).collect()
}
/// Static validation: every dep must reference a known flow; names must be
/// present and unique; no self-loops. This is **not** SQL validation — Spark
/// does that on submit.
pub fn validate(&self) -> Validation {
let flows = self.flows();
let known: HashSet<&str> = flows.iter().map(|f| f.name.as_str()).filter(|n| !n.is_empty()).collect();
let mut errors = Vec::new();
let mut seen: HashSet<&str> = HashSet::new();
for (i, f) in flows.iter().enumerate() {
if f.name.is_empty() {
errors.push(format!("flow #{} has an empty name", i + 1));
} else if !seen.insert(f.name.as_str()) {
errors.push(format!("duplicate flow name '{}'", f.name));
}
for d in &f.deps {
if d == &f.name {
errors.push(format!("flow '{}' depends on itself", f.name));
} else if !known.contains(d.as_str()) {
errors.push(format!("flow '{}' depends on unknown '{}'", f.name, d));
}
}
}
Validation { valid: errors.is_empty(), errors }
}
// ── facett Facet contract ────────────────────────────────────────────────
/// The **shared demo fixture** — a two-flow ETL pipeline (a source with no
/// deps and a downstream dependent) that validates clean, so both the Edit
/// list and the Graph DAG preview paint real geometry. Consumed by the render
/// and matrix tests (L5: one fixture, not one per test file).
pub fn local() -> Self {
Self::new("etl").with_flows([
Flow::new("raw", Vec::<String>::new(), "SELECT * FROM read_files('/in')"),
Flow::new("clean", ["raw"], "SELECT * FROM raw WHERE ok"),
])
}
pub fn title(&self) -> &str {
&self.title
}
/// Observable state as JSON:
/// ```json
/// { "title": "..", "mode": "Edit",
/// "flows": [ { "name": "..", "deps": [".."], "query": ".." } ],
/// "valid": true, "errors": [] }
/// ```
pub fn state_json(&self) -> serde_json::Value {
let v = self.validate();
serde_json::json!({
"title": self.title,
"mode": format!("{:?}", self.mode),
"flows": self.flows(),
"valid": v.valid,
"errors": v.errors,
})
}
pub fn ui(&mut self, ui: &mut Ui) {
ui.horizontal(|ui| {
ui.selectable_value(&mut self.mode, Mode::Edit, "✎ Edit");
ui.selectable_value(&mut self.mode, Mode::Graph, "◈ Graph");
ui.separator();
let v = self.validate();
if v.valid {
ui.label(egui::RichText::new(format!("✔ valid · {} flow(s)", self.rows.len())).small().color(OK));
} else {
ui.label(egui::RichText::new(format!("✖ {} error(s)", v.errors.len())).small().color(ERR));
}
});
ui.separator();
match self.mode {
Mode::Edit => self.edit_ui(ui),
Mode::Graph => self.graph_ui(ui),
}
}
fn edit_ui(&mut self, ui: &mut Ui) {
let mut remove: Option<usize> = None;
egui::ScrollArea::vertical().show(ui, |ui| {
for (i, row) in self.rows.iter_mut().enumerate() {
egui::Frame::group(ui.style()).show(ui, |ui| {
ui.horizontal(|ui| {
let label = if row.name.trim().is_empty() { format!("flow #{}", i + 1) } else { row.name.trim().to_string() };
ui.label(egui::RichText::new(label).strong().color(ACCENT));
if ui.button("✕ Remove").clicked() {
remove = Some(i);
}
});
ui.horizontal(|ui| {
ui.label("name");
ui.add(egui::TextEdit::singleline(&mut row.name).hint_text("flow name").desired_width(180.0));
ui.label("deps");
ui.add(
egui::TextEdit::singleline(&mut row.deps_raw)
.hint_text("upstream, comma-separated")
.desired_width(f32::INFINITY),
);
});
ui.add(
egui::TextEdit::multiline(&mut row.query)
.code_editor()
.desired_width(f32::INFINITY)
.desired_rows(4)
.hint_text("SELECT … FROM … (Spark SQL)"),
);
});
ui.add_space(4.0);
}
if ui.button("＋ Add flow").clicked() {
self.rows.push(Row::default());
}
});
if let Some(i) = remove {
self.rows.remove(i);
}
}
/// A tiny inline DAG preview: nodes laid out in columns by dependency depth,
/// edges drawn from each known dep. Self-contained (no facett-graph) — a
/// pragmatic sketch, not a full graph viewer.
fn graph_ui(&mut self, ui: &mut Ui) {
let flows = self.flows();
if flows.is_empty() {
ui.weak("No flows yet — add one in Edit mode.");
return;
}
let depth = depths(&flows);
// Group indices by depth (column), preserving flow order within a column.
let mut cols: HashMap<usize, Vec<usize>> = HashMap::new();
for (i, f) in flows.iter().enumerate() {
cols.entry(*depth.get(&f.name).unwrap_or(&0)).or_default().push(i);
}
let max_depth = depth.values().copied().max().unwrap_or(0);
let max_rows = cols.values().map(|v| v.len()).max().unwrap_or(1);
let col_w = 150.0;
let row_h = 56.0;
let node_r = 8.0;
let width = (max_depth as f32 + 1.0) * col_w + 40.0;
let height = (max_rows as f32).max(1.0) * row_h + 40.0;
let (resp, painter) = ui.allocate_painter(egui::vec2(width.max(200.0), height.max(80.0)), egui::Sense::hover());
let origin = resp.rect.min + egui::vec2(24.0, 24.0);
// Position every flow by (column = depth, row = order within column).
let mut pos: HashMap<String, egui::Pos2> = HashMap::new();
for (d, idxs) in &cols {
for (r, &i) in idxs.iter().enumerate() {
let p = origin + egui::vec2(*d as f32 * col_w, r as f32 * row_h);
pos.insert(flows[i].name.clone(), p);
}
}
// Edges: dep → flow (only for deps that are known flows).
for f in &flows {
if let Some(&to) = pos.get(&f.name) {
for dep in &f.deps {
if let Some(&from) = pos.get(dep) {
painter.line_segment([from + egui::vec2(node_r, 0.0), to - egui::vec2(node_r, 0.0)], egui::Stroke::new(1.5_f32, EDGE));
}
}
}
}
// Nodes.
for f in &flows {
if let Some(&p) = pos.get(&f.name) {
painter.circle_filled(p, node_r, NODE);
painter.circle_stroke(p, node_r, egui::Stroke::new(1.5_f32, ACCENT));
painter.text(
p + egui::vec2(node_r + 4.0, 0.0),
egui::Align2::LEFT_CENTER,
&f.name,
egui::FontId::proportional(13.0),
TEXT,
);
}
}
}
}
/// Parse a comma/whitespace-separated deps line into unique, trimmed names
/// (first-seen order preserved).
fn parse_deps(raw: &str) -> Vec<String> {
let mut out: Vec<String> = Vec::new();
for tok in raw.split(|c: char| c == ',' || c.is_whitespace()) {
let t = tok.trim();
if !t.is_empty() && !out.iter().any(|s| s == t) {
out.push(t.to_string());
}
}
out
}
/// Longest-path depth of each flow over the known-dep edges. Cycle-safe: the
/// relaxation is bounded to `flows.len()` passes, so a cycle just caps out instead
/// of looping forever.
fn depths(flows: &[Flow]) -> HashMap<String, usize> {
let known: HashSet<&str> = flows.iter().map(|f| f.name.as_str()).collect();
let mut depth: HashMap<String, usize> = flows.iter().map(|f| (f.name.clone(), 0usize)).collect();
for _ in 0..flows.len() {
let mut changed = false;
for f in flows {
let mut d = 0usize;
for dep in &f.deps {
if known.contains(dep.as_str()) && dep != &f.name {
d = d.max(depth.get(dep).copied().unwrap_or(0) + 1);
}
}
if d > *depth.get(&f.name).unwrap_or(&0) {
depth.insert(f.name.clone(), d);
changed = true;
}
}
if !changed {
break;
}
}
depth
}
// ── The host-facing `Facet` contract ───────────────────────────────────────────
impl facett_core::Facet for SdpEditor {
fn title(&self) -> &str {
SdpEditor::title(self)
}
fn ui(&mut self, ui: &mut Ui) {
SdpEditor::ui(self, ui)
}
fn state_json(&self) -> serde_json::Value {
SdpEditor::state_json(self)
}
/// The **robot drive seam**: `"graph"` / `"edit"` flip the mode toggle,
/// `{"add_flow":{"name":..,"deps":[..],"query":".."}}` is the Add button,
/// `{"remove_flow":<i>}` the Remove button, and
/// `{"set_query":{"index":<i>,"query":".."}}` retypes a SQL box.
fn update_json(&mut self, msg_json: &str) {
let Ok(v) = serde_json::from_str::<serde_json::Value>(msg_json) else { return };
match v.as_str() {
Some("graph") => return self.set_mode(Mode::Graph),
Some("edit") => return self.set_mode(Mode::Edit),
Some("add_empty_flow") => return self.add_empty_flow(),
_ => {}
}
if let Some(i) = v.get("remove_flow").and_then(|n| n.as_u64()) {
return self.remove_flow(i as usize);
}
if let Some(f) = v.get("add_flow") {
let name = f.get("name").and_then(|s| s.as_str()).unwrap_or("");
let query = f.get("query").and_then(|s| s.as_str()).unwrap_or("");
let deps: Vec<String> = f
.get("deps")
.and_then(|d| d.as_array())
.map(|a| a.iter().filter_map(|d| d.as_str().map(|s| s.to_string())).collect())
.unwrap_or_default();
return self.add_flow(Flow::new(name, deps, query));
}
if let Some(q) = v.get("set_query")
&& let (Some(i), Some(text)) =
(q.get("index").and_then(|n| n.as_u64()), q.get("query").and_then(|s| s.as_str()))
{
self.set_flow_query(i as usize, text);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample() -> SdpEditor {
SdpEditor::new("etl").with_flows([
Flow::new("raw", Vec::<String>::new(), "SELECT * FROM read_files('/in')"),
Flow::new("clean", ["raw"], "SELECT * FROM raw WHERE ok"),
Flow::new("marts", ["clean"], "SELECT k, count(*) FROM clean GROUP BY k"),
])
}
#[test]
fn parse_deps_splits_and_dedups() {
assert_eq!(parse_deps("a, b c,,a"), vec!["a", "b", "c"]);
assert!(parse_deps(" ").is_empty());
}
#[test]
fn flows_roundtrip_through_rows() {
let ed = sample();
let f = ed.flows();
assert_eq!(f.len(), 3);
assert_eq!(f[1].name, "clean");
assert_eq!(f[1].deps, vec!["raw"]);
assert!(f[2].query.contains("GROUP BY"));
}
#[test]
fn valid_pipeline_is_valid() {
let v = sample().validate();
assert!(v.valid, "errors: {:?}", v.errors);
assert!(v.errors.is_empty());
}
#[test]
fn unknown_dep_is_flagged_invalid() {
let mut ed = sample();
ed.set_flow_deps(1, ["nope"]);
let v = ed.validate();
assert!(!v.valid);
assert!(v.errors.iter().any(|e| e.contains("unknown 'nope'")));
}
#[test]
fn duplicate_name_and_self_dep_flagged() {
let ed = SdpEditor::new("x").with_flows([
Flow::new("a", ["a"], "SELECT 1"),
Flow::new("a", Vec::<String>::new(), "SELECT 2"),
]);
let v = ed.validate();
assert!(!v.valid);
assert!(v.errors.iter().any(|e| e.contains("depends on itself")));
assert!(v.errors.iter().any(|e| e.contains("duplicate flow name 'a'")));
}
#[test]
fn state_json_shape_and_reflects_edit() {
let mut ed = sample();
ed.set_flow_query(0, "SELECT 42 AS answer");
let s = ed.state_json();
assert_eq!(s["mode"], "Edit");
assert_eq!(s["valid"], true);
assert_eq!(s["flows"][0]["name"], "raw");
assert_eq!(s["flows"][0]["query"], "SELECT 42 AS answer");
assert_eq!(s["flows"][1]["deps"][0], "raw");
assert!(s["errors"].as_array().unwrap().is_empty());
}
#[test]
fn depths_are_layered_and_cycle_safe() {
let d = depths(&sample().flows());
assert_eq!(d["raw"], 0);
assert_eq!(d["clean"], 1);
assert_eq!(d["marts"], 2);
// a 2-cycle must terminate (bounded to len passes) — reaching here proves it
let cyc = vec![Flow::new("a", ["b"], ""), Flow::new("b", ["a"], "")];
let dc = depths(&cyc);
assert!(dc.contains_key("a") && dc.contains_key("b"));
}
}