Oracle XStream

7.3 and later

The Oracle XStream origin uses Oracle XStream to process change data capture (CDC) information that is stored in Oracle redo logs. For information about supported versions, see Supported Systems and Versions in the Data Collector documentation.

The Oracle XStream origin connects to an Oracle XStream outbound server and generates a record for each DML event that is fetched from the server. Use this origin when you want to use Oracle XStream to process high volumes of Oracle change data. To use LogMiner to process change data, try the Oracle CDC origin.

The Oracle XStream origin can use existing XStream components or can provision the necessary XStream components to access the data.

The origin includes the CRUD operation type in a record header attribute so generated records can be easily processed by CRUD-enabled destinations. For an overview of Data Collector changed data processing and a list of CRUD-enabled destinations, see Processing changed data.

When you configure the Oracle XStream origin, you configure the operation mode to use existing components or to provision components, then you define related properties. You specify connection information and optionally configure properties to tune performance. You can also use a connection to configure the origin.

The origin can generate events for an event stream. For more information about dataflow triggers and the event framework, see Dataflow triggers overview.

Before you use the Oracle XStream origin, you must complete the required Oracle and Data Collector prerequisite tasks.

Oracle prerequisites

Complete the following prerequisite tasks in Oracle before you use the Oracle XStream origin:

  1. Set database parameters
  2. Create the CDC user
  3. Create the outbound server (existing XStream components only)
  4. Start the outbound server (existing XStream components only)
  5. Prepare the database and tables
  6. Run verification commands
Note: This documentation includes commands that can help you perform the required tasks if your source data resides in a pluggable database. The required commands can differ if you want to use a different database type or based on your Oracle environment and configuration. For definitive steps and commands, see the Oracle documentation and contact your database administrator (DBA).
This documentation uses the following terms, which comply with the Oracle documentation at this time:
  • CDB or multitenant database - A multitenant container database (CDB) that includes one or more pluggable databases (PDB).
  • PDB or pluggable database - A database that is part of a CDB.

Requirements

The following table lists the requirements for using the Oracle XStream origin to process Oracle change capture data in a pluggable database. For other database types, see the Oracle documentation for details on performing similar tasks.

You can use the following verification commands to determine whether a pluggable database meets all requirements.
Requirement Sample verification command Expected result
Supported version of Oracle

SELECT banner FROM v$version WHERE ROWNUM = 1;

Supported version of Oracle. For more information, see Supported Systems and Versions in the Data Collector documentation.

Container database (CDB)

SELECT CDB FROM v$database;

Yes
Target pdb exists

SELECT name, open_mode FROM v$pdbs;

Name of the pdb
Target pdb is OPEN READ WRITE

SELECT open_mode FROM v$pdbs WHERE name='<pdb_name>';

READ WRITE
Database is in ARCHIVELOG mode SELECT log_mode FROM v$database; ARCHIVELOG
If needed, enable ARCHIVELOG mode. For example, you might use the following commands to enable ARCHIVELOG mode for a pluggable database:
  1. Connect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Run the following commands to enable ARCHIVELOG mode:
    SHUTDOWN IMMEDIATE;
    STARTUP MOUNT;
    ALTER DATABASE ARCHIVELOG;
    ALTER DATABASE OPEN;

Command tools and syntax

The prerequisite tasks provide commands for SQL*Plus. If you prefer to use a JDBC SQL client tool, use the following replacement commands:
SQL*Plus syntax JDBC SQL syntax
EXEC procedure(); BEGIN procedure(); END;
Use a slash (/) after PL/SQL blocks Omit the slash, simply run the PL/SQL blocks
Run the entire block of code at once Execute each statement separately

Task 1. Set database parameters

