Bitemporal AUTO CDC

Bitemporal AUTO CDC is a change data capture (CDC) storage type that records the history of each record across two timelines: when a change was true in the real world, and when your system learned about it. Use it when you need to answer both "What was true on a given date?" and "What did the system believe on a given date?", for example, for audits, regulatory reporting, or reproducing a past decision.

This page explains how bitemporal tables work, walks through a complete example, and shows how to process snapshot sources. For a conceptual comparison with slowly changing dimension (SCD) Type 1 and SCD Type 2, see Bitemporal: History across business time and system time. For the other AUTO CDC storage types, see The AUTO CDC APIs: Simplify change data capture with pipelines.

How bitemporal tables work

SCD Type 1 and SCD Type 2 tables track changes across a single timeline, the order defined by the SEQUENCE BY column. A bitemporal table tracks two timelines:

  • Business time (also called valid time): when a change took effect in the real world. For example, the date an employee's promotion became effective. You specify it with SEQUENCE BY.
  • System time (also called transaction time): when the system recorded the change. For example, the time the HR system sent the promotion to your pipeline. You specify it with SYSTEM SEQUENCE BY.

When a change arrives late, out of order, or as a correction, the pipeline doesn't overwrite history. Instead, it closes the rows that the system previously believed, at the new system time, and adds rows that reflect the corrected business history. You can reconstruct both the latest view of history and the view the system held at any earlier time.

The target table contains the columns from the source plus the following four columns:

Column Description
__START_AT The business time at which this version of the record became valid.
__END_AT The business time at which this version of the record stopped being valid. null if it is still valid.
__SYSTEM_START_AT The system time at which the system started to believe this row.
__SYSTEM_END_AT The system time at which the system stopped believing this row, because a later change replaced it. null if the system still believes it.

Rows where __SYSTEM_END_AT is null represent what the system currently believes. Rows with a non-null __SYSTEM_END_AT are earlier beliefs that the pipeline retains for auditing.

Requirements

  • Your pipeline must meet the requirements for the AUTO CDC APIs.
  • The source must provide a business time column for SEQUENCE BY and a system time column for SYSTEM SEQUENCE BY. Both columns must be sortable data types, and they can have different data types.

For features that bitemporal tables don't support, see AUTO CDC limitations.

Example: Track employee roles with bitemporal AUTO CDC

This walkthrough tracks employee roles from an HR system. Each change event has an effectiveDate column, which is the business time when the role change took effect, and a recordedAt column, which is the system time when the HR system sent the event. You create sample data, run a pipeline, and then add an in-order update, an out-of-order update, and a correction, running the pipeline after each step to see how the target table changes.

Step 1: Create sample data

Run the following statements to create the source table with an initial load of two employees. This code is not intended to be run as part of a pipeline definition. Run it from the exploration folder of your pipeline, rather than the transformations folder.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE OR REPLACE TABLE main.cdc_tutorial.employees_cdf (
  employeeId INT,
  name STRING,
  role STRING,
  operation STRING,
  effectiveDate DATE,
  recordedAt TIMESTAMP
);

INSERT INTO main.cdc_tutorial.employees_cdf VALUES
  -- Initial load. Dana's role is recorded incorrectly. Step 5 corrects it.
  (1, 'Chris', 'Owner',    'INSERT', DATE'2025-01-01', TIMESTAMP'2025-01-02 09:00:00'),
  (2, 'Dana',  'Engineer', 'INSERT', DATE'2025-01-01', TIMESTAMP'2025-01-02 09:00:00');

Step 2: Create and run the pipeline

The following code reads the change data feed and applies the changes to a bitemporal streaming table. SEQUENCE BY sets the business time and SYSTEM SEQUENCE BY sets the system time. You need a pipeline to run this code. See Develop and debug ETL pipelines with the Lakeflow Pipelines Editor.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import expr

@dp.view
def employees():
  return spark.readStream.table("main.cdc_tutorial.employees_cdf")

dp.create_streaming_table("employees_bitemporal")

dp.create_auto_cdc_flow(
  target = "employees_bitemporal",
  source = "employees",
  keys = ["employeeId"],
  sequence_by = "effectiveDate",
  system_sequence_by = "recordedAt",
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation"],
  stored_as_scd_type = "bitemporal"
)

