facett-plandag/src/lib.rs
//! **facett-plandag** — a **metric-bearing operator DAG**: the Spark **SQL query
//! plan** / Databricks **Query Profile graph**. Each node carries a multi-line
//! **metric body** (time / memory / rows / spill); a **cost-heat** switch
//! (Time / Memory / Rows) fills nodes by relative cost; **WholeStageCodegen groups**
//! collapse into super-nodes; and an **AQE initial-vs-final** toggle flips between
//! the two plan versions. The four things the doc says no single facett graph had
//! at once.
//!
//! Shared layered layout places producers on the left. Falsett (`facett-flow`)
//! draws persistent geometry; this component owns metric overlays and navigation.
//!
//! ## FC contract
//! - **Serializable state** (`state_json`): the resolved graph (every node's metric
//! body + its **cost-heat as DATA**), the active metric/version, the collapse set,
//! the selection.
//! - **[`Msg`] enum + [`PlanDagView::update`]**: the robot-/CLI-addressable control
//! surface — switch the cost metric, flip the AQE version, collapse a group,
//! select a node — the same effect the header controls drive.
//! - **Render-proof**: the painter and `state_json` both read from one
//! [`model::resolve`] call, so the asserted bodies/heat ARE the pixels.
pub mod model;
pub use model::{
CostMetric, Metrics, OpStatus, PlanEdge, PlanModel, PlanNode, PlanVersion, Resolved,
ResolvedNode, resolve,
};
use egui::{
Align2, Color32, CornerRadius, FontId, Pos2, Rect, Sense, Stroke, StrokeKind, Vec2, WidgetType,
};
use facett_core::layered::{Align, LayeredOpts, RowOrder, layered, place};
use facett_core::nav::{NavGesture, Navigable};
use facett_core::{Elm, Facet, FacetCaps, a11y, theme};
use facett_flow::attr::{Sizes, Status as FlowStatus, StatusTable};
use facett_flow::{Animation, CpuStyle, Edge, FlowScene, Graph, Node, PaintOpts, View};
use serde::{Deserialize, Serialize};
const NODE_W: f32 = 184.0;
const NODE_H: f32 = 84.0;
const COL_GAP: f32 = 240.0;
const ROW_GAP: f32 = 110.0;
const LEFT_PAD: f32 = 24.0;
const TOP_PAD: f32 = 16.0;
/// Which plan version is shown (the AQE toggle).
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
pub enum Version {
/// The plan Catalyst produced before runtime stats (`AdaptiveSparkPlan`, initial).
#[default]
Initial,
/// The post-AQE plan (`isFinalPlan = true`) — coalesced shuffles, SMJ→BHJ, ….
Final,
}
impl Version {
pub fn key(self) -> &'static str {
match self {
Version::Initial => "initial",
Version::Final => "final",
}
}
}
/// A robot-/CLI-addressable control message. Applied by [`PlanDagView::update`].
#[non_exhaustive]
#[derive(Clone, Debug, PartialEq)]
pub enum Msg {
Pan(Vec2),
Zoom {
factor: f32,
anchor: Pos2,
},
Fit(Rect),
RequestRefit,
/// Switch the cost-heat metric (Time / Memory / Rows).
SetMetric(CostMetric),
/// Flip the AQE plan version (no-op to Final if the model has no final plan).
SetVersion(Version),
/// Collapse / expand one WholeStageCodegen group by its id.
ToggleGroup(String),
/// Collapse every group.
CollapseAll,
/// Expand every group.
ExpandAll,
/// Select a resolved node by its id.
Select(String),
/// Clear the selection.
ClearSelection,
/// Toggle the constraint overlay (bottleneck badge · downstream emphasis ·
/// critical-path highlight).
SetConstraints(bool),
}
/// Side work as data (FC-8). Uninhabited on purpose: this component renders a
/// host-fed plan and performs **no** I/O, so it has no effect to emit.
#[derive(Clone, Debug, PartialEq)]
pub enum Effect {}
/// **FC-3** — the complete observable state of the view in one serializable,
/// round-trippable struct: the injected [`PlanModel`] plus the mutable controls
/// (active version, cost metric, collapse set, selection). The **only** thing that
/// mutates it is [`PlanDagView::update`].
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct PlanDagState {
/// The plan data (initial + optional AQE-final version).
pub model: PlanModel,
/// The AQE plan version currently shown.
pub version: Version,
/// The active cost-heat metric.
pub metric: CostMetric,
/// Collapsed WholeStageCodegen group ids.
pub collapsed: Vec<String>,
/// Selected resolved-node id (stable key, FC-5).
pub selected: Option<String>,
/// Constraint overlay on: badge the top bottleneck, weight node emphasis by
/// downstream reach, highlight the critical-path chain. Subtle + always-on by
/// default; the header toggle turns it off. `#[serde(default)]` so pre-overlay
/// serialized states still load (defaulting the overlay on).
#[serde(default = "default_show_constraints")]
pub show_constraints: bool,
#[serde(default)]
pub nav: Navigable,
#[serde(default)]
pub fitted: bool,
}
fn default_show_constraints() -> bool {
true
}
/// The metric-bearing operator-DAG [`Facet`].
pub struct PlanDagView {
pub title: String,
state: PlanDagState,
scene: FlowScene,
}
impl PlanDagView {
pub fn new(title: impl Into<String>, model: PlanModel) -> Self {
let resolved = resolve(&model.initial, &[], CostMetric::Time);
let scene = Self::build_scene(&resolved);
Self {
title: title.into(),
state: PlanDagState {
model,
version: Version::Initial,
metric: CostMetric::Time,
collapsed: Vec::new(),
selected: None,
show_constraints: true,
nav: Navigable::default(),
fitted: false,
},
scene,
}
}
fn build_scene(resolved: &Resolved) -> FlowScene {
let edges: Vec<_> = resolved
.edges
.iter()
.map(|&(a, b)| (a as u32, b as u32))
.collect();
let layout = layered(
resolved.nodes.len(),
&edges,
LayeredOpts {
align: Align::Sources,
rows: RowOrder::Input,
},
);
let positions = place(
&layout,
[COL_GAP, ROW_GAP],
[LEFT_PAD + NODE_W / 2.0, TOP_PAD + NODE_H / 2.0],
);
let graph = Graph {
nodes: resolved
.nodes
.iter()
.enumerate()
.map(|(i, n)| Node {
id: format!("p{i}"),
label: n.title.clone(),
position: positions[i],
group: "plan".into(),
kind: 0,
})
.collect(),
edges: resolved
.edges
.iter()
.map(|&(a, b)| Edge {
from: format!("p{a}"),
to: format!("p{b}"),
})
.collect(),
};
let mut scene = FlowScene::with_key(graph, 0).expect("resolved plan graph");
scene
.set_sizes(Sizes {
nodes: vec![[NODE_W, NODE_H]; resolved.nodes.len()],
})
.expect("plan sizes");
scene
.set_status(StatusTable {
nodes: resolved
.nodes
.iter()
.map(|n| match n.status {
OpStatus::Done => Some(FlowStatus::Ok),
OpStatus::Failed => Some(FlowStatus::Error),
_ => None,
})
.collect(),
edges: vec![None; resolved.edges.len()],
})
.expect("plan statuses");
scene.set_arrows(true);
scene
}
pub fn flow_scene(&self) -> &FlowScene {
&self.scene
}
pub fn nav(&self) -> &Navigable {
&self.state.nav
}
/// **FC-3** — read the complete observable state at any frame boundary.
pub fn state(&self) -> &PlanDagState {
&self.state
}
/// The plan data (initial + optional final version).
pub fn model(&self) -> &PlanModel {
&self.state.model
}
/// The plan version currently shown (falls back to initial if Final is asked for
/// but the model has none).
pub fn active_version(&self) -> &PlanVersion {
match self.state.version {
Version::Final => self
.state
.model
.final_plan
.as_ref()
.unwrap_or(&self.state.model.initial),
Version::Initial => &self.state.model.initial,
}
}
/// Resolve the active version under the current collapse set + cost metric — the
/// single source of truth for the painter and `state_json`.
pub fn resolved(&self) -> Resolved {
resolve(
self.active_version(),
&self.state.collapsed,
self.state.metric,
)
}
/// **FC-2 / FC-8** — the single mutation path. Apply one [`Msg`]; returns the
/// (always empty) [`Effect`]s the host should run.
pub fn update(&mut self, msg: Msg) -> Vec<Effect> {
let geometry_changed = matches!(
&msg,
Msg::SetVersion(_) | Msg::ToggleGroup(_) | Msg::CollapseAll | Msg::ExpandAll
);
match msg {
Msg::Pan(delta) => self.state.nav.pan(delta),
Msg::Zoom { factor, anchor } => self.state.nav.zoom_to(factor, anchor),
Msg::Fit(rect) => {
self.state.nav.fit(self.scene.bounds(), rect, 0.08);
self.state.fitted = true;
}
Msg::RequestRefit => self.state.fitted = false,
Msg::SetMetric(m) => self.state.metric = m,
Msg::SetVersion(v) => {
// Asking for Final with no final plan is a no-op.
if v != Version::Final || self.state.model.has_aqe() {
self.state.version = v;
}
}
Msg::ToggleGroup(g) => {
if let Some(i) = self.state.collapsed.iter().position(|c| *c == g) {
self.state.collapsed.remove(i);
} else {
self.state.collapsed.push(g);
}
}
Msg::CollapseAll => self.state.collapsed = self.active_version().groups(),
Msg::ExpandAll => self.state.collapsed.clear(),
Msg::Select(id) => {
let exists = self.resolved().nodes.iter().any(|n| n.id == id);
if exists {
self.state.selected = Some(id);
}
}
Msg::ClearSelection => {
self.state.selected = None;
}
Msg::SetConstraints(on) => self.state.show_constraints = on,
}
if geometry_changed {
self.scene = Self::build_scene(&self.resolved());
self.state.fitted = false;
}
Vec::new()
}
// Read accessors (handy for hosts / tests).
pub fn metric(&self) -> CostMetric {
self.state.metric
}
pub fn version(&self) -> Version {
self.state.version
}
pub fn is_collapsed(&self, group: &str) -> bool {
self.state.collapsed.iter().any(|c| c == group)
}
pub fn show_constraints(&self) -> bool {
self.state.show_constraints
}
/// The cost-heat fill colour for a heat fraction `[0,1]`: cool grey-blue → amber
/// → hot red (the Query-Profile hotspot ramp).
fn heat_color(h: f32) -> Color32 {
use facett_core::look::color::lerp_color as lerp;
let cool = Color32::from_rgb(70, 90, 120);
let amber = Color32::from_rgb(220, 160, 60);
let hot = Color32::from_rgb(220, 70, 60);
if h < 0.5 {
lerp(cool, amber, h * 2.0)
} else {
lerp(amber, hot, (h - 0.5) * 2.0)
}
}
/// Status → themed colour. Mirrors `facett_pipeline::Status::theme_color`
/// (done/running ride the palette's `point`/`accent`, failed a warm red, idle
/// dimmed) — inlined to keep this crate off the cross-repo pipeline dep cycle.
fn op_status_color(s: OpStatus, th: &facett_core::Theme) -> Color32 {
match s {
OpStatus::Done => th.point,
OpStatus::Running => th.accent,
OpStatus::Failed => facett_core::look::RED,
OpStatus::Idle => th.text_dim,
}
}
}
/// Short human byte count.
fn human_bytes(b: u64) -> String {
const U: [&str; 5] = ["B", "K", "M", "G", "T"];
let mut v = b as f64;
let mut i = 0;
while v >= 1024.0 && i < U.len() - 1 {
v /= 1024.0;
i += 1;
}
if i == 0 {
format!("{b} B")
} else {
format!("{v:.1}{}", U[i])
}
}
/// Short human row/count.
fn human_count(n: u64) -> String {
if n >= 1_000_000 {
format!("{:.1}M", n as f64 / 1e6)
} else if n >= 1_000 {
format!("{:.1}K", n as f64 / 1e3)
} else {
n.to_string()
}
}
impl PlanDagView {
/// A self-contained **demo plan** with an AQE re-plan: the initial plan has a
/// sort-merge join behind an Exchange; the final (post-AQE) plan coalesced the
/// shuffle and flipped to a broadcast hash join — and a WholeStageCodegen group
/// to collapse. Exercises every capability (metric body, cost-heat, collapse,
/// AQE toggle, Photon) with no host.
pub fn local() -> Self {
let m = |t: f64, mem: u64, rows: u64, spill: u64| Metrics {
time_ms: t,
peak_mem: mem,
rows,
spill,
};
// initial: scan → filter →[wscg 1] hashagg → exchange → smj
let initial = PlanVersion::new(
"initial",
vec![
PlanNode::new("scan", "BatchScan")
.photon()
.metrics(m(40.0, 1 << 20, 2_000_000, 0)),
PlanNode::new("filter", "Filter")
.group("1")
.photon()
.metrics(m(25.0, 1 << 18, 700_000, 0)),
PlanNode::new("hashagg", "HashAggregate")
.group("1")
.metrics(m(280.0, 1 << 24, 12_000, 1 << 21)),
PlanNode::new("exchange", "Exchange")
.status(OpStatus::Running)
.metrics(m(160.0, 1 << 20, 12_000, 0)),
PlanNode::new("smj", "SortMergeJoin").metrics(m(210.0, 1 << 23, 12_000, 1 << 20)),
],
vec![
PlanEdge { from: 0, to: 1 },
PlanEdge { from: 1, to: 2 },
PlanEdge { from: 2, to: 3 },
PlanEdge { from: 3, to: 4 },
],
);
// final: AQE coalesced the shuffle + flipped SMJ → BroadcastHashJoin.
let final_plan = PlanVersion::new(
"final",
vec![
PlanNode::new("scan", "BatchScan")
.photon()
.metrics(m(40.0, 1 << 20, 2_000_000, 0)),
PlanNode::new("filter", "Filter")
.group("1")
.photon()
.metrics(m(25.0, 1 << 18, 700_000, 0)),
PlanNode::new("hashagg", "HashAggregate")
.group("1")
.metrics(m(280.0, 1 << 24, 12_000, 0)),
PlanNode::new("aqeread", "AQEShuffleRead").metrics(m(30.0, 1 << 18, 12_000, 0)),
PlanNode::new("bhj", "BroadcastHashJoin").metrics(m(60.0, 1 << 22, 12_000, 0)),
],
vec![
PlanEdge { from: 0, to: 1 },
PlanEdge { from: 1, to: 2 },
PlanEdge { from: 2, to: 3 },
PlanEdge { from: 3, to: 4 },
],
);
Self::new("plandag", PlanModel::new(initial).with_final(final_plan))
}
}
impl PlanDagView {
/// The copyable text: the selected node's `title status heat` line when a node
/// is selected, else the resolved node list as a TSV rectangle
/// (`id \t title \t status \t heat`).
pub fn copy_text(&self) -> Option<String> {
let resolved = self.resolved();
if resolved.nodes.is_empty() {
return None;
}
if let Some(n) = self
.state
.selected
.as_deref()
.and_then(|id| resolved.nodes.iter().find(|n| n.id == id))
{
return Some(format!("{}\t{}\t{:.3}", n.title, n.status.label(), n.heat));
}
let mut out = String::from("id\ttitle\tstatus\theat");
for n in &resolved.nodes {
out.push('\n');
out.push_str(&format!(
"{}\t{}\t{}\t{:.3}",
n.id,
n.title,
n.status.label(),
n.heat
));
}
Some(out)
}
}
// ── typed copy (§16) — read-only: selected node or resolved node-list rows ─────
impl facett_core::clip::CopySource for PlanDagView {
fn copy_kinds(&self) -> &[facett_core::clip::ClipKind] {
use facett_core::clip::ClipKind;
&[ClipKind::Rows, ClipKind::Text]
}
fn copy_payload(&self) -> Option<facett_core::clip::ClipPayload> {
self.copy_text().map(facett_core::clip::ClipPayload::Rows)
}
}
impl PlanDagView {
/// **FC-9 render** — a **pure** function of `&self`: it paints the header
/// controls + the operator DAG and *returns* the [`Msg`]s the interactions
/// produced. It MUST NOT mutate the model; [`Facet::ui`] applies the returned
/// messages through [`update`](Self::update). This is the same body the old
/// `&mut self` `ui()` ran, with every inline `self.update(..)` replaced by a
/// `msgs.push(..)` — the only change that makes the view pure.
pub fn render(&self, ui: &mut egui::Ui) -> Vec<Msg> {
let mut msgs: Vec<Msg> = Vec::new();
let th = theme(ui);
// ── header: cost-heat switch · AQE toggle · collapse controls ────────────
ui.horizontal_wrapped(|ui| {
if ui.button("⤢ Fit").on_hover_text("Frame the plan (F / Ctrl-0)").clicked() { msgs.push(Msg::RequestRefit); }
ui.label(egui::RichText::new("cost-heat:").color(th.text_dim));
for m in CostMetric::ALL {
let sel = self.state.metric == m;
if ui.selectable_label(sel, m.label()).clicked() {
msgs.push(Msg::SetMetric(m));
}
}
if self.state.model.has_aqe() {
ui.separator();
ui.label(egui::RichText::new("plan:").color(th.text_dim));
for (v, lbl) in [(Version::Initial, "initial"), (Version::Final, "final (AQE)")] {
if ui.selectable_label(self.state.version == v, lbl).clicked() {
msgs.push(Msg::SetVersion(v));
}
}
}
ui.separator();
let sc = self.state.show_constraints;
if ui
.selectable_label(sc, "⚑ constraints")
.on_hover_text("badge the top bottleneck · emphasise nodes by downstream reach · highlight the critical path")
.clicked()
{
msgs.push(Msg::SetConstraints(!sc));
}
});
let groups = self.active_version().groups();
if !groups.is_empty() {
ui.horizontal_wrapped(|ui| {
ui.label(egui::RichText::new("WholeStageCodegen:").color(th.text_dim));
for g in &groups {
let mut on = self.is_collapsed(g);
if ui
.checkbox(&mut on, format!("[{g}]"))
.on_hover_text("collapse this codegen group")
.changed()
{
msgs.push(Msg::ToggleGroup(g.clone()));
}
}
if ui.small_button("collapse all").clicked() {
msgs.push(Msg::CollapseAll);
}
if ui.small_button("expand all").clicked() {
msgs.push(Msg::ExpandAll);
}
});
}
ui.separator();
// Resolve metric overlays; persistent Falsett geometry already holds layout.
let resolved = self.resolved();
if resolved.nodes.is_empty() {
ui.label("empty plan");
return msgs;
}
let (rect, resp) = ui.allocate_exact_size(
ui.available_size().max(Vec2::new(120.0, 320.0)),
Sense::click_and_drag(),
);
if !self.state.fitted {
msgs.push(Msg::Fit(rect));
}
msgs.extend(
facett_core::nav::gestures(ui, rect, &resp)
.into_iter()
.map(|g| match g {
NavGesture::Pan(d) => Msg::Pan(d),
NavGesture::Zoom { factor, anchor } => Msg::Zoom { factor, anchor },
NavGesture::Fit => Msg::Fit(rect),
}),
);
let painter = ui.painter_at(rect);
painter.rect_filled(rect, CornerRadius::ZERO, th.bg);
let view = View::from_navigable(&self.state.nav, rect);
let painted = facett_flow::paint(
ui,
ui.id().with("plandag-flow"),
rect,
&self.scene,
&view,
&PaintOpts {
cpu: CpuStyle::Full {
max_edges: usize::MAX,
},
animation: Animation::None,
..Default::default()
},
);
let scale = self.state.nav.scale;
// ── constraint overlay (bottleneck badge · downstream emphasis · critical
// path) — computed once from the SAME resolved graph as `state_json`, so
// the badged/emphasised pixels ARE the asserted data. Off ⇒ empty. ──────
let show = self.state.show_constraints;
let bottlenecks = if show {
resolved.bottlenecks()
} else {
Vec::new()
};
let top_bottleneck = if show {
resolved.top_bottleneck()
} else {
None
};
let crit = if show {
resolved.critical_path()
} else {
Vec::new()
};
let id_to_idx: std::collections::HashMap<&str, usize> = resolved
.nodes
.iter()
.enumerate()
.map(|(i, n)| (n.id.as_str(), i))
.collect();
let downstream_of: std::collections::HashMap<&str, usize> = bottlenecks
.iter()
.map(|b| (b.id.as_str(), b.downstream))
.collect();
let max_downstream = bottlenecks.iter().map(|b| b.downstream).max().unwrap_or(0);
let crit_edges: std::collections::HashSet<(usize, usize)> = crit
.windows(2)
.filter_map(|w| {
Some((
*id_to_idx.get(w[0].as_str())?,
*id_to_idx.get(w[1].as_str())?,
))
})
.collect();
for (ei, &(a, b)) in resolved.edges.iter().enumerate() {
if crit_edges.contains(&(a, b)) {
let pts: Vec<_> = (0..=16)
.filter_map(|j| self.scene.edge_point(ei, j as f32 / 16.0))
.map(|p| view.screen(p))
.collect();
painter.add(egui::Shape::line(
pts,
Stroke::new(4.0 * scale, th.accent.gamma_multiply(0.45)),
));
}
}
// nodes — heat fill + metric body + status/photon border + a11y node.
let base_id = ui.id().with("plandag-nodes");
// Reserve semantic slots before input-order capping so an active node
// cannot disappear from the accessibility tree in a crowded viewport.
let focused = ui.memory(|memory| memory.focused());
let priority: Vec<usize> = painted
.visible_nodes
.iter()
.filter_map(|shown| {
let key = &resolved.nodes[shown.node].id;
(self.state.selected.as_deref() == Some(key.as_str())
|| focused == Some(facett_core::a11y::stable_id(base_id, key)))
.then_some(shown.node)
})
.collect();
let mut semantic_slots = 2_000usize.saturating_sub(priority.len());
let mut clicked: Option<String> = None;
for visible in &painted.visible_nodes {
let n = &resolved.nodes[visible.node];
let nrect = visible.rect;
let p = nrect.min;
let is_sel = self.state.selected.as_deref() == Some(n.id.as_str());
// downstream-reach emphasis: an outer accent ring whose weight/alpha grows
// with how much work is stalled behind this node (subtle, drawn outside).
let downstream = if show {
downstream_of.get(n.id.as_str()).copied().unwrap_or(0)
} else {
0
};
if max_downstream > 0 && downstream > 0 {
let frac = downstream as f32 / max_downstream as f32;
let ring = nrect.expand(2.0 + 4.0 * frac);
let a = (40.0 + 130.0 * frac) as u8;
painter.rect_stroke(
ring,
CornerRadius::same(9),
Stroke::new(
1.5_f32,
Color32::from_rgba_unmultiplied(
th.accent.r(),
th.accent.g(),
th.accent.b(),
a,
),
),
StrokeKind::Outside,
);
}
// cost-heat fill over the node bg.
painter.rect_filled(nrect, CornerRadius::same(6), th.node_fill);
let heat = Self::heat_color(n.heat);
let alpha = (0.12 + 0.6 * n.heat).clamp(0.0, 1.0);
painter.rect_filled(
nrect,
CornerRadius::same(6),
Color32::from_rgba_unmultiplied(
heat.r(),
heat.g(),
heat.b(),
(alpha * 255.0) as u8,
),
);
// border: Photon → purple, else status colour; selected → accent.
let (bw, bc) = if is_sel {
(3.0_f32, th.accent)
} else if n.photon {
(2.0_f32, Color32::from_rgb(170, 120, 220))
} else {
(2.0_f32, Self::op_status_color(n.status, &th))
};
painter.rect_stroke(
nrect,
CornerRadius::same(6),
Stroke::new(bw, bc),
StrokeKind::Inside,
);
// top-bottleneck badge — the "clear this next" constraint (the single
// node that unblocks the most downstream work). A warm flag pinned on the
// top edge + a stronger border, plus its own AccessKit node so a robot can
// `get_by_label("⚑ bottleneck · <title>")` and read the downstream reach.
if show && top_bottleneck.as_deref() == Some(n.id.as_str()) {
let badge_c = Color32::from_rgb(240, 180, 40);
painter.rect_stroke(
nrect,
CornerRadius::same(6),
Stroke::new(3.0_f32, badge_c),
StrokeKind::Inside,
);
let center = Pos2::new(nrect.center().x, nrect.top());
painter.circle_filled(center, 9.0, badge_c);
painter.text(
center,
Align2::CENTER_CENTER,
"⚑",
FontId::proportional(11.0 * scale),
Color32::from_rgb(30, 25, 10),
);
let brect = Rect::from_center_size(center, Vec2::splat(18.0));
let reach = downstream_of.get(n.id.as_str()).copied().unwrap_or(0) as f64;
let sem =
a11y::Semantics::new(WidgetType::Label, format!("⚑ bottleneck · {}", n.title))
.value(reach);
let _ = a11y::node(
ui,
base_id.with("bottleneck"),
&n.id,
Sense::hover(),
brect,
sem,
);
}
// title row (+ collapse glyph for a super-node).
let title = if n.collapsed {
format!("▣ {}", n.title)
} else {
n.title.clone()
};
painter.text(
p + Vec2::new(8.0, 7.0) * scale,
Align2::LEFT_TOP,
&title,
FontId::proportional(13.0 * scale),
th.text,
);
if n.photon {
painter.text(
p + Vec2::new(NODE_W - 8.0, 7.0) * scale,
Align2::RIGHT_TOP,
"⚡",
FontId::proportional(12.0 * scale),
Color32::from_rgb(190, 150, 235),
);
}
// metric body — the multi-line operator readout.
let body = [
format!("time {:.0} ms", n.metrics.time_ms),
format!("mem {}", human_bytes(n.metrics.peak_mem)),
format!("rows {}", human_count(n.metrics.rows)),
];
let mut y = 28.0;
for line in &body {
painter.text(
p + Vec2::new(8.0, y) * scale,
Align2::LEFT_TOP,
line,
FontId::monospace(11.0 * scale),
th.text_dim,
);
y += 14.0;
}
if n.metrics.spill > 0 {
painter.text(
p + Vec2::new(NODE_W - 8.0, NODE_H - 16.0) * scale,
Align2::RIGHT_TOP,
format!("spill {}", human_bytes(n.metrics.spill)),
FontId::monospace(10.0 * scale),
Color32::from_rgb(224, 120, 90),
);
}
// Label stays stable (heat rides as the node's numeric value, queryable
// separately) so a robot/test can resolve a node by `title · status`.
let sem = a11y::Semantics::button(format!("{} · {}", n.title, n.status.label()))
.selected(is_sel)
.value(n.heat as f64);
let semantic = priority.contains(&visible.node)
|| if semantic_slots > 0 {
semantic_slots -= 1;
true
} else {
false
};
let resp = if semantic {
a11y::node(ui, base_id, &n.id, Sense::click(), nrect, sem)
} else {
ui.interact(nrect, base_id.with(&n.id), Sense::click())
};
if resp.clicked() {
clicked = Some(n.id.clone());
}
// double-click a collapsed super-node expands it (+ a hover hint).
if n.collapsed
&& let Some(g) = &n.group
{
if resp.double_clicked() {
msgs.push(Msg::ToggleGroup(g.clone()));
}
resp.on_hover_text(format!(
"WholeStageCodegen[{g}] — {} ops (double-click to expand)",
n.members
));
}
}
if let Some(id) = clicked {
// Toggle: clicking the open node clears it, else select it. Reads the
// CURRENT selection (`&self`) — no mutation happens in the pure view.
if self.state.selected.as_deref() == Some(id.as_str()) {
msgs.push(Msg::ClearSelection);
} else {
msgs.push(Msg::Select(id));
}
}
// selected-node detail line.
if let Some(id) = self.state.selected.clone()
&& let Some(n) = resolved.nodes.iter().find(|n| n.id == id)
{
ui.separator();
ui.horizontal_wrapped(|ui| {
ui.strong(&n.title);
ui.colored_label(Self::op_status_color(n.status, &th), n.status.label());
ui.label(
egui::RichText::new(format!(
"{:.0} ms · {} · {} rows",
n.metrics.time_ms,
human_bytes(n.metrics.peak_mem),
human_count(n.metrics.rows)
))
.monospace()
.color(th.text_dim),
);
if n.metrics.spill > 0 {
ui.colored_label(
Color32::from_rgb(224, 120, 90),
format!("spill {}", human_bytes(n.metrics.spill)),
);
}
if n.photon {
ui.colored_label(Color32::from_rgb(190, 150, 235), "⚡ Photon");
}
});
}
#[cfg(feature = "testmatrix")]
facett_core::testmatrix::emit(
"facett-plandag::PlanDagView::render",
"ui_render",
!resolved.nodes.is_empty(),
&format!(
"metric={} version={} nodes={} constraints={} top_bottleneck={:?} critical_path_len={}",
self.state.metric.key(),
self.state.version.key(),
resolved.nodes.len(),
self.state.show_constraints,
top_bottleneck,
crit.len(),
),
);
msgs
}
}
// ── FC-2/FC-9 canonical trait: PlanDagView IS an `Elm` component ──────────────
//
// The Model/Msg/Effect/pure-view split is implemented on the inherent methods; the
// `Elm` impl delegates to them verbatim.
impl Elm for PlanDagView {
type Model = PlanDagState;
type Msg = Msg;
type Effect = Effect;
fn title(&self) -> &str {
&self.title
}
fn state(&self) -> &PlanDagState {
&self.state
}
fn update(&mut self, msg: Msg) -> Vec<Effect> {
PlanDagView::update(self, msg)
}
fn view(&self, ui: &mut egui::Ui) -> Vec<Msg> {
facett_core::navgraph::declare_own_draw_order(ui); // atoms follow the data order (navgraph §9)
self.render(ui)
}
}
// The hand-written `impl Facet` is intentionally KEPT (not replaced by
// `impl_facet_via_elm!`) because this view publishes a **richer** `state_json`
// (the resolved graph + per-node cost-heat as data + version/collapse/selection)
// plus `selection_json` / `portable_state` / `load_state` / typed `copy` — the
// macro emits only the plain `serde(state())` default and cannot override it. Same
// call the `facett-security` reference makes for the same reason (see
// `facett-core/src/elm.rs`). The `ui` below IS the macro's FC-9 bridge, verbatim.
impl Facet for PlanDagView {
fn title(&self) -> &str {
&self.title
}
fn copy(&mut self) -> Option<String> {
use facett_core::clip::CopySource as _;
self.copy_payload().map(|p| p.as_text())
}
fn caps(&self) -> FacetCaps {
FacetCaps::NONE
.themeable()
.selectable()
.resizable()
.copyable()
.navigable()
}
/// FC-9: the view is pure (`render` paints + emits `Msg`s); FC-2: mutation only
/// via `update`. The immediate-mode trait just bridges the two.
fn ui(&mut self, ui: &mut egui::Ui) {
facett_core::navgraph::declare_own_draw_order(ui); // atoms follow the data order (navgraph §9)
for m in self.render(ui) {
let _ = self.update(m);
}
}
fn state_json(&self) -> serde_json::Value {
let resolved = self.resolved();
let mut graph = resolved.state_json(self.state.metric);
if let Some(obj) = graph.as_object_mut() {
obj.insert("nav".into(), serde_json::json!(self.state.nav));
obj.insert("fitted".into(), serde_json::json!(self.state.fitted));
obj.insert("title".into(), serde_json::json!(self.title));
obj.insert(
"version".into(),
serde_json::json!(self.state.version.key()),
);
obj.insert(
"has_aqe".into(),
serde_json::json!(self.state.model.has_aqe()),
);
obj.insert("collapsed".into(), serde_json::json!(self.state.collapsed));
obj.insert(
"groups".into(),
serde_json::json!(self.active_version().groups()),
);
obj.insert("selected".into(), serde_json::json!(self.state.selected));
obj.insert(
"show_constraints".into(),
serde_json::json!(self.state.show_constraints),
);
}
graph
}
fn selection_json(&self) -> serde_json::Value {
let resolved = self.resolved();
match self
.state
.selected
.as_deref()
.and_then(|id| resolved.nodes.iter().find(|n| n.id == id))
{
Some(n) => {
serde_json::json!({ "node": n.id, "title": n.title, "heat": n.heat, "status": n.status.label() })
}
None => serde_json::Value::Null,
}
}
fn kind(&self) -> &'static str {
"plandag"
}
fn portable_state(&self) -> Option<serde_json::Value> {
Some(serde_json::json!({
"metric": self.state.metric.key(),
"version": self.state.version.key(),
"collapsed": self.state.collapsed,
"selected": self.state.selected,
"show_constraints": self.state.show_constraints,
"nav": self.state.nav,
"fitted": self.state.fitted,
}))
}
fn load_state(&mut self, state: &serde_json::Value) -> bool {
let mut changed = false;
if let Some(k) = state.get("metric").and_then(|v| v.as_str()) {
self.state.metric = CostMetric::ALL
.iter()
.copied()
.find(|m| m.key() == k)
.unwrap_or(self.state.metric);
changed = true;
}
if let Some(k) = state.get("version").and_then(|v| v.as_str()) {
self.state.version = if k == "final" {
Version::Final
} else {
Version::Initial
};
changed = true;
}
if let Some(arr) = state.get("collapsed").and_then(|v| v.as_array()) {
self.state.collapsed = arr
.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect();
changed = true;
}
if let Some(sel) = state.get("selected") {
self.state.selected = sel.as_str().map(|s| s.to_string());
changed = true;
}
if let Some(sc) = state.get("show_constraints").and_then(|v| v.as_bool()) {
self.state.show_constraints = sc;
changed = true;
}
if changed {
self.scene = Self::build_scene(&self.resolved());
self.state.fitted = false;
}
if let Some(nav) = state
.get("nav")
.and_then(|v| serde_json::from_value::<Navigable>(v.clone()).ok())
&& nav.scale.is_finite()
&& nav.scale > 0.0
&& nav.min_scale.is_finite()
&& nav.min_scale > 0.0
&& nav.max_scale.is_finite()
&& nav.max_scale >= nav.min_scale
&& nav.offset.iter().all(|v| v.is_finite())
{
self.state.nav = nav;
self.state.fitted = state
.get("fitted")
.and_then(|v| v.as_bool())
.unwrap_or(false);
changed = true;
}
changed
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn typed_copy_is_selected_node_or_resolved_rows() {
use facett_core::clip::{ClipKind, CopySource};
let mut v = PlanDagView::local();
// No selection -> resolved node list as a TSV rectangle.
let p = v.copy_payload().expect("resolved nodes copy");
assert_eq!(p.kind(), ClipKind::Rows);
assert!(
p.as_text().starts_with("id\ttitle\tstatus\theat"),
"{}",
p.as_text()
);
// Select the first resolved node -> copy narrows to its line.
let first_id = v.resolved().nodes[0].id.clone();
v.update(Msg::Select(first_id.clone()));
let sel = v.copy_payload().unwrap().as_text();
assert_eq!(
sel.split('\t').count(),
3,
"selected copy is one node line: {sel}"
);
assert!(!sel.contains('\n'), "single line: {sel}");
}
#[test]
fn cost_metric_switch_is_observable() {
let mut v = PlanDagView::local();
assert_eq!(v.metric(), CostMetric::Time);
// FC-8: applying a Msg yields no effects (this component does no I/O).
assert!(v.update(Msg::SetMetric(CostMetric::Rows)).is_empty());
assert_eq!(v.metric(), CostMetric::Rows);
assert_eq!(v.state_json()["metric"], "rows");
}
#[test]
fn collapse_group_reduces_node_count() {
let mut v = PlanDagView::local();
let before = v.state_json()["node_count"].as_u64().unwrap();
v.update(Msg::ToggleGroup("1".into()));
let after = v.state_json()["node_count"].as_u64().unwrap();
assert_eq!(
after,
before - 1,
"collapsing WSCG[1] folds two ops into one super-node"
);
let st = v.state_json();
assert_eq!(st["collapsed"], serde_json::json!(["1"]));
assert!(
st["nodes"]
.as_array()
.unwrap()
.iter()
.any(|n| n["collapsed"] == true),
"a collapsed super-node is present"
);
}
#[test]
fn aqe_toggle_swaps_the_plan() {
let mut v = PlanDagView::local();
assert_eq!(v.state_json()["version"], "initial");
// initial plan has SortMergeJoin; final has BroadcastHashJoin.
let titles_init: Vec<String> = v.resolved().nodes.iter().map(|n| n.title.clone()).collect();
assert!(titles_init.iter().any(|t| t.contains("SortMergeJoin")));
v.update(Msg::SetVersion(Version::Final));
assert_eq!(v.state_json()["version"], "final");
let titles_final: Vec<String> =
v.resolved().nodes.iter().map(|n| n.title.clone()).collect();
assert!(
titles_final.iter().any(|t| t.contains("BroadcastHashJoin")),
"AQE final flipped SMJ → BHJ"
);
assert!(!titles_final.iter().any(|t| t.contains("SortMergeJoin")));
}
#[test]
fn nodes_carry_metric_bodies_and_heat_as_data() {
let v = PlanDagView::local();
let st = v.state_json();
let nodes = st["nodes"].as_array().unwrap();
assert!(!nodes.is_empty());
for n in nodes {
// every node has a metric body...
assert!(n["metrics"]["time_ms"].is_number());
assert!(n["metrics"]["rows"].is_number());
// ...and a cost-heat fraction in [0,1] (the fill, as DATA).
let h = n["heat"].as_f64().unwrap();
assert!((0.0..=1.0).contains(&h), "heat in range");
}
// exactly one node is the hottest (heat == 1.0) for the active metric.
assert!(
nodes
.iter()
.any(|n| (n["heat"].as_f64().unwrap() - 1.0).abs() < 1e-6),
"a hottest node exists"
);
}
#[test]
fn final_version_no_op_without_aqe() {
let mut v = PlanDagView::new(
"p",
PlanModel::new(PlanDagView::local().model().initial.clone()),
);
assert!(!v.model().has_aqe());
v.update(Msg::SetVersion(Version::Final));
assert_eq!(
v.version(),
Version::Initial,
"no final plan → toggle is a no-op"
);
}
#[test]
fn portable_state_round_trips() {
let mut a = PlanDagView::local();
a.update(Msg::SetMetric(CostMetric::Memory));
a.update(Msg::SetVersion(Version::Final));
a.update(Msg::ToggleGroup("1".into()));
let p = a.portable_state().unwrap();
let mut b = PlanDagView::local();
assert!(b.load_state(&p));
assert_eq!(b.metric(), CostMetric::Memory);
assert_eq!(b.version(), Version::Final);
assert!(b.is_collapsed("1"));
}
// ── FC-2 → FC-3 as a tested property: drive Msg sequences through the headless
// `facett_core::harness` (no egui, no GPU) and assert the resulting Model. ──
#[test]
fn harness_drives_metric_version_and_collapse_transitions() {
use facett_core::harness;
let mut v = PlanDagView::local();
// metric switch + AQE flip + one group collapse, in one Msg sequence.
let snap = harness::snapshot(
&mut v,
[
Msg::SetMetric(CostMetric::Memory),
Msg::SetVersion(Version::Final),
Msg::ToggleGroup("1".into()),
],
);
assert_eq!(snap.metric, CostMetric::Memory);
assert_eq!(snap.version, Version::Final);
assert_eq!(snap.collapsed, vec!["1".to_string()]);
// ExpandAll clears the collapse set.
let snap = harness::snapshot(&mut v, [Msg::ExpandAll]);
assert!(snap.collapsed.is_empty());
// CollapseAll re-collapses every group in the (now Final) active version.
let snap = harness::snapshot(&mut v, [Msg::CollapseAll]);
assert_eq!(snap.collapsed, v.active_version().groups());
}
#[test]
fn harness_node_selection_transitions() {
use facett_core::harness;
let mut v = PlanDagView::local();
let first = v.resolved().nodes[0].id.clone();
// Select an existing resolved node → it becomes the selection.
let snap = harness::snapshot(&mut v, [Msg::Select(first.clone())]);
assert_eq!(snap.selected.as_deref(), Some(first.as_str()));
// Selecting a non-existent id is ignored (selection unchanged).
let snap = harness::snapshot(&mut v, [Msg::Select("node:nope".into())]);
assert_eq!(snap.selected.as_deref(), Some(first.as_str()));
// ClearSelection empties it.
let snap = harness::snapshot(&mut v, [Msg::ClearSelection]);
assert!(snap.selected.is_none());
}
#[test]
fn harness_drive_yields_no_effects_fc8() {
use facett_core::harness;
let mut v = PlanDagView::local();
let effects = harness::drive(
&mut v,
[
Msg::SetMetric(CostMetric::Rows),
Msg::CollapseAll,
Msg::ExpandAll,
],
);
assert!(
effects.is_empty(),
"plandag performs no I/O — `Effect` is uninhabited"
);
}
#[test]
fn model_round_trips_through_serde_fc3() {
use facett_core::harness;
let mut v = PlanDagView::local();
let model = harness::snapshot(
&mut v,
[
Msg::SetMetric(CostMetric::Memory),
Msg::SetVersion(Version::Final),
Msg::ToggleGroup("1".into()),
],
);
let json = serde_json::to_string(&model).unwrap();
let back: PlanDagState = serde_json::from_str(&json).unwrap();
assert_eq!(
back, model,
"the observable Model round-trips through serde"
);
}
}