Use the following steps to enable the database to use GoldenGate replication and generate supplemental log data for change data capture.
  1. Connect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Use the following commands to enable GoldenGate replication, configure the streams pool, and enable writing change data to redo logs:
    ALTER SYSTEM SET enable_goldengate_replication=TRUE SCOPE=BOTH;
    ALTER SYSTEM SET streams_pool_size=256M SCOPE=BOTH;
    ALTER DATABASE FORCE LOGGING;
    ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

Task 2. Create the CDC user

Use the following steps to create a common user to use to access CDC data and to grant the user all necessary access and privileges.
  1. If necessary, reconnect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Use the following commands to create a user and password:
    CREATE USER <cdc_user> IDENTIFIED BY "<password>"
      DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS CONTAINER=ALL;
    Note: According to current Oracle guidelines, the username must begin with C## or c##.
  3. If the source data is in a pluggable database, use the following commands to provide the required access to the database for the new CDC user:
    ALTER USER <cdc_user> SET CONTAINER_DATA = (CDB$ROOT, <pdb>) FOR CDB_TAB_COLUMNS CONTAINER = CURRENT; 
    ALTER USER <cdc_user> SET CONTAINER_DATA = (CDB$ROOT, <pdb>) FOR CDB_CONSTRAINTS CONTAINER = CURRENT; 
    ALTER USER <cdc_user> SET CONTAINER_DATA = (CDB$ROOT, <pdb>) FOR CDB_CONS_COLUMNS CONTAINER = CURRENT;
  4. Use the following commands to create a session, switch to the appropriate container, and grant the required privileges:
    GRANT CREATE SESSION, SET CONTAINER TO <cdc_user> CONTAINER=ALL;
    GRANT LOGMINING TO <cdc_user> CONTAINER=ALL;
    
    BEGIN
      DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
        grantee                 => '<cdc_user>',
        privilege_type          => 'CAPTURE',
        grant_select_privileges => TRUE,
        container               => 'ALL'
      );
    END;
    /
    
    BEGIN
      DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
        grantee                 => '<cdc_user>',
        privilege_type          => 'APPLY',
        grant_select_privileges => TRUE,
        container               => 'ALL'
      );
    END;
    /
    
    GRANT EXECUTE ON DBMS_XSTREAM_ADM TO <cdc_user>;
    GRANT EXECUTE ON DBMS_CAPTURE_ADM TO <cdc_user>;
    GRANT EXECUTE ON DBMS_AQADM TO <cdc_user>;
    GRANT EXECUTE ON DBMS_AQ TO <cdc_user>;
    When using a JDBC SQL client tool, omit the slash.

Task 3. Create the outbound server (existing components only)

When you use existing XStream components instead of having the origin provision XStream components, you must create an outbound server. If you configure the origin to provision XStream components, you can skip this task.

When you create the outbound server, you can perform a global capture to include all tables or you can perform a selective capture to include selected tables.

Include all tables

Use the following steps to create an outbound server that uses global capture to include all existing tables.

  1. If necessary, reconnect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Use the following commands to include all tables in the outbound server.
    Important: Initialize arrays with a null element at index 1. Passing null directly can cause errors.
    DECLARE
      l_tables  DBMS_UTILITY.UNCL_ARRAY;
      l_schemas DBMS_UTILITY.UNCL_ARRAY;
    
    BEGIN
      
      l_tables(1)  := NULL;
      l_schemas(1) := NULL;
    
      DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
        server_name           => '<outbound_server_name>',
        source_container_name => '<target_pdb>',
        table_names           => l_tables,
        schema_names          => l_schemas,
        connect_user          => '<cdc_user>',
        include_dml           => TRUE,
        include_ddl           => TRUE
      );
    END;
    /
    When using a JDBC SQL client tool, omit the slash.

Include specified tables

