title: Prepare Oracle and Oracle RAC for RDI

title: Overview

text:
Oracle provides two main systems that Debezium can use to capture data changes:

- [LogMiner](#logminer)
- [XStream](#xstream)

The sections below explain how to configure each system for use with Debezium and RDI.
The checklists summarize the steps you should follow to configure each system.
You may find it helpful to use them to track your progress as you work through the steps.

**LogMiner**

[code example]

**XStream**

[code example]

**Optional: XMLTYPE Support**

[code example]

title: LogMiner

text:
Follow the steps below to configure
[LogMiner](https://docs.oracle.com/en/database/oracle/oracle-database/19/sutil/oracle-logminer-utility.html)
and prepare your database for use with RDI.

title: 1. Configure Oracle LogMiner

text:
The following example shows the configuration for Oracle LogMiner.

[Amazon RDS for Oracle](https://aws.amazon.com/rds/oracle/)
doesn't let you execute the commands
in the example below or let you log in as `sysdba`. See the
separate example below to [configure Amazon RDS for Oracle](#config-aws).

[code example]

#### Configure Amazon RDS for Oracle {#config-aws}

AWS provides its own set of commands to configure LogMiner.

Before executing these commands,
you must enable backups on your Oracle AWS RDS instance.

Check that Oracle has backups enabled with the following command:

[code example]

The `LOG_MODE` should be set to `ARCHIVELOG`. If it isn't then you
should reboot your Oracle AWS RDS instance.

Once `LOG_MODE` is correctly set to ARCHIVELOG, execute the following
commands to complete the LogMiner configuration. The first command enables
archive logging and the second adds [supplemental logging](#supp-logging).

[code example]

title: 2. Enable supplemental logging {#supp-logging}

text:
You must enable supplemental logging for the tables you want to capture or
for the entire database. This lets Debezium capture the state of
database rows before and after changes occur.

The following example shows how to configure supplemental logging for all columns
in a single table called `inventory.customers`:

[code example]

If you enable supplemental logging for *all* table columns, you will
probably see the size of the Oracle redo logs increase dramatically. Avoid this
by using supplemental logging only when you need it.

You must also enable minimal supplemental logging at the database level with
the following command:

[code example]

title: 3. Check the redo log sizing

text:
Before you use the Debezium connector, you should check with your
database administrator that there are enough
redo logs with enough capacity to store the data dictionary for your
database. In general, the size of the data dictionary increases with the number
of tables and columns in the database. If you don't have enough capacity in
the logs then you might see performance problems with both the database and
the Debezium connector.

title: 4. Set the Archive log destination

text:
You can configure up to 31 different destinations for archive logs
(you must have administrator privileges to do this). You can set parameters for
each destination to specify its purpose, such as log shipping for physical
standbys, or external storage to allow for extended log retention. Oracle reports
details about archive log destinations in the `V$ARCHIVE_DEST_STATUS` view.

The Debezium Oracle connector only uses destinations that have a status of
`VALID` and a type of `LOCAL`. If you only have one destination with these
settings then Debezium will use it automatically.
If you have more than one destination with these settings,
then you should consult your database administrator about which one to
choose for Debezium.

Use the `log.mining.archive.destination.name` property in the connector configuration
to select the archive log destination for Debezium.

For example, suppose you have two archive destinations, `LOG_ARCHIVE_DEST_2` and
`LOG_ARCHIVE_DEST_3`, and they both have status set to `VALID` and type set to
`LOCAL`. Debezium could use either of these destinations, so you must select one
of them explicitly in the configuration. To select `LOG_ARCHIVE_DEST_3`, you would
use the following setting:

[code example]

title: 5. Create a user for the connector {#create-dbz-user}

text:
The Debezium Oracle connector must run as an Oracle LogMiner user with specific permissions.

Typically, when you create the Oracle account for the connector,
you grant the account a level of access that permits the connector to detect changes from all tables in the database.
However, in some environments, security policies might prohibit you from granting such a broad level of access.

The following example shows some SQL that creates an Oracle user account for the connector in a multi-tenant database model.
The grant settings in the example permit the Debezium user to access all user tables in the database.

To comply with security policies, you can modify the `SELECT ANY TABLE` and `FLASHBACK ANY TABLE` grants
so that the connector can access only those tables that you intend to capture.

Do not modify other grants, such as the `SELECT ANY TRANSACTION` grant,
or the set of `SELECT ON V_$` grants, which provide access to dynamic performance views (`V_$`).
These grants are required for the connector to function.

To prevent data loss, if you restrict the scope of the SELECT and FLASHBACK grants,
be sure that the modified scope is compatible with the settings in the connector’s include configuration.
The privileges that you set for the account must permit reading from all of the tables that you want the connector to capture.

This example uses `ORCLCDB` as the container database (CDB) name and `ORCLPDB1` as the pluggable database (PDB) name. Replace these with the CDB and PDB names from your own environment.

[code example]

| Role name                | Description                                                                                |
|--------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| CREATE SESSION           | Enables the connector to connect to Oracle.                                                                                |
| SET CONTAINER            | Enables the connector to switch between pluggable databases. This is only required when the Oracle installation has container database support (CDB) enabled.                                                                                |
| SELECT ON V_$DATABASE    | Enables the connector to read the V_$DATABASE table.                                                                                |
| FLASHBACK ANY TABLE      | Enables the connector to perform Flashback queries, which is how the connector performs the initial snapshot of data. Optionally, rather than granting FLASHBACK permission on all tables, you can grant the FLASHBACK privilege for specific tables only.                                                                                |
| SELECT ANY TABLE         | Enables the connector to read any table. Optionally, rather than granting SELECT permission on all tables, you can grant the SELECT privilege for specific tables only.                                                                                |
| SELECT_CATALOG_ROLE      | Enables the connector to read the data dictionary, which is needed by Oracle LogMiner sessions.                                                                                |
| EXECUTE_CATALOG_ROLE     | Enables the connector to write the data dictionary into the Oracle redo logs, which is needed to track schema changes.                                                                                |
| SELECT ANY TRANSACTION   | Enables the snapshot process to perform a Flashback snapshot query against any transaction so that the connector can read past changes from LogMiner. When FLASHBACK ANY TABLE is granted, this should also be granted. This grant is optional for Oracle 19c and later. In those later releases, the connector obtains the required privileges through the EXECUTE_CATALOG_ROLE and LOGMINING grants. |
| LOGMINING                | This role was added in newer versions of Oracle as a way to grant full access to Oracle LogMiner and its packages. On older versions of Oracle that don’t have this role, you can ignore this grant.                                                                                |
| CREATE TABLE             | Enables the connector to create its flush table in its default tablespace. The flush table allows the connector to explicitly control flushing of the LGWR internal buffers to disk.                                                                                |
| LOCK ANY TABLE           | Enables the connector to lock tables during schema snapshot. If snapshot locks are explicitly disabled via configuration, this grant can be safely ignored.                                                                                |
| CREATE SEQUENCE          | Enables the connector to create a sequence in its default tablespace.                                                                                |
| EXECUTE ON DBMS_LOGMNR   | Enables the connector to run methods in the DBMS_LOGMNR package. This is required to interact with Oracle LogMiner. On newer versions of Oracle this is granted via the LOGMINING role but on older versions, this must be explicitly granted.                                                                                |
| EXECUTE ON DBMS_LOGMNR_D | Enables the connector to run methods in the DBMS_LOGMNR_D package. This is required to interact with Oracle LogMiner. On newer versions of Oracle this is granted via the LOGMINING role but on older versions, this must be explicitly granted.                                                                                |
| SELECT ON V_$….          | Enables the connector to read these tables. The connector must be able to read information about the Oracle redo and archive logs, and the current transaction state, to prepare the Oracle LogMiner session. Without these grants, the connector cannot operate.                                                                                |

#### Limiting privileges

The privileges granted in the example above are convenient,
but you may prefer to restrict them further to improve security. In particular,
you might want to prevent the Debezium user from creating tables, or
selecting or locking any table.

The Debezium user needs the `CREATE TABLE` privilege to create the
`LOG_MINING_FLUSH` table when it connects for the first
time. After this point, it doesn't need to create any more tables,
so you can safely revoke this privilege with the following command:

[code example]

[The example above](#create-dbz-user) grants the `SELECT ANY TABLE` and
`FLASHBACK ANY TABLE` privileges for convenience, but only the tables synced to RDI
and the `V_$XXX` tables strictly need these privileges.
You can replace the `GRANT SELECT ANY TABLE` command with explicit
commands for each table. For example, you would use commands like the
following for the tables in our sample
[`chinook`](https://github.com/Redislabs-Solution-Architects/rdi-quickstart-postgres)
database. (Note that Oracle 19c requires you to run a separate `GRANT`
command for each table individually.)

[code example]

Similarly, instead of `GRANT FLASHBACK ANY TABLE`, you would use the following
commands:

[code example]

The `LOCK` privilege is automatically granted by the `SELECT`
privilege, so you can omit this command if you have granted `SELECT`
on specific tables.

#### Revoking existing privileges

If you initially set the Debezium user's privileges on all tables,
but you now want to restrict them, you can revoke the existing
privileges before resetting them as described in the
[Limiting privileges](#limiting-privileges) section.

Use the following commands to revoke and reset the `SELECT` privileges:

[code example]

The equivalent commands for `FLASHBACK` are:

[code example]

The `SELECT` privilege automatically includes the `LOCK`
privilege, so when you grant `SELECT` for specific tables
you should also revoke `LOCK` on all tables:

[code example]

title: 6. Configuration is complete {#logminer-complete}

text:
Once you have followed the steps above, your Oracle database is ready
for Debezium to use.

title: XStream

text:
[XStream](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/introduction-to-xstream.html#GUID-5939CB6C-8BA9-4594-8F96-B0453D246722)
is a set of database components and APIs to communicate change data to and
from an Oracle database. RDI specifically uses
[XStream Out](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/xstream-out.html)
to capture changes.

Follow the steps in the sections below to configure XStream to work with
Debezium and RDI.

You should run all database commands shown below as the `sysdba` user.

[code example]

title: 1. Configure recovery area

text:
**non-RAC (single-instance):**

Create a directory for the recovery area on the Oracle host (or Oracle container if you are using a containerized Oracle).

For single-instance (non-RAC) deployments on local disk, you can use:

[code example]

  [code example]

**Oracle RAC (multitenant environment):**

For Oracle RAC (cluster) environments, the Fast Recovery Area (FRA) must be on **shared storage** (ASM or a shared filesystem).

- **If using ASM (recommended for RAC):**

  [code example]

  Replace `+FRA` with the ASM disk group used for your FRA. No `mkdir` is required; ASM manages the storage.

- **If using a shared filesystem (NFS/OCFS2/etc.):**

  [code example]

  The path you use must:

  - Be on shared storage visible from all RAC nodes
  - Be mounted at the **same path** on all nodes
  - Be writable by the Oracle user

  Create the directory once on the shared filesystem (it will be visible from all nodes):

  [code example]

title: 2. Enable GoldenGate replication

text:
Check if `enable_goldengate_replication` is already set to `true`:

**Container database (CDB):**

For a CDB, check the parameter in the root container:

[code example]

If it is not set to `true`, you need to set it and **restart the database**:

[code example]

**Non-container database (Non-CDB):**

For a non-CDB, check the parameter in the database instance:

[code example]

If it is not set to `true`, you need to set it and **restart the database**:

[code example]

title: 3. Configure XStream

text:
Use the following SQL commands to configure XStream (using the
[`chinook`](https://github.com/Redislabs-Solution-Architects/rdi-quickstart-postgres/tree/main)
schema as an example):

**non-RAC (single-instance):**

[code example]

**Oracle RAC (multitenant environment):**

[code example]

title: 4. Enable supplemental logging

text:
**Container database (CDB):**

Enable supplemental logging for the tables you want to capture or
for the entire database. This lets Debezium capture the state of
database rows before and after changes occur.

[code example]

The example above uses `orclpdb1` as the PDB name. Replace it with the name of the pluggable database (PDB) that you use in your own environment.

**Non-container database (Non-CDB):**

Enable supplemental logging for the tables you want to capture or
for the entire database. This lets Debezium capture the state of
database rows before and after changes occur.

[code example]

You must configure supplemental logging explicitly for each table as shown
above. Otherwise, only the original sync will be performed and no change data will be
captured for the table.

title: 5. Create XStream users

text:
**Container database (CDB):**

The XStream user examples below use `ORCLCDB` as the CDB name and `ORCLPDB1`/`orclpdb1` as the PDB name in file paths and `ALTER SESSION SET CONTAINER` commands. Replace these with the CDB and PDB names from your own Oracle deployment.

Create an XStream administrator user with the following SQL:

[code example]

Then, create the XStream user:

[code example]

If you are using the
[Debezium XStream documentation](https://debezium.io/documentation/reference/stable/connectors/oracle.html#creating-xstream-users-for-the-connector),
you should note that it misses out the last two GRANT statements shown above:

[code example]

However, without these, no tables can be read by Debezium, so neither the initial snapshot nor any subsequent updates will produce any data.

**Non-container database (Non-CDB):**

The non-CDB architecture is deprecated in Oracle Database 12c and discontinued in Oracle Database 20c.

For a non-container (non-CDB) Oracle database:

- You have a single database with no containers.
- Users are regular database users without the `C##` prefix.
- Application data is stored directly in the database.
- You do not use `ALTER SESSION SET CONTAINER` statements or `CONTAINER=ALL` clauses.

**Create XStream users and tablespaces (non-CDB)**<br/>

Run the following script as the `sys` user:

[code example]

This script creates:

- The `xstream_adm_tbs` tablespace and `dbzadmin` XStream administrator user.
- The `xstream_tbs` tablespace and `dbzxsuser` XStream connect user.

title: 6. Create an XStream outbound server

text:
Create the outbound server with the following SQL.

**Container database (CDB):**

In this example, `ORCLCDB` is the CDB service name and `orclpdb1` is the PDB name. Replace them with the appropriate service and PDB names for your own environment.

Note that you must connect as the `c##dbzadmin` user created in the previous step,
not the `sys` user:

[code example]

[code example]

-

**Non-container database (Non-CDB):**

Note that you must connect as the `dbzadmin` user created in the previous step,
not the `sys` user:

[code example]

[code example]

title: 7. Add a custom Docker image for the Debezium server

text:
To support XStream connector, you must create a custom [Docker](https://www.docker.com/) image that includes the required Instant Client package libraries for Linux x64 from the Oracle website.

1.  On the Docker machine, download the Instant Client package for Linux x64 from the Oracle website

    [code example]

1.  Unzip it to the `./dbz-ora` directory:

    [code example]

1.  Create a `Dockerfile` in the `./dbz-ora` directory with the following contents:

    [code example]

1.  Create the custom image:

    [code example]

1.  Add the image to the K3s image registry using the following commands:

    [code example]

title: 8. Make RDI use the custom image (VM installation)

text:
Edit the `rdi-operator` configmap:

[code example]

In the editor, find the collector section and change the image settings:

[code example]

Save the configmap. Once it is saved, the operator will restart automatically and will apply the changes.

After upgrading to another RDI version,
the changes to the configmap will be lost. You must repeat the above steps after each upgrade.

title: 9. Enable the Oracle configuration in RDI

text:
Finally, you must update your `config.yaml` file to enable XStream.
The example below shows the relevant parts of the `sources` section:

[code example]

The values `ORCLCDB` and `ORCLPDB1` in the example above are sample CDB and PDB names. Set `database.dbname` and `database.pdb.name` to the CDB and PDB names for your own Oracle database.

See the
[Debezium Oracle documentation](https://debezium.io/documentation/reference/stable/connectors/oracle.html#oracle-connector-properties)
for a full list of properties you can use in the `advanced.source` subsection.

title: 10. Configuration is complete {#xstream-complete}

text:
After you have followed the steps above, your Oracle database is ready
for Debezium to use.

title: Support for Oracle XMLTYPE columns (optional)

role: compatibility

text:
If your Oracle database contains tables with columns of type
[`XMLTYPE`](https://docs.oracle.com/en/database/oracle/oracle-database/21/arpls/XMLTYPE.html),
you must configure additional libraries for Debezium Server to process these columns correctly.

title: Create a custom Debezium Server image

text:
To support `XMLTYPE` columns, you must create a custom [Docker](https://www.docker.com/) image
that includes the required Oracle XML libraries.

1. Download the required libraries from Maven Central:

   [code example]

2. Create a `Dockerfile` in the same directory:

   [code example]

3. Build the custom image:

   [code example]

4. Add the image to your K3s image repository:

   [code example]

title: Configure RDI for XMLTYPE support

role: configuration

text:
In your RDI configuration file, set the `lob.enabled` property to `true` in the
`advanced.source` section:

[code example]

The XMLTYPE configuration example uses `ORCLCDB` as the CDB name and `ORCLPDB1` as the PDB name. Replace these with your actual CDB and PDB names when configuring XMLTYPE support.

title: Test XMLTYPE support

text:
You can create a test table to verify that `XMLTYPE` columns work correctly
(using the
[`CHINOOK`](https://github.com/Redislabs-Solution-Architects/rdi-quickstart-postgres/tree/main)
schema as an example):

[code example]

After you run an initial
[snapshot](),
the XML data appears in your Redis target database:

title: FAQ

title: How does CDB differ from Non-CDB?

text:
**CDB (Container Database – multitenant)**

- Root container `CDB$ROOT` plus one or more pluggable databases (PDBs).
- Common users (for example `C##DBZADMIN`, `C##DBZXSUSER`) exist across all containers.
- Application data lives in PDBs.
- Many XStream-related operations use `CONTAINER=ALL` and `ALTER SESSION SET CONTAINER`.

**Non-CDB (traditional)**

- Single database, no containers.
- Users are regular database users without the `C##` prefix (for example `DBZADMIN`, `DBZXSUSER`).
- Application data lives directly in that single database.
- No `CONTAINER=ALL` or `ALTER SESSION SET CONTAINER` clauses are required.

From an RDI/Debezium point of view, both models are supported; the main differences are in the user names you create and whether you need container-related clauses in the SQL.

title: How does single-instance differ from RAC?

text:
**Single-instance**

- One Oracle instance running on one server.
- You run the LogMiner/XStream setup scripts once.
- All configuration (users, tablespaces, outbound server, logging) is stored in that database instance.

**RAC (Real Application Clusters)**

- Multiple instances (nodes) share the same database and storage.
- You still run the XStream setup only once on any one node.
- All nodes share the same:
  - Database files
  - Archive logs
  - XStream outbound server
  - Capture process

Important RAC points:

- Archive log mode is enabled at the database level and applies to all nodes.
- Tablespaces you create are in shared storage and visible from all nodes.
- Users you create exist in the shared database and can connect through any node.
- Debezium/RDI can connect to any RAC node or to a SCAN address; it does not need a specific node.

title: In RAC setups, do you need to execute steps on all nodes?

text:
No. The XStream configuration is database-level, so you execute the setup once on any node:

- Enable archive log mode: once (affects the whole database).
- Create XStream tablespaces: once (on shared storage).
- Create XStream users: once (in the shared database).
- Create the XStream outbound server: once (visible from all nodes).
- Enable supplemental logging: once (database-wide).

You may run client connection tests from multiple nodes, but the actual administrative steps only need to be executed once.

title: Why are two XStream tablespaces required?

text:
XStream is not just reading your tables - it's storing its own data:

**Customer's existing tablespace**:

[code example]

**XStream tablespaces**:

[code example]

#### Real Example: What Gets Stored Where

Scenario: Customer has CUSTOMERS table

Customer's existing tablespace (USERS):
[code example]

XStream tablespaces (NEW):
[code example]

#### Size Comparison

Customer's Application Tablespaces:

[code example]

XStream Tablespaces:

[code example]

#### General Recommendations

This guide uses two separate tablespaces:

- `xstream_adm_tbs` – XStream administrator tablespace.
- `xstream_tbs` – XStream user (connect) tablespace.

They serve different purposes and follow the principle of least privilege.

**1. Separation of concerns and security**

- *Administrator tablespace (`xstream_adm_tbs`)*
  - Used by the XStream admin user (for example `C##DBZADMIN` or `DBZADMIN`).
  - Stores XStream metadata and control structures:
    - Capture process metadata
    - Outbound server configuration
    - Queue tables and queues
    - Other XStream internal objects
  - Belongs to a highly privileged user that can create and manage XStream objects.

- *User tablespace (`xstream_tbs`)*
  - Used by the XStream connect user (for example `C##DBZXSUSER` or `DBZXSUSER`).
  - Lower-privileged account used by Debezium/RDI to read changes.
  - Keeps operational data and temporary objects separate from admin metadata.

**2. Resource management**

Having two tablespaces lets you:

- Monitor space usage separately for admin vs user workloads.
- Allow for different growth patterns (admin metadata typically grows more slowly).
- Apply separate quotas per user/tablespace.
- Tune storage parameters differently for each tablespace if needed.

**3. Operational benefits**

- Backup and recovery:
  - You can back up or restore admin and user data independently.
  - Critical XStream metadata is isolated in its own tablespace.
- Maintenance and troubleshooting:
  - You can perform maintenance on one tablespace without affecting the other.
  - Easier to diagnose which part of the XStream setup is consuming space.
- Security and auditing:
  - Clear separation of administrative vs operational data.
  - Easier to audit which users access which objects.

**4. Can you use a single tablespace?**

Technically you could place both users in a single tablespace, but this is not recommended for production:

- It weakens separation of duties and least-privilege design.
- It makes monitoring and capacity planning harder.
- It mixes admin metadata and application-facing data in one place.

Oracle and Debezium best practices recommend separating admin and user workloads into different tablespaces.

title: What is the XStream outbound server?

text:
The XStream outbound server is the component that streams database changes out of Oracle in a supported way. It:

- Captures committed changes (INSERT, UPDATE, DELETE) from redo logs.
- Converts them into logical change records (LCRs).
- Exposes an API that external consumers such as Debezium/RDI use.

**High-level architecture**

[code example]

**Key components**

- *Redo logs*: Contain the physical change records for the database.
- *Capture process*: Mines redo logs and turns them into logical change records.
- *Queue*: Temporarily stores captured changes.
- *XStream outbound server* (for example `DBZXOUT`):
  - Provides the streaming interface for consumers.
  - Applies filters (schemas/tables) configured when you create it.
- *Connect user* (for example `C##DBZXSUSER` or `DBZXSUSER`):
  - The user Debezium/RDI connects as.
  - Reads changes from the outbound server.

**Why it is needed**

Without an XStream outbound server:

- There is no supported, structured API for consuming change events.
- You would have to parse redo logs directly, which is complex and unsupported.
- You lose built-in filtering, transaction grouping, and restart/resume semantics.

With an XStream outbound server:

- Changes are available in near real time.
- You can capture only selected schemas/tables.
- Debezium/RDI uses Oracle’s supported XStream API.
- Transactions are preserved and grouped correctly.
- Offsets/positions allow clean resume after restarts.
- Multiple consumers can be attached to the same outbound server if needed.

code:
- [ ] [Configure Oracle LogMiner](#1-configure-oracle-logminer)
- [ ] [Enable supplemental logging](#supp-logging)
- [ ] [Check the redo log sizing](#3-check-the-redo-log-sizing)
- [ ] [Set the Archive log destination](#4-set-the-archive-log-destination)
- [ ] [Create a user for the connector](#create-dbz-user)

code:
- [ ] [Configure recovery area](#1-configure-recovery-area)
- [ ] [Enable GoldenGate replication](#2-enable-goldengate-replication)
- [ ] [Configure XStream](#3-configure-xstream)
- [ ] [Enable supplemental logging](#4-enable-supplemental-logging)
- [ ] [Create XStream users](#5-create-xstream-users)
- [ ] [Create an XStream outbound server](#6-create-an-xstream-outbound-server)
- [ ] [Add a custom Docker image for the Debezium server](#7-add-a-custom-docker-image-for-the-debezium-server)
- [ ] [Make RDI use the custom image](#8-make-rdi-use-the-custom-image-vm-installation)
- [ ] [Enable the Oracle configuration in RDI](#9-enable-the-oracle-configuration-in-rdi)

code:
- [ ] [Create a custom Debezium Server image](#create-a-custom-debezium-server-image)
- [ ] [Configure RDI for XMLTYPE support](#configure-rdi-for-xmltype-support)
- [ ] [Test XMLTYPE support](#test-xmltype-support)

code:
ORACLE_SID=ORACLCDB dbz_oracle sqlplus /nolog

CONNECT sys/top_secret AS SYSDBA
alter system set db_recovery_file_dest_size = 10G;
alter system set db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' scope=spfile;
-- ======================================================================================================================================================
-- !!!IMPORTANT!!!:
-- In order to avoid Oracle downtime, please check if the LOG_MODE on your database is already set to `ARCHIVELOG` before executing the following commands:
-- SELECT log_mode FROM v$database;
-- If the LOG_MODE is already `ARCHIVELOG`, then you can skip the rest of the commands in this script
-- ======================================================================================================================================================
shutdown immediate
startup mount
alter database archivelog;
alter database open;
-- You should now see "Database log mode: Archive Mode"
archive log list

exit;

code:
SQL> SELECT LOG_MODE FROM V$DATABASE;

LOG_MODE
------------
ARCHIVELOG

code:
exec rdsadmin.rdsadmin_util.set_configuration('archivelog retention hours',24);
exec rdsadmin.rdsadmin_util.alter_supplemental_logging('ADD');

code: ALTER TABLE inventory.customers ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

code: ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

code:
sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
CREATE TABLESPACE logminer_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/logminer_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
exit;

sqlplus sys/top_secret@//localhost:1521/ORCLPDB1 as sysdba
CREATE TABLESPACE logminer_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/logminer_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
exit;

sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba

CREATE USER c##dbzuser IDENTIFIED BY dbz
    DEFAULT TABLESPACE logminer_tbs
    QUOTA UNLIMITED ON logminer_tbs
    CONTAINER=ALL;

GRANT CREATE SESSION TO c##dbzuser CONTAINER=ALL;
GRANT SET CONTAINER TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$DATABASE to c##dbzuser CONTAINER=ALL;

-- See `Limiting privileges` below if the privileges
-- granted by these two commands raise security concerns.
GRANT FLASHBACK ANY TABLE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ANY TABLE TO c##dbzuser CONTAINER=ALL;
--

GRANT SELECT_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL;
GRANT EXECUTE_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ANY TRANSACTION TO c##dbzuser CONTAINER=ALL;
GRANT LOGMINING TO c##dbzuser CONTAINER=ALL;

-- See `Limiting privileges` below if the privileges
-- granted by these two commands raise security concerns.
GRANT CREATE TABLE TO c##dbzuser CONTAINER=ALL;
GRANT LOCK ANY TABLE TO c##dbzuser CONTAINER=ALL;
--

GRANT CREATE SEQUENCE TO c##dbzuser CONTAINER=ALL;

GRANT EXECUTE ON DBMS_LOGMNR TO c##dbzuser CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR_D TO c##dbzuser CONTAINER=ALL;

GRANT SELECT ON V_$LOG TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOG_HISTORY TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_LOGS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGFILE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVED_LOG TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$TRANSACTION TO c##dbzuser CONTAINER=ALL;

GRANT SELECT ON V_$MYSTAT TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$STATNAME TO c##dbzuser CONTAINER=ALL;

exit;

code: REVOKE CREATE TABLE FROM c##dbzuser container=all;

code:
GRANT SELECT ON chinook.album TO c##dbzuser;
GRANT SELECT ON chinook.artist TO c##dbzuser;
GRANT SELECT ON chinook.customer TO c##dbzuser;
...

code:
GRANT FLASHBACK ON chinook.album TO c##dbzuser;
GRANT FLASHBACK ON chinook.artist TO c##dbzuser;
GRANT FLASHBACK ON chinook.customer TO c##dbzuser;
...

code:
REVOKE SELECT ANY TABLE FROM c##dbzuser container=all;
ALTER SESSION SET container=orclpdb1;

GRANT SELECT ON chinook.album TO c##dbzuser;
-- ...etc

code:
REVOKE FLASHBACK ANY TABLE FROM c##dbzuser container=all;
ALTER SESSION SET container=orclpdb1;
GRANT FLASHBACK ON chinook.album TO c##dbzuser;

code: REVOKE LOCK ANY TABLE FROM c##dbzuser container=all;

code: sqlplus sys/<PASSWORD> as sysdba

section_id: xstream

code:
ALTER SYSTEM SET db_recovery_file_dest_size = 10G; -- Adjust size as needed
  ALTER SYSTEM SET db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' SCOPE=BOTH;

code:
ALTER SYSTEM SET db_recovery_file_dest_size = 10G; -- Adjust size as needed
  ALTER SYSTEM SET db_recovery_file_dest = '+FRA' SCOPE=BOTH;

code:
ALTER SYSTEM SET db_recovery_file_dest_size = 10G; -- Adjust size as needed
  ALTER SYSTEM SET db_recovery_file_dest = '/u02/oradata/recovery_area' SCOPE=BOTH;

code:
mkdir -p /u02/oradata/recovery_area
  chown oracle:oinstall /u02/oradata/recovery_area
  chmod 755 /u02/oradata/recovery_area

code: SELECT VALUE FROM V$PARAMETER WHERE NAME = 'enable_goldengate_replication';

code: ALTER SYSTEM SET enable_goldengate_replication=true SCOPE=BOTH CONTAINER=ALL;

code: ALTER SYSTEM SET enable_goldengate_replication=true SCOPE=BOTH;

code:
-- ======================================================================================================================================================
-- !!!IMPORTANT!!!:
-- In order to avoid Oracle downtime, please check if the LOG_MODE on your database is already set to `ARCHIVELOG` before executing the following commands:
-- SELECT log_mode FROM v$database;
-- If the LOG_MODE is already `ARCHIVELOG`, then you can skip the rest of the commands in this script
-- ======================================================================================================================================================
SQL> shutdown immediate
Database closed.
Database dismounted.
ORACLE instance shut down.
SQL> startup mount
ORACLE instance started.

Total System Global Area 1476391776 bytes
Fixed Size     9134944 bytes
Variable Size 1006632960 bytes
Database Buffers   452984832 bytes
Redo Buffers     7639040 bytes
Database mounted.
SQL> alter database archivelog;
SQL> alter database open;
SQL> archive log list
Database log mode       Archive Mode
Automatic archival       Enabled
Archive destination       USE_DB_RECOVERY_FILE_DEST
Oldest online log sequence     10
Next log sequence to archive   12
Current log sequence       12

-- Confirm the database is in ARCHIVELOG mode.
SQL> SELECT LOG_MODE FROM V$DATABASE;

code:
-- ======================================================================================================================================================
-- !!!IMPORTANT!!!:
-- In order to avoid Oracle downtime, please check if the LOG_MODE on your database is already set to `ARCHIVELOG` before executing the following commands:
-- SELECT log_mode FROM v$database;
-- If the LOG_MODE is already `ARCHIVELOG`, then you can skip the rest of the commands in this script
-- ======================================================================================================================================================

-- Stop all database instances
SQL> srvctl stop database -d <db_Name>

-- Start the database in mount state.
SQL> srvctl start database -d <db_Name> -o mount

-- Enable archive log mode
SQL> alter database archivelog;

-- Restart all database instances.
SQL> srvctl stop database -d <db_Name>
SQL> srvctl start database -d <db_Name>

-- Confirm the database is in ARCHIVELOG mode.
SQL> SELECT LOG_MODE FROM V$DATABASE;

code:
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

SQL> alter session set container=orclpdb1;

SQL> ALTER TABLE CHINOOK.ALBUM ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.ARTIST ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.CUSTOMER ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.EMPLOYEE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.GENRE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.INVOICE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.INVOICELINE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.MEDIATYPE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.PLAYLIST ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.PLAYLISTTRACK ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.TRACK ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

code:
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

SQL> ALTER TABLE CHINOOK.ALBUM ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.ARTIST ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.CUSTOMER ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.EMPLOYEE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.GENRE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.INVOICE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.INVOICELINE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.MEDIATYPE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.PLAYLIST ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.PLAYLISTTRACK ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER TABLE CHINOOK.TRACK ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

code:
sqlplus sys/<PASSWORD> as sysdba

SQL> ALTER SESSION SET CONTAINER=CDB$ROOT;
SQL> CREATE TABLESPACE xstream_adm_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/xstream_adm_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

SQL> alter session set container=orclpdb1;

SQL> CREATE TABLESPACE xstream_adm_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/xstream_adm_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

SQL> alter session set container=cdb$root;

SQL> CREATE USER c##dbzadmin IDENTIFIED BY dbz
    DEFAULT TABLESPACE xstream_adm_tbs
    QUOTA UNLIMITED ON xstream_adm_tbs
    CONTAINER=ALL;

SQL> GRANT CREATE SESSION, SET CONTAINER TO c##dbzadmin CONTAINER=ALL;

SQL> BEGIN
     DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
        grantee                 => 'c##dbzadmin',
        privilege_type          => 'CAPTURE',
        grant_select_privileges => TRUE,
        container               => 'ALL'
     );
  END;
  /

code:
sqlplus sys/<PASSWORD> as sysdba

SQL> CREATE TABLESPACE xstream_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/xstream_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

SQL> alter session set container=orclpdb1;

SQL> CREATE TABLESPACE xstream_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/xstream_tbs.dbf'
    SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

SQL> alter session set container=cdb$root;

SQL> CREATE USER c##dbzxsuser IDENTIFIED BY dbz
    DEFAULT TABLESPACE xstream_tbs
    QUOTA UNLIMITED ON xstream_tbs
    CONTAINER=ALL;

SQL> GRANT CREATE SESSION TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT SET CONTAINER TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT SELECT ON V_$DATABASE to c##dbzxsuser CONTAINER=ALL;
SQL> GRANT FLASHBACK ANY TABLE TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT SELECT_CATALOG_ROLE TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT EXECUTE_CATALOG_ROLE TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT SELECT ANY TABLE TO c##dbzxsuser CONTAINER=ALL;
SQL> GRANT LOCK ANY TABLE TO c##dbzxsuser CONTAINER=ALL;

code:
GRANT SELECT ANY TABLE TO c##dbzxsuser CONTAINER=ALL;
GRANT LOCK ANY TABLE TO c##dbzxsuser CONTAINER=ALL;

code:
-- =====================================================
-- Create XStream Users - FOR NON-CDB ENVIRONMENT
-- =====================================================
-- This script creates the XStream administrator and user accounts
-- Run as: sqlplus sys/<PASSWORD> as sysdba
-- =====================================================

-- Step 1: Create XStream Administrator Tablespace
CREATE TABLESPACE xstream_adm_tbs DATAFILE '/u01/app/oracle/oradata/ORCL/xstream_adm_tbs.dbf'
  SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

PROMPT Tablespace xstream_adm_tbs created

-- Step 2: Create XStream Administrator User
CREATE USER dbzadmin IDENTIFIED BY dbz
  DEFAULT TABLESPACE xstream_adm_tbs
  QUOTA UNLIMITED ON xstream_adm_tbs;

GRANT CREATE SESSION TO dbzadmin;

PROMPT User dbzadmin created

-- Step 3: Grant XStream Admin Privileges
BEGIN
  DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee                 => 'dbzadmin',
    privilege_type          => 'CAPTURE',
    grant_select_privileges => TRUE
  );
END;
/

PROMPT XStream admin privileges granted to dbzadmin

-- Step 4: Create XStream User Tablespace
CREATE TABLESPACE xstream_tbs DATAFILE '/u01/app/oracle/oradata/ORCL/xstream_tbs.dbf'
  SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

PROMPT Tablespace xstream_tbs created

-- Step 5: Create XStream User
CREATE USER dbzxsuser IDENTIFIED BY dbz
  DEFAULT TABLESPACE xstream_tbs
  QUOTA UNLIMITED ON xstream_tbs;

PROMPT User dbzxsuser created

-- Step 6: Grant necessary privileges to XStream User
GRANT CREATE SESSION TO dbzxsuser;
GRANT SELECT ON V_$DATABASE TO dbzxsuser;
GRANT FLASHBACK ANY TABLE TO dbzxsuser;
GRANT SELECT_CATALOG_ROLE TO dbzxsuser;
GRANT EXECUTE_CATALOG_ROLE TO dbzxsuser;
GRANT SELECT ANY TABLE TO dbzxsuser;
GRANT LOCK ANY TABLE TO dbzxsuser;

PROMPT Privileges granted to dbzxsuser

-- Verification: Check users were created
SELECT username, account_status, default_tablespace
FROM dba_users
WHERE username IN ('DBZADMIN', 'DBZXSUSER');

PROMPT
PROMPT =====================================================
PROMPT XStream Users Created Successfully!
PROMPT =====================================================
PROMPT Created users:
PROMPT   - dbzadmin (XStream Administrator)
PROMPT   - dbzxsuser (XStream User)
PROMPT
PROMPT Next step:
PROMPT Run 03_create_xstream_outbound.sql to create outbound server
PROMPT =====================================================

code:
-- =====================================================
-- Create XStream Outbound Server
-- =====================================================
-- This script creates the XStream outbound server for CDC
-- Run as: sqlplus c##dbzadmin/dbz@localhost:1521/ORCLCDB
-- =====================================================

-- Step 1: Create XStream Outbound Server
DECLARE
  tables  DBMS_UTILITY.UNCL_ARRAY;
  schemas DBMS_UTILITY.UNCL_ARRAY;
BEGIN
  tables(1)  := NULL;
  schemas(1) := 'C##DBZUSER';

  DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
    server_name           => 'dbzxout',
    source_container_name => 'XEPDB1',
    table_names           => tables,
    schema_names          => schemas
  );
END;
/

-- Step 2: Configure XStream User to connect to outbound server
-- This must be run as sys user
CONNECT sys/oracle AS SYSDBA

BEGIN
  DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name  => 'dbzxout',
    connect_user => 'c##dbzxsuser'
  );
END;
/

-- Verification: Check outbound server was created
SELECT server_name, source_database, capture_name, capture_user, connect_user, queue_owner, queue_name
FROM dba_xstream_outbound;

-- Verification: Check capture process
SELECT capture_name, status, capture_type, source_database
FROM dba_capture;

PROMPT
PROMPT =====================================================
PROMPT XStream Outbound Server Created Successfully!
PROMPT =====================================================
PROMPT Server name: dbzxout
PROMPT Schema: C##DBZUSER
PROMPT PDB: XEPDB1
PROMPT =====================================================

code:
-- =====================================================
-- Create XStream Outbound Server - FOR NON-CDB ENVIRONMENT
-- =====================================================
-- This script creates the XStream outbound server for CDC
-- Run as: sqlplus dbzadmin/dbz@localhost:1521/ORCL
-- =====================================================

-- Step 1: Create XStream Outbound Server
DECLARE
  tables  DBMS_UTILITY.UNCL_ARRAY;
  schemas DBMS_UTILITY.UNCL_ARRAY;
BEGIN
  tables(1)  := NULL;
  schemas(1) := 'chinook';  -- Replace with actual schema name

  DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
    server_name     => 'dbzxout',
    table_names     => tables,
    schema_names    => schemas
  );
END;
/

PROMPT XStream outbound server created

-- Step 2: Configure XStream User to connect to outbound server
-- This must be run as sys user
CONNECT sys/<PASSWORD> AS SYSDBA

BEGIN
  DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name  => 'dbzxout',
    connect_user => 'dbzxsuser'
  );
END;
/

PROMPT Connect user configured

-- Verification: Check outbound server was created
SELECT server_name, source_database, capture_name, capture_user, connect_user, queue_owner, queue_name
FROM dba_xstream_outbound;

-- Verification: Check capture process
SELECT capture_name, status, capture_type, source_database
FROM dba_capture;

PROMPT
PROMPT =====================================================
PROMPT XStream Outbound Server Created Successfully!
PROMPT =====================================================
PROMPT Server name: dbzxout
PROMPT =====================================================

code: unzip instantclient-basic-linux.x64-23.8.0.25.04.zip -d ./dbz-ora

code:
FROM debezium/server:3.0.8.Final

    USER root

    RUN microdnf -y install libaio \
    && microdnf clean all \
    && mkdir -p /opt/oracle/instant_client \
    && rm -f /debezium/lib/ojdbc11*.jar

    COPY instantclient_23_8/* /opt/oracle/instant_client

    USER jboss

    COPY instantclient_23_8/xstreams.jar /debezium/lib
    COPY instantclient_23_8/ojdbc11.jar /debezium/lib

    ENV LD_LIBRARY_PATH=/opt/oracle/instant_client

code: docker build -t dbz-ora dbz-ora

code:
docker tag dbz-ora quay.io/debezium/server:3.0.8.Final
    docker image save quay.io/debezium/server:3.0.8.Final -o dbz3.0.8-xstream-linux-amd.tar
    sudo k3s ctr images import dbz3.0.8-xstream-linux-amd.tar all

code: kubectl edit configmap rdi-operator -n rdi

code:
collector:
        image:
          pullPolicy: IfNotPresent
          registry: docker.io # change this to `quay.io`
          repository: redislabs/debezium-server # Change this to `debezium/server`
          tag: 3.0.8.Final-rdi.1 # Change this to `3.0.8.Final`

code:
sources:
  oracle:
    type: cdc
    logging:
      level: info
    connection:
      type: oracle
      host: host.docker.internal
      port: 1521
      user: ${SOURCE_DB_USERNAME}
      password: ${SOURCE_DB_PASSWORD}
    advanced:
      source:
        database.dbname: ORCLCDB
        database.pdb.name: ORCLPDB1
        database.connection.adapter: xstream
        database.out.server.name: dbzxout

code:
mkdir xml
   cd xml
   wget https://repo.maven.apache.org/maven2/com/oracle/database/xml/xdb/19.27.0.0/xdb-19.27.0.0.jar
   wget https://repo.maven.apache.org/maven2/com/oracle/database/xml/xmlparserv2/19.27.0.0/xmlparserv2-19.27.0.0.jar
   mv xdb-19.27.0.0.jar xdb.jar
   mv xmlparserv2-19.27.0.0.jar xmlparserv2.jar

section_id: create-a-custom-debezium-server-image

code:
FROM quay.io/debezium/server:3.0.8.Final

   USER root

   COPY xdb.jar /debezium/lib
   COPY xmlparserv2.jar /debezium/lib

code:
cd ..
   docker build -t dbz-xml xml
   docker tag dbz-xml quay.io/debezium/server:3.0.8.Final
   docker image save quay.io/debezium/server:3.0.8.Final -o dbz3.0.8-xml-linux-amd.tar

code: sudo k3s ctr images import dbz3.0.8-xml-linux-amd.tar all

code:
sources:
  oracle:
    type: cdc
    logging:
      level: info
    connection:
      type: oracle
      host: oracle
      port: 1521
      user: ${SOURCE_DB_USERNAME}
      password: ${SOURCE_DB_PASSWORD}
      database: ORCLCDB
    advanced:
      source:
        database.pdb.name: ORCLPDB1
        lob.enabled: true

section_id: configure-rdi-for-xmltype-support

code:
CREATE TABLE tab1 (
  xmlid INT NOT NULL,
  col1  SYS.XMLTYPE,
  CONSTRAINT PK_tab1 PRIMARY KEY (xmlid)
);

DECLARE
  v_xml   SYS.XMLTYPE;
  v_doc   CLOB;
BEGIN
  -- XMLTYPE created from a CLOB
  v_doc := '<?xml version="1.0"?>' || Chr(10) || ' <TABLE_NAME>MY_TABLE</TABLE_NAME>';
  v_xml := SYS.XMLTYPE.createXML(v_doc);

  INSERT INTO tab1 (xmlid, col1) VALUES (1, v_xml);

  -- XMLTYPE created from a query
  SELECT SYS_XMLGEN(table_name)
  INTO   v_xml
  FROM   user_tables
  WHERE  rownum = 1;

  INSERT INTO tab1 (xmlid, col1) VALUES (2, v_xml);

  COMMIT;
END;
/

ALTER TABLE CHINOOK.TAB1 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

section_id: test-xmltype-support

code:
graph TB
    USERS["USERS

- CUSTOMERS table
- ORDERS table
- PRODUCTS table
- TRANSACTIONS table

(Application data)"]

code:
graph TB
    ADM["XSTREAM_ADM_TBS

- Capture process metadata
- Outbound server config
- Checkpoint information
- Position tracking"]
    USER_TS["XSTREAM_TBS

- LCR Queue change records
- Buffered changes
- Transaction state"]

code:
-- Application data
SELECT * FROM MY_APP.CUSTOMERS;
-- Returns: customer_id, name, address, balance, etc.

code:
-- XStream metadata (in XSTREAM_ADM_TBS)
SELECT capture_name, status, start_scn, checkpoint_scn
FROM dba_capture;
-- Returns: dbzxout_capture, ENABLED, 12345678, 12345690

-- XStream queue (in XSTREAM_TBS)
SELECT queue_name, enqueue_time, dequeue_time, state
FROM dba_queues;
-- Returns: dbzxout_queue, <timestamp>, <timestamp>, READY

code:
-- Typical size: Large
SELECT tablespace_name,
       ROUND(SUM(bytes)/1024/1024/1024, 2) AS size_gb
FROM dba_data_files
WHERE tablespace_name = 'USERS'
GROUP BY tablespace_name;

-- Result: 50 GB, 100 GB, 500 GB, etc.

code:
-- Typical size: Small
SELECT tablespace_name,
       ROUND(SUM(bytes)/1024/1024, 2) AS size_mb
FROM dba_data_files
WHERE tablespace_name IN ('XSTREAM_ADM_TBS', 'XSTREAM_TBS')
GROUP BY tablespace_name;

-- Result:
-- XSTREAM_ADM_TBS: 100-500 MB
-- XSTREAM_TBS: 500 MB - 2 GB (depends on queue size)

code:
graph LR
    RedoLogs["Redo Logs"]
    Capture["Capture Process"]
    Queue["Queue"]
    Outbound["XStream Outbound
Server"]
    Consumer["Debezium/RDI"]

    RedoLogs --> Capture
    Capture --> Queue
    Queue --> Outbound
    Outbound --> Consumer

section_id: what-is-the-xstream-outbound-server


title: Prepare PostgreSQL/Supabase for RDI

summary: Prepare PostgreSQL databases (including Supabase) to work with RDI

page_type: content

title: Overview

role: overview

text:
[Supabase](https://supabase.com/docs/guides/database/overview) uses PostgreSQL as
its database engine, so the instructions below also apply to Supabase. However, RDI
doesn't currently support cloud deployments of Supabase with AWS or GCP.

PostgreSQL supports several
[logical decoding plug-ins](https://wiki.postgresql.org/wiki/Logical_Decoding_Plugins)
to enable CDC. If you don't want to use the native `pgoutput` logical replication stream support
then you must install your preferred plug-in into the PostgreSQL server. Once you have done this,
you must enable a replication slot, and configure a user with privileges to perform the replication.

If you are using a service like [Heroku Postgres](https://www.heroku.com/postgres) to host
your database then this might restrict the plug-ins you can use. If you can't use your preferred
plug-in then could try the `pgoutput` decoder if you are using PostgreSQL 10 or above.
If this doesn't work for you then you won't be able to use RDI with your database.

The following checklist summarizes the steps to prepare a PostgreSQL
database for RDI, with links to the sections that explain the steps in
full detail. You may find it helpful to track your progress with the
checklist as you complete each step.

[code example]

title: Amazon RDS for PostgreSQL

role: content

text:
Follow the steps below to enable CDC with [Amazon RDS for PostgreSQL](https://aws.amazon.com/rds/postgresql/):

[code example]

1.
    Set the instance parameter `rds.logical_replication` to 1.

1.
    Check that the `wal_level` parameter is set to `logical` by running the query `SHOW wal_level`
    as the database RDS master user. The parameter might not have this value in multi-zone replication
    setups. You can't change the value manually but it should change automatically when you set the
    `rds.logical_replication` parameter to 1. If it doesn't change then you probably just need to
    restart your database instance. You can restart manually or wait until a restart occurs
    during your maintenance window.

1.
    Set the Debezium `plugin.name` parameter to `pgoutput`.

1.
    Initiate logical replication from an AWS account that has the `rds_replication` role. The role grants
    permissions to manage logical slots and to stream data using logical slots. By default, only the master user account on AWS has the `rds_replication` role on Amazon RDS, but if you have administrator privileges,
    you can grant the role to other accounts using a query like the following:

    [code example]

    To enable accounts other than the master account to create an initial snapshot, you must grant `SELECT`
    permission to the accounts on the tables to be captured. See the documentation about
    [security for PostgreSQL logical replication](https://www.postgresql.org/docs/current/logical-replication-security.html)
    for more information.

role: setup

text:
As of PostgreSQL 9.4, the only way to read changes to the write-ahead-log is to
[install a logical decoding output plug-in](https://debezium.io/documentation/reference/2.6/postgres-plugins.html).
These plug-ins are written in C using PostgreSQL-specific APIs, as described in the
[PostgreSQL documentation](https://www.postgresql.org/docs/current/logicaldecoding-output-plugin.html).
The PostgreSQL connector uses one of Debezium’s supported logical decoding
plug-ins to receive change events from the database in either the default
[`pgoutput`](https://github.com/postgres/postgres/blob/master/src/backend/replication/pgoutput/pgoutput.c) format (supplied with PostgreSQL) or the
[`Protobuf`](https://github.com/protocolbuffers/protobuf) format.
See the
[decoderbufs Protobuf plug-in documentation](https://github.com/debezium/postgres-decoderbufs)
for more details about how to compile it and also its requirements and limitations.

For simplicity, Debezium also provides a container image that compiles and installs the plug-ins
on top of the upstream PostgreSQL server image. Use this image as an example of the steps
involved in the installation.

 The Debezium logical decoding plug-ins have been tested on Linux machines, but if you are
using Windows or other operating systems, the installation steps might be different from
those listed here.

title: Plug-in differences

text:
Plug-ins don't all behave in exactly the same way. All of them refresh information about
the database schema when they detect that it has changed, but the `pgoutput` plug-in is
more "eager" than some other plug-ins to do this. For example, `pgoutput` will refresh
when it detects a change to the default value of a column but other plug-ins won't
notice this until another, more significant change happens (such as adding a new table
column).

The Debezium project maintains a
[Java class](https://github.com/debezium/debezium/blob/main/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/DecoderDifferences.java) that tracks the known differences between plug-ins.

title: Configure the PostgreSQL server

text:
If you want to use a logical decoding plug-in other than the default `pgoutput` then
you must first configure it in the `postgresql.conf` file. Set the `shared_preload_libraries`
parameter to load your plug-in at startup. For example, to load the `decoderbufs`
plug-in, you would add the following line:

[code example]

Add the line below to configure the replication slot (for any plug-in).
This instructs the server to use logical decoding with the write-ahead log.

[code example]

You can also set other PostgreSQL streaming replication parameters if you need them.
For example, you can use `max_wal_senders` and `max_replication_slots` to increase
the number of connectors that can access the sending server concurrently,
and `wal_keep_size` to limit the maximum WAL size that a replication slot retains.
The
[configuration parameters](https://www.postgresql.org/docs/current/runtime-config-replication.html#RUNTIME-CONFIG-REPLICATION-SENDER)
documentation describes all the parameters you can use.

PostgreSQL’s logical decoding uses replication slots. These are guaranteed to retain all the WAL
segments that Debezium needs even when Debezium suffers an outage. You should monitor replication
slots carefully to avoid excessive disk consumption and other conditions such as catalog bloat that can arise
if a replication slot is used infrequently. See the PostgreSQL documentation about
[replication slots](https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS)
for more information.
If you are using a `synchronous_commit` setting other than `on`, then you should set `wal_writer_delay`
to a value of about 10 milliseconds to ensure a low latency for change events. If you don't set this then
the default value of about 200 milliseconds will apply.

This guide summarizes the operation of the PostgreSQL write-ahead log, but we strongly
recommend you consult the [PostgreSQL write-ahead log](https://www.postgresql.org/docs/current/wal-configuration.html)
documentation to get a better understanding.

title: Set up permissions

text:
The Debezium connector needs a database user that has the REPLICATION and LOGIN roles so that it
can perform replications. By default, a superuser has these roles but for security reasons, you
should give the minimum necessary permissions to the Debezium user rather than full superuser
permissions.

If you have administrator privileges then you can create a role for your Debezium user
using a query like the following. Note that these are the *minimum* permissions the user
needs to perform replications, but you might also need to grant other permissions.

[code example]

title: Set privileges for Debezium to create PostgreSQL publications with `pgoutput`

text:
The Debezium user needs specific permissions to work with the `pgoutput` plug-in.
The plug-in captures change events from the
[*publications*](https://www.postgresql.org/docs/current/logical-replication-publication.html)
that PostgreSQL produces for your chosen source tables. A publication contains change events from
one or more tables that are filtered using criteria from a *publication specification*.

If you have administrator privileges, you can create the publication specification
manually or you can grant the Debezium user the privileges to create the specification
automatically. The required privileges are:

-   Replication privileges in the database to add the table to a publication.
-   `CREATE` privileges on the database to add publications.
-   `SELECT` privileges on the tables to copy the initial table data. Table owners
    automatically have `SELECT` permission for the table.

To add a table to a publication, the user must be an owner of the table. However, in
this case, the source table already exists, so you must use a PostgreSQL replication
group to share ownership between the Debezium user and the original owner. Configure
the replication group using the following commands:

[code example]

1.
    Create the replication group (the name `replication_group` here is
    just an example):

    [code example]

1.
    Add the original owner of the table to the group:

    [code example]

1.
    Add the Debezium replication user to the group:

    [code example]

1.
    Transfer ownership of the table to `replication_group`:

    [code example]

You must also set the value of the `publication.autocreate.mode` parameter to `filtered`
to allow Debezium to specify the publication configuration. See the
[Debezium documentation for `publication.autocreate.mode`](https://debezium.io/documentation/reference/2.6/connectors/postgresql.html#postgresql-publication-autocreate-mode)
to learn more about this setting.

text:
You must configure the database to allow replication with the host that runs
the PostgreSQL Debezium connector. To do this, add an entry to the
host-based authentication file, `pg_hba.conf`, for each client that needs to
use replication. For example, to enable replication for `<youruser>` locally,
on the server machine, you would add a line like the following:

[code example]

To allow `<youruser>` on localhost to receive replication changes using IPV4,
add the line:

[code example]

To allow `<youruser>` on localhost to receive replication changes using IPV6,
add the line:

[code example]

Find out more from the PostgreSQL pages about
[`pg_hba.conf`](https://www.postgresql.org/docs/10/auth-pg-hba-conf.html)
and
[network address types](https://www.postgresql.org/docs/current/datatype-net-types.html).

title: Supported PostgreSQL topologies

text:
You can use the Debezium PostgreSQL connector with a standalone PostgreSQL server or
with a cluster of servers.
For versions 12 and below, PostgreSQL supports logical replication slots on only primary servers.
This means that Debezium can only connect to a primary server for CDC and the connection will
stop if this server fails. If the same server is promoted to primary when service resumes
then you can simply restart the Debezium connector. However, if a different server is
promoted to primary, then you must reconfigure Debezium to use the new server
before restarting. Also, make sure the new server has the correct plug-in and configuration
for Debezium.

code:
- [ ] [Install the logical decoding output plug-in](#install-the-logical-decoding-output-plug-in)
- [ ] [Configure the PostgreSQL server](#configure-the-postgresql-server)
- [ ] [Set up permissions](#set-up-permissions)
- [ ] [Set privileges for Debezium to create PostgreSQL publications with pgoutput](#set-privileges-for-debezium-to-create-postgresql-publications-with-pgoutput)
- [ ] [Configure PostgreSQL for replication with the Debezium connector host](#configure-postgresql-for-replication-with-the-debezium-connector-host)

section_id: overview

language: checklist {id="postgresawslist" nointeractive="true" }

code:
- [ ] [Set the instance parameter rds.logical_replication to 1](#1-set-the-instance-parameter-rdslogicalreplication-to-1)
- [ ] [Check that the wal_level parameter is set to logical](#2-check-that-the-wal_level-parameter-is-set-to-logical)
- [ ] [Set the Debezium plugin.name parameter to pgoutput](#3-set-the-debezium-pluginname-parameter-to-pgoutput)
- [ ] [Initiate logical replication from an AWS account that has the rds_replication role](#4-initiate-logical-replication-from-an-aws-account-that-has-the-rdsreplication-role)

section_id: amazon-rds-for-postgresql

code: GRANT rds_replication TO <my_user>

code:
# MODULES
shared_preload_libraries = 'decoderbufs'

section_id: configure-the-postgresql-server

code:
# REPLICATION
wal_level = logical

code: CREATE ROLE <name> REPLICATION LOGIN;

section_id: set-up-permissions

language: checklist {id="postgrespgoutputlist" nointeractive="true" }

code:
- [ ] [Create the replication group](#1-create-the-replication-group)
- [ ] [Add the original owner of the table to the group](#2-add-the-original-owner-of-the-table-to-the-group)
- [ ] [Add the Debezium replication user to the group](#3-add-the-debezium-replication-user-to-the-group)
- [ ] [Transfer ownership of the table to the replication group](#4-transfer-ownership-of-the-table-to-the-replication-group)

code: CREATE ROLE replication_group;

code: GRANT replication_group TO original_owner;

code: GRANT replication_group TO replication_user;

code: ALTER TABLE table_name OWNER TO replication_group;

code: local   replication     <youruser>                          trust

code: host    replication     <youruser>  127.0.0.1/32            trust

code: host    replication     <youruser>  ::1/128                 trust


title: Prepare Spanner for RDI

summary: Prepare Google Cloud Spanner databases to work with RDI

title: Overview

text:
Google Cloud Spanner requires specific configuration to enable change data capture (CDC) with RDI.
RDI operates in two phases with Spanner: snapshot (initial sync) and streaming. During the snapshot
phase, RDI uses the JDBC driver to connect directly to Spanner and read the current state of the
database. In the streaming phase, RDI uses [Spanner's Change Streams](https://cloud.google.com/spanner/docs/change-streams) to capture changes related to
the monitored schemas and tables.

Spanner is only supported with RDI deployed on Kubernetes/Helm. RDI VM mode does not support Spanner as a source database.

The following checklist summarizes the steps to prepare a Spanner
database for RDI, with links to the sections that explain the steps in
full detail. You may find it helpful to track your progress with the
checklist as you complete each step.

[code example]

title: 1. Prepare for snapshot

text:
During the snapshot phase, RDI executes multiple transactions to capture data at an exact point
in time that remains consistent across all queries. This is achieved using a Spanner feature called
[Timestamp bounds with exact staleness](https://cloud.google.com/spanner/docs/timestamp-bounds#exact_staleness).

This feature relies on the
[version_retention_period](https://cloud.google.com/spanner/docs/reference/rest/v1/projects.instances.databases#Database.FIELDS.version_retention_period),
which is set to one hour by default. Depending on the database tier, the volume of data to be
ingested into RDI, and the load on the database, this setting may need to be increased. You can
update it using [this method](https://cloud.google.com/spanner/docs/use-pitr#set-period).

title: 2. Prepare for streaming

text:
To enable streaming, you must create a change stream in Spanner at the database level. Use the
option `value_capture_type = 'NEW_ROW_AND_OLD_VALUES'` to capture both the previous and updated
row values.

Be sure to specify only the tables you want to ingest from and, optionally, the specific columns
you're interested in. Here's an example using Google SQL syntax:

[code example]

Refer to the [official documentation](https://cloud.google.com/spanner/docs/change-streams/manage#googlesql)
for more details, including additional configuration options and dialect-specific syntax.

title: 3. Create a service account

text:
To allow RDI to access the Spanner instance, you'll need to create a service account with the
appropriate permissions. By default, RDI uses Google Cloud Workload Identity authentication. In this case RDI will assume the [service account is assigned to the GKE cluster](https://cloud.google.com/kubernetes-engine/docs/how-to/workload-identity#enable_on_clusters_and_node_pools). Alternatively, you can provide the
service account credentials as a Kubernetes secret (see step 4 for details).

[code example]

1.
   Create the service account

    [code example]

1.
   Grant required roles:

    **Database Reader** (read access to Spanner data):

    [code example]

    **Database User** (query execution and metadata access):

    [code example]

    **Viewer** (viewing instance and database configuration):

    [code example]

1.
   Download the service account key:

    Save the credentials locally so they can be used later by RDI:

    [code example]

title: Authentication methods

role: security

text:
RDI supports two authentication methods for accessing Spanner:

1. **Workload Identity (default)**: The service account is assigned to the GKE cluster, and RDI
   automatically uses the cluster's identity to authenticate. This is the recommended approach
   as it's more secure and doesn't require managing credential files.

2. **Service account credentials file**: You provide the service account key file as a Kubernetes
   secret. This method requires setting `use_credentials_file: true` in your RDI configuration.

title: 4. Set up secrets for Kubernetes deployment (optional)

text:
Before deploying the RDI pipeline, you need to configure the necessary secrets for the target
database. Instructions for setting up the target database secrets are available in the
[RDI deployment guide]().

**Optional**: If you prefer to use a service account credentials file instead of Workload Identity
authentication, you'll need to create a Spanner-specific secret named `source-db-credentials`.
This secret should contain the service account key file generated during the Spanner setup phase.
Use the command below to create it:

[code example]

Be sure to adjust the file path (`~/spanner-reader-account.json`) if your service account key is
stored elsewhere.

If you create the `source-db-credentials` secret, you must also set `use_credentials_file: true`
in your RDI configuration to use the credentials file instead of Workload Identity authentication.

title: 5. Configure RDI for Spanner

text:
When configuring your RDI pipeline for Spanner, use the following example configuration in your
`config.yaml` file:

[code example]

Make sure to replace the relevant connection details with your own for both the Spanner and target
Redis databases.

title: 6. Additional Kubernetes configuration

text:
In your `rdi-values.yaml` file for Kubernetes deployment, make sure to configure the `dataPlane`
section like this:

[code example]

title: 7. Configuration is complete

text: Once you have followed the steps above, your Google Spanner database is ready for RDI to use.

code:
- [ ] [Prepare for snapshot](#1-prepare-for-snapshot)
- [ ] [Prepare for streaming](#2-prepare-for-streaming)
- [ ] [Create a service account](#3-create-a-service-account)
- [ ] [Set up secrets for Kubernetes deployment (optional)](#4-set-up-secrets-for-kubernetes-deployment-optional)
- [ ] [Configure RDI for Spanner](#5-configure-rdi-for-spanner)
- [ ] [Additional Kubernetes configuration](#6-additional-kubernetes-configuration)

code:
CREATE CHANGE STREAM change_stream_table1_and_table2
  FOR table1, table2
  OPTIONS (
    value_capture_type = 'NEW_ROW_AND_OLD_VALUES'
  );

language: checklist {id="spanner-service-account" nointeractive="true" }

code:
- [ ] [Create the service account](#create-the-service-account)
- [ ] [Grant required roles](#grant-required-roles)
- [ ] [Download the service account key](#download-the-service-account-key)

code:
gcloud iam service-accounts create spanner-reader-account \
        --display-name="Spanner Reader Service Account" \
        --description="Service account for reading from Spanner databases" \
        --project=YOUR_PROJECT_ID

code:
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
        --member="serviceAccount:spanner-reader-account@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
        --role="roles/spanner.databaseReader"

code:
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
        --member="serviceAccount:spanner-reader-account@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
        --role="roles/spanner.databaseUser"

code:
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
        --member="serviceAccount:spanner-reader-account@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
        --role="roles/spanner.viewer"

code:
gcloud iam service-accounts keys create ~/spanner-reader-account.json \
        --iam-account=spanner-reader-account@YOUR_PROJECT_ID.iam.gserviceaccount.com \
        --project=YOUR_PROJECT_ID

code:
kubectl create secret generic source-db-credentials --namespace=rdi \
--from-file=gcp-service-account.json=~/spanner-reader-account.json \
--save-config --dry-run=client -o yaml | kubectl apply -f -

code:
sources:
  source:
    type: flink
    connection:
      type: spanner
      project_id: your-project-id
      instance_id: your-spanner-instance
      database_id: your-spanner-database
      # use_credentials_file: false  # Default: uses Workload Identity. Set to true to use service account credentials file instead
      change_streams:
        change_stream_all:
          {}
          # retention_hours: 24
    # schemas:
    #  - DEFAULT
    # tables:
    #   products: {}
    #   orders: {}
    #   order_items: {}
    # logging:
    #   level: debug
    # advanced:
    #   source:
    #     spanner.change.stream.retention.hours: 24
    #     spanner.fetch.timeout.milliseconds: 20000
    #     spanner.dialect: POSTGRESQL
    #   flink:
    #     jobmanager.rpc.port: 7123
    #     jobmanager.memory.process.size: 1024m
    #     taskmanager.numberOfTaskSlots: 3
    #     taskmanager.rpc.port: 7122
    #     taskmanager.memory.process.size: 2g
    #     blob.server.port: 7124
    #     rest.port: 8082
    #     parallelism.default: 4
    #     restart-strategy.type: fixed-delay
    #     restart-strategy.fixed-delay.attempts: 3
targets:
  target:
    connection:
      type: redis
      host: ${HOST_IP}
      port: 12000
      user: ${TARGET_DB_USERNAME}
      password: ${TARGET_DB_PASSWORD}
processors:
  target_data_type: hash

code:
operator:
  dataPlane:
    flinkCollector:
      enabled: true
      jobManager:
        ingress:
          enabled: true
          className: traefik # Replace with your ingress controller
          hosts:
            - hostname # Replace with your desired ingress hostname


title: Prepare SQL Server for RDI

summary: Prepare SQL Server databases to work with RDI

title: Overview

text:
To prepare your SQL Server database for Debezium, you must first create a dedicated Debezium user,
run a script to enable CDC globally, and then separately enable CDC for each table you want to
capture. You need administrator privileges to do this.

Once you enable CDC, it captures all of the INSERT, UPDATE, and DELETE operations
on your chosen tables. The Debezium connector can then emit these events to RDI.

The following checklist summarizes the steps to prepare a SQL Server
database for RDI, with links to the sections that explain the steps in
full detail. You may find it helpful to track your progress with the
checklist as you complete each step.

[code example]

title: 1. Create a Debezium user

text:
It is strongly recommended to create a dedicated Debezium user for the connection between RDI
and the source database. When using an existing user, ensure that the required
permissions are granted and that the user is added to the CDC role.

[code example]

1.
  Create the Debezium user with the Transact-SQL below:

    [code example]

    Replace `MyUser`, `My_Password` and `MyDB` with your chosen values.

1.
  Grant the user the necessary permissions:

    [code example]

title: 2. Enable CDC on the database

text:
There are two system stored procedures to enable CDC (you need
administrator privileges to run these). Use `sys.sp_cdc_enable_db`
to enable CDC for the whole database and then `sys.sp_cdc_enable_table` to enable CDC for individual tables.

Before running the procedures, ensure that:

- You are a member of the `sysadmin` fixed server role for the SQL Server.
- You are a `db_owner` of the database.
- The SQL Server Agent is running.

Then, assuming your database is called `MyDB`, run the script below to enable CDC:

[code example]

For SQL Server on AWS RDS, you must use a different stored procedure:
[code example]

When you enable CDC for the database, it creates a schema called `cdc` and also
a CDC user, metadata tables, and other system objects.

title: 3. Enable CDC for the tables you want to capture

text:
[code example]

1.
    You must also enable CDC on the tables you want Debezium to capture using the
    following commands (again, you need administrator privileges for this):

    [code example]

    Repeat this for every table you want to capture.

    The value for `@role_name` can’t be a fixed database role, such as `db_datareader`.
    Specifying a new name will create a corresponding database role that has full access to the
    captured change data.

1.
    Add the Debezium user to the CDC role:

    [code example]

title: 4. Check that you have access to the CDC table

text:
You can use another stored procedure `sys.sp_cdc_help_change_data_capture`
to query the CDC information for the database and check you have enabled
it correctly. To do this, connect as the Debezium user you created previously (`MyUser`).

[code example]

1.
  Run the `sys.sp_cdc_help_change_data_capture` stored procedure to query
    the CDC configuration. For example, if your database was called `MyDB` then you would
    run the following:

    [code example]

1.
    The query returns configuration information for each table in the database that
    has CDC enabled and that contains change data that you are authorized to
    access. If the result is empty then you should check that you have privileges
    to access both the capture instance and the CDC tables.

title: Troubleshooting

role: errors

text:
If no CDC is happening then it might mean that SQL Server Agent is down. You can check for this using the SQL query shown below:

[code example]

If the query returns a result of 0, you need to need to start SQL Server Agent using the following commands:

[code example]

title: SQL Server capture job agent configuration parameters

text:
In SQL Server, the parameters that control the behavior of the capture job agent
are defined in the SQL Server table `msdb.dbo.cdc_jobs`. If you experience performance
problems while running the capture job agent then you can adjust the capture jobs
settings to reduce CPU load. To do this, run the `sys.sp_cdc_change_job` stored procedure
with your new parameter values.

A full guide to configuring the SQL Server capture job agent parameters
is outside the scope of the Redis documentation.

The following parameters are the most important ones for modifying the capture agent behavior
of the Debezium SQL Server connector:

* `pollinginterval`: This specifies the number of seconds that the capture agent
  waits between log scan cycles. A higher value reduces the load on the database
  host, but increases latency.  A value of 0 specifies no wait between scans.
  The default value is 5.
* `maxtrans`: This specifies the maximum number of transactions to process during
  each log scan cycle. After the capture job processes the specified number of
  transactions, it pauses for the length of time that `pollinginterval` specifies
  before the next scan begins. A lower value reduces the load on the database host,
  but increases latency. The default value is 500.
* `maxscans`: This specifies a limit on the number of scan cycles that the capture
  job can attempt when capturing the full contents of the database transaction log.
  If the continuous parameter is set to 1, the job pauses for the length of time
  that the `pollinginterval` specifies before it resumes scanning. A lower values
  reduces the load on the database host, but increases latency. The default value is 10.

See the SQL Server documentation for more information about capture agent parameters.

title: SQL Server on Azure

text:
You can also use the Debezium SQL Server connector with SQL Server on Azure.
See Microsoft's guide to
[configuring SQL Server on Azure for CDC with Debezium](https://learn.microsoft.com/en-us/samples/azure-samples/azure-sql-db-change-stream-debezium/azure-sql%2D%2Dsql-server-change-stream-with-debezium/)
for more information.

title: Handling changes to the schema

text:
RDI can't adapt automatically when you change the schema of a CDC table in SQL Server. For example,
if you add a new column to a table you are capturing then RDI will generate errors
instead of capturing the changes correctly. See Debezium's
[SQL Server schema evolution](https://debezium.io/documentation/reference/stable/connectors/sqlserver.html#sqlserver-schema-evolution)
docs for more information.

If you have administrator privileges, you can follow the steps below to update RDI after
a schema change and resume CDC. See the
[online schema updates](https://debezium.io/documentation/reference/stable/connectors/sqlserver.html#online-schema-updates)
documentation for further details.

[code example]

1.
    Make your changes to the source table schema.

1.
  Create a new capture table for the updated source table by running the `sys.sp_cdc_enable_table` stored
    procedure with a new, unique value for the parameter `@capture_instance`. For example, if the old value
    was `dbo_MyTable`, you could replace it with `dbo_MyTable_v2` (you can see the existing values by running
    stored procedure `sys.sp_cdc_help_change_data_capture`):

    [code example]

1.
    When Debezium starts streaming from the new capture table, drop the old capture table by running
    the `sys.sp_cdc_disable_table` stored procedure with the parameter `@capture_instance` set to the old
    capture instance name, `dbo_MyTable`:

    [code example]

RDI will *not* correctly capture changes that happen in the time gap between changing
the source schema (step 1 above) and updating the value of `@capture_instance` (step 2).
Try to keep the gap as short as possible or perform the update at a time when you expect
few changes to the data.

code:
- [ ] [Create a Debezium user](#1-create-a-debezium-user)
- [ ] [Enable CDC on the database](#2-enable-cdc-on-the-database)
- [ ] [Enable CDC for the tables you want to capture](#3-enable-cdc-for-the-tables-you-want-to-capture)
- [ ] [Check that you have access to the CDC table](#4-check-that-you-have-access-to-the-cdc-table)

language: checklist {id="sqlserver-create-debezium-user" nointeractive="true" }

code:
- [ ] [Create the Debezium user](#create-the-debezium-user)
- [ ] [Grant the user the necessary permissions](#grant-the-user-the-necessary-permissions)

code:
USE master
    GO
    CREATE LOGIN MyUser WITH PASSWORD = 'My_Password'
    GO
    USE MyDB
    GO
    CREATE USER MyUser FOR LOGIN MyUser
    GO

code:
USE master
    GO
    GRANT VIEW SERVER STATE TO MyUser
    GO
    USE MyDB
    GO
    EXEC sp_addrolemember N'db_datareader', N'MyUser'
    GO

code:
USE MyDB
GO
EXEC sys.sp_cdc_enable_db
GO

code:
EXEC msdb.dbo.rds_cdc_enable_db 'Chinook'
GO

language: checklist {id="sqlserver-enable-cdc-tables" nointeractive="true" }

code:
- [ ] [Enable CDC on the tables you want to capture](#enable-cdc-on-the-tables-you-want-to-capture)
- [ ] [Add the Debezium user to the CDC role](#add-the-debezium-user-to-the-cdc-role)

code:
USE MyDB
    GO

    EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name   = N'MyTable',
    @role_name     = N'MyRole',
    @supports_net_changes = 0
    GO

code:
USE MyDB
    GO
    EXEC sp_addrolemember N'MyRole', N'MyUser'
    GO

language: checklist {id="sqlserver-check-cdc-table" nointeractive="true" }

code:
- [ ] [Run the stored procedure to query the CDC configuration](#run-the-stored-procedure-to-query-the-cdc-configuration)
- [ ] [Check the results](#check-the-results)

code:
USE MyDB;
    GO
    EXEC sys.sp_cdc_help_change_data_capture
    GO

code:
IF EXISTS (SELECT 1
           FROM master.dbo.sysprocesses
           WHERE program_name = N'SQLAgent - Generic Refresher')
BEGIN
  SELECT @@SERVERNAME AS 'InstanceName', 1 AS 'SQLServerAgentRunning'
END
ELSE
BEGIN
  SELECT @@SERVERNAME AS 'InstanceName', 0 AS 'SQLServerAgentRunning'
END

section_id: troubleshooting

code:
EXEC xp_servicecontrol N'START',N'SQLServerAGENT';
GO

language: checklist {id="sqlserver-schema-changes" nointeractive="true" }

code:
- [ ] [Make your changes to the source table schema](#make-your-changes-to-the-source-table-schema)
- [ ] [Create a new capture table for the updated source table](#create-a-new-capture-table-for-the-updated-source-table)
- [ ] [Drop the old capture table](#drop-the-old-capture-table)

section_id: handling-changes-to-the-schema

code:
EXEC sys.sp_cdc_enable_table
    @source_schema    = N'dbo',
    @source_name      = N'MyTable',
    @role_name        = N'MyRole',
    @capture_instance = N'dbo_MyTable_v2',
    @supports_net_changes = 0
    GO

code:
EXEC sys.sp_cdc_disable_table
    @source_schema    = N'dbo',
    @source_name      = N'MyTable',
    @capture_instance = N'dbo_MyTable'
    GO


title: Formatting date and time values

title: Overview

text: The way you format date and time values depends on the source database, the data type of the field, and how it is represented in the incoming record. Below are some examples for different databases and data types.

title: Oracle

text:
Oracle supports the following date and time data types:

- `DATE` - represented by Debezium as a 64-bit integer representing the milliseconds since epoch
  [code example]
- `TIMESTAMP` - the value is represented by Debezium as a 64-bit integer and depends on the number of decimal places of precision of the column, representing fractions of a second. For example, if the column is defined as `TIMESTAMP(6)`, there are six decimal places and so the value is represented as microseconds since epoch (since there are 10^6 microseconds in each second).
You can format it similarly to `DATE`, but you need to divide the value by the appropriate factor based on the precision.

- `TIMESTAMP WITH TIME ZONE` - the value is represented as a string containing the timestamp and time zone.

- `TIMESTAMP WITH LOCAL TIME ZONE` - the value is represented as a string containing the timestamp and local time zone.

  SQLite supports both `TIMESTAMP WITH TIME ZONE` and `TIMESTAMP WITH LOCAL TIME ZONE`. You can format them using the `STRFTIME` function.

  [code example]

----

title: SQL Server

text:
SQL Server supports the following date and time data types:

- `date` - represented by Debezium as number of days since epoch (1970-01-01). You can multiply the value by 86400 (the number of seconds in a day) to convert it to seconds since epoch and then use the `STRFTIME` or `DATE` functions to format it.
  [code example]

- `datetime`, `smalldatetime` - represented by Debezium as number of milliseconds since epoch. Divide the value by 1000 to convert it to seconds since epoch and then use the `STRFTIME` function to format it.
  [code example]

- `datetime2` - similar to `datetime` but with higher precision. For `datetime2(0-3)`, the representation is the same as for `datetime`. For `datetime2(4-6)`, it is the number of microseconds since epoch. For `datetime2(7)`, it is the number of nanoseconds since epoch. To convert to another time unit, you can use the same approach as for `datetime` but you need to divide by 1000, 1000000 or 1000000000 depending on the precision.

- `time` - the number of milliseconds since midnight.
  [code example]

- `datetimeoffset` - represented as a timestamp with timezone information (for example, `2025-05-27T15:21:42.864Z` or `2025-01-02T14:45:30.123+05:00`).
  [code example]

<!-- TODO [ilianiliev-redis]: Test and document the dynamic expressions for the rest of the supported databases - MySQL, PostgresSQL, MongoDB -->

----

title: PostgreSQL

text:
PostgreSQL supports the following date and time data types:

- `date` - represented by Debezium as number of days since epoch (1970-01-01). You can multiply the value by 86400 (the number of seconds in a day) to convert it to seconds since epoch and then use the `STRFTIME` or `DATE` functions to format it.
  [code example]

- `time` - the time of microseconds since midnight.
  [code example]

- `time with time zone` - a string representation of the time with timezone information, where the timezone is GMT (for example, `07:15:00Z`).
  [code example]

- `timestamp` - represented by Debezium as a 64-bit integer containing the microseconds since epoch. You can use the `STRFTIME` function to format it.
  [code example]

- `timestamp with time zone` - represented by Debezium as a string containing the timestamp with time zone information, where the timezone is GMT (for example, `2025-06-07T10:15:00.000000Z`).
  [code example]

code:
name: Format Oracle DATE field
  transform:
  - uses: add_field
    with:
      fields:
        - field: formatted_date
          language: sql
          # Date is stored as a Unix timestamp in milliseconds so you need to
          # divide it by 1000 to convert it to seconds.
          expression: STRFTIME('%Y-%m-%d %H:%M:%S', DATE / 1000, 'unixepoch')
          # Example: 1749047572000 is transformed to 2025-06-04 14:32:52

section_id: oracle

code:
name: Format Oracle TIMESTAMP WITH TIME ZONE
  transform:
    - uses: add_field
      with:
        fields:
          - field: seconds_since_epoch
            language: sql
            # Convert the timestamp with local time zone to seconds since epoch.
            expression: STRFTIME('%s', TIMESTAMP_FIELD)

          - field: date_from_timestamp
            language: sql
            # Convert the timestamp with local time zone to date and time.
            expression: STRFTIME('%Y-%m-%d %H:%M:%S', TIMESTAMP_FIELD)

code:
name: Format SQL Server date field
  transform:
    - uses: add_field
      with:
        fields:
          - field: with_default_date_format
            language: sql
            # Uses the default DATE format
            expression: DATE(event_date * 86400, 'unixepoch')

          - field: with_custom_date_format
            language: sql
            # Uses the default DATE format
            expression: STRFTIME('%Y/%m/%d', event_date * 86400, 'unixepoch')

section_id: sql-server

code:
name: Format SQL Server datetime field
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_datetime
            language: sql
            expression: STRFTIME('%Y-%m-%d %H:%M:%S', event_datetime / 1000, 'unixepoch')

code:
name: Format SQL Server time field
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_time
            language: sql
            expression: TIME(event_time, 'unixepoch', 'utc')

code:
name: Format SQL Server datetimeoffset field
  transform:
  - uses: add_field
    with:
      fields:
        - field: formatted_datetimeoffset
          language: sql
          expression: STRFTIME('%Y-%m-%d %H:%M:%S', event_datetimeoffset)

code:
name: Format PostgreSQL date field
    transform:
        - uses: add_field
          with:
            fields:
              - field: with_default_date_format
                language: sql
                # Uses the default DATE format
                expression: DATE(event_date * 86400, 'unixepoch')

section_id: postgresql

code:
name: Format PostgreSQL time field
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_time
            language: sql
            # Divide by 1000000 to convert microseconds to seconds
            expression: TIME(event_time / 1000000, 'unixepoch', 'utc')

code:
name: Format PostgreSQL time with time zone
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_time_with_tz
            language: sql
            expression: STRFTIME('%H:%M:%S', event_time_with_time_zone)

code:
name: Format PostgreSQL timestamp field
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_timestamp
            language: sql
            # Divide by 1000000 to convert microseconds to seconds
            expression: STRFTIME('%Y-%m-%d %H:%M:%S', event_timestamp / 1000000, 'unixepoch')

code:
name: Format PostgreSQL timestamp with time zone
  transform:
    - uses: add_field
      with:
        fields:
          - field: formatted_timestamp_with_tz
            language: sql
            # Divide by 1000000 to convert microseconds to seconds
            expression: STRFTIME('%Y-%m-%d %H:%M:%S', event_timestamp_with_time_zone)