Architecture¶

Components¶

flowchart TB
  subgraph src/mssql_cdc
    DS[source.MssqlCdcDataSource<br/>schema, streamReader]
    RD[source.MssqlCdcStreamReader<br/>offsets, partitions, read]
    SN[source.MssqlCdcSnapshotReader<br/>key ranges, read]
    CL[client.CdcClient]
    SQL[client.SqlCdcClient<br/>T-SQL]
    BE1[MssqlPythonBackend<br/>cursor.arrow_batch]
    BE2[ArrowOdbcBackend<br/>read_arrow_batches]
    FK[fake.FakeCdcClient<br/>files]
    SK[sink.delta_sink<br/>foreachBatch]
    FN[finalization<br/>advance / is_final]
    PL[pipeline.CdcStream<br/>to_delta, snapshot, generations]
    BF[pipeline.CdcStream.backfill<br/>chunked snapshot waves]
    SV[silver.apply_changes<br/>MERGE, waves, rebuild]
    RC[reconcile<br/>bucket counts, row hashes]
  end
  PL --> DS
  PL --> SK
  BF --> SN
  RC --> SN
  RC --> CL
  DS --> RD --> CL
  DS --> SN --> CL
  CL --> SQL --> BE1 & BE2
  CL --> FK
  RD -. micro-batches .-> SK
  SN -. snapshot rows .-> PL
  SK -. after commit .-> FN
  SK -. bronze .-> SV
  BF -. chunk waves .-> SV
  SV -. after MERGE .-> FN
  SV -. silver .-> RC

• Source: Spark Python DataSource V2 (pyspark.sql.datasource). One stream per source table: its capture instance, and the newer one it switches to (ADR 0023). mssql_cdc_snapshot is its batch sibling: the tracked table's current rows in the same schema, stamped with the LSN recorded before they are read, for the initial load (ADR 0016). A whole snapshot has one stamp; a chunked one (snapshotChunks) a stamp per wave, each at or after the snapshot's own LSN S, so its rows do not share one _start_lsn (ADR 0028).
• Client: the only code that knows T-SQL. The reader depends on the CdcClient interface, so the fake can replace SQL Server in tests.
• Backends: turn a query into Arrow record batches. Interchangeable because every LSN is a hex string on the wire (invariant 6 in CLAUDE.md).
• Sink: an optional, Delta-specific foreachBatch writer. The source works with any sink.
• Finalization: a control table with one row per target table.
• Events: the facts table's event rows (snapshots opened, planned, read in chunks and completed; schema changes, capture instance switches, data skipped) are written and read back through one module, events: their names, the rows each writer appends, and the readers the pipeline, silver and reconcile share. A chunked snapshot's key values go to a plans table next to bronze through it too, and its readers complete the facts rows from it.
• Backfill: CdcStream.backfill reads a chunked snapshot that to_delta opened, in waves next to the running stream: one bronze commit and one facts row per chunk each (ADR 0028).
• Silver: apply_changes reads bronze (and the facts, for re-snapshots and chunked snapshots) and MERGEs the latest image per key into a current-state table; its position and verdict are its row in the control table (ADR 0019).
• Reconcile: reconcile compares silver with its SQL Server table, by row count and key sum per key range and then row by row through the snapshot reader, and reports what differs.

Where code runs¶

One micro-batch¶

sequenceDiagram
  participant E as Spark engine
  participant R as Reader (driver)
  participant X as read() (executor)
  participant S as SQL Server
  participant D as Delta sink
  participant F as Finalization

  E->>R: latestOffset(start, ReadMaxRows(n))
  R->>S: fn_cdc_get_max_lsn()  (or AvailableNow target)
  R->>S: n-th start_lsn after start in lsn_time_mapping
  R->>S: fn_cdc_map_lsn_to_time(end)  -> commit_ts (UTC)
  R-->>E: end = {lsn, commit_ts}
  Note over E: offset log written (checkpoint)
  E->>R: partitions(start, end)
  R->>S: sp_cdc_help_change_data_capture: the table's instances, their start_lsn
  R->>S: sp_cdc_get_ddl_history(ci): DDL in (start, end]  (a type change: SchemaChangedError)
  R->>S: fn_cdc_increment_lsn(start), fn_cdc_get_min_lsn(ci) per instance read
  R->>S: NTILE split points on cdc.ci_CT (numPartitions > 1)
  R-->>E: [LsnRange(ci, from, to), ...], cut at a newer instance's start  or DataLossError
  Note over R: events (schema_change, capture_instance_switched, data_skipped) as files in metricsPath
  E->>X: read(LsnRange)
  X->>S: cdc.ci_CT WHERE start_lsn BETWEEN from AND to, JOIN lsn_time_mapping
  S-->>X: Arrow record batches
  X->>S: fn_cdc_get_min_lsn(ci)  (cleanup during the read? DataLossError, or with failOnDataLoss=false a possible data_skipped in its metrics file)
  X-->>E: batches cast to the Spark schema
  E->>D: foreachBatch(df, batch_id)
  D->>D: append (txnAppId, txnVersion=batch_id, userMetadata=facts, mergeSchema)
  D->>D: facts: the batch's row and its event rows, one commit
  Note over E: commit log written
  E->>F: after the query (AvailableNow) or on progress
  F->>F: MERGE finalized_until = GREATEST(old, trunc(end.commit_ts))

