OceanBase CDC Connector

The OceanBase CDC connector allows for reading snapshot data and incremental data from OceanBase. This document describes how to set up the OceanBase CDC connector to run SQL queries against OceanBase.

OceanBase CDC Solutions

Glossary:

• OceanBase CE: OceanBase Community Edition. It's compatible with MySQL and has been open sourced at https://github.com/oceanbase/oceanbase.
• OceanBase EE: OceanBase Enterprise Edition. It supports two compatibility modes: MySQL and Oracle. See https://en.oceanbase.com.
• OceanBase Cloud: OceanBase Enterprise Edition on Cloud. See https://en.oceanbase.com/product/cloud.
• Binlog Service CE: OceanBase Binlog Service Community Edition. It is a solution of OceanBase CE that is compatible with the MySQL replication protocol. See the docs of OceanBase Binlog Service for details.
• Binlog Service EE: OceanBase Binlog Service Enterprise Edition. It is a solution of OceanBase EE MySQL compatible mode that is compatible with the MySQL replication protocol, and it's only available for users of Alibaba Cloud, see User Guide.
• MySQL Driver: mysql-connector-java which can be used with OceanBase CE and OceanBase EE MySQL compatible mode.
• OceanBase Driver: The JDBC driver for OceanBase, which supports both MySQL compatible mode and Oracle compatible mode of all OceanBase versions. It's open sourced at https://github.com/oceanbase/obconnector-j.

CDC Source Solutions for OceanBase:

Database	Supported Driver	CDC Source Connector	Other Required Components
OceanBase CE	MySQL Driver: 8.0.x	OceanBase CDC Connector	Binlog Service CE
	MySQL Driver: 8.0.x	MySQL CDC Connector	Binlog Service CE
OceanBase EE (MySQL Compatible Mode)	MySQL Driver: 8.0.x	OceanBase CDC Connector	Binlog Service EE
	MySQL Driver: 8.0.x	MySQL CDC Connector	Binlog Service EE
OceanBase EE (Oracle Compatible Mode)		The incremental data subscription service for OceanBase Oracle Compatibility Mode is not currently supported. Please contact Enterprise Support for assistance.

Important Notes: The oceanbase-cdc connector has undergone significant changes starting from version 3.5:

• The previous connector implementation based on the OceanBase Log Proxy service has been officially removed. The current version now only supports connecting to the OceanBase Binlog Service.
• The current oceanbase-cdc connector is rebuilt based on the mysql-cdc connector, offering enhanced compatibility with the OceanBase Binlog Service, including critical bug fixes. Upgrading to this version is strongly recommended.
• Since the OceanBase Binlog Service is compatible with MySQL replication protocols, you may still use the MySQL CDC connector to connect to the OceanBase Binlog Service.
• The incremental data subscription service for OceanBase Oracle Compatibility Mode is not currently supported. Please contact Enterprise Support for assistance.

Dependencies

In order to set up the OceanBase CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

Maven dependency

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-oceanbase-cdc</artifactId>
    <version>3.8-SNAPSHOT</version>
</dependency>

SQL Client JAR

Download flink-sql-connector-oceanbase-cdc and put it under <FLINK_HOME>/lib/.

Note: Refer to flink-sql-connector-oceanbase-cdc, more released versions will be available in the Maven central warehouse.

Since the licenses of MySQL Driver and OceanBase Driver are incompatible with Flink CDC project, we can't provide them in prebuilt connector jar packages. You may need to configure the following dependencies manually.

Dependency Item	Description
mysql:mysql-connector-java:8.0.27	Used for connecting to MySQL tenant of OceanBase.

Deploy OceanBase database and Binlog service

1. Deploy the OceanBase database according to the document.
2. Deploy the OceanBase Binlog service according to the document.

Usage Documentation

Notes:

• The current version of the oceanbase-cdc connector is built on top of the mysql-cdc connector, with only minor internal implementation changes.
  • External interfaces and configuration parameters are fully consistent with those of the mysql-cdc connector.
  • For detailed usage instructions, refer to the MySQL CDC Usage Documentation.

How to create a OceanBase CDC table

-- checkpoint every 3000 milliseconds                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   

