Oracle Streams Schema Replication

The below procedure sets one way replication from source to target database

Initialization Parameters For Streams
Set the below intialization paramters


COMPATIBLE =10.1.0 or higher.

GLOBAL_NAMES = TRUE

JOB_QUEUE_PROCESSES= 4 or higher.

PARALLEL_MAX_SERVERS= 2 or higher.

SHARED_POOL_SIZE= 200 MB

OPEN_LINKS= 4 or higher.

TIMED_STATISTICS= true.

LOG_ARCHIVE_DEST_n= set atleast one arch destination, where n is 1, 2, 3, ... 10.

LOG_ARCHIVE_DEST_STATE_n= set to enable.

UNDO_RETENTION >900

_job_queue_interval=1


Create streams adminstrator and grant privileges on both Source and Target

create user STRMADMIN identified by STRMADMIN;
ALTER USER STRMADMIN DEFAULT TABLESACE STRMTAB;
ALTER USER STRMADMIN QUOTA UNLIMITED ON STRMTAB;
GRANT CONNECT, RESOURCE, AQ_ADMINISTRATOR_ROLE to STRMADMIN;
GRANT SELECT ANY DICTIONARY TO STRMADMIN;
GRANT EXECUTE ON DBMS_AQ TO STRMADMIN;
GRANT EXECUTE ON DBMS_AQADM TO STRMADMIN;
GRANT EXECUTE ON DBMS_FLASHBACK TO STRMADMIN;
GRANT EXECUTE ON DBMS_STREAMS_ADM TO STRMADMIN;
GRANT EXECUTE ON DBMS_CAPTURE_ADM TO STRMADMIN;
GRANT EXECUTE ON DBMS_APPLY_ADM TO STRMADMIN;
GRANT EXECUTE ON DBMS_RULE_ADM TO STRMADMIN;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO STRMADMIN;
GRANT DBA to STRMADMIN;
execute DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE('STRMADMIN');


Create streams queue

connect STRMADMIN/STRMADMIN

BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE(
queue_table => 'STREAMS_QUEUE_TABLE',
queue_name => 'STREAMS_QUEUE',
queue_user => 'STRMADMIN');
END;
/


On Target database ORCDDB2

Apply rules for the Schema at the destination database :

BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'SCOTT',
streams_type => 'APPLY ',
streams_name => 'STRMADMIN_APPLY',
queue_name => 'STRMADMIN.STREAMS_QUEUE',
include_dml => true,
include_ddl => true,
source_database => 'ORCLDB1');
END;
/

Specify an 'APPLY USER' at the destination database

User which needs to apply all DML statements and DDL statements.
The user specified in the APPLY_USER parameter must have the necessary
privileges to perform DML and DDL changes on the apply objects.

BEGIN
DBMS_APPLY_ADM.ALTER_APPLY(
apply_name => 'STRMADMIN_APPLY',
apply_user => 'SCOTT');
END;
/

Start the Apply process

DECLARE
v_started number;
BEGIN
SELECT decode(status, 'ENABLED', 1, 0) INTO v_started
FROM DBA_APPLY WHERE APPLY_NAME = 'STRMADMIN_APPLY';
if (v_started = 0) then
DBMS_APPLY_ADM.START_APPLY(apply_name => 'STRMADMIN_APPLY');
end if;
END;
/


Steps to be carried out at the Source Database ORCLDB1

Move LogMiner tables from SYSTEM tablespace:

By default, all LogMiner tables are created in the SYSTEM tablespace.
It is a good practice to create an alternate tablespace for the LogMiner
tables.

CREATE TABLESPACE LOGMNRTS DATAFILE 'logmnrts.dbf' SIZE 25M AUTOEXTEND ON
MAXSIZE UNLIMITED;

BEGIN
DBMS_LOGMNR_D.SET_TABLESPACE('LOGMNRTS');
END;
/

Turn on supplemental logging at database level

connect SYS/password as SYSDBA

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY, UNIQUE INDEX) COLUMNS;


Create a database link to the source and destination database

connect STRMADMIN/STRMADMIN

CREATE DATABASE LINK ORCLDB1 connect to
STRMADMIN identified by STRMADMIN using 'ORCLDB1';

Test the database link to be working properly by querying against the
destination database.
Eg : select * from global_name@ORCLDB1;

connect STRMADMIN/STRMADMIN

CREATE DATABASE LINK ORCLDB2 connect to
STRMADMIN identified by STRMADMIN using 'ORCLDB2';

