# Oracle
## Description
Change Data Capture (CDC) enables you to ingest data changes from tables in an Oracle [Data Store](/overview/core-concepts/store.md) and publish them to a sink. In a CDC pipeline, data changes in the source relational table are captured at the row-level operations (`INSERT`, `DELETE`, `UPDATE`) based on the table's `PRIMARY KEY`. Then they're pushed to the downstream sink in real-time.
{% hint style="info" %}
**Note** Oracle stores support CDC flows, and changes can be captured from Oracle tables and streamed into DeltaStream for real-time processing.
{% endhint %}
## How a CDC Pipeline Works
### Source Record Structure
DeltaStream uses [Debezium](https://debezium.io/) to capture changes in the source relational table via Oracle LogMiner. This means when you define a CDC source backed by a given table in an ORACLE store, you can use this JSON skeleton:
```
{
"op": ...,
"ts_ms": ...,
"before": {...},
"after": {...},
"source": {...}
}
```
* **op:** Describes the row-level operation that caused a change in the source. It can be `c` for `CREATE`, `u` for `UPDATE`, `d` for `DELETE` or `r` for `READ`.
* **ts\_ms:** Shows the time at which the change event is processed by the CDC pipeline, in the source.
* **before** and **after**: These specify the state of the row at which the change occurred, before and/or after the change, depending on the semantics of the operation.
* **source**: Shows the metadata in the source about the operation. Some of its included fields are:
* **db** (String): The Oracle database name containing the source table.
* **schema** (String): The Oracle schema name containing the source table.
* **table** (String): The Oracle source table name.
* **scn** (BigInt): The Oracle System Change Number (SCN) of the change.
### Requirements for an Oracle Source Store
DeltaStream uses Oracle LogMiner to stream changes from the source relational table.
* When you create the source Oracle [Data Store](/overview/core-concepts/store.md):
* the Oracle instance must be running in **ARCHIVELOG** mode, which is required for LogMiner to access historical redo data.
* **supplemental logging** must be enabled at the database level (at minimum, minimal supplemental logging). For tables captured by CDC, enable supplemental logging for all columns to ensure complete before/after images.
* the username you provide must have sufficient privileges to use Oracle LogMiner and read change events from the source tables (e.g. `CREATE SESSION`, `SELECT` on source tables, `LOGMINING`, `SELECT ANY TRANSACTION`, etc.).
* the URI must use the `oracle://` scheme and include exactly one host and port (e.g. `oracle://oracle-host:1521`).
* `oracle.service_name`, `oracle.username`, and `oracle.password` are required.
To verify that ARCHIVELOG mode is enabled in the Oracle instance, connect as `SYSDBA` and run:
```sql
SELECT LOG_MODE FROM V$DATABASE;
```
To enable minimal supplemental logging at the database level:
```sql
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
```
To enable supplemental logging for all columns on a specific table:
```sql
ALTER TABLE schema_name.table_name ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
```
{% hint style="info" %}
**Note** For more information on Oracle LogMiner and supplemental logging, please see:
* [Oracle LogMiner](https://docs.oracle.com/en/database/oracle/oracle-database/19/sutil/oracle-logminer-utility.html)
* [Supplemental Logging](https://docs.oracle.com/en/database/oracle/oracle-database/19/sutil/oracle-logminer-utility.html#GUID-D857AF96-AC24-4CA1-B620-8EA3DF30D1BB)
{% endhint %}
## How to Create a CDC Pipeline
You create a CDC pipeline in 2 steps:
1. Define the CDC source
2. Define the sink and the query to write CDC changes into it
### Step 1. Define DDL for the CDC source
Using [CREATE STREAM](/reference/sql-syntax/ddl/create-stream.md), define a [Database](/overview/core-concepts/databases.md#stream) backed by the source relational table. The parameters below are used as CDC parameters in the `WITH` clause in [CREATE STREAM](/reference/sql-syntax/ddl/create-stream.md):
| Parameter Name | Description |
| ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `oracle.cdc.table.list` | Comma-separated list of fully qualified source tables to capture using CDC. Each entry must be specified as schema\_name.table\_name. If a table name contains special characters (such as a dot) or uppercase letters, enclose it in double quotes — for example, HR."orders.v1". Regex patterns are supported to match multiple tables dynamically (for example, HR.ORDERS\_.\* matches all tables in the HR schema with names starting with ORDERS\_).
Note: This property can be specified either in the DDL when defining the CDC source, or as a source property at query time. If it is provided in both places, the value specified at query time takes precedence.
Required: No
Type: String
Valid values: See /pages/ZVNhODGeUbLHKEa7VuaP
|
| `value.format` | Format of the CDC record coming from the source.
Required: Yes
Type: String
Valid values: JSON
|
| `store` | Name of the ORACLE store that hosts the relational table backing the CDC source.
Required: Yes
Type: String
Valid values: See /pages/yGCSID06tqSuDcvkGevU
|
### Step 2. Define the CREATE STREAM AS SELECT (CSAS) for CDC Sink Stream and Query
Using [CREATE STREAM AS SELECT](/reference/sql-syntax/query/create-stream-as.md), define a [Database](/overview/core-concepts/databases.md#stream) backed by the sink [Data Store](/overview/core-concepts/store.md#streaming-entity) and the query to insert CDC records into it.
#### Sink parameters
The CDC sink can write into an existing entity in the sink store or create a new entity. The below parameters are used in the WITH clause for the sink in CSAS to create the desired behavior:
| Parameter Name | Description |
| ------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `topic` | Name of the /pages/khAma1ENd997ICjDGcJW#streaming-entity into which the data for the CDC sink is written. If the entity doesn't exist, the system creates an entity with this name in the corresponding store.
Required: No
Default value: Lowercase sink name
Type: String
|
| `topic.partitions` | The number of partitions to use when creating the sink entity, if applicable. If the entity already exists, this value must equal the number of partitions in the existing entity.
Required: Yes, unless topic already exists
Type: Integer
Valid values: \[1, ...]
|
| `topic.replicas` | The number of replicas to use when creating the sink entity, if applicable. If the entity already exists, this value must equal the number of replicas in the existing entity.
Required: Yes, unless topic already exists
Type: Integer
Valid values: \[1, ...]
|
#### Source parameters
The source stream in the CDC pipeline captures the changes from a relational table in the source ORACLE [Data Store](/overview/core-concepts/store.md). Here are the parameters you can specify in the WITH clause for the source in CSAS:
| Parameter Name | Description |
| ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `scan.startup.option` | Startup mode for Oracle CDC consumer when starting.
Required: No
Default value: initial
Type: String
Valid values: initial, latest-offset
|
| `oracle.cdc.table.list` | Comma-separated list of fully qualified source tables to capture using CDC. Each entry must be specified as schema\_name.table\_name. If a table name contains special characters (such as a dot) or uppercase letters, enclose it in double quotes — for example, HR."orders.v1". Regex patterns are supported to match multiple tables dynamically (for example, HR.ORDERS\_.\* matches all tables in the HR schema with names starting with ORDERS\_).
Note: This property can be specified either in the DDL when defining the CDC source, or as a source property at query time. If it is provided in both places, the value specified at query time takes precedence.
Required: No
Type: String
Valid values: See /pages/ZVNhODGeUbLHKEa7VuaP
|
#### Passthrough Source parameters
**Passthrough CDC properties**
Passthrough CDC properties are supported for the CDC source stream. These properties must be specified using the `cdc.` prefix in their names and must be assigned a string value. They are listed in the table below. Refer to the [CDC connector options](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/docs/connectors/flink-sources/oracle-cdc/) for more details.
| Parameter Name (with prefix) | Description |
| ---------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `cdc.split.size` | The number of rows of table snapshot. Captured tables are split into multiple splits when read the snapshot of table.
Required: No
Default value: '1000'
Type: String
|
| `cdc.fetch.size` | The maximum fetch size for per poll when read table snapshot.
Required: No
Default value: '1024'
Type: String
|
| `cdc.scan.incremental.close-idle-reader.enabled` | Whether to close idle readers at the end of the snapshot phase.
Required: No
Default value: 'false'
Type: String
|
| `cdc.chunk-key.even-distribution.factor.lower-bound` | The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distributed or not. The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven. The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.
Required: No
Default value: '0.05d'
Type: String
|
| `cdc.chunk-key.even-distribution.factor.upper-bound` | The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distributed or not. The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven. The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.
Required: No
Default value: '1000.0d'
Type: String
|
| `cdc.connect.pool.size` | The connection pool size.
Required: No
Default value: '30'
Type: String
|
| `cdc.connect.max-retries` | The max retry times that the connector should retry to build database server connection.
Required: No
Default value: '3'
Type: String
|
| `cdc.connect.timeout` | The maximum time that the connector should wait after trying to connect to the Oracle database server before timing out.
Required: No
Default value: '30s'
Type: String
|
| `cdc.scan.newly-added-table.enabled` | Enables automatic discovery of new tables when resuming an Oracle CDC pipeline. When set to true, any tables that match the configured tables list patterns, but were not part of the previous run, are automatically detected. When set to false (default), the pipeline only continues reading the tables that were part of the original state and ignores any newly created matching tables. This option has an effect only when resuming a CDC pipeline from a previous run's state and does not change the behavior of a fresh initial run.
Required: No
Default value: 'false'
Type: String
|
**Passthrough Debezium properties**
CDC source also supports passthrough Debezium properties. These properties must be specified using the `cdc.debezium.` prefix. For additional details on Debezium passthrough options, refer to [Debezium's Oracle Connector properties](https://debezium.io/documentation/reference/1.9/connectors/oracle.html#oracle-connector-properties)
## Supported Data Types in an Oracle CDC Pipeline
| NUMBER(p, s <= 0), p - s < 3 | TINYINT |
| ------------------------------------------------------------------------------------------------------------------ | ------------------------------------ |
| NUMBER(p, s <= 0), p - s < 5 | SMALLINT |
| NUMBER(p, s <= 0), p - s < 10 | INT |
| NUMBER(p, s <= 0), p - s < 19 | BIGINT |
| NUMBER(p, s <= 0), 19 <= p - s <= 38
| DECIMAL(p - s, 0) |
| NUMBER(p, s > 0) | DECIMAL(p, s) |
| NUMBER(p, s <= 0), p - s > 38 | STRING |
| FLOAT
BINARY\_FLOAT
| FLOAT |
| DOUBLE PRECISION
BINARY\_DOUBLE
| DOUBLE |
| NUMBER(1) | BOOLEAN |
| DATE
TIMESTAMP \[(p)]
| TIMESTAMP \[(p)] \[WITHOUT TIMEZONE] |
| TIMESTAMP \[(p)] WITH TIME ZONE | TIMESTAMP \[(p)] WITH TIME ZONE |
| TIMESTAMP \[(p)] WITH LOCAL TIME ZONE | TIMESTAMP\_LTZ \[(p)] |
| CHAR(n)
NCHAR(n)
NVARCHAR2(n)
VARCHAR(n)
VARCHAR2(n)
CLOB
NCLOB
XMLType
SYS.XMLTYPE
| STRING |
| BLOB
ROWID
| BYTES |
| INTERVAL DAY TO SECOND
INTERVAL YEAR TO MONTH
| BIGINT |
***Attention***: Oracle data types that are not listed in the table above are not supported by the DeltaStream Flink CDC Connector.
## Example
**Create CDC pipeline capturing changes from an Oracle table**
Assume an `ORACLE` [Data Store](/overview/core-concepts/store.md) is defined with the name `oracle_store` and it has a table named `PAGEVIEWS` under the schema named `HR` in the database accessed via service `FREEPDB1`. Here is how the data displays in `PAGEVIEWS`. Each row has 3 columns — `VIEWTIME`, `USERID`, and `PAGEID` — and shows when a given page was visited by a specific user:
```sh
demodb.public/oracle_store# PRINT ENTITY HR.PAGEVIEWS;
+-----------------+-------------+-------------+
| VIEWTIME | USERID | PAGEID |
+=================+=============+=============+
| 1694124853651 | User_4 | Page_29 |
+-----------------+-------------+-------------+
| 1694124856731 | User_1 | Page_59 |
+-----------------+-------------+-------------+
| 1694124857732 | User_1 | Page_63 |
+-----------------+-------------+-------------+
```
First, create the Oracle store:
```sql
CREATE STORE oracle_store WITH (
'type' = ORACLE,
'uris' = 'oracle://oracle-host:1521',
'oracle.service_name' = 'FREEPDB1',
'oracle.username' = 'scott',
'oracle.password' = 'tiger'
);
```
**Create CDC pipeline capturing changes from multiple tables**
Now consider a scenario similar to the previous example, except that besides the `PAGEVIEWS` table there is another table called `USERS` under the `STAFF` schema within the same Oracle database in the `oracle_store` store.
Suppose you want to create a CDC pipeline to capture changes for **multiple tables**. First, define a stream as the CDC source. When capturing changes from multiple tables with different schemas, the `before` and `after` fields are defined as `BYTES` for flexibility:
```sql
CREATE STREAM cdc_source(
op VARCHAR,
ts_ms BIGINT,
`before` BYTES,
`after` BYTES,
`source` STRUCT<`schema` VARCHAR, `table` VARCHAR>)
WITH (
'store' = 'oracle_store',
'value.format' = 'json',
'oracle.cdc.table.list' = 'HR.PAGEVIEWS,STAFF.USERS');
```
The `cdc_source` stream acts as the CDC source in your pipeline. Whenever an `INSERT`, `DELETE`, or `UPDATE` happens on either the `PAGEVIEWS` or `USERS` table in the source database, a corresponding change event is captured and written into this stream.
Now, use the following CSAS statement to define a downstream stream as the sink and run a query to write these changes into it. Each output record includes the operation type and timestamp (`op` and `ts_ms`), along with the source schema and table name, which indicate where the changes came from.
You can use these fields to filter or route changes for each table in subsequent queries that read from `cdc_sink`.
```sql
CREATE STREAM cdc_sink WITH (
'topic.partitions' = 1,
'topic.replicas' = 1) AS
SELECT
`source`->`schema` AS src_schema,
`source`->`table` AS src_table,
`before`,
`after`,
`op`,
`ts_ms`
FROM cdc_source;
```
Run one query per table to write the captured changes into their own dedicated Kafka topic.
```sql
-- Query 1
CREATE STREAM pageviews_raw_cdc WITH ('store' = 'kafka_store') AS
SELECT
`before`,
`after`,
`op`,
`ts_ms`
FROM cdc_sink
WHERE src_schema = 'HR' AND src_table = 'PAGEVIEWS';
-- Query 2
CREATE STREAM users_raw_cdc WITH ('store' = 'kafka_store') AS
SELECT
`before`,
`after`,
`op`,
`ts_ms`
FROM cdc_sink
WHERE src_schema = 'STAFF' AND src_table = 'USERS';
```
You can verify the per-table streams are receiving the expected CDC events using `PRINT ENTITY`:
```sh
demodb.public/kafka_store# PRINT ENTITY pageviews_raw_cdc;
+-------------------------------------------------------------------+-------------------------------------------------------------------+----+---------------+
| before | after | op | ts_ms |
+===================================================================+===================================================================+====+===============+
| null | {"VIEWTIME":1694124853651,"USERID":"User_4","PAGEID":"Page_29"} | c | 1694124853700 |
+-------------------------------------------------------------------+-------------------------------------------------------------------+----+---------------+
| {"VIEWTIME":1694124853651,"USERID":"User_4","PAGEID":"Page_29"} | {"VIEWTIME":1694124853651,"USERID":"User_4","PAGEID":"Page_30"} | u | 1694124860012 |
+-------------------------------------------------------------------+-------------------------------------------------------------------+----+---------------+
```
```sh
demodb.public/kafka_store# PRINT ENTITY users_raw_cdc;
+---------------------------------------------------------------+---------------------------------------------------------------+----+---------------+
| before | after | op | ts_ms |
+===============================================================+===============================================================+====+===============+
| null | {"UID":"u001","USER_NAME":"Alice","CITY":"NYC","BALANCE":500} | c | 1694124870200 |
+---------------------------------------------------------------+---------------------------------------------------------------+----+---------------+
| {"UID":"u001","USER_NAME":"Alice","CITY":"NYC","BALANCE":500} | null | d | 1694124890350 |
+---------------------------------------------------------------+---------------------------------------------------------------+----+---------------+
```
Now run a DDL per topic to define a table-specific CDC source for the table.
```sql
CREATE STREAM pageviews_cdc(
op VARCHAR,
ts_ms BIGINT,
`before` STRUCT,
`after` STRUCT
) WITH (
'store' = 'kafka_store',
'value.format' = 'json',
'topic' = 'pageviews_raw_cdc');
```
```sql
CREATE STREAM users_cdc(
op VARCHAR,
ts_ms BIGINT,
`before` STRUCT,
`after` STRUCT
) WITH (
'store' = 'kafka_store',
'value.format' = 'json',
'topic' = 'users_raw_cdc');
```
You could write further queries on the above source relations, again defined per table on dedicated topics, to process and write results into the sink of your choice, such as a Snowflake table.
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://docs.deltastream.io/reference/sql-syntax/query/change-data-capture-cdc/oracle.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.