The AUTO CDC APIs: Simplify change data capture with pipelines

Lakeflow pipelines simplify change data capture (CDC) with the AUTO CDC and AUTO CDC FROM SNAPSHOT APIs. These APIs automate the complexity of computing slowly changing dimensions (SCD) Type 1 and Type 2 from either a CDC feed or database snapshots, and the AUTO CDC API also computes bitemporal history from a CDC feed. To learn more about SCD Type 1, SCD Type 2, and bitemporal, see Change data capture and snapshots.

note

The AUTO CDC APIs replace the APPLY CHANGES APIs and have the same syntax. The APPLY CHANGES APIs are still available, but Databricks recommends using the AUTO CDC APIs in their place.

The API you use depends on the source of your change data:

AUTO CDC: Use this when the source database has a CDC feed enabled. AUTO CDC processes changes from a change data feed (CDF). It is supported in both the pipeline SQL and Python interfaces.
AUTO CDC FROM SNAPSHOT: Use this when CDC is not enabled on the source database and only snapshots are available. This API compares snapshots to determine changes and then processes them. It is supported in both the pipeline SQL and Python interfaces.

For SCD Type 1 targets, you can also combine an AUTO CDC FROM SNAPSHOT flow with one or more AUTO CDC flows. Use this pattern when a source provides periodic authoritative snapshots and a lower-latency CDC feed between snapshots.

The APIs support the following storage types for target tables:

SCD Type 1: Update records directly. History is not retained for updated records. Supported by both APIs.
SCD Type 2: Retain a history of records, either on all updates or on updates to a specified set of columns. Supported by both APIs.
Bitemporal: Retain a history of records across two timelines: business time, when a change took effect, and system time, when the system recorded it. Supported by AUTO CDC only. See Bitemporal AUTO CDC.

AUTO CDC also supports partial updates, where a change record updates only a subset of columns. See Apply partial updates.

The AUTO CDC APIs are not supported by Apache Spark Declarative Pipelines.

For syntax and other references, see AUTO CDC INTO (pipelines), create_auto_cdc_flow, and create_auto_cdc_from_snapshot_flow.

note

Use these APIs to update tables in your pipelines based on changes in source data. To learn how to record and query row-level change information for Delta tables, see Use change data feed on Databricks.

Requirements

To use the CDC APIs, your pipeline must be configured to use serverless Lakeflow pipelines or the Lakeflow pipelines Pro or Advanced editions.

How AUTO CDC works

To perform CDC processing with AUTO CDC, create a streaming table and then use the AUTO CDC ... INTO statement in SQL or the create_auto_cdc_flow() function in Python to specify the source, keys, and sequencing for the change feed. For an explanation of how sequencing and SCD logic work, see Change data capture and snapshots. See the AUTO CDC examples.

For initial hydration from a source with a change feed, use AUTO CDC with a once flow and then continue processing the change feed. See Replicate an external RDBMS table using AUTO CDC.

For syntax details, see AUTO CDC INTO (pipelines) or create_auto_cdc_flow.

How AUTO CDC FROM SNAPSHOT works

AUTO CDC FROM SNAPSHOT determines changes in source data by comparing in-order snapshots. You can read snapshots from a Delta table, cloud storage files, or JDBC directly. It is supported in both the pipeline SQL and Python interfaces.

To perform CDC processing with AUTO CDC FROM SNAPSHOT, create a streaming table and then define the flow:

In SQL, use the AUTO CDC ... FROM SNAPSHOT form of the CREATE FLOW statement or the embedded FLOW AUTO CDC clause of CREATE STREAMING TABLE. The source is a required FROM SNAPSHOT (snapshot_query) clause and an optional WITH VERSION (version_query) clause that selects the next snapshot version to process. See CREATE FLOW (pipelines).
In Python, use the create_auto_cdc_from_snapshot_flow() function to specify the snapshot, keys, and other arguments. See create_auto_cdc_from_snapshot_flow.

For details on the two ingestion patterns and when to use each, see Snapshot processing patterns. See the AUTO CDC FROM SNAPSHOT examples.

Combine AUTO CDC FROM SNAPSHOT and AUTO CDC

The snapshot version and the CDC sequencing column form one ordering domain. A CDC event newer than the latest snapshot takes precedence over the snapshot. A newer snapshot takes precedence over older CDC events and treats keys that are absent from the snapshot as deleted at the snapshot version. Newer CDC events can preserve or restore those keys.