-- register a OceanBase table 'orders' in Flink SQL
Flink SQL> CREATE TABLE orders (
     order_id INT,
     order_date TIMESTAMP(0),
     customer_name STRING,
     price DECIMAL(10, 5),
     product_id INT,
     order_status BOOLEAN,
     PRIMARY KEY(order_id) NOT ENFORCED
     ) WITH (
     'connector' = 'oceanbase-cdc',
     'hostname' = 'localhost',
     'port' = '2881',
     'username' = 'root',
     'password' = '123456',
     'database-name' = 'mydb',
     'table-name' = 'orders');
  
-- read snapshot and binlogs from orders table
Flink SQL> SELECT * FROM orders;

Connector Options

Option	Required	Default	Type	Description
connector	required	(none)	String	Specify what connector to use, here should be 'oceanbase-cdc'.
hostname	required	(none)	String	IP address or hostname of the OceanBase database server.
username	required	(none)	String	Name of the OceanBase database to use when connecting to the OceanBase database server.
password	required	(none)	String	Password to use when connecting to the OceanBase database server.
database-name	required	(none)	String	Database name of the OceanBase server to monitor. The database-name also supports regular expressions to monitor multiple tables matches the regular expression.
table-name	required	(none)	String	Table name of the OceanBase database to monitor. The table-name also supports regular expressions to monitor multiple tables that satisfy the regular expressions. Note: When the OceanBase CDC connector regularly matches the table name, it will concat the database-name and table-name filled in by the user through the string `\\.` to form a full-path regular expression, and then use the regular expression to match the fully qualified name of the table in the OceanBase database.
port	optional	2881	Integer	Integer port number of the OceanBase database server.
server-id	optional	(none)	String	A numeric ID or a numeric ID range of this database client, The numeric ID syntax is like '5400', the numeric ID range syntax is like '5400-5408', The numeric ID range syntax is recommended when 'scan.incremental.snapshot.enabled' enabled. Every ID must be unique across all currently-running database processes in the OceanBase cluster. This connector joins the OceanBase cluster as another server (with this unique ID) so it can read the binlog. By default, a random number is generated between 5400 and 6400, though we recommend setting an explicit value.
scan.incremental.snapshot.enabled	optional	true	Boolean	Incremental snapshot is a new mechanism to read snapshot of a table. Compared to the old snapshot mechanism, the incremental snapshot has many advantages, including: (1) source can be parallel during snapshot reading, (2) source can perform checkpoints in the chunk granularity during snapshot reading, (3) source doesn't need to acquire global read lock (FLUSH TABLES WITH READ LOCK) before snapshot reading. If you would like the source to run in parallel, each parallel reader should have a unique server id, so the 'server-id' must be a range like '5400-6400', and the range must be larger than the parallelism. Please see Incremental Snapshot Readingsection for more detailed information.
scan.incremental.snapshot.chunk.size	optional	8096	Integer	The chunk size (number of rows) of table snapshot, captured tables are split into multiple chunks when read the snapshot of table.
scan.snapshot.fetch.size	optional	1024	Integer	The maximum fetch size for per poll when read table snapshot.
scan.incremental.snapshot.chunk.key-column	optional	(none)	String	The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table. By default, the chunk key is the first column of the primary key. A column that is not part of the primary key can be used as a chunk key, but this may lead to slower query performance. Warning: Using a non-primary key column as a chunk key may lead to data inconsistencies. Please see Warning for details.
scan.startup.mode	optional	initial	String	Optional startup mode for OceanBase CDC consumer, valid enumerations are "initial", "earliest-offset", "latest-offset", "specific-offset", "timestamp" and "snapshot". Please see Startup Reading Position section for more detailed information.
scan.startup.specific-offset.file	optional	(none)	String	Optional binlog file name used in case of "specific-offset" startup mode
scan.startup.specific-offset.pos	optional	(none)	Long	Optional binlog file position used in case of "specific-offset" startup mode
scan.startup.specific-offset.gtid-set	optional	(none)	String	Optional GTID set used in case of "specific-offset" startup mode
scan.startup.timestamp-millis	optional	(none)	Long	Optional millisecond timestamp used in case of "timestamp" startup mode.
scan.startup.specific-offset.skip-events	optional	(none)	Long	Optional number of events to skip after the specific starting offset
scan.startup.specific-offset.skip-rows	optional	(none)	Long	Optional number of rows to skip after the specific starting offset
server-time-zone	optional	(none)	String	The session time zone in database server, e.g. "Asia/Shanghai". It controls how the TIMESTAMP type in OceanBase converted to STRING. See more here. If not set, then ZoneId.systemDefault() is used to determine the server time zone.
debezium.min.row.count.to.stream.result	optional	1000	Integer	During a snapshot operation, the connector will query each included table to produce a read event for all rows in that table. This parameter determines whether the OceanBase connection will pull all results for a table into memory (which is fast but requires large amounts of memory), or whether the results will instead be streamed (can be slower, but will work for very large tables). The value specifies the minimum number of rows a table must contain before the connector will stream results, and defaults to 1,000. Set this parameter to '0' to skip all table size checks and always stream all results during a snapshot.
connect.timeout	optional	30s	Duration	The maximum time that the connector should wait after trying to connect to the OceanBase database server before timing out. This value cannot be less than 250ms.
connect.max-retries	optional	3	Integer	The max retry times that the connector should retry to build OceanBase database server connection.
connection.pool.size	optional	20	Integer	The connection pool size.
jdbc.properties.*	optional	20	String	Option to pass custom JDBC URL properties. User can pass custom properties like 'jdbc.properties.useSSL' = 'false'.
heartbeat.interval	optional	30s	Duration	The interval of sending heartbeat event for tracing the latest available binlog offsets.
debezium.*	optional	(none)	String	Pass-through Debezium's properties to Debezium Embedded Engine which is used to capture data changes from OceanBase server. For example: 'debezium.snapshot.mode' = 'never'. See more about the Debezium's MySQL Connector properties
scan.incremental.close-idle-reader.enabled	optional	false	Boolean	Whether to close idle readers at the end of the snapshot phase. The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true. If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true, so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
debezium.binary.handling.mode	optional	(none)	String	debezium.binary.handling.mode can be set to one of the following values: none: No processing is performed, and the binary data type is transmitted as a byte array (byte array). base64: The binary data type is converted to a Base64-encoded string and transmitted. hex: The binary data type is converted to a hexadecimal string and transmitted. The default value is none. Depending on your requirements and data types, you can choose the appropriate processing mode. If your database contains a large number of binary data types, it is recommended to use base64 or hex mode to make it easier to handle during transmission.
scan.parse.online.schema.changes.enabled	optional	false	Boolean	Whether to parse "online" schema changes generated by gh-ost or pt-osc. Schema change events are applied to a "shadow" table and then swapped with the original table later. This is an experimental feature, and subject to change in the future.
use.legacy.json.format	optional	true	Boolean	Whether to use legacy JSON format to cast JSON type data in binlog. It determines whether to use the legacy JSON format when retrieving JSON type data in binlog. If the user configures 'use.legacy.json.format' = 'true', whitespace before values and after commas in the JSON type data is removed. For example, JSON type data {"key1": "value1", "key2": "value2"} in binlog would be converted to {"key1":"value1","key2":"value2"}. When 'use.legacy.json.format' = 'false', the data would be converted to {"key1": "value1", "key2": "value2"}, with whitespace before values and after commas preserved.
scan.incremental.snapshot.unbounded-chunk-first.enabled	optional	false	Boolean	Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Experimental option, defaults to false.
scan.read-changelog-as-append-only.enabled	optional	false	Boolean	Whether to convert the changelog stream to an append-only stream. This feature is only used in special scenarios where you need to save upstream table deletion messages. For example, in a logical deletion scenario, users are not allowed to physically delete downstream messages. In this case, this feature is used in conjunction with the row_kind metadata field. Therefore, the downstream can save all detailed data at first, and then use the row_kind field to determine whether to perform logical deletion. The option values are as follows: true: All types of messages (including INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER) will be converted into INSERT messages. false (default): All types of messages are sent as is.
scan.incremental.snapshot.backfill.skip	optional	false	Boolean	Whether to skip backfill in snapshot reading phase. If backfill is skipped, changes on captured tables during snapshot phase will be consumed later in change log reading phase instead of being merged into the snapshot. WARNING: Skipping backfill might lead to data inconsistency because some change log events happened within the snapshot phase might be replayed (only at-least-once semantic is promised). For example updating an already updated value in snapshot, or deleting an already deleted entry in snapshot. These replayed change log events should be handled specially.