Use the following steps to create an outbound server that uses selective capture to include only the listed tables.

  1. If necessary, reconnect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Use the following commands to include the listed tables and schemas in the outbound server:
    Add additional l_tables(<number>) := '<schema_name>.<table_name>'; lines for each table that you want to include.
    Important: Initialize arrays with a null element at index 1. Passing null directly can cause errors.
    DECLARE
      l_tables  DBMS_UTILITY.UNCL_ARRAY;
      l_schemas DBMS_UTILITY.UNCL_ARRAY;
    BEGIN
      l_tables(1) := '<schema_name>.<table_name>';
      l_tables(2) := '<schema_name>.<table_name>'; 
      l_schemas(1) := NULL; 
    
      DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
        server_name           => '<outbound_server_name>',
        source_container_name => '<target_pdb>',
        table_names           => l_tables,
        schema_names          => l_schemas,
        connect_user          => '<cdc_user>',
        include_dml           => TRUE,
        include_ddl           => FALSE
      );
    END;
    /
    When using a JDBC SQL client tool, omit the slash.

Task 4. Start the outbound server (existing components only)

When you use existing XStream components instead of having the origin provision XStream components, the outbound server must be running before you run the pipeline. If you configure the origin to provision XStream components, you can skip this task.

Use the following steps to start the outbound server.

  1. If necessary, reconnect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. Use one of the following commands to start the outbound server:
    SQL*Plus syntax:
    EXEC DBMS_XSTREAM_ADM.START_OUTBOUND('<outbound_server_name>');
    JDBC SQL client syntax:
    BEGIN
      DBMS_XSTREAM_ADM.START_OUTBOUND(server_name => '<outbound_server_name>');
    END;

Task 5. Prepare the database and tables

Use the following steps to prepare tables in the database for processing.

  1. Use the following command to connect to a pluggable database:
    ALTER SESSION SET CONTAINER = <target_pdb>;
  2. Use the following command to enable supplemental logging for the specified table. Repeat this command for each table that you want to use.
    ALTER TABLE <schema>.<table> ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
  3. Use one of the following commands to prepare the specified table for instantiation. Repeat this command for each table that you want to use.
    SQL*Plus syntax:
    EXEC DBMS_CAPTURE_ADM.PREPARE_TABLE_INSTANTIATION('<schema>.<table>','all');
    JDBC SQL client syntax:
    BEGIN
      DBMS_CAPTURE_ADM.PREPARE_TABLE_INSTANTIATION(
        table_name          => '<schema>.<table>',
        supplemental_logging => 'all'
      );
    END;
  4. Use the following command to grant the CDC user access to the table. Repeat for every table that you want to use.
    GRANT SELECT, FLASHBACK ON <schema>.<table> TO <cdc_user>;
  5. Use the following command to reconnect to the CBD root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;

Task 6. Run verification commands

Use the following commands to verify that the prerequisite tasks were completed.

  1. If necessary, reconnect to the CDB root container CDB$ROOT with SYSDBA privileges:
    ALTER SESSION SET CONTAINER = CDB$ROOT;
  2. When using existing XStream components, use the following command to check the status of the outbound server:
    SELECT server_name, connect_user, capture_name FROM all_xstream_outbound;
  3. Use the following command to verify that the capture process is enabled:
    SELECT capture_name, status FROM dba_capture WHERE capture_name LIKE 'CAP$_SSTREAMS%';
  4. Use the following command to verify the CDC user privileges:
    SELECT username, privilege_type, grant_select_privileges FROM dba_xstream_administrator WHERE username='<cdc_user>';
  5. Use the following command to verify the database parameters:
    SELECT name, value FROM v$parameter WHERE name IN ('enable_goldengate_replication','streams_pool_size');

Data Collector prerequisite

Complete the following Data Collector prerequisite task to allow the Oracle XStream origin to use Oracle Instant Client libraries to access Oracle.

The Oracle XStream origin uses Oracle Instant Client libraries to access Oracle. Docker deployments include the Instant Client libraries and require no additional prerequisite tasks.

