Change Data Capture

Why Use CDC

CDC allows you to:

• Replicate your SingleStore databases to both homogeneous (SingleStore) and heterogeneous (for example, Kafka) systems.

• Maintain a copy of your SingleStore database or a subset of tables in open source formats to external storage options like Iceberg, Delta Lake, Apache Hudi, etc.

• Enable real-time data consumption by streaming changes to data processing pipelines, analytical systems, and incrementally updated dashboards.

• Support enterprise-grade high-availability node configuration through data replication.

• Filter data before moving/replicating to downstream systems.

Enable CDC

To enable CDC:

1. Upgrade to the latest SingleStore 8.5.x version.

2. Set the enable_observe_queries global variable to 1.

   Warning

   Once the enable_observe_queries variable is enabled, you can no longer rollback to a SingleStore version earlier than 8.5.x.

   SQL

   Copy

   SET GLOBAL enable_observe_queries = 1;

3. After upgrading SingleStore, create a snapshot of the relevant databases. This snapshot serves as the initial offset for OBSERVE queries, which allows any OBSERVE query to resume from this offset. Refer to SNAPSHOT DATABASE for information on creating a snapshot.

If enable_observe_queries is disabled, repeat the procedure to re-enable it.

Syntax

OBSERVE field_filter 
    FROM table_filter 
    [ AS format]
    [ BEGIN AT offset_config ]
    [ END AT TAIL ]


field_filter: [table.]field, ...
format: SQL
offset_config: [offset | NULL], ...  // number of partitions

Remarks

• The OBSERVE command only streams committed data. Rolled back (aborted) and failed transactions are filtered and excluded.

• To manually stop an OBSERVE query, use the KILL QUERY command.

• To prevent long-running OBSERVE queries from blocking DDL operations, use the observe_leaf_timeout_secs engine variable. This engine variable specifies the maximum time an OBSERVE query can remain idle on leaf nodes before closing the connection. This timeout is only applicable to leaf partitions.

Offset

An offset is an opaque identifier that represents a position in a database snapshot or a database write-ahead log (WAL). Within a partition, each offset represents a unique position which is always strictly increasing. Offset values are immutable and do not change between queries.

The OBSERVE query can resume from either the last observed offset or a specific offset. The offset in the BEGIN AT clause indicates that all the changes before the specified offset have been observed.

The offset specified in the END AT clause indicates that the observation must stop once this offset is reached. The END AT TAIL clause continues observation until the offset reaches the end of database WAL. If the END AT clause is not specified, the OBSERVE query runs indefinitely.

Consider the following query:

OBSERVE * FROM test 
BEGIN AT ('0000000000000088000000000000000C000000000000C087', 
NULL,
'0000000000000078000000000000000C000000000000C087', 
NULL);

This query starts observing from the test table at the following logical position for each partition:

• Partition 0: 0000000000000088000000000000000C000000000000C087

• Partition 1: Latest snapshot

• Partition 2: 0000000000000078000000000000000C000000000000C087

• Partition 3: Latest snapshot

Specifying NULL for a partition in the BEGIN AT clause indicates that observation starts from the latest snapshot for the partition. Omitting the BEGIN AT clause is equivalent to specifying NULL for each partition.

For each database partition, the OBSERVE query scans each offset in the transaction log. The result set is then multiplexed and relayed to the client.

The OBSERVE query automatically transitions from a snapshot to the next logical position in the database WAL, eliminating the need for an out-of-band snapshot scan or stopping write queries on the database.

Use the snapshots_to_keep and snapshot_trigger_size engine variables to control the number and size of snapshots. Refer to List of Engine Variables for more information.

Note: Snapshots are only observed at the start of an OBSERVE query.

Transactions

All changes are contained within a set of transaction boundary records, BeginTransaction and CommitTransaction. These events (or changes) specify the beginning and ending positions for all the changes applied within a transaction. There are no nested transactions. Each transaction is identified by a TxID, which is unique within a database. It can be shared between partitions in the instance of a two-phase commit (2PC). All 2PC transactions have a TxPartitions value greater than 1, which implies that this transaction writes data to at most TxPartitions number of partitions. All individual change records within a transaction share the same TxID.

The OBSERVE query only retrieves committed transactions. Rolled back (aborted) or uncommitted transactions are not observed.

Snapshots also contain boundary records, namely BeginSnapshot and CommitSnapshot, which indicate that the data is sourced from a snapshot. Snapshots can have transactions within them. BeginSnapshot and CommitSnapshot records are always returned even if they do not contain any records.

Output Format

Replicate Tables

Limitations

• DDL changes are not supported. The OBSERVE query stops once a DDL event is observed on all the partitions, effectively aligning all partition streams within the database.

