Oracle Streams One Way Table Replication 101

If you're looking to jump start Oracle Streams, this is probably the right post: a quick setup, a lot of shortcuts, one only database and probably the simplest configuration you will ever see. Obviously, that doesn't address everything; you'll always need to deep dive into the concepts and principals of Streams. But you should be able to get a running Streams configuration in less than 30 minutes and use it as a basis to build more complex test cases. Hopefully that will be as useful to you as it is to me.

In this example, we'll maintain a copy of a table named T1 located in the SOURCE schema, in the DESTINATION schema of the same database. In order to get to that result, we'll dig into the details of the steps below:

• Build The Sample Schema and Table
  
• Configure Database Logging and the Multi-Version Data Dictionary
  
• Create the Streams Administrator and The Streams Queue
• Create the Capture Process and Define the Capture Rules
• Create an Apply Process
• Transform the SQL Captured so that changes to SOURCE are applied to DESTINATION
• Instantiate the Table in the DESTINATION Schema
• Start the Capture and Apply Processes
• Test the Table Replication
• Stop and Suppress the Oracle Streams Configuration

Note:
We've tested this post on a 11.1.0.7 and on a 10.2.0.4 Enterprise Edition version of Oracle running on Linux 32bits.

Step 1: Build The Sample Schema and Table

In order to start, we'll create a schema named SOURCE and a table named T1. You'll find below the script to perform the associated operations:

connect / as sysdba

create user source
      identified by source
      default tablespace users
      temporary tablespace temp;

grant connect,resource to source;

connect source/source

create table t1(
      id   number primary key,
      text varchar2(80));

insert into t1(id, text)
      values (1,'Text 1');

insert into t1(id, text)
      values (2,'Text 2');

commit;

Step 2: Configure Database Logging and the Multi-Version Data Dictionary
Streams capture processes use the content of the redologs/archivelogs to build Logical Change Records (LCR) that are propagated to the destination database via buffered queues. Once propagated, the LCRs are applied to the destination database by the apply processes that transform them into a set of SQL commands. As you know the redo logs don't contain SQL but binary data. Those data are built to be used by a recovery process and, by default, don't contain the necessary data to be applied on a different database. Here are some of the consequences of using the redo logs to capture changes made on a source database:

• The redologs don't contain the SQL commands that have been run on the database; instead, they contain the rowids of the changed rows with the before and after image of those changes. So if you update, say 1 column for all the rows of a table, in one update command, Streams will NOT replicate that command. Instead, capture processes generate one LCR per changed row. Each LCR includes the before and after value of the changes. Saying it in an other way, one update on the source database may lead to millions LCRs and updates on the destination database.
• RowIds are used to uniquely identified changed rows and recover the archivelogs and redologs. But RowIds of rows differ on a destination database that is not a physical copy of the source database! To uniquely identify changed rows on a destination database, we must add to the LCR the value of a Unique or a Primary Key with every change. To add them to the LCR we must add them to the redo logs; we'll use supplemental logging for that purpose!
  
• The schemes and names of the objects that are changed are not stored in the redo logs either; Instead, what is stored is the object identifiers. Those identifiers are stored in the LCRs and you need a copy of the dictionary from the source database on the destination database to "reverse engineer" LCRs into SQL. That copy of the dictionary is part of the Multi-Version Data Dictionary (MVDD). It needs to be captured from the source database and sent to the destination database.
  

For the reasons above, we need to prepare the source database to be compatible with Streams Capture.

To uniquely identified rows on the destination database, you need the database to be in ARCHIVELOG mode. Add supplemental logging on the source database and log the unique and primary keys of all the changed rows. That can be done with the atomicity of one object but, as discussed earlier, this post takes some shortcuts and we'll add the supplemental logging at the database level. Below is the script to perform that operation:

connect / as sysdba

alter database add supplemental log
     data (primary key, unique index) columns;

select SUPPLEMENTAL_LOG_DATA_MIN,
       SUPPLEMENTAL_LOG_DATA_PK,
       SUPPLEMENTAL_LOG_DATA_UI
  from gv$database;

SUPPLEME SUP SUP
-------- --- ---
IMPLICIT YES YES

To push a copy of the data dictionary from the source to the destination database, we need to run DBMS_CAPTURE_ADM.BUILD as below:

connect / as sysdba

var first_scn number;

set serveroutput on

DECLARE
   scn NUMBER;
BEGIN
   DBMS_CAPTURE_ADM.BUILD(
          first_scn => scn);
   DBMS_OUTPUT.PUT_LINE('First SCN Value = ' || scn);
   :first_scn := scn;
END;
/

First SCN Value = 49042254018

It is important to keep track of the SCN returned by DBMS_CAPTURE_ADM.BUILD. You'll use that SCN as the first SCN of the capture processes so that the meta data from the source are pushed and kept in the destination MVDD. Note that the last command captures the minimum to add to the MVDD and you have to add the meta data of all the objects you want to replicate. In order to do that, you must use one of the prepare_xxxx_instantiation procedure of the dbms_capture_adm package as below:

exec dbms_capture_adm.prepare_table_instantiation(-
    table_name=>'source.t1');

