Change data capture in BigQuery

BigQuery CDC applies streamed UPSERT and DELETE rows to a table by primary key, in the background. max_staleness trades freshness for cost, keys aren't enforced, and CDC tables reject UPDATE, DELETE and MERGE. APPENDS and CHANGES read a table's own change history.

Change data capture (CDC) keeps a BigQuery table in step with a source database: every insert, update and delete in the source is sent to BigQuery, which applies it to the copy. BigQuery supports this directly. You stream rows marked UPSERT or DELETE through the Storage Write API, and BigQuery merges them into the table by primary key, without you running a MERGE. In the other direction, the APPENDS and CHANGES table functions return what changed in a BigQuery table during a time range.

Streaming changes into a table

The destination table needs a primary key, which tells BigQuery which row a change applies to:

CREATE TABLE mydataset.customers (
  customer_id INT64,
  name STRING,
  credit_limit INT64,
  PRIMARY KEY (customer_id) NOT ENFORCED
)
OPTIONS (max_staleness = INTERVAL 15 MINUTE);

Each streamed row carries the pseudocolumn _CHANGE_TYPE, set to UPSERT (insert the row, or replace the row with the same key) or DELETE (remove the row with that key). Streaming these rows:

removes customer 100, replaces customer 101 and adds 105. The requirements:

• The Storage Write API’s gRPC interface, default stream, protocol buffer format. See Streaming.
• A primary key of up to 16 columns, declared NOT ENFORCED. BigQuery doesn’t check it, so the key must really be unique in the source.
• _CHANGE_TYPE values UPSERT and DELETE only. Mixing them with plain inserts on the same connection fails.

Google’s Datastream service can replicate transactional databases into BigQuery this way, without you writing the streaming code.

Order of changes

By default, when two changes arrive for the same key, the one BigQuery received last wins. If your source can deliver changes out of order, send _CHANGE_SEQUENCE_NUMBER with each row: a string of hexadecimal sections separated by /, for example a log position. BigQuery then keeps the change with the larger sequence number, whatever the arrival order.

Freshness, max_staleness and cost

Streamed changes aren’t merged into the table immediately. BigQuery applies them with background jobs. What a query sees depends on the table’s max_staleness option:

• Without max_staleness, every query merges all changes not yet applied at query time, so results are always current but each query is slower and costs more.
• With max_staleness = INTERVAL 15 MINUTE, a query returns the table as of the last background apply, as long as that was less than 15 minutes ago. If the background work has fallen further behind, the query merges the pending changes itself (a runtime merge) and returns current data at the higher cost.

Google suggests setting max_staleness to the larger of the staleness you can accept and twice the time the background apply jobs take, plus a buffer. A runtime merge on a partitioned table scans the whole table, whatever partitions the query asks for, so a max_staleness that is too low makes queries both slower and more expensive.

The background apply jobs use BigQuery compute. With on-demand pricing they run on internal resources and are billed as on-demand compute. With capacity pricing you can assign them to a BACKGROUND reservation; if it is too small, changes, deletions included, are applied late.

What a CDC table can’t do

While CDC is active on a table:

• UPDATE, DELETE and MERGE statements on it aren’t allowed. All changes come through the stream.
• It can’t be queried through a wildcard table and can’t have a search index.
• It can have at most 2,000 top-level columns.
• If queries are doing runtime merges, table copies, clones, snapshots, the Storage Read API and require_partition_filter stop working on it.
• An extract job exports only the changes already applied. Use EXPORT DATA to export the current state.

Deletes are kept for two days to handle late, out-of-order changes, and that data is billed as storage meanwhile.

Reading a table’s changes: APPENDS and CHANGES

The other half of CDC is getting changes out of a BigQuery table, for example to update a copy elsewhere without exporting the whole table each time:

-- Rows added to the table in the last hour
SELECT *
FROM APPENDS(TABLE mydataset.orders, TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR), NULL);

-- Every row inserted, updated or deleted in the last hour
SELECT *
FROM CHANGES(TABLE mydataset.orders, TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR), NULL);

• APPENDS returns rows added by inserts, loads, streaming and the insert part of MERGE. It works on any table, within the time travel window.
• CHANGES also returns updated and deleted rows, and changes from TRUNCATE, overwrite loads and dropped partitions. It needs the table option enable_change_history = TRUE set before the changes happen, and that option stores extra change metadata, which is billed.
• Both add pseudocolumns describing the change, such as _CHANGE_TYPE and _CHANGE_TIMESTAMP.
• Both are billed for processing all data written to the table in the time range, not only the rows you select. Keep the ranges short and run them often.
• Users need bigquery.tables.getData on the table. If the table has, or ever had, row-level access policies, only a table admin can read its change history.