A unified target must meet the following requirements:

The target uses SCD Type 1.
The target has exactly one AUTO CDC FROM SNAPSHOT flow and one or more AUTO CDC flows.
Each AUTO CDC flow has a unique name.
All flows use the same number of keys in the same order. Snapshot-flow key names are compared case-insensitively with AUTO CDC key names. Multiple AUTO CDC flows must use identical key names and casing.
The snapshot version and each CDC sequencing column have exactly the same data type.
The AUTO CDC FROM SNAPSHOT flow does not define expectations.
The AUTO CDC flows do not use IGNORE NULL UPDATES or its column-list variants.
The pipeline uses triggered mode.
If you add an AUTO CDC flow to an existing AUTO CDC FROM SNAPSHOT target, the target must not contain user columns whose names conflict with reserved AUTO CDC system columns.

Both AUTO CDC and AUTO CDC FROM SNAPSHOT flows can use the Python or SQL pipeline interface. You can mix SQL and Python flows in the same target.

Use once=True in Python or ONCE in SQL on the AUTO CDC FROM SNAPSHOT flow to backfill a snapshot one time while the AUTO CDC flows continue to process new events on subsequent updates. A full refresh of the target re-runs the one-time snapshot flow. For a complete example, see Add a backfill to an AUTO CDC SCD Type 1 table.

This pattern does not support SCD Type 2 or bitemporal targets.

Use multiple columns for sequencing

To sequence by multiple columns (for example, a timestamp and an ID to break ties), use a STRUCT to combine them. The API orders by the first field first, and in the event of a tie, considers the second field, and so on.

SQL
Python

SQL
SEQUENCE BY STRUCT(timestamp_col, id_col)

Python
sequence_by = struct("timestamp_col", "id_col")

AUTO CDC examples

The following examples demonstrate SCD Type 1 and Type 2 processing using a change data feed source. The sample data creates new user records, deletes a user record, and updates user records. In the SCD Type 1 example, the last UPDATE operations arrive late and are dropped from the target table, demonstrating out-of-order event handling.

The following are the input records used in these examples. This data is created by running the query in the Create sample data section.

userId	name	city	operation	sequenceNum
124	Raul	Oaxaca	INSERT	1
123	Isabel	Monterrey	INSERT	1
125	Mercedes	Tijuana	INSERT	2
126	Lily	Cancun	INSERT	2
123	null	null	DELETE	6
125	Mercedes	Guadalajara	UPDATE	6
125	Mercedes	Mexicali	UPDATE	5
123	Isabel	Chihuahua	UPDATE	5

userId	name	city	operation	sequenceNum
124	Raul	Oaxaca	INSERT	1
123	Isabel	Monterrey	INSERT	1
125	Mercedes	Tijuana	INSERT	2
126	Lily	Cancun	INSERT	2
123	null	null	DELETE	6
125	Mercedes	Guadalajara	UPDATE	6
125	Mercedes	Mexicali	UPDATE	5
123	Isabel	Chihuahua	UPDATE	5

If you uncomment the final row in the sample data generation query, it inserts the following record that specifies to truncate the table (clear the table) at sequenceNum=3:

userId	name	city	operation	sequenceNum
null	null	null	TRUNCATE	3

note

All the following examples include options to specify both DELETE and TRUNCATE operations, but each is optional.

Create sample data

Run the following statements to create a sample dataset. 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.

SQL
CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Process SCD Type 1 updates

SCD Type 1 keeps only the latest version of each record. The following example reads from the change data feed created above and applies changes to a streaming table target. You need a pipeline to run this code. See Develop and debug ETL pipelines with the Lakeflow Pipelines Editor.

Python
SQL

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

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

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL
CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

After running the SCD Type 1 example, the target table contains the following records:

userId	name	city
124	Raul	Oaxaca
125	Mercedes	Guadalajara
126	Lily	Cancun

User 123 (Isabel) was deleted and does not appear. User 125 (Mercedes) shows only the latest city (Guadalajara) because SCD Type 1 overwrites previous values. The earlier UPDATE at sequenceNum=5 was dropped because a later update at sequenceNum=6 arrived.

After running the example with the TRUNCATE record uncommented, the table is cleared at sequenceNum=3. This means that records 124 and 126 are not in the table, and the final target table contains only the following record:

userId	name	city
125	Mercedes	Guadalajara

Process SCD Type 2 updates

