变更数据流分区、记录和查询

本页详细介绍了变更数据流的以下属性：

• 它的基于分屏的分区模型
• 变更数据流记录的格式和内容
• 用于查询这些记录的低层级语法
• 查询工作流示例

本页面上的信息最适合使用 Spanner API 直接查询变更数据流。使用 Dataflow 读取变更数据流数据的应用无需直接使用此处所述的数据模型。

如需获得关于变更数据流的更宽泛的入门指南，请参阅变更数据流概览。

更改数据流分区

当变更流所监控的表发生更改时，Spanner 会在数据更改所在的事务中同步写入相应的变更数据流记录。这可以保证如果事务成功，Spanner 也会成功捕获并保留更改。在内部，Spanner 会将变更数据流记录和数据更改存储在一起，以便由同一台服务器处理它们，从而最大限度地减少写入开销。

作为特定分块的 DML 的一部分，Spanner 会将写入附加到同一事务中的相应变更数据流数据分块。由于这种主机托管，变更流不会在服务资源之间增加额外的协调，从而最大限度地减少事务提交开销。

Spanner 可根据数据库负载和大小动态拆分和合并数据，并在服务资源之间分配分片，从而实现扩缩。

为了使变更数据流写入和读取能够实现规模化，Spanner 会对内部变更数据流存储与数据库数据进行拆分和合并，自动避免热点。为了支持随着数据库写入规模的实时读取变更数据流记录，Spanner API 支持使用变更数据流分区并发查询变更数据流。变更数据流分区会映射到包含变更数据流记录的数据流数据分块。变更数据流的分区会随时间动态变化，并且与 Spanner 动态拆分和合并数据库数据的方式相关。

变更数据流分区包含特定时间范围内不可变键范围的记录。任何变更数据流分区都可以拆分为一个或多个变更数据流分区，或者与其他变更数据流分区合并。当发生这些拆分或合并事件时，系统会创建子分区，以捕获下一个时间范围内各自不可变键范围的更改。除了数据变更记录之外，变更数据流查询还会返回子分区记录以通知读取器需要查询的新变更数据流分区，以及检测信号记录以指示在最近未发生写入操作时前进的进度。

查询特定的更改流分区时，系统会按提交时间戳顺序返回更改记录。每条变更记录仅返回一次。在各个变更数据流分区之间，变更记录的顺序没有保证的顺序。在特定时间范围内，系统仅在一个分区上返回特定主键的变更记录。

由于父子分区沿袭，为了按提交时间戳顺序处理特定键的更改，应仅在处理完所有父分区中的记录后处理从子分区返回的记录。

更改数据流读取函数和查询语法

GoogleSQL

您可以使用 ExecuteStreamingSql API 查询变更数据流。Spanner 会自动创建一个特殊的读取函数以及变更数据流。读取函数提供对变更流的记录的访问权限。读取函数的命名惯例为 READ_change_stream_name。

假设数据库中存在变更数据流 SingersNameStream，GoogleSQL 的查询语法如下：

SELECT ChangeRecord
FROM READ_SingersNameStream (
    start_timestamp,
    end_timestamp,
    partition_token,
    heartbeat_milliseconds,
    read_options
)

read 函数接受以下参数：

参数名称	类型	是否必需？	说明
start_timestamp	TIMESTAMP	需要	指定应返回 commit_timestamp 大于或等于 start_timestamp 的记录。该值必须在变更数据流保留期限内，且应小于或等于当前时间，且大于或等于变更数据流的创建时间戳。
end_timestamp	TIMESTAMP	可选（默认值：NULL）	指定应返回 commit_timestamp 小于或等于 end_timestamp 的记录。该值必须在变更数据流保留期限内，且大于或等于 start_timestamp。查询会在返回直到 end_timestamp 的所有 ChangeRecord 之后完成，或者用户终止连接。如果为 NULL 或不指定，则查询会一直执行，直到返回所有 ChangeRecords 或用户终止连接。
partition_token	STRING	可选（默认值：NULL）	根据子分区记录的内容，指定要查询的变更数据流分区。如果未指定 NULL，则表示读取器是首次查询变更数据流，并且尚未获取任何要查询的特定分区令牌。
heartbeat_milliseconds	INT64	需要	确定在此分区中未提交事务的情况下返回检测信号 ChangeRecord 的频率。 该值必须介于 1,000（1 秒）到 30,0000（5 分钟）之间。
read_options	ARRAY	可选（默认值：NULL）	预留供将来使用的其他读取选项。目前，唯一允许使用的值为 NULL。