Step 3: Create the Streams Administrator and The Streams Queue

To use Oracle Streams, you must create a Streams Administrator in all the databases that directly involved with a capture, propagation or apply process; the script below does it. It creates a tablespace to store the queues: it is require to have a separate tablespace for queues. Then it creates a user STRMADMIN, grants it the privileges and roles as well as the buffered queue with will be used by the capture and the apply processes to share the LCR:

connect / as sysdba

CREATE TABLESPACE streams_tbs
      DATAFILE '/u01/app/oracle/oradata/BLACK/streams_tbs.dbf' size 25M
      AUTOEXTEND ON MAXSIZE 256M;

CREATE USER strmadmin IDENTIFIED BY strmadmin
      DEFAULT TABLESPACE streams_tbs
      QUOTA UNLIMITED ON streams_tbs;

grant dba to strmadmin;

BEGIN
   DBMS_STREAMS_ADM.SET_UP_QUEUE(
        queue_table => 'strmadmin.streams_queue_table',
        queue_name  => 'strmadmin.streams_queue');
END;
/

Step 4: Create the Capture Process and Define the Capture Rules
Once the source and the destination databases setup, you can create the capture process and add the source table to it. The first_scn parameter must match the one returned from DBMS_CAPTURE_ADM.BUILD; the source_database parameter must match the db_unique_name (or db_name, if db_unique_name is not used) of the source database; this is the script to create the capture process:

connect strmadmin/strmadmin

accept first_scn prompt "Enter the First SCN of the Capture: "
Enter the First SCN of the Capture: 49042254018

var first_scn number;
exec :first_scn:=&&first_scn

BEGIN
   DBMS_CAPTURE_ADM.CREATE_CAPTURE(
        queue_name         => 'strmadmin.streams_queue',
        capture_name       => 'streams_capture',
        rule_set_name      => NULL,
        source_database    => 'BLACK',
        use_database_link  => false,
        first_scn          => :first_scn,
        logfile_assignment => 'implicit');
END;
/

col capture_name format a15
col queue_name format a13
col first_scn format 999999999999
col start_scn format 999999999999
col rule_set_name format a11

select capture_name,
       queue_name,
       first_scn,
       start_scn,
       rule_set_name
  from dba_capture;

CAPTURE_NAME QUEUE_NAME FIRST_SCN START_SCN RULE_SET_N
--------------- ------------- ------------- ------------- ----------
STREAMS_CAPTURE STREAMS_QUEUE 49042254018 49042254018 RULESET$_7

BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
        table_name         => 'source.t1',
        streams_type       => 'capture',
        streams_name       => 'streams_capture',
        queue_name         => 'strmadmin.streams_queue',
        include_dml        => true,
        include_ddl        => false,
        include_tagged_lcr => true,
        source_database    => 'BLACK',
        inclusion_rule     => true);
END;
/

set lines 120
col streams_name format a16
col streams_type format a9
col table_owner format a10
col table_name format a15
col rule_type format a8
col rule_name format a15

select rule_owner,
       STREAMS_NAME,
       STREAMS_TYPE,
       TABLE_OWNER,
       TABLE_NAME,
       RULE_TYPE,
       RULE_NAME
  from DBA_STREAMS_TABLE_RULES;

STREAMS_NAME    STREAMS_T TABLE_OWNE TABLE_NAME RULE_TYP RULE_NAME
--------------- --------- ---------- ---------- -------- ---------
STREAMS_CAPTURE CAPTURE   SOURCE     T1         DML      T16

Step 5: Create an Apply Process

You must create an apply process that will subscribe the Streams queue and apply the changes on the SOURCE.T1 table:

connect strmadmin/strmadmin

BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
       table_name         => 'source.t1',
       streams_type       => 'apply',
       streams_name       => 'streams_apply',
       queue_name         => 'strmadmin.streams_queue',
       include_dml        => true,
       include_ddl        => false,
       include_tagged_lcr => true,
       source_database    => 'BLACK',
       inclusion_rule     => true);
END;
/


col apply_name format a13
col queue_name format a13
col rule_set_name format a11

select apply_name,
       queue_name,
       rule_set_name,
       status,
       message_delivery_mode
  from dba_apply;

APPLY_NAME    QUEUE_NAME    RULE_SET_NA STATUS   MESSAGE_DE
------------- ------------- ----------- -------- ----------
STREAMS_APPLY STREAMS_QUEUE RULESET$_10 DISABLED CAPTURED

Step 6: Transform the SQL Captured so that changes on SOURCE are applied to DESTINATION

In this example the source and the destination databases are the same; as a result the source and target tables cannot have the same schema or name. What we'll do now is add a transformation rule so that the schema name is transformed from SOURCE to DESTINATION. The script below add the transformation to the apply:

connect strmadmin/strmadmin

col rule_name format a20 new_value rulename

select rule_owner,
       STREAMS_NAME,
       STREAMS_TYPE,
       TABLE_OWNER,
       TABLE_NAME,
       RULE_TYPE,
       RULE_NAME
  from DBA_STREAMS_TABLE_RULES
 where streams_name='STREAMS_APPLY'
   and streams_type='APPLY'
   and rule_type='DML';

