Schema changes¶

SQL Server CDC does not follow DDL by itself: a capture instance keeps the column list it was enabled with, and a table can have at most two. This page covers what the stream does when the source table changes, how to start capturing a new column, and what to do when a column's type changes. The why behind all of it is ADR 0023.

The smallest setup¶

There is nothing to turn on. Each time the stream plans a micro-batch it asks SQL Server for the table's capture instances and for the DDL recorded on the table inside that batch (sys.sp_cdc_get_ddl_history). Give it a facts table so that what it finds is recorded:

from mssql_cdc import stream

query = stream(spark, options).to_delta(
    "bronze.orders",
    app_id="orders-v1",
    checkpoint="/Volumes/cat/sch/vol/ckpt/orders",
    facts_table="ops.ingestion_facts",
)

Each DDL statement and each move to a newer capture instance becomes a facts row of the batch that read past it. A switch's detail is JSON (old, new, start_lsn, not_captured, and text, the line rows written before 1.0.0rc1 hold instead), so the query reads its text, or the DDL statement:

SELECT app_id, batch_id, event,
       coalesce(get_json_object(detail, '$.text'), detail) AS detail, min_lsn, written_at
FROM ops.ingestion_facts
WHERE event IN ('schema_change', 'capture_instance_switched')
ORDER BY written_at DESC;

These rows travel through the metrics directory, which to_delta sets up by itself for a local or Volume checkpoint. With a URI checkpoint (abfss://, dbfs:/) set the metricsPath option, or the events only show up as warnings in the driver log.

Two things stay manual: a DBA creates and drops capture instances, and the owner of the bronze table enables Delta type widening. The library does neither.

What the stream does with each kind of DDL¶

The schemaChangePolicy option sets the reaction. classify, the default, does what the table says. fail fails the batch on any DDL; since the replayed batch holds the same DDL, restart once with classify to go past it.

Capturing a new column¶

A new column needs a new capture instance with the new column list. The stream then moves to it on its own: there is no option to change and nothing to restart unless the running query's schema lacks the new column.

How the stream follows a newer capture instance¶

The newer instance starts at S, its start_lsn, the commit LSN of its enable. From S on every commit is in both instances. Each batch reads the older instance below S and the newer one from S, both in one batch if it crosses S, so every commit is read once and offsets and checkpoints do not change (Architecture).

• At load(), the inferred schema is the union of both instances' columns.
• A query already running when the new instance appears switches in place if its schema holds every column the new instance captures. Otherwise it stops at S with SchemaChangedError, before reading past it; restarted, it infers the new columns and resumes there.
• With the columns option the declared list decides: the query switches in place and logs a warning naming captured columns it leaves out, which the batch's facts row keeps too (Monitoring).
• A column the query reads that the new instance does not capture reads NULL from S. The warning and the event's detail name it.
• Bronze gains the new column (every append uses mergeSchema); older rows read NULL for it. _capture_instance on each row says which instance it came from. A name Delta takes only with column mapping needs one step of yours first (below).

The procedure¶

The DBA runs it in the source database as db_owner:

1. Enable the new instance with the new column list: sys.sp_cdc_enable_table with a new @capture_instance.
2. Grant the reader SELECT on the new change table, cdc.[<new instance>_CT]. Each capture instance has its own change table; without the grant the first batch that reaches it fails with a PermissionError naming it (Permissions).
3. Check that sys.sp_cdc_help_change_data_capture lists both instances. The new one's sys.fn_cdc_get_min_lsn stays 0x00... until capture has processed the enable (seconds; up to about 5 minutes on a quiet database).
4. Wait until every stream that reads the table has a capture_instance_switched facts row, and one batch more: Spark commits the batch after its facts row, and a replay would read the old instance again. A stream already at or past S when it started (for example a bootstrap after the enable) never reads the old instance, writes no event and needs no wait.
5. Disable the old instance.

For step 4, a stream is done when it has a row with a larger batch_id than its switch:

SELECT s.app_id, s.batch_id AS switched_in, MAX(f.batch_id) AS last_batch
FROM ops.ingestion_facts s
JOIN ops.ingestion_facts f ON f.app_id = s.app_id
WHERE s.event = 'capture_instance_switched' AND s.target = 'bronze.orders'
GROUP BY s.app_id, s.batch_id;
-- ready when last_batch > switched_in for every stream of the table

A stream missing from the result has not reached S yet, unless it started past it.

-- Switch a CDC-tracked table to a new capture instance, for example to capture a column added
-- to the table: a capture instance keeps the column list it was enabled with (ADR 0023).
-- A procedure to follow step by step in the source database as db_owner, not a script to run
-- unattended. mssql-cdc-pyspark reads both instances and moves to the new one on its own at
-- the new one's start LSN; it never creates or drops capture instances.
--
-- Replace the names first: the table dbo.orders, its current instance dbo_orders, the new
-- one dbo_orders_v2, and cdc_reader, the database user the streams connect as.
-- (Lab check t9 runs this file as written, with those names replaced and split at GO.)

-- 1. Enable the new instance with the new column list. A table has at most two instances.
--    The enable waits for open transactions that already wrote the table; they end up in the
--    old instance only, below the new one's start LSN, which the stream reads from the old.
EXEC sys.sp_cdc_enable_table
     @source_schema        = N'dbo',
     @source_name          = N'orders',
     @capture_instance     = N'dbo_orders_v2',
     @role_name            = NULL,   -- or the old instance's gating role (the reader is a member)
     @captured_column_list = NULL,   -- NULL: every column; or N'order_id, status, note'
     @supports_net_changes = 0;
GO

-- 2. Let the reader read the new change table. Each capture instance has its own; without
--    this grant the first batch that reaches the new instance fails with a PermissionError
--    naming it. Nothing else changes for the reader (ADR 0009).
GRANT SELECT ON cdc.[dbo_orders_v2_CT] TO [cdc_reader];
GO

-- 3. Check: both instances of the table, and the new one's start LSN. Its min_lsn stays
--    0x00... until the capture job has processed the enable (seconds; up to ~5 minutes on a
--    quiet database).
EXEC sys.sp_cdc_help_change_data_capture @source_schema = N'dbo', @source_name = N'orders';
SELECT sys.fn_cdc_get_min_lsn(N'dbo_orders_v2') AS new_min_lsn;
GO

-- 4. Wait for every stream that reads the table. Each one writes a facts row with
--    event = 'capture_instance_switched' (detail {"old": "dbo_orders", "new": "dbo_orders_v2",
--    ...}; before 1.0.0rc1 the text 'dbo_orders -> dbo_orders_v2') once it has
--    read past the new instance's start. Spark commits that batch after its facts row: wait
--    for the stream's next batch too (a row of the same app_id with a larger batch_id). The
--    event reaches the facts only through metricsPath, which to_delta sets on its own only for
--    a local or Volume checkpoint; without it, look for the warning in the stream's log. A
--    stream already at or past the new instance's start (started after the enable, e.g. with
--    bootstrap=True: its facts end_lsn >= new_min_lsn) never reads the old instance, writes
--    no event and needs no wait. A stream whose schema lacks a column the new instance
--    captures stops there with SchemaChangedError instead: restart it, and it infers the new
--    columns and goes on.
--    In the lakehouse, for example:
--      SELECT app_id, batch_id, coalesce(get_json_object(detail, '$.text'), detail) AS detail,
--             written_at
--      FROM ops.ingestion_facts
--      WHERE event = 'capture_instance_switched' AND target = 'bronze.orders';
--    Until step 5 both instances capture every change: twice the change-table writes.