Available Metadata

The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.

Key	DataType	Description
table_name	STRING NOT NULL	Name of the table that contain the row.
database_name	STRING NOT NULL	Name of the database that contain the row.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	It indicates the time that the change was made in the database. If the record is read from snapshot of the table instead of the binlog, the value is always 0.
row_kind	STRING NOT NULL	It indicates the row kind of the changelog,Note: The downstream SQL operator may fail to compare due to this new added column when processing the row retraction if the source operator chooses to output the 'row_kind' column for each record. It is recommended to use this metadata column only in simple synchronization jobs. '+I' means INSERT message, '-D' means DELETE message, '-U' means UPDATE_BEFORE message and '+U' means UPDATE_AFTER message.

The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:

CREATE TABLE products
(
    db_name       STRING METADATA FROM 'database_name' VIRTUAL,
    table_name    STRING METADATA FROM 'table_name' VIRTUAL,
    operation_ts  TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    operation     STRING METADATA FROM 'row_kind' VIRTUAL,
    order_id      INT,
    order_date    TIMESTAMP(0),
    customer_name STRING,
    price         DECIMAL(10, 5),
    product_id    INT,
    order_status  BOOLEAN,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
      'connector' = 'oceanbase-cdc',
      'hostname' = 'localhost',
      'port' = '2881',
      'username' = 'root',
      'password' = '123456',
      'database-name' = 'mydb',
      'table-name' = 'orders'
      );

