Advanced AUTO CDC topics

Beyond the basic AUTO CDC and AUTO CDC FROM SNAPSHOT APIs, you can run DML on target tables, read change data feeds from CDC targets, monitor processing metrics, apply partial updates, and track changes with bitemporal storage. For an introduction to the AUTO CDC APIs, see The AUTO CDC APIs: Simplify change data capture with pipelines.

Add, change, or delete data in a target streaming table

If your pipeline publishes tables to Unity Catalog, you can use data manipulation language (DML) statements, including insert, update, delete, and merge statements, to modify the target streaming tables created by AUTO CDC ... INTO statements.

Note

• DML statements that modify the table schema of a streaming table are not supported. Ensure that your DML statements do not attempt to evolve the table schema.
• DML statements that update a streaming table can be run only in a shared Unity Catalog cluster or a SQL warehouse using Databricks Runtime 13.3 LTS and above.
• Because streaming requires append-only data sources, if your processing requires streaming from a source streaming table with changes (for example, by DML statements), set the skipChangeCommits flag when reading the source streaming table. When skipChangeCommits is set, transactions that delete or modify records on the source table are ignored. If your processing does not require a streaming table, you can use a materialized view (which does not have the append-only restriction) as the target table.

Because the pipeline uses a specified SEQUENCE BY column and propagates appropriate sequencing values to the __START_AT and __END_AT columns of the target table (for SCD Type 2), you must ensure that DML statements use valid values for these columns to maintain the proper ordering of records. See How AUTO CDC works.

For more information about using DML statements with streaming tables, see Add, change, or delete data in a streaming table.

The following example inserts an active record with a start sequence of 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Tip

If you need to rename the __START_AT and __END_AT columns in your SCD Type 2 target table (for example, to match downstream schema requirements), create a view over the target table:

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

Read a change data feed from an AUTO CDC target table

In Databricks Runtime 15.2 and above, you can read a change data feed from a streaming table that is the target of AUTO CDC or AUTO CDC FROM SNAPSHOT queries in the same way that you read a change data feed from other Delta tables. The following are required to read the change data feed from a target streaming table:

• The target streaming table must be published to Unity Catalog. See Use Unity Catalog with pipelines.
• To read the change data feed from the target streaming table, you must use Databricks Runtime 15.2 or above. To read the change data feed in a different pipeline, the pipeline must be configured to use Databricks Runtime 15.2 or above.

You read the change data feed from a target streaming table that was created in a Lakeflow pipeline the same way as reading a change data feed from other Delta tables. To learn more about using the Delta change data feed functionality, including examples in Python and SQL, see Use change data feed on Azure Databricks.

Note

The change data feed record includes metadata identifying the type of change event. When a record is updated in a table, the metadata for the associated change records typically includes _change_type values set to update_preimage and update_postimage events.

However, the _change_type values are different if updates are made to the target streaming table that include changing primary key values. When changes include updates to primary keys, the _change_type metadata fields are set to insert and delete events. Changes to primary keys can occur when manual updates are made to one of the key fields with an UPDATE or MERGE statement or, for SCD type 2 tables, when the __start_at field changes to reflect an earlier start sequence value.

The AUTO CDC query determines the primary key values, which differ for SCD type 1 and SCD type 2 processing:

SCD Type	Primary key
SCD type 1, and the pipelines Python interface	The primary key is the value of the keys parameter in the create_auto_cdc_flow() function. For the SQL interface the primary key is the columns defined by the KEYS clause in the AUTO CDC ... INTO statement.
SCD type 2	The primary key is the keys parameter or KEYS clause plus the return value from the coalesce(__START_AT, __END_AT) operation, where __START_AT and __END_AT are the corresponding columns from the target streaming table. This uses __START_AT when available, and __END_AT when __START_AT is null (for example, the initial record).

Get data about records processed by a CDC query in pipelines

Note

The following metrics are captured only by AUTO CDC queries and not by AUTO CDC FROM SNAPSHOT queries.

The following metrics are captured by AUTO CDC queries:

• num_upserted_rows: The number of output rows upserted into the dataset during an update.
• num_deleted_rows: The number of existing output rows deleted from the dataset during an update.

The num_output_rows metric, output for non-CDC flows, is not captured for AUTO CDC queries.

Apply partial updates

When a source sends only the columns that changed, AUTO CDC must distinguish between a column that is absent from a change record, which should leave the target value unchanged, and a column that is explicitly set to null, which should overwrite the target value with null. By default, IGNORE NULL UPDATES treats every null as a "do not update" marker, so it cannot apply an explicit null. To resolve this ambiguity, choose one of the following three methods:

Method	When to use	Behavior
IGNORE NULL UPDATES ON columnList	A small, fixed set of columns should ignore null values, while all other columns apply explicit null values.	The listed columns keep their existing target value when the incoming value is null. All other columns apply explicit null values.
IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList)	Most columns should ignore null values, and only a few should apply explicit null values.	The listed columns apply explicit null values. All other columns keep their existing target value when the incoming value is null.
COLUMNS TO UPDATE	Each change record updates a different set of columns, or the set of updatable columns changes over time.	A source column names the columns to update for each change record. The listed columns are written from the source, including explicit null values. Columns that are not listed keep their existing target value.