我们建议提供了一种便利方法来构建读取函数查询的文本并将参数绑定到该文本，如以下示例所示。

Java

private static final String SINGERS_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.
public static Statement getChangeStreamQuery(
      String partitionToken,
      Timestamp startTimestamp,
      Timestamp endTimestamp,
      long heartbeatMillis) {
  return Statement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE)
                    .bind("startTimestamp")
                    .to(startTimestamp)
                    .bind("endTimestamp")
                    .to(endTimestamp)
                    .bind("partitionToken")
                    .to(partitionToken)
                    .bind("heartbeatMillis")
                    .to(heartbeatMillis)
                    .build();
}

PostgreSQL

您可以使用 ExecuteStreamingSql API 查询变更数据流。Spanner 会自动创建一个特殊的读取函数以及变更数据流。读取函数提供对变更流的记录的访问权限。读取函数的命名惯例为 spanner.read_json_change_stream_name。

假设数据库中存在变更数据流 SingersNameStream，PostgreSQL 的查询语法如下：

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

read 函数接受以下参数：

参数名称	类型	是否必需？	说明
start_timestamp	timestamp with time zone	需要	指定应返回 commit_timestamp 大于或等于 start_timestamp 的变更记录。该值必须在变更数据流保留期限内，且应小于或等于当前时间，且大于或等于变更数据流的创建时间戳。
end_timestamp	timestamp with timezone	可选（默认值：NULL）	指定应返回 commit_timestamp 小于或等于 end_timestamp 的变更记录。该值必须在变更数据流保留期限内，且大于或等于 start_timestamp。返回截至 end_timestamp 的所有变更记录后，查询完成，或者用户终止连接。如果为 NULL，则查询会一直执行，直到返回所有更改历史记录或用户终止连接。
partition_token	text	可选（默认值：NULL）	根据子分区记录的内容，指定要查询的变更数据流分区。如果未指定 NULL，则表示读取器是首次查询变更数据流，并且尚未获取任何要查询的特定分区令牌。
heartbeat_milliseconds	bigint	需要	确定在此分区中未提交事务的情况下返回检测信号 ChangeRecord 的频率。 该值必须介于 1,000（1 秒）到 300,000（5 分钟）之间。
null	null	需要	留待将来使用

我们建议提供了一种便捷方法来构建读取函数的文本并将参数绑定到该函数，如以下示例所示。

Java

private static final String SINGERS_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.
public static Statement getChangeStreamQuery(
      String partitionToken,
      Timestamp startTimestamp,
      Timestamp endTimestamp,
      long heartbeatMillis) {

  return Statement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE)
                    .bind("p1")
                    .to(startTimestamp)
                    .bind("p2")
                    .to(endTimestamp)
                    .bind("p3")
                    .to(partitionToken)
                    .bind("p4")
                    .to(heartbeatMillis)
                    .build();
}

变更数据流记录格式

GoogleSQL

变更数据流读取函数会返回类型为 ARRAY<STRUCT<...>> 的单个 ChangeRecord 列。在每一行中，此数组始终包含一个元素。

数组元素具有以下类型：

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

此结构体中有三个字段：data_change_record、heartbeat_record 和 child_partitions_record，每个字段的类型均为 ARRAY<STRUCT<...>>。在变更数据流读取函数返回的任何行中，只有这三个字段中的一个包含值；另外两个字段为空或 NULL。这些数组字段最多包含一个元素。

以下各部分分别介绍了这三种记录类型。

PostgreSQL

变更数据流读取函数会返回 JSON 类型的单个 ChangeRecord 列，其结构如下：

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

此对象中有三个可能的键：data_change_record、heartbeat_record 和 child_partitions_record，对应的值类型为 JSON。在变更数据流读取函数返回的任何行中，只有这三个键中的一个存在。

以下各部分分别介绍了这三种记录类型。

