<!-- nornir:class: design -->
# CDC — one seam, three backends, picked by what the server actually is
> Rickard's ask: *"for Spark 4.2 use its CDC, and for 4.1 use Iceberg's CDC. Want to see this."*
>
> This is the chosen design. Alternatives and the research trail live in
> `cdc-reasoning.md`.
## f0 — the whole
knut reads "what changed in this table since snapshot *N*" three different ways.
Which one is *available* is a property of the server on the other end of the
socket, not of anything an operator types into a YAML file. So there is **one
trait**, `cdc::CdcBackend`, three implementations of it, and a **probe** that
picks one.
```
┌──────────────────────────────────────┐
ChangeRequest ─────►│ cdc::CdcBackend │────► ChangeStream
(table, from, to, │ (the ONE seam — the listener and │ (op, before,
columns, filter, │ the UI never branch on engine) │ after,
net_changes, └───────┬────────────┬─────────────┬───┘ snapshot, ts)
identity) │ │ │
SparkNativeCdc SparkIcebergCdc SkadeCdc
Spark 4.2's Iceberg's skade's
CHANGES clause create_changelog_ read_changelog
view procedure (pure Rust)
```
## f1 — what each Spark version really offers
Established from primary sources, not memory. Citations in `cdc-reasoning.md`.
### Spark 4.2 — native CDC exists
Spark 4.2.0 (GA 2026-07-14) landed **SPARK-55668, "Change Data Capture (CDC)
Support"**, resolved with fix-version 4.2.0. Concretely:
* **A SQL `CHANGES` clause.** From Spark's own grammar at tag `v4.2.0`
(`sql/api/…/SqlBaseParser.g4`, rule `changesClause`):
```
CHANGES FROM (SYSTEM_VERSION | VERSION) startingVersion (INCLUSIVE | EXCLUSIVE)?
(TO (SYSTEM_VERSION | VERSION) endingVersion (INCLUSIVE | EXCLUSIVE)?)?
| CHANGES FROM (SYSTEM_TIME | TIMESTAMP) startingTimestamp (INCLUSIVE | EXCLUSIVE)?
(TO (SYSTEM_TIME | TIMESTAMP) endingTimestamp (INCLUSIVE | EXCLUSIVE)?)?
```
plus `streamChangesClause` (open-ended, optional start) for `SELECT … FROM
STREAM t CHANGES FROM VERSION 0`.
* **A DSv2 connector interface**, `org.apache.spark.sql.connector.catalog.Changelog`,
marked `@since 4.2.0`. A connector implementing it must expose
`_change_type` (STRING: `insert` / `delete` / `update_preimage` /
`update_postimage`), `_commit_version` (Long **or** String) and
`_commit_timestamp` (TIMESTAMP).
* **Engine-side post-processing** — carry-over removal, update detection,
net-change collapse — in `ResolveChangelogTable`, so the same query behaves the
same across any CDC-capable connector.
* **Spark Connect bindings** (SPARK-55949, "Add DataFrame API and Spark Connect
support for CDC queries"). And independently of those bindings, the `CHANGES`
clause is *plain SQL*, so it travels over `SparkSession::sql` on any Connect
client — including knut's Rust one.
### …and it is NOT usable against an Iceberg table today
This is the part that must be said plainly rather than stubbed around:
* Apache Iceberg publishes **no `iceberg-spark-runtime-4.2` artifact**. As of
2026-07-22 Iceberg 1.11.0 (the latest release) ships
`iceberg-spark-runtime-{3.4,3.5,4.0,4.1}` only — Maven Central returns zero
results for the 4.2 coordinate. In the Iceberg tree, `spark/` contains
`v3.5`, `v4.0`, `v4.1` and nothing else.
* apache/iceberg#14984 *"Spark: Add support for 4.2.0"* is **still open**;
#16751 *"Spark: Add Spark 4.2 module (copy of Spark 4.1)"* was **closed
unmerged** (2026-06-13).
* No source in apache/iceberg references
`org.apache.spark.sql.connector.catalog.Changelog` — a GitHub code search over
the repo returns 0 hits.
So on a Spark 4.2 server holding Iceberg tables, the native CDC engine is
**present** and, for our data, **unreachable**. `CHANGES` will not resolve.
### Spark 4.1 and earlier — Iceberg's CDC
`CALL system.create_changelog_view(table => …, options => map('start-snapshot-id',
…, 'end-snapshot-id', …), identifier_columns => array(…), net_changes => …,
changelog_view => …)` materialises a temp view whose rows carry `_change_type`
(`INSERT` / `DELETE` / `UPDATE_BEFORE` / `UPDATE_AFTER` — **upper case**),
`_change_ordinal` and `_commit_snapshot_id`.
This is what knut's blessed pair (Spark 4.1.2 GA +
`iceberg-spark-runtime-4.1_2.13:1.11.0`) uses, and it is also what a 4.2 server
must fall back to for Iceberg.
### No Spark at all
`skade::Table::read_changelog` walks the Iceberg manifest lineage, tags appended
rows `INSERT` and equality-deleted identities `DELETE`, and re-reads the whole
target snapshot on an overwrite / position-delete window. knut **consumes** it;
it does not keep a second incremental reader (L5).
## f2 — detection, and honest degradation
`cdc::detect(session, table)` is two steps, both observable:
1. **Ask.** `SparkSession::version()` (the Connect `AnalyzePlan` SparkVersion
request) → `parse_major_minor` → `has_native_cdc_engine((4,2))`.
2. **Probe.** Even at 4.2, run a zero-row
`SELECT * FROM {table} CHANGES FROM VERSION 0 LIMIT 0`. A version number is
not a capability: the clause only resolves for a connector implementing
`Changelog`.
The verdict is a `CdcCapability`:
| field | meaning |
|---|---|
| `chosen` | `spark_native` \| `iceberg_changelog_view` \| `skade_delta` |
| `server_version` | the server's version string, verbatim |
| `server_major_minor` | parsed `(major, minor)` |
| `native_expected` | server is >= 4.2 (the engine exists) |
| `native_usable` | the probe actually resolved for this table |
| `fell_back` | `native_expected && !native_usable` — the honest degrade |
| `reason` | human-readable, **including the engine's own error text** |
A fallback is never silent. It is:
* **logged** — `tracing::warn!(target: "bifrost::cdc", event = "cdc_backend",
fell_back = true, reason = …)` (an ordinary selection logs at `info`);
* **persisted** — `TableState::cdc` is stamped after every changelog round, so
the ledger file (`state_json`) names the backend that ran and why;
* **cached** — detection costs one version request plus one probe, per table,
for the lifetime of the source; a poll loop does not re-probe.
Given the Iceberg state of the world above, the *expected* production verdict on
a 4.2 server today is `chosen = iceberg_changelog_view, fell_back = true` — and
knut says so out loud rather than pretending the native path ran.
## f3 — the renderable row (what a UI binds to)
Every backend normalises onto ONE Arrow schema, `cdc::change_row_schema`:
| column | type | null? | meaning |
|---|---|---|---|
| `op` | `Utf8` | no | `insert` \| `delete` \| `update_before` \| `update_after` \| `update` |
| `before` | `Struct<data columns, all nullable>` | yes | the pre-image; **null** for an insert |
| `after` | `Struct<data columns, all nullable>` | yes | the post-image; **null** for a delete |
| `snapshot` | `Int64` | yes | Iceberg snapshot id / Spark `_commit_version` |
| `ts` | `Timestamp(Microsecond, "UTC")` | yes | commit timestamp |
* `before` and `after` share **one** `Fields` instance, so they are guaranteed
field-identical — a UI renders a diff by zipping them positionally.
* Changelog metadata columns never leak into the images.
* `op` is knut's canonical spelling. `ChangeOp::parse` accepts **both** engine
dialects (Iceberg's `UPDATE_BEFORE`, Spark's `update_preimage`, …) so the two
never reach the UI as different vocabularies. An unrecognised value is a hard
error — a silently dropped change row is the failure that matters.
* `cdc::fold_updates(rows, identity)` optionally pairs an `update_before` with
its `update_after` on `(snapshot, identity)` into a single `op = "update"` row
carrying **both** images. Unmatched halves pass through, in order; nothing is
dropped.
Delivery is a `cdc::ChangeStream` — a bounded `tokio::sync::mpsc` of renderable
batches, so a pane paints changes as they arrive and a slow consumer applies
backpressure instead of growing a buffer. `ChangeStream::next()` for the
incremental case, `collect()` for tests and benches.
**Caveat, stated:** Spark Connect's `collect` materialises the whole result before
returning, so on the two Spark backends the "streaming" is incremental *delivery
of the decode*, not of the network. The skade backend is the same shape. The
seam is right; the wire is not yet.
## f4 — pushdown into the CDC path
Before this, `SyncMode::Incremental` / `SyncMode::Cdc` were full-width **by
construction**: the Spark path did `SELECT * FROM {changelog_view}`, and the
skade path called `read_changelog` with no projection. The `Full`-mode pushdown
that landed in `00f6598` did not reach them at all.
Now, per backend:
| backend | projection | predicate | where it acts |
|---|---|---|---|
| `SparkNativeCdc` | select-list on the `CHANGES` relation | `WHERE` | **in Spark** |
| `SparkIcebergCdc` | select-list on the changelog view | `WHERE` | **in Spark** |
| `SkadeCdc` | `ChangelogBatch::project_columns` | `cdc::post_filter` | **in knut, post-decode** |
Two invariants:
1. **Metadata always survives.** `_change_type` is load-bearing (the listener
routes deletes on it); `_change_ordinal` / `_commit_snapshot_id` /
`_commit_version` / `_commit_timestamp` carry order and provenance. The filter's
own column is added to the projection too, or the `WHERE` would reference a
column the select-list dropped.
2. **Unprojected stays unprojected.** An empty projection and no filter emit
exactly the statement knut emitted before, so the narrowing can only ever
apply where it was asked for.
**Honest accounting.** On the skade path this is *not* plan-time file skipping:
`project_columns` is a post-decode slice and skade's `read_delta` decodes whole
parquet files (`ParquetRecordBatchReaderBuilder::try_new(bytes)?.build()?` takes
no projection mask), and the predicate is a post-scan row prune. The win is in
the encode, the transforms, the sink and the memory — not in parquet I/O. Pushing
either into the manifest-delta scan is skade-side work and is logged as such
(`post_filtered = true` in the trace event, `pushdown_kind` in the bench). The
genuinely plan-level numbers live in `knut.iceberg_pushdown_read`; the two must
not be conflated.
**One more invariant, tested:** a `DELETE` row whose filter column is **null** —
which is exactly how skade widens an equality-delete to the table schema — is
**kept**. Dropping it would silently lose a deletion.
## f5 — where it plugs in
* `source::TableSource::cdc_capability(table)` — default `None`; `IcebergSource`
answers `skade()`, `SparkConnectSource` answers its cached verdict.
* `Listener::sync_table` stamps that verdict onto the `TableState` it persists,
for `Incremental` and `Cdc` modes.
* `SparkConnectSource::changes(&ChangeRequest)` — the UI-facing renderable feed,
served by whichever backend detection chose.
* `transform::partition_change_rows` now parses through `cdc::ChangeOp`, so a
Spark 4.2 native changelog (lower-case ops) routes identically to an Iceberg
one. Before, it string-matched upper case only and a native feed would have
failed the round.
## f6 — the numbers
`knut.cdc_pushdown_read`, **NORNIR_MACHINE=oden** (Threadripper PRO 3975WX,
32 physical cores, SMT off, 499 GB, RTX 4090). **NEW HOST ROW** — never merged
with the `loki · 12 cores` table.
Change window: 1000000 appended rows (4 data files, 11 columns) plus a real
equality-delete snapshot (3 deletes). The mapping reads 2 of 11 columns.
| arm | s | rows/s | vs full |
|---|--:|--:|--:|
| full-width (pre-fix, unprojected changelog read) | 1.065778 | 938284 | 1.00 |
| projected (2 of 11 columns) | 0.652894 | 1531646 | **1.63** |
| projected + `country == SE` (250003 of 1000003 rows) | 0.617988 | 404543 | **1.72** |
| renderable `(op, before, after, snapshot, ts)` + update pairing | 0.582049 | 1718072 | **1.83** |
`peak_rss_bytes` 2299731968 · `warehouse_bytes` 711262775 ·
`pushdown_kind = post_decode_projection_and_post_scan_predicate`.
Two caveats that belong next to the numbers rather than in a footnote:
1. **The host was contended.** Sibling agents were building and benching
throughout (1-minute load average 58–276 on 32 cores; `telem_cpu_pct_avg`
79.19 for the run above). Three runs of the same code gave
`projection_speedup_x` of 1.46, 2.41 and 1.63. The *shape* is established;
the *magnitudes* are provisional until the bencher re-runs them uncontended.
2. **This is not `knut.iceberg_pushdown_read`.** That arm's 12.11x / 29.12x are
**plan-time** wins on a `SyncMode::Full` read — data files never opened
(`files_baseline` 4 → `files_planned` 1). The CDC arm's projection is a
post-decode slice and its predicate a post-scan row prune, because skade's
`read_delta` decodes whole parquet files and neither changelog surface takes a
scan predicate. A ~1.6–1.8x here and a 12x there are measuring different
things, which is why the metric carries `pushdown_kind`.