If you use tarball deployments, perform the following steps:
  1. Download and install Oracle Instant Client to the Data Collector engine machine.

    For more information, see the Oracle Instant Client Downloads page.

  2. From the Oracle Instant Client installation location, copy the xstream.jar and ojdbc<version>.jar files, and then install them as external resources for the Oracle XStream origin.

    Best practice is to use an archive file for the external resources. For information about installing external resources, see the Control Hub documentation.

  3. Add the following lines to the $SDC_INSTALLATION/libexec/sdc-env.sh file:
    • export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:"<instantclient_installdir>"
    • export ORACLE_INSTANTCLIENT_HOME="<instantclient_installdir>"
  4. Install additional dependencies required by Oracle XStream. Use the command appropriate for your Linux distribution:

    For Red Hat Enterprise: sudo yum install libaio

    For Ubuntu or Debian: sudo apt-get install libaio1 libnsl2

Operation mode

The operation mode determines how the Oracle XStream origin accesses Oracle data. The origin provides the following modes:
Provision new XStream components
The origin provisions all XStream components that are required to process changed data. Use this option for early development or testing.
When you use this option, you define the start mode to use and specify whether the query is case-sensitive for schema and table names.
You also define the schemas and tables to be included in the read. In the table filter properties, you can specify individual schema and table names or you can use regular expressions to define groups of schemas and tables. You can also use regular expressions to exclude groups of schemas or tables. You can configure multiple sets of properties to define the data to read.
For example, to process all tables in the Sales schema except the EU table, you can define the table filter properties as follows:
  • Schema inclusion pattern: Sales
  • Schema exclusion pattern: Use default value, ^$, which excludes nothing
  • Table inclusion pattern: Use default value, ^.*$, which reads all tables in the schema
  • Table exclusion pattern: EU
Before you use this mode, complete the required subset of Oracle prerequisite tasks and the additional prerequisite tasks.
Use existing XStream components
The origin uses existing XStream components to process changed data. Use this option for testing and production to can ensure that the origin processes the exact data set that you want.
When you use this option, you define name of the XStream outbound server and the start mode to use.
Before you use this mode, complete all Oracle and Data Collector prerequisite tasks.

Start mode

When the Oracle XStream origin uses existing XStream components to process data, you specify where the origin starts processing by configuring the Start Mode property and related settings. You can configure the origin to start at the current change or instant in time so that it processes all available changes. Or, you can specify a specific change or time to start processing.

The origin uses the Start Mode and any specified initial change or time only after you start the pipeline for the first time or after you reset the origin. At other times, the origin uses the last-saved offset to ensure that it processes all appropriate data.

Note that the Oracle XStream origin processes only change capture data. To process existing data, you might use a JDBC Multitable Consumer origin in a separate pipeline to read table data before you start an Oracle XStream pipeline.

The origin provides the following start modes:

From the latest change
The origin processes all changes that occur after you start the pipeline.
From a specified change
The origin processes all changes that occurred in the specified change (SCN) and later. When the origin uses the SCN, the origin started processing with the timestamp that is associated with the SCN. If the SCN cannot be found in the redo logs, the origin continues reading from the next higher SCN that is available in the redo logs.
Typically, a database admin can provide the SCN to use.
From a specified datetime
The origin processes all changes that occurred at the specified datetime and later. Use the following format: YYYY-MM-DD HH24:MM:SS.

CRUD operation header attributes

The Oracle XStream origin specifies the operation type for generated records in the following record header attributes:
sdc.operation.type
The origin evaluates the operation type associated with each entry that it processes. When appropriate, it writes the operation type to the sdc.operation.type record header attribute.
The origin uses the following values in the attribute to represent the operation type:
  • 1 for INSERT
  • 2 for DELETE
  • 3 for UPDATE

If you use a CRUD-enabled destination in the pipeline such as JDBC Producer or Elasticsearch, the destination can use the operation type when writing to destination systems. When necessary, you can use an Expression Evaluator processor or any scripting processor to manipulate the value in the header attribute. For an overview of Data Collector changed data processing and a list of CRUD-enabled destinations, see Processing changed data.

