nordisk/skade

CDC · streaming · lineage — guide

Class: guide · Function: cdc-stream. The what is in cdc-stream-design.md; the dead-simple 10-line version is in 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.

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.

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.

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.

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.

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):

// 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:

// 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:

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:

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):

cargo test --test streaming    # CommitEvent broadcast, commits_since, gapless commit_seq