数据变更记录

数据更改记录包含对表的一组更改，这些更改的修改类型（插入、更新或删除）的提交时间戳是针对同一事务的一个更改流分区中提交的。可以跨多个变更数据流分区的同一事务返回多个数据变更记录。

所有数据变更记录都具有 commit_timestamp、server_transaction_id 和 record_sequence 字段，这些字段共同决定了流记录在变更流中的顺序。这三个字段足以推导更改的顺序并提供外部一致性。

请注意，如果多个事务涉及不重叠的数据，则它们可以具有相同的提交时间戳。server_transaction_id 字段用于区分是在同一事务中发出的哪组更改（可能跨变更数据流分区）。通过将其与 record_sequence 和 number_of_records_in_transaction 字段配对，您还可以缓冲特定事务中的所有记录并对其进行排序。

数据更改记录的字段包括：

GoogleSQL

字段	类型	说明
commit_timestamp	TIMESTAMP	提交更改的时间戳。
record_sequence	STRING	事务中记录的序列号。在事务中，序列号必须唯一且单调递增（但不一定是连续的）。按 record_sequence 对同一 server_transaction_id 的记录进行排序，以重构事务中更改的排序。
server_transaction_id	STRING	一个全局唯一字符串，表示在其中提交更改的事务。该值应仅用于处理变更数据流记录的上下文，且与 Spanner API 中的事务 ID 无关。
is_last_record_in_transaction_in_partition	BOOL	指示这是否是当前分区中事务的最后一条记录。
table_name	STRING	受更改影响的表的名称。
value_capture_type	STRING	描述捕获此更改时在变更数据流配置中指定的值捕获类型。 值捕获类型可以是 "OLD_AND_NEW_VALUES"、"NEW_ROW"、"NEW_VALUES" 或 "NEW_ROW_AND_OLD_VALUES"。 默认为 "OLD_AND_NEW_VALUES"。如需了解详情，请参阅值捕获类型。
column_types	ARRAY<STRUCT< name STRING,  type JSON,  is_primary_key BOOL,  ordinal_position INT64 >>	列的名称、列类型、是否为主键，以及架构中定义的列的位置（“ordinal_position”）。架构中表的第一列的序号位置为“1”。列类型可以嵌套在数组列中。该格式与 Spanner API 参考文档中所述的类型结构相符。
mods	ARRAY<STRUCT< keys JSON,  new_values JSON,  old_values JSON >>	描述所做的更改，包括主键值对、旧值以及已更改的列或跟踪的列的新值。旧值和新值的可用性和内容将取决于已配置的 value_capture_type。new_values 和 old_values 字段仅包含非键列。
mod_type	STRING	说明更改的类型。其值为 INSERT、UPDATE 或 DELETE。
number_of_records_in_transaction	INT64	所有变更数据流分区中属于此事务的数据变更记录数量。
number_of_partitions_in_transaction	INT64	将返回此事务的数据更改记录的分区数量。
transaction_tag	STRING	与此交易关联的 交易代码。
is_system_transaction	BOOL	指示事务是否为系统事务。

PostgreSQL