When using CRUD-enabled destinations, the destination looks for the operation type in this attribute before checking the following attribute.
oracle.cdc.operation
The origin also writes the CRUD operation type to an oracle.cdc.operation attribute. This attribute was implemented in an earlier release and is supported for backward compatibility.
The origin uses the following values in the attribute to represent the operation type:
  • INSERT
  • UPDATE
  • DELETE
CRUD-enabled destinations check this attribute for the operation type only if the sdc.operation.type attribute is not set.

CDC header attributes

In addition to the CRUD operation header attributes, Oracle XStream origin provides the following CDC record header attributes for each record:
  • schema - Schema where the change occurred
  • oracle.cdc.table - Table where the change occurred.
  • oracle.cdc.xid - Transaction ID for the change.
  • oracle.cdc.scn - System Change Number for the change.
  • oracle.cdc.timestamp - Timestamp of the DDL record.

You can use the record:attribute or record:attributeOrDefault functions to access the information in the attributes. For more information about working with record header attributes, see Working with header attributes.

Field attributes

The Oracle XStream origin can generate the following field attributes, when appropriate:
  • lob_value_omitted - Set to true when the origin encounters an object larger than the configured Max large object size property.
  • nanoSeconds - Includes the nanosecond value from timestamp data.
  • old_value - Includes the original value of the column for columns that were updated.
  • oracle_type - Provides the original Oracle data type for each field.
  • oracle_type_code - Provides the code for the original Oracle data type for each field.

You can use the record:fieldAttribute or record:fieldAttributeOrDefault functions to access the information in the attributes. For more information about working with field attributes, see Field attributes.

Event generation

The Oracle XStream origin can generate events that you can use in an event stream when the origin receives data from new tables and when existing tables change.
Important: To create table-related events, the XStream outbound server and capture process must be configured to capture DDL events.
Oracle XStream events can be used in any logical way. For example:

For more information about dataflow triggers and the event framework, see Dataflow triggers overview.

Event records

All event records generated by the Oracle XStream origin include the following event-related record headers:
Record header attribute Description
sdc.event.type Event type. Uses one of the following types:
  • ALTER - Generated when the origin receives a DDL event that indicates an existing table has changed.
  • CREATE - Generated when the origin receives a DDL event that indicates a new table has been created.
  • DROP - Generated when the origin receives a DDL event that indicates a table has been dropped.
  • TRUNCATE - Generated when the origin receives a DDL event that indicates a table has been truncated.
  • CLEAN_UP_SUCCESS - Generated when provisioned components are successfully shut down and deleted.
  • CLEAN_UP_FAILURE - Generated when provisioned components are not successfully shut down and deleted.
sdc.event.version Integer that indicates the version of the event record type.
sdc.event.creation_timestamp Epoch timestamp when the stage created the event.
DDL-generated events
The DDL-generated events - alter table, create table, drop table, and truncate table - are created when the origin reads a DDL statement.
DDL-generated events include the following additional record header attributes in event records:
Event header attribute Description
oracle.cdc.timestamp Timestamp of the originating DDL operation.
schema Originating schema name.
oracle.cdc.table Originating source table name.
oracle.cdc.xid Originating transaction ID.
oracle.cdc.scn Originating system changes number.
DDL-generated events also include the following field in event records:
Event record field Description
SQL SQL statement associated with the DDL event.
Cleanup success events
The origin creates a cleanup success event when you stop a pipeline that provisions XStream components, and the provisioned components are successfully shut down and deleted.
The cleanup success event does not include record fields. It includes only the following record header attribute:
Event header attribute Description
component_name Component that was successfully shut down and deleted. Can include the following values: CAPT, OUTB, QUEUE, QTBL.
Cleanup failure events
The origin creates a cleanup failure event when you stop a pipeline that provisions XStream components, and the provisioned components are not successfully shut down and deleted.
The cleanup failure event does not include record fields. It includes only the following record header attribute:
Event header attribute Description
component_drop_error Error message associated with the failure.