-- 5. Disable the old instance. Its change table goes at once. A stream that had not read up
--    to the new instance's start loses the changes only the old one held: it fails with
--    DataLossError, and to_delta(on_data_loss="resnapshot") recovers with a snapshot.
EXEC sys.sp_cdc_disable_table
     @source_schema    = N'dbo',
     @source_name      = N'orders',
     @capture_instance = N'dbo_orders';
GO

Keep the capture instance name¶

Leave captureInstance at the old name when it is SQL Server's default, <schema>_<table>: the stream follows the table's newest instance, also after the old one is disabled. With any other name, set it to the new one once the old one is gone, since a disabled custom name no longer leads to its table.

Renaming has a cost. SQL Server no longer lists the old name, so rows stamped with it, the bootstrap snapshot included, no longer count as this table's: a rerun with bootstrap=True takes one more snapshot, and a silver table built from scratch fails on the old rows.

Rows unchanged since the switch¶

The latest image of a row that has not changed since S has NULL in a column only the new instance captures. To fill it, ask for a snapshot right after the switch:

query = stream(spark, options).to_delta(
    "bronze.orders",
    app_id="orders-v1",
    checkpoint="/Volumes/cat/sch/vol/ckpt/orders",
    facts_table="ops.ingestion_facts",
    snapshot_on_switch=True,
)

After the batch that first reads the new instance, the sink appends a snapshot of the table, so every row's latest image carries the new column. It reads the whole table inside that batch. With a URI checkpoint it needs metricsPath, and to_delta raises a ValueError without it. A crash right after the switch batch takes the snapshot again on the replay: harmless, only slower.

For a table too big to snapshot, add snapshot="chunked" (with its facts_table): the batch then only opens a chunked snapshot at the new instance's start LSN, which the stream has already read past, and backfill(), run in a task of its own, reads it next to the stream as it reads a chunked bootstrap:

query = stream(spark, options).to_delta(
    "bronze.orders",
    app_id="orders-v1",
    checkpoint="/Volumes/cat/sch/vol/ckpt/orders",
    facts_table="ops.ingestion_facts",
    snapshot="chunked",
    snapshot_on_switch=True,
)
# in another task, until done
stream(spark, options).backfill(
    "bronze.orders", app_id="orders-v1", facts_table="ops.ingestion_facts"
)

Its snapshot_open facts row has switch_batch_id, the batch_id of the switch's capture_instance_switched row, and its last chunk writes a bootstrap row: no change was lost, and silver rebuilds from it then, applying its waves before. A chunked snapshot of the stream still open is superseded by it (its chunks so far are read again, now with the new column), and a run with snapshot="full" raises until it completes. A replay of the switch batch finds the snapshot it opened.