字段	类型	说明
commit_timestamp	STRING	提交更改的时间戳。
record_sequence	STRING	事务中记录的序列号。在事务中，序列号必须唯一且单调递增（但不一定是连续的）。按“record_sequence”对同一 `server_transaction_id` 的记录进行排序，以重建事务内更改的顺序。
server_transaction_id	STRING	一个全局唯一字符串，表示在其中提交更改的事务。该值应仅用于处理变更数据流记录的上下文，且不与 Spanner API 中的事务 ID 相关联
is_last_record_in_transaction_in_partition	BOOLEAN	指示这是否是当前分区中事务的最后一条记录。
table_name	STRING	受更改影响的表的名称。
value_capture_type	STRING	描述捕获此更改时在变更数据流配置中指定的值捕获类型。 值捕获类型可以是 "OLD_AND_NEW_VALUES"、"NEW_ROW"、"NEW_VALUES" 或 "NEW_ROW_AND_OLD_VALUES"。 默认为 "OLD_AND_NEW_VALUES"。如需了解详情，请参阅值捕获类型。
column_types	[ { "name": <STRING>, "type": { "code": <STRING> }, "is_primary_key": <BOOLEAN>, "ordinal_position": <NUMBER> }, ... ]	列的名称、列类型、是否为主键，以及架构中定义的列的位置（“ordinal_position”）。架构中表的第一列的序号位置为“1”。列类型可以嵌套在数组列中。该格式与 Spanner API 参考文档中所述的类型结构相符。
mods	[ { "keys": {<STRING> : <STRING>}, "new_values": { <STRING> : <VALUE-TYPE>, [...] }, "old_values": { <STRING> : <VALUE-TYPE>, [...] }, }, [...] ]	描述所做的更改，包括主键值对、旧值以及已更改或跟踪的列的新值。旧值和新值的可用性和内容将取决于配置的 value_capture_type。new_values 和 old_values 字段仅包含非键列。
mod_type	STRING	说明更改的类型。其值为 INSERT、UPDATE 或 DELETE。
number_of_records_in_transaction	INT64	所有变更数据流分区中属于此事务的数据变更记录数量。
number_of_partitions_in_transaction	NUMBER	将返回此事务的数据更改记录的分区数量。
transaction_tag	STRING	与此交易关联的 交易代码。
is_system_transaction	BOOLEAN	指示事务是否为系统事务。

接下来是一对示例数据更改记录。它们描述了一笔交易，即两个帐号之间进行转移。请注意，这两个帐号位于单独的变更数据流分区中。

"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,
}

以下数据更改记录是具有值捕获类型为 "NEW_VALUES" 的记录示例。请注意，系统只会填充新值。 只修改了 "LastUpdate" 列，因此系统仅返回了该列。

"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
}

以下数据更改记录是具有值捕获类型为 "NEW_ROW" 的记录示例。只修改了 "LastUpdate" 列，但会返回所有跟踪的列。

"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
}

以下数据更改记录是具有值捕获类型为 "NEW_ROW_AND_OLD_VALUES" 的记录示例。只修改了 "LastUpdate" 列，但会返回所有跟踪的列。此值捕获类型会捕获 LastUpdate 的新值和旧值。

"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
}

检测信号记录

当返回检测信号记录时，表示已返回 commit_timestamp 小于或等于检测信号记录的 timestamp 的所有更改，并且此分区中未来数据记录的提交时间戳必须高于检测信号记录返回的提交时间戳。当没有数据更改写入分区时，系统会返回检测信号记录。当有写入分区的数据更改时，可以使用 data_change_record.commit_timestamp（而非 heartbeat_record.timestamp）来告知读取器正在读取分区。

您可以使用在分区上返回的检测信号记录在所有分区之间同步读取器。一旦所有读取器都收到了大于或等于某个时间戳 A 的检测信号，或者已接收到大于或等于时间戳 A 的数据或子分区记录，则读取器知道自己已收到在该时间戳 A 或之前提交的所有记录，并且可以开始处理已缓冲的记录，例如，按时间戳对跨分区记录进行排序并按 server_transaction_id 对它们进行分组。

检测信号记录仅包含一个字段：

GoogleSQL

字段	类型	说明
timestamp	TIMESTAMP	检测信号记录的时间戳。

PostgreSQL

字段	类型	说明
timestamp	STRING	检测信号记录的时间戳。

以下心跳记录示例已返回时间戳小于或等于此记录的时间戳的所有记录：

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

子分区记录

子分区记录会返回子分区的相关信息：子分区的分区令牌、父分区的令牌，以及表示子分区包含变更记录的最早时间戳的 start_timestamp。提交时间戳早于 child_partitions_record.start_timestamp 的记录将在当前分区中返回。返回此分区的所有子分区记录后，此查询将返回成功状态，表示已返回此分区的所有记录。

子分区记录的字段包括：

GoogleSQL

