This page describes change streams in Spanner for GoogleSQL-dialect databases andPostgreSQL-dialect databases, including:

• The split-based partitioning model
• The format and content of change stream records
• The low-level syntax used to query those records
• An example of the query workflow

You use the Spanner API toquery change streams directly.Applications that insteaduse Dataflow to read change streamdata don't need to work directly with the data model describedhere.

For a broader introductory guide to change streams, seeChange streamsoverview.

Change stream partitions

When a change occurs on a table that is watched by a change stream,Spanner writes a corresponding change stream record in thedatabase, synchronously in the same transaction as the data change. Thismeans that if the transaction succeeds, Spanner has alsosuccessfully captured and persisted the change. Internally,Spanner co-locates the change stream record and the data changeso that they are processed by the same server to minimize write overhead.

As part of the DML to a particular split, Spannerappends the write to the corresponding change stream datasplit in the same transaction. Because of this colocation, changestreams don't add extra coordination across serving resources, whichminimizes the transaction commit overhead.

Spanner scales by dynamically splitting and merging data basedon database load and size, and distributing splits across serving resources.

To enable change streams writes and reads to scale, Spannersplits and merges the internal change stream storage along with the databasedata, automatically avoidinghotspots. To supportreading change stream records in near real-time as database writes scale, theSpanner API is designed for a change stream to be queriedconcurrently using change stream partitions. Change stream partitions map tochange stream data splits that contain the change stream records. A changestream's partitions change dynamically over time and are correlated to howSpanner dynamically splits and merges the database data.

A change stream partition contains records for an immutable key range for aspecific time range. Any change stream partition can split into one or morechange stream partitions, or be merged with other change stream partitions. Whenthese split or merge events happen, child partitions are created to capture thechanges for their respective immutable key ranges for the next time range. Inaddition to data change records, a change stream query returns child partitionrecords to notify readers of new change stream partitions that need to bequeried, as well as heartbeat records to indicate forward progress when nowrites have occurred recently.

When querying a particular change stream partition, the change records arereturned in commit timestamp order. Each change record is returned exactlyonce. Across change stream partitions, change record ordering is not guaranteed.Change records for a particular primary key are returned only on onepartition for a particular time range.

Due to the parent-child partition lineage, in order to process changes for aparticular key in commit timestamp order, records returned from childpartitions should be processed only after records from all parentpartitions have been processed.

Change stream read functions and query syntax

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp,end_timestamp,partition_token,heartbeat_milliseconds,read_options)

privatestaticfinalStringSINGERS_NAME_STREAM_QUERY_TEMPLATE="SELECT ChangeRecord FROM READ_SingersNameStream"+"("+"   start_timestamp => @startTimestamp,"+"   end_timestamp => @endTimestamp,"+"   partition_token => @partitionToken,"+"   heartbeat_milliseconds => @heartbeatMillis"+")";// Helper method to conveniently create change stream query texts and// bind parameters.publicstaticStatementgetChangeStreamQuery(StringpartitionToken,TimestampstartTimestamp,TimestampendTimestamp,longheartbeatMillis){returnStatement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE).bind("startTimestamp").to(startTimestamp).bind("endTimestamp").to(endTimestamp).bind("partitionToken").to(partitionToken).bind("heartbeatMillis").to(heartbeatMillis).build();}

SELECT*FROM"spanner"."read_json_SingersNameStream"(start_timestamp,end_timestamp,partition_token,heartbeat_milliseconds,null)

privatestaticfinalStringSINGERS_NAME_STREAM_QUERY_TEMPLATE="SELECT * FROM \"spanner\".\"read_json_SingersNameStream\""+"($1, $2, $3, $4, null)";// Helper method to conveniently create change stream query texts and// bind parameters.publicstaticStatementgetChangeStreamQuery(StringpartitionToken,TimestampstartTimestamp,TimestampendTimestamp,longheartbeatMillis){returnStatement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE).bind("p1").to(startTimestamp).bind("p2").to(endTimestamp).bind("p3").to(partitionToken).bind("p4").to(heartbeatMillis).build();}

Change streams record format

STRUCT <  data_change_record ARRAY<STRUCT<...>>,  heartbeat_record ARRAY<STRUCT<...>>,  child_partitions_record ARRAY<STRUCT<...>>>

{"data_change_record":{},"heartbeat_record":{},"child_partitions_record":{}}

Data change records

A data change record contains a set of changes to a table with thesame modification type (insert, update, or delete) committed at the samecommit timestamp in one change stream partition for the sametransaction. Multiple data change records can be returned for the sametransaction across multiple change stream partitions.

All data change records havecommit_timestamp,server_transaction_id,andrecord_sequence fields, which together determine the order in the changestream for a stream record. These three fields are sufficient to derivethe ordering of changes and provide external consistency.