SQL

CREATE OR REFRESH STREAMING TABLE employees_bitemporal;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  employees_bitemporal
FROM
  stream(main.cdc_tutorial.employees_cdf)
KEYS
  (employeeId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  effectiveDate
SYSTEM SEQUENCE BY
  recordedAt
COLUMNS * EXCEPT
  (operation)
STORED AS
  BITEMPORAL;

Run the pipeline. The tables in this walkthrough show the employeeId, name, and role columns and the four bitemporal columns. The target table also contains the effectiveDate and recordedAt source columns. After the first run, the target table contains the following records:

employeeId name role __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
1 Chris Owner 2025-01-01 null 2025-01-02 09:00:00 null
2 Dana Engineer 2025-01-01 null 2025-01-02 09:00:00 null

Both roles are valid from January 1 with no known end. The system learned about them on January 2.

Step 3: Add an in-order update

Chris is promoted to Manager, effective March 1. The HR system sends the change the same day. Add the event to the source table:

INSERT INTO main.cdc_tutorial.employees_cdf VALUES
  (1, 'Chris', 'Manager', 'UPDATE', DATE'2025-03-01', TIMESTAMP'2025-03-01 09:00:00');

Run the pipeline again. The target table contains the following records for Chris. Dana's record doesn't change.

employeeId name role __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
1 Chris Owner 2025-01-01 null 2025-01-02 09:00:00 2025-03-01 09:00:00
1 Chris Owner 2025-01-01 2025-03-01 2025-03-01 09:00:00 null
1 Chris Manager 2025-03-01 null 2025-03-01 09:00:00 null

The pipeline doesn't edit the original Owner row. It closes that row in system time at March 1, 09:00, which records that the system believed "Owner with no end date" until then. It adds a row with the updated belief, Owner from January 1 until March 1, and a row for Manager from March 1 with no known end.

Step 4: Add an out-of-order update

On March 15, the HR system sends an event that it missed earlier. Chris was a Team Lead between the Owner and Manager roles, effective February 1. This event arrives after the March 1 promotion in system time, but its business time is earlier. Add the event to the source table:

INSERT INTO main.cdc_tutorial.employees_cdf VALUES
  (1, 'Chris', 'Team Lead', 'UPDATE', DATE'2025-02-01', TIMESTAMP'2025-03-15 09:00:00');

Run the pipeline again. The target table contains the following records for Chris:

employeeId name role __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
1 Chris Owner 2025-01-01 null 2025-01-02 09:00:00 2025-03-01 09:00:00
1 Chris Owner 2025-01-01 2025-03-01 2025-03-01 09:00:00 2025-03-15 09:00:00
1 Chris Owner 2025-01-01 2025-02-01 2025-03-15 09:00:00 null
1 Chris Team Lead 2025-02-01 2025-03-01 2025-03-15 09:00:00 null
1 Chris Manager 2025-03-01 null 2025-03-01 09:00:00 null

The pipeline inserts Team Lead into the business-time history instead of appending it to the end. The belief that Chris was Owner until March 1 is closed at March 15, 09:00. From that time, the system believes that Chris was Owner until February 1 and Team Lead from February 1 until March 1. The Manager row doesn't change, because the late event doesn't affect it.

Step 5: Correct an erroneous value

Dana's role was recorded incorrectly in the initial load. Dana has been an Analyst since January 1, not an Engineer. To correct a value, add a record with the corrected value, the same business time as the erroneous record, and a later system time:

INSERT INTO main.cdc_tutorial.employees_cdf VALUES
  (2, 'Dana', 'Analyst', 'UPDATE', DATE'2025-01-01', TIMESTAMP'2025-04-01 09:00:00');

Run the pipeline again. The target table contains the following records for Dana. Chris's records don't change.

employeeId name role __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
2 Dana Engineer 2025-01-01 null 2025-01-02 09:00:00 2025-04-01 09:00:00
2 Dana Analyst 2025-01-01 null 2025-04-01 09:00:00 null

Both rows have the same business-time range. The erroneous row's system-time range ends at April 1, 09:00, when the corrected row's system-time range starts. Queries that filter on __SYSTEM_END_AT IS NULL return only the corrected value, but the table still records that the system believed Dana was an Engineer from January 2 until April 1.

Delete records

To delete a record, send an event that matches the APPLY AS DELETE WHEN condition, with the business time when the record stopped existing and the system time when the system learned about it. The pipeline closes the record's business-time range at the delete's business time, effective from the delete's system time, and retains the earlier belief.

For example, if the source sends (1, null, null, 'DELETE', DATE'2025-06-30', TIMESTAMP'2025-06-30 17:00:00'), the Manager row for Chris is closed at system time 2025-06-30 17:00:00. A new Manager row with a business-time range from 2025-03-01 to 2025-06-30 is added from that system time.

Query a bitemporal table

A bitemporal table has two validity ranges per row, so filter on both timelines. The following queries use the employees_bitemporal table after step 5 of the walkthrough.

Query the current state

The following query returns the current state, as the system currently believes it:

SELECT employeeId, name, role
FROM employees_bitemporal
WHERE __END_AT IS NULL AND __SYSTEM_END_AT IS NULL;

This query returns Chris as Manager and Dana as Analyst.

Query business-time history

The following query returns the business-time history, as the system currently believes it:

SELECT employeeId, name, role, __START_AT, __END_AT
FROM employees_bitemporal
WHERE __SYSTEM_END_AT IS NULL
ORDER BY employeeId, __START_AT;

This query returns the corrected history: Chris as Owner from 2025-01-01 to 2025-02-01, Team Lead from 2025-02-01 to 2025-03-01, and Manager from 2025-03-01, and Dana as Analyst from 2025-01-01.

Query a past belief about a past business time

The following query asks what role each employee had on February 15, according to what the system knew on March 10. The business time is 2025-02-15 and the system time is 2025-03-10 00:00:00:

SELECT employeeId, name, role
FROM employees_bitemporal
WHERE __START_AT <= DATE'2025-02-15'
  AND (__END_AT > DATE'2025-02-15' OR __END_AT IS NULL)
  AND __SYSTEM_START_AT <= TIMESTAMP'2025-03-10 00:00:00'
  AND (__SYSTEM_END_AT > TIMESTAMP'2025-03-10 00:00:00' OR __SYSTEM_END_AT IS NULL);

This query returns Chris as Owner and Dana as Engineer, because on March 10 the system had not yet received the Team Lead event or Dana's correction. To see what the system believes today, change the system time to the current time. The query then returns Chris as Team Lead and Dana as Analyst.

Process snapshots with bitemporal AUTO CDC

Bitemporal tables don't support AUTO CDC FROM SNAPSHOT. If your source only provides periodic full snapshots, derive a change feed from the snapshots and process that change feed with bitemporal AUTO CDC.

Tip

If you only need one timeline, use AUTO CDC FROM SNAPSHOT with SCD Type 2 instead. See AUTO CDC FROM SNAPSHOT examples.

Option 1: Merge snapshots into a table with change data feed enabled

If you control how snapshots land in Azure Databricks, merge each snapshot into a Delta table that has change data feed enabled. The MERGE statement writes only the rows that were inserted, updated, or removed, so the table's change data feed is a CDC feed that AUTO CDC can process.

  1. Create the table, and enable change data feed. The isDeleted column marks records that are absent from the latest snapshot:

    CREATE TABLE main.cdc_tutorial.employees_current (
  employeeId INT,
  name STRING,
  role STRING,
  snapshotDate DATE,
  isDeleted BOOLEAN
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

  2. Merge each new snapshot into the table. The following example reads a table named main.cdc_tutorial.employees_snapshots that accumulates snapshots. Each row has a snapshotDate column with the business date that its snapshot represents, and each employeeId can appear only one time per snapshot. The script merges each snapshot that is newer than the latest merged snapshot, one at a time, in ascending snapshotDate order:

    BEGIN
  FOR snap AS
    SELECT DISTINCT snapshotDate
    FROM main.cdc_tutorial.employees_snapshots
    WHERE snapshotDate > (
      SELECT COALESCE(MAX(snapshotDate), DATE'0001-01-01')
      FROM main.cdc_tutorial.employees_current)
    ORDER BY snapshotDate
  DO
    MERGE INTO main.cdc_tutorial.employees_current AS t
    USING (
      SELECT * FROM main.cdc_tutorial.employees_snapshots
      WHERE snapshotDate = snap.snapshotDate
    ) AS s
    ON t.employeeId = s.employeeId
    WHEN MATCHED AND (t.isDeleted OR NOT (t.name <=> s.name AND t.role <=> s.role)) THEN
      UPDATE SET name = s.name, role = s.role, snapshotDate = s.snapshotDate, isDeleted = false
    WHEN NOT MATCHED THEN
      INSERT (employeeId, name, role, snapshotDate, isDeleted)
      VALUES (s.employeeId, s.name, s.role, s.snapshotDate, false)
    WHEN NOT MATCHED BY SOURCE AND NOT t.isDeleted THEN
      UPDATE SET snapshotDate = snap.snapshotDate, isDeleted = true;
  END FOR;
END;

    The script uses SQL scripting, which requires Databricks SQL or Databricks Runtime 16.3 and above. It behaves as follows:

    • Each snapshot is merged in its own commit, so each snapshot gets its own _commit_timestamp. If a run is skipped, the next run merges every pending snapshot in order.
    • The merge marks removed records with isDeleted = true instead of deleting them. As a result, each removal appears in the change data feed with the date of the snapshot that removed it, and that date becomes the business time of the delete.
    • The latest merged snapshotDate in employees_current records progress, and it is written in the same commit as the data. Running the script again without new snapshots doesn't add change data feed rows.
    • Snapshots older than the latest merged snapshot are skipped, because merging one would move the current state backward.

  3. In your pipeline, read the change data feed and apply it with bitemporal AUTO CDC. The _commit_timestamp metadata column, which records when the merge committed, is the system time. Skip update_preimage rows, which contain the values before each update. Set the starting version to 0. Otherwise, the first pipeline run reads the table's current contents as inserts, and every earlier merge collapses into one system time.

    Python

    from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def employees_changes():
  return (spark.readStream
    .option("readChangeFeed", "true")
    .option("startingVersion", 0)
    .table("main.cdc_tutorial.employees_current")
    .filter(col("_change_type") != "update_preimage")
  )

dp.create_streaming_table("employees_bitemporal_from_snapshots")

dp.create_auto_cdc_flow(
  target = "employees_bitemporal_from_snapshots",
  source = "employees_changes",
  keys = ["employeeId"],
  sequence_by = "snapshotDate",
  system_sequence_by = "_commit_timestamp",
  apply_as_deletes = expr("isDeleted"),
  except_column_list = ["isDeleted", "_change_type", "_commit_version"],
  stored_as_scd_type = "bitemporal"
)

    SQL

    CREATE OR REFRESH STREAMING TABLE employees_bitemporal_from_snapshots;

CREATE TEMPORARY VIEW employees_changes AS
SELECT *
FROM STREAM main.cdc_tutorial.employees_current WITH (readChangeFeed = true, startingVersion = 0)
WHERE _change_type != 'update_preimage';

CREATE FLOW apply_snapshot_changes AS AUTO CDC INTO
  employees_bitemporal_from_snapshots
FROM
  stream(employees_changes)
KEYS
  (employeeId)
APPLY AS DELETE WHEN
  isDeleted
SEQUENCE BY
  snapshotDate
SYSTEM SEQUENCE BY
  _commit_timestamp
COLUMNS * EXCEPT
  (isDeleted, _change_type, _commit_version)
STORED AS
  BITEMPORAL;

Option 2: Compare consecutive snapshots in a job

If the snapshots are stored together in one table, with a column that identifies each snapshot, compare consecutive snapshots in a job task that runs before the pipeline, and append the differences to a change table. The pipeline then reads the change table as the source for bitemporal AUTO CDC.

  1. Create the change table:

    CREATE TABLE main.cdc_tutorial.employees_snapshot_changes (
  employeeId INT,
  name STRING,
  role STRING,
  operation STRING,
  snapshotDate DATE,
  recordedAt TIMESTAMP
);

  2. In a job task, compare each new snapshot with the snapshot before it. The following example reads a table named main.cdc_tutorial.employees_snapshots, which has a snapshotDate column. Each snapshot's date is the business time, and the time the job runs is the system time:

    INSERT INTO main.cdc_tutorial.employees_snapshot_changes
WITH snapshot_pairs AS (
  SELECT snapshotDate, LAG(snapshotDate) OVER (ORDER BY snapshotDate) AS prevSnapshotDate
  FROM (SELECT DISTINCT snapshotDate FROM main.cdc_tutorial.employees_snapshots)
),
pending AS (
  SELECT * FROM snapshot_pairs
  WHERE snapshotDate > (
    SELECT COALESCE(MAX(snapshotDate), DATE'0001-01-01')
    FROM main.cdc_tutorial.employees_snapshot_changes)
),
curr AS (
  SELECT p.snapshotDate, s.employeeId, s.name, s.role
  FROM pending p
  JOIN main.cdc_tutorial.employees_snapshots s ON s.snapshotDate = p.snapshotDate
),
prev AS (
  SELECT p.snapshotDate, s.employeeId, s.name, s.role
  FROM pending p
  JOIN main.cdc_tutorial.employees_snapshots s ON s.snapshotDate = p.prevSnapshotDate
)
SELECT
  COALESCE(curr.employeeId, prev.employeeId) AS employeeId,
  curr.name,
  curr.role,
  CASE
    WHEN prev.employeeId IS NULL THEN 'INSERT'
    WHEN curr.employeeId IS NULL THEN 'DELETE'
    ELSE 'UPDATE'
  END AS operation,
  COALESCE(curr.snapshotDate, prev.snapshotDate) AS snapshotDate,
  current_timestamp() AS recordedAt
FROM curr
FULL OUTER JOIN prev
  ON curr.snapshotDate = prev.snapshotDate AND curr.employeeId = prev.employeeId
WHERE prev.employeeId IS NULL
  OR curr.employeeId IS NULL
  OR NOT (curr.name <=> prev.name AND curr.role <=> prev.role);

    The statement behaves as follows:

    • Every snapshot that is newer than the latest processed snapshot is compared with the snapshot just before it. If a run is skipped, or several snapshots arrive together, the next run compares every pair in order. The first snapshot has no earlier snapshot, so all of its rows become INSERT events.
    • The latest snapshotDate in the change table records progress, and the whole run is one INSERT, so the progress and the change rows are committed together. Running the statement again without new snapshots inserts nothing.
    • Snapshots compared in the same run share one recordedAt value, because the system learned about them at the same time.
    • Snapshots older than the latest processed snapshot are ignored.

  3. In your pipeline, apply the change table with bitemporal AUTO CDC:

    Python

    from pyspark import pipelines as dp
from pyspark.sql.functions import expr

@dp.view
def employees_snapshot_changes():
  return spark.readStream.table("main.cdc_tutorial.employees_snapshot_changes")

dp.create_streaming_table("employees_bitemporal_from_snapshot_diffs")

dp.create_auto_cdc_flow(
  target = "employees_bitemporal_from_snapshot_diffs",
  source = "employees_snapshot_changes",
  keys = ["employeeId"],
  sequence_by = "snapshotDate",
  system_sequence_by = "recordedAt",
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation"],
  stored_as_scd_type = "bitemporal"
)

    SQL

    CREATE OR REFRESH STREAMING TABLE employees_bitemporal_from_snapshot_diffs;

CREATE FLOW apply_snapshot_diffs AS AUTO CDC INTO
  employees_bitemporal_from_snapshot_diffs
FROM
  stream(main.cdc_tutorial.employees_snapshot_changes)
KEYS
  (employeeId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  snapshotDate
SYSTEM SEQUENCE BY
  recordedAt
COLUMNS * EXCEPT
  (operation)
STORED AS
  BITEMPORAL;

Note

  • To use a column with the date that each change took effect as the business time, add it to the change table as a separate column and use it for SEQUENCE BY. Keep snapshotDate, because the statement uses it to record progress.
  • The statement can't detect a snapshot with zero rows, and a snapshot that is still loading when the job runs produces incorrect deletes. To handle both, have the loader write each snapshot's date to a table of completed snapshots after the load finishes. Then build snapshot_pairs from that table instead of from DISTINCT snapshotDate.

Additional resources

  • อัปเดตล่าสุดเมื่อ