nordisk/knut

CDC guide — reading a change feed out of knut

The 30-second version

use knut_bifrost::cdc::{CdcBackend, ChangeRequest, SkadeCdc, col};

// Pure Rust, no Spark: read every change to `people` since snapshot 42.
let cdc = SkadeCdc::new("/var/lib/lake");
let mut feed = cdc.changes(&ChangeRequest::new("people", Some(42))).await?;

while let Some(batch) = feed.next().await {
    let batch = batch?;                  // (op, before, after, snapshot, ts)
    render(&batch);                      // paint it — the next batch is still coming
}

Over Spark, the backend is chosen for you:

use knut_bifrost::source::SparkConnectSource;

let src = SparkConnectSource::connect("sc://spark:15002").await?;
// Detects the server version, probes the table, logs the verdict, then serves
// the feed from Spark 4.2's CHANGES clause or Iceberg's changelog view.
let mut feed = src.changes(&ChangeRequest::new("lake.social.people", Some(42))).await?;
println!("backend = {}", feed.capability.chosen.as_str());

The row a UI renders

Column names are constants — use cdc::col::{OP, BEFORE, AFTER, SNAPSHOT, TS}, don't string-match.

before and after carry the same fields, in the same order, so a diff pane is a positional zip:

let before = batch.column_by_name(col::BEFORE).unwrap()
    .as_any().downcast_ref::<StructArray>().unwrap();
let after  = batch.column_by_name(col::AFTER).unwrap()
    .as_any().downcast_ref::<StructArray>().unwrap();

for row in 0..batch.num_rows() {
    match (before.is_valid(row), after.is_valid(row)) {
        (false, true)  => paint_added(after, row),
        (true, false)  => paint_removed(before, row),
        (true, true)   => paint_diff(before, after, row),   // a folded `update`
        (false, false) => unreachable!("every change row fills at least one side"),
    }
}

Getting update rows with both sides filled

Engines emit an update as two rows (update_before, then update_after), and Iceberg v2 sometimes only as delete + insert. Name the identity columns and knut pairs them for you:

ChangeRequest::new("people", Some(42)).with_identity(vec!["id".into()])

Then a matched pair becomes one row with op = "update", before = old, after = new. Unmatched halves pass through untouched — nothing is ever dropped to make the output prettier.

Narrowing the feed

ChangeRequest::new("people", Some(42))
    .to(Some(99))                                        // window end (inclusive)
    .with_columns(vec!["id".into(), "name".into()])       // projection
    .with_filter(Some(RowFilter::eq("country", "SE")))    // predicate
    .with_identity(vec!["id".into()])                     // pair updates

• An empty projection means every column — exactly the pre-pushdown behaviour, so nothing changes unless you ask.
• The changelog metadata columns and the filter's own column are added back automatically; you cannot accidentally project away the thing that says a row was deleted.
• A filter that names none of eq / in / ge+le is rejected, not silently treated as "match everything".
• net_changes: true asks the engine to collapse intermediate states per row identity. Leave it false (the default) for a live feed — on a log table the collapse cancels an insert-then-delete pair, which is right for a net-change report and wrong for a change pane.

Which backend am I on?

let cap = feed.capability;
println!("{}", serde_json::to_string_pretty(&cap.to_json())?);

{
  "cdc_backend": "iceberg_changelog_view",
  "server_version": "4.2.0",
  "server_major_minor": "4.2",
  "native_expected": true,
  "native_usable": false,
  "fell_back": true,
  "reason": "Spark 4.2 has the native CDC engine but the CHANGES clause did not resolve for `lake.social.people` (the connector implements no org.apache.spark.sql.connector.catalog.Changelog) — FELL BACK to Iceberg's create_changelog_view. Engine said: …"
}

Read it out of the log:

grep '"event":"cdc_backend"' bifrost.log

…or out of the persisted ledger, where the listener stamps it after every changelog round:

$ jq '."lake.social.people".cdc' bifrost-state.json
{ "chosen": "iceberg_changelog_view", "fell_back": true, ... }

fell_back: true is the load-bearing field. It means knut expected Spark 4.2's native CDC and could not use it — today, that is the normal answer for an Iceberg table, because Iceberg ships no Spark 4.2 runtime yet (see cdc-design.md §f1). It is never a silent substitution.

The listener side

Nothing to configure. A mapping that says

tables:
  - table: lake.social.people
    mode: cdc
    columns: [id, name]                  # optional; derived from the pipeline otherwise
    filter: { column: country, eq: SE }  # optional
    pipeline:
      - vertex: { label: Person, id_column: id, properties: [name] }

gets the projected + filtered changelog read on whichever backend the server supports, INSERT/UPDATE_AFTER rows become graph upserts and DELETE rows become DETACH DELETEs — in either engine's dialect, because the router parses through the shared cdc::ChangeOp.

What to watch out for

• Streaming is per-batch delivery, not per-row wire streaming. Spark Connect's collect materialises the result before returning; the ChangeStream hands it over batch by batch. Real wire-level streaming needs a Connect execute-plan stream, which is not wired yet.
• On the skade path the pushdown is post-decode. project_columns slices after the parquet read and the predicate prunes rows after the scan, so the win is encode/transform/sink/memory — not I/O. The plan-time file-skipping numbers are knut.iceberg_pushdown_read, a different arm; don't quote one as the other.
• A DELETE row's non-identity columns are null. That is how an Iceberg equality-delete widens to the table schema. post_filter therefore keeps a row whose filter column is null, on purpose.