# CDC · streaming · lineage — guide
> **Class:** `guide` · **Function:** `cdc-stream`. The *what* is in
> [`cdc-stream-design.md`](cdc-stream-design.md); the dead-simple 10-line
> version is in [`cdc-stream-for-idiots.md`](cdc-stream-for-idiots.md).
Four things you can do with a skade table's change history: **read a changelog
window**, **tail changes live**, **do a point-in-time lookup**, and **emit
lineage**. The first and third are always available; the live tail is behind the
`stream` feature and lineage behind `lineage` / `lineage-http`.
## Read a changelog window (`read_changelog`)
Always on. Every changed row carries a `_change_type` tag (`INSERT` /
`UPDATE_BEFORE` / `UPDATE_AFTER` / `DELETE` — the `skade::change_type`
constants). `from = None` reads full history up to `to`; a snapshot id reads only
the window after it.
```rust
use skade::change_type;
let to = t.current_snapshot_id().expect("a snapshot");
let cl = t.read_changelog(None, to).await?; // full history as a changelog
for b in &cl.rows {
let tags = b.column_by_name(change_type::COLUMN).unwrap(); // "_change_type"
// … each row is an INSERT (or DELETE, widened to the schema with nulls).
}
println!("{} change rows to snapshot {}", cl.num_rows(), cl.to_snapshot);
```
**Net change + column pruning.** `.collapsed()` folds the window to net change
(insert-then-delete cancels on a log table; delete+insert becomes
`UPDATE_BEFORE`/`UPDATE_AFTER` on a primary-key table). `.project_columns()`
keeps only the columns you name (the `_change_type` sentinel is always kept).
Collapse **before** you prune — collapse needs the identity columns.
```rust
let net = t.read_changelog(from, to).await?
.collapsed()? // net change
.project_columns(&["id", "name"])?; // Fluss columnar win
```
Whether a table collapses as a log or a PK table comes from its Iceberg
identifier-field-ids (`skade::TableKind::of`). When a window can't be expressed
as an additive delta (an overwrite, or a position delete), `read_changelog`
transparently re-reads the whole `to` snapshot as an `INSERT` stream —
`cl.plan.needs_full_reload` tells you it did.
## Tail a table's changes live (`ChangeStream`, feature `stream`)
`t.changes(from)` returns a live `BoxStream` of changelog windows — one per
commit — driven by the catalog's in-process commit broadcast (sub-millisecond,
no Kafka). `from = None` starts at "now" (future commits only); `Some(s)` first
emits a catch-up window `read_changelog(s, current)`, then tails.
```rust
use futures::StreamExt;
use skade::ChangeStream; // feature = "stream"
let mut stream = t.changes(None).await?;
t.append(&[batch_a]).await?; // (another task / handle commits)
t.append(&[batch_b]).await?;
let w1 = stream.next().await.unwrap()?; // window for commit A
let w2 = stream.next().await.unwrap()?; // window for commit B
```
**Exactly-once resume.** A window's `to_snapshot` is the resume key. Persist a
`ConsumerCursor`, restart with `changes(cursor.snapshot_id)`, and the first
window covers only rows strictly after it — nothing is delivered twice.
```rust
use skade::ConsumerCursor;
let w = stream.next().await.unwrap()?;
let cursor = ConsumerCursor::after("main.events", &w); // save cursor.snapshot_id
// … process crashes / restarts …
let mut resumed = t.changes(cursor.snapshot_id).await?; // resumes after the cursor
```
**No broadcast?** `t.changes_polling(from, interval)` is the same stream with a
pure timer poll instead of the broadcast — for a consumer whose producer runs in
another process. If a subscriber lags and the broadcast ring drops events, the
push stream self-heals by catching up to the current snapshot in one window (the
durable catalog is the truth), so you never miss a change either way.
## Point-in-time lookup (`Table::lookup`)
Always on. Resolve the current row for an equality key, pinned to a snapshot
(time-travel) or the latest. It runs merge-on-read, so an upserted key resolves
to its new row and a deleted key resolves to `None`.
```rust
use skade::Scalar;
let latest = t.lookup(&[("id", Scalar::I64(7))], None).await?; // latest
let pinned = t.lookup(&[("id", Scalar::I64(7))], Some(snap)).await?; // as of `snap`
// Multi-column key: &[("region", "eu".into()), ("id", Scalar::I64(7))]
```
This is the Flink/Fluss "delta-join" probe done embedded — the catalog *is* the
point-lookup store, no external state service.
## Emit lineage (`WarehouseLineageSink` / `HttpLineageSink`)
Turn on the `lineage` feature and attach a sink to a table handle; every
`append` / `ingest*` / `delete_equality` then emits one best-effort
`LineageEvent` post-commit (a sink error never fails the write).
**Historize into a `lineage_events` table** (skade writing skade — it then gets
time-travel and CDC like any other table):
```rust
// feature = "lineage"
let sink = wh.lineage_sink().await?; // ensures the reserved table exists
let mut t = wh.create_table("events", &schema).await?
.with_lineage_sink(sink) // Arc<dyn LineageSink>
.with_actor("nornir/ingest-7"); // who did it (default "skade")
t.append(&[batch]).await?; // → one Append row in lineage_events
```
A self-reference guard drops any event whose output *is* `lineage_events`, so
the lineage feed never recurses. For an atomic multi-table release, emit one
`Release` event naming every table with
`wh.atomic_release_with_lineage(commits, actor, &sink).await?`.
**Post to an OpenLineage collector** (Marquez et al.) — feature `lineage-http`,
so skade commits show up next to Spark/Airflow runs:
```rust
// feature = "lineage-http"
use skade::HttpLineageSink;
use std::sync::Arc;
let sink: Arc<dyn skade::LineageSink> =
Arc::new(HttpLineageSink::new("http://localhost:5000/api/v1/lineage"));
let mut t = wh.create_table("events", &schema).await?.with_lineage_sink(sink);
// each append POSTs an OpenLineage RunEvent (eventType COMPLETE), fenced onto
// spawn_blocking so the async runtime never stalls.
```
Inspect the exact wire body without a live POST via
`HttpLineageSink::run_event(&event)` (it returns the `serde_json::Value`).
## Subscribing at the catalog level
Below the per-table `ChangeStream`, `RedbCatalog` exposes the raw spine — a
receiver plus a durable replay for catch-up:
```rust
let (cursor, mut rx) = catalog.subscribe_commits().await?; // (commit_seq, Receiver)
let key = catalog.commit_table_key(&ident); // filter the stream
while let Ok(ev) = rx.recv().await {
if ev.table_key.as_ref() == key { /* ev.snapshot_id, ev.commit_seq, … */ }
}
// After a RecvError::Lagged, replay the gap from the durable log:
for ev in catalog.commits_since(cursor).await? { /* … */ }
```
## Delivering a window into a query engine
skade computes the diff; it deliberately does **not** deliver it. Delivery lives
in `knut-bifrost`, which already polls, reads `read_changelog`, splits the
`_change_type` tags and keeps a durable cursor — see knut's
`.nornir/doris-sink-design.md` for the `TableSink` seam and its Doris
implementation.
Two things a consumer must get right, documented there and repeated here because
they are skade-side facts:
* **Commit the consumer cursor only after the sink acknowledges.** The other
order loses a window silently on a crash.
* **Check `DeltaPlan::needs_full_reload` via `read_delta` *before* calling
`read_changelog`.** `read_changelog` handles a full-reload window by re-reading
the whole `to` snapshot (by design — see `cdc-stream-design.md` f2), so a
consumer that checks the flag on the returned batch has already paid for the
read it wanted to avoid. `read_delta` returns the flag after a
snapshot-lineage walk and nothing else.
## Running the tests (detached `skade/` workspace)
The CDC/streaming/lineage tests live in the detached `skade/` workspace. From
the repo root, `cd skade`, then:
```bash
cargo test --test changelog # read_changelog + collapse + pruning
cargo test --test lookup # point lookup + eq-delete full-scan
cargo test --test stream --features stream # ChangeStream push + poll + resume
cargo test --test lineage --features lineage # emit hooks + WarehouseLineageSink
cargo test --test lineage_http --features lineage-http # OpenLineage RunEvent mapping + POST
```
The Phase-0 catalog spine is tested in the top-level `skade-katalog` crate
(repo root):
```bash
cargo test --test streaming # CommitEvent broadcast, commits_since, gapless commit_seq
```