facett-lineage/src/lib.rs
//! **facett-lineage** — the data-**LINEAGE** viewer facet.
//!
//! Reads skade's reserved **`lineage_events`** table (the "who did what, to which
//! tables, between which systems, when" fact log) and renders it as a **lineage graph**:
//! **datasets are nodes**, **runs are edges** (`input → output`, labelled by the
//! operation, carrying the actor / commit cursor / timestamp). The whole graph-DB IDE
//! surface — query bar, laid-out canvas, node/edge inspector, layout switcher, styling,
//! shortest-path highlighting, minimap, export — is **reused unchanged** from
//! [`facett_graphnav::GraphNav`]; this crate is only the *adapter* from the
//! `lineage_events` column shape to a [`PropGraph`].
//!
//! # The column shape (modelled locally — no skade dependency)
//! facett is a **leaf peer** of skade, so we do **not** depend on skade. Instead the
//! reserved [`lineage_events_schema`] and the [`LineageRow`] shape are mirrored here,
//! byte-for-byte against skade's `lineage_events_schema()` (design §4.3):
//! `event_id, actor, operation, input_table, input_snapshot, output_table,
//! output_snapshot, source_system, source_instance, target_system, target_instance,
//! ts_micros, commit_seq`. A skade warehouse's `lineage_events` `RecordBatch` drops
//! straight into [`read_rows`]; nothing skade-side has to change.
//!
//! This is the "facett viewer adapter" the skade lineage design defers to facett.
use std::collections::BTreeMap;
use arrow_array::{Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use facett_core::{Facet, FacetCaps};
use facett_graphnav::{Edge, GraphNav, Node, PropGraph};
use serde::{Deserialize, Serialize};
/// OpenLineage / OpenMetadata-standard ingest — read lineage that arrived over
/// the open `RunEvent` wire (the format skade's `HttpLineageSink` emits and
/// OpenMetadata/Marquez/DataHub ingest) back into [`LineageRow`]s, so one viewer
/// renders both the skade-direct Arrow feed and a standard-collector feed.
pub mod openlineage;
pub use facett_graphnav::{ExportFormat, LayoutKind, Msg, StyleBy};
/// One row of the reserved `lineage_events` table — the flattened, single-input row
/// form skade's `WarehouseLineageSink` appends (a multi-input event becomes one row per
/// declared input). Mirrors skade's `LineageEvent::to_arrow_row` column shape exactly.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LineageRow {
/// Stable per-event id (process-unique, time-ordered).
pub event_id: String,
/// Identity that performed the write (`"skade"` when unattributed).
pub actor: String,
/// What the write did: `append`/`delete`/`overwrite`/`create`/`release`.
pub operation: String,
/// The (first) upstream dataset consumed, if any.
pub input_table: Option<String>,
/// The upstream snapshot read, if known.
pub input_snapshot: Option<i64>,
/// The dataset produced (always present).
pub output_table: String,
/// The produced snapshot, if known.
pub output_snapshot: Option<i64>,
/// The source (input) system kind, e.g. `"spark"` — `None` = this warehouse.
pub source_system: Option<String>,
/// The source system instance (a Spark app id, a broker cluster, …).
pub source_instance: Option<String>,
/// The target (output) system kind — `None` = this warehouse.
pub target_system: Option<String>,
/// The target system instance.
pub target_instance: Option<String>,
/// Event time, microseconds since the Unix epoch.
pub ts_micros: i64,
/// The durable monotonic catalog commit cursor, when known.
pub commit_seq: Option<i64>,
}
/// The reserved **`lineage_events`** Arrow schema — the exact column shape skade's
/// `lineage_events_schema()` produces, mirrored here so a skade `RecordBatch` reads with
/// no adapter on either side. One row per (flattened) [`LineageRow`].
#[must_use]
pub fn lineage_events_schema() -> ArrowSchema {
ArrowSchema::new(vec![
Field::new("event_id", DataType::Utf8, false),
Field::new("actor", DataType::Utf8, false),
Field::new("operation", DataType::Utf8, false),
Field::new("input_table", DataType::Utf8, true),
Field::new("input_snapshot", DataType::Int64, true),
Field::new("output_table", DataType::Utf8, false),
Field::new("output_snapshot", DataType::Int64, true),
Field::new("source_system", DataType::Utf8, true),
Field::new("source_instance", DataType::Utf8, true),
Field::new("target_system", DataType::Utf8, true),
Field::new("target_instance", DataType::Utf8, true),
Field::new("ts_micros", DataType::Int64, false),
Field::new("commit_seq", DataType::Int64, true),
])
}
/// The `lineage_events` table name (skade's reserved table).
pub const LINEAGE_EVENTS_TABLE: &str = "lineage_events";
/// **Read a `lineage_events` [`RecordBatch`] into [`LineageRow`]s** — the by-name,
/// null-tolerant reader (the inverse of skade's `to_arrow_row`, over a whole batch).
/// Missing `event_id`/`actor`/`operation`/`output_table`/`ts_micros` (the non-null
/// columns) is an error; the optional columns default to `None`.
///
/// # Errors
/// Returns a message if a required column is absent, mistyped, or null in some row.
pub fn read_rows(batch: &RecordBatch) -> Result<Vec<LineageRow>, String> {
let n = batch.num_rows();
let sget = |name: &str, row: usize| -> Option<String> {
batch
.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.filter(|a| a.is_valid(row))
.map(|a| a.value(row).to_string())
};
let iget = |name: &str, row: usize| -> Option<i64> {
batch
.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<Int64Array>())
.filter(|a| a.is_valid(row))
.map(|a| a.value(row))
};
// Fail loudly if a required column is entirely missing/mistyped.
for col in ["event_id", "actor", "operation", "output_table"] {
if batch.column_by_name(col).and_then(|c| c.as_any().downcast_ref::<StringArray>()).is_none() {
return Err(format!("lineage_events: missing/mistyped Utf8 column `{col}`"));
}
}
if batch.column_by_name("ts_micros").and_then(|c| c.as_any().downcast_ref::<Int64Array>()).is_none() {
return Err("lineage_events: missing/mistyped Int64 column `ts_micros`".to_string());
}
let mut rows = Vec::with_capacity(n);
for row in 0..n {
let req = |v: Option<String>, name: &str| v.ok_or_else(|| format!("lineage_events.{name} is null at row {row}"));
rows.push(LineageRow {
event_id: req(sget("event_id", row), "event_id")?,
actor: req(sget("actor", row), "actor")?,
operation: req(sget("operation", row), "operation")?,
input_table: sget("input_table", row),
input_snapshot: iget("input_snapshot", row),
output_table: req(sget("output_table", row), "output_table")?,
output_snapshot: iget("output_snapshot", row),
source_system: sget("source_system", row),
source_instance: sget("source_instance", row),
target_system: sget("target_system", row),
target_instance: sget("target_instance", row),
ts_micros: iget("ts_micros", row).ok_or_else(|| format!("lineage_events.ts_micros is null at row {row}"))?,
commit_seq: iget("commit_seq", row),
});
}
Ok(rows)
}
/// Encode rows back into a single [`RecordBatch`] with [`lineage_events_schema`] — the
/// round-trip partner of [`read_rows`] (chiefly for tests / re-export).
///
/// # Errors
/// Propagates any Arrow batch-construction error.
pub fn to_record_batch(rows: &[LineageRow]) -> Result<RecordBatch, String> {
use std::sync::Arc;
let s = |f: fn(&LineageRow) -> Option<String>| -> Arc<dyn Array> {
Arc::new(StringArray::from_iter(rows.iter().map(f)))
};
let sv = |f: fn(&LineageRow) -> String| -> Arc<dyn Array> {
Arc::new(StringArray::from_iter_values(rows.iter().map(f)))
};
let i = |f: fn(&LineageRow) -> Option<i64>| -> Arc<dyn Array> {
Arc::new(Int64Array::from_iter(rows.iter().map(f)))
};
let iv = |f: fn(&LineageRow) -> i64| -> Arc<dyn Array> {
Arc::new(Int64Array::from_iter_values(rows.iter().map(f)))
};
let cols: Vec<Arc<dyn Array>> = vec![
sv(|r| r.event_id.clone()),
sv(|r| r.actor.clone()),
sv(|r| r.operation.clone()),
s(|r| r.input_table.clone()),
i(|r| r.input_snapshot),
sv(|r| r.output_table.clone()),
i(|r| r.output_snapshot),
s(|r| r.source_system.clone()),
s(|r| r.source_instance.clone()),
s(|r| r.target_system.clone()),
s(|r| r.target_instance.clone()),
iv(|r| r.ts_micros),
i(|r| r.commit_seq),
];
RecordBatch::try_new(Arc::new(lineage_events_schema()), cols).map_err(|e| format!("lineage_events batch: {e}"))
}
/// The stable node id for a dataset: system-qualified so a `spark` `events` and a skade
/// `main.events` never collide. `"table"` for a warehouse-local dataset,
/// `"kind:instance/table"` for a foreign one.
fn dataset_key(system: Option<&str>, instance: Option<&str>, table: &str) -> String {
match (system, instance) {
(Some(k), Some(i)) => format!("{k}:{i}/{table}"),
(Some(k), None) => format!("{k}/{table}"),
_ => table.to_string(),
}
}
/// Build the **lineage [`PropGraph`]** from rows: one node per distinct dataset
/// (labelled by its system kind — `"skade"` when local — so styling-by-label colours
/// datasets by system), one directed edge per **run** that declared an input
/// (`input → output`, edge type = the uppercased operation, carrying
/// actor/commit_seq/ts/event_id). Input-less events (external ingest, table create)
/// still contribute their output node.
#[must_use]
pub fn build_graph(rows: &[LineageRow]) -> PropGraph {
let mut g = PropGraph::new();
let mut seen: BTreeMap<String, ()> = BTreeMap::new();
let mut ensure = |g: &mut PropGraph, key: &str, table: &str, system: Option<&str>, snap: Option<i64>| {
if seen.insert(key.to_string(), ()).is_none() {
let label = system.unwrap_or("skade");
let mut node = Node::new(key, label).with("name", table).with("system", label);
if let Some(s) = snap {
node = node.with("snapshot", s.to_string());
}
g.add_node(node);
}
};
for r in rows {
let out_key = dataset_key(r.target_system.as_deref(), r.target_instance.as_deref(), &r.output_table);
ensure(&mut g, &out_key, &r.output_table, r.target_system.as_deref(), r.output_snapshot);
if let Some(in_table) = &r.input_table {
let in_key = dataset_key(r.source_system.as_deref(), r.source_instance.as_deref(), in_table);
ensure(&mut g, &in_key, in_table, r.source_system.as_deref(), r.input_snapshot);
let mut e = Edge::new(in_key, out_key, r.operation.to_uppercase())
.with("actor", r.actor.as_str())
.with("event_id", r.event_id.as_str())
.with("ts_micros", r.ts_micros.to_string());
if let Some(cs) = r.commit_seq {
e = e.with("commit_seq", cs.to_string());
}
g.add_edge(e);
}
}
g
}
/// The **LINEAGE viewer** [`Facet`] — a [`GraphNav`] over the lineage [`PropGraph`],
/// plus the source rows for a lineage-specific `state_json` summary. Every graph-DB IDE
/// gesture (query / select / layout / path / style / export) drives the inner navigator
/// directly.
#[derive(Clone, Debug)]
pub struct LineageView {
title: String,
rows: Vec<LineageRow>,
nav: GraphNav,
}
impl LineageView {
/// Build a lineage view from already-decoded rows.
#[must_use]
pub fn from_rows(title: impl Into<String>, rows: Vec<LineageRow>) -> Self {
let title = title.into();
let graph = build_graph(&rows);
let nav = GraphNav::new(title.clone(), graph);
Self { title, rows, nav }
}
/// Build a lineage view directly from a `lineage_events` [`RecordBatch`].
///
/// # Errors
/// Propagates a [`read_rows`] decode error.
pub fn from_record_batch(title: impl Into<String>, batch: &RecordBatch) -> Result<Self, String> {
let rows = read_rows(batch)?;
Ok(Self::from_rows(title, rows))
}
/// Build a lineage view from an **OpenLineage / OpenMetadata**-standard
/// `RunEvent` payload (a single event object or a top-level array) — the
/// open-wire twin of [`from_record_batch`](Self::from_record_batch), so a
/// feed POSTed to an OpenLineage collector or an OpenMetadata OpenLineage
/// ingestion endpoint renders through the exact same graph surface.
///
/// # Errors
/// Propagates an [`openlineage::read_run_events`] decode error.
pub fn from_openlineage(
title: impl Into<String>,
value: &serde_json::Value,
) -> Result<Self, String> {
let rows = openlineage::read_run_events(value)?;
Ok(Self::from_rows(title, rows))
}
/// A self-contained **demo lineage**: a Spark app ingesting into skade `main.raw`,
/// two transform hops fanning out to `main.clean` / `main.metrics`, a promotion to
/// `main.gold`, and an atomic `release` — exercising multiple systems, every
/// operation, a multi-hop DAG, cross-system edges, and commit ordering.
#[must_use]
pub fn demo() -> Self {
Self::from_rows("lineage", demo_rows())
}
/// The decoded source rows.
#[must_use]
pub fn rows(&self) -> &[LineageRow] {
&self.rows
}
/// The inner GRAPH NAVIGATOR (read).
#[must_use]
pub fn nav(&self) -> &GraphNav {
&self.nav
}
/// The inner GRAPH NAVIGATOR (mutable) — drive it with [`Msg`]s.
pub fn nav_mut(&mut self) -> &mut GraphNav {
&mut self.nav
}
/// Apply a navigator [`Msg`] (the robot-/CLI-addressable channel).
pub fn update(&mut self, msg: Msg) -> bool {
self.nav.update(msg)
}
/// The lineage-specific summary folded into [`state_json`](Facet::state_json) under
/// `"lineage"`: event/dataset/run counts, the operation histogram, the distinct
/// systems + actors, and the max commit cursor.
#[must_use]
pub fn lineage_json(&self) -> serde_json::Value {
let mut operations: BTreeMap<String, usize> = BTreeMap::new();
let mut systems: BTreeMap<String, ()> = BTreeMap::new();
let mut actors: BTreeMap<String, ()> = BTreeMap::new();
let mut runs = 0usize;
let mut max_seq: Option<i64> = None;
for r in &self.rows {
*operations.entry(r.operation.clone()).or_default() += 1;
actors.insert(r.actor.clone(), ());
systems.insert(r.source_system.clone().unwrap_or_else(|| "skade".into()), ());
systems.insert(r.target_system.clone().unwrap_or_else(|| "skade".into()), ());
if r.input_table.is_some() {
runs += 1;
}
if let Some(cs) = r.commit_seq {
max_seq = Some(max_seq.map_or(cs, |m| m.max(cs)));
}
}
serde_json::json!({
"events": self.rows.len(),
"datasets": self.nav.graph.nodes.len(),
"runs": runs,
"operations": operations,
"systems": systems.keys().cloned().collect::<Vec<_>>(),
"actors": actors.keys().cloned().collect::<Vec<_>>(),
"commit_seq_max": max_seq,
})
}
}
impl Facet for LineageView {
fn title(&self) -> &str {
&self.title
}
fn caps(&self) -> FacetCaps {
FacetCaps::NONE.themeable().selectable().resizable()
}
fn kind(&self) -> &'static str {
"lineage"
}
fn ui(&mut self, ui: &mut egui::Ui) {
// The whole GRAPH NAVIGATOR IDE renders the lineage DAG.
self.nav.ui(ui);
}
fn selection_json(&self) -> serde_json::Value {
self.nav.selection_json()
}
fn state_json(&self) -> serde_json::Value {
let mut v = self.nav.state_json();
if let Some(obj) = v.as_object_mut() {
obj.insert("lineage".to_string(), self.lineage_json());
obj.insert("title".to_string(), serde_json::Value::String(self.title.clone()));
}
v
}
}
/// The demo lineage rows (deterministic ts derived from `commit_seq`).
fn demo_rows() -> Vec<LineageRow> {
// ts is deterministic: a fixed epoch base + seq, so `state_json` is byte-stable.
let ts = |seq: i64| 1_700_000_000_000_000 + seq * 1_000_000;
let row = |seq: i64, actor: &str, op: &str, output: &str| LineageRow {
event_id: format!("{:016x}-{:08x}", ts(seq), seq),
actor: actor.into(),
operation: op.into(),
input_table: None,
input_snapshot: None,
output_table: output.into(),
output_snapshot: Some(seq),
source_system: None,
source_instance: None,
target_system: None,
target_instance: None,
ts_micros: ts(seq),
commit_seq: Some(seq),
};
let with_input = |mut r: LineageRow, table: &str, snap: Option<i64>| -> LineageRow {
r.input_table = Some(table.into());
r.input_snapshot = snap;
r
};
let from_system = |mut r: LineageRow, kind: &str, inst: &str| -> LineageRow {
r.source_system = Some(kind.into());
r.source_instance = Some(inst.into());
r
};
vec![
// 1. Table birth (no upstream).
row(1, "nornir/etl", "create", "main.raw"),
// 2. A Spark app ingests external events INTO skade main.raw (cross-system).
from_system(with_input(row(2, "spark/app-42", "append", "main.raw"), "events", None), "spark", "app-42"),
// 3. Clean transform: raw → clean (overwrite in place).
with_input(row(3, "nornir/etl", "overwrite", "main.clean"), "main.raw", Some(2)),
// 4. Metrics fan-out: raw → metrics.
with_input(row(4, "nornir/etl", "append", "main.metrics"), "main.raw", Some(2)),
// 5. Promote: clean → gold.
with_input(row(5, "nornir/etl", "append", "main.gold"), "main.clean", Some(3)),
// 6. Atomic release of gold (from clean).
with_input(row(6, "nornir/release-9", "release", "main.gold"), "main.clean", Some(3)),
]
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn schema_matches_the_reserved_lineage_events_columns() {
let s = lineage_events_schema();
let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names,
vec![
"event_id", "actor", "operation", "input_table", "input_snapshot", "output_table",
"output_snapshot", "source_system", "source_instance", "target_system",
"target_instance", "ts_micros", "commit_seq",
]
);
// The non-null columns match skade's nullability.
for req in ["event_id", "actor", "operation", "output_table", "ts_micros"] {
assert!(!s.field_with_name(req).unwrap().is_nullable(), "{req} is non-null");
}
}
#[test]
fn rows_round_trip_through_the_arrow_batch() {
let rows = demo_rows();
let batch = to_record_batch(&rows).unwrap();
assert_eq!(batch.schema().as_ref(), &lineage_events_schema());
let back = read_rows(&batch).unwrap();
assert_eq!(back, rows, "read_rows inverts to_record_batch");
}
#[test]
fn graph_maps_datasets_to_nodes_and_runs_to_edges() {
let rows = demo_rows();
let g = build_graph(&rows);
// 5 distinct datasets: spark:app-42/events + main.raw/clean/metrics/gold.
assert_eq!(g.nodes.len(), 5, "one node per distinct dataset");
// 5 runs declared an input (the create has none) → 5 edges.
assert_eq!(g.edges.len(), 5, "one edge per run with an input");
// The cross-system ingest edge exists: spark events → main.raw.
assert!(g.edges.iter().any(|e| e.from == "spark:app-42/events" && e.to == "main.raw"));
// Edge type carries the operation; the release edge is labelled RELEASE.
assert!(g.edges.iter().any(|e| e.etype == "RELEASE"));
}
#[test]
fn reader_rejects_a_batch_missing_a_required_column() {
use std::sync::Arc;
use arrow_schema::Schema;
let bad = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("actor", DataType::Utf8, false)])),
vec![Arc::new(StringArray::from_iter_values(["x"]))],
)
.unwrap();
assert!(read_rows(&bad).is_err(), "a batch without event_id/operation/output_table is rejected");
}
#[test]
fn state_json_carries_the_lineage_summary() {
let v = LineageView::demo();
let st = v.state_json();
let lin = &st["lineage"];
assert_eq!(lin["events"], 6);
assert_eq!(lin["datasets"], 5);
assert_eq!(lin["runs"], 5);
assert_eq!(lin["operations"]["append"], 3);
assert_eq!(lin["operations"]["release"], 1);
assert_eq!(lin["commit_seq_max"], 6);
// Two systems: skade (local) + spark (the ingest).
let systems = lin["systems"].as_array().unwrap();
assert!(systems.iter().any(|s| s == "spark"));
assert!(systems.iter().any(|s| s == "skade"));
// The graphnav surface still rides along (its canvas section is present).
assert!(st["canvas"]["total_nodes"] == 5);
}
#[test]
fn state_is_deterministic() {
assert_eq!(LineageView::demo().state_json(), LineageView::demo().state_json());
}
#[test]
fn demo_renders_headless_through_the_facet_harness() {
// The fixture the `lineage_panel` demo mounts must actually paint: render it
// offscreen (no GPU) and assert it tessellated to vertices and reported its
// lineage summary — the same panel the wasm/native host runs.
use facett_core::harness;
let mut view = LineageView::demo();
let r = harness::headless_render(&mut view);
assert_eq!(r.title, "lineage");
assert!(r.drew(), "the lineage DAG tessellates to vertices");
assert_eq!(r.state["lineage"]["datasets"], 5);
}
}