<!-- nornir:class: reasoning -->
# skade — CDC + Streaming, Lineage Emission, and a facett Dataflow System-Map
> **Status:** design (2026-06-30, Rickard item 10 "this is big"). Author: research
> agent. Companion to `.nornir/streaming-composability-lineage-proposal.md` (the
> commit-event/OpenLineage spine — call it **SCL**) and
> `.nornir/incremental-read.md` (`read_delta` CDC). This doc makes three things
> concrete and *incrementally buildable*: (a) a CDC/changelog + streaming API off
> skade's Iceberg snapshots, (b) a **lineage event model** skade emits on commit
> and historizes into a `lineage_events` Iceberg table, and (c) a **facett
> dataflow / lineage system-map viewer** over a subset of skade tables that fuses
> **Spark and skade pipelines** into one DAG.
>
> A **compiling spike** of (b) already exists on branch `agent/skade-lineage-spike`
> (worktree `../skade-lineage-spike`): `skade::lineage` — the event model +
> `LineageSink` trait + an `append` emit hook gated behind a default-off `lineage`
> feature, one green unit test. See §6.
>
> **Grounding rule:** every concrete claim cites `file:line` or a sibling repo
> shape. The three workstreams are independent; a recommended order is in §7.
---
## 0. Executive summary
skade already has the hard parts: one durable commit chokepoint with a monotonic
`commit_seq` (`src/store.rs:233` `group_commit`, `:111` `record_commit`,
`:82` `META_COMMIT_SEQ`), an immutable `COMMITS` log (`src/store.rs:77`), and a
pure-Rust incremental/CDC read — `read_delta` + `DeltaPlan { snapshots,
added_files, delete_files, needs_full_reload }` with real equality-delete CDC
(`skade/src/read.rs:369`, `.nornir/incremental-read.md`). knut-bifrost *already
consumes this exact shape* against skade (`knut-bifrost/src/source.rs` `mod
iceberg`, feature `skade`). So this is **not** a from-scratch build — it is three
thin layers over existing chokepoints:
1. **CDC + streaming (§3).** Promote `read_delta` to a *changelog* (`_change_type`
rows, knut-bifrost's vocabulary) and add a **push** subscription
(`subscribe_commits` → `CommitEvent` broadcast, SCL §1) so consumers stop
polling `commit_seq()`. The streaming *append/consume* API is a small trait —
`ChangeStream` (push delta windows) + a `Changelog` row encoder.
2. **Lineage emission (§4).** A `LineageEvent { actor, operation, inputs[],
output, ts_micros, commit_seq, system }` minted at the **same post-commit hook**
as the `CommitEvent`, landed via a `LineageSink` into a historized
**`lineage_events` Iceberg table** (skade writing skade — dogfood). The
*between-which-systems* axis is `SystemRef` on each `DatasetRef`. This is the
spike (§6).
3. **facett viewer (§5).** A `LineageView` built on **`facett-pipeline`**
(`PipelineModel` + `PipelineAdapter`, `facett-pipeline/src/model.rs`): tables =
`Stage` nodes, lineage edges = `PipelineEdge{from,to}`, systems = node groups.
Reuse `facett-graphview::downstream_of` for impact-trace, `facett-icebergview`
for node drill-down, `facett-warehousedeck` as the shell. **A skade
`PipelineAdapter` and a Spark/SDP `PipelineAdapter` feed the same view** — that
is the "system map of actual pipelines (Spark + skade) tied together".
**One hook, three payoffs:** the post-commit point in `group_commit` already
fires the `CommitEvent` (SCL §1); the lineage event and the changelog watermark
ride the *same* moment. Build that once, and CDC-push, lineage, and the viewer are
all downstream subscribers.
---
## 1. What already exists (don't reinvent)
| Capability | Where | Note |
|---|---|---|
| Durable commit chokepoint + monotonic cursor | `src/store.rs:233`,`:111`,`:82` | every mutation funnels here |
| Immutable commit log (`{table_key}\x1f{snap_be} → loc`) | `src/store.rs:77`,`:99` | the durable truth behind any stream |
| Incremental read (manifest diff) | `skade/src/read.rs:387` `read_delta` | inserts **and** equality deletes |
| `DeltaPlan` (snapshots/added/delete/full-reload) | `skade/src/read.rs:369` | the CDC unit, already public |
| Equality-delete CDC writer (no Spark) | `skade/src/delete.rs` `delete_equality` | real Iceberg `Delete` snapshot |
| Commit-event broadcast + `subscribe_commits` | **SCL §1** (designed, not yet built) | `CommitEvent{table,snapshot_id,commit_seq,ts}` |
| OpenLineage `RunEvent` mapping | **SCL §4** (designed) | this doc's lineage model is the in-warehouse twin |
| Git-like snapshot/branch/tag surface | `.nornir/git-like-catalog.md` | lineage edges can carry a `ref` (branch) later |
**knut precedents to mirror (sibling repo `/home/rickard/git/knut`):**
- **CDC change record = Arrow row + `_change_type` string column** (`"INSERT"`,
`"UPDATE_AFTER"`, `"UPDATE_BEFORE"`, `"DELETE"`), not a bespoke struct —
`knut-bifrost/src/mapping.rs:43` (`SyncMode::{Full,Incremental,Cdc}`,
`CHANGE_TYPE_COLUMN`). Offset/LSN is **table-level** = Iceberg `snapshot_id`
(`knut-bifrost/src/state.rs:25` `TableState{snapshot_id, watermark, …}`).
- **Routing** = `partition_change_rows(&[RecordBatch]) -> ChangeRows{upserts,
deletes}` (`knut-bifrost/src/transform.rs:185`); the **listener** polls
`TableSource::fetch_new(spec, state) -> Option<(batches, new_state)>`
(`source.rs:9`), one `sync_once()` per round (`listener.rs:127`).
- **A skade Iceberg-CDC source already exists** in bifrost (`source.rs` `mod
iceberg`, feature `skade`): its `DeltaPlan` is a structural copy of skade's, and
it tags equality-deletes `_change_type=DELETE` via `tag_change_type`. **skade
should own this shape so bifrost imports it instead of re-declaring it.**
- **Pipeline DAG model** = `DataflowGraph{ datasets: Vec<Dataset>, flows:
Vec<Flow> }`, lineage = each `Flow.reads -> Flow.target`
(`knut-pipelines/src/model.rs:87,132,185`); `OutputType{MaterializedView,Table,
TemporaryView,Sink}` marks system boundaries; live state folds into
`ViewerState`/`ViewerUpdate` (`progress.rs:187,397`). An engine-neutral adapter
to facett already exists: `knut-pipelines/src/facett_adapter.rs` maps
`ViewerState → facett PipelineModel`. **The skade lineage source is another such
adapter.**
---
## 2. The shape of the whole thing
```
WRITE PATH (skade) READ/STREAM PATH VIEW (facett)
append/ingest/delete_equality ┌────────────────────┐
│ │ LineageView │
▼ │ (facett-pipeline) │
persist_commit ─► group_commit ──post-commit hook──┬─► CommitEvent broadcast ──┤ Stage = table │
(catalog.rs:317) (store.rs:233) │ (SCL §1) │ Edge = lineage │
│ writes COMMITS + commit_seq │ └─► ChangeStream ────┤ Group = system │
▼ │ (push read_delta) │ downstream_of() │
lineage_events ◄──LineageSink.emit──◄─────────────┴─► LineageEvent ───────────┤ icebergview drill │
(Iceberg table) (this doc §4 / spike) {actor,op,in,out,sys} └────────────────────┘
▲
Spark/SDP run ──(OpenLineage RunEvent / SDP ViewerState)──► lineage_events ───────────────┘
(knut-pipelines) (same table, system="spark")
```
The lineage table is the **fusion point**: both skade commits *and* Spark runs
write rows into it, the viewer reads it as one DAG.
---
## 3. Workstream A — CDC + streaming
### 3.1 The change-data-capture event shape
Adopt knut-bifrost's vocabulary verbatim so a skade changelog and a bifrost
consumer agree with **zero translation**. A change "record" is an **Arrow row
carrying a `_change_type` Utf8 column**; the unit of work is a `DeltaPlan` window
keyed by `snapshot_id`.
```rust
// skade/src/read.rs — promote DeltaPlan into a changelog view (additive).
/// CDC op vocabulary — identical strings to knut-bifrost's mapping.rs:43.
pub mod change_type {
pub const INSERT: &str = "INSERT";
pub const UPDATE_BEFORE: &str = "UPDATE_BEFORE"; // pre-image (usually dropped)
pub const UPDATE_AFTER: &str = "UPDATE_AFTER";
pub const DELETE: &str = "DELETE";
pub const COLUMN: &str = "_change_type"; // sentinel column name
}
/// A changelog batch set for a [`DeltaPlan`] window: insert rows tagged
/// `_change_type=INSERT`, equality-deleted identities tagged `DELETE`. Built from
/// the existing `read_delta` + `read_equality_deletes`, so this is a thin
/// re-tagging layer, NOT a new scan path.
pub struct ChangelogBatch {
pub rows: Vec<RecordBatch>, // carries the _change_type column
pub plan: DeltaPlan, // the snapshots/files this window covered
pub to_snapshot: i64, // the new cursor (advance to here)
}
/// `read_changelog(table, from, to)` = `read_delta` inserts tagged INSERT
/// ∪ `read_equality_deletes` identities tagged DELETE, schema-aligned. When
/// `plan.needs_full_reload` is set (overwrite / position-delete), the caller
/// re-reads `to` in full — exactly bifrost's rule.
pub async fn read_changelog(table: &IceTable, from: Option<i64>, to: i64)
-> Result<ChangelogBatch> { /* wraps read_delta + read_equality_deletes */ }
```
This is the natural home for the code bifrost currently keeps private
(`knut-bifrost/src/source.rs` `tag_change_type`/`widen_to_schema`): **lift it into
skade** so bifrost depends on `skade::read_changelog` and deletes its copy.
### 3.2 Streaming consume API — push, not poll
Today every CDC consumer (bifrost `listener.rs:293` `run`, modgunn's world-delta
watcher, `.nornir/incremental-read.md`) **polls** `commit_seq()` then
`read_delta(cursor, latest)`. Replace the poll with SCL §1's broadcast and wrap it
in a stream:
```rust
// skade/src/stream.rs (new, feature `stream`) — push CDC over the commit broadcast.
/// A live changelog stream for one table: each item is a CDC window from the
/// consumer's last cursor to a freshly-committed snapshot. Built on
/// `RedbCatalog::subscribe_commits()` (SCL §1) — a `CommitEvent` for this table
/// triggers a `read_changelog(last, event.snapshot_id)`. Degrades to polling if
/// the catalog has no broadcast (older builds): the same trait, a poll backend.
#[async_trait]
pub trait ChangeStream: Send + Sync {
/// Yield CDC windows as the table advances. `from = None` starts at "now".
async fn changes(&self, from: Option<i64>)
-> Result<BoxStream<'static, Result<ChangelogBatch>>>;
}
```
`impl ChangeStream for Table` = `subscribe_commits()` filtered to this table →
for each event `read_changelog(last_cursor, event.snapshot_id)` → yield, advance.
A lagged subscriber recovers via SCL's `commits_since(cursor)` (the durable
`COMMITS` log is the truth; the broadcast is the hint). This is **exactly**
bifrost's `Listener::run` loop, inverted from poll to push — bifrost reimplements
itself as `let mut s = table.changes(Some(cursor)).await?; while let Some(w) =
s.next().await { sink.apply(w) }`.
### 3.3 Streaming append / produce
The write side already streams: `ingest_pipelined` (`skade/src/write.rs:435`) is a
bounded-channel gatling writer, and `delete_equality` (`skade/src/delete.rs`)
produces CDC deletes. The only addition is that each commit **wakes** the
`ChangeStream` (via the broadcast) instead of waiting for the next poll tick — so
a producer→consumer hop is sub-millisecond locally (SCL §1.2). No new write API is
needed; the streaming-ness is the *push* on the read side.
### 3.4 Effort / risk
- `read_changelog` + `ChangelogBatch`: ~0.5d (re-tag over existing `read_delta`).
Low risk; bifrost's logic is the reference.
- `ChangeStream` + `impl for Table` (feature `stream`): ~1d, depends on SCL §1's
broadcast. Low risk (additive; poll fallback).
- **Reuse win:** bifrost deletes ~200 lines of private Iceberg-CDC code and
imports skade's. Validates the shape by a real second consumer.
---
## 4. Workstream B — Lineage emission
### 4.1 The event model (built — see spike §6)
```rust
// skade/src/lineage.rs (spike: branch agent/skade-lineage-spike)
pub enum Operation { Append, Delete, Overwrite, Create, Release }
pub struct SystemRef { pub kind: String, pub instance: String } // "spark","skade",…
pub struct DatasetRef { pub table: String, pub snapshot_id: Option<i64>,
pub system: Option<SystemRef> } // None = this warehouse
pub struct LineageEvent {
pub event_id: String, // process-unique, time-ordered
pub actor: String, // who: user / service / pipeline-run id
pub operation: Operation, // what
pub inputs: Vec<DatasetRef>, // which tables in (empty = external ingest)
pub output: DatasetRef, // which table out
pub ts_micros: i64, // when (post-commit)
pub commit_seq: Option<u64>, // ties edge to the exact catalog commit
}
#[async_trait] pub trait LineageSink { async fn emit(&self, e: &LineageEvent) -> Result<()>; }
pub struct CapturingSink { /* Arc<Mutex<Vec<LineageEvent>>> */ } // tests + viewer live-tail
```
**Why this shape:** it is the prompt's model `{actor, operation, input
tables/snapshots, output table/snapshot, timestamp, optional source/target
SYSTEM}` made precise, and it is **isomorphic to the viewer's edge**: a
`LineageEvent` with `inputs=[A]`, `output=B` *is* the facett `PipelineEdge{from:A,
to:B}` (§5) and the knut-pipelines `Flow{reads:[A], target:B}`
(`knut-pipelines/src/model.rs:132`). The `system` axis is the
`OutputType::Sink`/external-read distinction made explicit. It is also the
in-warehouse twin of SCL §4's OpenLineage `RunEvent` (one mapping function both
ways: `inputs→inputs[]`, `output→outputs[]`, `actor→job.namespace`).
### 4.2 Where it is emitted
The same **post-commit moment** as the `CommitEvent` (SCL §1.1,
`src/store.rs:263` after `write.commit()`), surfaced ergonomically per write API:
- **Spike level (built):** `Table::append` emits an `Append` event when a
`LineageSink` is attached (`Table::with_lineage_sink`, feature `lineage`,
default off). `inputs` empty (plain ingest); `output` = the table's new
snapshot. Best-effort: a sink error never fails the write.
- **Next:** `delete_equality` → `Delete`; `ingest*` → one `Append` per commit (or
one summarizing event per ingest run); `atomic_release` (`src/atomic.rs:46`) →
one `Release` event whose `inputs`/`output` name all N tables (the atomic batch
*is* one logical job — semantically perfect, matches SCL §4.2).
- **Transform hops** (the interesting lineage): when a caller reads tables A,B and
writes C, it supplies `inputs=[A@snap, B@snap]` via a builder
(`LineageEvent::new(op, out).with_input(A).with_input(B)`). skade can't infer
arbitrary upstreams, so the *producing job* declares them — exactly how SDP's
`Flow.reads` is declared, not inferred.
For the **catalog-level** path (multi-table `atomic_release`, `create_table`,
`register_table`), the emit hook lives in skade-katalog and rides the *same*
broadcast the `CommitEvent` does — a `LineageSink` is **just another subscriber**
to `subscribe_commits()`, so lineage never couples to the writer (SCL §4.3).
### 4.3 Where it lands — a historized `lineage_events` Iceberg table
skade writes skade. A `WarehouseLineageSink` appends each `LineageEvent` as a row
into a reserved **`lineage_events`** Iceberg table in the warehouse:
```
lineage_events schema (Iceberg / Arrow):
event_id string actor string operation string
input_table string (repeated / first input; or a child `inputs` list column)
input_snapshot long output_table string output_snapshot long
source_system string source_instance string
target_system string target_instance string
ts_micros long commit_seq long
```
Because it is a normal Iceberg table, it gets **time-travel, `read_delta`, the
git-like snapshot log, and the CDC stream for free** — the viewer (§5) reads it
with the very same `read_changelog`/`ChangeStream` this doc defines, and modgunn's
world-delta watcher can watch lineage the same way it watches any table. The sink
batches rows (one append per N events or per flush interval) so high-rate ingest
doesn't commit per-event. A `WarehouseLineageSink` plus an `HttpLineageSink`
(OpenLineage POST, SCL §4.3) plus the `CapturingSink` (tests) are the three
`LineageSink` impls; the trait keeps them interchangeable.
> **Self-reference guard:** the lineage sink must NOT emit lineage for writes to
> `lineage_events` itself (filter `output.table == "lineage_events"`), or it
> recurses. One-line guard in `WarehouseLineageSink::emit`.
### 4.4 Effort / risk
- Event model + `LineageSink` + `CapturingSink` + `append` hook: **done** (spike).
- `WarehouseLineageSink` (append into `lineage_events`, batched, self-guarded):
~1d. Low risk (uses skade's own `Table::append`).
- Hooks on `delete_equality`/`ingest*`/`atomic_release` + transform-input builder:
~1d. Low risk (additive).
- OpenLineage `HttpLineageSink` + `RunEvent` map: SCL §4 (~1.5d), reuses this model.
---
## 5. Workstream C — facett dataflow / lineage system-map viewer
### 5.1 Reuse, don't build (survey result)
The facett survey (over `facett-graph`, `-graphview`, `-pipeline`, `-syschart`,
`-icebergview`, `-warehousedeck`, and the parity doc) is unambiguous: **build on
`facett-pipeline`** and feed it via a `PipelineAdapter`.
| Need | Component (today) | Path |
|---|---|---|
| Tables as nodes, lineage as **directed** edges, toposort L→R, pan/zoom, animated dataflow direction, per-node status/metrics, click-preview | **`facett-pipeline::PipelineView` + `PipelineModel`** | `facett-pipeline/src/{lib,model}.rs` |
| Engine-agnostic feed (the seam) | **`PipelineAdapter::to_pipeline_model()`** | `facett-pipeline/src/model.rs` |
| "Light up everything downstream of table X" (impact trace) | `facett-graphview::downstream_of` + `DecoratedGraphView` | `facett-graphview/src/decorated_view.rs` |
| Large graphs / caller-positioned swimlanes / status rings+badges | `facett-graphview` (`GraphModel` world-space `Pos` + `Decorations`) | `facett-graphview/src/model.rs` |
| Node drill-down → table preview + snapshot time-travel | `facett-icebergview::IcebergView` | `facett-icebergview/src/lib.rs` |
| Multi-pane shell + table→view registry | `facett-warehousedeck` / `FacetDeck` | `facett-warehousedeck/src/lib.rs` |
`PipelineModel` *is* the lineage model:
```rust
// facett-pipeline/src/model.rs — already exists
Stage { id, name, status: StageStatus, kind: Option<String>, error, metrics }
PipelineEdge { from: String /*upstream*/, to: String /*downstream*/, label }
PipelineModel { title, stages, edges, metrics }
trait PipelineAdapter { fn to_pipeline_model(&self) -> PipelineModel; }
```
Map: a distinct `(system, table)` → one `Stage` (id = `"{system}:{table}"`,
`kind` = the `system`); each `LineageEvent` → `PipelineEdge` per `input→output`;
`Stage.metrics.rows`/`status` from the latest snapshot / freshness. Done.
### 5.2 `facett-lineage` (new, thin) — the adapter + the view
A small new crate (or a module in `facett-warehousedeck`) with **one job**: read a
subset of skade tables — the **`lineage_events`** table (§4.3) plus the catalog's
table list for node metadata — and implement `PipelineAdapter`:
```rust
// facett-lineage/src/lib.rs (sketch)
pub struct LineageSource { // reads skade's lineage_events table
pub events: Vec<LineageEvent>, // from read_all / read_changelog (live tail)
pub tables: Vec<TableNodeMeta>, // name, latest snapshot, rows, last_write_ts
}
impl PipelineAdapter for LineageSource {
fn to_pipeline_model(&self) -> PipelineModel {
// nodes = distinct (system, table); edges = events' input→output;
// group/colour by system; status from freshness; rows from table meta.
}
}
pub struct LineageView { /* PipelineView::from_adapter(&LineageSource) */ }
```
`LineageView` then = `PipelineView::from_adapter(&source)` for free: directional
animated edges, pan/zoom, click-select, `take_preview_request()` → open a
`facett-icebergview` pane on the clicked table (snapshot time-travel as
drill-down). Impact-trace ("what breaks if `raw_samples` is wrong?") =
`facett-graphview::downstream_of(model, "skade:raw_samples")`.
### 5.3 Systems as lanes — the honest state
The survey found **no existing swimlane/lane primitive**; lanes are design-only in
the parity doc (`facett-timeline` = temporal Gantt lanes; `facett-plandag` =
collapsible spatial groups). Two pragmatic paths, both incremental:
- **Now (spatial grouping):** colour + cluster nodes by `system` using
`facett-graphview`'s caller-owned `Pos` — compute per-system Y-bands, feed world
positions into `GraphModel`. Gives "Spark lane above, skade lane below" without
a new primitive. This is the `facett-plandag` "collapsible system-groups" idea
approximated today.
- **Later (temporal):** when `facett-timeline` lands (parity doc P1), a second
pane shows the *same* lineage as wall-clock `Lane{label=system, rows:[Bar]}` —
when each table/snapshot was produced. Reuses the identical `LineageSource`.
### 5.4 Effort / risk
- `facett-lineage` (`LineageSource` + `PipelineAdapter` + `LineageView` wrapper):
~1.5-2d. Low risk — it's an adapter over `PipelineView`, the heavy lifting
(layout/render/nav/animation) already exists in `facett-pipeline`.
- icebergview drill-down wiring + `downstream_of` impact trace: ~1d.
- system-lane spatial grouping via `facett-graphview`: ~1-2d (optional polish).
---
## 6. The spike (built, branch `agent/skade-lineage-spike`)
Worktree `../skade-lineage-spike`, branch `agent/skade-lineage-spike`, **not
pushed**. Files (`skade` data-plane crate):
- `skade/src/lineage.rs` — `Operation`, `SystemRef`, `DatasetRef`, `LineageEvent`
(+ builders `with_actor`/`with_input`/`with_commit_seq`/`with_target_system`),
`LineageSink` trait, `CapturingSink`. **Types always compile** (only `serde` +
`async-trait`, both already deps).
- `skade/src/lib.rs` — `pub mod lineage;` + re-exports.
- `skade/src/table.rs` — `Table::with_lineage_sink` + a `#[cfg(feature =
"lineage")]` emit in `append` (post-commit, best-effort).
- `skade/Cargo.toml` — `lineage = []` feature (default off) + `serde_json`
dev-dep.
**Status:** `cargo build` green, `cargo build --features lineage` green, `cargo
build --tests --features lineage` green. One unit test
(`lineage::tests::append_event_captures_actor_op_output_and_input`) passes **both**
with and without the feature — inject-and-assert on every field, the `CapturingSink`
async round-trip, id uniqueness, a cross-system target edge, and the serde
round-trip (the `lineage_events`-row form). Per the project LAW, wire a
`functional_status` row under `--features testmatrix` when productionized.
**Next commits off this branch:** `WarehouseLineageSink` (§4.3), the
`delete_equality`/`atomic_release` hooks (§4.2), then `read_changelog` (§3.1).
---
## 7. How it composes + build order
### 7.1 Spark + skade tied together (the prompt's headline)
The fusion is the **`lineage_events` table** (§4.3) read through **one
`PipelineAdapter`** (§5.1):
- **skade side:** every commit emits a `LineageEvent` (`system=None`/`"skade"`).
- **Spark/SDP side:** knut-pipelines already models the SDP DAG (`DataflowGraph`,
`model.rs:185`) and already maps it to facett (`facett_adapter.rs`). Point that
adapter — or a small Spark-OpenLineage ingester — at the **same**
`lineage_events` table with `system="spark"`, `instance=<app-id>`. Spark reads
skade tables via the Iceberg REST catalog (SCL §3.3, `bench/src/rest_shim.rs`),
so a Spark `Flow{reads:["skade.raw"], target:"gold"}` becomes a lineage edge
whose **input is a skade table and output is a Spark dataset** — a real
cross-system edge with `source_system=skade`, `target_system=spark`.
- **The viewer renders both** as one DAG: skade nodes and Spark nodes in different
system-lanes, edges crossing between them. "A system map with actual pipelines
(Spark and skade) tied together as a dataflow" — exactly the deliverable.
### 7.2 Composition with the Spark-UI parity work
This shares the parity work's spine: both consume **`facett-pipeline`'s
`PipelineModel`/`PipelineAdapter`** and the proposed `facett-plandag`/
`facett-timeline` (parity doc). The lineage viewer is a *third* `PipelineAdapter`
alongside the SDP `ViewerState` adapter — same render path, same nav, same theming
(`facett_core::nav`, `effects::glow_rect`). When `facett-plandag` (metric-bearing,
collapsible groups) and `facett-timeline` (swimlanes) land, the lineage viewer
gets system-grouping and a temporal lane view **for free** by feeding them the
same `LineageSource`. Net: build the lineage adapter now on `PipelineView`; inherit
the parity components as they arrive.
### 7.3 Composition with modgunn
modgunn's world-snapshot / behaviour-delta is lineage-shaped: its Engine-B
world-delta watcher already polls `commit_seq()` + `read_delta`
(`.nornir/incremental-read.md`; `modgunn/.nornir/modgunn-master-plan.md` §1 CDC
seam). Two free wins: (1) modgunn switches its poll to the §3.2 `ChangeStream`
push; (2) modgunn's behaviour-deltas become `LineageEvent`s
(`actor=modgunn-run`, inputs=world tables, output=behaviour table) in the **same**
`lineage_events` table, so the facett viewer shows modgunn runs as just another
system-lane next to Spark and skade.
### 7.4 Recommended build order (incremental, each shippable)
1. **SCL §1 commit-event broadcast** (the spine — `subscribe_commits` +
`commits_since`). Everything pushes off this. ~1.5d.
2. **Lineage emit, finish the spike (§4):** `WarehouseLineageSink` into
`lineage_events` + `delete_equality`/`atomic_release` hooks. Rides §1 broadcast.
~2d. *(Event model + append hook already done.)*
3. **CDC changelog (§3.1) `read_changelog`** — lift bifrost's tagging into skade;
bifrost imports it. ~0.5d. Then **`ChangeStream` (§3.2)** push. ~1d.
4. **facett `LineageSource` + `LineageView` (§5.2)** over `lineage_events` via
`PipelineAdapter`. ~2d. Visible payoff: the dataflow map.
5. **Spark fusion (§7.1):** route knut-pipelines / Spark-OpenLineage into
`lineage_events` with `system="spark"`. ~1-2d. Now the map is cross-system.
6. **Polish:** system-lanes (§5.3), icebergview drill-down, `downstream_of` impact
trace; inherit `facett-plandag`/`facett-timeline` from parity work.
**~10 working days** to a cross-system dataflow map, each step independently
useful. Step 0's spike is already green.
---
## 8. Open questions (for the owner)
1. **`lineage_events` granularity:** one row per commit (cheap, append-rate) vs one
per logical job/run (needs a run-id grouping)? This doc assumes per-commit with
`commit_seq` as the natural key; a `run_id` column can group later.
2. **Transform-input declaration:** skade can't infer upstreams — the producing job
declares `inputs` (like SDP `Flow.reads`). OK, or do we want an opt-in
"lineage context" handle that records reads automatically within a scope?
3. **Spark→skade lineage source:** prefer (a) the SDP `ViewerState` adapter writing
`lineage_events`, or (b) a generic OpenLineage HTTP *ingest* endpoint skade
exposes that lands any producer's `RunEvent` as a row? (b) is more universal.
4. **Viewer crate home:** new `facett-lineage` crate, or a module inside
`facett-warehousedeck` (which already has a `release_lineage` table in its
`default_nordisk()` registry)?
5. **System taxonomy:** fixed `SystemRef.kind` vocabulary (`skade`/`spark`/
`kafka`/`falkordb`/`modgunn`) or free-form? Free-form now, lint later.
---
### Appendix — file:line index
- skade commit chokepoint: `src/store.rs:233`,`:111`,`:99`,`:77`,`:82`,`:263`;
catalog `src/catalog.rs:317`,`:79`; atomic `src/atomic.rs:46`.
- skade CDC read: `skade/src/read.rs:369` (`DeltaPlan`),`:387` (`read_delta`),
delete `skade/src/delete.rs`; doc `.nornir/incremental-read.md`.
- skade write: `skade/src/write.rs:281` (`append_props`),`:435` (`ingest_pipelined`);
table `skade/src/table.rs:115` (`append`),`:278` (`current_snapshot_id`).
- SCL (sibling design): `.nornir/streaming-composability-lineage-proposal.md`
§1 (broadcast), §3 (traits), §4 (OpenLineage).
- knut CDC: `knut-bifrost/src/mapping.rs:43`, `transform.rs:185`, `state.rs:25`,
`source.rs:9` + `mod iceberg`, `listener.rs:127`,`:293`, `airflow.rs`.
- knut pipelines: `knut-pipelines/src/model.rs:87,132,185,316`, `progress.rs:187,397`,
`facett_adapter.rs`, `client.rs`.
- facett: `facett-pipeline/src/{lib,model}.rs`, `facett-graphview/src/{model,
decorated_view,depgraph_layout}.rs`, `facett-icebergview/src/lib.rs`,
`facett-warehousedeck/src/lib.rs`, `.nornir/databricks-spark-ui-parity.md`
(`facett-plandag`, `facett-timeline` — design-only).
- spike: branch `agent/skade-lineage-spike`, `skade/src/lineage.rs`,
`skade/src/table.rs`, `skade/src/lib.rs`, `skade/Cargo.toml`.
```