Tables¶

The Delta tables the library writes. You name each one; a name such as ops.ingestion_facts is a catalog table, and anything with a / or a : is a path.

Each is created on first use with the DeltaTable builder: explicit types, a comment on the table and on every column the library defines, which DESCRIBE TABLE shows (ADR 0012). The comments below are the ones the tables carry, copied from the code (mssql_cdc.sink, mssql_cdc.events, mssql_cdc.finalization, mssql_cdc.silver, mssql_cdc.reconcile and mssql_cdc.migrations), and tests/test_docs_tables.py fails when they drift. How the pieces fit: Architecture.

Bronze¶

The change log, append-only.

Append-only change rows from SQL Server CDC, written by mssql-cdc-pyspark's delta_sink. One row per change: an update is two rows (operation 3, the row before; 4, the row after). Order changes by (_start_lsn, _command_id, _seqval, _operation). Rows with operation 0 are a snapshot of the source table, the one _snapshot names: rebuild from it with its rows and the changes with a larger _start_lsn. A whole snapshot's rows share its _start_lsn; a chunked snapshot's (_chunk set) are stamped per chunk, at or after it. A column the source table gained through a newer capture instance is added when the stream first reads it (older rows read NULL); a column it lost stays, NULL from then on.

_command_id is absent with includeCommandId=false. A column a newer capture instance captures is added after the existing ones.

How it is written:

• One append per micro-batch that read rows, with txnAppId = app_id and txnVersion = the batch id, so a replayed batch is skipped. Its userMetadata holds the batch's facts as JSON: rows, min_lsn, max_lsn, min_commit_ts, max_commit_ts, deletes, inserts, updates, batch_id and app_id. A batch that read no rows writes no commit. Where the platform refuses to cache the batch (Databricks serverless), the append reads it, so its userMetadata is written first: rows to updates are null, and the facts table has them; and a batch with no rows writes an empty commit unless the batch before it had none too (for a run's first batch, as the facts table's last row says; without a facts table, it writes one) (ADR 0032).
• A snapshot is one append whose userMetadata is {"snapshot": <capture instance>, "lsn": ..., "commit_ts": ...}; a seed's, from a copy you already had, is {"seed": <capture instance>, "lsn": ..., "commit_ts": ...} (Bootstrap). Its rows have _snapshot = _start_lsn and no _chunk.
• A chunked snapshot is one append per wave of backfill(), with txnAppId <app_id>#snap.<S> and txnVersion the wave, so a rerun of a wave is skipped. Its userMetadata is {"backfill": "<app_id>#snap.<S>", "wave", "lsn", "attempt", "chunks"}: the wave's stamp and, per chunk, its bounds and rows. Every row has _snapshot = S, its chunk in _chunk and the wave's stamp in _start_lsn, at or after S (Bootstrap). Where the platform refuses to cache the wave, a chunk's rows, high_lsn, read_seconds and read_mb are null there; its snapshot_chunk facts row has them (ADR 0032).
• Every append uses mergeSchema. A changed column type fails it with SchemaChangedError unless the table has delta.enableTypeWidening and the change widens (Schema changes).

DESCRIBE HISTORY bronze.orders;  -- userMetadata of each append

Facts¶

What each micro-batch did, plus one row per snapshot, per source change and per skip past purged changes. Optional, but monitoring, the data-loss recovery and the schema-change events all need it.

One row per micro-batch written by mssql-cdc-pyspark's delta_sink, including batches that read no change rows (rows = 0), so a current stream on a quiet table keeps writing rows: what was written (counts, LSN and commit-time ranges), how far the stream had read (end_lsn, end_commit_ts) and how long it took. Where the platform caches, the same facts are in each target commit's userMetadata (batches with rows only); elsewhere the commit's userMetadata has the keys with null counts (ADR 0032). Delta log cleanup eventually drops it. Each snapshot stream().to_delta takes (bootstrap or re-snapshot), each schema change on the source, each switch to a newer capture instance and each skip past purged changes (failOnDataLoss=false) adds one row, with event set (see its comment); a snapshot adds one more when it opens; a chunked one also adds one with the chunks its first stream().backfill() call plans and one per chunk it reads.

The facts hold no source key values. A chunked snapshot's key values, its key's extent and the bounds of its chunks (with a composite or non-integer key, real keys of the table every chunk_rows rows), are in the snapshot plans table next to bronze, and in the userMetadata of each wave's commit in bronze's history; give both bronze's access policy. A snapshot opened before 0.7.0 keeps them in its facts rows to its end, as that release wrote them, so jobs of both releases that share the facts read them; give such a facts table bronze's access policy until its snapshots complete (ADR 0028).

The event values, the keys of these JSON payloads and those of a wave's userMetadata are state a later release reads back: a release only adds keys, never renames or removes one, so a query that parses detail keeps working across upgrades (ADR 0021). The one exception is the key values above, which 0.7.0 moved to the snapshot plans table, whose detail holds each payload whole.

Their types, TypedDicts exported from mssql_cdc, for Python that reads them (d: SnapshotChunkDetail = json.loads(row["detail"])):

The kinds of row, and the Delta txnAppId and txnVersion that make each write idempotent:

A chunked snapshot's rows in its plans table are each written just before their facts rows, with the same key in that table, but for 'snapshot_open': its row there has <app_id>#snapopen.<S>, 0, so a rerun that opens it at another S writes its own.