Note that multiple transactions can have the same commit timestamp ifthey touch non-overlapping data. Theserver_transaction_id fieldoffers the ability to distinguish which set of changes (potentiallyacross change stream partitions) were issued within the sametransaction. Pairing it with therecord_sequence andnumber_of_records_in_transaction fields allows you buffer and orderall the records from a particular transaction, as well.

The fields of a data change record include the following:

[  {      "name": "STRING",      "type": {        "code": "STRING"      },      "is_primary_key":BOOLEAN      "ordinal_position":NUMBER    },    ...]

[  {    "keys": {"STRING" : "STRING"},    "new_values": {      "STRING" : "VALUE-TYPE",      [...]    },    "old_values": {      "STRING" : "VALUE-TYPE",      [...]    },  },  [...]]

[  {      "name": "STRING",      "type": {        "code": "STRING"      },      "is_primary_key":BOOLEAN      "ordinal_position":NUMBER    },    ...]

[  {    "keys": {"STRING" : "STRING"},    "new_values": {      "STRING" : "VALUE-TYPE",      [...]    },    "old_values": {      "STRING" : "VALUE-TYPE",      [...]    },  },  [...]]

Example data change record

A pair of example data change records follow. They describe a single transactionwhere there is a transfer between two accounts. The two accounts are in separatechange stream partitions.