prompt &&rulename
T16

begin
  dbms_streams_adm.rename_schema(
      rule_name        => '&&rulename' ,
      from_schema_name => 'SOURCE',
      to_schema_name   => 'DESTINATION',
      step_number      => 0,
      operation        => 'add');
end;
/

col rule_name format A6
col from_schema_name format a6
col to_schema_name format a12

select rule_name,
       transform_type,
       from_schema_name,
       to_schema_name,
       declarative_type
  from dba_streams_transformations;

RULE_N TRANSFORM_TYPE             FROM_S TO_SCHEMA_NA DECLARATIVE_T
------ -------------------------- ------ ------------ -------------
T16    DECLARATIVE TRANSFORMATION SOURCE DESTINATION  RENAME SCHEMA

Step 7: Instantiate the Table in the DESTINATION Schema

Streams is now configured. Before you start the different processes, you must instantiate the table in the DESTINATION schema and store the instantiation SCN so that the apply process knows what changes exactly it can apply. To instantiate the table, you can use the method of your choice: RMAN, DataPump, exp or Flashback Query:

connect / as sysdba

create user destination
       identified by destination
       default tablespace users
       temporary tablespace temp;

grant connect,resource to destination;

create table destination.t1(
       id number primary key,
       text varchar2(80));
col apply_scn format 999999999999 new_value instantiation_scn

select dbms_flashback.get_system_change_number apply_scn
 from dual;

APPLY_SCN
-----------
49042261443

prompt Enter the Instantiation: &&instantiation_scn
Enter the Instantiation: 49042261443

insert into destination.t1
      (select * from source.t1 as of scn &&instantiation_scn);

commit;

Once the table instantiated, use the dbms_apply_adm.set_table_instantiation_scn procedure to store the instantiation SCN for the apply process like below:

begin
  dbms_apply_adm.set_table_instantiation_scn(
          source_object_name   => 'source.t1',
          source_database_name => 'BLACK',
          instantiation_scn    => &&instantiation_scn);
end;
/

col SOURCE_DATABASE format a6
col OBJECT format a10
col INSTANTIATION_SCN format 999999999999

select source_database,
       source_object_owner||'.'||source_object_name object,
       instantiation_scn
  from dba_apply_instantiated_objects;

SOURCE OBJECT     INSTANTIATION_SCN
------ ---------- -----------------
BLACK SOURCE.T1   49042261443

Step 8: Start the Capture and Apply Processes

You can start the capture and the apply processes :

exec dbms_capture_adm.start_capture('streams_capture');
exec dbms_apply_adm.start_apply('streams_apply');

Step 9: Test the Table Replication

To test the replication, update the source table and check the changes are propagated to the destination table. Below is a simple test that shows the replication is working as it should:

insert into source.t1(id, text)
       values (3,'Text 3');

commit;

pause

col id format 99
col text format a6

select id,
       text
  from destination.t1;

ID TEXT
--- ------
1 Text 1
2 Text 2
3 Text 3

I guess troubleshooting the Streams configuration is out of the scope of this post. However, you can query the fixed views and catalog views if you need to get some of details about the configuration and its state.

Step 10: Stop and Suppress the Oracle Streams Configuration

In this last step, we'll clean up the environment and leave it as it was when we've started. In order to do it it, we'll stop and drop the capture and apply processes; we'll drop the Streams queue, the Streams administrator and the associated tablespace; we'll drop the 2 schemas and we'll disabled the supplemental logging:

exec dbms_capture_adm.stop_capture('streams_capture');
exec dbms_apply_adm.stop_apply('streams_apply');

exec dbms_capture_adm.drop_capture('streams_capture',true);
exec dbms_apply_adm.drop_apply('streams_apply',true);

exec dbms_streams_adm.remove_queue('strmadmin.streams_queue',true,true);

drop user strmadmin cascade;

drop tablespace streams_tbs
     including contents and datafiles;

drop user destination cascade;
drop user source cascade;

begin
  for i in (select source_schema name
             from dba_apply_instantiated_schemas
            where source_schema in ('SOURCE','DESTINATION'))
     loop  
     dbms_apply_adm.set_schema_instantiation_scn(
          source_schema_name   => i.name,
          source_database_name => 'BLACK',
          instantiation_scn    => null);
     end loop;
  for i in (select source_object_owner||'.'||
                   source_object_name name
             from dba_apply_instantiated_objects
            where source_object_owner in ('SOURCE','DESTINATION'))
     loop  
     dbms_apply_adm.set_table_instantiation_scn(
          source_object_name   => i.name,
          source_database_name => 'BLACK',
          instantiation_scn    => null);
     end loop;
end;
/
alter database drop supplemental log
     data (primary key, unique index) columns;

select SUPPLEMENTAL_LOG_DATA_MIN,
       SUPPLEMENTAL_LOG_DATA_PK,
       SUPPLEMENTAL_LOG_DATA_UI
  from gv$database;

SUPPLEME SUP SUP
-------- --- ---
NO       NO  NO