字段	类型	说明
start_timestamp	TIMESTAMP	从此子分区记录中的子分区返回的数据变更记录的提交时间戳大于或等于 start_timestamp。查询子分区时，查询应指定子分区令牌以及大于或等于 child_partitions_token.start_timestamp 的 start_timestamp。一个分区返回的所有子分区记录都具有相同的 start_timestamp，并且时间戳始终介于查询的指定 start_timestamp 和 end_timestamp 之间。
record_sequence	STRING	单调递增的序列号，可用于在特定分区中返回具有相同 start_timestamp 的多个子分区记录时定义子分区记录的顺序。分区令牌 start_timestamp 和 record_sequence 唯一标识子分区记录。
child_partitions	ARRAY<STRUCT< token STRING, parent_partition_tokens ARRAY<STRING> >>	返回一组子分区及其相关信息。这包括用于在查询中标识子分区的分区令牌字符串，及其父分区的令牌。

PostgreSQL

字段	类型	说明
start_timestamp	STRING	从这个子分区记录中的子分区返回的数据变更记录的提交时间戳大于或等于 start_timestamp。查询子分区时，查询应指定子分区令牌以及大于或等于 child_partitions_token.start_timestamp 的 start_timestamp。一个分区返回的所有子分区记录都具有相同的 start_timestamp，并且时间戳始终介于查询的指定 start_timestamp 和 end_timestamp 之间。
record_sequence	STRING	单调递增的序列号，可用于在特定分区中返回具有相同 start_timestamp 的多个子分区记录时定义子分区记录的顺序。分区令牌 start_timestamp 和 record_sequence 唯一标识子分区记录。
child_partitions	[ { "token": <STRING>, "parent_partition_tokens": [<STRING>], }, [...] ]	返回子分区及其相关信息的数组。这包括用于在查询中标识子分区的分区令牌字符串，及其父分区的令牌。

以下是子分区记录的示例：

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"]
    }
  ],
}

变更数据流查询工作流

使用 ExecuteStreamingSql API 运行变更数据流查询，并提供一次性只读事务和强时间戳边界。通过变更数据流读取函数，您可以指定相关时间范围的 start_timestamp 和 end_timestamp。可以使用强只读时间戳边界访问保留期限内的所有更改历史记录。

所有其他 TransactionOptions 对于变更数据流查询无效。此外，如果 TransactionOptions.read_only.return_read_timestamp 设置为 true，则在描述事务的 Transaction 消息中会返回特殊值 kint64max - 1，而不是有效的读取时间戳。此特殊值应该被舍弃，并且不用于任何后续查询。

每个变更数据流查询都可以返回任意数量的行，每行都包含数据更改记录、检测信号记录或子分区记录。您无需为请求设置截止时间。

示例：

流式查询工作流首先通过将 partition_token 指定为 NULL 来发出第一个变更数据流查询。该查询需要指定变更数据流的读取函数、感兴趣的开始和结束时间戳，以及检测信号间隔。当 end_timestamp 为 NULL 时，查询会不断返回数据更改，直到分区结束。

GoogleSQL

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

PostgreSQL

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

处理此查询中的数据记录，直到返回子分区记录。在下面的示例中，返回了两个子分区记录和三个分区令牌，然后查询终止。特定查询中的子分区记录始终共用同一 start_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]
    }
  ],
}

如需在 2022-05-01T09:00:01Z 之后处理将来的更改，请创建三个新查询并并行运行这些查询。这三个查询结合起来会返回其父级涵盖的同一键范围的未来数据更改。请务必将 start_timestamp 设置为同一子分区记录中的 start_timestamp，并使用相同的 end_timestamp 和检测信号间隔在所有查询中一致地处理记录。

GoogleSQL

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

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

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

PostgreSQL

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

一段时间后，对 child_token_2 的查询在返回另一个子分区记录后完成，此记录表示从 2022-05-01T09:30:15Z 开始，新分区将涵盖 child_token_2 和 child_token_3 的未来更改。针对 child_token_3 的查询将返回完全相同的记录，因为两者都是新 child_token_4 的父分区。为了保证对特定键的数据记录进行严格排序的处理，只有在所有父级都完成后（在本例中为 child_token_2 和 child_token_3）才能开始对 child_token_4 的查询。请仅为每个子分区令牌创建一个查询，在查询工作流设计中应指定一个父项来等待并在 child_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],
    }
  ],
}

GoogleSQL

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

PostgreSQL

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

在 GitHub 上的 Apache Beam SpannerIO Dataflow 连接器中，查找处理和解析变更数据流记录的示例。