Oracle streams implementation at schema level

conn sys/password@dba1 as sysdba

create tablespace demo_TS datafile 'E:\TestDB\oradata\DBA1\demo_DAT.dbf' size 50M autoextend off extent management local;
create user demo IDENTIFIED BY demo DEFAULT TABLESPACE demo_TS TEMPORARY TABLESPACE TEMP;
ALTER USER demo QUOTA UNLIMITED on demo_TS;
grant connect to demo;


conn sys/password@dba2 as sysdba

create tablespace demo_TS datafile 'E:\TestDB\oradata\DBA2\demo_DAT.dbf' size 50M autoextend off extent management local;
create user demo IDENTIFIED BY demo DEFAULT TABLESPACE demo_TS TEMPORARY TABLESPACE TEMP;
ALTER USER demo QUOTA UNLIMITED on demo_TS;
grant connect to demo;

conn demo/demo@dba1

create table department
(departmentNO NUMBER(2),
DNAME VARCHAR2(14),
LOC VARCHAR2(13));

create table emp
(EMPNO NUMBER(2),
DNO VARCHAR2(14),
LOC VARCHAR2(13));

insert into department values ( 10, 'ACCOUNTING', 'NEW YORK');
insert into department values ( 20, 'RESEARCH', 'DALLAS');
insert into department values ( 30, 'SALES' , 'CHICAGO');
insert into department values ( 40, 'OPERATIONS', 'BOSTON');


conn sys/password@dba1 as sysdba

ALTER SYSTEM SET JOB_QUEUE_PROCESSES=1;
ALTER SYSTEM SET AQ_TM_PROCESSES=1;
ALTER SYSTEM SET GLOBAL_NAMES=TRUE;
ALTER SYSTEM SET COMPATIBLE='9.2.0' SCOPE=SPFILE;
ALTER SYSTEM SET LOG_PARALLELISM=1 SCOPE=SPFILE;
SHUTDOWN IMMEDIATE;
STARTUP;


CONN sys/password@DBA1 AS SYSDBA

CREATE USER strmadmin_dept IDENTIFIED BY strmadminpw
DEFAULT TABLESPACE users QUOTA UNLIMITED ON users;

GRANT CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin_dept;

GRANT EXECUTE ON DBMS_AQADM TO strmadmin_dept;

GRANT EXECUTE ON DBMS_CAPTURE_ADM TO strmadmin_dept;

GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO strmadmin_dept;

GRANT EXECUTE ON DBMS_STREAMS_ADM TO strmadmin_dept;

GRANT EXECUTE ON DBMS_APPLY_ADM TO strmadmin_dept;

GRANT EXECUTE ON DBMS_FLASHBACK TO strmadmin_dept;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ,
grantee => 'strmadmin_dept',
grant_option => FALSE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_OBJ,
grantee => 'strmadmin_dept',
grant_option => FALSE);
END;
/

CONNECT strmadmin_dept/strmadminpw@DBA1
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE();

CREATE DATABASE LINK dba2 CONNECT TO strmadmin_dept IDENTIFIED BY strmadminpw USING 'DBA2';

conn sys/password@dba2 as sysdba

ALTER SYSTEM SET JOB_QUEUE_PROCESSES=1;
ALTER SYSTEM SET AQ_TM_PROCESSES=1;
ALTER SYSTEM SET GLOBAL_NAMES=TRUE;
ALTER SYSTEM SET COMPATIBLE='9.2.0' SCOPE=SPFILE;
ALTER SYSTEM SET LOG_PARALLELISM=1 SCOPE=SPFILE;
SHUTDOWN IMMEDIATE;
STARTUP;


CONN sys/password@DBA2 AS SYSDBA

CREATE USER strmadmin_dept IDENTIFIED BY strmadminpw
DEFAULT TABLESPACE users QUOTA UNLIMITED ON users;

GRANT CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin_dept;

