diff --git a/ticdc/ticdc-debezium.md b/ticdc/ticdc-debezium.md
index 773d01394494..b83addc64bbb 100644
--- a/ticdc/ticdc-debezium.md
+++ b/ticdc/ticdc-debezium.md
@@ -5,13 +5,30 @@ summary: 了解 TiCDC Debezium Protocol 的概念和使用方法。
# TiCDC Debezium Protocol
-[Debezium](https://debezium.io/) 是一个用于捕获数据库变更的工具。它会将捕获的数据库变更的每一条记录转化为一个被称为“事件” (event) 的消息，并将这些事件发送到 Kafka 中。从 v8.0.0 起，TiCDC 支持将 TiDB 的变更以 Debezium 的格式直接传输到 Kafka，为之前使用 Debezium 的 MySQL 集成的用户简化了从 MySQL 数据库迁移的过程。
+[Debezium](https://debezium.io/) 是一个用于捕获数据库变更的工具。它会将捕获的数据库变更的每一条记录转化为一个被称为“事件” (event) 的消息 (message)，并将这些事件发送到 Kafka 中。从 v8.0.0 起，TiCDC 支持将 TiDB 的行数据变更（DML 事件）以 Debezium 的格式直接传输到 Kafka，为之前使用 Debezium 的 MySQL 集成的用户简化了从 MySQL 数据库迁移的过程。从 [TiCDC v8.5.4-release.1](https://github.com/pingcap/ticdc/releases/tag/v8.5.4-release.1)（[TiCDC 新架构](/ticdc/ticdc-architecture.md)）起，TiCDC 支持以 Debezium 的格式同步 DDL 事件和 WATERMARK 事件。
## 使用 Debezium 消息格式
-当使用 Kafka 作为下游 Sink 时，你可以将 `sink-uri` 的 `protocol` 字段指定为 `debezium`，TiCDC 将以 Event 为基本单位封装构造 Debezium 消息，向下游发送 TiDB 的数据变更事件。
+当使用 Kafka 作为下游 Sink 时，你可以将 `sink-uri` 的 `protocol` 字段指定为 `debezium`，TiCDC 将以 event 为基本单位封装构造 Debezium 消息，向下游发送 TiDB 的数据变更事件。
-目前，Debezium 协议只支持 Row Changed Event，会直接忽略 DDL Event 和 WATERMARK Event。Row Changed Event 代表一行的数据变化，在行发生变更时该 Event 被发出，包含该行在变更前后的相关信息。WATERMARK Event 用于标记一个表的同步进度，所有早于 watermark 的事件都已经发送给下游。
+
+
+
+要使用 TiCDC 新架构，请将 TiCDC 配置项 [`newarch`](/ticdc/ticdc-server-config.md#newarch-从-v854-release1-版本开始引入) 设置为 `true`。
+
+对于 TiCDC 新架构，Debezium 协议支持以下类型的事件：
+
+- DDL 事件：表示 DDL 变更记录。在上游 DDL 语句成功执行后，DDL 事件被发送到每个 MQ (Message Queue) 分区。
+- DML 事件：表示一行数据变更记录。在行变更发生时，DML 事件被发出，包含变更后该行的相关信息。
+- WATERMARK 事件：表示一个特殊的时间点。在这个时间点之前收到的事件是完整的。仅适用于 TiDB 扩展字段，当你在 `sink-uri` 中设置 [`enable-tidb-extension`](/ticdc/ticdc-sink-to-kafka.md#sink-uri-配置-kafka) 为 `true` 时生效。
+
+
+
+
+对于 TiCDC 老架构，Debezium 协议仅支持 Row Changed 事件，并会直接忽略 DDL 事件和 WATERMARK 事件。Row changed 事件表示一行中的数据变更。当某一行发生变更时，会发送 Row Changed 事件，其中包含该行变更前后的相关信息。WATERMARK 事件标记表的复制进度，表示所有早于该 watermark 的事件都已发送到下游。
+
+
+
使用 Debezium 消息格式时的配置样例如下所示：
@@ -23,32 +40,427 @@ Debezium 输出格式中包含当前行的 Schema 信息，以便下游消费者
此外，Debezium 原有格式中并不包含 TiDB 专有的 `CommitTS` 事务唯一标识等重要字段。为了保证数据的完整性，TiCDC 在 Debezium 格式中增加了 `CommitTs` 和 `ClusterID` 两个字段，用于标识 TiDB 数据变更的相关信息。
-## Message 格式定义
+## 消息格式定义
+
+本节介绍 DDL 事件、DML 事件和 WATERMARK 事件的消息格式。
-下面将介绍 Debezium 格式输出的 DML Event 的格式定义。
+### DDL 事件（TiCDC 新架构）
-### DML Event
+> **注意：**
+>
+> DDL 事件仅在 [TiCDC 新架构](/ticdc/ticdc-architecture.md)中支持。在 [TiCDC 老架构](/ticdc/ticdc-classic-architecture.md)中，DDL 事件会被忽略。
-TiCDC 会将一个 DML 事件转换为一个 Kafka 事件，其中事件的 key 和 value 都按照 Debezium 协议进行编码。
+TiCDC 会将一个 DDL 事件转换为一条 Kafka 消息，其中消息的 key 和 value 都按照 Debezium 协议进行编码。
#### Key 数据格式
```json
{
"payload": {
- "a": 4
+ "databaseName": "test"
},
"schema": {
+ "type": "struct",
+ "name": "io.debezium.connector.mysql.SchemaChangeKey",
+ "optional": false,
+ "version": 1,
"fields": [
{
- "field": "a",
+ "field": "databaseName",
+ "optional": false,
+ "type": "string"
+ }
+ ]
+ }
+}
+```
+
+Key 中的字段仅包含数据库名称。字段解释如下：
+
+| 字段 | 类型 | 说明 |
+|:--------------|:-----------|:---------------------------------------------|
+| `payload` | JSON | 数据库名称信息。 |
+| `schema.fields` | JSON | `payload` 中各个字段的类型信息。 |
+| `schema.type` | 字符串 | 字段类型。 |
+| `schema.optional` | 布尔值 | 该字段是否为选填项。值为 `true` 表示该字段为选填项。 |
+| `schema.version` | 字符串 | schema 的版本。 |
+
+#### Value 数据格式
+
+```json
+{
+ "payload": {
+ "source": {
+ "version": "2.4.0.Final",
+ "connector": "TiCDC",
+ "name": "test_cluster",
+ "ts_ms": 0,
+ "snapshot": "false",
+ "db": "test",
+ "table": "table1",
+ "server_id": 0,
+ "gtid": null,
+ "file": "",
+ "pos": 0,
+ "row": 0,
+ "thread": 0,
+ "query": null,
+ "commit_ts": 1,
+ "cluster_id": "test_cluster"
+ },
+ "ts_ms": 1701326309000,
+ "databaseName": "test",
+ "schemaName": null,
+ "ddl": "RENAME TABLE test.table1 to test.table2",
+ "tableChanges": [
+ {
+ "type": "ALTER",
+ "id": "\"test\".\"table2\",\"test\".\"table1\"",
+ "table": {
+ "defaultCharsetName": "",
+ "primaryKeyColumnNames": [
+ "id"
+ ],
+ "columns": [
+ {
+ "name": "id",
+ "jdbcType": 4,
+ "nativeType": null,
+ "comment": null,
+ "defaultValueExpression": null,
+ "enumValues": null,
+ "typeName": "INT",
+ "typeExpression": "INT",
+ "charsetName": null,
+ "length": 0,
+ "scale": null,
+ "position": 1,
+ "optional": false,
+ "autoIncremented": false,
+ "generated": false
+ }
+ ],
+ "comment": null
+ }
+ }
+ ]
+ },
+ "schema": {
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "name": "io.debezium.connector.mysql.SchemaChangeValue",
+ "fields": [
+ {
+ "field": "source",
+ "name": "io.debezium.connector.mysql.Source",
+ "optional": false,
+ "type": "struct",
+ "fields": [
+ {
+ "field": "version",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "connector",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "name",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "ts_ms",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "snapshot",
+ "optional": true,
+ "type": "string",
+ "parameters": {
+ "allowed": "true,last,false,incremental"
+ },
+ "default": "false",
+ "name": "io.debezium.data.Enum",
+ "version": 1
+ },
+ {
+ "field": "db",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "sequence",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "table",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "server_id",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "gtid",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "file",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "pos",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "row",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "thread",
+ "optional": true,
+ "type": "int64"
+ },
+ {
+ "field": "query",
+ "optional": true,
+ "type": "string"
+ }
+ ]
+ },
+ {
+ "field": "ts_ms",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "databaseName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "schemaName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "ddl",
"optional": true,
- "type": "int32"
+ "type": "string"
+ },
+ {
+ "field": "tableChanges",
+ "optional": false,
+ "type": "array",
+ "items": {
+ "name": "io.debezium.connector.schema.Change",
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "fields": [
+ {
+ "field": "type",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "id",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "table",
+ "optional": true,
+ "type": "struct",
+ "name": "io.debezium.connector.schema.Table",
+ "version": 1,
+ "fields": [
+ {
+ "field": "defaultCharsetName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "primaryKeyColumnNames",
+ "optional": true,
+ "type": "array",
+ "items": {
+ "type": "string",
+ "optional": false
+ }
+ },
+ {
+ "field": "columns",
+ "optional": false,
+ "type": "array",
+ "items": {
+ "name": "io.debezium.connector.schema.Column",
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "fields": [
+ {
+ "field": "name",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "jdbcType",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "nativeType",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "typeName",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "typeExpression",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "charsetName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "length",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "scale",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "position",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "optional",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "autoIncremented",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "generated",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "comment",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "defaultValueExpression",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "enumValues",
+ "optional": true,
+ "type": "array",
+ "items": {
+ "type": "string",
+ "optional": false
+ }
+ }
+ ]
+ }
+ },
+ {
+ "field": "comment",
+ "optional": true,
+ "type": "string"
+ }
+ ]
+ }
+ ]
+ }
}
+ ]
+ }
+}
+```
+
+以上 JSON 数据的重点字段解释如下：
+
+| 字段 | 类型 | 说明 |
+|:----------|:-------|:-------------------------------------------------------|
+| `payload.ts_ms` | 数值 | TiCDC 生成这条信息的时间戳（毫秒级别）。 |
+| `payload.ddl` | 字符串 | DDL 事件的 SQL 语句。 |
+| `payload.databaseName` | 字符串 | 事件发生的数据库的名称。 |
+| `payload.source.commit_ts` | 数值 | 该事件的 `CommitTs` 值。 |
+| `payload.source.db` | 字符串 | 事件发生的数据库的名称。 |
+| `payload.source.table` | 字符串 | 事件发生的数据表的名称。 |
+| `payload.tableChanges` | 数组 | 在 schema 变更后的整个表 schema 的结构化表示。`tableChanges` 字段包含一个数组，其中包括表中每一列的条目。由于结构化表示以 JSON 或 Avro 格式呈现数据，因此消费者可以在不通过 DDL 解析器处理的情况下轻松读取消息。 |
+| `payload.tableChanges.type` | 字符串 | 描述变更的类型。值为以下之一：`CREATE`，表示表已创建；`ALTER`，表示表已修改；`DROP`，表示表已删除。 |
+| `payload.tableChanges.id` | 字符串 | 被创建、修改或删除的表的完整标识符。如果是表重命名，则该标识符是 <old> 和 <new> 表名的拼接。 |
+| `payload.tableChanges.table.defaultCharsetName` | 字符串 | 事件发生的表的字符集。 |
+| `payload.tableChanges.table.primaryKeyColumnNames` | 字符串 | 组成表主键的列的名称列表。 |
+| `payload.tableChanges.table.columns` | 数组 | 变更的表中每一列的元数据。 |
+| `payload.tableChanges.table.columns.name` | 字符串 | 列的名称。 |
+| `payload.tableChanges.table.columns.jdbcType` | 数值 | 列的 JDBC 类型。 |
+| `payload.tableChanges.table.columns.comment` | 字符串 | 列的注释。 |
+| `payload.tableChanges.table.columns.defaultValueExpression` | 字符串 | 列的默认值。 |
+| `payload.tableChanges.table.columns.enumValues` | 字符串 | 列的枚举值。格式为 `['e1', 'e2']`。 |
+| `payload.tableChanges.table.columns.charsetName` | 字符串 | 列的字符集。 |
+| `payload.tableChanges.table.columns.length` | 数值 | 列的长度。 |
+| `payload.tableChanges.table.columns.scale` | 数值 | 列的精度。 |
+| `payload.tableChanges.table.columns.position` | 数值 | 列的位置。 |
+| `payload.tableChanges.table.columns.optional` | 布尔值 | 是否为可选列。值为 `true` 表示为可选列。 |
+| `schema.fields` | JSON | `payload` 每个字段的类型信息，包括变更表的列 schema 信息。 |
+| `schema.name` | 字符串 | schema 的名称，格式为 `"{cluster-name}.{schema-name}.{table-name}.SchemaChangeValue"`。 |
+| `schema.optional` | 布尔值 | 该字段是否为选填项。值为 `true` 表示该字段为选填项。 |
+| `schema.type` | 字符串 | 字段的数据类型。 |
+
+### DML 事件
+
+TiCDC 会将一个 DML 事件转换为一条 Kafka 消息，其中消息的 key 和 value 都按照 Debezium 协议进行编码。
+
+#### Key 数据格式
+
+```json
+{
+ "payload": {
+ "tiny": 1
+ },
+ "schema": {
+ "fields": [
+ {
+ "field":"tiny",
+ "optional":true,
+ "type":"int16"
+ }
],
- "name": "default.test.t2.Key",
+ "name": "test_cluster.test.table1.Key",
"optional": false,
- "type": "struct"
+ "type":"struct"
}
}
```
@@ -57,93 +469,317 @@ Key 中的字段只包含主键或唯一索引列。字段解释如下：
| 字段 | 类型 | 说明 |
|:----------|:-------|:-------------------------------------------------------------------------|
-| payload | JSON | 主键或唯一索引列的信息。每个字段的 key 和 value 分别为列名和当前值 |
-| schema.fields | JSON | payload 中各个字段的类型信息，包括对应行数据变更前后 schema 的信息等 |
-| schema.name | 字符串 | schema 的名称，格式为 `"{cluster-name}.{schema-name}.{table-name}.Key"` |
-| schema.optional | 布尔值 | optional 为 `true` 时表示该字段为选填项 |
-| schema.type | 字符串 | 表示该字段的数据类型 |
+| `payload` | JSON | 主键或唯一索引列的信息。每个字段的 key 和 value 分别为列名和当前值。 |
+| `schema.fields` | JSON | `payload` 中各个字段的类型信息，包括对应行数据变更前后 schema 的信息。 |
+| `schema.name` | 字符串 | schema 的名称，格式为 `"{cluster-name}.{schema-name}.{table-name}.Key"`。 |
+| `schema.optional` | 布尔值 | 该字段是否为选填项。值为 `true` 表示该字段为选填项。 |
+| `schema.type` | 字符串 | 字段的数据类型。 |
#### Value 数据格式
```json
{
- "payload":{
- "ts_ms":1707103832957,
- "transaction":null,
- "op":"c",
- "before":null,
- "after":{
- "a":4,
- "b":2
+ "payload": {
+ "source": {
+ "version": "2.4.0.Final",
+ "connector": "TiCDC",
+ "name": "test_cluster",
+ "ts_ms": 0,
+ "snapshot": "false",
+ "db": "test",
+ "table": "table1",
+ "server_id": 0,
+ "gtid": null,
+ "file": "",
+ "pos": 0,
+ "row": 0,
+ "thread": 0,
+ "query": null,
+ "commit_ts": 1,
+ "cluster_id": "test_cluster"
},
- "source":{
- "version":"2.4.0.Final",
- "connector":"TiCDC",
- "name":"default",
- "ts_ms":1707103832263,
- "snapshot":"false",
- "db":"test",
- "table":"t2",
- "server_id":0,
- "gtid":null,
- "file":"",
- "pos":0,
- "row":0,
- "thread":0,
- "query":null,
- "commit_ts":447507027004751877,
- "cluster_id":"default"
- }
+ "ts_ms": 1701326309000,
+ "transaction": null,
+ "op": "u",
+ "before": { "tiny": 2 },
+ "after": { "tiny": 1 }
},
- "schema":{
- "type":"struct",
- "optional":false,
- "name":"default.test.t2.Envelope",
- "version":1,
- "fields":{
- {
- "type":"struct",
- "optional":true,
- "name":"default.test.t2.Value",
- "field":"before",
- "fields":[
- {
- "type":"int32",
- "optional":false,
- "field":"a"
- },
- {
- "type":"int32",
- "optional":true,
- "field":"b"
- }
- ]
+ "schema": {
+ "type": "struct",
+ "optional": false,
+ "name": "test_cluster.test.table1.Envelope",
+ "version": 1,
+ "fields": [
+ {
+ "type": "struct",
+ "optional": true,
+ "name": "test_cluster.test.table1.Value",
+ "field": "before",
+ "fields": [{ "type": "int16", "optional": true, "field": "tiny" }]
+ },
+ {
+ "type": "struct",
+ "optional": true,
+ "name": "test_cluster.test.table1.Value",
+ "field": "after",
+ "fields": [{ "type": "int16", "optional": true, "field": "tiny" }]
},
{
- "type":"struct",
- "optional":true,
- "name":"default.test.t2.Value",
- "field":"after",
- "fields":[
+ "type": "struct",
+ "fields": [
+ { "type": "string", "optional": false, "field": "version" },
+ { "type": "string", "optional": false, "field": "connector" },
+ { "type": "string", "optional": false, "field": "name" },
+ { "type": "int64", "optional": false, "field": "ts_ms" },
{
- "type":"int32",
- "optional":false,
- "field":"a"
+ "type": "string",
+ "optional": true,
+ "name": "io.debezium.data.Enum",
+ "version": 1,
+ "parameters": { "allowed": "true,last,false,incremental" },
+ "default": "false",
+ "field": "snapshot"
},
+ { "type": "string", "optional": false, "field": "db" },
+ { "type": "string", "optional": true, "field": "sequence" },
+ { "type": "string", "optional": true, "field": "table" },
+ { "type": "int64", "optional": false, "field": "server_id" },
+ { "type": "string", "optional": true, "field": "gtid" },
+ { "type": "string", "optional": false, "field": "file" },
+ { "type": "int64", "optional": false, "field": "pos" },
+ { "type": "int32", "optional": false, "field": "row" },
+ { "type": "int64", "optional": true, "field": "thread" },
+ { "type": "string", "optional": true, "field": "query" }
+ ],
+ "optional": false,
+ "name": "io.debezium.connector.mysql.Source",
+ "field": "source"
+ },
+ { "type": "string", "optional": false, "field": "op" },
+ { "type": "int64", "optional": true, "field": "ts_ms" },
+ {
+ "type": "struct",
+ "fields": [
+ { "type": "string", "optional": false, "field": "id" },
+ { "type": "int64", "optional": false, "field": "total_order" },
{
- "type":"int32",
- "optional":true,
- "field":"b"
+ "type": "int64",
+ "optional": false,
+ "field": "data_collection_order"
}
- ]
+ ],
+ "optional": true,
+ "name": "event.block",
+ "version": 1,
+ "field": "transaction"
+ }
+ ]
+ }
+}
+```
+
+以上 JSON 数据的重点字段解释如下：
+
+| 字段 | 类型 | 说明 |
+|:----------|:-------|:-------------------------------------------------------------------------|
+| `payload.op` | 字符串 | 变更事件类型。`"c"` 表示 `INSERT` 事件，`"u"` 表示 `UPDATE` 事件，`"d"` 表示 `DELETE` 事件。 |
+| `payload.ts_ms` | 数值 | TiCDC 生成这条信息的时间戳（毫秒级别）。 |
+| `payload.before` | JSON | 这条事件语句变更前的数据值。对于 `"c"` 事件，`before` 字段的值为 `null`。 |
+| `payload.after` | JSON | 这条事件语句变更后的数据值。对于 `"d"` 事件，`after` 字段的值为 `null`。 |
+| `payload.source.commit_ts` | 数值 | 该事件的 `CommitTs` 值。 |
+| `payload.source.db` | 字符串 | 事件发生的数据库的名称。 |
+| `payload.source.table` | 字符串 | 事件发生的数据表的名称。 |
+| `schema.fields` | JSON | `payload` 中各个字段的类型信息，包括对应行数据变更前后 schema 的信息。 |
+| `schema.fields[1].fields[n].tidb_type` | 字符串 | `payload.after` 中每列的 TiDB 类型。仅在 `enable-tidb-extension = true` 时存在。 |
+| `schema.name` | 字符串 | schema 的名称，格式为 `"{cluster-name}.{schema-name}.{table-name}.Envelope"`。 |
+| `schema.optional` | 布尔值 | 该字段是否为选填项。值为 `true` 表示该字段为选填项。 |
+| `schema.type` | 字符串 | 字段的类型。 |
+
+### WATERMARK 事件（TiCDC 新架构）
+
+> **注意：**
+>
+> WATERMARK 事件仅在 [TiCDC 新架构](/ticdc/ticdc-architecture.md)中支持。在 [TiCDC 老架构](/ticdc/ticdc-classic-architecture.md)中，WATERMARK 事件会被忽略。
+
+TiCDC 会将一个 WATERMARK 事件转换为一条 Kafka 消息，其中消息的 key 和 value 都按照 Debezium 协议进行编码。
+
+#### Key 数据格式
+
+```json
+{
+ "payload": {},
+ "schema": {
+ "fields": [],
+ "optional": false,
+ "name": "test_cluster.watermark.Key",
+ "type": "struct"
+ }
+}
+```
+
+Key 中的字段解释如下：
+
+| 字段 | 类型 | 说明 |
+|:-----------|:--------|:---------------------------------------------------------------|
+| `schema.name` | 字符串 | schema 的名称，格式为 `"{cluster-name}.watermark.Key"`。 |
+
+#### Value 数据格式
+
+```json
+{
+ "payload": {
+ "source": {
+ "version": "2.4.0.Final",
+ "connector": "TiCDC",
+ "name": "test_cluster",
+ "ts_ms": 0,
+ "snapshot": "false",
+ "db": "",
+ "table": "",
+ "server_id": 0,
+ "gtid": null,
+ "file": "",
+ "pos": 0,
+ "row": 0,
+ "thread": 0,
+ "query": null,
+ "commit_ts": 3,
+ "cluster_id": "test_cluster"
+ },
+ "op": "m",
+ "ts_ms": 1701326309000,
+ "transaction": null
+ },
+ "schema": {
+ "type": "struct",
+ "optional": false,
+ "name": "test_cluster.watermark.Envelope",
+ "version": 1,
+ "fields": [
+ {
+ "type": "struct",
+ "fields": [
+ {
+ "type": "string",
+ "optional": false,
+ "field": "version"
+ },
+ {
+ "type": "string",
+ "optional": false,
+ "field": "connector"
+ },
+ {
+ "type": "string",
+ "optional": false,
+ "field": "name"
+ },
+ {
+ "type": "int64",
+ "optional": false,
+ "field": "ts_ms"
+ },
+ {
+ "type": "string",
+ "optional": true,
+ "name": "io.debezium.data.Enum",
+ "version": 1,
+ "parameters": {
+ "allowed": "true,last,false,incremental"
+ },
+ "default": "false",
+ "field": "snapshot"
+ },
+ {
+ "type": "string",
+ "optional": false,
+ "field": "db"
+ },
+ {
+ "type": "string",
+ "optional": true,
+ "field": "sequence"
+ },
+ {
+ "type": "string",
+ "optional": true,
+ "field": "table"
+ },
+ {
+ "type": "int64",
+ "optional": false,
+ "field": "server_id"
+ },
+ {
+ "type": "string",
+ "optional": true,
+ "field": "gtid"
+ },
+ {
+ "type": "string",
+ "optional": false,
+ "field": "file"
+ },
+ {
+ "type": "int64",
+ "optional": false,
+ "field": "pos"
+ },
+ {
+ "type": "int32",
+ "optional": false,
+ "field": "row"
+ },
+ {
+ "type": "int64",
+ "optional": true,
+ "field": "thread"
+ },
+ {
+ "type": "string",
+ "optional": true,
+ "field": "query"
+ }
+ ],
+ "optional": false,
+ "name": "io.debezium.connector.mysql.Source",
+ "field": "source"
},
{
- "type":"string",
- "optional":false,
- "field":"op"
+ "type": "string",
+ "optional": false,
+ "field": "op"
},
- ...
- }
+ {
+ "type": "int64",
+ "optional": true,
+ "field": "ts_ms"
+ },
+ {
+ "type": "struct",
+ "fields": [
+ {
+ "type": "string",
+ "optional": false,
+ "field": "id"
+ },
+ {
+ "type": "int64",
+ "optional": false,
+ "field": "total_order"
+ },
+ {
+ "type": "int64",
+ "optional": false,
+ "field": "data_collection_order"
+ }
+ ],
+ "optional": true,
+ "name": "event.block",
+ "version": 1,
+ "field": "transaction"
+ }
+ ]
}
}
```
@@ -152,17 +788,15 @@ Key 中的字段只包含主键或唯一索引列。字段解释如下：
| 字段 | 类型 | 说明 |
|:----------|:-------|:-------------------------------------------------------------------------|
-| payload.op | 字符串 | 变更事件类型。`"c"` 表示这是一个 `INSERT` 事件，`"u"` 表示这是一个 `UPDATE` 事件，`"d"` 表示这是一个 `DELETE` 事件 |
-| payload.ts_ms | 数值 | TiCDC 生成这条信息的时间戳（毫秒级别） |
-| payload.before | JSON | 这条事件语句变更前的数据值，对于 `"c"` 事件，`before` 字段的值为 `null` |
-| payload.after | JSON | 这条事件语句变更后的数据值，对于 `"d"` 事件，`after` 字段的值为 `null` |
-| payload.source.commit_ts | 数值 | TiCDC 生成这条信息时的 `CommitTs` 标识 |
-| payload.source.db | 字符串 | 事件发生的数据库的名称 |
-| payload.source.table | 字符串 | 事件发生的数据表的名称 |
-| schema.fields | JSON | payload 中各个字段的类型信息，包括对应行数据变更前后 schema 的信息等 |
-| schema.name | 字符串 | schema 的名称，格式为 `"{cluster}.{schema}.{table}.Envelope"` |
-| schema.optional | 布尔值 | optional 为 `true` 时表示该字段为选填项 |
-| schema.type | 字符串 | 表示该字段的类型 |
+| `payload.op` | 字符串 | 变更事件类型。`"m"` 表示 WATERMARK 事件。 |
+| `payload.ts_ms` | 数值 | TiCDC 生成这条信息的时间戳（毫秒级别）。 |
+| `payload.source.commit_ts` | 数值 | 该事件的 `CommitTs` 值。 |
+| `payload.source.db` | 字符串 | 事件发生的数据库的名称。 |
+| `payload.source.table` | 字符串 | 事件发生的数据表的名称。 |
+| `schema.fields` | JSON | `payload` 中各个字段的类型信息，包括对应行数据变更前后 schema 的信息。 |
+| `schema.name` | 字符串 | schema 的名称，格式为 `"{cluster-name}.watermark.Envelope"`。 |
+| `schema.optional` | 布尔值 | 该字段是否为选填项。值为 `true` 表示该字段为选填项。 |
+| `schema.type` | 字符串 | 字段的类型。 |
### 数据类型映射
@@ -172,4 +806,18 @@ TiCDC Debezium 消息中的数据格式映射基本遵循 [Debezium 的数据类
- 对于 String-likes 的数据类型，包括 Varchar、String、VarString、TinyBlob、MediumBlob、BLOB、LongBlob 等，当该列具有 BINARY 标志时，TiCDC 会将其按照 Base64 编码后以 String 类型表示；当该列没有 BINARY 标志时，TiCDC 则直接将其编码为 String 类型。而原生的 Debezium Connector 会根据 `binary.handling.mode` 以不同的编码方式进行编码。
-- 对于 Decimal 数据类型，包括 `DECIMAL` 和 `NUMERIC`，TiCDC 均会使用 float64 类型来表示。而原生的 Debezium Connector 会根据数据类型的不同精度采用 float32 或者 float64 的方式进行编码。
+- 对于 Decimal 数据类型，包括 DECIMAL 和 NUMERIC，TiCDC 均会使用 float64 类型来表示。而原生的 Debezium Connector 会根据数据类型的不同精度采用 float32 或者 float64 的方式进行编码。
+
+- TiCDC 将 REAL 转换为 DOUBLE；当长度为 1 时，将 BOOLEAN 转换为 TINYINT(1)。
+
+- 在 TiCDC 中，BLOB、TEXT、GEOMETRY、JSON 列没有默认值。
+
+- Debezium 将 FLOAT 类型的 `"5.61"` 转换为 `"5.610000133514404"`，但 TiCDC 不会。
+
+- TiCDC 在处理 FLOAT 时打印了错误的 `flen` [tidb#57060](https://github.com/pingcap/tidb/issues/57060)。
+
+- 当列的排序规则为 `utf8_unicode_ci` 且字符集为 null 时，Debezium 将 `charsetName` 转换为 `"utf8mb4"`，但 TiCDC 不会。
+
+- TiCDC 将 ENUM 元素中的 `\` 视为转义引号，但 Debezium 不会。例如，TiCDC 将 ENUM 元素 `("c,\'d','g,''h")` 编码为 `('c,'d', 'g,''h')`。
+
+- TiCDC 将 TIME 类型的默认值如 `'1000-00-00 01:00:00.000'` 转换为 `"1000-00-00"`，但 Debezium 不会。