integrations/knut-flow.patch
diff --git a/crates/knut-bifrost/Cargo.toml b/crates/knut-bifrost/Cargo.toml
index c874925..d556b68 100644
--- a/crates/knut-bifrost/Cargo.toml
+++ b/crates/knut-bifrost/Cargo.toml
@@ -9,6 +9,10 @@ repository.workspace = true
[features]
default = []
+# Delivery outcomes belong to the skade source. Rendering is optional and does
+# not pull wgpu into the delivery engine. Development uses the sibling checkout;
+# change to the published requirement when the Falsett release is available.
+facett-flow = ["dep:facett-flow", "skade"]
# `testmatrix` — wire the in-process functional-status emitter into the
# component self-test / headless-render paths so a broken render surfaces as a
# RED row in `nornir test` / the viz 🧪 Test pane. Passes the feature straight
@@ -111,6 +115,7 @@ iceberg-doris = ["doris", "skade-rest", "skade-grpc"]
service = ["cli", "iceberg-doris", "dep:skidbladnir", "dep:anyhow"]
[dependencies]
+facett-flow = { path = "../../../facett/facett-flow", version = "0.2.0", optional = true, features = ["layered", "status", "rate", "labels", "compact-state"] }
graphar = { path = "../graphar" , version = "0.1.0" }
graphar-falkordb = { path = "../graphar-falkordb" , version = "0.1.13" }
diff --git a/crates/knut-bifrost/src/lib.rs b/crates/knut-bifrost/src/lib.rs
index 69e09a1..75c3267 100644
--- a/crates/knut-bifrost/src/lib.rs
+++ b/crates/knut-bifrost/src/lib.rs
@@ -54,6 +54,8 @@ pub mod delivery;
#[cfg(feature = "service")]
pub mod doris_cli;
pub mod error;
+#[cfg(feature = "facett-flow")]
+pub mod flow_adapter;
#[cfg(feature = "flight")]
pub mod flight;
pub mod listener;
diff --git a/crates/knut-pipelines/Cargo.toml b/crates/knut-pipelines/Cargo.toml
index ad8809f..ec9241b 100644
--- a/crates/knut-pipelines/Cargo.toml
+++ b/crates/knut-pipelines/Cargo.toml
@@ -113,7 +113,7 @@ knut-tracing = { path = "../knut-tracing", version = "0.1" }
# Default-features off so this pulls just the model/view crate (no eframe example
# deps). It shares knut's egui 0.34 rather than standing up a second tree — that
# isolation was only ever forced by egui-snarl 0.4 holding `gui` on 0.28.
-facett-pipeline = { version = ">=0.1.19, <0.3", default-features = false, optional = true }
+facett-pipeline = { version = ">=0.1.19, <0.4", default-features = false, optional = true }
# Airflow (AIP-82) asset-event model + emitter trait, only with `airflow`.
# default-features off: the `airflow` module (AssetEvent/AssetEmitter/
diff --git a/crates/knut-bifrost/src/flow_adapter.rs b/crates/knut-bifrost/src/flow_adapter.rs
new file mode 100644
index 0000000..5058137
--- /dev/null
+++ b/crates/knut-bifrost/src/flow_adapter.rs
@@ -0,0 +1,321 @@
+//! Pure Falsett adapters for delivery rounds and ordered Iceberg snapshots.
+//!
+//! The optional `facett-flow` feature also enables `skade`, which owns the
+//! actual [`RoundOutcome`] type. This module performs no I/O and owns no clock.
+use std::collections::{BTreeMap, BTreeSet};
+
+use crate::{delivery::RoundOutcome, listener::RoundReport};
+use facett_flow::attr::{Rates, Status, StatusTable};
+use facett_flow::{Edge, Graph, LayeredOpts, Node};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct DeliveryRoute {
+ pub source: String,
+ pub sink: String,
+ pub table: String,
+}
+
+/// Input supplied by the host, not an upstream Iceberg metadata type.
+/// Pass snapshots in ancestry order, oldest first. Snapshot IDs are opaque;
+/// their numeric ordering says nothing about commit order.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct SnapshotInfo {
+ pub id: i64,
+}
+
+#[derive(Clone, Debug)]
+pub struct FlowParts {
+ pub graph: Graph,
+ pub status: StatusTable,
+ pub rates: Rates,
+}
+
+fn layout(graph: &mut Graph) {
+ // Keep even 100k routes within Falsett's finite coordinate budget. Uniform
+ // scaling preserves the aspect ratio and producer-to-consumer ordering.
+ let scale = (1_000_000.0 / (graph.nodes.len().max(1) as f32 * 240.0)).min(1.0);
+ graph
+ .layered(LayeredOpts {
+ pitch: [240.0 * scale, 120.0 * scale],
+ ..Default::default()
+ })
+ .expect("adapter constructs valid bounded layered geometry");
+}
+
+fn combine(a: Option<Status>, b: Option<Status>) -> Option<Status> {
+ match (a, b) {
+ (Some(Status::Error), _) | (_, Some(Status::Error)) => Some(Status::Error),
+ (Some(Status::Slow), _) | (_, Some(Status::Slow)) => Some(Status::Slow),
+ (Some(Status::Ok), _) | (_, Some(Status::Ok)) => Some(Status::Ok),
+ _ => None,
+ }
+}
+
+/// Build one table node per `(source, table)`, one destination node per sink,
+/// and one edge per distinct route. Outcomes and failed/synced reports are keyed
+/// by `DeliveryRoute.table`. Failure takes precedence over lag and success.
+///
+/// Missing outcomes have unknown rate. A measured zero-row round has rate zero,
+/// even if its duration is zero; a positive-row round with zero duration has an
+/// unknown rate rather than infinity. Destination status is the worst incoming
+/// route status. Node/edge ordering follows first occurrence in `routes`.
+pub fn delivery_graph(
+ routes: &[DeliveryRoute],
+ outcomes: &BTreeMap<String, RoundOutcome>,
+ report: Option<&RoundReport>,
+) -> FlowParts {
+ let failed: BTreeSet<&str> = report
+ .into_iter()
+ .flat_map(|r| r.failed.iter().map(|(table, _)| table.as_str()))
+ .collect();
+ let synced: BTreeSet<&str> = report
+ .into_iter()
+ .flat_map(|r| r.synced.iter().map(|(table, _, _)| table.as_str()))
+ .collect();
+ let mut graph = Graph {
+ nodes: Vec::new(),
+ edges: Vec::new(),
+ };
+ let mut status = StatusTable {
+ nodes: Vec::new(),
+ edges: Vec::new(),
+ };
+ let mut rates = Rates { edges: Vec::new() };
+ let mut node_indices = BTreeMap::<String, usize>::new();
+ let mut seen_routes = BTreeSet::new();
+ for route in routes {
+ // Length-delimited table identity avoids delimiter collisions; source and
+ // sink namespaces remain disjoint even when their human names coincide.
+ let source_id = format!(
+ "table:{}:{}{}",
+ route.source.len(),
+ route.source,
+ route.table
+ );
+ let sink_id = format!("sink:{}", route.sink);
+ if !seen_routes.insert((source_id.clone(), sink_id.clone())) {
+ continue;
+ }
+ let outcome = outcomes.get(&route.table);
+ let value = if failed.contains(route.table.as_str()) {
+ Some(Status::Error)
+ } else if outcome.is_some_and(|o| o.behind > 0) {
+ Some(Status::Slow)
+ } else if outcome.is_some() || synced.contains(route.table.as_str()) {
+ Some(Status::Ok)
+ } else {
+ None
+ };
+ for (id, label, group, kind) in [
+ (&source_id, &route.table, &route.source, 0),
+ (&sink_id, &route.sink, &route.sink, 2),
+ ] {
+ let index = *node_indices.entry(id.clone()).or_insert_with(|| {
+ let index = graph.nodes.len();
+ graph.nodes.push(Node {
+ id: id.clone(),
+ label: label.clone(),
+ position: [0.0, 0.0],
+ group: group.clone(),
+ kind,
+ });
+ status.nodes.push(None);
+ index
+ });
+ status.nodes[index] = combine(status.nodes[index], value);
+ }
+ graph.edges.push(Edge {
+ from: source_id,
+ to: sink_id,
+ });
+ status.edges.push(value);
+ rates.edges.push(outcome.and_then(|o| {
+ if o.rows_loaded == 0 {
+ return Some(0.0);
+ }
+ let seconds = o.read.as_secs_f64() + o.load.as_secs_f64();
+ (seconds > 0.0).then(|| (o.rows_loaded as f64 / seconds).min(f32::MAX as f64) as f32)
+ }));
+ }
+ layout(&mut graph);
+ FlowParts {
+ graph,
+ status,
+ rates,
+ }
+}
+
+/// Build an oldest-to-newest snapshot chain. Duplicate IDs are ignored after
+/// their first occurrence. A missing/expired cursor marks no snapshot as done.
+pub fn snapshot_chain(table: &str, snaps: &[SnapshotInfo], cursor: Option<i64>) -> FlowParts {
+ let mut seen = BTreeSet::new();
+ let ordered: Vec<_> = snaps.iter().filter(|s| seen.insert(s.id)).collect();
+ let at = cursor.and_then(|id| ordered.iter().position(|s| s.id == id));
+ let mut graph = Graph {
+ nodes: ordered
+ .iter()
+ .map(|s| Node {
+ id: format!("snapshot:{}:{table}:{}", table.len(), s.id),
+ label: s.id.to_string(),
+ position: [0.0, 0.0],
+ group: table.into(),
+ kind: 0,
+ })
+ .collect(),
+ edges: Vec::new(),
+ };
+ graph.edges = graph
+ .nodes
+ .windows(2)
+ .map(|pair| Edge {
+ from: pair[0].id.clone(),
+ to: pair[1].id.clone(),
+ })
+ .collect();
+ let status = StatusTable {
+ nodes: (0..ordered.len())
+ .map(|i| at.filter(|&a| i <= a).map(|_| Status::Ok))
+ .collect(),
+ edges: (1..ordered.len())
+ .map(|i| at.filter(|&a| i <= a).map(|_| Status::Ok))
+ .collect(),
+ };
+ // Snapshots describe history, not observed throughput; keep their edges still.
+ let rates = Rates {
+ edges: vec![Some(0.0); graph.edges.len()],
+ };
+ layout(&mut graph);
+ FlowParts {
+ graph,
+ status,
+ rates,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::time::Duration;
+
+ fn route(table: &str) -> DeliveryRoute {
+ DeliveryRoute {
+ source: "lake".into(),
+ sink: "warehouse".into(),
+ table: table.into(),
+ }
+ }
+
+ #[test]
+ fn real_round_fixtures_map_counts_status_and_elapsed_rate() {
+ let outcomes = BTreeMap::from([
+ (
+ "orders".into(),
+ RoundOutcome {
+ rows_loaded: 120,
+ behind: 2,
+ read: Duration::from_secs(1),
+ load: Duration::from_secs(2),
+ ..Default::default()
+ },
+ ),
+ ("users".into(), RoundOutcome::default()),
+ ]);
+ let report = RoundReport {
+ failed: vec![("orders".into(), "sink unavailable".into())],
+ ..Default::default()
+ };
+ let parts = delivery_graph(&[route("orders"), route("users")], &outcomes, Some(&report));
+ parts.graph.validate().unwrap();
+ assert_eq!((parts.graph.nodes.len(), parts.graph.edges.len()), (3, 2));
+ assert_eq!(
+ parts.status.nodes,
+ [Some(Status::Error), Some(Status::Error), Some(Status::Ok)]
+ );
+ assert_eq!(parts.status.edges, [Some(Status::Error), Some(Status::Ok)]);
+ assert_eq!(parts.rates.edges, [Some(40.0), Some(0.0)]);
+ let slow = delivery_graph(&[route("orders")], &outcomes, None);
+ assert_eq!(slow.status.nodes, [Some(Status::Slow), Some(Status::Slow)]);
+ for edge in &parts.graph.edges {
+ let a = parts
+ .graph
+ .nodes
+ .iter()
+ .find(|n| n.id == edge.from)
+ .unwrap();
+ let b = parts.graph.nodes.iter().find(|n| n.id == edge.to).unwrap();
+ assert!(a.position[0] < b.position[0]);
+ assert_eq!((a.kind, b.kind), (0, 2));
+ }
+ }
+
+ #[test]
+ fn missing_and_zero_duration_samples_are_not_infinite_or_fabricated() {
+ let outcomes = BTreeMap::from([(
+ "orders".into(),
+ RoundOutcome {
+ rows_loaded: 10,
+ ..Default::default()
+ },
+ )]);
+ let parts = delivery_graph(
+ &[route("orders"), route("unknown"), route("orders")],
+ &outcomes,
+ None,
+ );
+ assert_eq!(parts.graph.edges.len(), 2);
+ assert_eq!(parts.rates.edges, [None, None]);
+ assert_eq!(parts.status.nodes[2], None);
+ }
+
+ #[test]
+ fn snapshot_cursor_uses_ancestry_order_not_random_numeric_ids() {
+ let snaps = [
+ SnapshotInfo { id: 90 },
+ SnapshotInfo { id: 7 },
+ SnapshotInfo { id: 40 },
+ ];
+ let parts = snapshot_chain("orders", &snaps, Some(7));
+ parts.graph.validate().unwrap();
+ assert_eq!(
+ parts.status.nodes,
+ [Some(Status::Ok), Some(Status::Ok), None]
+ );
+ assert_eq!(parts.status.edges, [Some(Status::Ok), None]);
+ assert_eq!(parts.rates.edges, [Some(0.0), Some(0.0)]);
+ assert!(
+ parts
+ .graph
+ .nodes
+ .windows(2)
+ .all(|w| w[0].position[0] < w[1].position[0])
+ );
+ assert!(
+ snapshot_chain("orders", &snaps, Some(12))
+ .status
+ .nodes
+ .iter()
+ .all(Option::is_none)
+ );
+ assert!(snapshot_chain("orders", &[], None).graph.nodes.is_empty());
+ }
+
+ #[test]
+ fn one_hundred_thousand_routes_have_linear_output_and_finite_positions() {
+ let routes: Vec<_> = (0..100_000).map(|i| route(&format!("table{i}"))).collect();
+ let started = std::time::Instant::now();
+ let parts = delivery_graph(&routes, &BTreeMap::new(), None);
+ parts.graph.validate().unwrap();
+ assert_eq!(parts.graph.nodes.len(), 100_001);
+ assert_eq!(parts.graph.edges.len(), 100_000);
+ assert_eq!(parts.status.nodes.len(), parts.graph.nodes.len());
+ assert_eq!(parts.rates.edges.len(), parts.graph.edges.len());
+ assert!(
+ parts
+ .graph
+ .nodes
+ .iter()
+ .all(|n| n.position.iter().all(|p| p.is_finite()))
+ );
+ eprintln!("100k delivery routes: {:?}", started.elapsed());
+ }
+}