This topic describes the serialization methods and data formats for transmitting data from the database to the text protocol.
Format of the serialization method
When you use OceanBase Migration Service (OMS) to synchronize data from a source to a destination such as Kafka, DataHub (BLOB type), or RocketMQ, you can control the serialization method to determine the message format during data synchronization. The supported serialization methods are Default, Canal, DataWorks (V2.0), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten, DebeziumSmt, and Avro.
Note
Currently, only the MySQL compatible mode of OceanBase Database supports the
Debezium,DebeziumFlatten, andDebeziumSmtserialization methods.Currently, only data from the MySQL compatible mode of OceanBase Database can be synchronized to Kafka by using the
Avroserialization method.
Default JSON message format
When data is synchronized to Kafka, DataHub (BLOB type), and RocketMQ, the default JSON message format is used.
{
"prevStruct": { // The data before the change.
"col1": "val1" // Key-value pairs that contain all keys and values.
},
"postStruct": { // The data after the change.
"col1": "val1" // Key-value pairs that contain all keys and values.
},
"allMetaData"{
"checkpoint": "STRING", // The current synchronization point. In incremental synchronization, it indicates the time point to which data is synchronized (in seconds). In full synchronization, it indicates the primary key-value pairs.
"record_primary_key": "STRING", // The name of the primary key column. If multiple columns are used, they are separated by \u0001.
"record_primary_value": "STRING", // The value of the primary key. If multiple columns are used, they are separated by \u0001.
"source_identity": "STRING", // The source identifier. In incremental synchronization, it indicates the subtopic. In full synchronization, it is an irrelevant sequence number.
"dbType": "STRING", // The type of the database. Valid values: MYSQL, ORACLE, OCEANBASE (legacy mode, compatible with new mode), OB_IN_ORACLE_MODE (legacy mode, compatible with new mode), DB2 (legacy mode, compatible with new mode), OB_MYSQL, OB_ORACLE, and DB2_LUW.
"storeDataSequence": "LONG", // This field exists only in incremental synchronization and is specified in the source.json file as sequenceEnabled=true. The default value is true. It is used for sorting. The generation rule is a timestamp plus an incrementing sequence number of up to five digits. The format is {timestamp}{incrementing sequence number}.
"table_name": "STRING", // The name of the table that is changed by using an SQL statement.
"db": "STRING", // The name of the database that is changed by using an SQL statement. If the database is OceanBase Database, it contains the tenant name. The format is {tenant}.{database}.
"timestamp": "STRING", // The timestamp of the data change in seconds. This field exists only in incremental synchronization.
"uniqueId": "STRING", // The transaction sequence number identifier passed down by STORE in incremental synchronization.
"transId": "STRING", // In OceanBase Database V4.x and earlier versions, this field indicates the transaction ID. inc is an incrementing number, addr is the address of the coordinator, and ts is the timestamp when the transaction ID is generated. hash is the hash value of the three values. If the transaction is incomplete, this field is null. For example, "transId":"{hash:614827533592190202, inc:1384316437, addr:\"xxx.xxx.xxx.xxx:2882\", t:1756190533588977}". In OceanBase Database V4.x and later versions, this field indicates the tenant ID and transaction ID. For example, "transId":"1xxx_3152xxx".
"clusterId": "STRING", // In OceanBase Database, this field indicates the cluster ID. In MySQL, this field indicates the thread ID. In Oracle, this field does not exist.
"ddlType": "STRING", // The specific type of DDL operations. This field is supported in OMS V4.0.1 and later versions.
},
"recordType": "INSERT/UPDATE/DELETE/HEARTBEAT/DDL/ROW" // The type of the change.
}
In the DDL record, only the key named "ddl" exists, and the value is the DDL statement.
The prevStruct and postStruct fields indicate the data before and after the change, respectively:
prevStruct: The data before the change, that is, the data before the SQL statement is executed.postStruct: The data after the change, that is, the data after the SQL statement is executed.
DELETEonly exists inprevStruct,INSERTandDDLonly exist inpostStruct,UPDATEexists in bothprevStructandpostStruct, andHEARTBEAT(periodic heartbeat messages) does not exist inpostStruct.The
clusterIdfield indicates the following:In OceanBase Database,
ob_org_cluster_idis used to set the cluster ID at the session level. This value is persisted in the transaction log. If you setob_org_cluster_idwhen you write data, the value is used. Otherwise, the default value is used. This field exists only in incremental synchronization. For more information, see ob_org_cluster_id.In MySQL,
clusterIdindicates the thread ID in the MySQL binlog event. MySQL Server assigns an incrementing thread ID to each thread on a connection.
The following example shows the data format:
An example of INSERT (insert) data
{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This example applies to OceanBase Database V4.x and later versions. For OceanBase Database V4.x and earlier versions, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "INSERT", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } }When the
recordTypefield isROW, the data is full data. The data format is the same as that ofINSERT.{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This example applies to OceanBase Database V4.x and later versions. For OceanBase Database V4.x and earlier versions, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "source_identity": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "ROW", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } }An example of UPDATE (update) data
{ "allMetaData": { "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx27, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This example applies to OceanBase Database V4.x and later versions. For OceanBase Database V4.x and earlier versions, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", }, "recordType": "UPDATE", "postStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", } }An example of DELETE (delete) data
{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx27, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This example applies to OceanBase Database V4.x and later versions. For OceanBase Database V4.x and earlier versions, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col16": 1.2222, "col7": 9.99999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345" }, "recordType": "DELETE", "postStruct": null }DDL example
ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT "test" COMMENT 'test';{ "prevStruct": null, "postStruct": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" }, "allMetaData": { "checkpoint": "1671177057", "dbType": "OB_MYSQL", "storeDataSequence": null, "db": "connector_test", "timestamp": "1671177057", "uniqueId": null, "ddlType": "ALTER_TABLE", "record_primary_key": null, "source_identity": null, "record_primary_value": null, "table_name": "all_mysql_type_test" }, "recordType": "DDL" }
JSON message format of Canal
When data is synchronized to Kafka, DataHub (BLOB type), and RocketMQ, Canal uses the following JSON message format for serialization.
{
"database": "STRING", // The name of the database that was changed by using an SQL statement. If the database is OceanBase Database, only the database name is provided, and the tenant name is not required.
"sqlType": {
"col1": "INTEGER", // The type of the changed column. The value is referenced from java.sql.Types.
},
"data": [ // The key-value pairs of the data after the change. Only one message is provided.
{
"col1": "val1"
}
],
"pkNames": [ // The names of the primary key columns.
"col1"
],
"old": [ // This field is provided only for UPDATE messages. It indicates the values of the columns that were changed by the UPDATE statement, that is, the values before the change.
{
"col1": "val1"
}
],
"mysqlType": { // The column type description.
"col": "STRING"
},
"type": "STRING", // The type of the change.
"table": "STRING", // The name of the table that was changed by using an SQL statement.
"es": "LONG", // The change time, in milliseconds.
"isDdl": "BOOLEAN", // Indicates whether the change is a DDL operation.
"ts": "LONG", // The write time of the destination.
"sql": "STRING" // This field is empty.
}
Here are some examples:
INSERT(insert) data example{ "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":1.2222, "col4":9.999999999999999093266253, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.9999999999999990, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "col1", "col2" ], "old":null, "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"INSERT", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618323429026, "sql":"" }UPDATE(update) data example{ "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world 2020", "col3":1.2222, "col4":9.999999999999999093266253372484, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.9999999999999990932662, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "col1", "col2" ], "old":[ { "string":"hello world" } ], "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"UPDATE", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618364572908, "sql":"" }DELETE(delete) data example{ "database":"database", "sqlType":{ "col1":93, "col2":12, "col3":6, "col4":8, "col5":5, "col6":92, "col7":4, "col8":-5, "col9":2004, "col10":-6, "col11":91, "col12":3, "col13":-5, "col14":93 }, "data":[ { "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":1.2222, "col4":9.99999999999999909326625, "col5":129, "col6":"00:01:02", "col7":2147483646, "col8":9223372036854775806, "col9":"aGVsbG8gd29ybGQ=", "col10":3, "col11":"2020-11-25", "col12":9.99999999999999909326625, "col13":10223372036854775806, "col14":"1606233662.012345" } ], "pkNames":[ "int8", "int16" ], "old":null, "mysqlType":{ "col1":"datetime", "col2":"varchar", "col3":"float", "col4":"double", "col5":"smallint", "col6":"time", "col7":"int", "col8":"int64", "col9":"blob", "col10":"tinyint", "col11":"date", "col12":"decimal", "col13":"bigint", "col14":"timestamp" }, "type":"DELETE", "table":"table", "es":1609344671000, "isDdl":false, "ts":1618364660278, "sql":"" }DDL example
ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT "test" COMMENT 'test';{ "database": "connector_test", "sqlType": null, "data": null, "pkNames": null, "old": null, "mysqlType": null, "type": "ALTER", "table": "all_mysql_type_test", "es": 1671177209000, "isDdl": true, "ts": 1671177291475, "sql": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" }
JSON message format of DataWorks
When data is synchronized to Kafka, DataHub (BLOB type), or RocketMQ, DataWorks uses the following JSON message format for data serialization.
{
"version":"2.0", //The protocol version. DataWorks 2.0 is only supported now.
"schema": { //The changed metadata information, specifying only the name and type of the column.
"source": {//Change source information
"dbType": "mysql", //Database type
"dbVersion": "5.7.35", //Database version
"dbName": "myDatabase", //The database name.
"schema": "mySchema", //Schema name. This parameter is required when a system contains a schema.
"table": "tableName" //Table name
},
"column": [//Information about the changed columns, the target table records that need to be updated.
{
"name": "id",
"type": "bigint"
},
{
"name": "name",
"type": "varchar(20)"
},
{
"name": "mydata",
"type": "binary"
},
{
"name": "ts",
"type": "datetime"
}
],
"pk": [//The field is required if the primary key or unique key exists, otherwise it is not required.
"pkName1",
"pkName2"
]
},
"payload": {
"before": {
"data":{
"id": 111,
"name":"scooter",
"mydata": "[base64 string]", // if the data type is binary, base64 encode it
"ts": 1590315269000.123456789 // timestamp, in which the integer part is 13 digits and the decimal part is 9 digits
}
},
"after": {
"data":{
"id": 222,
"name":"donald",
"mydata": "[base64 string]",
"ts": 1590315269000
}
},
"op":"INSERT/UPDATE/DELETE/HEARTBEAT/TRANSACTION_BEGIN/TRANSACTION_END/CREATE/ALTER/ERASE/QUERY/TRUNCATE/RENAME/CINDEX/DINDEX/GTID/XACOMMIT/XAROLLBACK/...",//case sensitive
"timestamp": {
"eventTime": 1620457659000 // The timestamp of when the event occurred, in milliseconds with 13 digits.
},
"ddl": {
"text": "ADD COLUMN ..."
},
"scn": "Self-incremental ID"
},
"extend": { //Extend fields are for future expansion needs. If not required, this can be omitted.
"load_fm":"CIBS" // Indicates the source system. For example, "CIBS".
}
}
Heartbeat messages sent for synchronous tasks:
{
"version": "2.0", //Protocol version
"payload": {
"timestamp": {
"eventTime": 1620457659000 //Heartbeat packet time
},
"op": "HEARTBEAT" //Indicates that the request is a heartbeat request.
}
}
Here is an example of the data:
INSERT(inserts) data example
{
"version":"2.0",
"schema":{
"source":{
"dbType":"ob_mysql",
"dbVersion":null,
"dbName":"db",
"schema":null,
"table":"tab"
},
"column":[
{
"name":"int8",
"type":"TINYINT"
},
{
"name":"int16",
"type":"SMALLINT"
},
{
"name":"int32",
"type":"INT"
},
{
"name":"int64",
"type":"INT64"
},
{
"name":"float32",
"type":"FLOAT"
},
{
"name":"float64",
"type":"DOUBLE"
},
{
"name":"bigInt",
"type":"BIGINT"
},
{
"name":"boolean",
"type":"BOOLEAN"
},
{
"name":"string",
"type":"VARCHAR"
},
{
"name":"bytes",
"type":"BLOB"
},
{
"name":"decimal",
"type":"DECIMAL"
},
{
"name":"localDate",
"type":"DATE"
},
{
"name":"localTime",
"type":"TIME"
},
{
"name":"localDateTime",
"type":"DATETIME"
},
{
"name":"timestamp",
"type":"TIMESTAMP"
},
{
"name":"zonedDateTime",
"type":"ZONED_DATETIME"
},
{
"name":"intervalDayToSecond",
"type":"INTERVAL_DAY_TO_SECOND"
},
{
"name":"intervalYearToMonth",
"type":"INTERVAL_YEAR_TO_MONTH"
}
],
"pk":[
"pkName1",
"pkName12"
]
},
"payload":{
"before":null,
"after":{
"data":{
"col1":3,
"col2":129,
"col3":2147483646,
"col4":9223372036854775806,
"col5":1.2222,
"col6":0.0000000000000008125,
"col7":10223372036854775806,
"col8":1,
"col9":"hello world",
"col10":"aGVsbG8gd29ybGQ=",
"col11":0.000000000000125,
"col12":"2020-11-25",
"col13":"00:01:02",
"col14":"2020-11-25 00:01:02",
"col15":"1606233662.012345",
"col16":"2020-11-25 00:01:02.012345 Asia/Shanghai",
"col17":"INTERVAL '3' DAY",
"col18":"INTERVAL '4' YEAR"
}
},
"op":"INSERT",
"timestamp":{
"eventTime":1647581000000,
"systemTime":1647581000795,
"checkpointTime":1647581000
},
"ddl":null,
"scn":"null"
},
"extend":{
"load_fm": "test"
}
}
UPDATEstatements used to update data.
{
"version":"2.0",
"schema":{
"source":{
"dbType":"ob_mysql",
"dbVersion":null,
"dbName":"db",
"schema":null,
"table":"tab"
},
"column":[
{
"name":"int8",
"type":"TINYINT"
},
{
"name":"int16",
"type":"SMALLINT"
},
{
"name":"int32",
"type":"INT"
},
{
"name":"int64",
"type":"INT64"
},
{
"name":"float32",
"type":"FLOAT"
},
{
"name":"float64",
"type":"DOUBLE"
},
{
"name":"bigInt",
"type":"BIGINT"
},
{
"name":"boolean",
"type":"BOOLEAN"
},
{
"name":"string",
"type":"VARCHAR"
},
{
"name":"bytes",
"type":"BLOB"
},
{
"name":"decimal",
"type":"DECIMAL"
},
{
"name":"localDate",
"type":"DATE"
},
{
"name":"localTime",
"type":"TIME"
},
{
"name":"localDateTime",
"type":"DATETIME"
},
{
"name":"timestamp",
"type":"TIMESTAMP"
},
{
"name":"zonedDateTime",
"type":"ZONED_DATETIME"
},
{
"name":"intervalDayToSecond",
"type":"INTERVAL_DAY_TO_SECOND"
},
{
"name":"intervalYearToMonth",
"type":"INTERVAL_YEAR_TO_MONTH"
}
],
"pk":[
"pkName1",
"pkName2"
]
},
"payload":{
"before":{
"data":{
"col1":3,
"col2":129,
"col3":2147483646,
"col4":9223372036854775806,
"col5":1.2222,
"col6":0.000000000000000125,
"col7":10223372036854775806,
"col8":1,
"col9":"hello world",
"col10":"aGVsbG8gd29ybGQ=",
"col11":0.0000000000125,
"col12":"2020-11-25",
"col13":"00:01:02",
"col14":"2020-11-25 00:01:02",
"col15":"1606233662.012345",
"col16":"2020-11-25 00:01:02.012345 Asia/Shanghai",
"col17":"INTERVAL '3' DAY",
"col18":"INTERVAL '4' YEAR"
}
},
"after":{
"data":{
"col1":3,
"col2":129,
"col3":2147483646,
"col4":9223372036854775806,
"col5":1.2222,
"col6":0.00000000008125,
"col7":10223372036854775806,
"col8":1,
"col9":"hello world 2020",
"col10":"aGVsbG8gd29ybGQ=",
"col11":0.000000000125,
"col12":"2020-11-25",
"col13":"00:01:02",
"col14":"2020-11-25 00:01:02",
"col15":"1606233662.012345",
"col16":"2020-11-25 00:01:02.012345 Asia/Shanghai",
"col17":"INTERVAL '3' DAY",
"col18":"INTERVAL '4' YEAR"
}
},
"op":"UPDATE",
"timestamp":{
"eventTime":1647581038000,
"systemTime":1647581038674,
"checkpointTime":1647581038
},
"ddl":null,
"scn":"null"
},
"extend":{
"load_fm": "test"
}
}
- An example of deleting data using
DELETE.
{
"version":"2.0",
"schema":{
"source":{
"dbType":"ob_mysql",
"dbVersion":null,
"dbName":"db",
"schema":null,
"table":"tab"
},
"column":[
{
"name":"int8",
"type":"TINYINT"
},
{
"name":"int16",
"type":"SMALLINT"
},
{
"name":"int32",
"type":"INT"
},
{
"name":"int64",
"type":"INT64"
},
{
"name":"float32",
"type":"FLOAT"
},
{
"name":"float64",
"type":"DOUBLE"
},
{
"name":"bigInt",
"type":"BIGINT"
},
{
"name":"boolean",
"type":"BOOLEAN"
},
{
"name":"string",
"type":"VARCHAR"
},
{
"name":"bytes",
"type":"BLOB"
},
{
"name":"decimal",
"type":"DECIMAL"
},
{
"name":"localDate",
"type":"DATE"
},
{
"name":"localTime",
"type":"TIME"
},
{
"name":"localDateTime",
"type":"DATETIME"
},
{
"name":"timestamp",
"type":"TIMESTAMP"
},
{
"name":"zonedDateTime",
"type":"ZONED_DATETIME"
},
{
"name":"intervalDayToSecond",
"type":"INTERVAL_DAY_TO_SECOND"
},
{
"name":"intervalYearToMonth",
"type":"INTERVAL_YEAR_TO_MONTH"
}
],
"pk":[
"pkName1",
"pkName2"
]
},
"payload":{
"before":{
"data":{
"col1":3,
"col2":129,
"col3":2147483646,
"col4":9223372036854775806,
"col5":1.2222,
"col6":0.0000000000125,
"col7":10223372036854775806,
"col8":1,
"col9":"hello world",
"col10":"aGVsbG8gd29ybGQ=",
"col11":0.0000000000125,
"col12":"2020-11-25",
"col13":"00:01:02",
"col14":"2020-11-25 00:01:02",
"col15":"1606233662.012345",
"col16":"2020-11-25 00:01:02.012345 Asia/Shanghai",
"col17":"INTERVAL '3' DAY",
"col18":"INTERVAL '4' YEAR"
}
},
"after":null,
"op":"DELETE",
"timestamp":{
"eventTime":1647581072000,
"systemTime":1647581072976,
"checkpointTime":1647581072
},
"ddl":null,
"scn":"null"
},
"extend":{
"load_fm": "test"
}
}
A DDL operation example
ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT "test" COMMENT 'test';{ "version": "2.0", "schema": { "source": { "dbType": "ob_mysql", "dbVersion": null, "dbName": "connector_test", "schema": null, "table": "all_mysql_type_test" }, "column": null, "pk": null }, "payload": { "before": null, "after": null, "op": "ALTER", "timestamp": { "eventTime": 1671177209000, "systemTime": 1671177291485, "checkpointTime": 1671177200 }, "ddl": { "text": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'" }, "scn": "null" }, "extend": {} }
JSON message format of SharePlex
When data is synchronized to Kafka, DataHub (BLOB type), or RocketMQ, SharePlex uses the following JSON message format for serialization.
{
"data": { // Change data key-value pairs. For INSERT and DELETE operations, the values are complete. For UPDATE operations, only the changed values are included.
"col1": "val1"
},
"meta": {
"time": "YYYY-MM-DDTHH:mm:ss", // Change time
"op": "", // Change type, which can be ins, upd, del, or ddl.
"posttime": "YYYY-MM-DDTHH:mm:ss", // Time when the data is written to the target database
"idx": "STRING", // The index of the message in the transaction or the number of messages indexed in the transaction. This parameter is deprecated.
"size": "NUMBER", // The number of messages in the transaction. This parameter is deprecated.
"seq": "STRING", // The sequence number. This parameter is valid only if transactionEnabled is enabled on the source.
"table": "STRING", // The name of the SQL database table in the format {database}.{table}
"rowid": "STRING", // The name of the changed database table and the primary key value separated by \u0001.
"trans": "STRING", // The transaction ID
"scn": "STRING" // This field exists only in incremental scenarios where sequenceEnabled is set to true in the source.json file. By default, sequenceEnabled is set to true. It is used for sorting. The generation rule is a timestamp plus an incrementing sequence number of up to five digits in a synchronization process.
},
"key": { // This field exists only for UPDATE operations and indicates the value before the change.
},
"sql": {
"ddl": ""
} // This field exists only for DDL operations. If op is set to ddl, the DDL statement is written to this field.
}
Here are some examples of data:
Example of
INSERT(insert) data{ "data":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.999999999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999999308, "col17":10223372036854775806, "col18":"1606233662.012345" }, "meta":{ "posttime":"2020-12-07T13:22:00", "op":"ins", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3129", "trans":"shareplex_transaction_id", "scn":"123456789" } }Example of
UPDATE(update) data{ "data":{ "string":"hello world 2020" }, "meta":{ "posttime":"2020-12-07T13:59:09", "op":"upd", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3\u0001129", "trans":"shareplex_transaction_id", "scn":"123456789" }, "key":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.9999999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999999308, "col17":10223372036854775806, "col18":"1606233662.012345" } }Example of
DELETE(delete) data{ "data":{ "col1":"2020-11-25 00:01:02", "col2":"hello world", "col3":"INTERVAL '3' DAY", "col4":1.2222, "col5":9.9999999999308, "col6":129, "col7":"00:01:02", "col8":1, "col9":"2020-11-25 00:01:02.012345 Asia/Shanghai", "col10":2147483646, "col11":9223372036854775806, "col12":"aGVsbG8gd29ybGQ=", "col13":"INTERVAL '4' YEAR", "col14":3, "col15":"2020-11-25", "col16":9.9999999308, "col17":10223372036854775806, "col18":"1606233662.012345" }, "meta":{ "posttime":"2020-12-07T13:34:10", "op":"del", "size":10, "time":"2020-11-25T00:01:02", "idx":"1/10", "seq":1, "table":"mock_database.mock_table", "rowid":"mock_database.mock_table-3\u0001129", "trans":"shareplex_transaction_id", "scn":"123456789" } }Example of DDL data
ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT "test" COMMENT 'test';{ "data": {}, "meta": { "posttime": "2022-12-16T15:54:51", "op": "ddl", "size": 0, "time": "2022-12-16T15:53:29", "idx": "0/0", "seq": 0, "table": "connector_test.all_mysql_type_test", "rowid": "connector_test.all_mysql_type_test-", "trans": null, "scn": "null" }, "sql": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD COLUMN c90 VARCHAR(30) DEFAULT \"test\" comment 'test'" } }
JSON message format of DefaultExtendColumnType
When data is synchronized to Kafka, DataHub (BLOB type), and RocketMQ, the serialization method of DefaultExtendColumnType uses the following JSON message format.
The DefaultExtendColumnType JSON message format adds a field __light_type to the DEFAULT format, which is used to indicate the data type of the field.
{
"prevStruct": { // The image before the change.
},
"postStruct": { // The image after the change.
"__light_type": {
"col": { // The name of the field.
"schemaType": "type" // The type of the value.
}
}
},
"allMetaData": {
}
}
Here are some examples of data:
INSERT(insert) data{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:11039xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This is an example for OceanBase Database V4.x and later versions. For OceanBase Database earlier than V4.x, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": null, "recordType": "INSERT", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.99999999999999909326625337248, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999999909326625337248461995470488734032045693707225049338, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long":{ "schemaType":"TIMESTAMP" } } } }When
recordTypeisROW, it indicates that the data is fully transmitted, and the format is the same as that ofINSERT.{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": null, "transId":"1xxx_3152xxx", // This is an example for OceanBase Database V4.x and later versions. For OceanBase Database earlier than V4.x, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": null, "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": null }, "prevStruct": null, "recordType": "ROW", "postStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999, "col8": "hello world", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.99999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long":{ "schemaType":"TIMESTAMP" } } } }UPDATE(update) data{ "allMetaData": { "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:1103909342127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This is an example for OceanBase Database V4.x and later versions. For OceanBase Database earlier than V4.x, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType":"OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type": { "int8": { "schemaType": "TINYINT" }, "int16": { "schemaType": "SMALLINT" }, "int32": { "schemaType": "INT" }, "int64": { "schemaType": "INT64" }, "bigInt": { "schemaType": "BIGINT" }, "float32": { "schemaType": "FLOAT" }, "float64": { "schemaType": "DOUBLE" }, "string": { "schemaType": "VARCHAR" }, "bytes": { "schemaType": "BLOB" }, "decimal": { "schemaType": "DECIMAL" }, "localDate": { "schemaType": "DATE" }, "localTime": { "schemaType": "TIME" }, "localDateTime": { "schemaType": "DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } }, "recordType": "UPDATE", "postStruct": { "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type": { "int8": { "schemaType": "TINYINT" }, "int16": { "schemaType": "SMALLINT" }, "int32": { "schemaType": "INT" }, "int64": { "schemaType": "INT64" }, "bigInt": { "schemaType": "BIGINT" }, "float32": { "schemaType": "FLOAT" }, "float64": { "schemaType": "DOUBLE" }, "string": { "schemaType": "VARCHAR" }, "bytes": { "schemaType": "BLOB" }, "decimal": { "schemaType": "DECIMAL" }, "localDate": { "schemaType": "DATE" }, "localTime": { "schemaType": "TIME" }, "localDateTime": { "schemaType": "DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } } }DELETE(delete) data{ "allMetaData":{ "checkpoint": null, "record_primary_key": "int8\u0001int16", "source_identity": null, "uniqueId": "{tid:1103xxxx127, partition_id:0, part_cnt:0},5917,391,0", "transId":"1xxx_3152xxx", // This is an example for OceanBase Database V4.x and later versions. For OceanBase Database earlier than V4.x, the value is "transId": "{hash:123456, inc:1234, addr:\"xxx.xxx.xxx.xxx:2883\", t:123456}". "clusterId": "123456", "record_primary_value": "3\u0001129", "dbType": "OB_MYSQL", "table_name": "table", "db": "tenant.database", "timestamp": "1609344671" }, "prevStruct":{ "col1": 3, "col2": 129, "col3": 2147483646, "col4": 9223372036854775806, "col5": 10223372036854775806, "col6": 1.2222, "col7": 9.999999999999, "col8": "hello world 2020", "col9": "aGVsbG8gd29ybGQ=", "col10": 9.999999999999, "col11": "2020-11-25", "col12": "00:01:02", "col13": "2020-11-25 00:01:02", "col14": "1606233662.012345", "__light_type":{ "int8":{ "schemaType":"TINYINT" }, "int16":{ "schemaType":"SMALLINT" }, "int32":{ "schemaType":"INT" }, "int64":{ "schemaType":"INT64" }, "bigInt":{ "schemaType":"BIGINT" }, "float32":{ "schemaType":"FLOAT" }, "float64":{ "schemaType":"DOUBLE" }, "string":{ "schemaType":"VARCHAR" }, "bytes":{ "schemaType":"BLOB" }, "decimal":{ "schemaType":"DECIMAL" }, "localDate":{ "schemaType":"DATE" }, "localTime":{ "schemaType":"TIME" }, "localDateTime":{ "schemaType":"DATETIME" }, "timestamp_in_long": { "schemaType": "TIMESTAMP" } } }, "recordType":"DELETE", "postStruct":null }DDL example
ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT "test" COMMENT 'test';{ "prevStruct": null, "postStruct": { "ddl": "ALTER TABLE connector_test.all_mysql_type_test ADD column c90 VARCHAR(30) DEFAULT \"test\" COMMENT 'test'", "__light_type": { "ddl": { "schemaType": "VAR_STRING" } } }, "allMetaData": { "checkpoint": "1671177200", "dbType": "OB_MYSQL", "storeDataSequence": null, "db": "connector_test", "timestamp": "1671177209", "uniqueId": null, "ddlType": "ALTER_TABLE", "record_primary_key": null, "source_identity": null, "record_primary_value": null, "table_name": "all_mysql_type_test" }, "recordType": "DDL" }
JSON message format of Debezium
When you synchronize data from OceanBase Database in MySQL compatible mode to Kafka, DataHub (BLOB type), and RocketMQ, Debezium uses the following JSON message format. The format contains two types. By default, only the structure in the payload field is displayed.
The
schemaandpayloadfields exist.{ "schema": { //The structure that describes the payload fields. This structure is not displayed by default. "type": "struct", //The struct field indicates that the field contains a structure. "optional": false, //Indicates whether the field is required. "fields": [ { "type": "int64", //The type of the field. "optional": false, //Indicates whether the field is required. "field": "ts_ms" //The name of the field. } ] }, "payload": { "op": "c", //The data modification type. Valid values: c (full or insert), u (update), d (delete), and HEARTBEAT (heartbeat message). "source": { "version": "", //The version of OMS. "connector": "OB_MYSQL", //The type of the data source. "name": "OMS", //The fixed value OMS. "ts_ms": 0, //The second-level timestamp of data change. This field exists only in incremental data. "db": "test", //The name of the database whose data is changed by using an SQL statement. If the database is OceanBase Database, only the database name is displayed, and the tenant name is not displayed. "table": "testTab", //The name of the table whose data is changed by using an SQL statement. "pos": "553132@1668496109" //The position in the binlog file [binlog file name]@[binlog file name offset]. }, "before": { //The image before the data change. "column": "value" //The key-value pair that contains the full key-value. }, "after": { //The image after the data change. "column": "value" //The key-value pair that contains the full key-value. }, "ts_ms": 1668497367188 //The timestamp of data processing. } }Only the
payloadfield exists.{ "payload": { "op": "c", //The data modification type. Valid values: c (full or insert), u (update), d (delete), and HEARTBEAT (heartbeat message). "source": { "version": "", //The version of OMS. "connector": "OB_MYSQL", //The type of the data source. "name": "OMS", //The fixed value OMS. "ts_ms": 0, //The second-level timestamp of data change. This field exists only in incremental data. "db": "test", //The name of the database whose data is changed by using an SQL statement. If the database is OceanBase Database, only the database name is displayed, and the tenant name is not displayed. "table": "testTab", //The name of the table whose data is changed by using an SQL statement. "pos": "553132@16684****" //The position in the binlog file [binlog file name]@[binlog file name offset]. }, "before": { //The image before the data change. "column": "value" //The key-value pair that contains the full key-value. }, "after": { //The image after the data change. "column": "value" //The key-value pair that contains the full key-value. }, "ts_ms": 1668497367188 //The timestamp of data processing. } }
Here are some examples:
Example of
INSERT(insert) data{ "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"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":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"c", "source":{ "connector":"OB_MYSQL", "pos":"703223@166849****", "name":"OMS", "version":"", "ts_ms":1668491621000, "db":"test", "table":"table_name" }, "after":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":2, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495423594 } }Example of
UPDATE(update) data{ "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"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":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"u", "before":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"6900000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"436999@166849****", "name":"OMS", "version":"", "ts_ms":1668495861000, "db":"test", "table":"table_name" }, "after":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495906356 } }Example of
DELETE(delete) data{ "schema":{ "optional":false, "type":"STRUCT", "fields":[ { "field":"before", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"after", "optional":false, "type":"struct", "fields":[ { "field":"c01", "optional":false, "type":"int32" }, { "field":"c02", "optional":false, "type":"string" }, { "field":"c03", "optional":false, "type":"string" }, { "field":"c04", "optional":false, "type":"bytes" }, { "field":"c05", "optional":false, "type":"int16" }, { "field":"c06", "optional":false, "type":"int16" }, { "field":"c07", "optional":false, "type":"int32" }, { "field":"c08", "optional":false, "type":"int64" }, { "field":"c09", "optional":false, "type":"float64" }, { "field":"c10", "optional":false, "type":"float64" }, { "field":"c11", "optional":false, "type":"string" }, { "field":"c12", "optional":false, "type":"string" }, { "field":"c13", "optional":false, "type":"string" }, { "field":"c14", "optional":false, "type":"string" }, { "field":"c15", "optional":false, "type":"bytes" }, { "field":"c16", "optional":false, "type":"string" }, { "field":"c17", "optional":false, "type":"bytes" }, { "field":"c18", "optional":false, "type":"bytes" }, { "field":"c19", "optional":false, "type":"bytes" }, { "field":"c20", "optional":false, "type":"bytes" }, { "field":"c21", "optional":false, "type":"string" }, { "field":"c22", "optional":false, "type":"int32" }, { "field":"c23", "optional":false, "type":"int64" }, { "field":"c24", "optional":false, "type":"string" }, { "field":"c25", "optional":false, "type":"int32" }, { "field":"c26", "optional":false, "type":"bytes" } ] }, { "field":"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":"db", "optional":false, "type":"string" }, { "field":"table", "optional":false, "type":"string" }, { "field":"server_id", "optional":false, "type":"int64" }, { "field":"pos", "optional":false, "type":"string" } ] }, { "field":"op", "optional":false, "type":"string" }, { "field":"ts_ms", "optional":false, "type":"int64" } ] }, "payload":{ "op":"d", "before":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"553132@1668****", "name":"OMS", "version":"", "ts_ms":1668496109000, "db":"test", "table":"table_name" }, "ts_ms":1668496119717 } }
JSON message format of DebeziumFlatten
When you synchronize data from the MySQL compatible mode of OceanBase Database to Kafka, DataHub (BLOB type), and RocketMQ, the JSON message format of the DebeziumFlatten serialization method is as follows. Compared with the Debezium serialization method, the schema and payload fields are not filled in.
{
"op": "c", //The data modification type. Valid values: c (full or insert), u (update), d (delete), and HEARTBEAT (heartbeat message).
"source": {
"version": "", //The version of OMS.
"connector": "OB_MYSQL", //The type of the data source.
"name": "OMS", //The fixed value OMS.
"ts_ms": 0, //The second-level timestamp of data modification. This field exists only in incremental data.
"db": "test", //The name of the database to which the SQL statement is applied. If the database is OceanBase Database, only the database name is provided, and the tenant name is not required.
"table": "testTab", //The name of the table to which the SQL statement is applied.
"pos": "553132@16684****" //The position in the binlog file [binlog file name]@[binlog file name offset]
},
"before": { //The image before the modification.
"column": "value" //A key-value pair that contains the full key-value.
},
"after": { //The image after the modification.
"column": "value" //A key-value pair that contains the full key-value.
},
"ts_ms": 1668497367188 //The timestamp of data processing.
}
Data examples:
Example of
INSERT(insert) data{ "op":"c", "source":{ "connector":"OB_MYSQL", "pos":"703223@166849****", "name":"OMS", "version":"", "ts_ms":1668491621000, "db":"test", "table":"table_name" }, "after":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":2, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495423594 }Example of
UPDATE(update) data{ "op":"u", "before":{ "c11":"a", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"690000000000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"436999@166849****", "name":"OMS", "version":"", "ts_ms":1668495861000, "db":"test", "table":"table_name" }, "after":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "ts_ms":1668495906356 }Example of
DELETE(delete) data{ "op":"d", "before":{ "c11":"aa", "c10":2.4212412, "c13":"c", "c12":"b", "c15":"65", "c14":"d", "c17":"67", "c16":"f", "c19":"69000000000000000000000000000", "c18":"68", "c20":"6A", "c22":19311, "c21":"2022-11-15T05:12:11Z", "c02":"12312", "c24":1668489131000, "c01":1, "c23":36060000000, "c04":"61", "c26":"6B", "c03":"1241.41000", "c25":2022, "c06":141, "c05":11, "c08":412124124, "c07":4241, "c09":2.11111 }, "source":{ "connector":"OB_MYSQL", "pos":"553132@1668****", "name":"OMS", "version":"", "ts_ms":1668496109000, "db":"test", "table":"table_name" }, "ts_ms":1668496119717 }
JSON message format of DebeziumSmt
DebeziumSmt is a configuration method provided by Debezium. It uses Single Message Transform (SMT) to convert and process single messages. When you synchronize data from the MySQL compatible mode of OceanBase Database to Kafka, DataHub (BLOB type), and RocketMQ, the JSON message format of the DebeziumSmt serialization method only displays the key:value in the after field.
For example, you can use the Debezium serialization method to update data:
{
"op": "u",
"source": {
"connector": "OB_MYSQL",
"name": "OMS"
},
"ts_ms": 1668496119717,
"before": {
"field1": "before_value1",
"field2": "before_value2"
},
"after": {
"field1": "after_value1",
"field2": "after_value2"
}
}
After SMT processes the message in the preceding example, the message format is simplified. That is, when you use the DebeziumSmt serialization method, the JSON message format is as follows:
{
"field1": "after_value1",
"field2": "after_value2"
}
Data examples:
Example of
INSERT(insert) data{ "field1": "after_value1", "field2": "after_value2", "__deleted": "false" }Example of
UPDATE(update) data{ "field1": "after_value1", "field2": "after_value2", "__deleted": "false" }Example of
DELETE(delete) data{ "field1": "after_value1", "field2": "after_value2", "__deleted": "true" }
JSON message format of Avro
When you synchronize data from OceanBase Database in MySQL compatible mode to Apache Kafka, the Avro serialization method uses the following JSON message format.
Full migration
{ "version": 1, "id": 0, "sourceTimestamp": 1702371565, // The safe source position. "sourcePosition": "", // Full migration does not contain position information. "safeSourcePosition": "", "sourceTxid": "", "source": { "sourceType": "MySQL", // The value is fixed to MySQL. "version": "OBMySQL" // The value is fixed to OBMySQL. }, "operation": "INIT", // The full migration type is INIT. "objectName": "test***", "processTimestamps": [ 1702371565238 ], // The timestamp of the delivery. "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" // Only the primary key type is supported. }, "fields": [ { "name": "id", "dataTypeNumber": 246 }, // The type of each column. { "name": "bid", "dataTypeNumber": 3 }, { "name": "name", "dataTypeNumber": 15 }, { "name": "address", "dataTypeNumber": 254 } ], "beforeImages": null, // The before image of full migration is empty. "afterImages": [ // The after image. The precision of the INTEGER type is 8, and the precision and scale of the FLOAT type are 8 and 64, respectively. { "value": "1", "precision": 1, "scale": 0 }, { "precision": 8, "value": "11" }, { "charset": "utf8mb4", "value": { "bytes": "yyy" } }, null ] }Incremental synchronization of DML operations
Example of
INSERT(insertion) data{ "version": 1, "id": 170236922143600000, "sourceTimestamp": 1702369092, "sourcePosition": "1702369080", // The checkpoint of OceanBase Database in MySQL compatible mode. "safeSourcePosition": "1702369080", // The checkpoint of OceanBase Database in MySQL compatible mode. "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "INSERT", "objectName": "test***", "processTimestamps": [1702369221480], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": null, // The before image of INSERT is empty. "afterImages": [ {"precision": 8, "value": "2"}, {"precision": 8, "value": "12"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"} } ] }Example of
UPDATE(update) data{ "version": 1, "id": 170236975822100001, "sourceTimestamp": 1702369757, "sourcePosition": "1702369756", "safeSourcePosition": "1702369756", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "UPDATE", "objectName": "test***", "processTimestamps": [1702369758237], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": [ // The before image and after image of UPDATE exist. {"precision": 8, "value": "3"}, {"precision": 8, "value": "22"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ], "afterImages": [ {"precision": 8, "value": "3"}, {"precision": 8, "value": "44"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ] }Example of
DELETE(delete) data{ "version": 1, "id": 170236976527500000, "sourceTimestamp": 1702369764, "sourcePosition": "1702369763", "safeSourcePosition": "1702369763", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "DELETE", "objectName": "test***", "processTimestamps": [1702369765287], "tags": { "pk_uk_info": "{\"PRIMARY\":[\"id\"]}" }, "fields": [ {"name": "id", "dataTypeNumber": 8}, {"name": "bid", "dataTypeNumber": 3}, {"name": "name", "dataTypeNumber": 15} ], "beforeImages": [ {"precision": 8, "value": "3"}, {"precision": 8, "value": "44"}, {"charset": "utf8mb4", "value": {"bytes": "xxx"}} ], "afterImages": null // The after image of DELETE is empty. }
Incremental synchronization of DDL operations
{ "version": 1, "id": 170236979372400000, "sourceTimestamp": 1702369793, "sourcePosition": "1702369792", "safeSourcePosition": "1702369792", "sourceTxid": "", "source": { "sourceType": "MySQL", "version": "OBMySQL" }, "operation": "DDL", "objectName": "test***", "processTimestamps": [1702369794543], "tags": {}, "fields": null, // Incremental synchronization of DDL operations does not contain fields and beforeImages. "beforeImages": null, "afterImages": "alter table multi_db_multi_tbl add column address char(20) default null" // The afterImages of the STRING type is the DDL statement. }
Format description of data transmission to the text protocol
When you synchronize data from an OceanBase database to Kafka, DataHub (BLOB type), or RocketMQ
If the serialization method is Default, Canal, DataWorks (V2.0), SharePlex, or DefaultExtendColumnType, the mapping of the two tenants of the OceanBase database is as follows.
OceanBase Database in MySQL compatible mode
Data typeMapped typeDescriptionTINYINT
SMALLINT
MEDIUMINT
INT
INTEGER
YEAR
BOOL
BOOLEANLong Integer with a bit length of 64 or less.
Normal values, such as 1000, are not displayed in scientific notation.
For BOOL/BOOLEAN, true = 1 and false = 0.DECIMAL
NUMERICBigDecimal Exact decimal values and integers with a bit length exceeding 64. For integers, decimal points and decimals are not displayed. For decimal values, the number of digits is displayed based on the data from the database, and trailing zeros are not removed. Scientific notation is used. FLOAT
DOUBLEDouble Floating-point numbers
The number of significant digits is determined by the source data type, which is FLOAT or DOUBLE. FLOAT has 7 significant digits, and DOUBLE has 16 significant digits.CHAR
VARCHAR
TINYTEXT
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
SETString String. TINYBLOB
BLOB
MEDIUMBLOB
LONGBLOB
BINARY
VARBINARY
BITBytes Byte array. By default, it is displayed in BASE64 encoding.
Description For BIT, the leading zeros are removed after incremental data is received, but they are not removed during full data synchronization. Therefore, the BASE64 encoding may be inconsistent. However, the actual results are consistent, and the decoding results are consistent.DATE Date Date type, in the YYYY-MM-DDformat. If the time is invalid, the original string is displayed.TIME Time Time type, in the HH:mm:ss[.nnnnnnnnn]format.
For time values with a precision less than seconds, up to 9 digits are displayed. For time values with a precision less than seconds, all non-zero digits are displayed. If the time is invalid, the original string is displayed.DATETIME DateTime Date and time type, including time zone. The format is YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId].
For time values with a precision less than seconds, up to 9 digits are displayed. For time values with a precision less than seconds, all non-zero digits are displayed. If the time is invalid, the original string is displayed.TIMESTAMP Timestamp Timestamp type, in the [seconds timestamp][.nnnnnnnnn]format.
For time values with a precision less than seconds, up to 9 digits are displayed. For time values with a precision less than seconds, all non-zero digits are displayed. If the time is invalid, it is displayed in the0000-00-00 00:00:00format.OceanBase Database in Oracle compatible mode
Data typeMapped typeDescriptionINTEGER Long Integer with a bit length of 64 or less.
Normal values, such as 1000, are not displayed in scientific notation.NUMBER
FLOATBigDecimal Exact decimal values and integers with a bit length exceeding 64. BINARY_FLOAT BINARY_DOUBLE Double Floating-point numbers
The number of significant digits is determined by the source data type, which is FLOAT or DOUBLE. FLOAT has 7 significant digits, and DOUBLE has 16 significant digits.VARCHAR2
NVARCHAR2
INTERVAL YEAR TO MOTH
INTERVAL DAY TO SECOND
CLOB
NCLOB
ROWID
UROWIDString String. BLOB
BFILE
RAWBytes Byte array
By default, it is displayed in BASE64 encoding.DATE
TIMESTAMP
TIMESTAMP WITH TIME ZONE
TIMESTAMP WITH LOCAL TIME ZONEDateTime Date and time type, including time zone. The format is YYYY-MM-DD HH:mm:ss[.nnnnnnnnn] [zoneId].
For time values with a precision less than seconds, up to 9 digits are displayed. For time values with a precision less than seconds, all non-zero digits are displayed. If the time is invalid, the original string is displayed.
If the serialization method is Debezium, the mapping of OceanBase Database in MySQL compatible mode is as follows.
Notice
When you synchronize data from an OceanBase database in Oracle compatible mode to Kafka, DataHub (BLOB type), or RocketMQ, you cannot select the Debezium serialization method.
Data typeMapped typeDescriptionBOOLEAN
BOOLBOOLEAN The value can be true or false. TINYINT
SMALLINT
MEDIUMINT
INT/INTEGER
BIGINT
YEARLONG Integer in the range of -2^63^ to 2^63^, excluding 2^63^ and -2^63^ . BIGINT STRING The data is displayed in the string format. FLOAT
DOUBLEDOUBLE Floating-point number. DECIMAL
NUMERICSTRING The data is displayed in the string format. For decimal values, the number of digits is displayed based on the data from the database, and trailing zeros are not removed. Scientific notation is used. BIT
BINARY
VARBINARY
TINYBLOB
BLOB
MEDIUMBLOB
LONGBLOBBYTES Byte array, encoded in base16. CHAR
VARCHAR
TINYTEXT
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
SETSTRING String. TIMESTAMP STRING The format is YYYY-MM-DDTHH:mm:ss[.nnnnnnnnn]Z, and the time zone is UTC. DATE LONG The number of days elapsed since January 1, 1970. TIME LONG The number of microseconds elapsed since 00:00:00, without time zone information. DATETIME LONG The number of milliseconds elapsed since January 1, 1970 00:00:00, without time zone information. If the serialization method is Avro, the mapping of OceanBase Database in MySQL compatible mode is as follows.
Notice
When you synchronize data from an OceanBase database in MySQL compatible mode to Kafka, you can select the Avro serialization method.
Type nameMapped typeTINYINT
BOOLEAN
SMALLINT
MEDIUMINT
INT
BIGINT
BITINTEGER FLOAT
DOUBLEFLOAT DECIMAL
NUMERICDECIMAL VARCHAR
CHAR
TINYTEXT
MEDIUMTEXT
LONGTEXT
TEXTCHARACTER BINARY
VARBINARY
TINYBLOB
MEDIUMBLOB
LONGBLOB
BLOBBinaryObject TIMESTAMP TimestampObject
Description For TIMESTAMP type, full data and incremental data are converted to timestamps. The timestamp of an invalid time is-9223372022400L.
Except for invalid timestamps, you can use theInstant.ofEpochSecond(ts, nanos)method in Java to obtain the correct wall time.DATE
TIME
DATETIME
YEARDATETIME JSON
ENUM
SETTextObject GEOMETRY TextGeometry
Description Data transmission uses the EWKT format, so the type is mapped to TextGeometry.