nordisk/knut

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:

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:

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

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.

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.