Test the database link to be working properly by querying against the
destination database.
Eg : select * from global_name@ORCLDB2;

Add capture rules for the schema SCOTT at the source database

BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'SCOTT',
streams_type => 'CAPTURE',
streams_name => 'STREAM_CAPTURE',
queue_name => 'STRMADMIN.STREAMS_QUEUE',
include_dml => true,
include_ddl => true,
source_database => 'ORCLDB1');
END;
/

Add propagation rules for the schema SCOTT at the source database

This step will also create a propagation job to the destination database.

BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
schema_name => 'SCOTT',
streams_name => 'STREAM_PROPAGATE',
source_queue_name => 'STRMADMIN.STREAMS_QUEUE',
destination_queue_name => 'STRMADMIN.STREAMS_QUEUE@ORCLDB2',
include_dml => true,
include_ddl => true,
source_database => 'ORCLDB1');
END;
/

Export, import and instantiation of tables from Source to Destination Database

exp USERID=SYSTEM/manager@ORCLDB1 OWNER=SCOTT FILE=scott.dmp LOG=exportTables.log OBJECT_CONSISTENT=Y STATISTICS = NONE

imp USERID=SYSTEM@ORCLDB2 FILE=scott.dmp LOG=importTables.log FULL=Y CONSTRAINTS=Y IGNORE=Y COMMIT=Y STREAMS_INSTANTIATION=Y


Start the Capture process on source database ORCLDB1


begin
DBMS_CAPTURE_ADM.START_CAPTURE(capture_name => 'STREAM_CAPTURE');
end;
/


The setup is now ready to replicate data between the two databases using
Oracle Streams.


Removing a Streams Configuration

To Remove streams configuration use DBMS_STREAMS_ADM.REMOVE_STREAMS_CONFIGURATION and drop Streams user,

CONN SYS/MANAGER AS SYSDBA

EXEC DBMS_STREAMS_ADM.REMOVE_STREAMS_CONFIGURATION();

DROP USER STRMADMIN CASCADE;

SQL Queries for Monitoring Streams

Determining Capture to Dequeue Latency for a Message

COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A17
COLUMN LATENCY HEADING 'Latency|in|Seconds' FORMAT 9999
COLUMN CREATION HEADING 'Message Creation' FORMAT A17
COLUMN LAST_DEQUEUE HEADING 'Last Dequeue Time' FORMAT A20
COLUMN DEQUEUED_MESSAGE_NUMBER HEADING 'Dequeued|Message Number' FORMAT 999999

SELECT APPLY_NAME,
(DEQUEUE_TIME-DEQUEUED_MESSAGE_CREATE_TIME)*86400 LATENCY,
TO_CHAR(DEQUEUED_MESSAGE_CREATE_TIME,'HH24:MI:SS MM/DD/YY') CREATION,
TO_CHAR(DEQUEUE_TIME,'HH24:MI:SS MM/DD/YY') LAST_DEQUEUE,
DEQUEUED_MESSAGE_NUMBER
FROM V$STREAMS_APPLY_READER;



General information about each apply process in a database

COLUMN APPLY_NAME HEADING 'Apply|Process|Name' FORMAT A15
COLUMN QUEUE_NAME HEADING 'Apply|Process|Queue' FORMAT A15
COLUMN RULE_SET_NAME HEADING 'Positive|Rule Set' FORMAT A15
COLUMN NEGATIVE_RULE_SET_NAME HEADING 'Negative|Rule Set' FORMAT A15
COLUMN STATUS HEADING 'Apply|Process|Status' FORMAT A15

SELECT APPLY_NAME,
QUEUE_NAME,
RULE_SET_NAME,
NEGATIVE_RULE_SET_NAME,
STATUS
FROM DBA_APPLY;


COLUMN APPLY_NAME HEADING 'Apply Process Name' FORMAT A20
COLUMN APPLY_CAPTURED HEADING 'Type of Messages Applied' FORMAT A25
COLUMN APPLY_USER HEADING 'Apply User' FORMAT A30

SELECT APPLY_NAME,
DECODE(APPLY_CAPTURED,
'YES', 'Captured',
'NO', 'User-Enqueued') APPLY_CAPTURED,
APPLY_USER
FROM DBA_APPLY;


List Parameter Settings for Each Apply Process


COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A15
COLUMN PARAMETER HEADING 'Parameter' FORMAT A25
COLUMN VALUE HEADING 'Value' FORMAT A20
COLUMN SET_BY_USER HEADING 'Set by User?' FORMAT A15

