.nornir/cdc-stream-for-idiots.md

Watching changes stream out — for idiots

Class: idiot · Function axis: f0 (zero → a live change feed). No prior knowledge assumed. The fuller version is in cdc-stream-guide.md; if you have never used skade at all, start with for-idiots.md first.

The idea in one sentence

You put rows into a table. Someone else (or a future you) wants to know every time a row is added or removed, the moment it happens — that is a change stream, and skade hands it to you as a normal Rust loop. No Kafka, no server, no database to run. It all happens inside your own program.

Step 1 — add skade with the `stream` feature

In your project's Cargo.toml, under [dependencies]:

skade = { version = "0.5", features = ["stream"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
futures = "0.3"
anyhow = "1"

The features = ["stream"] part is what turns on the live change feed.

Step 2 — the whole program

Open src/main.rs, delete everything, paste this in:

use std::sync::Arc;
use futures::StreamExt;                 // lets us call .next() on the stream
use skade::ChangeStream;                // the live-changes trait
use skade::arrow_array::{Int64Array, RecordBatch};
use skade::arrow_schema::{DataType, Field, Schema};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // 1. Open a lakehouse in a folder, and make a table with one column `id`.
    let wh = skade::open("/tmp/my-lake").await?;
    let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]);
    let mut table = wh.table_or_create("events", &schema).await?;

    // 2. Start watching for changes *before* we add anything.
    let mut changes = table.changes(None).await?;

    // 3. Append three rows (this is one "commit").
    let batch = RecordBatch::try_new(
        Arc::new(schema),
        vec![Arc::new(Int64Array::from(vec![1, 2, 3]))],
    )?;
    table.append(&[batch]).await?;

    // 4. The change feed hands us a "window" for that commit. Print each row's tag.
    let window = changes.next().await.unwrap()?;
    for b in &window.rows {
        let tags = b.column_by_name("_change_type").unwrap();
        println!("{tags:?}");   // three INSERT tags — one per row we added
    }
    println!("Saw {} changes.", window.num_rows());
    Ok(())
}

Step 3 — run it

cargo run

The first build downloads and compiles the ingredients (a few minutes — normal). Then you'll see three INSERT tags printed and Saw 3 changes.

What just happened

You made a table and subscribed to its changes with one line: table.changes(None).
You added rows. The change feed woke up and handed you a window describing exactly what changed — three rows, each tagged INSERT.
If you had removed a row instead (with delete_equality), it would show up tagged DELETE. Add more rows in a loop and each changes.next().await gives you the next batch, live, the instant it commits.

That is change-data-capture: your program reacts to every table change as it happens, and it all ran inside this one process — no server anywhere.

Where to go next

Want the changes to survive a restart exactly once? That is the ConsumerCursor in the guide.
Want to look up one row by its id (now, or as it was in the past)? table.lookup(&[("id", skade::Scalar::I64(7))], None) — also in the guide.
Want to record "who changed what, when" for a lineage graph? Turn on the lineage feature and attach a sink — the guide's last section.