Oracle Streams - replication data on schema base

Oracle Streams is one of the solution to replication. Oracle has several solutions : Replication, Golden Gate, Streams. Streams has a lot of advantages, one and main is data can be replicate between Oracle to non Oracle services. Brief info about streams.

Oracle Streams captures database changes at a source database, stages the changes, propagates the changes to one or more destination databases, and then applies the changes at the destination database(s). Using Oracle Streams, enterprise systems can capture, propagate, and apply information as follows:

• Within an Oracle database
• Between two Oracle databases
• Among multiple Oracle databases
• Between an Oracle database and a non-Oracle database

How Streams Works

Oracle Streams begins by capturing changes. The changes (to data, tables, schemas, and so on) that take place in a database are recorded in the database redo log files. The Streams capture process extracts these changes from the redo log files and formats each change into a logical change record (LCR). The LCRs are then stored in a queue (staged). Next, Streams propagates LCRs from one queue (the producer queue) to another (the consumer queue) and can then apply (or consume) the LCRs from the consumer queue to the destination database.

In this tutorial I will show you how stream works on base one database between schema. Oracle version 10.2.x:

Objectives:  Replicate HR.EMPLOYEES table with DEMO.EMP in the same database.



Note: Database should be work on ARCHIVELOG mode and global_names value should be TRUE

SQL> alter system set global_names = TRUE scope=both;

1.  Prepare source database parameters

    
2.  Create Streams Administrator          

SQL> create tablespace streams_tbs datafile '/home/oracle/oracle/product/10.2.0/oradata/mydb/streams_tbs01.dbf' size 25M autoextend on maxsize unlimited;

Tablespace created.

SQL>  grant dba to strmadmin identified by streams;

Grant succeeded.

SQL>  alter user strmadmin default tablespace streams_tbs quota unlimited on streams_tbs;

User altered.

SQL>

SQL> exec dbms_streams_auth.grant_admin_privilege(grantee  => 'strmadmin', grant_privileges => true);

PL/SQL procedure successfully completed.

SQL> 

   

3.  Prepare HR.EMPLOYEES and DEMO.EMP objects.

SQL> grant dba to demo identified by demo;    


Grant succeeded.
   

 SQL> create table DEMO.EMP as select * from HR.EMPLOYEES;


Table created.

 Add constraint emp_emp_id_pk primary key (employee_id);

SQL> alter table DEMO.EMP add constraint emp_emp_id_pk primary key (employee_id);


Table altered.

And give select permission to demo user

SQL > grant select on hr.employees to demo

4.  Setup staging QUEUE

   

SQL> begin

      dbms_streams_adm.set_up_queue (

           queue_table => 'strmadmin.streams_queue_table',

           queue_name => 'strmadmin.streams_queue');

    end;

    /  2    3    4    5    6  

PL/SQL procedure successfully completed.

SQL> select name, queue_table from dba_queues where owner='STRMADMIN';

NAME                                        QUEUE_TABLE

------------------------------                 ------------------------------

STREAMS_QUEUE                        STREAMS_QUEUE_TABLE

AQ$_STREAMS_QUEUE_TABLE_E    STREAMS_QUEUE_TABLE

SQL> set linesize 150

col rule_owner for a10

SQL> select rule_owner, streams_type, streams_name, rule_set_name, rule_name from dba_streams_rules;


RULE_OWNER STREAMS_TYP  STREAMS_NAME                         RULE_SET_NAME                  RULE_NAME
----------          -----------         ------------------------------                 ------------------------------ ------------------------------
SYS               CAPTURE       CAPTURE_EMP                             RULESET$_13                    EMPLOYEES12

SQL> 

5.  Define the CAPTURE process and associated rules.

SQL> BEGIN
 dbms_streams_adm.add_table_rules(
   table_name  => 'HR.EMPLOYEES',
   streams_type  => 'CAPTURE',
   streams_name  => 'CAPTURE_EMP',
   queue_name  => 'STRMADMIN.STREAMS_QUEUE',
   include_dml  => TRUE,
   include_ddl  => FALSE,
   inclusion_rule => TRUE);
END;
    /  2    3    4    5    6    7    8    9   10   11  


PL/SQL procedure successfully completed.


SQL> 

SQL> select capture_name, rule_set_name,capture_user from dba_capture;


CAPTURE_NAME                   RULE_SET_NAME         CAPTURE_USER
------------------------------          ------------------------------ ------------------------------
CAPTURE_EMP                     RULESET$_13             SYS


SQL> 

-- Optionally, we can include extra attributes in the capture streams.

       

SQL> BEGIN
        DBMS_CAPTURE_ADM.INCLUDE_EXTRA_ATTRIBUTE(
             capture_name   => 'CAPTURE_EMP',
             attribute_name => 'USERNAME',
             include  => true);
        END;
    /  2    3    4    5    6    7  


PL/SQL procedure successfully completed.


SQL> 