The sink's app_id is the one passed to to_delta, or <app_id>.g<n> in generation n. Statistics over micro-batches filter event IS NULL; the snapshots downstream rebuilds from are event IN ('bootstrap', 'resnapshot'). A chunked snapshot's bootstrap or resnapshot row comes once backfill() has read its last chunk, with the same key as a whole one's: its min_lsn = max_lsn is S, rows the sum of its chunks. The one a switch to a newer capture instance opens has keys by its S instead, as its generation's own may be taken; its snapshot_open detail has switch_batch_id, and S is the newer instance's start LSN, which the stream had read past (the min_lsn of its capture_instance_switched row).

SELECT app_id, batch_id, rows, end_commit_ts, retention_headroom_hours, ingestion_lag_seconds
FROM ops.ingestion_facts
WHERE target = 'bronze.orders' AND event IS NULL
ORDER BY written_at DESC
LIMIT 20;

What to alert on: Monitoring. The metric columns and end_lsn are NULL without metricsPath.

Snapshot plans¶

The key values of a chunked snapshot, next to bronze: what its 'snapshot_open', 'snapshot_plan' and 'snapshot_chunk' facts rows leave out (ADR 0028). You do not name it: to_delta creates it when it opens a chunked snapshot of target, as <target>_snapshot_plans (bronze.orders_snapshot_plans, or the sibling directory .../orders_snapshot_plans of a path), and backfill(), apply_changes and reconcile find it by bronze's name.

Key values of the chunked snapshots stream().backfill() reads into a bronze table, kept by mssql-cdc-pyspark next to it, under its name with _snapshot_plans appended: one row per 'snapshot_open', 'snapshot_plan' and 'snapshot_chunk' row of the ingestion facts table, with that row's detail whole. The facts rows leave these values out; they are values of the source table's key, so give this table bronze's access policy.

Control¶

The completeness verdict, finalized_until, one row per table, and the position of each silver table.

Completeness verdict per CDC-fed table, kept by mssql-cdc-pyspark: one row per table. Gate downstream work on finalized_until, e.g. with finalization.is_final().

finalization.advance MERGEs on table_name and updates the row only when the new verdict is later, so it never moves back. apply_changes records applied_lsn and snapshot_lsn for its silver table (and, while a chunked snapshot is open, open_snapshot_lsn and snapshot_wave), then advances that table's verdict. Every table shares the control table, and in OSS Delta two MERGEs on it conflict even when they change other rows, so both retry a MERGE that loses to a concurrent commit, for up to a minute. See Finalization and Silver.

SELECT table_name, finalized_until, end_lsn, applied_lsn, updated_at
FROM ops.table_finalization;

Silver¶

The current state of the source table, built by apply_changes.

Current state of a SQL Server table, one row per key, built from the bronze change log by mssql-cdc-pyspark's apply_changes. Deleted rows are removed. How far it is applied is applied_lsn in the control table; finalized_until there says which periods are complete.

A column bronze gains is added to silver (older rows read NULL); a column's type never changes (ADR 0019).

Reconcile report¶

What each reconcile() run found, appended when it is given a report_table (Validation).

Report of mssql-cdc-pyspark's reconcile(), which compares a silver table with its SQL Server table. Each run adds one row per bucket of the key's range (the rows and the key sum on each side) and one row per key whose row differs in the buckets compared row by row; run_id groups them.

Three kinds of row share it, told apart by which columns are set:

SELECT status, count(*) FROM ops.reconcile_report
WHERE run_id = '<run_id>' AND status IS NOT NULL GROUP BY status;

Schema versions¶

Each table carries the table property mssql_cdc.schema_version: the number of migrations its kind has had. A new table is created at the latest version. An older one gets the missing migrations, in order, the next time a stream, finalization.advance or apply_changes opens it (ADR 0013). The versions, from mssql_cdc.migrations.<kind>.MIGRATIONS:

A migration that adds columns appends them, so in a table created by an older release they come after written_at (facts), updated_at (control) or _batch_id and the captured columns (bronze): select columns by name. The other table properties the library sets, and the ones it leaves to you, are below.

Jobs that share a table can upgrade one at a time: a job on an older release keeps writing a table a newer release migrated, and logs a WARNING. A future migration that older releases would misread will set mssql_cdc.min_version, and releases that know fewer migrations than it names refuse the table with a ValueError; no migration sets it yet.

SHOW TBLPROPERTIES ops.ingestion_facts ('mssql_cdc.schema_version');

Table properties¶

Besides mssql_cdc.schema_version, the library sets one property, and only where a table cannot work without it (ADR 0012): delta.columnMapping.mode = 'name', on a bronze or silver table it creates with a captured column whose name Delta refuses otherwise: one with a space or one of ,;{}()=, such as [Unit Price] or [Qty (kg)]. Column mapping raises the table's Delta protocol, so every reader and writer of that table needs a Delta that supports it. A table created without it does not get it later: when a newer capture instance adds such a column, the stream and apply_changes fail before writing, until you enable it (Schema changes).

Everything else is yours to enable, where every reader and writer of the table supports it, since each one changes the table's protocol or how its files are laid out:

• Compaction: a scheduled OPTIMIZE (and VACUUM) on bronze, silver and the facts table, or optimized writes and auto compaction where the runtime has them (Streaming).
• Deletion vectors on silver (delta.enableDeletionVectors = true): the apply_changes MERGE marks the rows it replaces instead of rewriting whole files. On Databricks, adding row tracking (delta.enableRowTracking = true) lets concurrent writes to different rows go through without conflict.
• Clustering silver by its keys, with liquid clustering (ALTER TABLE ... CLUSTER BY) where the runtime has it, or OPTIMIZE ... ZORDER BY the keys: a key range lives in fewer files, so each MERGE rewrites fewer.
• Type widening (delta.enableTypeWidening = true) on bronze, so a widened source column does not stop the stream (Schema changes).