COLUMNS TO UPDATE cannot be combined with IGNORE NULL UPDATES, and is not supported for bitemporal tables.

As a rule of thumb, choose COLUMNS TO UPDATE when the producer knows which columns changed in each record and can carry that information in a source column, such as when multiple producers write to the same source or the set of updatable columns grows over time. Choose IGNORE NULL UPDATES ON when the pipeline owner knows the fixed set of updatable columns in advance and prefers to control them in pipeline code.

The following example uses a source column named columnsToUpdate to control which columns each change record updates, including columns set explicitly to null:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

For the full parameter reference, see AUTO CDC INTO (pipelines) and create_auto_cdc_flow.

Bitemporal AUTO CDC

Important

Bitemporal AUTO CDC is in Beta.

SCD Type 1 and Type 2 are unitemporal: they track changes across a single time dimension. Bitemporal extends SCD Type 2 history to track changes across two time dimensions and distinguish between two perspectives:

• Business time: when the event actually happened.
• System time: when the system recorded or ingested the event.

Like SCD Type 2, bitemporal preserves a full history of records. It adds a second timeline so you can reconstruct both what the data showed and what the system believed at any point in the past.

For example, a hedge fund ingests stock data from a source system. Acme Corp's stock price changes on January 1, but the fund does not ingest that update until January 5. Bitemporal AUTO CDC lets the fund answer two distinct questions: what Acme Corp's actual stock price was on January 1 (business time), and what price the system believed when the fund made trading decisions on January 3 (system time). The ability to distinguish between these timelines is useful for auditing, regulatory reporting, and financial decision-making.

To enable bitemporal processing, set STORED AS BITEMPORAL (SQL) or stored_as_scd_type="bitemporal" (Python), use SEQUENCE BY for the business time column, and use SYSTEM SEQUENCE BY for the system time column. The target table adds __SYSTEM_START_AT and __SYSTEM_END_AT columns alongside the SCD Type 2 __START_AT and __END_AT columns. For syntax details, see AUTO CDC INTO (pipelines) or create_auto_cdc_flow.

Bitemporal AUTO CDC examples

The following example creates a bitemporal target table from a small set of synthetic CDC events. The bt column has the business time and the st column has the system time.

Python

from pyspark import pipelines as dp

# Source: synthetic CDC events
dp.create_streaming_table(name="cdc_source")

@dp.append_flow(target="cdc_source", once=True)
def load_cdc_source():
  return spark.createDataFrame(
    [
      (1, "x10", "y10", 10, 100),
      (1, "x20", "y20", 20, 200)
    ],
    schema="id INT, x STRING, y STRING, bt INT, st INT",
  )

# Target: bitemporal table
dp.create_streaming_table(name="target_bitemporal")

dp.create_auto_cdc_flow(
  target = "target_bitemporal",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "bt",
  system_sequence_by = "st",
  stored_as_scd_type = "bitemporal"
)

SQL

-- Source: synthetic CDC events
CREATE OR REFRESH STREAMING TABLE cdc_source_sql;

CREATE FLOW cdc_source_sql AS INSERT INTO ONCE
  cdc_source_sql BY NAME
SELECT * FROM VALUES
  (1, 'x10', 'y10', 10, 100),
  (1, 'x20', 'y20', 20, 200)
  AS t(id, x, y, bt, st);

-- Target: bitemporal table
CREATE OR REFRESH STREAMING TABLE target_bitemporal_sql;

CREATE FLOW target_bitemporal_sql AS AUTO CDC INTO
  target_bitemporal_sql
FROM
  stream(cdc_source_sql)
KEYS
  (id)
SEQUENCE BY
  bt
SYSTEM SEQUENCE BY
  st
STORED AS
  BITEMPORAL;

The following sequence of changes shows how a bitemporal table records an insert, an update, an out-of-order update, and a delete for a single company. The sequencing column generates the __START_AT and __END_AT (business time) columns, and the system sequencing column generates the __SYSTEM_START_AT and __SYSTEM_END_AT (system time) columns:

Column	Description
__START_AT	The business time at which this row became valid.
__END_AT	The business time at which this row's validity ends. null if valid indefinitely.
__SYSTEM_START_AT	The system time at which this row's data and business time interval are known to be true.
__SYSTEM_END_AT	The system time at which this row's data and business time interval are known to be invalidated. null if known to be true indefinitely.

The system handles events that arrive in any order across both timelines. When an event arrives with an earlier business time or system time than events already processed, the system corrects the affected history rather than appending only to the end.

Change 1: Insert

Company A is added at 7/18/2025 10:01:00 (business time) but is not ingested until 10:05:00 (system time).

Input:

CompanyId	Data Point	Sequencing	System Sequencing	Operation
A	XFv1	7/18/2025 10:01:00	7/18/2025 10:05:00	INSERT

Output:

CompanyId	Data Point	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	7/18/2025 10:01:00	NULL	7/18/2025 10:05:00	NULL

XFv1 is valid starting at 10:01:00 with no known end. The system learned of this fact at system time 10:05:00, with no known end.

Change 2: Update

Company A is updated at 7/18/2025 12:15:43 (business time), and the system consumes the event at 12:20:00 (system time). The system preserves both what it believed before the update was known and the corrected business history after the update is ingested.

Input:

CompanyId	Data Point	Sequencing	System Sequencing	Operation
A	XFv2	7/18/2025 12:15:43	7/18/2025 12:20:00	UPDATE

Output:

CompanyId	Data Point	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	7/18/2025 10:01:00	NULL	7/18/2025 10:05:00	7/18/2025 12:20:00
A	XFv1	7/18/2025 10:01:00	7/18/2025 12:15:43	7/18/2025 12:20:00	NULL
A	XFv2	7/18/2025 12:15:43	NULL	7/18/2025 12:20:00	NULL

XFv1 was believed valid from 10:01:00 with no known end, and the system held that belief from 10:05:00 until 12:20:00. XFv1 is now known to be valid only until 12:15:43, a corrected history effective from system time 12:20:00 with no known end. XFv2 is valid starting at 12:15:43 with no known end, and was learned at system time 12:20:00.

Change 3: Out-of-order update

An out-of-order update arrives indicating that Company A was actually updated at 7/18/2025 12:05:00 (business time), but it is not ingested until 12:25:00 (system time). When an update arrives later in system time but with a preceding business time, the system corrects the historical business time and preserves both what it believed before the out-of-order update and the corrected history.

Input:

CompanyId	Data Point	Sequencing	System Sequencing	Operation
A	XFv3	7/18/2025 12:05:00	7/18/2025 12:25:00	UPDATE

Output:

CompanyId	Data Point	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	7/18/2025 10:01:00	NULL	7/18/2025 10:05:00	7/18/2025 12:20:00
A	XFv1	7/18/2025 10:01:00	7/18/2025 12:15:43	7/18/2025 12:20:00	7/18/2025 12:25:00
A	XFv1	7/18/2025 10:01:00	7/18/2025 12:05:00	7/18/2025 12:25:00	NULL
A	XFv3	7/18/2025 12:05:00	7/18/2025 12:15:43	7/18/2025 12:25:00	NULL
A	XFv2	7/18/2025 12:15:43	NULL	7/18/2025 12:20:00	NULL

XFv1 was believed valid from 10:01:00 to 12:15:43, and that belief is now valid in system time up to 12:25:00. The new update corrects XFv1's business validity to end at 12:05:00, a corrected history effective from system time 12:25:00. XFv3 is now known to be valid from 12:05:00 until 12:15:43, a belief valid in system time from 12:25:00 with no known end.

Change 4: Delete

Company A is deleted at 7/18/2025 12:30:00, and the system consumes the event at 12:30:00. Because a delete operation represents the end of the entity's business existence, the system creates no replacement row. XFv2 appears in two rows, preserving a complete audit trail of both when the company ceased to exist and when the system learned of the deletion.

Input:

CompanyId	Data Point	Sequencing	System Sequencing	Operation
A	XFv2	7/18/2025 12:30:00	7/18/2025 12:30:00	DELETE

Output:

CompanyId	Data Point	__START_AT	__END_AT	__SYSTEM_START_AT	__SYSTEM_END_AT
A	XFv1	7/18/2025 10:01:00	NULL	7/18/2025 10:05:00	7/18/2025 12:20:00
A	XFv1	7/18/2025 10:01:00	7/18/2025 12:15:43	7/18/2025 12:20:00	7/18/2025 12:25:00
A	XFv1	7/18/2025 10:01:00	7/18/2025 12:05:00	7/18/2025 12:25:00	NULL
A	XFv3	7/18/2025 12:05:00	7/18/2025 12:15:43	7/18/2025 12:25:00	NULL
A	XFv2	7/18/2025 12:15:43	NULL	7/18/2025 12:20:00	7/18/2025 12:30:00
A	XFv2	7/18/2025 12:15:43	7/18/2025 12:30:00	7/18/2025 12:30:00	NULL

XFv2 was valid from 12:15:43 with no known end, and the system held that belief from 12:20:00 to 12:30:00. After the delete is ingested, XFv2 is known to be valid only until 12:30:00, a corrected history effective from system time 12:30:00.

What data objects are used for CDC processing in a pipeline?

When you declare the target table in the Hive metastore, two data structures are created:

• A view using the name assigned to the target table.
• An internal backing table used by the pipeline to manage CDC processing. This table is named by prepending __apply_changes_storage_ to the target table name.

For example, if you declare a target table named dp_cdc_target, you see a view named dp_cdc_target and a table named __apply_changes_storage_dp_cdc_target in the metastore. Query the view to access the processed data. Do not modify the backing table directly.

Note

These data structures apply only to AUTO CDC processing, not AUTO CDC FROM SNAPSHOT processing. They also apply only to Hive metastore, not Unity Catalog.