SELECT APPLY_NAME,
PARAMETER,
VALUE,
SET_BY_USER
FROM DBA_APPLY_PARAMETERS;

Displays All of the Error Handlers for Local Apply Processes

COLUMN OBJECT_OWNER HEADING 'Table|Owner' FORMAT A5
COLUMN OBJECT_NAME HEADING 'Table Name' FORMAT A10
COLUMN OPERATION_NAME HEADING 'Operation' FORMAT A10
COLUMN USER_PROCEDURE HEADING 'Handler Procedure' FORMAT A30
COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A15

SELECT OBJECT_OWNER,
OBJECT_NAME,
OPERATION_NAME,
USER_PROCEDURE,
APPLY_NAME
FROM DBA_APPLY_DML_HANDLERS
WHERE ERROR_HANDLER = 'Y'
ORDER BY OBJECT_OWNER, OBJECT_NAME;


Displays the Message Handler for Each Apply Process

COLUMN APPLY_NAME HEADING 'Apply Process Name' FORMAT A20
COLUMN MESSAGE_HANDLER HEADING 'Message Handler' FORMAT A20

SELECT APPLY_NAME, MESSAGE_HANDLER FROM DBA_APPLY
WHERE MESSAGE_HANDLER IS NOT NULL;

Displays Precommit Handler for Each Apply Process

COLUMN APPLY_NAME HEADING 'Apply Process Name' FORMAT A20
COLUMN PRECOMMIT_HANDLER HEADING 'Precommit Handler' FORMAT A30
COLUMN APPLY_CAPTURED HEADING 'Type of|Messages|Applied' FORMAT A15

SELECT APPLY_NAME,
PRECOMMIT_HANDLER,
DECODE(APPLY_CAPTURED,
'YES', 'Captured',
'NO', 'User-Enqueued') APPLY_CAPTURED
FROM DBA_APPLY
WHERE PRECOMMIT_HANDLER IS NOT NULL;


Displays Information About the Reader Server for Each Apply Process

COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A15
COLUMN APPLY_CAPTURED HEADING 'Apply Type' FORMAT A22
COLUMN PROCESS_NAME HEADING 'Process|Name' FORMAT A7
COLUMN STATE HEADING 'State' FORMAT A17
COLUMN TOTAL_MESSAGES_DEQUEUED HEADING 'Total Messages|Dequeued' FORMAT 99999999

SELECT r.APPLY_NAME,
DECODE(ap.APPLY_CAPTURED,
'YES','Captured LCRS',
'NO','User-enqueued messages','UNKNOWN') APPLY_CAPTURED,
SUBSTR(s.PROGRAM,INSTR(s.PROGRAM,'(')+1,4) PROCESS_NAME,
r.STATE,
r.TOTAL_MESSAGES_DEQUEUED
FROM V$STREAMS_APPLY_READER r, V$SESSION s, DBA_APPLY ap
WHERE r.SID = s.SID AND
r.SERIAL# = s.SERIAL# AND
r.APPLY_NAME = ap.APPLY_NAME;

Monitoring Transactions and Messages Spilled by Each Apply Process


COLUMN APPLY_NAME HEADING 'Apply Name' FORMAT A20
COLUMN 'Transaction ID' HEADING 'Transaction ID' FORMAT A15
COLUMN FIRST_SCN HEADING 'First SCN' FORMAT 99999999
COLUMN MESSAGE_COUNT HEADING 'Message Count' FORMAT 99999999

SELECT APPLY_NAME,
XIDUSN ||'.'||
XIDSLT ||'.'||
XIDSQN "Transaction ID",
FIRST_SCN,
MESSAGE_COUNT
FROM DBA_APPLY_SPILL_TXN;


COLUMN APPLY_NAME HEADING 'Apply Name' FORMAT A15
COLUMN TOTAL_MESSAGES_SPILLED HEADING 'Total|Spilled Messages' FORMAT 99999999
COLUMN ELAPSED_SPILL_TIME HEADING 'Elapsed Time|Spilling Messages' FORMAT 99999999.99

SELECT APPLY_NAME,
TOTAL_MESSAGES_SPILLED,
(ELAPSED_SPILL_TIME/100) ELAPSED_SPILL_TIME
FROM V$STREAMS_APPLY_READER;


Displaying General Information About Each Coordinator Process

COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A17
COLUMN LATENCY HEADING 'Latency|in|Seconds' FORMAT 9999
COLUMN CREATION HEADING 'Message Creation' FORMAT A17
COLUMN LAST_DEQUEUE HEADING 'Last Dequeue Time' FORMAT A20
COLUMN DEQUEUED_MESSAGE_NUMBER HEADING 'Dequeued|Message Number' FORMAT 999999

SELECT APPLY_NAME,
(DEQUEUE_TIME-DEQUEUED_MESSAGE_CREATE_TIME)*86400 LATENCY,
TO_CHAR(DEQUEUED_MESSAGE_CREATE_TIME,'HH24:MI:SS MM/DD/YY') CREATION,
TO_CHAR(DEQUEUE_TIME,'HH24:MI:SS MM/DD/YY') LAST_DEQUEUE,
DEQUEUED_MESSAGE_NUMBER
FROM V$STREAMS_APPLY_READER;


Displaying Information About Transactions Received and Applied

COLUMN APPLY_NAME HEADING 'Apply Process Name' FORMAT A25
COLUMN TOTAL_RECEIVED HEADING 'Total|Trans|Received' FORMAT 99999999
COLUMN TOTAL_APPLIED HEADING 'Total|Trans|Applied' FORMAT 99999999
COLUMN TOTAL_ERRORS HEADING 'Total|Apply|Errors' FORMAT 9999
COLUMN BEING_APPLIED HEADING 'Total|Trans Being|Applied' FORMAT 99999999
COLUMN TOTAL_IGNORED HEADING 'Total|Trans|Ignored' FORMAT 99999999

SELECT APPLY_NAME,
TOTAL_RECEIVED,
TOTAL_APPLIED,
TOTAL_ERRORS,
(TOTAL_ASSIGNED - (TOTAL_ROLLBACKS + TOTAL_APPLIED)) BEING_APPLIED,
TOTAL_IGNORED
FROM V$STREAMS_APPLY_COORDINATOR;

V$STREAMS_APPLY_COORDINATOR Query for Latency

COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A17
COLUMN 'Latency in Seconds' FORMAT 999999
COLUMN 'Message Creation' FORMAT A17
COLUMN 'Apply Time' FORMAT A17
COLUMN HWM_MESSAGE_NUMBER HEADING 'Applied|Message|Number' FORMAT 999999

SELECT APPLY_NAME,
(HWM_TIME-HWM_MESSAGE_CREATE_TIME)*86400 "Latency in Seconds",
TO_CHAR(HWM_MESSAGE_CREATE_TIME,'HH24:MI:SS MM/DD/YY')
"Message Creation",
TO_CHAR(HWM_TIME,'HH24:MI:SS MM/DD/YY') "Apply Time",
HWM_MESSAGE_NUMBER
FROM V$STREAMS_APPLY_COORDINATOR;

DBA_APPLY_PROGRESS Query for Latency

COLUMN APPLY_NAME HEADING 'Apply Process|Name' FORMAT A17
COLUMN 'Latency in Seconds' FORMAT 999999
COLUMN 'Message Creation' FORMAT A17
COLUMN 'Apply Time' FORMAT A17
COLUMN APPLIED_MESSAGE_NUMBER HEADING 'Applied|Message|Number' FORMAT 999999

SELECT APPLY_NAME,
(APPLY_TIME-APPLIED_MESSAGE_CREATE_TIME)*86400 "Latency in Seconds",
TO_CHAR(APPLIED_MESSAGE_CREATE_TIME,'HH24:MI:SS MM/DD/YY')
"Message Creation",
TO_CHAR(APPLY_TIME,'HH24:MI:SS MM/DD/YY') "Apply Time",
APPLIED_MESSAGE_NUMBER
FROM DBA_APPLY_PROGRESS;

Checking for Apply Errors

COLUMN APPLY_NAME HEADING 'Apply|Process|Name' FORMAT A10
COLUMN SOURCE_DATABASE HEADING 'Source|Database' FORMAT A10
COLUMN LOCAL_TRANSACTION_ID HEADING 'Local|Transaction|ID' FORMAT A11
COLUMN ERROR_NUMBER HEADING 'Error Number' FORMAT 99999999
COLUMN ERROR_MESSAGE HEADING 'Error Message' FORMAT A20
COLUMN MESSAGE_COUNT HEADING 'Messages in|Error|Transaction' FORMAT 99999999

SELECT APPLY_NAME,
SOURCE_DATABASE,
LOCAL_TRANSACTION_ID,
ERROR_NUMBER,
ERROR_MESSAGE,
MESSAGE_COUNT
FROM DBA_APPLY_ERROR;