SCD Type 2 preserves a complete history of changes by creating new rows for each version of a record, with __START_AT and __END_AT columns indicating when each version was active.

Python
SQL

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

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

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL
CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

After running the SCD Type 2 example, the target table contains the following records:

userId	name	city	__START_AT	__END_AT
123	Isabel	Monterrey	1	5
123	Isabel	Chihuahua	5	6
124	Raul	Oaxaca	1	null
125	Mercedes	Tijuana	2	5
125	Mercedes	Mexicali	5	6
125	Mercedes	Guadalajara	6	null
126	Lily	Cancun	2	null

The table preserves complete history. User 123 has two versions (ended at sequence 6 when deleted). User 125 has three versions showing city changes. Records with __END_AT = null are currently active.

Track a column subset with SCD Type 2

By default, SCD Type 2 creates a new version whenever any column value changes. You can specify a subset of columns to track, so that changes to other columns update the current version in place rather than generating a new history record.

The following example excludes the city column from history tracking:

Python
SQL

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

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

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL
CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Because city changes are not tracked, city updates overwrite the current row instead of creating a new version. The target table contains the following records:

userId	name	city	__START_AT	__END_AT
123	Isabel	Chihuahua	1	6
124	Raul	Oaxaca	1	null
125	Mercedes	Guadalajara	2	null
126	Lily	Cancun	2	null

Process bitemporal updates

Bitemporal tables track when each change took effect (business time) and when the system recorded it (system time). Use SEQUENCE BY for the business time column and SYSTEM SEQUENCE BY (system_sequence_by in Python) for the system time column. Bitemporal targets require both. The following snippets show only the flow definition. For complete pipeline code, see Example: Track employee roles with bitemporal AUTO CDC.

Python
SQL

Python
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 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;

For the columns that bitemporal targets add, and a full walkthrough with sample data, out-of-order updates, corrections, and queries, see Bitemporal AUTO CDC.

AUTO CDC FROM SNAPSHOT examples

The following sections provide examples of using AUTO CDC FROM SNAPSHOT to process snapshots into SCD Type 1 or Type 2 target tables. For background on when to use this API, see Change data capture and snapshots.

The examples in this section include Python and SQL implementations. For complete SQL syntax, see CREATE FLOW (pipelines).

Example: Process periodic snapshots

Use this approach when a source system exports a full snapshot on a regular schedule, such as one file every hour, and the snapshots arrive in order. In this example, each snapshot is a CSV file whose name includes the date and hour of the export, such as users_2026-10-15-09.csv.

The SQL implementation uses a WITH VERSION query to find the next snapshot file and uses the date and hour in the file name as the snapshot version. The pipeline tracks the last processed version.
The Python implementation uses a version function to find the next snapshot file and uses the date and hour in the file name as the snapshot version.

You can also read snapshots from other source types, such as a table in Python or SQL, or a JDBC connection in Python. See Initialize an empty target from one snapshot.

Step 1: Create sample snapshot files

Create a volume and write two hourly snapshot files to it. Run the following code from a notebook in the explorations folder of your pipeline:

Python
spark.sql("CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial")
spark.sql("CREATE VOLUME IF NOT EXISTS main.cdc_tutorial.hourly_snapshots")

dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-09.csv",
  "userId,city\n1,Oaxaca\n2,Monterrey\n3,Tijuana\n",
  overwrite=True
)

dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-10.csv",
  "userId,city\n2,Carmel\n3,Los Angeles\n4,Death Valley\n6,Kings Canyon\n",
  overwrite=True
)

Step 2: Run AUTO CDC FROM SNAPSHOT

You need a pipeline to run the code in this step. See Develop and debug ETL pipelines with the Lakeflow Pipelines Editor.

In a pipeline, create a new Python or SQL file in the transformations folder, add the following code, and then run the pipeline:

Python
SQL

Python
import re
from typing import Optional, Tuple

from pyspark import pipelines as dp
from pyspark.sql import DataFrame

snapshot_dir = "/Volumes/main/cdc_tutorial/hourly_snapshots/"
snapshot_pattern = re.compile(r"users_([0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2})[.]csv")