6.  Tell Oracle where to start the catpure.  Instantiate the scn for hr.employees.

SQL> select source_object_owner, source_object_name, instantiation_scn from dba_apply_instantiated_objects;


SOURCE_OBJECT_OWNER            SOURCE_OBJECT_NAME             INSTANTIATION_SCN
------------------------------                ------------------------------               -----------------
HR                                          EMPLOYEES                              1591945


SQL> 

SQL>  conn strmadmin/streams

Connected.

    

SQL> DECLARE

         iscn  NUMBER;

         BEGIN

               iscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_nUMBER();

               DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN(

                      source_object_name => 'HR.EMPLOYEES',

                      source_database_name => 'MYDB',

                      instantiation_scn => iscn);

         END;

    /  2    3    4    5    6    7    8    9   10  

PL/SQL procedure successfully completed.

SQL> 

7.  Create the APPLY process.

SQL>  conn strmadmin/streams

Connected.

SQL> SET SERVEROUTPUT ON

DECLARE

 emp_rule_name_dml VARCHAR2(30);

 emp_rule_name_ddl VARCHAR2(30);

BEGIN

 DBMS_STREAMS_ADM.ADD_TABLE_RULES(

   table_name   => 'hr.employees',

   streams_type   => 'apply',

   streams_name   => 'apply_emp',

   queue_name   => 'strmadmin.streams_queue',

   include_dml   => true,

   include_ddl   => false,

   source_database  => 'MYDB',

   dml_rule_name  => emp_rule_name_dml,

   ddl_rule_name  => emp_rule_name_ddl);

        DBMS_OUTPUT.PUT_LINE('DML rule name: '||emp_rule_name_dml);

        DBMS_OUTPUT.PUT_LINE('DDL rule name: '||emp_rule_name_ddl);

END;

/SQL>   2    3    4    5    6    7    8    9   10   11   12   13   14   15   16   17   18   19  

DML rule name: "STRMADMIN"."EMPLOYEES20"

DDL rule name:

PL/SQL procedure successfully completed.

SQL> 

--We don't want to stop applying changes when there is an error, so:

SQL> BEGIN

 DBMS_APPLY_ADM.SET_PARAMETER(

   apply_name => 'apply_emp',

   parameter => 'disable_on_error',

   value => 'n');

END;

/  2    3    4    5    6    7  

PL/SQL procedure successfully completed.

SQL> 

--Tranform the CAPTURED event.

--Find the APPLY rule name and change the rule declarativly.

   
SQL> SELECT a.apply_name, a.rule_set_name, r.rule_owner, r.rule_name
FROM   dba_apply a, dba_streams_rules r
WHERE  a.rule_set_name=r.rule_set_name;  2    3  


APPLY_NAME              RULE_SET_NAME        RULE_OWNER    RULE_NAME
------------------------------ ------------------------------ --------------------- ------------------------------
APPLY_EMP                RULESET$_21             STRMADMIN      EMPLOYEES20


SQL>    

take rule_name of the above select and put below

SQL> BEGIN

DBMS_STREAMS_ADM.RENAME_TABLE(

   rule_name   => 'STRMADMIN.EMPLOYEES20',

   from_table_name => 'HR.EMPLOYEES',

    to_table_name  => 'DEMO.EMP',

   operation  => 'ADD');  -- can be ADD or REMOVE

END;

/  2    3    4    5    6    7    8  

PL/SQL procedure successfully completed.

SQL> 

8.  Turn on the apply process:

    

 SQL>BEGIN

DBMS_APPLY_ADM.START_APPLY(apply_name => 'apply_emp');

 END;

 /

9.  Turn on the capture process:

 SQL> BEGIN
 DBMS_CAPTURE_ADM.START_CAPTURE(capture_name => 'capture_emp');
END;
/  2    3    4  


PL/SQL procedure successfully completed.


SQL> 

Check if hr`s user password was changed or account locked: 
select account_status from dba_users where username='HR'

alter user hr identified by hr;

alter user hr account unlock;

10. Test

SQL> select count(1) from hr.employees;


  COUNT(1)
----------
  107
SQL> 


 AS HR user add a row to the employees table

SQL> conn hr/hr

Connected.

SQL> insert into employees values 

(300,'Ulfet','Tanriverdiyev','teulfet@azercell.com','123456789',sysdate,'IT_PROG',30000,0,103,60); 

1 row created.

SQL> commit;

Commit complete.

SQL> 

As DEMO user check the EMP table.

SQL> select employee_id, first_name, email, job_id from emp where employee_id=300;


EMPLOYEE_ID   FIRST_NAME       EMAIL                        JOB_ID
-------------------- --------------------   -------------------------        ----------
300                 Ulfet                teulfet@azercell.com   IT_PROG


SQL> 

Data captured, propagated and finally applied. 
P.S: Possible data not immediately applied, please to be patient :) Next time I will show you using streams technique between databases.

Good luck :)