anchor.diff:        "commit_scn": "1513734",
         "lcr_position" : null
       },
  -    "databaseName": "ORCLPDB1",
  -    "schemaName": "DEBEZIUM",
  -    "ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n   (    \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n    \"FIRST_NAME\" VARCHAR2(255), \n    \"LAST_NAME" VARCHAR2(255), \n    \"EMAIL\" VARCHAR2(255), \n     PRIMARY KEY (\"ID\") ENABLE, \n     SUPPLEMENTAL LOG DATA (ALL) COLUMNS\n   ) SEGMENT CREATION IMMEDIATE \n  PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 \n NOCOMPRESS LOGGING\n  STORAGE(INITIAL 65536 NEXT 1048576 MINEXTENTS 1 MAXEXTENTS 2147483645\n  PCTINCREASE 0 FREELISTS 1 FREELIST GROUPS 1\n  BUFFER_POOL DEFAULT FLASH_CACHE DEFAULT CELL_FLASH_CACHE DEFAULT)\n  TABLESPACE \"USERS\" ",
  -    "tableChanges": [
  +    "databaseName": "ORCLPDB1", // <1>
  +    "schemaName": "DEBEZIUM", // <1>
  +    "ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n   (    \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n    \"FIRST_NAME\" VARCHAR2(255), \n    \"LAST_NAME" VARCHAR2(255), \n    \"EMAIL\" VARCHAR2(255), \n     PRIMARY KEY (\"ID\") ENABLE, \n     SUPPLEMENTAL LOG DATA (ALL) COLUMNS\n   ) SEGMENT CREATION IMMEDIATE \n  PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 \n NOCOMPRESS LOGGING\n  STORAGE(INITIAL 65536 NEXT 1048576 MINEXTENTS 1 MAXEXTENTS 2147483645\n  PCTINCREASE 0 FREELISTS 1 FREELIST GROUPS 1\n  BUFFER_POOL DEFAULT FLASH_CACHE DEFAULT CELL_FLASH_CACHE DEFAULT)\n  TABLESPACE \"USERS\" ", // <2>
  +    "tableChanges": [ // <3>
         {
  -        "type": "CREATE",
  -        "id": "\"ORCLPDB1\".\"DEBEZIUM\".\"CUSTOMERS\"",
  -        "table": {
  +        "type": "CREATE", // <4>
  +        "id": "\"ORCLPDB1\".\"DEBEZIUM\".\"CUSTOMERS\"", // <5>
  +        "table": { // <6>
             "defaultCharsetName": null,
  -          "primaryKeyColumnNames": [
  +          "primaryKeyColumnNames": [ // <7>
               "ID"
             ],
  -          "columns": [
  +          "columns": [ // <8>
               {
                 "name": "ID",
                 "jdbcType": 2,
  
positive.code: }
  ----
  
  [[selecting-the-adapter]]
  == Selecting the adapter
  
  {prodname} provides multiple ways to ingest change events from Oracle.
  By default {prodname} uses the XStream API but this isn't always applicable for every installation.
  
negative.code: But only if the message value is `null` will Kafka know that it can remove _all messages_ with that same key.
  To make this possible, {prodname}'s Oracle connector always follows the _delete_ event with a special _tombstone_ event that has the same key but `null` value.
  
  [[oracle-data-type-mappings]]
  == Data Type mappings
  
  The Oracle conenctor represents changes to rows with events that are structured like the table in which the rows exists. The event contains a field for each column value. How that value is represented in the event depends on the Oracle data type of the column. The following sections describe how the connector maps oracle data types to a _litearl type_ and a _semantic type_ in event fields.
  
  * _litearl type_ describes how the value is literally represented using Kafka Connect schema types: `INT8`, `INT16`, `INT32`, `INT64`, `FLOAT32`, `FLOAT64`, `BOOLEAN`, `STRING`, `BYTES`, `ARRAY`, `MAP`, and `STRUCT`.
  
  * _semantic type_ describes how the Kafka Connect schema captures the _meaning_ of the field using the name of the Kafka Connect schema for the field.
  
  ifdef::product[]
  Details are in the following sections:
  
  * xref:oracle-character-types[]
  * xref:oracle-numeric-types[]
  * xref:oracle-decimal-types[]
  * xref:oracle-temporal-types[]
  
  endif::product[]
  
  ifndef::product[]
  Support for further data types will be added in subsequent releases.
  Please file a {jira-url}/browse/DBZ[JIRA issue] for any specific types that may be missing.
  endif::[]
  
  [id="oracle-character-types"]
  === Character types
  
  The following table describes how the connector maps character types.
  
  .Mappings for Oracle character data types
  [cols="20%a,15%a,55%a",options="header"]
  |===
  |Oracle Data Type
  |Literal type (schema type)
  |Semantic type (schema name) and Notes
  
  |`CHAR[(M)]`
  |`STRING`
  |n/a
  
  |`NCHAR[(M)]`
  |`STRING`
  |n/a
  
  |`NVARCHAR2[(M)]`
  |`STRING`
  |n/a
  
  |`VARCHAR[(M)]`
  |`STRING`
  |n/a
  
  |`VARCHAR2[(M)]`
  |`STRING`
  |n/a
  
  |===
  
  [id="oracle-numeric-types"]
  === Numeric types
  
  The following table describes how the connector maps numeric types.
  
  .Mappings for Oracle numeric data types
  [cols="20%a,15%a,55%a",options="header"]
  |===
  |Oracle Data Type
  |Literal type (schema type)
  |Semantic type (schema name) and Notes
  
  |`BINARY_FLOAT`
  |`FLOAT32`
  |n/a
  
  |`BINARY_DOUBLE`
  |`FLOAT64`
  |n/a
  
  |`DECIMAL[(P, S)]`
  |`BYTES` / `INT8` / `INT16` / `INT32` / `INT64`
  |`org.apache.kafka.connect.data.Decimal` if using `BYTES` +
   +
  Handled equivalently to `NUMBER` (note that S defaults to 0 for `DECIMAL`).
  
  |`DOUBLE PRECISION`
  |`STRUCT`
  |`io.debezium.data.VariableScaleDecimal` +
   +
  Contains a structure with two fields: `scale` of type `INT32` that contains the scale of the transferred value and `value` of type `BYTES` containing the original value in an unscaled form.
  
  |`FLOAT[(P)]`
  |`STRUCT`
  |`io.debezium.data.VariableScaleDecimal` +
   +
  Contains a structure with two fields: `scale` of type `INT32` that contains the scale of the transferred value and `value` of type `BYTES` containing the original value in an unscaled form.
  
  |`INTEGER`, `INT`
  |`BYTES`
  |`org.apache.kafka.connect.data.Decimal` +
   +
  `INTEGER` is mapped in Oracle to NUMBER(38,0) and hence can hold values larger than any of the `INT` types could store
  
  |`LONG`
  |`INT64`
  |n/a
  
  |`NUMBER(1,0)`
  |`BOOLEAN`
  |n/a
  
  |`NUMBER[(P[, *])]`
  |`STRUCT`
  |`io.debezium.data.VariableScaleDecimal` +
   +
  Contains a structure with two fields: `scale` of type `INT32` that contains the scale of the transferred value and `value` of type `BYTES` containing the original value in an unscaled form.
  
  |`NUMBER(P, S > 0)`
  |`BYTES`
  |`org.apache.kafka.connect.data.Decimal`
  
  |`NUMBER(P, S <= 0)`
  |`INT8` / `INT16` / `INT32` / `INT64`
  |`NUMBER` columns with a scale of 0 represent integer numbers; a negative scale indicates rounding in Oracle, e.g. a scale of -2 will cause rounding to hundreds. +
   +
  Depending on the precision and scale, a matching Kafka Connect integer type will be chosen: +
  
   * P - S < 3, `INT8` +
   * P - S < 5, `INT16` +
   * P - S < 10, `INT32` +
   * P - S < 19, `INT64` +
   * P - S >= 19, `BYTES` (`org.apache.kafka.connect.data.Decimal`).
  
  |`NUMERIC[(P, S)]`
  |`BYTES` / `INT8` / `INT16` / `INT32` / `INT64`
  |`org.apache.kafka.connect.data.Decimal` if using `BYTES` +
   +
  Handled equivalently to `NUMBER` (note that S defaults to 0 for `NUMERIC`).
  
  |`SMALLINT`
  |`BYTES`
  |`org.apache.kafka.connect.data.Decimal` +
   +
  `SMALLINT` is mapped in Oracle to NUMBER(38,0) and hence can hold values larger than any of the `INT` types could store
  
  |`REAL`
  |`STRUCT`
  |`io.debezium.data.VariableScaleDecimal` +
   +
  Contains a structure with two fields: `scale` of type `INT32` that contains the scale of the transferred value and `value` of type `BYTES` containing the original value in an unscaled form.
  
  |===
  
  [id="oracle-decimal-types"]
  === Decimal types
  The setting of the Oracle connector configuration property, `decimal.handling.mode` determines how the connector maps decimal types.
  
  When the `decimal.handling.mode` property is set to `precise`, the connector uses Kafka Connect `org.apache.kafka.connect.data.Decimal` logical type for all `DECIMAL` and `NUMERIC` columns.
  This is the default mode.
  
  However, when the `decimal.handling.mode` property is set to `double`, the connector will represent the values as Java double values with schema type `FLOAT64`.
  
  The last possible setting for the `decimal.handling.mode` configuration property is `string`.
  In this case, the connector reprsents `DECIMAL` and `NUMERIC` values as their formatted string representation with schema type `STRING`.
  
  [id="oracle-temporal-types"]
  === Temporal types
  
  The following table describes how the connector maps temporal types.
  
  .Mappings for Oracle temporal data types
  [cols="25%a,20%a,55%a",options="header"]
  |===
  |Oracle data type |Literal type (schema type) |Semantic type (schema name) and Notes
  
  |`DATE`
  |`INT64`
  |`io.debezium.time.Timestamp` +
   +
  Represents the number of milliseconds past epoch, and does not include timezone information.
  
  |`INTERVAL DAY[(M)] TO SECOND`
  |`FLOAT64`
  |`io.debezium.time.MicroDuration` +
  +
  The number of micro seconds for a time interval using the `365.25 / 12.0` formula for days per month average
  
  |`INTERVAL YEAR[(M)] TO MONTH`
  |`FLOAT64`
  |`io.debezium.time.MicroDuration` +
   +
  The number of micro seconds for a time interval using the `365.25 / 12.0` formula for days per month average
  
  |`TIMESTAMP(0 - 3)`
  |`INT64`
  |`io.debezium.time.Timestamp` +
   +
  Represents the number of milliseconds past epoch, and does not include timezone information.
  
  |`TIMESTAMP, TIMESTAMP(4 - 6)`
  |`INT64`
  |`io.debezium.time.MicroTimestamp` +
   +
  Represents the number of microseconds past epoch, and does not include timezone information.
  
  |`TIMESTAMP(7 - 9)` +
   |`INT64`
  |`io.debezium.time.NanoTimestamp` +
   +
  Represents the number of nanoseconds past epoch, and does not include timezone information.
  
  |`TIMESTAMP WITH TIME ZONE`
  |`STRING`
  |`io.debezium.time.ZonedTimestamp` +
   +
  A string representation of a timestamp with timezone information
  
  |`TIMESTAMP WITH LOCAL TIME ZONE`
  |`STRING`
  |`io.debezium.time.ZonedTimestamp` +
   +
  A string representation of a timestamp with timezone information
  
  |===
  
  [[setting-up-oracle]]
  == Setting up Oracle
  
  The following database set up steps are necessary to use the {prodname} Oracle connector.
  These steps assume the use of the multitenancy configuration with a container database and at least one pluggable database.
  If you intend to use a non-multitenancy configuration, the following steps may require adjustment.
  
  You can find a template for setting up Oracle in a virtual machine (via Vagrant) in the https://github.com/debezium/oracle-vagrant-box/[oracle-vagrant-box/] repository.
  
  === Preparing the Database
  
  .Configuration needed for Oracle XStream
  [source,indent=0]
  ----
  ORACLE_SID=ORCLCDB dbz_oracle sqlplus /nolog
  
  CONNECT sys/top_secret AS SYSDBA
  alter system set db_recovery_file_dest_size = 5G;
  alter system set db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' scope=spfile;
  alter system set enable_goldengate_replication=true;
  shutdown immediate
  startup mount
  alter database archivelog;
  alter database open;
  -- Should show "Database log mode: Archive Mode"
  archive log list
  
  exit;
  ----
  
  .Configuration needed for Oracle LogMiner
  [source,indent=0]
  ----
  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/oradta/recovery_area' scope=spfile;
  shutdown immediate
  startup mount
  alter database archivelog;
  alter database open;
  -- Should now "Database log mode: Archive Mode"
  archive log list
  
  exit;
  ----
  
  In addition, supplemental logging must be enabled for captured tables or the database in order for data changes to capture the _before_ state of changed database rows.
  The following illustrates how to configure this on a specific table, which is the ideal choice to minimize the amount of information captured in the Oracle redo logs.
  
  [source,indent=0]
  ----
  ALTER TABLE inventory.customers ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
  ----
  
  === Creating Users for the connector
  
  The {prodname} Oracle connector requires that users accounts be set up with specific permissions so that the connector can capture change events.
  The following briefly describes these user configurations using a multi-tenant database model.
  
  * <<oracle-create-users-xstream, `Creating users for Oracle XStream`>>
  * <<oracle-create-users-logminer, `Creating users for Oracle LogMiner`>>
  
  [[oracle-create-users-xstream]]
  .Creating an XStream Administrator user
  [source,indent=0]
  ----
  sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
    CREATE TABLESPACE xstream_adm_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/xstream_adm_tbs.dbf'
      SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
    exit;
  
  sqlplus sys/top_secret@//localhost:1521/ORCLPDB1 as sysdba
    CREATE TABLESPACE xstream_adm_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/xstream_adm_tbs.dbf'
      SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
    exit;
  
  sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
    CREATE USER c##xstrmadmin IDENTIFIED BY xsa
      DEFAULT TABLESPACE xstream_adm_tbs
      QUOTA UNLIMITED ON xstream_adm_tbs
      CONTAINER=ALL;
  
    GRANT CREATE SESSION, SET CONTAINER TO c##xstrmadmin CONTAINER=ALL;
  
    BEGIN
       DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
          grantee                 => 'c##xstrmadmin',
          privilege_type          => 'CAPTURE',
          grant_select_privileges => TRUE,
          container               => 'ALL'
       );
    END;
    /
  
    exit;
  ----
  
  .Creating the connector's XStream user
  [source,indent=0]
  ----
  sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
    CREATE TABLESPACE xstream_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/xstream_tbs.dbf'
      SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
    exit;
  
  sqlplus sys/top_secret@//localhost:1521/ORCLPDB1 as sysdba
    CREATE TABLESPACE xstream_tbs DATAFILE '/opt/oracle/oradata/ORCLCDB/ORCLPDB1/xstream_tbs.dbf'
      SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
    exit;
  
  sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
    CREATE USER c##xstrm IDENTIFIED BY xs
      DEFAULT TABLESPACE xstream_tbs
      QUOTA UNLIMITED ON xstream_tbs
      CONTAINER=ALL;
  
    GRANT CREATE SESSION TO c##xstrm CONTAINER=ALL;
    GRANT SET CONTAINER TO c##xstrm CONTAINER=ALL;
    GRANT SELECT ON V_$DATABASE to c##xstrm CONTAINER=ALL;
    GRANT FLASHBACK ANY TABLE TO c##xstrm CONTAINER=ALL;
    GRANT SELECT_CATALOG_ROLE TO c##xstrm CONTAINER=ALL;
    GRANT EXECUTE_CATALOG_ROLE TO c##xstrm CONTAINER=ALL;
    exit;
  ----
  
  [[oracle-create-users-logminer]]
  .Creating the connector's LogMiner user
  [source,indent=0]
  ----
  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##logminer IDENTIFIED BY lm
      DEFAULT TABLESPACE logminer_tbs
      QUOTA UNLIMITED ON logminer_tbs
      CONTAINER=ALL;
  
    GRANT CREATE SESSION TO c##logminer CONTAINER=ALL;
    GRANT SET CONTAINER TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$DATABASE to c##logminer CONTAINER=ALL;
    GRANT FLASHBACK ANY TABLE TO c##logminer CONTAINER=ALL;
    GRANT SELECT ANY TABLE TO c##logminer CONTAINER=ALL;
    GRANT SELECT_CATALOG_ROLE TO c##logminer CONTAINER=ALL;
    GRANT EXECUTE_CATALOG_ROLE TO c##logminer CONTAINER=ALL;
    GRANT SELECT ANY TRANSACTION TO c##logminer CONTAINER=ALL;
    GRANT LOGMINING TO c##logminer CONTAINER=ALL;
  
    GRANT CREATE TABLE TO c##logminer CONTAINER=ALL;
    GRANT LOCK ANY TABLE TO c##logminer CONTAINER=ALL;
    GRANT ALTER ANY TABLE TO c##logminer CONTAINER=ALL;
    GRANT CREATE SEQUENCE TO c##logminer CONTAINER=ALL;
  
    GRANT EXECUTE ON DBMS_LOGMNR TO c##logminer CONTAINER=ALL;
    GRANT EXECUTE ON DBMS_LOGMNR_D TO c##logminer CONTAINER=ALL;
  
    GRANT SELECT ON V_$LOG TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$LOG_HISTORY TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$LOGMNR_LOGS TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$LOGFILE TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$ARCHIVED_LOG TO c##logminer CONTAINER=ALL;
    GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##logminer CONTAINER=ALL;
  
    exit;
  ----
  
  === Create an XStream Outbound Server
  
  [NOTE]
  ====
  If you're using the LogMiner implementation, this step is not necessary.
  ====
  
  Create an https://docs.oracle.com/cd/E11882_01/server.112/e16545/xstrm_cncpt.htm#XSTRM1088[XStream Outbound server]
  (given the right privileges, this may be done automatically by the connector going forward, see {jira-url}/browse/DBZ-721[DBZ-721]):
  
  .Create an XStream Outbound Server
  [source,indent=0]
  ----
  sqlplus c##xstrmadmin/xsa@//localhost:1521/ORCLCDB
  DECLARE
    tables  DBMS_UTILITY.UNCL_ARRAY;
    schemas DBMS_UTILITY.UNCL_ARRAY;
  BEGIN
      tables(1)  := NULL;
      schemas(1) := 'debezium';
    DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
      server_name     =>  'dbzxout',
      table_names     =>  tables,
      schema_names    =>  schemas);
  END;
  /
  exit;
  ----
  
  .Configure the XStream user account to connect to the XStream Outbound Server
  [source,indent=0]
  ----
  sqlplus sys/top_secret@//localhost:1521/ORCLCDB as sysdba
  BEGIN
    DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
      server_name  => 'dbzxout',
      connect_user => 'c##xstrm');
  END;
  /
  exit;
  ----
  
  [NOTE]
  ====
  A single XStream Outbound server cannot be shared by multiple {prodname} Oracle connectors.
  Each connector requires a unique XStream Outbound connector to be configured.
  ====
  
  [[oracle-deploying-a-connector]]
  == Deploying a Connector
  
  Due to licensing requirements, the {prodname} Oracle Connector does not ship with the Oracle JDBC driver and the XStream API JAR.
  You can obtain them for free by downloading the https://www.oracle.com/database/technologies/instant-client/downloads.html[Oracle Instant Client].
  
  Extract the archive and navigate to the Instant Client directory.
  Copy `ojdbc<version>.jar` and `xstreams.jar` into Kafka's _libs_ directory.
  Lastly, create an environment variable, `LD_LIBRARY_PATH`, that points to the instant client directory, as shown below:
  
  [source,bash,indent=0]
  ----
  