• OBSERVE query plans are not removed by default when a table is updated or a database is dropped. You must manually drop the OBSERVE query plans:

  SQL

  Copy

  SELECT PLAN_ID FROM information_schema.mv_plancache
  WHERE QUERY_TEXT LIKE 'OBSERVE %'; 
  
  DROP <plan_id> FROM PLANCACHE;

• The earliest possible offset for a database is its restore point.

• The TxPartitions specifies the number of partitions affected by the current transaction. It represents an upper bound on the number of partitions that return records for a transaction. The OBSERVE query may not return an empty transaction even if the distributed transaction did not write to any of the partitions.

• For Delete operations on columnstore tables, only the internal ID (InternalId) is populated in the change record.

• The OBSERVE command does not support the JOIN clause.

Example

CREATE DATABASE dbTest;
USE dbTest;

CREATE TABLE tblEx(
    ID INT NOT NULL PRIMARY KEY,
    Code VARCHAR(4));

INSERT INTO tblEx(ID, Code) VALUES (1, "KorE");
INSERT INTO tblEx(ID, Code) VALUES (2, "PamY");
INSERT INTO tblEx(ID, Code) VALUES (3, "TabK");
UPDATE tblEx SET Code = "JonA" WHERE ID = 2;

OBSERVE * FROM tblEx;

+----------------------------------------------------+-------------+--------------------+--------+--------------------------------------------------------------------------------------+--------------+----------------------+------+------+
| Offset                                             | PartitionId | Type               | Table  | TxId                                                                                 | TxPartitions | InternalId           | ID   | Code |
+----------------------------------------------------+-------------+--------------------+--------+--------------------------------------------------------------------------------------+--------------+----------------------+------+------+
| 0x0000000000000077000000000000000E000000000000E06E |           4 | BeginTransaction   |        | 0x0100000000000000040000000000000077000000000000000E000000000000E06E0000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000077000000000000000E000000000000E087 |           4 | Insert             | tblEx  | 0x0100000000000000040000000000000077000000000000000E000000000000E06E0000000000000000 |            1 |  1152921504606846977 |    1 | KorE |
| 0x0000000000000077000000000000000E000000000000E088 |           4 | CommitTransaction  |        | 0x0100000000000000040000000000000077000000000000000E000000000000E06E0000000000000000 |            1 |                    0 |.NULL | NULL |
| 0x0000000000000077000000000000000F000000000000F039 |           4 | BeginTransaction   |        | 0x0100000000000000040000000000000077000000000000000F000000000000F0390000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000077000000000000000F000000000000F052 |           4 | Insert             | tblEx  | 0x0100000000000000040000000000000077000000000000000F000000000000F0390000000000000000 |            1 |  1152921504606846978 |    3 | TabK |
| 0x0000000000000077000000000000000F000000000000F053 |           4 | CommitTransaction  |        | 0x0100000000000000040000000000000077000000000000000F000000000000F0390000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000078000000000000000C000000000000C06E |           3 | BeginTransaction   |        | 0x0100000000000000030000000000000078000000000000000C000000000000C06E0000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000078000000000000000C000000000000C087 |           3 | Insert             | tblEx  | 0x0100000000000000030000000000000078000000000000000C000000000000C06E0000000000000000 |            1 |  1152921504606846977 |    2 | PamY |
| 0x0000000000000078000000000000000C000000000000C088 |           3 | CommitTransaction  |        | 0x0100000000000000030000000000000078000000000000000C000000000000C06E0000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000078000000000000000E000000000000E039 |           3 | BeginTransaction   |        | 0x0100000000000000030000000000000078000000000000000E000000000000E0390000000000000000 |            1 |                    0 | NULL | NULL |
| 0x0000000000000078000000000000000E000000000000E052 |           3 | Update             | tblEx  | 0x0100000000000000030000000000000078000000000000000E000000000000E0390000000000000000 |            1 |  1152921504606846977 |    2 | JonA |
| 0x0000000000000078000000000000000E000000000000E053 |           3 | CommitTransaction  |        | 0x0100000000000000030000000000000078000000000000000E000000000000E0390000000000000000 |            1 |                    0 | NULL | NULL |
+----------------------------------------------------+-------------+--------------------+--------+--------------------------------------------------------------------------------------+--------------+----------------------+------+------+

OBSERVE * FROM test 
    AS SQL
    BEGIN AT 
        ('0000000000000088000000000000000E000000000000E06E', 
        NULL,
        '0000000000000088000000000000000E000000000000E053', 
        NULL)
    END AT TAIL;

References

• SingleStore Connector for Debezium

• OBSERVE