"data_change_record":{"commit_timestamp":"2022-09-27T12:30:00.123456Z",// record_sequence is unique and monotonically increasing within a// transaction, across all partitions."record_sequence":"00000000","server_transaction_id":"6329047911","is_last_record_in_transaction_in_partition":true,"table_name":"AccountBalance","column_types":[{"name":"AccountId","type":{"code":"STRING"},"is_primary_key":true,"ordinal_position":1},{"name":"LastUpdate","type":{"code":"TIMESTAMP"},"is_primary_key":false,"ordinal_position":2},{"name":"Balance","type":{"code":"INT"},"is_primary_key":false,"ordinal_position":3}],"mods":[{"keys":{"AccountId":"Id1"},"new_values":{"LastUpdate":"2022-09-27T12:30:00.123456Z","Balance":1000},"old_values":{"LastUpdate":"2022-09-26T11:28:00.189413Z","Balance":1500},}],"mod_type":"UPDATE",// options are INSERT, UPDATE, DELETE"value_capture_type":"OLD_AND_NEW_VALUES","number_of_records_in_transaction":2,"number_of_partitions_in_transaction":2,"transaction_tag":"app=banking,env=prod,action=update","is_system_transaction":false,}

"data_change_record":{"commit_timestamp":"2022-09-27T12:30:00.123456Z","record_sequence":"00000001","server_transaction_id":"6329047911","is_last_record_in_transaction_in_partition":true,"table_name":"AccountBalance","column_types":[{"name":"AccountId","type":{"code":"STRING"},"is_primary_key":true,"ordinal_position":1},{"name":"LastUpdate","type":{"code":"TIMESTAMP"},"is_primary_key":false,"ordinal_position":2},{"name":"Balance","type":{"code":"INT"},"is_primary_key":false,"ordinal_position":3}],"mods":[{"keys":{"AccountId":"Id2"},"new_values":{"LastUpdate":"2022-09-27T12:30:00.123456Z","Balance":2000},"old_values":{"LastUpdate":"2022-01-20T11:25:00.199915Z","Balance":1500},},...],"mod_type":"UPDATE",// options are INSERT, UPDATE, DELETE"value_capture_type":"OLD_AND_NEW_VALUES","number_of_records_in_transaction":2,"number_of_partitions_in_transaction":2,"transaction_tag":"app=banking,env=prod,action=update","is_system_transaction":false,}

The following data change record is an example of a record with the valuecapture typeNEW_VALUES. Note that only new values are populated.Only theLastUpdate column was modified, so only that columnwas returned.

"data_change_record":{"commit_timestamp":"2022-09-27T12:30:00.123456Z",// record_sequence is unique and monotonically increasing within a// transaction, across all partitions."record_sequence":"00000000","server_transaction_id":"6329047911","is_last_record_in_transaction_in_partition":true,"table_name":"AccountBalance","column_types":[{"name":"AccountId","type":{"code":"STRING"},"is_primary_key":true,"ordinal_position":1},{"name":"LastUpdate","type":{"code":"TIMESTAMP"},"is_primary_key":false,"ordinal_position":2}],"mods":[{"keys":{"AccountId":"Id1"},"new_values":{"LastUpdate":"2022-09-27T12:30:00.123456Z"},"old_values":{}}],"mod_type":"UPDATE",// options are INSERT, UPDATE, DELETE"value_capture_type":"NEW_VALUES","number_of_records_in_transaction":1,"number_of_partitions_in_transaction":1,"transaction_tag":"app=banking,env=prod,action=update","is_system_transaction":false}

The following data change record is an example of a record with the valuecapture typeNEW_ROW. Only theLastUpdatecolumn was modified, but all tracked columns are returned.

"data_change_record":{"commit_timestamp":"2022-09-27T12:30:00.123456Z",// record_sequence is unique and monotonically increasing within a// transaction, across all partitions."record_sequence":"00000000","server_transaction_id":"6329047911","is_last_record_in_transaction_in_partition":true,"table_name":"AccountBalance","column_types":[{"name":"AccountId","type":{"code":"STRING"},"is_primary_key":true,"ordinal_position":1},{"name":"LastUpdate","type":{"code":"TIMESTAMP"},"is_primary_key":false,"ordinal_position":2},{"name":"Balance","type":{"code":"INT"},"is_primary_key":false,"ordinal_position":3}],"mods":[{"keys":{"AccountId":"Id1"},"new_values":{"LastUpdate":"2022-09-27T12:30:00.123456Z","Balance":1000},"old_values":{}}],"mod_type":"UPDATE",// options are INSERT, UPDATE, DELETE"value_capture_type":"NEW_ROW","number_of_records_in_transaction":1,"number_of_partitions_in_transaction":1,"transaction_tag":"app=banking,env=prod,action=update","is_system_transaction":false}

The following data change record is an example of a record with the valuecapture typeNEW_ROW_AND_OLD_VALUES. Only theLastUpdate column wasmodified, but all tracked columns are returned. This value capture type capturesthe new value and old value ofLastUpdate.

"data_change_record":{"commit_timestamp":"2022-09-27T12:30:00.123456Z",// record_sequence is unique and monotonically increasing within a// transaction, across all partitions."record_sequence":"00000000","server_transaction_id":"6329047911","is_last_record_in_transaction_in_partition":true,"table_name":"AccountBalance","column_types":[{"name":"AccountId","type":{"code":"STRING"},"is_primary_key":true,"ordinal_position":1},{"name":"LastUpdate","type":{"code":"TIMESTAMP"},"is_primary_key":false,"ordinal_position":2},{"name":"Balance","type":{"code":"INT"},"is_primary_key":false,"ordinal_position":3}],"mods":[{"keys":{"AccountId":"Id1"},"new_values":{"LastUpdate":"2022-09-27T12:30:00.123456Z","Balance":1000},"old_values":{"LastUpdate":"2022-09-26T11:28:00.189413Z"}}],"mod_type":"UPDATE",// options are INSERT, UPDATE, DELETE"value_capture_type":"NEW_ROW_AND_OLD_VALUES","number_of_records_in_transaction":1,"number_of_partitions_in_transaction":1,"transaction_tag":"app=banking,env=prod,action=update","is_system_transaction":false}

Heartbeat records

When a heartbeat record is returned, it indicates that all changes withcommit_timestamp less than or equal to the heartbeat record'stimestamp have been returned, and future data records in thispartition must have higher commit timestamps than that returned by theheartbeat record. Heartbeat records are returned when there are no datachanges written to a partition. When there are data changes written tothe partition,data_change_record.commit_timestamp can be used insteadofheartbeat_record.timestamp to tell that the reader is making forwardprogress in reading the partition.

You can use heartbeat records returned on partitions to synchronizereaders across all partitions. Once all readers have received either aheartbeat greater than or equal to some timestampA or have received data orchild partition records greater than or equal to timestampA, the readers knowthey have received all records committed at or before that timestampA and canstart processing the buffered records—for example, sorting the cross-partitionrecords by timestamp and grouping them byserver_transaction_id.

A heartbeat record contains only one field:

Example heartbeat record

An example heartbeat record, communicating that all records with timestampsless or equal than this record's timestamp have been returned:

heartbeat_record:{"timestamp":"2022-09-27T12:35:00.312486Z"}

Child partition records

Child partition records returns information about child partitions: theirpartition tokens, the tokens of their parent partitions, and thestart_timestamp that represents the earliest timestamp that the childpartitions contain change records for. Records whose commit timestampsare immediately prior to thechild_partitions_record.start_timestamp arereturned in the current partition. After returning all thechild partition records for this partition, this query returns witha success status, indicating all records have been returned for thispartition.

The fields of a child partition record includes the following:

[  {    "token" : "STRING",    "parent_partition_tokens" : ["STRING"]  }]

[  {    "token": "STRING",    "parent_partition_tokens": ["STRING"],  }, [...]]

Example child partition record

The following is an example of a child partition record:

child_partitions_record:{"start_timestamp":"2022-09-27T12:40:00.562986Z","record_sequence":"00000001","child_partitions":[{"token":"child_token_1",// To make sure changes for a key is processed in timestamp// order, wait until the records returned from all parents// have been processed."parent_partition_tokens":["parent_token_1","parent_token_2"]}],}

Change streams query workflow

Run change stream queries using theExecuteStreamingSql API, with a single-useread-onlytransaction and astrongtimestamp bound. The changestream read function lets you specify thestart_timestamp andend_timestamp for the time range of interest. All change recordswithin the retention period are accessible using the strong read-onlytimestamp bound.

All otherTransactionOptionsare invalid for change stream queries. In addition,ifTransactionOptions.read_only.return_read_timestamp is set totrue,a special value ofkint64max - 1 is returned in theTransactionmessage that describes the transaction, instead of a valid readtimestamp. This special value should be discarded and not used for anysubsequent queries.

Each change stream query can return any number of rows, each containingeither a data change record, heartbeat record, or child partitionsrecord. There is no need to set a deadline for the request.

Example change stream query workflow

The streaming query workflow begins with issuing the very first change streamquery by specifying thepartition_token toNULL. The query needs to specifythe read function for the change stream, start and end timestamp of interest,and the heartbeat interval. When theend_timestamp isNULL, the query keepsreturning data changes until the partition ends.

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp=>"2022-05-01T09:00:00Z",end_timestamp=>NULL,partition_token=>NULL,heartbeat_milliseconds=>10000);

SELECT*FROM"spanner"."read_json_SingersNameStream"('2022-05-01T09:00:00Z',NULL,NULL,10000,NULL);

Process data records from this query until all child partition records arereturned. In the following example, two child partition records and threepartition tokens are returned, then the query terminates. Child partitionrecords from a specific query always shares the samestart_timestamp.

child_partitions_record:{"record_type":"child_partitions","start_timestamp":"2022-05-01T09:00:01Z","record_sequence":"1000012389","child_partitions":[{"token":"child_token_1",// Note parent tokens are null for child partitions returned// from the initial change stream queries."parent_partition_tokens":[NULL]}{"token":"child_token_2","parent_partition_tokens":[NULL]}],}

child_partitions_record:{"record_type":"child_partitions","start_timestamp":"2022-05-01T09:00:01Z","record_sequence":"1000012390","child_partitions":[{"token":"child_token_3","parent_partition_tokens":[NULL]}],}

To process changes after2022-05-01T09:00:01Z, create three new queries andrun them in parallel. Used together, the three queries return data changes forthe same key range their parent covers. Always set thestart_timestamp to thestart_timestamp in the same child partition record and use the sameend_timestamp and heartbeat interval to process the records consistentlyacross all queries.

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp=>"2022-05-01T09:00:01Z",end_timestamp=>NULL,partition_token=>"child_token_1",heartbeat_milliseconds=>10000);

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp=>"2022-05-01T09:00:01Z",end_timestamp=>NULL,partition_token=>"child_token_2",heartbeat_milliseconds=>10000);

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp=>"2022-05-01T09:00:01Z",end_timestamp=>NULL,partition_token=>"child_token_3",heartbeat_milliseconds=>10000);

SELECT*FROM"spanner"."read_json_SingersNameStream"('2022-05-01T09:00:01Z',NULL,'child_token_1',10000,NULL);

SELECT*FROM"spanner"."read_json_SingersNameStream"('2022-05-01T09:00:01Z',NULL,'child_token_2',10000,NULL);

SELECT*FROM"spanner"."read_json_SingersNameStream"('2022-05-01T09:00:01Z',NULL,'child_token_3',10000,NULL);

The query onchild_token_2 finishes after returning another child partitionrecord. This record indicates that a new partition is covering changes for bothchild_token_2 andchild_token_3 starting at2022-05-01T09:30:15Z. Theexact same record is returned by the query onchild_token_3, because both arethe parent partitions of the newchild_token_4. To ensure a strict orderedprocessing of data records for a particular key, the query onchild_token_4must start after all the parents have finished. In this case, the parents arechild_token_2 andchild_token_3. Only create one query for each childpartition token. The query workflow design should appoint one parent to wait andschedule the query onchild_token_4.

child_partitions_record:{"record_type":"child_partitions","start_timestamp":"2022-05-01T09:30:15Z","record_sequence":"1000012389","child_partitions":[{"token":"child_token_4","parent_partition_tokens":["child_token_2","child_token_3"],}],}

SELECTChangeRecordFROMREAD_SingersNameStream(start_timestamp=>"2022-05-01T09:30:15Z",end_timestamp=>NULL,partition_token=>"child_token_4",heartbeat_milliseconds=>10000);

SELECT*FROM"spanner"."read_json_SingersNameStream"('2022-05-01T09:30:15Z',NULL,'child_token_4',10000,NULL);

Find examples of handling and parsing change stream records in the Apache BeamSpannerIO Dataflow connector onGitHub.