此页面由 Cloud Translation API 翻译。

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

本页介绍了适用于 GoogleSQL 方言数据库和 PostgreSQL 方言数据库的 Spanner 中的变更数据流，包括：

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

您可以使用 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` 之前的所有 `ChangeRecords` 后或您终止连接时完成。如果 `end_timestamp` 设置为 `NULL` 或未指定，则查询会继续执行，直到返回所有 `ChangeRecords` 或您终止连接为止。
`partition_token`	`STRING`	可选（默认值：`NULL`）	根据子分区记录的内容指定要查询的变更数据流分区。如果为 `NULL` 或未指定，则表示读取器是首次查询更改流，并且尚未获取任何要查询的特定分区令牌。
`heartbeat_milliseconds`	`INT64`	必填	确定在没有在此分区中提交任何事务的情况下，返回心跳 `ChangeRecord` 的频率。该值必须介于 `1,000`（1 秒）和 `300,000`（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<...>>
>

此 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` 的记录进行排序，以重建事务中更改的顺序。Spanner 可能会优化此排序以提升性能，并且此排序可能并不总是与您提供的原始排序一致。
`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`	[ { "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`	`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`	[ { "token" : "`STRING`", "parent_partition_tokens" : ["`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_4 进行查询。在本例中，父级为 child_token_2 和 child_token_3。请仅为每个子分区令牌创建一个查询。查询工作流设计应指定一个父级等待并在 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 连接器。