Offsets and the checkpoint¶

{"lsn": "0x0000002A000001F00003", "commit_ts": "2026-09-28T14:03:12.117"}

• lsn: last processed commit LSN. The initial offset for startingLsn=earliest is fn_cdc_decrement_lsn(min_lsn), so the first read starts at min_lsn.
• commit_ts: commit time of lsn in UTC. It is stored even when the batch has no rows, because end can be an idle "dummy" entry. This is what lets finalization advance on quiet tables.
• Replays: Spark re-runs an uncommitted batch with the same (start, end). read() is deterministic for a range as long as CDC cleanup has not purged it; if it has, the guard fails the query.

A second capture instance¶

A table can have two capture instances, the newer one usually capturing a column the older does not (ADR 0023). The newer one's start_lsn S is the commit LSN of its enable, and from S on every commit is in both. partitions() lists the table's instances on every planning and cuts the batch at S:

start                S - 1 | S                        end
  |------ older: cdc.dbo_orders_CT ------|------ newer: cdc.dbo_orders_v2_CT ------|

Offsets stay database-wide LSNs, so the checkpoint needs nothing new, and a replay cuts at the same place. Each range reads its own instance's columns, NULL for the ones it lacks; load() infers the union of both. The first batch that reads the newer instance leaves a capture_instance_switched event for the facts. Its facts row is written before Spark commits that batch, and a replay still reads the older instance, so the DBA disables the older one only after the stream's next batch (sql/switch_capture_instance.sql). A declared column that no instance captures fails the first planning instead of reading NULL. DDL inside a batch leaves a schema_change event; a changed type fails the batch before it reads anything (SchemaChangedError).

Generations (to_delta)¶

to_delta reads <checkpoint>/_mssql_cdc_generation.json on every call to find the live generation. Without the file it is generation 0: the checkpoint and app_id as given.

<checkpoint>/
  offsets/ commits/ metadata ...   generation 0 (Spark's files)
  _mssql_cdc_metrics/              generation 0 metrics (default metricsPath)
  _mssql_cdc_generation.json       {"generation": n, "snapshot_lsn", "commit_ts", "at"},
                                   plus "recovering" and "failed_at" while a recovery is open
  _generations/<n>/                generation n: Spark checkpoint and metrics, app_id <app_id>.g<n>

With on_data_loss="resnapshot", before starting the query it takes the last processed offset from the live checkpoint (line 3 of offsets/<n> for the highest committed n, Spark's offset log format v1) and, when max_lsn is past it, applies the driver guard's test. When the next range is purged it records that offset as recovering, snapshots the target, records a 'resnapshot' facts row, writes the state file for n + 1 (atomically; the commit point) and starts generation n + 1 from the snapshot's LSN. At most one automatic re-snapshot, or failed one, per resnapshot_interval_days (ADR 0018).

Output schema¶

Six metadata columns, then the captured columns; every column, the ordering key and the default type mapping are in Output schema. Captured columns come from the columns option (DDL) or, when it is omitted, from CDC metadata at load() time on the driver: sys.sp_cdc_get_captured_columns, sorted by column_ordinal (SqlCdcClient.captured_columns), for every capture instance of the table, joined by name, older instance first. It needs only the permissions of the CDC query functions. The fake infers them too when its capture instances record their columns (FakeCdcDatabase(columns=...)), and needs columns otherwise.

Tables written by the sink, finalization, silver and reconcile¶

All six are created on first use with DeltaTable.createIfNotExists: explicit types, and a comment on the table and on every control, facts, plans and report column and every bronze and silver metadata column (DESCRIBE TABLE shows them), and stamped with the table property mssql_cdc.schema_version. Existing tables get the schema migrations of their kind that they have not had yet (ADR 0012, ADR 0013).

Extension points¶

• New backend: a class with batches, scalar and close, the Backend protocol (subclassing it inherits scalar and close, ADR 0030), added to make_client.
• Other sinks: the source is sink-agnostic; any writeStream target works. Idempotency and facts are then the sink's job.
• Continuous mode: finalization.track advances the verdict from a StreamingQueryListener (ADR 0026); where a listener cannot run, call finalization.advance from a separate job reading the checkpoint's committed offsets.