Oracle Connector

Oracle connector allows reading snapshot data and incremental data from Oracle database and provides end-to-end full-database data synchronization capabilities. This document describes how to set up the Oracle connector.

Example

An example of the pipeline for reading data from Oracle and sink to Doris can be defined as follows:

source:
   type: oracle
   name: Oracle Source
   hostname: 127.0.0.1
   port: 1521
   username: debezium
   password: password
   database: ORCLDB
   tables: testdb.\.*, testdb.user_table_[0-9]+, [app|web].order_\.*

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: password

pipeline:
   name: Oracle to Doris Pipeline
   parallelism: 4

Connector Options

Option	Required	Default	Type	Description
hostname	required	(none)	String	IP address or hostname of the Oracle database server.
port	required	1521	Integer	Integer port number of the Oracle database server.
username	required	(none)	String	Name of the Oracle database to use when connecting to the Oracle database server.
password	required	(none)	String	Password to use when connecting to the Oracle database server.
tables	required	(none)	String	Table name of the Oracle database to monitor. The table-name also supports regular expressions to monitor multiple tables that satisfy the regular expressions. It is important to note that the dot (.) is treated as a delimiter for database and table names. If there is a need to use a dot (.) in a regular expression to match any character, it is necessary to escape the dot with a backslash. eg. db0.\.*, db1.user_table_[0-9]+, db[1-2].[app|web]order_\.*
schema-change.enabled	optional	true	Boolean	Whether to send schema change events, so that downstream sinks can respond to schema changes and achieve table structure synchronization.
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.startup.mode	optional	initial	String	Optional startup mode for Oracle CDC consumer, valid enumerations are "initial", "latest-offset" and "snapshot".
debezium.*	optional	(none)	String	Pass-through Debezium's properties to Debezium Embedded Engine which is used to capture data changes from Oracle server. For example: 'debezium.snapshot.mode' = 'never'. See more about the Debezium's Oracle 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'
metadata.list	optional	false	String	List of readable metadata from SourceRecord to be passed to downstream and could be used in transform module, split by `,`. Available readable metadata are: op_ts.
jdbc.url	optional	(none)	String	The url for Oracle JDBC, the url will be used preferentially, if no url is configured, then use "jdbc:oracle:thin:@localhost:1521:orcl", but oracle 19c url is "jdbc:oracle:thin:@//localhost:1521/pdb1", so the url property is option to adapt to different versions of Oracle.
database	required	(none)	String	Database name of the Oracle server to monitor.
server-time-zone	optional	(none)	String	The session time zone in database server. If not set, then ZoneId.systemDefault() is used to determine the server time zone.
connection.pool.size	optional	20	Integer	The connection pool size.
connect.timeout	optional	30	Integer	The maximum time that the connector should wait after trying to connect to the oracle database server before timing out.
connect.max-retries	optional	3	Integer	The max retry times that the connector should retry to build oracle database server connection.
chunk-meta.group.size	optional	1000	Integer	The group size of chunk meta, if the meta size exceeds the group size, the meta will be divided into multiple groups.
chunk-key.even-distribution.factor.upper-bound	optional	1000.0d	Double	The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the" table is evenly distribution or not." The table chunks would use evenly calculation optimization when the data distribution is even," and the query oracle for splitting would happen when it is uneven. The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.
chunk-key.even-distribution.factor.lower-bound	optional	0.05d	Double	The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distribution or not." The table chunks would use evenly calculation optimization when the data distribution is even," and the query oracle for splitting would happen when it is uneven. The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.
debezium.log.mining.strategy	optional	online_catalog	String	A strategy in log data analysis or mining.
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.
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 incremental reading phase instead of being merged into the snapshot.WARNING: Skipping backfill might lead to data inconsistency because some incremental 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 binlog events should be handled specially.

Startup Reading Position

The config option scan.startup.mode specifies the startup mode for Oracle CDC consumer. The valid enumerations are:

• initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest archive log.
• latest-offset: On first startup, no snapshots of the monitored database tables are performed, and the connector only starts reading from the end of the archive log, which means the connector can only read data changes after the connector starts.
• snapshot: Only the snapshot phase is performed and exits after the snapshot phase reading is completed.

For example in YAML definition:

source:
  type: oracle
  scan.startup.mode: earliest-offset                    # Start from earliest offset
  scan.startup.mode: latest-offset                      # Start from latest offset
  scan.startup.mode: latest-offset                      # Read snapshot only
  # ...

Data Type Mapping

Oracle type	CDC type	NOTE
NUMBER(p,s)	DECIMAL(p, s)/BIGINT	When s is greater than 0, use DECIMAL (p, s); Otherwise, use BIGINT.
LONG	BIGINT
DATE	DATE
FLOAT BINARY_FLOAT REAL	FLOAT
BINARY_DOUBLE DOUBLE	DOUBLE
TIMESTAMP(p)	TIMESTAMP [(p)]
TIMESTAMP(p) WITH TIME ZONE	TIMESTAMP_TZ [(p)]
TIMESTAMP(p) WITH LOCAL TIME ZONE	TIMESTAMP_LTZ [(p)]
INTERVAL YEAR(2) TO MONTH INTERVAL DAY(3) TO SECOND(2) BIGINT	BIGINT
NCHAR(n) CHAR(n)	CHAR(n)
VARCHAR(n) VARCHAR2(n) NVARCHAR2(n)	VARCHAR(n)
CLOB NLOB NCLOB SDO_GEOMETRY XMLTYPE	STRING	Currently, only blob data types with a length not exceeding 2147483647 (2 * * 31-1) are supported in Oracle.
BLOB RAW BFILE	BYTES
INTEGER SMALLINT TINYINT	INT
BOOLEAN	BOOLEAN