facett-plandag/src/model.rs
//! The **engine-agnostic operator-plan model** — the neutral shape a host maps a
//! Spark **SQL query plan** / a Databricks **Query Profile graph** / an **AQE
//! adaptive plan** onto, and [`PlanDagView`](crate::PlanDagView) renders. Pure +
//! egui-free: inject operators + metrics, [`resolve`] them under a cost metric and
//! a collapse set, assert the per-node bodies + cost-heat as DATA — no painter.
//!
//! ## Shape
//! ```text
//! PlanModel { initial: PlanVersion, final_plan: Option<PlanVersion> } // AQE: two plans
//! PlanVersion { label, nodes, edges }
//! PlanNode { id, op, group?, photon, status, metrics } // group = WholeStageCodegen[id]
//! PlanEdge { from, to } // indices into the version's nodes
//! Metrics { time_ms, peak_mem, rows, spill }
//! ```
//! The host adapter is the only thing that knows what an "operator" is; the model
//! speaks `String` ids/ops, a neutral [`OpStatus`], and a numeric [`Metrics`] bag.
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
/// Which metric drives the **cost-heat** node fill — the Query-Profile Time / Memory
/// peak / Rows switch.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
pub enum CostMetric {
/// Wall-clock time spent in the operator.
#[default]
Time,
/// Peak memory the operator held.
Memory,
/// Rows the operator produced.
Rows,
}
impl CostMetric {
/// A short, stable key for JSON / a toggle.
pub fn key(self) -> &'static str {
match self {
CostMetric::Time => "time",
CostMetric::Memory => "memory",
CostMetric::Rows => "rows",
}
}
/// A human label for the switch.
pub fn label(self) -> &'static str {
match self {
CostMetric::Time => "Time",
CostMetric::Memory => "Memory",
CostMetric::Rows => "Rows",
}
}
/// The switch in toggle order.
pub const ALL: [CostMetric; 3] = [CostMetric::Time, CostMetric::Memory, CostMetric::Rows];
}
/// Neutral operator run status (host maps its engine status onto this).
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default, Serialize, Deserialize)]
pub enum OpStatus {
#[default]
Idle,
Running,
Done,
Failed,
}
impl OpStatus {
pub fn label(self) -> &'static str {
match self {
OpStatus::Idle => "idle",
OpStatus::Running => "running",
OpStatus::Done => "done",
OpStatus::Failed => "failed",
}
}
/// Severity for aggregating a collapsed group (failed wins, then running, …).
fn rank(self) -> u8 {
match self {
OpStatus::Failed => 3,
OpStatus::Running => 2,
OpStatus::Done => 1,
OpStatus::Idle => 0,
}
}
}
/// The per-operator metric bag (the multi-line node body + the cost-heat source).
#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct Metrics {
/// Wall-clock time in the operator (ms).
pub time_ms: f64,
/// Peak memory held (bytes).
pub peak_mem: u64,
/// Output rows produced.
pub rows: u64,
/// Bytes spilled to disk (0 = none).
pub spill: u64,
}
impl Metrics {
/// The scalar this metric contributes to the cost-heat for `m`.
pub fn value_for(&self, m: CostMetric) -> f64 {
match m {
CostMetric::Time => self.time_ms,
CostMetric::Memory => self.peak_mem as f64,
CostMetric::Rows => self.rows as f64,
}
}
/// Element-wise sum (used to aggregate a collapsed WholeStageCodegen group).
fn add(&self, o: &Metrics) -> Metrics {
Metrics {
time_ms: self.time_ms + o.time_ms,
peak_mem: self.peak_mem + o.peak_mem,
rows: self.rows.max(o.rows), // a group's output rows ≈ its widest operator
spill: self.spill + o.spill,
}
}
}
/// One operator node.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct PlanNode {
/// Stable identity (selection + collapse key, FC-5).
pub id: String,
/// Operator name (e.g. `HashAggregate`, `Exchange`, `BatchScan`).
pub op: String,
/// The WholeStageCodegen / Photon group this node belongs to (collapsible).
pub group: Option<String>,
/// Vectorized (Photon) operator — drives the purple/orange palette.
pub photon: bool,
/// Run status.
pub status: OpStatus,
/// The operator's metrics.
pub metrics: Metrics,
}
impl PlanNode {
/// A node with an id + operator name; metrics/status set via the setters.
pub fn new(id: impl Into<String>, op: impl Into<String>) -> Self {
Self { id: id.into(), op: op.into(), group: None, photon: false, status: OpStatus::Done, metrics: Metrics::default() }
}
pub fn group(mut self, g: impl Into<String>) -> Self {
self.group = Some(g.into());
self
}
pub fn photon(mut self) -> Self {
self.photon = true;
self
}
pub fn status(mut self, s: OpStatus) -> Self {
self.status = s;
self
}
pub fn metrics(mut self, m: Metrics) -> Self {
self.metrics = m;
self
}
}
/// A directed edge `from → to` (indices into the version's `nodes`).
#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
pub struct PlanEdge {
pub from: usize,
pub to: usize,
}
/// One plan **version** — the operator graph of a single plan (AQE's *initial* or
/// *final*).
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct PlanVersion {
/// e.g. `"initial"` / `"final"`.
pub label: String,
pub nodes: Vec<PlanNode>,
pub edges: Vec<PlanEdge>,
}
impl PlanVersion {
pub fn new(label: impl Into<String>, nodes: Vec<PlanNode>, edges: Vec<PlanEdge>) -> Self {
Self { label: label.into(), nodes, edges }
}
/// The distinct collapsible group ids present, in first-seen order.
pub fn groups(&self) -> Vec<String> {
let mut seen = Vec::new();
for n in &self.nodes {
if let Some(g) = &n.group && !seen.contains(g) {
seen.push(g.clone());
}
}
seen
}
}
/// The whole plan: the initial graph + an optional **final** (post-AQE) graph.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct PlanModel {
pub initial: PlanVersion,
/// `None` ⇒ no AQE re-plan happened (the toggle is hidden).
pub final_plan: Option<PlanVersion>,
}
impl PlanModel {
pub fn new(initial: PlanVersion) -> Self {
Self { initial, final_plan: None }
}
pub fn with_final(mut self, final_plan: PlanVersion) -> Self {
self.final_plan = Some(final_plan);
self
}
pub fn has_aqe(&self) -> bool {
self.final_plan.is_some()
}
}
/// A **resolved** node — what the painter draws + `state_json` dumps after applying
/// the collapse set + cost metric. The body lines, the status (aggregated for a
/// collapsed group), and the **`heat`** (the cost-heat fraction `[0,1]` for the
/// active metric) are all here, so a headless test asserts the picture as DATA.
#[derive(Clone, Debug, PartialEq)]
pub struct ResolvedNode {
/// Stable id (a single op's id, or `group:<g>` for a collapsed super-node).
pub id: String,
/// Display title (the op name, or `WholeStageCodegen[g] ×N` collapsed).
pub title: String,
/// Aggregated status.
pub status: OpStatus,
/// Aggregated metrics.
pub metrics: Metrics,
/// How many original operators this node represents (1 unless collapsed).
pub members: usize,
/// Photon (vectorized) — true if every member is Photon.
pub photon: bool,
/// The collapsible group id this node is/was in (for the collapse toggle).
pub group: Option<String>,
/// `true` when this node is a collapsed WholeStageCodegen super-node.
pub collapsed: bool,
/// Cost-heat fraction `[0,1]` for the active [`CostMetric`] (value ÷ max).
pub heat: f32,
}
/// A **constraint ranking** entry — the resolved-DAG analog of
/// `funnel_core::Bottleneck`. Scores one node by its **transitive-descendant cone**
/// (`downstream` = every operator stalled behind it, the "unblocks the most work"
/// metric) plus local `fan_in`/`fan_out` degree.
///
/// This mirrors `funnel_core::PlanDag::bottlenecks()` deliberately rather than
/// depending on funnel-core: `facett-plandag` renders a *host-fed* generic operator
/// DAG (`PlanModel` ≠ a funnel `Funnel`) and is pinned to the buildable
/// dependency-island of `facett-core`/`facett-graph` (see this crate's `Cargo.toml`
/// note — the same reason it inlines `facett-pipeline`'s status colours). The
/// nornir-viz layer maps a funnel `PlanDag` → this `PlanModel`; the ranking rule
/// (downstream desc → fan_out → fan_in → id) is identical so the two agree.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Bottleneck {
/// Resolved-node id (`node:<id>` or `group:<g>`).
pub id: String,
/// Incoming producer edges.
pub fan_in: usize,
/// Outgoing consumer edges.
pub fan_out: usize,
/// Transitive descendants reachable following consumer edges (the reach).
pub downstream: usize,
}
/// The resolved graph (nodes + remapped edges) — the single source of truth for the
/// painter and `state_json`.
#[derive(Clone, Debug, Default)]
pub struct Resolved {
pub nodes: Vec<ResolvedNode>,
/// Edges as `(from_idx, to_idx)` into `nodes` (internal collapsed edges dropped).
pub edges: Vec<(usize, usize)>,
}
impl Resolved {
/// Producer→consumer adjacency over the resolved edge set.
fn adjacency(&self) -> Vec<Vec<usize>> {
let n = self.nodes.len();
let mut out = vec![Vec::new(); n];
for &(a, b) in &self.edges {
if a < n && b < n {
out[a].push(b);
}
}
out
}
/// **Rank every node by downstream reach** — the constraint ranking the funnel
/// pane surfaces to answer "*which node* unblocks the most work". `downstream` is
/// the transitive-descendant cone (all work stalled behind the node); the vector
/// is sorted `downstream` desc → `fan_out` desc → `fan_in` desc → id asc, so
/// `bottlenecks().first()` is the single biggest constraint. Mirrors
/// `funnel_core::PlanDag::bottlenecks()` (see [`Bottleneck`]). One descendant-set
/// pass per node, O(V·E) on the cold render path; empty/cyclic-free graphs are
/// fine (a cycle simply over-counts a saturated cone, still bounded by V).
pub fn bottlenecks(&self) -> Vec<Bottleneck> {
let n = self.nodes.len();
if n == 0 {
return Vec::new();
}
let out = self.adjacency();
let mut fan_in = vec![0usize; n];
let mut fan_out = vec![0usize; n];
for &(a, b) in &self.edges {
if a < n && b < n {
fan_out[a] += 1;
fan_in[b] += 1;
}
}
let mut ranked: Vec<Bottleneck> = (0..n)
.map(|i| {
// transitive descendants of i (iterative DFS over consumer edges).
let mut seen = vec![false; n];
let mut stack: Vec<usize> = out[i].clone();
let mut downstream = 0usize;
while let Some(x) = stack.pop() {
if x < n && !seen[x] {
seen[x] = true;
downstream += 1;
stack.extend(out[x].iter().copied());
}
}
Bottleneck { id: self.nodes[i].id.clone(), fan_in: fan_in[i], fan_out: fan_out[i], downstream }
})
.collect();
ranked.sort_by(|a, b| {
b.downstream
.cmp(&a.downstream)
.then(b.fan_out.cmp(&a.fan_out))
.then(b.fan_in.cmp(&a.fan_in))
.then(a.id.cmp(&b.id))
});
ranked
}
/// The **top interior bottleneck** id — the highest-ranked node that has an
/// incoming edge (`fan_in > 0`). Skipping the pure sources drops the root
/// operator whose cone is trivially the *whole* plan (the generic analog of the
/// funnel "mother idea" the funnel-core note says a renderer filters out), so the
/// answer is the genuine "clear this next" constraint. Falls back to the overall
/// `first()` if every node is a source; `None` for the empty graph. This is the
/// id the view **badges**.
pub fn top_bottleneck(&self) -> Option<String> {
let ranked = self.bottlenecks();
ranked
.iter()
.find(|b| b.fan_in > 0)
.or_else(|| ranked.first())
.map(|b| b.id.clone())
}
/// The **critical path** — the longest producer→consumer chain by hops, as
/// resolved-node ids in order. Mirrors `funnel_core::PlanDag::critical_path()`
/// (structural longest path with a deterministic id tie-break: successors and the
/// start are chosen by ascending id). Empty for an empty graph.
pub fn critical_path(&self) -> Vec<String> {
let n = self.nodes.len();
if n == 0 {
return Vec::new();
}
let mut out = self.adjacency();
// deterministic tie-break: visit successors in ascending id order.
for adj in out.iter_mut() {
adj.sort_by(|&x, &y| self.nodes[x].id.cmp(&self.nodes[y].id));
}
let mut len = vec![0usize; n]; // longest chain (node count) starting at i
let mut nxt = vec![usize::MAX; n]; // chosen successor
let mut state = vec![0u8; n]; // 0 unvisited, 1 active (cycle guard), 2 done
fn walk(i: usize, out: &[Vec<usize>], len: &mut [usize], nxt: &mut [usize], state: &mut [u8]) -> usize {
match state[i] {
2 => return len[i],
1 => return 1, // cycle: treat as terminal (plans are DAGs)
_ => {}
}
state[i] = 1;
let mut best = 1usize;
let mut best_next = usize::MAX;
for &j in &out[i] {
let l = walk(j, out, len, nxt, state) + 1;
if l > best {
best = l;
best_next = j;
}
}
len[i] = best;
nxt[i] = best_next;
state[i] = 2;
best
}
for i in 0..n {
let _ = walk(i, &out, &mut len, &mut nxt, &mut state);
}
// start node = max chain length, smallest id tie-break.
let mut order: Vec<usize> = (0..n).collect();
order.sort_by(|&x, &y| self.nodes[x].id.cmp(&self.nodes[y].id));
let mut start = order[0];
let mut best = 0usize;
for &i in &order {
if len[i] > best {
best = len[i];
start = i;
}
}
let mut path = Vec::new();
let mut cur = start;
while cur != usize::MAX {
path.push(self.nodes[cur].id.clone());
cur = nxt[cur];
}
path
}
}
impl Resolved {
pub fn state_json(&self, metric: CostMetric) -> serde_json::Value {
let nodes: Vec<serde_json::Value> = self
.nodes
.iter()
.map(|n| {
serde_json::json!({
"id": n.id,
"title": n.title,
"status": n.status.label(),
"members": n.members,
"photon": n.photon,
"collapsed": n.collapsed,
"group": n.group,
// the cost-heat, AS DATA — what the fill intensity comes from
"heat": n.heat,
"metrics": {
"time_ms": n.metrics.time_ms,
"peak_mem": n.metrics.peak_mem,
"rows": n.metrics.rows,
"spill": n.metrics.spill,
},
})
})
.collect();
let bottlenecks: Vec<serde_json::Value> = self
.bottlenecks()
.iter()
.map(|b| serde_json::json!({ "id": b.id, "fan_in": b.fan_in, "fan_out": b.fan_out, "downstream": b.downstream }))
.collect();
serde_json::json!({
"metric": metric.key(),
"node_count": self.nodes.len(),
"edge_count": self.edges.len(),
"nodes": nodes,
// constraint analysis (mirrors funnel-core), AS DATA: the full ranking,
// the badged "clear next" node, and the critical-path chain.
"bottlenecks": bottlenecks,
"top_bottleneck": self.top_bottleneck(),
"critical_path": self.critical_path(),
})
}
}
/// **Resolve** a plan version under a collapse set + cost metric — the pure core the
/// view paints and `state_json` dumps. Collapses every node whose `group` is in
/// `collapsed` into one super-node (summed metrics, aggregated status), remaps the
/// edges (dropping intra-group ones), then computes the cost-heat `[0,1]` per node.
pub fn resolve(version: &PlanVersion, collapsed: &[String], metric: CostMetric) -> Resolved {
// Effective key per original node index: a collapsed group, or the node itself.
let key_of = |n: &PlanNode| -> String {
match &n.group {
Some(g) if collapsed.iter().any(|c| c == g) => format!("group:{g}"),
_ => format!("node:{}", n.id),
}
};
// Build effective nodes in first-seen order; aggregate members.
let mut order: Vec<String> = Vec::new();
let mut by_key: BTreeMap<String, ResolvedNode> = BTreeMap::new();
let mut idx_of_orig: Vec<usize> = Vec::with_capacity(version.nodes.len());
for n in &version.nodes {
let key = key_of(n);
if !by_key.contains_key(&key) {
order.push(key.clone());
let collapsed_group = key.starts_with("group:");
by_key.insert(
key.clone(),
ResolvedNode {
id: key.clone(),
title: if collapsed_group {
format!("WholeStageCodegen[{}]", n.group.clone().unwrap_or_default())
} else {
n.op.clone()
},
status: n.status,
metrics: Metrics::default(),
members: 0,
photon: true,
group: n.group.clone(),
collapsed: collapsed_group,
heat: 0.0,
},
);
}
let rn = by_key.get_mut(&key).unwrap();
rn.metrics = rn.metrics.add(&n.metrics);
rn.members += 1;
rn.photon = rn.photon && n.photon;
if n.status.rank() > rn.status.rank() {
rn.status = n.status;
}
idx_of_orig.push(0); // placeholder; filled after order is final
}
// Map effective key → final index.
let pos_of: BTreeMap<&String, usize> = order.iter().enumerate().map(|(i, k)| (k, i)).collect();
// Recompute idx_of_orig now that order is final.
for (oi, n) in version.nodes.iter().enumerate() {
idx_of_orig[oi] = pos_of[&key_of(n)];
}
// Collapsed super-node titles get a ×N count.
let nodes: Vec<ResolvedNode> = order
.iter()
.map(|k| {
let mut rn = by_key.remove(k).unwrap();
if rn.collapsed {
rn.title = format!("{} ×{}", rn.title, rn.members);
}
rn
})
.collect();
let mut resolved = Resolved { nodes, edges: Vec::new() };
// Remap edges; drop self-loops (intra-group) + duplicates.
let mut seen_edges: Vec<(usize, usize)> = Vec::new();
for e in &version.edges {
if e.from >= idx_of_orig.len() || e.to >= idx_of_orig.len() {
continue;
}
let a = idx_of_orig[e.from];
let b = idx_of_orig[e.to];
if a != b && !seen_edges.contains(&(a, b)) {
seen_edges.push((a, b));
}
}
resolved.edges = seen_edges;
// Cost-heat: value ÷ max across the resolved nodes for the active metric.
let max = resolved.nodes.iter().map(|n| n.metrics.value_for(metric)).fold(0.0_f64, f64::max);
if max > 0.0 {
for n in &mut resolved.nodes {
n.heat = (n.metrics.value_for(metric) / max).clamp(0.0, 1.0) as f32;
}
}
resolved
}
#[cfg(test)]
mod tests {
use super::*;
fn demo() -> PlanVersion {
// scan → filter → (agg in WSCG group "1") → exchange → (final agg in WSCG "1")
let nodes = vec![
PlanNode::new("scan", "BatchScan").metrics(Metrics { time_ms: 50.0, peak_mem: 1 << 20, rows: 1_000_000, spill: 0 }),
PlanNode::new("filter", "Filter").group("1").metrics(Metrics { time_ms: 20.0, peak_mem: 1 << 18, rows: 400_000, spill: 0 }),
PlanNode::new("hashagg", "HashAggregate").group("1").metrics(Metrics { time_ms: 300.0, peak_mem: 1 << 24, rows: 5_000, spill: 1 << 20 }),
PlanNode::new("exchange", "Exchange").metrics(Metrics { time_ms: 120.0, peak_mem: 1 << 19, rows: 5_000, spill: 0 }),
];
let edges = vec![
PlanEdge { from: 0, to: 1 },
PlanEdge { from: 1, to: 2 },
PlanEdge { from: 2, to: 3 },
];
PlanVersion::new("initial", nodes, edges)
}
#[test]
fn resolve_uncollapsed_keeps_all_nodes() {
let v = demo();
let r = resolve(&v, &[], CostMetric::Time);
assert_eq!(r.nodes.len(), 4);
assert_eq!(r.edges.len(), 3);
// heat: hashagg (300ms) is the max → heat 1.0; scan (50) → ~0.166.
let agg = r.nodes.iter().find(|n| n.title == "HashAggregate").unwrap();
assert!((agg.heat - 1.0).abs() < 1e-6, "the slowest op is the hottest");
let scan = r.nodes.iter().find(|n| n.title == "BatchScan").unwrap();
assert!((scan.heat - (50.0 / 300.0) as f32).abs() < 1e-5);
}
#[test]
fn collapse_group_merges_metrics_and_edges() {
let v = demo();
let r = resolve(&v, &["1".into()], CostMetric::Time);
// filter+hashagg fold into one WholeStageCodegen[1] super-node → 3 nodes.
assert_eq!(r.nodes.len(), 3, "the WSCG group collapsed two ops into one");
let g = r.nodes.iter().find(|n| n.collapsed).unwrap();
assert_eq!(g.members, 2);
assert!(g.title.contains("WholeStageCodegen[1] ×2"));
// summed time = 20 + 300 = 320 ms.
assert!((g.metrics.time_ms - 320.0).abs() < 1e-6);
// edges: scan→group, group→exchange (the intra-group filter→agg edge dropped).
assert_eq!(r.edges.len(), 2);
}
#[test]
fn cost_metric_switch_changes_heat() {
let v = demo();
// By rows, scan (1M) dominates; by time, hashagg dominates.
let by_rows = resolve(&v, &[], CostMetric::Rows);
let scan = by_rows.nodes.iter().find(|n| n.title == "BatchScan").unwrap();
assert!((scan.heat - 1.0).abs() < 1e-6, "by rows, the scan is the hottest");
let by_time = resolve(&v, &[], CostMetric::Time);
let scan_t = by_time.nodes.iter().find(|n| n.title == "BatchScan").unwrap();
assert!(scan_t.heat < 0.5, "by time, the scan is cool");
}
/// A **wide-vs-deep** DAG: a source `root` feeds a WIDE node (fan_out 3, three
/// dead-end leaves) and a DEEP node (fan_out 1, heading a 4-long chain). The deep
/// node stalls MORE work (downstream 4) than the wide node (downstream 3) despite
/// its smaller fan-out — the reach-beats-fan-out discriminator.
fn wide_vs_deep() -> PlanVersion {
let n = |id: &str, op: &str| PlanNode::new(id, op);
let nodes = vec![
n("root", "Root"),
n("wide", "Wide"),
n("w1", "W1"),
n("w2", "W2"),
n("w3", "W3"),
n("deep", "Deep"),
n("d1", "D1"),
n("d2", "D2"),
n("d3", "D3"),
n("d4", "D4"),
];
let e = |from: usize, to: usize| PlanEdge { from, to };
let edges = vec![
e(0, 1), // root → wide
e(0, 5), // root → deep
e(1, 2), // wide → w1
e(1, 3), // wide → w2
e(1, 4), // wide → w3
e(5, 6), // deep → d1
e(6, 7), // d1 → d2
e(7, 8), // d2 → d3
e(8, 9), // d3 → d4
];
PlanVersion::new("initial", nodes, edges)
}
#[test]
fn bottlenecks_rank_by_downstream_reach_not_fanout() {
let r = resolve(&wide_vs_deep(), &[], CostMetric::Time);
let ranked = r.bottlenecks();
// The root's cone is the whole plan → it is the unfiltered #1.
assert_eq!(ranked.first().unwrap().id, "node:root");
assert_eq!(ranked.first().unwrap().downstream, 9);
let by = |id: &str| ranked.iter().find(|b| b.id == format!("node:{id}")).unwrap();
let deep = by("deep");
let wide = by("wide");
// RED-when-broken: reach beats fan-out. `deep` (fan_out 1, downstream 4) must
// outrank `wide` (fan_out 3, downstream 3). A fan_out-primary sort flips this.
assert_eq!(deep.downstream, 4);
assert_eq!(wide.downstream, 3);
assert_eq!(wide.fan_out, 3);
assert_eq!(deep.fan_out, 1);
let pos = |id: &str| ranked.iter().position(|b| b.id == format!("node:{id}")).unwrap();
assert!(pos("deep") < pos("wide"), "deep chain outranks the wide fan by reach");
// The badged "clear next" node skips the source root → it is `deep`.
assert_eq!(r.top_bottleneck().as_deref(), Some("node:deep"));
}
#[test]
fn critical_path_is_the_longest_chain() {
let r = resolve(&wide_vs_deep(), &[], CostMetric::Time);
let cp = r.critical_path();
// root → deep → d1 → d2 → d3 → d4 (6 nodes), NOT the 3-node wide branch.
assert_eq!(cp, vec!["node:root", "node:deep", "node:d1", "node:d2", "node:d3", "node:d4"]);
}
#[test]
fn state_json_carries_the_constraint_analysis() {
let r = resolve(&wide_vs_deep(), &[], CostMetric::Time);
let j = r.state_json(CostMetric::Time);
assert_eq!(j["top_bottleneck"], "node:deep");
assert_eq!(j["critical_path"].as_array().unwrap().len(), 6);
let bn = j["bottlenecks"].as_array().unwrap();
assert_eq!(bn.len(), 10);
assert_eq!(bn[0]["id"], "node:root", "the ranking leads with the whole-plan cone");
assert!(bn.iter().all(|b| b["downstream"].is_number() && b["fan_in"].is_number()));
}
#[test]
fn empty_graph_has_no_constraints() {
let r = resolve(&PlanVersion::default(), &[], CostMetric::Time);
assert!(r.bottlenecks().is_empty());
assert!(r.top_bottleneck().is_none());
assert!(r.critical_path().is_empty());
}
#[test]
fn state_json_carries_heat_and_metrics() {
let v = demo();
let r = resolve(&v, &[], CostMetric::Memory);
let j = r.state_json(CostMetric::Memory);
assert_eq!(j["metric"], "memory");
assert_eq!(j["node_count"], 4);
let nodes = j["nodes"].as_array().unwrap();
assert!(nodes.iter().all(|n| n["heat"].is_number()), "every node carries cost-heat as data");
assert!(nodes.iter().any(|n| n["metrics"]["spill"].as_u64().unwrap() > 0), "spill surfaced");
}
}