def get_next_snapshot(
  latest_snapshot_version: Optional[str]
) -> Optional[Tuple[DataFrame, str]]:
  snapshots = []

  for file in dbutils.fs.ls(snapshot_dir):
    match = snapshot_pattern.fullmatch(file.name)
    if match:
      version = match.group(1)
      if latest_snapshot_version is None or version > latest_snapshot_version:
        snapshots.append((version, file.path))

  if not snapshots:
    return None

  next_version, snapshot_path = min(snapshots)
  snapshot = (
    spark.read
      .format("csv")
      .option("header", True)
      .schema("userId INT, city STRING")
      .load(snapshot_path)
  )
  return (snapshot, next_version)

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = get_next_snapshot,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL
CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
-- Read the snapshot file selected by WITH VERSION.
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/hourly_snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'users_',
    (SELECT version FROM current_snapshot_version()),
    '.csv'
  )
)
-- Select the earliest snapshot file that is newer than the last processed version.
-- The version is the fixed-width date and hour string in the file name.
WITH VERSION (
  WITH snapshot_files AS (
    SELECT regexp_extract(
      path,
      'users_([0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2})[.]csv$',
      1
    ) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/hourly_snapshots/')
    WHERE path RLIKE 'users_[0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2}[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Run the pipeline. During the update, the flow processes the 09:00 snapshot followed by the 10:00 snapshot. AUTO CDC FROM SNAPSHOT compares the two snapshots and detects that user 1 was deleted, users 2 and 3 were updated, and users 4 and 6 were inserted. The snapshot versions parsed from the file names are stored in the __START_AT and __END_AT columns.

With SCD Type 2, the target table contains the following records:

userId	city	__START_AT	__END_AT
1	Oaxaca	2026-10-15-09	2026-10-15-10
2	Monterrey	2026-10-15-09	2026-10-15-10
2	Carmel	2026-10-15-10	null
3	Tijuana	2026-10-15-09	2026-10-15-10
3	Los Angeles	2026-10-15-10	null
4	Death Valley	2026-10-15-10	null
6	Kings Canyon	2026-10-15-10	null

userId	city	__START_AT	__END_AT
1	Oaxaca	2026-10-15-09	2026-10-15-10
2	Monterrey	2026-10-15-09	2026-10-15-10
2	Carmel	2026-10-15-10	null
3	Tijuana	2026-10-15-09	2026-10-15-10
3	Los Angeles	2026-10-15-10	null
4	Death Valley	2026-10-15-10	null
6	Kings Canyon	2026-10-15-10	null

User 1 was ended (deleted). Users 2 and 3 each have two versions showing their city changes. Users 4 and 6 were newly inserted.

note

To use SCD Type 1 instead of SCD Type 2, set stored_as_scd_type=1 in Python or STORED AS SCD TYPE 1 in SQL. The SCD Type 1 target table does not include __START_AT and __END_AT columns.

With SCD Type 1, the target table shows only the current state:

userId	city
2	Carmel
3	Los Angeles
4	Death Valley
6	Kings Canyon

Step 3: Simulate a new snapshot and rerun

Write another hourly snapshot file to simulate a new export arriving. Run this code from a notebook in the explorations folder of your pipeline:

Python
dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-11.csv",
  "userId,city\n2,Carmel\n3,San Diego\n4,Death Valley\n6,Kings Canyon\n7,Joshua Tree\n",
  overwrite=True
)

Run the pipeline again. The flow selects only the 11:00 snapshot because the earlier versions were already processed. User 3 is updated, user 7 is inserted, and the other records are unchanged.

If several hourly files arrive between pipeline updates, both implementations process each available file in version order during the next update.

Example: Initialize an empty target from one snapshot

Use this approach to initialize a new, empty target from one authoritative snapshot. Set once=True in Python or use the ONCE keyword in SQL to make this a one-time flow. After the flow commits successfully, subsequent incremental pipeline updates skip it. A full refresh clears the target and runs the flow again.

This example uses a table as the snapshot source and stores the current state with SCD Type 1. To add a snapshot to a target that also receives events from an AUTO CDC flow, see Add a backfill to an AUTO CDC SCD Type 1 table.

Step 1: Create the source snapshot

Create a table that contains the full current state of the source:

SQL
CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE OR REPLACE TABLE main.cdc_tutorial.initial_users (
  userId INT,
  city STRING
);

INSERT INTO main.cdc_tutorial.initial_users VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

Step 2: Create the one-time snapshot flow

In a pipeline, create a Python or SQL file in the transformations folder, add the following code, and run the pipeline. See Develop and debug ETL pipelines with the Lakeflow Pipelines Editor.

Python
SQL

Python
from pyspark import pipelines as dp

dp.create_streaming_table("users")

dp.create_auto_cdc_from_snapshot_flow(
  target = "users",
  source = "main.cdc_tutorial.initial_users",
  keys = ["userId"],
  stored_as_scd_type = 1,
  once = True
)

SQL
CREATE OR REFRESH STREAMING TABLE users;

CREATE FLOW initial_users_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.initial_users
)
KEYS (userId)
STORED AS SCD TYPE 1;

