Output schema¶

The columns of the stream, spark.readStream.format("mssql_cdc"), and of the snapshot, spark.read.format("mssql_cdc_snapshot"). Both have the same schema: six metadata columns, then the captured columns. What the sink adds on top, and the other tables, are in Tables.

from mssql_cdc import register

register(spark)  # stream() does this for you
changes = spark.readStream.format("mssql_cdc").options(**options).load()
current = spark.read.format("mssql_cdc_snapshot").options(**options).load()
changes.printSchema()

Metadata columns¶

An update is two rows, operation 3 and operation 4.

A source column whose name equals one of these, or one the sink adds (_batch_id, _snapshot, _chunk), ignoring case, raises ValueError when the stream or snapshot loads: leave it out with columns.

Ordering changes¶

Order rows with (_start_lsn, _command_id, _seqval, _operation), for example in bronze:

history = spark.read.table("bronze.orders").orderBy(
    "_start_lsn", "_command_id", "_seqval", "_operation"
)

After a switch to a newer capture instance, the two instances number the same change differently in _command_id, but all the rows of one commit come from one instance, so the order holds. Across instances, (_start_lsn, _seqval, _operation) identifies a change. The rows of a whole snapshot share one _start_lsn, below every change read after them, so a MERGE that keeps the latest image per key absorbs the overlap between a snapshot and the stream (Silver does that for you). A chunked snapshot's rows carry their wave's stamp instead, at or after the snapshot's LSN, and can hold a commit newer than it: rebuild from the snapshot's rows and every change after its LSN, and do not read a chunk row as the table's state at its stamp (ADR 0028).

Captured columns¶

Without the columns option, the driver infers them at load() from sys.sp_cdc_get_captured_columns, in capture order. When the table has two capture instances it takes the union by name, older instance first; a column both capture with different types takes the newer type when it holds the older's values, otherwise load() fails with SchemaChangedError. A range of an instance that lacks a column reads it as NULL.

A computed column is left out, and load() logs a warning naming it: CDC stores NULL for it in every change row (below), so its value never reaches a change. Compute it downstream from the columns it is computed from. The client tells it by sys.columns.is_computed, which a login that can SELECT the table sees.

Default types, from mssql_cdc.client:

Alias types, such as sysname, map through their base type. Any other type (sql_variant, geography, geometry, hierarchyid) fails at load() with no default Spark type for SQL Server type ...: pass columns, leaving that column out. Only _commit_ts is converted to UTC; captured datetime and datetime2 values arrive as stored. Every batch is cast to this schema before Spark sees it.

With columns, the declared list and types are the schema, for example order_id INT, status STRING, amount DECIMAL(18,2); the read casts to them. A computed column listed there stays in the schema, NULL in every row, snapshot rows too, and the first planning logs a warning naming it.

NULLs in change rows¶

SQL Server CDC does not store every value in the change table, so some change rows hold NULL where the source row had a value (cdc._CT):

The snapshot reads the source table, so its rows hold the large values these change rows lack. A computed column is not read: the inferred schema leaves it out, and one listed in columns reads NULL in snapshot rows too, so every row agrees. reconcile() does not compare computed columns.

Snapshot rows¶

The snapshot reads the source table itself, under READ COMMITTED (never NOLOCK), in the same schema. A captured column the source table no longer has (dropped, matched by column_id) reads NULL, as its change rows do since the drop; so does a computed column listed in columns. Written by snapshot() or to_delta, snapshot rows also get _batch_id NULL in bronze, and _snapshot, the LSN of the snapshot they belong to (ADR 0016).

With snapshotChunks the snapshot reads only those chunks and has one more column, _chunk INT, the chunk each row was read in. backfill() writes them to bronze with _snapshot set to the chunked snapshot's LSN S, while each wave's _start_lsn is its own stamp, at or after S (ADR 0028).