更改流分区、记录和查询

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

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

此页面上的信息与使用 Spanner API 直接变更数据流最为相关。改为使用 Dataflow 读取变更数据流数据的应用无需直接使用此处所述的数据模型。

如需了解更宽泛的数据流变更入门指南，请参阅变更数据流概览。

更改流分区

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

作为 DML 应用于特定拆分的一部分，Spanner 会将写入操作附加到同一事务中的相应更改流数据拆分。由于具有这种对接位置，变更流不会跨服务资源添加额外的协调特性，从而最大限度地减少事务提交开销。

Spanner 会根据数据库负载和大小动态拆分和合并数据，并跨服务资源分布拆分，从而进行扩缩。

为了支持变更数据流写入和扩缩，Spanner 会拆分内部变更流的存储及数据库数据，并自动避免热点问题。为了支持在数据库写入扩缩时近乎实时地读取更改流记录，Spanner API 旨在使用更改流分区并发查询更改流。变更数据流分区映射以更改包含变更数据流记录的数据流数据拆分。变更流的分区会随时间动态变化，并且与 Spanner 动态拆分和合并数据库数据的方式相关。

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

查询特定更改流分区时，系统会以提交时间戳顺序返回更改记录。每个更改记录只返回一次。在各个变更流分区中，无法保证变更记录的顺序。特定主键的更改记录仅在特定时间范围内只有一个分区返回。

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

更改流查询语法

GoogleSQL

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

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

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

该函数接受以下参数：

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

我们建议您制定一种便捷的方法来构建 TVF 查询的文本，并为其绑定参数，如以下示例所示。

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.read_json_change_stream_name。

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

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

该函数接受以下参数：

参数名称	类型	是否必需？	说明
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

变更流 TVF 会返回一个 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<...>>。在从变更流 TVF 返回的任何行中，只有这三个字段之一包含值；其他两个字段为空或 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"。默认为 "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	交易中记录的序列号。序列号在事务中可保证是唯一的，并且单调递增（但不一定是连续的）。按记录记录对同一 `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"。默认为 "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
}

检测信号记录

返回检测信号记录后，这表示 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 运行变更流查询，并使用一次性只读事务和强大的时间戳边界。变更流 TVF 允许用户为感兴趣的时间范围指定 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-01 09:00:00-00",
  end_timestamp => NULL,
  partition_token => NULL,
  heartbeat_milliseconds => 10000
);

PostgreSQL

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

处理此查询中的数据记录，直到返回子分区记录。在以下示例中，返回了两个子分区记录和三个分区令牌，然后查询终止。特定查询的子分区记录始终共用相同的 start_timestamp。

child_partitions_record: {
  "record_type": "child_partitions",
  "start_timestamp": "2022-05-01 09:00:01-00",
  "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-01 09:00:01-00",
  "record_sequence": 1000012390,
  "child_partitions": [
    {
      "token": "child_token_3",
      "parent_partition_tokens": [NULL]
    }
  ],
}

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

GoogleSQL

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

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

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

PostgreSQL

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

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

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

一段时间后，在返回另一个子分区记录后，对 child_token_2 的查询完成，此记录表明，从 2022-05-01 09:30:15-00 开始，新的分区将覆盖 child_token_2 和 child_token_3 的未来更改。该查询会针对 child_token_3 返回完全相同的记录，因为两个记录都是新 child_token_4 的父分区。为了确保特定键的数据记录具有严格有序的处理，则对 child_token_4 的查询必须在所有父项都完成后才能开始（在本例中为 child_token_2 和 child_token_3）。只需为每个子分区令牌创建一个查询，查询工作流设计应指定一位用户等待，并在 child_token_4 上安排查询。

child partitions record: {
  "record_type": "child_partitions",
  "start_timestamp": "2022-05-01 09:30:15-00",
  "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-01 09:30:15-00",
  end_timestamp => NULL,
  partition_token => "child_token_4",
  heartbeat_milliseconds => 10000
);

PostgreSQL

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

您可以在 GitHub 上找到 Apache Beam SpannerIO Dataflow 连接器处理和解析更改流记录的示例。