<!-- nornir:class: guide -->
# CDC guide — reading a change feed out of knut
## The 30-second version
```rust
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:
```rust
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 | Arrow type | meaning |
|---|---|---|
| `op` | `Utf8`, non-null | `insert` · `delete` · `update_before` · `update_after` · `update` |
| `before` | `Struct<…>`, nullable | pre-image; **null** on an insert |
| `after` | `Struct<…>`, nullable | post-image; **null** on a delete |
| `snapshot` | `Int64`, nullable | Iceberg snapshot id / Spark `_commit_version` |
| `ts` | `Timestamp(us, "UTC")`, nullable | commit timestamp |
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:
```rust
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:
```rust
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
```rust
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?
```rust
let cap = feed.capability;
println!("{}", serde_json::to_string_pretty(&cap.to_json())?);
```
```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:
```console
$ 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
```yaml
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 DELETE`s — 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.