The extended CREATE TABLE example demonstrates the usage of regex to match multi-tables:

CREATE TABLE products
(
    db_name       STRING METADATA FROM 'database_name' VIRTUAL,
    table_name    STRING METADATA FROM 'table_name' VIRTUAL,
    operation_ts  TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    operation     STRING METADATA FROM 'row_kind' VIRTUAL,
    order_id      INT,
    order_date    TIMESTAMP(0),
    customer_name STRING,
    price         DECIMAL(10, 5),
    product_id    INT,
    order_status  BOOLEAN,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
      'connector' = 'oceanbase-cdc',
      'hostname' = 'localhost',
      'port' = '2881',
      'username' = 'root',
      'password' = '123456',
      'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})',
      'table-name' = '(t[5-8]|tt)'
      );

example	expression	description
prefix match	^(test).*	This matches the database name or table name starts with prefix of test, e.g test1、test2.
suffix match	.*[p$]	This matches the database name or table name ends with suffix of p, e.g cdcp、edcp.
specific match	txc	This matches the database name or table name according to a specific name, e.g txc.

It will use database-name\\.table-name as a pattern to match tables, as above examples using pattern (^(test).*|^(tpc).*|txc|.*[p$]|t{2})\\.(t[5-8]|tt) matches txc.tt、test2.test5.

Data Type Mapping

