Streams Synchronous Capture 101

Streams Synchronous Capture is a New Feature of Oracle 11g; It doesn't rely on archive logs or redo logs to capture changes. Instead it captures changes when they are made. It uses a row level trigger-based mechanism to perform this task. As its name suggests, the only part that is "synchronous" is the capture; the changes continue to be applied asynchronously to the destination objects and the databases stay loosely coupled. As a matter of fact, propagation and apply processes don't change so much.

In addition to the capture itself, the main change associated with synchronous capture is the queue usage. Messages are persistent/commit time instead of buffered. The reason is that the capture process won't get any second chance to capture a change and, as consequence, it cannot afford loosing anything in a queue. As a result, every change of a synchronous capture is logged in all the databases it transit through and the fingerprint can be quite significant. It's worth to notice too, that synchronous capture can only propagate DML changes, it only works with a small subset of datatypes that doesn't include LOBs, and the capture process have to run on the source database. However, synchronous capture has its own set of advantages too:

• It can capture changes on a Standard Edition or a Standard Edition One 11g database
• It can be used even if the source database is running in noarchivelog mode and doesn't require you add any supplemental logging
• It's probably simpler to setup, though I feel like a regular Streams setup is pretty simple too
• It's a new feature of 11.1 ;-)
  

In this post, we'll reproduce the same setup we've built in our previous Oracle Streams One Way Table Replication 101 post, with synchronous capture this time; in order to do it, we'll go through the steps below:

• Build The Sample Schema and Table
  
• Create the Streams Administrator and Queue
• Create an Apply Process
• Create the Capture Process and Define the Capture Rules
• Transform the SQL Captured so that changes to SOURCE are applied to DESTINATION
• Instantiate the Table in the DESTINATION Schema
• Start the Apply Process
• Test/Validate the Replication
• Stop and Suppress the Oracle Streams Configuration

Step 1: Build The Sample Schema and Table

In order to start, we'll create a schema named SOURCE and a table named T2 the same way we've done it in the previous post. 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 t2(
  id number primary key,
  text varchar2(80));

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

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

commit;

Step 2: Create the Streams Administrator and Queue

The database doesn't require any specific settings. It doesn't rely either on the Multi Version Data Dictionary. We'll create a Streams Administrator in all the databases that are involved with the capture, propagation or apply processes; the script below does it. It creates a tablespace to store the queues. It creates a user STRMADMIN, grants it the privileges and roles as well as it creates the queue that will be used by the capture and the apply processes to share the LCR:

connect / as sysdba

CREATE TABLESPACE streams_tbs
  DATAFILE 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 3: Create an Apply Process

IMPORTANT :
Opposite to the asynchronous capture, synchronous capture is always running. You cannot start it, and more importantly you cannot stop it! The reason is that it has only one chance to capture changes and you could miss changes if you had that ability. A consequence of that and of the fact a running capture process must have one apply process is that you MUST create the apply process before you create the capture process. If you don't, you may encounter an error like the one below when you try to change your source tables:

insert into source.t1 values (5,'Text 5')
*
ERROR at line 1:
ORA-26694: error while enqueueing into queue STRMADMIN.STREAMS_QUEUE
ORA-24033: no recipients for message

Before rush on the creation of the apply process, one important point to understand:

• You'll want the consumed messages to be the ones captured by a synchronous capture and NOT by an asynchronous capture. That means that you want the apply process to subscribe to the persistent message part of the queue and NOT to the buffered part of it. The way you specify that is to set the applied_captured parameter to FALSE. That's for the subtle part: you should name that parameter "applied asynchronous captured" for yourself ;-). And here comes the confusing part, the 11.1 documentation for dbms_apply_adm.create_capture suggests that FALSE is the default value. That's not the case; the default is TRUE and that parameter has to be specified in the case of a synchronous capture.

The script below creates an apply process and configures it so that it subscribes to the strmadmin.streams_queue queue:

connect strmadmin/strmadmin

begin
  dbms_apply_adm.create_apply(
     queue_name      => 'strmadmin.streams_queue',
     apply_name      => 'sync_apply',
     apply_captured  => false,
     source_database => 'BLACK');
end;
/

begin
  dbms_streams_adm.add_table_rules(
    table_name         => 'source.t2',
    streams_type       => 'apply',
    streams_name       => 'sync_apply',
    queue_name         => 'strmadmin.streams_queue',
    include_dml        => true,
    include_ddl        => false,
    include_tagged_lcr => false,
    source_database    => 'BLACK',
    inclusion_rule     => true);
end;
/


set lines 120
col streams_name format a12
col streams_type format a12
col table_owner format a11
col table_name format a5
col rule_type format a8
col rule_name format a5

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

STREAMS_NAME STREAMS_TYPE TABLE_OWNER TABLE RULE_TYP RULE
------------ ------------ ----------- ----- -------- ----
SYNC_APPLY   APPLY        DESTINATION T2    DML      T226

Step 4: Create a Capture Process

You can now create the synchronous capture process. You can simply use DBMS_STREAMS_ADM.ADD_TABLE_RULES with streams_type=>'sync_capture' like below:

connect strmadmin/strmadmin

BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
     table_name   => 'source.t2',
     streams_type => 'sync_capture',
     streams_name => 'sync_capture',
     queue_name   => 'strmadmin.streams_queue');
END;
/

set lines 120
col streams_name format a12
col streams_type format a12
col table_owner format a11
col table_name format a5
col rule_type format a8
col rule_name format a4

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

STREAMS_NAME STREAMS_TYPE TABLE_OWNER TABLE RULE_TYP RULE
------------ ------------ ----------- ----- -------- ----
SYNC_CAPTURE SYNC_CAPTURE SOURCE      T2    DML      T228
SYNC_APPLY   APPLY        SOURCE      T2    DML      T226

Step 5: Add a Transformation Rule to the Table Apply Rule

Add the Transformation associated with the dml apply rule so that the schema name is transformed from SOURCE to DESTINATION. The script below adds the transformation to the apply. It is the same as the one for asynchronous capture:

begin
  dbms_streams_adm.rename_schema(
     rule_name        => 'T226' ,
     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
------ -------------------------- ------ ------------ -------------
T226   DECLARATIVE TRANSFORMATION SOURCE DESTINATION  RENAME SCHEMA

Step 6: 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. Here is an example with 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.t2(
  id   number primary key,
  text varchar2(80));

col apply_scn format 999999999999 new_value apply_scn

select dbms_flashback.get_system_change_number apply_scn
 from dual;

 APPLY_SCN
-----------
49042368875

insert into destination.t2
  (select *
     from source.t2 as of scn &&apply_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.t2',
     source_database_name => 'BLACK',
     instantiation_scn    => &&apply_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.T2   49042368875

Step 7: Start the Apply Processes

You can finally start the capture and the apply processes :

exec dbms_apply_adm.start_apply('sync_apply');

Step 8: Test/Validate the 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.t2(id, text)
  values (3,'Text 3');

commit;

pause

col id format 99
col text format a6

select id,
      text
from destination.t2;

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

Step 9: 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 also drop the 2 schemes:

connect / as sysdba

exec dbms_apply_adm.stop_apply('sync_apply');

exec dbms_capture_adm.drop_capture('sync_capture',true);
exec dbms_apply_adm.drop_apply('sync_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,
                  source_database
            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 => i.source_database,
         instantiation_scn    => null);
    end loop;
 for i in (select source_object_owner||'.'||
                  source_object_name name,
                  source_database
            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 => i.source_database,
         instantiation_scn    => null);
    end loop;
end;
/