A column name that needs column mapping¶

SQL Server allows a space or one of ,;{}()= in a bracketed column name, such as [Unit Price]; Delta takes such a name only on a table with column mapping. A bronze or silver table created with one has it (Table properties). When a newer capture instance adds one to a table created without it, the stream fails before writing the batch, and apply_changes before changing silver:

SchemaChangedError: bronze.orders: column mapping is off on this table, and Delta takes
the new column(s) 'Unit Price' only with it. Enable it, then run again: ALTER TABLE
bronze.orders SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name',
'delta.minReaderVersion' = '2', 'delta.minWriterVersion' = '5'). That upgrades the
table's Delta protocol: every reader and writer of the table then needs a Delta that
supports column mapping.

The library never enables it itself, because the upgrade locks out every reader and writer of the table on a Delta without column mapping. Check them, run the statement on bronze, restart the query, and run it on silver before the next apply_changes:

ALTER TABLE bronze.orders SET TBLPROPERTIES (
  'delta.columnMapping.mode' = 'name',
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5'
);

Changing a column type¶

A type change goes on as a schema_change event when the type the query reads the column as still holds every value of the new type. That is the case when both map to the same Spark type, or when the new type widens to the query's along one of these steps:

• a smaller integer to a larger one: SMALLINT, INT, BIGINT;
• SMALLINT or INT to DOUBLE;
• an integer to DECIMAL(p,s) with at least 5 (SMALLINT), 10 (INT) or 20 (BIGINT) digits before the point;
• DECIMAL(p,s) to a DECIMAL with at least as many digits before the point and after it;
• FLOAT to DOUBLE, DATE to TIMESTAMP_NTZ.

So a varchar length change, or a type made narrower, goes on. A type made wider on the source (int to bigint, decimal(9,2) to decimal(18,4)) fails the batch that holds the DDL while it is planned, before any of it is read:

SchemaChangedError: dbo_orders: the type of captured column(s) amount DECIMAL(18,4)
(read as decimal(9,2)) changed inside the batch (...). Restart the query to re-infer the
schema (and enable delta.enableTypeWidening on bronze for a widening).

To go on:

1. Enable type widening on bronze. The library never changes a table's properties:

   ALTER TABLE bronze.orders SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true');
   

2. Restart the query. It infers the new type, replays the batch, and the append widens the bronze column.

3. On silver, the property alone is not enough. apply_changes adds the columns bronze gains but never changes a column's type: its MERGE runs without schema evolution, so silver keeps the old type. Widen silver's column yourself, with Delta's own statement, before the next call:

   ALTER TABLE silver.orders SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true');
   ALTER TABLE silver.orders ALTER COLUMN amount TYPE DECIMAL(18,4);
   

Without the property, or for a change Delta cannot widen (a type that is not a widening, such as int to varchar), the append fails with a SchemaChangedError that says so: write to a new bronze table, or rewrite this one.

The same rule applies to a newer capture instance: if it captures a column with a type the query's does not hold, the query stops at S and the restart infers it. Two instances that type one column in incompatible ways fail load() itself; pass columns with a type that holds both, or disable the older instance once the stream has read past the newer one's start.

With the columns option¶

With columns, the check compares against the types SQL Server reported when the query started, since the declared ones are your own conversions. A type change made while the query runs is caught at planning, as above. One made while it was stopped is not: the read fails casting to the declared type. Update columns and restart.

Dropping a column¶

CDC keeps a dropped column captured and writes NULL in it from then on, so bronze keeps the column and its new rows read NULL. A snapshot (bootstrap=True, a re-snapshot, snapshot_on_switch) reads NULL for it too, matched by column_id: a column dropped and added back under the same name is another column. To stop carrying it, capture the table again with a new instance without it; the switch event's detail then lists it under "no longer captured, read as NULL".

Errors¶

Pitfalls¶

• Disabling the old instance too early. The changes below S that only it held are gone; the stream fails with DataLossError, which names CDC cleanup and the disabled instance as the possible causes because it cannot tell them apart.
• Changes to a new column made before the new instance exists are lost: an update that touches only that column writes no change row. Enable the new instance soon after the ALTER, or snapshot afterwards (snapshot_on_switch=True).
• Until the old instance is disabled, both capture every change: twice the change-table writes.
• A URI checkpoint without metricsPath writes no events to the facts; step 4 then relies on the warning in the stream's log.
• Facts consumers that look for snapshots must filter event IN ('bootstrap', 'resnapshot'), not event IS NOT NULL: event rows of schema changes are no snapshots (Tables).
• None of this has run on Databricks yet; see Databricks for the differences there.

See also¶

• Options: schemaChangePolicy, columns, metricsPath, snapshot_on_switch.
• Tables: the event and detail columns.
• Permissions, Bootstrap, Data loss, Silver.
• ADR 0023: the measurements on SQL Server 2022 and 2017 behind these rules. Lab check t9 (LAB.md) runs the procedure above under a continuous writer on both.