OceanBase type	Flink SQL type	NOTE
TINYINT	TINYINT
SMALLINT TINYINT UNSIGNED TINYINT UNSIGNED ZEROFILL	SMALLINT
INT MEDIUMINT SMALLINT UNSIGNED SMALLINT UNSIGNED ZEROFILL	INT
BIGINT INT UNSIGNED INT UNSIGNED ZEROFILL MEDIUMINT UNSIGNED MEDIUMINT UNSIGNED ZEROFILL	BIGINT
BIGINT UNSIGNED BIGINT UNSIGNED ZEROFILL SERIAL	DECIMAL(20, 0)
FLOAT FLOAT UNSIGNED FLOAT UNSIGNED ZEROFILL	FLOAT
REAL REAL UNSIGNED REAL UNSIGNED ZEROFILL DOUBLE DOUBLE UNSIGNED DOUBLE UNSIGNED ZEROFILL DOUBLE PRECISION DOUBLE PRECISION UNSIGNED DOUBLE PRECISION UNSIGNED ZEROFILL	DOUBLE
NUMERIC(p, s) NUMERIC(p, s) UNSIGNED NUMERIC(p, s) UNSIGNED ZEROFILL DECIMAL(p, s) DECIMAL(p, s) UNSIGNED DECIMAL(p, s) UNSIGNED ZEROFILL FIXED(p, s) FIXED(p, s) UNSIGNED FIXED(p, s) UNSIGNED ZEROFILL where p <= 38	DECIMAL(p, s)
NUMERIC(p, s) NUMERIC(p, s) UNSIGNED NUMERIC(p, s) UNSIGNED ZEROFILL DECIMAL(p, s) DECIMAL(p, s) UNSIGNED DECIMAL(p, s) UNSIGNED ZEROFILL FIXED(p, s) FIXED(p, s) UNSIGNED FIXED(p, s) UNSIGNED ZEROFILL where 38 < p <= 65	STRING	The precision for DECIMAL data type is up to 65 in OceanBase compatible mode, but the precision for DECIMAL is limited to 38 in Flink. So if you define a decimal column whose precision is greater than 38, you should map it to STRING to avoid precision loss.
BOOLEAN TINYINT(1) BIT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)]
TIMESTAMP [(p)] DATETIME [(p)]	TIMESTAMP [(p)]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈(n + 7) / 8⌉)
BINARY(n)	BINARY(n)
VARBINARY(N)	VARBINARY(N)
TINYTEXT TEXT MEDIUMTEXT LONGTEXT	STRING
TINYBLOB BLOB MEDIUMBLOB LONGBLOB	BYTES	Currently, for BLOB data type, only the blob whose length isn't greater than 2,147,483,647(2 ** 31 - 1) is supported.
YEAR	INT
ENUM	STRING
JSON	STRING	The JSON data type will be converted into STRING with JSON format in Flink.
SET	ARRAY<STRING>	As the SET data type is a string object that can have zero or more values, it should always be mapped to an array of string
GEOMETRY POINT LINESTRING POLYGON MULTIPOINT MULTILINESTRING MULTIPOLYGON GEOMETRYCOLLECTION	STRING	The spatial data types will be converted into STRING with a fixed Json format. Please see Spatial Data Types Mapping section for more detailed information.

Spatial Data Types Mapping

The spatial data types except for GEOMETRYCOLLECTION will be converted into Json String with a fixed format like:

{"srid": 0 , "type": "xxx", "coordinates": [0, 0]}

The field srid identifies the SRS in which the geometry is defined, SRID 0 is the default for new geometry values if no SRID is specified.

The field type identifies the spatial data type, such as POINT/LINESTRING/POLYGON.

The field coordinates represents the coordinates of the spatial data.

For GEOMETRYCOLLECTION, it will be converted into Json String with a fixed format like:

{"srid": 0 , "type": "GeometryCollection", "geometries": [{"type":"Point","coordinates":[10,10]}]}

The field geometries is an array contains all spatial data.

The example for different spatial data types mapping is as follows:

Spatial data	Json String converted in Flink
POINT(1 1)	{"coordinates":[1,1],"type":"Point","srid":0}
LINESTRING(3 0, 3 3, 3 5)	{"coordinates":[[3,0],[3,3],[3,5]],"type":"LineString","srid":0}
POLYGON((1 1, 2 1, 2 2, 1 2, 1 1))	{"coordinates":[[[1,1],[2,1],[2,2],[1,2],[1,1]]],"type":"Polygon","srid":0}
MULTIPOINT((1 1),(2 2))	{"coordinates":[[1,1],[2,2]],"type":"MultiPoint","srid":0}
MultiLineString((1 1,2 2,3 3),(4 4,5 5))	{"coordinates":[[[1,1],[2,2],[3,3]],[[4,4],[5,5]]],"type":"MultiLineString","srid":0}
MULTIPOLYGON(((0 0, 10 0, 10 10, 0 10, 0 0)), ((5 5, 7 5, 7 7, 5 7, 5 5)))	{"coordinates":[[[[0,0],[10,0],[10,10],[0,10],[0,0]]],[[[5,5],[7,5],[7,7],[5,7],[5,5]]]],"type":"MultiPolygon","srid":0}
GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))	{"geometries":[{"type":"Point","coordinates":[10,10]},{"type":"Point","coordinates":[30,30]},{"type":"LineString","coordinates":[[15,15],[20,20]]}],"type":"GeometryCollection","srid":0}