Use a JDBC source in Python

To initialize the target from a JDBC connection on classic compute, define a view that reads the source table and use the view name as the source argument:

Python
@dp.view(name="initial_users")
def initial_users():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Set source = "initial_users" and once = True when you create the flow.

After the update completes, the target contains the following rows:

userId	city
1	Oaxaca
2	Monterrey
3	Tijuana

Step 3: Verify the one-time flow behavior

Replace the source data with a different snapshot:

SQL
TRUNCATE TABLE main.cdc_tutorial.initial_users;

INSERT INTO main.cdc_tutorial.initial_users VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley');

Run another incremental pipeline update. The target remains unchanged because the one-time flow has already completed. If you run a full refresh, the pipeline clears the target and runs the flow again. The target then contains the replacement snapshot.

Limitations

The sequencing column must be a sortable data type. NULL sequencing values are not supported.
To stream data from the target of an AUTO CDC process, read from its change feed. For details, see Read a change data feed from an AUTO CDC target table.
Bitemporal targets have the following limitations:
- AUTO CDC FROM SNAPSHOT is not supported. To process snapshots, derive a change feed from them first. See Process snapshots with bitemporal AUTO CDC.
- Combined AUTO CDC FROM SNAPSHOT and AUTO CDC flows are not supported.
- Multiple AUTO CDC flows that write to the same bitemporal target are not supported.
- TRACK HISTORY ON (track_history_column_list and track_history_except_column_list in Python) is not supported. Bitemporal targets track history for every column.
- COLUMNS TO UPDATE and IGNORE NULL UPDATES are not supported.
- APPLY AS TRUNCATE WHEN is not supported.

Additional resources

Change data capture and snapshots: Learn about CDC concepts, snapshots, and SCD types.
Replicate an external RDBMS table using AUTO CDC: Learn how to perform initial hydration with a once flow and then continue processing changes.
Backfilling historical data with pipelines: Learn how to backfill append-only and AUTO CDC SCD Type 1 targets.
Bitemporal AUTO CDC: Learn how to track history across business time and system time.
Advanced AUTO CDC topics: Learn about change operations on AUTO CDC targets, reading change data feeds, and processing metrics.
Rewind and replay a pipeline: Learn how to restore an AUTO CDC target to an earlier point in time and reprocess changes.
Tutorial: Build an ETL pipeline using change data capture

Requirements​

How AUTO CDC works​

How AUTO CDC FROM SNAPSHOT works​

Combine AUTO CDC FROM SNAPSHOT and AUTO CDC​

Use multiple columns for sequencing​

AUTO CDC examples​

Create sample data​

Process SCD Type 1 updates​

Process SCD Type 2 updates​

Track a column subset with SCD Type 2​

Process bitemporal updates​

AUTO CDC FROM SNAPSHOT examples​

Example: Process periodic snapshots​

Step 1: Create sample snapshot files​

Step 2: Run AUTO CDC FROM SNAPSHOT​

Step 3: Simulate a new snapshot and rerun​

Example: Initialize an empty target from one snapshot​

Step 1: Create the source snapshot​

Step 2: Create the one-time snapshot flow​

Step 3: Verify the one-time flow behavior​

Limitations​

Additional resources​

Requirements

How AUTO CDC works

How AUTO CDC FROM SNAPSHOT works

Combine AUTO CDC FROM SNAPSHOT and AUTO CDC

Use multiple columns for sequencing

AUTO CDC examples

Create sample data

Process SCD Type 1 updates

Process SCD Type 2 updates

Track a column subset with SCD Type 2

Process bitemporal updates

AUTO CDC FROM SNAPSHOT examples

Example: Process periodic snapshots

Step 1: Create sample snapshot files

Step 2: Run AUTO CDC FROM SNAPSHOT

Step 3: Simulate a new snapshot and rerun

Example: Initialize an empty target from one snapshot

Step 1: Create the source snapshot

Step 2: Create the one-time snapshot flow

Step 3: Verify the one-time flow behavior

Limitations

Additional resources