Oracle Streams Unidirectional to Bidirectional Setup

In the last post I went through the steps to setup a unidirectional Streams replicated environment. Here are the steps I followed to take the environment and make it bidirectional.

You'll recall that currently we have in the SOURCE database the following:

• A capture queue called "streams_capture_queue"
• A Streams process called "streams_capture" - tied to the streams_capture_queue
• A Propagation process called "streams_propagation" - tied to the streams_capture_queue

And in the DESTINATION database we have:

• An apply queue called "streams_apply_queue
• A Streams process called "streams_apply" - tied to the streams_apply_queue

Basically we need to set all this up in reverse. In the words of that meerkat: "simples"!

And in reality it is pretty simple - we go through the same process to set up the queues/Streams processes in reverse. Obviously we will not be doing the data pump import as the data is already in the original source. What we will be doing however, is instantiating the original source schema so that any changes to existing objects made in the original destination database will propagate and be applied at the original source. Without this step any new objects created would be streamed back but any changes would not, you'd get an error that looks like this (found in the ERROR_MESSAGE column of the DBA_APPLY_ERROR view):

ORA-26687: no instantiation SCN provided for "TEST"."T" in source database "STREAMS2.WORLD"

So first we create the capture queue in the original DESTINATION database:

CONN streams admin/password@original destination database
BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE(
queue_table => 't_streams_capture',
queue_name => 'streams_capture_queue',
queue_user => 'streams admin');
END;
/

Then we create the apply queue in the original SOURCE database:

CONN streams admin/password@original source database
BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE(
queue_table => 't_streams_apply',
queue_name => 'streams_apply_queue',
queue_user => 'streams admin');
END;
/

Now we add apply rules for the schema to be streamed in the original SOURCE database:

CONN streams admin/password@original source database
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'schema to be streamed',
streams_type => 'apply',
streams_name => 'streams_apply',
queue_name => 'streams_apply_queue',
include_dml => true,
include_ddl => true,
source_database => 'original destination database');
END;
/

Next we specify the "apply user" (the schema we are looking to replicate) in the original SOURCE database:

CONN streams admin/password@original source database
BEGIN
DBMS_APPLY_ADM.ALTER_APPLY(
apply_name => 'streams_apply',
apply_user => 'schema to be streamed');
END;
/

and disable the "disable_on_error" parameter to stop any errors being show stoppers:

CONN streams admin/password@original source database
BEGIN
DBMS_APPLY_ADM.SET_PARAMETER(
apply_name => 'streams_apply',
parameter => 'disable_on_error',
value => 'n');
END;
/

Now we can start the apply process in the original SOURCE database:

CONN streams admin/password@original source database
EXECUTE DBMS_APPLY_ADM.START_APPLY(apply_name => 'streams_apply');

We now add capture rules for the schema to be streamed in the original DESTINATION database:

CONN streams admin/password@original destination database
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'schema to be streamed',
streams_type => 'capture',
streams_name => 'streams_capture',
queue_name => 'streams_capture_queue',
include_dml => true,
include_ddl => true,
source_database => 'original destination database');
END;
/

Once this has been done we can add propagation rules in the original DESTINATION database:

CONN streams admin/password@original destination database
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
schema_name => 'schema to be streamed',
streams_name => 'streams_propagation',
source_queue_name => 'streams_capture_queue',
destination_queue_name => 'streams_apply_queue@original source database',
include_dml => true,
include_ddl => true,
source_database => 'original destination database');
END;
/

Now to the important part - the instantiation of the schema in the original SOURCE database. First we need to get the current SCN from the original DESTINATION database:

CONN streams admin/password@original destination database
SELECT TO_CHAR(dbms_flashback.get_system_change_number()) FROM dual;

TO_CHAR(DBMS_FLASHBACK.GET_SYSTEM_CHANGE
----------------------------------------
7874142636454

Now prepare the schema for instantiation in the original SOURCE database:

CONN streams admin/password@original source database
BEGIN
DBMS_CAPTURE_ADM.PREPARE_SCHEMA_INSTANTIATION(
schema_name => 'schema to be streamed');
END;
/

And actually instantiate it with the SCN gathered previously:

CONN streams admin/password@original source database
BEGIN
DBMS_APPLY_ADM.SET_SCHEMA_INSTANTIATION_SCN(
source_schema_name => 'schema to be streamed',
source_database_name => 'original destination database',
instantiation_scn => '7874142636454',
recursive => true);
END;
/

We're done! All that's left is to begin the capture queue in the original DESTINATION database:

CONN streams admin/password@original destination database
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(capture_name => 'streams_capture');
END;
/

Cheers,

Ian