GRANT EXECUTE ON DBMS_AQADM TO strmadmin_dept;
GRANT EXECUTE ON DBMS_CAPTURE_ADM TO strmadmin_dept;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO strmadmin_dept;
GRANT EXECUTE ON DBMS_STREAMS_ADM TO strmadmin_dept;
GRANT EXECUTE ON DBMS_APPLY_ADM TO strmadmin_dept;
GRANT EXECUTE ON DBMS_FLASHBACK TO strmadmin_dept;

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ,
grantee => 'strmadmin_dept',
grant_option => FALSE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_OBJ,
grantee => 'strmadmin_dept',
grant_option => FALSE);
END;
/

CONNECT strmadmin_dept/strmadminpw@DBA2
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE();


CONN sys/password@DBA1 AS SYSDBA
GRANT ALL ON demo.department TO strmadmin_dept;
GRANT ALL ON demo.emp TO strmadmin_dept;

CONN sys/password@DBA1 AS SYSDBA

CREATE TABLESPACE logmnr_ts1 DATAFILE 'E:\TestDB\ORADATA\DBA1\logmnr02.dbf'
SIZE 10 M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('logmnr_ts1');

CONN sys/password@DBA1 AS SYSDBA

ALTER TABLE demo.department ADD SUPPLEMENTAL LOG GROUP log_group_department_pk1 (departmentno) ALWAYS;
ALTER TABLE demo.emp ADD SUPPLEMENTAL LOG GROUP log_group_emp_pk1 (empno) ALWAYS;

CONNECT strmadmin_dept/strmadminpw@DBA1


BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
schema_name => 'demo',
streams_name => 'dba1_to_dba2',
source_queue_name => 'strmadmin_dept.streams_queue',
destination_queue_name => 'strmadmin_dept.streams_queue@dba2',
include_dml => true,
include_ddl => true,
include_tagged_lcr => false,
source_database => 'dba1');
END;
/

CONNECT strmadmin_dept/strmadminpw@DBA1

BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'demo',
streams_type => 'capture',
streams_name => 'capture_demo',
queue_name => 'strmadmin_dept.streams_queue',
include_dml => true,
include_ddl => true,
source_database => 'dba1');
END;
/



exp userid=demo/demo@dba1 FILE=E:\TestDB\testDump\demo_instant.dmp OBJECT_CONSISTENT=y ROWS=n

imp userid=demo/demo@dba2 FILE=E:\TestDB\testDump\demo_instant.dmp IGNORE=y COMMIT=y LOG=import.log STREAMS_INSTANTIATION=y


CONN sys/password@DBA2 AS SYSDBA
ALTER TABLE demo.department DROP SUPPLEMENTAL LOG GROUP log_group_department_pk1;
ALTER TABLE demo.emp DROP SUPPLEMENTAL LOG GROUP log_group_emp_pk1;



CONNECT strmadmin_dept/strmadminpw@dba1
DECLARE
v_scn NUMBER;
BEGIN
v_scn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
DBMS_APPLY_ADM.SET_SCHEMA_INSTANTIATION_SCN@DBA2(
source_schema_name => 'demo',
source_database_name => 'dba1',
instantiation_scn => v_scn);
END;
/


CONNECT strmadmin_dept/strmadminpw@DBA2

BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'demo',
streams_type => 'apply',
streams_name => 'apply_demo',
queue_name => 'strmadmin_dept.streams_queue',
include_dml => true,
include_ddl => true,
source_database => 'dba1');
END;
/

CONNECT strmadmin_dept/strmadminpw@DBA2
BEGIN
DBMS_APPLY_ADM.SET_PARAMETER(
apply_name => 'apply_demo',
parameter => 'disable_on_error',
value => 'n');

DBMS_APPLY_ADM.START_APPLY(
apply_name => 'apply_demo');
END;
/


CONNECT strmadmin_dept/strmadminpw@DBA1
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(
capture_name => 'capture_demo');
END;
/