Configuring an Oracle XStream origin

Configure an Oracle XStream origin to use XStream to read Oracle CDC data from multiple tables.

  1. In the Properties panel, on the General tab, configure the following properties:
    General Property Description
    Name Stage name.
    Description Optional description.
    Produce Events Generates event records when events occur. Use for event handling.
    On Record Error Error record handling for the stage:
    • Discard - Discards the record.
    • Send to Error - Sends the record to the pipeline for error handling.
    • Stop Pipeline - Stops the pipeline.
  2. On the Oracle tab, configure the following properties:
    Oracle Property Description
    Operation mode Determines whether the origin provisions the XStream components to use for the read or uses existing components.
    Table filters Defines the schemas and tables to include in the read. Click Add to add one or more sets of the following properties:
    • Schema inclusion pattern - Defines the schemas to read. Specify a single schema or a regular expression to represent a group of schemas to read.

      Default is ^.*$, which includes all schemas in the database.

    • Schema exclusion pattern - Defines the schemas to exclude. Specify a single schema or a regular expression to represent a group of schemas to exclude from the read.

      Default is ^$, which excludes nothing.

    • Table inclusion pattern - Defines the tables to read in the specified schemas. Specify a single table or a regular expression to represent a group of tables to read.

      Default is ^.*$, which includes all tables in the specified schemas.

    • Table exclusion pattern - Defines the tables to exclude. Specify a single table or a regular expression to represent a group of tables to exclude from the read.

      Default is ^$, which excludes nothing.

    Available when the origin provisions XStream components.

    Case-sensitive names Determines whether the read query is case-sensitive for schema and table names. By default, the query is not case-sensitive.

    Available when the origin provisions XStream components.

    Outbound server name Name of the outbound server.

    Available when the origin uses existing XStream components.

    Start mode Starting point for the read. The origin starts processing from the specified start change or time when you start the pipeline for the first time or after you reset the origin.

    Use one of the following options:

    • From current change - Processes changes that arrive after you start the pipeline.
    • From specified change - Processes changes from the specified system change number (SCN).
    • From specified instant - Processes changes from the specified timestamp.

    Available when the origin uses existing XStream components.

    Initial system change number Initial system change number to use.

    Available when the origin reads from a specified change.

    Initial instant Start time for processing. Use the following format: YYYY-MM-DD HH24:MM:SS.

    Available when the origin reads from a specified instant.

  3. On the Connection tab, configure the following properties:
    Connection Property Description
    Connection

    7.5 and later

    Connection that defines the information that is required to connect to an external system.

    To connect to an external system, you can select a connection that contains the details, or you can directly enter the details in the pipeline. When you select a connection, Control Hub hides other properties so that you cannot directly enter connection details in the pipeline.

    JDBC Connection String Connection string used to connect to the database.

    Use the following format: jdbc:oracle:oci@<alias>.

    For example: jdbc:oracle:oci:@//myhost:3001/myservice.

    Username Username to connect to the database. Use the CDC user that you created in the prerequisite steps.
    Password Password for the user account.
  4. On the Tuning tab, configure the following properties as needed:
    Tuning property Description
    Max Batch Size Maximum number of records processed at one time. Honors values up to the Data Collector maximum batch size.

    Default is 1000. The Data Collector default is 1000.

    To increase throughput for frequent operations with minimal column data, consider increasing this setting to 10,000-50,000 records.

    For operations that affect many columns or include large objects, a smaller batch size might be more appropriate.

    Max large object size (MB) Maximum size of data to process for each record.

    When encountering larger objects, the origin sets the field value to null and sets a lob_value_omitted record header attribute to true.

    Default is 10 MB.