Streams Implementer's Guide

A book about Oracle is like a UFO: unprobable and exceptional. I wish I know better the 80 volumes of the Oracle Database documentation. However, a few times, a few books have provided me with such a different way to look and consider things that it has transformed my way of thinking. A book can turn something you couldn't get into something obvious. And I'm sure you know what and who I'm talking about: let's call it talent.

That's why, despite the many disappointments, I continue to buy books about Oracle and continue to hope, like you can discover new genius friends, that the next book I'll buy will stay near me for a while.

So when the Streams Implementer's Guide Editor sent me a free copy of this book, UFO among UFOs, I've started to hope and wish again. And like a made on purpose, I just got my copy right now at the time I'm leaving for holidays. I don't know yet if I'll spend a few minutes or days reading it but it's probably worth to mention it. And, by the way, "How can you get to the point where you would start a book about Oracle Streams ?"

I'll share my enthusiasm when I'll be back. Nice subject and challenge at a time Oracle replication technology is for sure living one of its greatest paradoxes since 9i Release 2...

To be continued!
Read more...

Oracle 11gR2, GoldenGate and Koala Jump Start

It was some time ago now, but Oracle GoldenGate statement of direction made me sad. Sometimes, I still wish the person who wrote it is somehow as far away as possible from the database server development team. But I know that's not the case. The sentence that is killing me for sure is:

"Given the strategic nature of Oracle GoldenGate, Oracle Streams will continue to be supported, but will not be actively enhanced. Rather, the best elements of Oracle Streams will be evaluated for inclusion with Oracle GoldenGate."

Some Background

Even today, a few weeks after I've read GoldenGate documentation, tested the product and I've listed GoldenGate benefits for myself (openness, simplicity, completeness, ...). I'm still getting some trouble swallowing I will miss some of the unique features of Streams.

Not that I'm intolerant or I'm stuck to the "good old time". I mean, I know there are bad and good things about Streams. During the past years, I've spent quite some energy^(*) on leveraging what is so unique in it and I believe most of its advantages come from its "built-inside" architecture. To a point where it could be very difficult for Oracle to provide Streams strengths to GoldenGate because their architectures are so different.

You may wonder what I am talking about. Well, I'm talking about:

• Streams in-memory end-to-end propagation that provides an extremely scalable framework without the need to log changes in another separate file (trail)
  
• The reliability of the staging (the database) and propagation (buffered queues).
  
• Streams lighweight footprint that relies (most of the time) on internal object ids, can easily be offloaded to a separate server without any database copy and doesn't access original data at all
  
• The ability Streams has to treat set of changes without "deferring" the constraints
  
• The flexibility of the framework that is running inside the database with hooks at every stage you can completely personalize and extend to your needs
  
• The openness for developers that can directly use the APIs to provide advanced features to their applications by leveraging their Oracle knowledge
  
• The omnipotence and ubiquity of the solution that is already installed with Oracle 11g, work with RAC no change, is extended by the Change Data Capture feature and easy to share with the outside with AQ or XStream.
  
• The maturity, ease and simplicity it has gained over the years of hard work by the development team and you can leverage by a simple SQL script
  
• The cost model that comes for free with Oracle Database Enterprise Edition and with a limited set of features with 11g Standard Edition
  

At some point, I was hoping XStream would be the missing link between Streams and GoldenGate so that GoldenGate could leverage the power of Oracle Database 11g. That would for sure have been my preferred scenario and the fact XStream is licenced with GoldenGate made me hope for a while. But the reading of the statement of direction suggests a very different end.

Now Oracle GoldenGate is very likely to, if not succeed, do a better job than Streams to bring real-time data integration to the mass ;-). Nevertheless, I have no clue how Oracle could, in a timely manner, provide Streams strengths to it. Interesting challenge for the product managers and development team... The future will tell us!

Koala jump start

As you can guess, time has come for all of us to learn more about Oracle GoldenGate. Since my laptop runs Karmic Koala, I've given it a try on Ubuntu 9.10. It's not supported but did not face any issue so far. I've downloaded and installed GoldenGate from Oracle E-Delivery and I've choosen the following distribution:

• Select a product Pack: "Oracle Fusion Middleware"
• Platform: "Linux x86"
• Description: "Oracle GoldenGate on Oracle Media Pack for Linux x86"
• Name: "Oracle GoldenGate V10.4.0.x for Oracle 11g on RedHat 5.0"

To perform the installation, I've just unzip/untar the file in a directory and I've set the environment variable so that I can access my Oracle Database:

. oraenv
ORACLE_SID = [WHITE] ? BLACK
The Oracle base for ORACLE_HOME=/u01/app/oracle/product/11.2.0/db_1 is /u01/app/oracle

That done, you should be ready to use Oracle GoldenGate; run the command line interface like below:

./ggsci 

Oracle GoldenGate Command Interpreter for Oracle
Version 10.4.0.19 Build 002
Linux, x86, 32bit (optimized), Oracle 11 on Sep 29 2009 08:50:50

Copyright (C) 1995, 2009, Oracle and/or its affiliates.  All rights reserved.

exit

Prepare the database

There are a few settings to check on the database. It must be in archivelog mode and GoldenGate must be able to access the archivelogs and redologs:

sqlplus / as sysdba

archive log list
Database log mode        Archive Mode
Automatic archival        Enabled
Archive destination        /u01/app/oracle/oradata/BLACK/archivelogs
Oldest online log sequence     1
Next log sequence to archive   3
Current log sequence        3

I've set the NLS_LANG so that it matches the one from my database:

sqlplus / as sysdba

select parameter, value
  from nls_database_parameters
 where parameter in ('NLS_LANGUAGE',
                     'NLS_TERRITORY',
                     'NLS_CHARACTERSET',
                     'NLS_LENGTH_SEMANTICS');

parameter            VALUE
-------------------- -------------
NLS_LANGUAGE         AMERICAN
NLS_TERRITORY        AMERICA
NLS_CHARACTERSET     WE8MSWIN1252
NLS_LENGTH_SEMANTICS BYTE

exit

export NLS_LANG=AMERICAN_AMERICA.WE8MSWIN1252

The configuration has to rely on a user with a high level of privileges to perform several operations, like extracting the content of an UDT or a LOBs with flashback queries. It should also be able to set supplemental log groups to the tables that are part of the replication:

sqlplus / as sysdba

create user gg
  identified by gg
  default tablespace users
  temporary tablespace temp;

grant create session, resource, dba to gg;

The database must also have the minimal supplemental logging enabled:

alter database add supplemental log data;

alter system switch logfile;

In my case, that was it. Obviously, based on what you want to do, your database may require more changes. It could require some tables to audit the DDL or to store checkpoints to be created. To get the complete list of what needs to be done, check GoldenGate's documentation and more specifically the "Oracle Installation and Setup Guide".

A Demo Schema

I did not try to setup anything advanced. I've just replicated scott.dept in a demo schema. To make the example even more simple, I've considered nobody was accessing the table and I did not pay any attention to pending transactions or out-of-sync instantiation. I've just created a table named demo.dept and loaded it with scott.dept's data:

create user demo identified by demo
       default tablespace users
       temporary tablespace temp
       quota unlimited on users;

grant connect, resource to demo;

create table demo.dept
 ( deptno number(2,0),
 dname  varchar2(14),
 loc    varchar2(13),
        constraint pk_dept primary key(deptno))
  tablespace users;

insert into demo.dept
   select * from scott.dept;

commit;

Configure GoldenGate Process Manager

Once the database configured, refer to the "Administration Guide" to continue and get more details about GoldenGate setup. First, it requires some directories to store its configuration, logs and trail files. You can choose an alternate location for them but that wasn't really my concern either:

./ggsci 

create subdirs

Creating subdirectories under current directory /gg

Parameter files                /gg/dirprm: created
Report files                   /gg/dirrpt: created
Checkpoint files               /gg/dirchk: created
Process status files           /gg/dirpcs: created
SQL script files               /gg/dirsql: created
Database definitions files     /gg/dirdef: created
Extract data files             /gg/dirdat: created
Temporary files                /gg/dirtmp: created
Veridata files                 /gg/dirver: created
Veridata Lock files            /gg/dirver/lock: created
Veridata Out-Of-Sync files     /gg/dirver/oos: created
Veridata Out-Of-Sync XML files /gg/dirver/oosxml: created
Veridata Parameter files       /gg/dirver/params: created
Veridata Report files          /gg/dirver/report: created
Veridata Status files          /gg/dirver/status: created
Veridata Trace files           /gg/dirver/trace: created
Stdout files                   /gg/dirout: created

Once done, I've edited the Manager configuration file named MGR to set the port parameter and I've started it:

edit params mgr

view params mgr
port 7809

start manager

status manager
Manager is running (IP port arkzoyd.7809).

Source Table Supplemental Log Group

Like Streams (there is no secret!), GoldenGate needs to be able to identify rows to apply captured changes. It provides some generic tools to enable and check additional logging to the tables from ggsci:

dblogin userid gg, password gg
Successfully logged into database.

add trandata scott.dept
Logging of supplemental redo data enabled for table SCOTT.DEPT.

info trandata scott.dept
Logging of supplemental redo log data is enabled for table SCOTT.DEPT

Parameter Files

I've named the extract scott. It captures changes made to the SCOTT.DEPT table and send them to the remote trail file that, in my case, is managed by the same manager. I've named the replicat demo. The parameter files for scott and demo looks like the ones below:

edit params scott

view params scott

extract scott
userid gg, password gg
rmthost localhost mgrport 7809
rmttrail SC
table SCOTT.DEPT;

edit params demo

view params demo

replicat demo
assumetargetdefs
userid gg, password gg
map SCOTT.DEPT, target DEMO.DEPT;

Note:
With Oracle, you have to use double-quote to manage case-sensitive table names. However, that's not the case with all the database engines. As a result, depending on the parameter, GoldenGate may or may not differentiate strings with different cases. To avoid any issue, I use uppercase for the parameter values, unless I want to specifically use a different case.

Extract and Replicat

Once the parameter files defined, I've added the extract, the replicat and the trail files from ggsci:

add extract scott, tranlog, begin now
EXTRACT added.

add rmttrail SC, extract scott
RMTTRAIL added.

add replicat demo, exttrail SC, nodbcheckpoint, begin now
REPLICAT added.

And I've started them both:

start er *
Sending START request to MANAGER ...
EXTRACT SCOTT starting

Sending START request to MANAGER ...
REPLICAT DEMO starting

info all

Program     Status      Group       Lag           Time Since Chkpt
MANAGER     RUNNING                                           
EXTRACT     RUNNING     SCOTT       00:00:00      00:00:02    
REPLICAT    RUNNING     DEMO        00:00:00      00:00:08

exit

Are Changes Replicated?

Once the setup completed, I've tested the replication with the script below:

sqlplus / as sysdba

update scott.dept 
   set dname='OPERATIONS2' 
 where deptno=40;

commit;

select dname from demo.dept
  where deptno=40;

DNAME
--------------
OPERATIONS2

update scott.dept 
   set dname='OPERATIONS' 
 where deptno=40;

commit;

select dname from demo.dept
  where deptno=40;

DNAME
--------------
OPERATIONS

exit

Configuration Cleanup

Obviously that's just a start. To avoid any issue with my next tests, I've cleaned up my configuration, once happy with it:

./ggsci

stop er *
Sending STOP request to EXTRACT SCOTT ...
Request processed.

Sending STOP request to REPLICAT DEMO ...
Request processed.

delete er *
Are you sure you want to delete all groups? y
Deleted EXTRACT SCOTT.
Deleted REPLICAT DEMO.

stop manager
Manager process is required by other GGS processes.
Are you sure you want to stop it (y/n)? y

Sending STOP request to MANAGER ...
Request processed.
Manager stopped.

exit

And I've dropped the demo and gg users:

sqlplus / as sysdba

drop user gg cascade;
drop user demo cascade;

exit

(*) I know at least one person that considers I'm a useless clown and the time I spend on Streams should be treated less mercifully. I admit it easily. I'm aware I'm not close to one percent of his knowledge about what stays an awesome technology. Anyway, good or bad, I've spent some energy on digging into Streams internals and I've always tried to listen to positive and negative feedback.
Read more...

LOB Chunks

To continue with the previous test case, I'll insert a large enough LOB so that the change record is splitted into several chunks. To perform that operation :

• Setup the example as described in Streams LCRs and LOBs
• Insert a LOB that is large enough to guaranty it will be deleted
  

What is interesting is less the output than the number of chunks and the size of each chunks.

Insert a large CLOB

The script below create a LOB as a temporary resource and, once built, insert it in SOURCE.T9:

connect source/source

declare
  z clob;
begin
  DBMS_LOB.CREATETEMPORARY(z,true,DBMS_LOB.SESSION);
  for i in 1..20 loop
     dbms_lob.append(z,rpad('Z',1024,'Z'));
  end loop;
  insert into source.T9 values (2,z);
end;
/
commit;

connect strmadmin/strmadmin
set lines 1000
set serveroutput on
exec print_xml_fromq('MYQUEUE');

The result

You'll find below the output that matches the set of LCRs generated by the one only insert:

---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
               http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>INSERT</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <new_values>
      <new_value>
         <column_name>ID</column_name>
         <data>
     <number>2</number>
         </data>
      </new_value>
      <new_value>
         <column_name>TEXT</column_name>
         <data>
     <varchar2 nil="true">
         </data>
         <lob_information>EMPTY LOB</lob_information>
      </new_value>
   </new_values>
</row_lcr>
---------------------------------------------------------
---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
               http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>LOB WRITE</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <new_values>
      <new_value>
         <column_name>ID</column_name>
         <data>
     <number>2</number>
         </data>
      </new_value>
      <new_value>
         <column_name>TEXT</column_name>
         <data>
            <varchar2>ZZZ[...]ZZZ</varchar2>
         </data>
         <lob_information>LAST LOB CHUNK</lob_information>
         <lob_offset>1</lob_offset>
      </new_value>
   </new_values>
</row_lcr>
---------------------------------------------------------
---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
               http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>LOB WRITE</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <new_values>
      <new_value>
         <column_name>ID</column_name>
         <data>
     <number>2</number>
         </data>
      </new_value>
      <new_value>
         <column_name>TEXT</column_name>
         <data>
            <varchar2>ZZZ[...]ZZZ</varchar2>
         </data>
         <lob_information>LAST LOB CHUNK</lob_information>
         <lob_offset>8061</lob_offset>
      </new_value>
   </new_values>
</row_lcr>
---------------------------------------------------------
---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
              http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>LOB WRITE</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <new_values>
      <new_value>
          <column_name>ID</column_name>
          <data>
      <number>2</number>
          </data>
      </new_value>
      <new_value>
          <column_name>TEXT</column_name>
          <data>
             <varchar2>ZZZ[...]ZZZ</varchar2>
          </data>
          <lob_information>LAST LOB CHUNK</lob_information>
          <lob_offset>16121</lob_offset>
      </new_value>
   </new_values>
</row_lcr>
---------------------------------------------------------
---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
              http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>LOB TRIM</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <new_values>
      <new_value>
         <column_name>ID</column_name>
         <data>
     <number>2</number>
         </data>
      </new_value>
      <new_value>
         <column_name>TEXT</column_name>
            <data>
        <varchar2 nil="true">
            </data>
         <lob_information>LAST LOB CHUNK</lob_information>
         <lob_operation_size>20480</lob_operation_size>
      </new_value>
   </new_values>
</row_lcr>
---------------------------------------------------------
---------------------------------------------------------
<ROW_LCR xmlns="http://xmlns.oracle.com/streams/schemas/lcr"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://xmlns.oracle.com/streams/schemas/lcr
               http://xmlns.oracle.com/streams/schemas/lcr/streamslcr.xsd">
   <source_database_name>BLACK</source_database_name>
   <command_type>UPDATE</command_type>
   <object_owner>SOURCE</object_owner>
   <object_name>T9</object_name>
   <transaction_id>5.5.913</transaction_id>
   <scn>1448259</scn>
   <old_values>
      <old_value>
         <column_name>ID</column_name>
         <data>
     <number>2</number>
         </data>
      </old_value>
   </old_values>
</row_lcr>
---------------------------------------------------------
No more messages

Read more...

The New Streams 11.2 SQL Generation Facility

I must admit, I'm kind of disappointed by the Oracle 11.2 Streams new features. Not that there is nothing important: the support for SecureFile LOBs and compressed tables are really things, I've been eagerly waiting for. This new release is huge and looks promising! However, I'm disappointed for the following 3 reasons:

• I did not manage to use statement handlers yet. Though I find them very easy to use for some specific use cases. But by hand or with dbms_streams_adm.maintain_change_table, they've always been failing with ORA-01008: not all variables bound
• I was kind of hoping for the following restriction to be handled: "The DBMS_STREAMS_ADVISOR_ADM package does not gather information about synchronous captures or messaging clients.".
• Last but not least, there are obvious hidden and undocumented features which raise my level of frustration to its maximum.

Anyway, I imagine I have no other choices than to be patient and wish those features can be fixed and documented, hopefully, as soon as possible. In the meantime, I propose we explore one of those new features called SQL Generation.

SQL Generation is the ability you get in a Custom DML Handler or any piece of code that deal with LCR to transform it into its canonical SQL command. What we'll do is write a very simple example that will use a DML handler to store that SQL command in a table instead of applying it to the destination schema.

Step 1: Create a sample Schema

To begin with the sample, create a source schema with a table and a few rows:

connect / as sysdba

create user source
  identified by source
   default tablespace users
   temporary tablespace temp;

grant connect,resource to source;

col dbname new_value dbname

select value dbname
 from v$parameter
where name='db_unique_name';

prompt &&dbname

connect source/source

create table t5(
 id    number primary key,
 text1 varchar2(80),
 text2 varchar2(80));

insert into t5(id, text1, text2)
 values (1,'Text 1','Text 1');

insert into t5(id, text1, text2)
 values (2,'Text 2','Text 2');

commit;

Step 2: Create the Streams administrator and a queue

Once done, you can create a Streams Administrator and a queue to use in your configuration:

connect / as sysdba

create tablespace streams_tbs
  datafile '/u01/app/oracle/oradata/BLACK/streams_tbs01.dbf'
  size 25M autoextend on maxsize 256M;

CREATE USER strmadmin
  IDENTIFIED BY strmadmin
  DEFAULT TABLESPACE streams_tbs
  QUOTA UNLIMITED ON streams_tbs
  temporary tablespace temp;

grant dba to strmadmin;

begin
  dbms_streams_adm.set_up_queue(
     queue_table =>'strmadmin.streams_queue_table',
     queue_name  =>'strmadmin.streams_queue');
end;
/

exec dbms_streams_auth.grant_admin_privilege('strmadmin', true);

select *
  from dba_streams_administrator;

USERNAME  LOC ACC
--------- --- ---
STRMADMIN YES YES

Step 3: Create a Capture Process

Then create a capture process on the source schema:

connect / as sysdba

var first_scn number;
set serveroutput on

DECLARE
 scn NUMBER;
BEGIN
 DBMS_CAPTURE_ADM.BUILD(
    first_scn => scn);
 DBMS_OUTPUT.PUT_LINE('First SCN Value = ' || scn);
 :first_scn := scn;
END;
/

col first_scn new_value first_scn
select :first_scn first_scn
 from dual;

connect strmadmin/strmadmin

prompt &&first_scn
prompt &&dbname

BEGIN
 DBMS_CAPTURE_ADM.CREATE_CAPTURE(
    queue_name         => 'strmadmin.streams_queue',
    capture_name       => 'SQLGEN_CAPTURE',
    rule_set_name      => NULL,
    source_database    => '&&dbname',
    use_database_link  => false,
    first_scn          => &&first_scn,
    logfile_assignment => 'implicit');
END;
/

col capture_name format a15
col queue_name format a13
col first_scn format 999999999999
col start_scn format 999999999999
col rule_set_name format a11

select capture_name,
      queue_name,
      first_scn,
      start_scn,
      rule_set_name
 from dba_capture;

BEGIN
 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name         => 'source.t5',
    streams_type       => 'capture',
    streams_name       => 'SQLGEN_CAPTURE',
    queue_name         => 'strmadmin.streams_queue',
    include_dml        => true,
    include_ddl        => false,
    include_tagged_lcr => true,
    source_database    => '&&dbname',
    inclusion_rule     => true);
END;
/

set lines 120
col streams_name format a16
col streams_type format a9
col table format a15
col rule_type    format a8
col rule_name    format a15
col rule_condition format a60 wor wra

select streams_name,
      streams_type,
      table_owner||'.'||table_name "TABLE",
      rule_type,
      rule_name,
      rule_condition
 from dba_streams_table_rules
where streams_name='SQLGEN_CAPTURE'
  and table_owner='SOURCE'
  and table_name='T5';

Step 4: Create a procedure and a table to store the SQL of the LCR

Don't apply the LCR on the destination; instead, create a table and a procedure that will use SQL Generation to store the change vector as a DML command in that table:

connect strmadmin/strmadmin

create table mydml_tab(
      id      number,
      sqltext clob);

create sequence mydml_seq;

create or replace procedure mydml_handler(in_any in anydata)
is
 lcr            sys.lcr$_row_record;
 v_sqltext      clob:=' /* COMMENT */';
 rc             number;
begin
 rc := in_any.GETOBJECT(lcr);
   lcr.get_row_text(v_sqltext);
 insert into mydml_tab
    values (mydml_seq.nextval,v_sqltext);
end;
/

Step 5: Create an Apply process with a DML handler

Create an apply process and add the DML handler to it:

connect strmadmin/strmadmin

BEGIN
 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
     table_name         => 'source.t5',
     streams_type       => 'apply',
     streams_name       => 'SQLGEN_APPLY',
     queue_name         => 'strmadmin.streams_queue',
     include_dml        => true,
     include_ddl        => false,
     include_tagged_lcr => true,
     source_database    => '&&dbname',
     inclusion_rule     => true);
END;
/
col apply_name format a13
col queue_name format a13
col rule_set_name format a11

select apply_name,
      queue_name,
      rule_set_name,
      status
 from dba_apply;

set lines 120
col streams_name format a16
col streams_type format a9
col table_owner  format a11
col table_name   format a15
col rule_type    format a8
col rule_name    format a15

select STREAMS_NAME,
      STREAMS_TYPE,
      TABLE_OWNER,
      TABLE_NAME,
      RULE_TYPE,
      RULE_NAME
 from DBA_STREAMS_TABLE_RULES
where STREAMS_NAME='SQLGEN_APPLY';

begin
 dbms_apply_adm.set_dml_handler(
    object_name    => 'SOURCE.T5',
    object_type    => 'TABLE',
    operation_name => 'INSERT',
    error_handler  => false,
    user_procedure => 'strmadmin.mydml_handler',
    apply_name     => 'SQLGEN_APPLY');
end;
/

begin
  dbms_apply_adm.set_dml_handler(
     object_name    => 'SOURCE.T5',
     object_type    => 'TABLE',
     operation_name => 'UPDATE',
     error_handler  => false,
     user_procedure => 'strmadmin.mydml_handler',
     apply_name     => 'SQLGEN_APPLY');
end;
/

begin
  dbms_apply_adm.set_dml_handler(
      object_name    => 'SOURCE.T5',
      object_type    => 'TABLE',
      operation_name => 'DELETE',
      error_handler  => false,
      user_procedure => 'strmadmin.mydml_handler',
      apply_name     => 'SQLGEN_APPLY');
end;
/

begin
  dbms_apply_adm.set_table_instantiation_scn(
      source_object_name   => 'source.t5',
      source_database_name => '&&dbname',
      instantiation_scn    => &&first_scn);
end;
/

col "OBJ3CT" format a15
col operation_name format a8
col user_procedure format a28
col apply_name format a13

select object_owner||'.'||object_name "OBJ3CT",
      operation_name,
      user_procedure,
      apply_name
 from dba_apply_dml_handlers;

col source_database format a10
col "OBJECT" format a15
set numwidth 15
select source_database,
      SOURCE_OBJECT_OWNER||'.'||SOURCE_OBJECT_NAME "OBJECT",
      instantiation_scn
 from dba_apply_instantiated_objects;

Step 6: Start the Apply and Capture processes

Now start the apply and the capture processes and wait for them to keep up with the current position of the log writer:

exec dbms_apply_adm.start_apply('SQLGEN_APPLY')
exec dbms_capture_adm.start_capture('SQLGEN_CAPTURE')

Step 7: Test the SQL Generation

You can test your settings by inserting or updating a row in the source table; After some time, query the table that stores the generated SQL:

insert into source.t5 values (3,'Hello','Hello');
commit;

-- Wait a few minutes
col id format 99
col sqltext format a50
set long 1000
set longchunksize 1000
select * from mydml_tab;

The table content should look like below:

 ID SQLTEXT
--- --------------------------------------------------
1   /* COMMENT */ INSERT INTO "SOURCE"."T5"("ID","TEX
    T1","TEXT2" ) VALUES ( 3,'Hello','Hello')

You can perform more tests with UPDATE or DELETE

Step 8: Drop the test environment

Like always with my example, I propose you drop the whole configuration before you leave:

connect / as sysdba

exec dbms_apply_adm.stop_apply('SQLGEN_APPLY')
exec dbms_capture_adm.stop_capture('SQLGEN_CAPTURE')

exec dbms_apply_adm.delete_all_errors('SQLGEN_APPLY');
exec dbms_apply_adm.drop_apply('SQLGEN_APPLY')
exec dbms_capture_adm.drop_capture('SQLGEN_CAPTURE')

begin
for i in (select source_object_owner||'.'||
           source_object_name name
     from dba_apply_instantiated_objects
    where source_object_owner in ('SOURCE'))
loop
dbms_apply_adm.set_table_instantiation_scn(
  source_object_name   => i.name,
  source_database_name => '&&dbname',
  instantiation_scn    => null);
end loop;
end;
/

begin
for i in (select object_owner||'.'||
           object_name name,
           operation_name,
           apply_name
     from dba_apply_dml_handlers
    where object_owner in ('SOURCE'))
loop
dbms_apply_adm.set_dml_handler(
  object_name   => i.name,
  object_type   => 'TABLE',
  operation_name=> i.operation_name,
  user_procedure=> null,
  apply_name    => i.apply_name);
end loop;
end;
/

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 source cascade;

Read more...

Tracking Streams Changes with V$STREAMS_MESSAGE_TRACKING

V$STREAMS_MESSAGE_TRACKING is one of the best new features of Oracle Streams 11g. For a full description and an example, check the "Tracking LCRs Through a Stream" section of the documentation. In this post, we'll be creating a procedure after setting a 1-way Streams replication and the message tracking. You'll see how easy to use it is. To shorten this post, I won't be describing how to configure the replication. The 2 scripts attached, should help you to quickly get to it:

• 05_StreamsSchema1WayReplication.sql, assuming you have 2 databases and you can connect with SQL*Plus as oracle, creates a 1-way streams replication of the DEMO schema
• 05_StreamsSchema1WayReplication_CleanUp.sql drops the whole configuration.

Connect to the source database (i.e. BLACK), set the label and create a procedure in the DEMO schema:

. oraenv
BLACK

sqlplus / as sysdba

set serveroutput on

declare
tracking_label  VARCHAR2(15);
begin
dbms_streams_adm.set_message_tracking(
tracking_label => 'ARKZOYD',
actions        => DBMS_STREAMS_ADM.ACTION_MEMORY);
END;
/

create or replace procedure demo.hizoyd is
begin
null;
end;
/

Check the view to see how the message has been managed by Streams:

set lines 120
col component_name format a19
col component_type format a18
col action format a23
col object_owner format a14
col object_name format a14
col command_type format a16
select component_name,
     component_type,
     action,
     object_owner,
     object_name,
     command_type
from V$STREAMS_MESSAGE_TRACKING
where tracking_label='ARKZOYD'
order by timestamp;

Here is the result:

COMPONENT_NAME     COMPONENT_TYPE      ACTION                     OBJECT_OWNER   OBJECT_NAME      COMMAND_TYPE
----------------- ----------------- ----------------------- ------------- -------------- ----------------
STREAMS_CAPTURE     CAPTURE             Created                    Missing MVDD:0 Missing MVDD:0 CREATE PROCEDURE
STREAMS_CAPTURE     CAPTURE             Rule evaluation            Missing MVDD:0 Missing MVDD:0 CREATE PROCEDURE
STREAMS_CAPTURE     CAPTURE             Enqueue                    Missing MVDD:0 Missing MVDD:0 CREATE PROCEDURE
STREAMS_CAPTURE     CAPTURE             Created                                                   COMMIT
STREAMS_CAPTURE     CAPTURE             Enqueue                                                   COMMIT
STREAMS_PROPAGATION PROPAGATION SENDER  Dequeued                   Missing MVDD:0 Missing MVDD:0 CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION SENDER  Propagation Sender sent    Missing MVDD:0 Missing MVDD:0 CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION SENDER  Dequeued                                                  COMMIT
STREAMS_PROPAGATION PROPAGATION SENDER  Propagation Sender sent                                  COMMIT

The tracking is propagated to the destination database (i.e. WHITE) like you can see below :

exit

. oraenv
WHITE

sqlplus / as sysdba

set pages 20
set lines 120
col component_name format a19
col component_type format a20
col action format a28
col object_owner format a14
col object_name format a14
col command_type format a16
select component_name,
     component_type,
     action,
     object_owner,
     object_name,
     command_type
from V$STREAMS_MESSAGE_TRACKING
where tracking_label='ARKZOYD'
order by timestamp;

Here is the result:

COMPONENT_NAME      COMPONENT_TYPE       ACTION                       OBJECT_OWNER    OBJECT_NAME    COMMAND_TYPE
----------------- ------------------ ------------------------- ------------- -------------- ----------------
STREAMS_PROPAGATION PROPAGATION RECEIVER Propagation receiver enqueue Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION RECEIVER Rule evaluation              Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION RECEIVER Rule evaluation              Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION RECEIVER Propagation receiver enqueue Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_PROPAGATION PROPAGATION RECEIVER Propagation receiver enqueue                                 COMMIT
STREAMS_PROPAGATION PROPAGATION RECEIVER Rule evaluation                                              COMMIT
STREAMS_PROPAGATION PROPAGATION RECEIVER Rule evaluation                                              COMMIT
STREAMS_PROPAGATION PROPAGATION RECEIVER Propagation receiver enqueue                                 COMMIT
STREAMS_APPLY       APPLY READER         Dequeued                      Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_APPLY       APPLY READER         Dequeued                                                     COMMIT
STREAMS_APPLY       APPLY SERVER         Apply executed                Missing MVDD:0 Missing MVDD:0  CREATE PROCEDURE
STREAMS_APPLY       APPLY SERVER         Commit                                                       COMMIT

And you can check the procedure has been replicated to the WHITE database:

select text    
 from dba_source
 where owner='DEMO'
   and name='HIZOYD'
 order by line;

TEXT
-------------------
procedure hizoyd is
 begin
   null;
 end;

Read more...

Are your triggers triggered by Streams?

That sounds like a fair question, don't you think? Like often with Oracle the answer is split between "It depends" and "That's not as easy as I thought first". Let's start with how it's supposed to work; From the documentation, How an apply process behaves depends on the fire_once property of the trigger:

• If the trigger fire_once is TRUE, then the trigger is not executed by an apply process
• If the trigger fire_once is FALSE, then the trigger is suppose to be always executed, wether that's a regular DML/DDL command or an Streams apply process performing the change

That sounds easy. Lets have a look with a concrete example!

To build the test case, we'll implement a simple Streams One-Way Replication. To do it in 10 minutes, assuming your database is in archivelog, follow Step 1 to Step 9 of the "Oracle Streams One Way Table Replication 101" post. Ready?

Test when fire_once is TRUE (i.e. Default)

To begin, we'll create a table and a trigger in the DESTINATION schema; we'll check if the trigger is supposed to fire ONCE or ALWAYS:

connect / as sysdba

create table destination.t1_x
  (id   number primary key,
   text varchar2(80));

create or replace trigger destination.t1_x_trigger
 after insert on destination.T1 for each row 
  begin
    insert into T1_X(id, text)
      values (:new.id, :new.text);
  end;
/

set serveroutput on
begin
 if (dbms_ddl.IS_TRIGGER_FIRE_ONCE('DESTINATION', 'T1_X_TRIGGER'))
   then
    dbms_output.put_line('Trigger FIRE_ONCE=TRUE');
  else
    dbms_output.put_line('Trigger FIRE_ONCE=FALSE');
  end if;
end;
/

Trigger FIRE_ONCE=TRUE

So that's really the default and that trigger is not supposed to be executed by an apply process. But before, we can test the trigger is working fine by inserting a value in the DESTINATION.T1 table manually (That will make the 2 tables diverge but who cares?):

insert into DESTINATION.T1
   values (9999, 'Text 9999');
commit;

select * 
  from DESTINATION.T1_X;

  ID TEXT
---- ---------
9999 Text 9999

Now that we are sure the tigger is working fine, we can test the trigger is not executed by the apply process. We'll change the SOURCE.T1table this time and will check the values in changed on the DESTINATION.T1 but not in DESTINATION.T1_X as expected:

insert into SOURCE.T1
   values (4, 'Text 4');
commit;

pause

select * 
  from DESTINATION.T1;

  ID TEXT
---- ---------
   1 Text 1
   2 Text 2
   3 Text 3
9999 Text 9999
   4 Text 4

select * 
  from DESTINATION.T1_X;

  ID TEXT
---- ---------
9999 Text 9999

Very good!

Test when fire_once is FALSE

Now we'll change the trigger's behavior and we'll perform another test to see if the trigger is executed when the change is applied to the DESTINATION.T1 table:

begin
  DBMS_DDL.SET_TRIGGER_FIRING_PROPERTY(
   trig_owner => 'DESTINATION', 
   trig_name  => 'T1_X_TRIGGER',
   fire_once  => false);
end;
/

set serveroutput on
begin
 if (dbms_ddl.IS_TRIGGER_FIRE_ONCE('DESTINATION', 'T1_X_TRIGGER'))
   then
    dbms_output.put_line('Trigger FIRE_ONCE=TRUE');
  else
    dbms_output.put_line('Trigger FIRE_ONCE=FALSE');
  end if;
end;
/

Trigger FIRE_ONCE=FALSE

insert into SOURCE.T1
   values (5, 'Text 5');
commit;

pause

select * 
  from DESTINATION.T1;

  ID TEXT
---- ---------
   1 Text 1
   2 Text 2
   3 Text 3
9999 Text 9999
   4 Text 4
   5 Text 5

select * 
  from DESTINATION.T1_X;

  ID TEXT
---- ---------
9999 Text 9999

But wait a minute! Wasn't that trigger supposed to be executed? Let's restart all the processes to see if anything changes:

exec dbms_capture_adm.stop_capture('STREAMS_CAPTURE');
exec dbms_apply_adm.stop_apply('STREAMS_APPLY');
exec dbms_capture_adm.start_capture('STREAMS_CAPTURE');
exec dbms_apply_adm.start_apply('STREAMS_APPLY');

set serveroutput on
begin
 if (dbms_ddl.IS_TRIGGER_FIRE_ONCE('DESTINATION', 'T1_X_TRIGGER'))
   then
    dbms_output.put_line('Trigger FIRE_ONCE=TRUE');
  else
    dbms_output.put_line('Trigger FIRE_ONCE=FALSE');
  end if;
end;
/

Trigger FIRE_ONCE=FALSE

insert into SOURCE.T1
   values (6, 'Text 6');

commit;

select * 
  from DESTINATION.T1;
  2  
  ID TEXT
---- ----------
   1 Text 1
   2 Text 2
   3 Text 3
9999 Text 9999
   4 Text 4
   5 Text 5
   6 Text 6

select * 
  from DESTINATION.T1_X;

  ID TEXT
---- ----------
9999 Text 9999

No, Same Story! Actually, I had to recreate the trigger to make it work:

exec dbms_apply_adm.stop_apply('STREAMS_APPLY');
drop trigger destination.t1_x_trigger;

create or replace trigger destination.t1_x_trigger
 after insert on destination.T1 for each row 
  begin
    insert into T1_X(id, text)
      values (:new.id, :new.text);
  end;
/

begin
  DBMS_DDL.SET_TRIGGER_FIRING_PROPERTY(
   trig_owner => 'DESTINATION', 
   trig_name  => 'T1_X_TRIGGER',
   fire_once  => false);
end;
/

set serveroutput on
begin
 if (dbms_ddl.IS_TRIGGER_FIRE_ONCE('DESTINATION', 'T1_X_TRIGGER'))
   then
    dbms_output.put_line('Trigger FIRE_ONCE=TRUE');
  else
    dbms_output.put_line('Trigger FIRE_ONCE=FALSE');
  end if;
end;
/

Trigger FIRE_ONCE=FALSE

exec dbms_apply_adm.start_apply('STREAMS_APPLY');

insert into SOURCE.T1
   values (7, 'Text 7');

commit;

select * 
  from DESTINATION.T1;

  ID TEXT
---- ----------
   1 Text 1
   2 Text 2
   3 Text 3
9999 Text 9999
   4 Text 4
   5 Text 5
   6 Text 6
   7 Text 7

select * 
  from DESTINATION.T1_X;

  ID TEXT
---- ----------
9999 Text 9999
   7 Text 7

Expected Behavior? Bug? I'm still split! What do you think? btw, don't forget to clean up your Streams configuration.
Read more...

Propagation Error And Exception Queue Management

If you've tried to use the example from my previous blog post named "Transfer Files From One Server To Another With Oracle Streams", you may have faced some propagation issues. Step 7 of the post helps to quickly work around any issue by stopping and restarting the propagation. However, this is just a work around and in a real case scenario, you would want to manage the message in the exception queue.

This post goes deeper into the right way to fix any propagation issue; to begin, it simulates a propagation error, just in case you did not hit any by yourself ;-). The next section shows how you could fix the situation and manage the exception message by your own.

1. Simulate a Propagation Error

Like with DBMS_FILE_TRANSFER, external files transported with Oracle Streams must be multiple of 512 bytes in size. We'll use that limit to make a propagation process fail. To set up the example, follow "Step 1" to "Step 5" of the previous post "Transfer Files From One Server To Another With Oracle Streams". Once done, instead of executing "Step 6", execute the step below that will make the propagation fail.
On BLACK, create a file that is NOT a multiple of 512 bytes in size, instantiate a BFILE that points to it and enqueue the message with that points to the file in the Streams queue:

connect strmadmin/strmadmin

select * from GLOBAL_NAME;

GLOBAL_NAME
----------
BLACK

!dd if=/dev/zero of=/tmp/black/demo2.zero bs=511 count=1

declare
v_demotyp demotyp:=demotyp(2, bfilename('tmp_dir','demo2.zero'));
v_any     anydata;
begin
v_any:=anydata.convertobject(v_demotyp);
dbms_streams_messaging.enqueue(
  'strmadmin.streams_queue',
  v_any);
end;
/
commit;

This time the propagation fails with a message like the one below:

col error_message format a100
set lines 100
set long 10000
set longchunksize 10000

select error_message
from dba_propagation
where propagation_name='BFILE_PROPAGATION';

ERROR_MESSAGE
-------------------------------------------------------
ORA-19505: failed to identify file "/tmp/black/demo2.zero"
ORA-27046: file size is not a multiple of logical block size
Additional information: 1

You'll find the same error message in the queue propagation schedule:

select last_error_msg
from dba_queue_schedules
where schema='STRMADMIN'
 and qname='STREAMS_QUEUE'
 and message_delivery_mode='PERSISTENT';

LAST_ERROR_MSG
-------------------------------------------------------
ORA-19505: failed to identify file "/tmp/black/demo2.zero"
ORA-27046: file size is not a multiple of logical block size
Additional information: 1   

2. Fix the Error

Fixing the error depends on what it is. Let's say in this case, you'll fix the error by padding the file with a "0x00" byte; You can do it with a dd and the seek parameter:

Important note:
Don't use such a command on a real file without evaluating first what the impact would be on the file itself!

!dd if=/dev/zero of=/tmp/black/demo2.zero count=1 bs=1 seek=511 

!ls -ltra /tmp/black/demo2.zero
-rw-r--r-- 1 oracle oinstall 512 2009-08-24 14:26 /tmp/black/demo2.zero

That done, dequeue the message from the exception queue and re-enqueue it in the Streams queue; in that case that's probably a good idea to have only one queue per queue table to know the original queue of the message in the exception queue. Here is how to do it; if you want more examples, refer to Metalink Note "233103.1 - Dequeuing Messages from an Exception Queue":

connect strmadmin/strmadmin

begin
dbms_aqadm.start_queue(
 queue_name => 'strmadmin.AQ$_STREAMS_QUEUE_TABLE_E',
 enqueue    => false,
 dequeue    => true);
end;
/

SET SERVEROUTPUT ON

DECLARE
 v_deq_options     DBMS_AQ.dequeue_options_t;
 v_msg_properties  DBMS_AQ.message_properties_t;
 o_msgid           RAW(16);
 v_any             anydata;
BEGIN
 v_deq_options.consumer_name:=null;
 v_deq_options.dequeue_mode := DBMS_AQ.remove;
 v_deq_options.navigation := DBMS_AQ.NEXT_TRANSACTION;
 DBMS_AQ.dequeue(
    queue_name          => 'strmadmin.AQ$_STREAMS_QUEUE_TABLE_E',
    dequeue_options     => v_deq_options,
    message_properties  => v_msg_properties,
    payload             => v_any,
    msgid               => o_msgid);
 dbms_streams_messaging.enqueue('strmadmin.streams_queue',v_any);
 COMMIT;
END;
/

begin
dbms_aqadm.stop_queue(
 queue_name => 'strmadmin.AQ$_STREAMS_QUEUE_TABLE_E' ,
 enqueue    => true,
 dequeue    => true);
end;
/

Once done, you'll see the error go away from the propagation and the schedule. The file, now of a correct size, will be transferred from BLACK to WHITE:

select error_message
 from dba_propagation
where propagation_name='BFILE_PROPAGATION';

ERROR_MESSAGE
-------------------------------------------------------

select last_error_msg
 from dba_queue_schedules
where schema='STRMADMIN'
  and qname='STREAMS_QUEUE'
  and message_delivery_mode='PERSISTENT';

LAST_ERROR_MSG
-------------------------------------------------------

Check the file has been received on the WHITE side:

!ls -ltra /tmp/white/demo2.zero
-rw-r----- 1 oracle oinstall 512 2009-08-24 14:33 /tmp/white/demo2.zero

This is it: a very simple way to fix propagation errors. Once again, clean up the environment before you leave it for your next demo; delete the demo2.zero file on both sides of your configuration and execute "Step 8 of the previous post" to delete queues, tablespaces, administrators and more...
Read more...

Transfer Files From One Server To Another With Oracle Streams

You can leverage Oracle Streams to transfer files stored on a file system from one server to another. Not only, those files can be images or documents, assuming they are multiple of 512 bytes, but they also can be Oracle files, like Datapump exports, backups or data files. As a matter of fact, to me, sharing files across servers is probably the #1 reason why you would want to use Streams user messages.

In this post, you'll find an easy to reproduce sample application. You'll create a User Define Type (UDT) with a BFILE (i.e. a file located outside the database on a file system or in ASM). Once done, you'll create an instance of that type, you'll wrap it in an ANYDATA type and will share it between databases with Streams. You'll figure out that, not only the message with a BFILE locator, but also the corresponding file will be transferred from the source database to the destination database.

This post is made of the following sections:

• Step 1: Configure Databases, Streams Administrators and Directories
• Step 2: Configure Directories and the User Defined Type
• Step 3: Create the Database Link and the Propagation Between the Queues
• Step 4: Create A Streams Message Consumer and a Dequeue Program
• Step 5: Test The Streams configuration is working
• Step 6: Create a File and Transfer It with Streams
• Step 7: Troubleshooting
• Step 8: Clean-up the environment

Step 1: Configure Databases and Streams Administrators

If you read this blog or the documentation, you've seen it dozens of time already! In order for Streams to work, you must setup a Streams Administrator, a queue and, though it's not mandatory, it's strongly recommended to set the global_names parameter of the two databases to TRUE. In this sample application, you'll assume we have 2 databases called BLACK, the source, and WHITE, the destination. You'll find below a table with the commands to run on the 2 databases to setup the Streams Administrator and the parameters:

On BLACK	On WHITE
select * from global_name; GLOBAL_NAME ----------- BLACK alter system set global_names=true;	select * from global_name; GLOBAL_NAME ----------- WHITE alter system set global_names=true;
connect / as sysdba CREATE TABLESPACE streams_tbs DATAFILE '/u01/app/oracle/oradata/BLACK/streams_tbs.dbf' 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; / begin dbms_streams_auth.grant_admin_privilege( grantee => 'strmadmin', grant_privileges => true); end; /	connect / as sysdba CREATE TABLESPACE streams_tbs DATAFILE '/u01/app/oracle/oradata/WHITE/streams_tbs.dbf' 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; / begin dbms_streams_auth.grant_admin_privilege( grantee => 'strmadmin', grant_privileges => true); end; /

Note:
Because we don't use any Streams capture process, those databases don't have to be in archivelog mode.

Step 2: Configure Directories and the User Defined Type

Once the Streams Administrator created, you'll create a directory with the same name on the 2 databases as well as a User Defined Type. To allow the instances associated with the 2 databases to run on the same server, the directory path for BLACK differs from the directory path for WHITE:

On BLACK	On WHITE
!mkdir -p /tmp/black create directory tmp_dir as '/tmp/black'; grant read, write, execute on directory tmp_dir to strmadmin;	!mkdir -p /tmp/white create directory tmp_dir as '/tmp/white'; grant read, write, execute on directory tmp_dir to strmadmin;
connect strmadmin/strmadmin create type demotyp as object (id number, myfile bfile); /	connect strmadmin/strmadmin create type demotyp as object (id number, myfile bfile); /

Step 3: Create the Database Link and the Propagation Between the Queues

Make sure you can add an alias WHITE to connect to the WHITE database in the Oracle*Net configuration of the BLACK database (e.g. in the tnsnames.ora file). Once done, you can create a database link from BLACK to WHITE and create a Propagation process to send messages from the source to the destination database. Connect to the BLACK and run the script below:

connect strmadmin/strmadmin

select * from GLOBAL_NAME;

GLOBAL_NAME
----------
BLACK

CREATE DATABASE LINK WHITE
  connect to STRMADMIN
  identified by strmadmin
  using 'WHITE';

BEGIN
  DBMS_PROPAGATION_ADM.CREATE_PROPAGATION(
    propagation_name   => 'bfile_propagation',
    source_queue       => 'strmadmin.streams_queue',
    destination_queue  => 'strmadmin.streams_queue',
    destination_dblink => 'white',
    rule_set_name      => null,
    queue_to_queue     => TRUE);
END;
/

Step 4: Create A Streams Message Consumer And A Dequeue Program

You can now configure the WHITE database so that a program of yours dequeues messages from its Streams queue. The script below creates a Streams DEQUEUE client and execute an anonymous PL/SQL block that receives and removes any messages (if ID>0):

connect strmadmin/strmadmin

begin
 DBMS_STREAMS_ADM.ADD_MESSAGE_RULE (
   message_type   => 'strmadmin.demotyp',
   rule_condition => ':MSG.ID > 0',
   streams_type   => 'DEQUEUE',
   streams_name   => 'demostrm',
   queue_name     => 'strmadmin.streams_queue');
end;
/

select streams_name,
      queue_owner||'.'||queue_name queue
 from DBA_STREAMS_MESSAGE_CONSUMERS;

set serveroutput on
declare
  v_demotyp demotyp;
  v_any     anydata;
  v_out     pls_integer;
begin
   dbms_streams_messaging.dequeue(
      queue_name   => 'strmadmin.streams_queue',
      streams_name => 'demostrm',
      payload      => v_any,
      dequeue_mode => 'REMOVE',
      navigation   => 'NEXT TRANSACTION',
      wait         => DBMS_STREAMS_MESSAGING.FOREVER);
   v_out := v_any.getobject(v_demotyp);
   dbms_output.put_line('Message Received: '||
                        to_char(v_demotyp.id));
   commit;
end;
/

Step 5: Test The Streams Configuration

You can now test your settings. In order to do it, enqueue a message on the BLACK side. For this first test, don't "attach" any file to the message; leave myfile to NULL:

connect strmadmin/strmadmin

declare
 v_demotyp demotyp:=demotyp(1, null);
 v_any     anydata;
begin
 v_any:=anydata.convertobject(v_demotyp);
 dbms_streams_messaging.enqueue(
   'strmadmin.streams_queue',
   v_any);
end;
/
commit;

The program running on the WHITE should display the following line and stop:

Message Received: 1

Step 6: Create a File and Transfer It with Streams

You'll perform the same test, but this time with a file attached. Restart the program on WHITE:

select * from GLOBAL_NAME; 

GLOBAL_NAME
----------
WHITE 

set serveroutput on

declare
  v_demotyp demotyp;
  v_any     anydata;
  v_out     pls_integer;
begin
  dbms_streams_messaging.dequeue(
     queue_name   => 'strmadmin.streams_queue',
     streams_name => 'demostrm',
     payload      => v_any,
     dequeue_mode => 'REMOVE',
     navigation   => 'NEXT TRANSACTION',
     wait         => DBMS_STREAMS_MESSAGING.FOREVER);
  v_out := v_any.getobject(v_demotyp);
  dbms_output.put_line('Message Received: '||
                       to_char(v_demotyp.id));
  commit;
end;
/

Create a file that is a multiple of 512 bytes in size on the BLACK side, allocate a BFILE that points to it and send it to WHITE with Streams:

connect strmadmin/strmadmin

select * from GLOBAL_NAME; 

GLOBAL_NAME
----------
BLACK

!dd if=/dev/zero of=/tmp/black/demo.zero bs=512 count=1 

declare
   v_demotyp demotyp:=demotyp(2, bfilename('tmp_dir','demo.zero')); 
   v_any     anydata;
begin 
   v_any:=anydata.convertobject(v_demotyp); 
   dbms_streams_messaging.enqueue(
      'strmadmin.streams_queue',
      v_any);
end;
/
commit;

The program you've restarted on WHITE should display something like:

Message Received: 2

And the file should now be present on the WHITE side too:

!ls -l /tmp/white
total 4
-rw-r----- 1 oracle oinstall 512 2009-08-24 00:25 demo.zero

Step 7: Troubleshooting

Troubleshooting this sample is beyond the scope of the post. However if the configuration is not working, that's very likely it's related to the propagation of the file (the file size, some privileges on the /tmp/white directory, etc). You can query DBA_PROPAGATION and the queue tables to know more about the status of your message. If there is any error at the propagation level, fix it and you restart the propagation:

exec dbms_propagation_adm.stop_propagation('bfile_propagation',true);
exec dbms_propagation_adm.start_propagation('bfile_propagation');

Step 8: Clean-up the environment

As always, leave the database as it was when you've first started. The scripts below removes all the components:

On BLACK

On WHITE

connect / as sysdba !rm /tmp/black/demo.zero begin dbms_propagation_adm.stop_propagation( 'bfile_propagation',true); dbms_propagation_adm.drop_propagation( 'bfile_propagation',true); end; / drop directory tmp_dir; begin DBMS_STREAMS_ADM.REMOVE_QUEUE( queue_name => 'strmadmin.streams_queue', cascade => true, drop_unused_queue_table=> true); end; / drop user strmadmin cascade; drop tablespace streams_tbs including contents and datafiles;

connect / as sysdba !rm /tmp/white/demo.zero col streams_name format a15 col queue format a40 select streams_name, queue_owner||'.'||queue_name queue from DBA_STREAMS_MESSAGE_CONSUMERS; begin for i in (select rule_owner, rule_name from DBA_STREAMS_RULES where STREAMS_TYPE='DEQUEUE' and STREAMS_NAME='DEMOSTRM') loop dbms_streams_adm.remove_rule( rule_name => i.rule_owner||'.'||i.rule_name, streams_type => 'DEQUEUE', streams_name => 'DEMOSTRM', drop_unused_rule => true); end loop; end; / col streams_name format a15 col queue format a40 select streams_name, queue_owner||'.'||queue_name queue from DBA_STREAMS_MESSAGE_CONSUMERS; drop directory tmp_dir; drop user strmadmin cascade; drop tablespace streams_tbs including contents and datafiles;

In the next post, we'll get a bit deeper in the troubleshooting part; we'll generate an error with the propagation and we'll see how to fix it. Stay tuned!
Read more...

User-Defined DDL LCRs

In "User-Defined DML LCRs", I've shown how to use lcr$_row_record.construct to build LCRs for DELETE, UPDATE and INSERT; In the same post, I've created an apply process to execute those LCRs. I wanted to add an example of a DDL LCR at the same time, but because of an error with my 11.1.0.7 AL32UTF8 database, I have not been able to.

Not a big deal! I've created a WE8ISO8859P15 database and I'll correct that miss fire right away. You'll find below a sample configuration that shows how to use lcr$_ddl_record.construct.

Streams Queue and Apply

We'll use the same schema for the queue and for the apply process and to apply the DDL. To speed up the setup, I've granted DBA to the schema owner; There are 3 points worth to mention:

• I've set apply_capture parameter to false to make sure the apply process dequeues persistent messages
• I did not add any rule set to the apply so that it dequeues every message in the queue
• There is no need to define an instantiation SCN or to setup a SCN in the LCR (In the case of user-defined LCRs)

begin
 dbms_streams_adm.set_up_queue(
      queue_table => 'custom_queue_table',
      queue_name  => 'custom_queue');
end;
/
declare
 v_name varchar2(256);
begin
  select value into v_name
    from v$parameter
   where name='db_unique_name';
dbms_apply_adm.create_apply(
    queue_name => 'custom_queue',
    apply_name => 'custom_apply',
    apply_captured => false,
    source_database => v_name );
end;
/

exec dbms_apply_adm.start_apply('CUSTOM_APPLY');

Create and Enqueue A DDL LCR

Once everything ready you can build the LCRs and enqueue them in the queue so that the CREATE TABLE can be consumed and executed by the apply processes:

declare
  v_name varchar2(256);
  v_any     anydata;
  lcr       sys.lcr$_ddl_record;
  rc        pls_integer;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name and SCN
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 lcr:=sys.lcr$_ddl_record.construct(
            source_database_name => v_name,
            command_type         => 'CREATE TABLE',
            object_owner         => user,
            object_name          => 'MYTABLE',
            object_type          => 'TABLE',
            ddl_text             =>
              'create table MYTABLE(id number primary key)',
            logon_user           => user,
            current_schema       => user,
            base_table_owner     => user,
            base_table_name      => 'MYTABLE',
            tag                  => null,
            transaction_id       => null,
            scn                  => null);

 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);
end;
/

commit;

desc mytable

Name Null?    Type
---- -------- ------
ID   NOT NULL NUMBER

declare
  v_name varchar2(256);
  v_any     anydata;
  lcr       sys.lcr$_ddl_record;
  rc        pls_integer;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name and SCN
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 lcr:=sys.lcr$_ddl_record.construct(
            source_database_name => v_name,
            command_type         => 'DROP TABLE',
            object_owner         => user,
            object_name          => 'MYTABLE',
            object_type          => 'TABLE',
            ddl_text             =>
               'drop table MYTABLE purge',
            logon_user           => user,
            current_schema       => user,
            base_table_owner     => user,
            base_table_name      => 'MYTABLE',
            tag                  => null,
            transaction_id       => null,
            scn                  => null);
 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);
end;
/

commit;

desc mytable

ERROR:
ORA-04043: object mytable does not exist

Conclusion

This is it. With a WE8ISO8859P15 database, it works both with 10g and 11g and it should even work with Oracle Standard Edition... At least in theory!
Read more...

Oracle Streams Can Do Things, Mortals Can't! The Sequel.

In my post "Oracle Streams Can Do Things, Mortals Can't!", I've shown one of the challenges of applying changes, row by row, like Streams does. It can be the case when one of the impacted columns holds a unique, a primary key or a referential integrity constraint and you change several rows as part of the same update. I've also shown that Streams can actually handle those cases while you would have to DEFER the validation of those constraints to the commit time to perform the same operations manually.

I've written another post that has just been published with my other articles on The Pythian Group Blog. It's entitled "Delete and Re-Insert a Row in the Same Statement" and it's actually the sequel of my previous post. It shows how you can leverage Streams ability to run several SQL statements as if they were running as part of the same statement. You could use it to move rows... Even if your table doesn't have ROW MOVEMENT ENABLE and the constraints are not DEFERRABLE.
Read more...

User-Defined DML LCRs

One of the great features that comes with Oracle Streams is the ability you have to use the APIs to extend the framework and build almost anything you have in mind. Streams provides an easy way to share data across databases and with applications. In this post, I'll be providing an example of PL/SQL code that publishes DML LCRs in a queue; An Apply Handler will execute the content of those LCRs.

Test Schema

For the purpose of this demonstration, you can build one table T1 in the schema of your choice. Below is the script that creates that table and displays its content:

create table T1
   (id         number,
    text       varchar2(10),
    constraint T1_PK primary key(id));

insert into T1 values (1, 'Text 1');

commit;

col id format 99
col text format a6

select id, text
 from T1;

ID TEXT
-- ------
1 Text 1

Streams Queue and Apply

I use the same schema for the table, the queue and the apply process. To speed up the setup, I've granted DBA to the schema owner; There are 3 points worth to mention:

• I've set apply_capture parameter to false to make sure the apply process dequeues persistent messages
• I did not add any rule set to the apply so that it dequeues every message in the queue
• There is no need to define an instantiation SCN or to setup a SCN in the LCR

begin
 dbms_streams_adm.set_up_queue(
      queue_table => 'custom_queue_table',
      queue_name  => 'custom_queue');
end;
/
declare
 v_name varchar2(256);
begin
  select value into v_name
    from v$parameter
   where name='db_unique_name';
dbms_apply_adm.create_apply(
    queue_name => 'custom_queue',
    apply_name => 'custom_apply',
    apply_captured => false,
    source_database => v_name );
end;
/

exec dbms_apply_adm.start_apply('CUSTOM_APPLY');

Create and Enqueue DML LCRs

Once everything ready you can build the LCRs and enqueue them in the queue so that the DELETE, UPDATE or INSERT can be consumed and executed by the apply processes; You'll find below 3 example of such user defined DML LCR

• User Defined DELETE LCR

declare
  v_name varchar2(256);
  lcr       sys.lcr$_row_record;
  v_oldlist sys.lcr$_row_list;
  v_oldid   sys.lcr$_row_unit;
  v_oldtxt  sys.lcr$_row_unit;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 v_oldid:=sys.lcr$_row_unit(
         'ID',
         anydata.convertnumber(1),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_oldtxt:=sys.lcr$_row_unit(
         'TEXT',
         anydata.convertvarchar2('Text 1'),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_oldlist:=sys.lcr$_row_list(v_oldid,v_oldtxt);
 lcr:=sys.lcr$_row_record.construct(
            source_database_name=>v_name,
            command_type => 'DELETE',
            object_owner=> user,
            object_name => 'T1',
            tag => null,
            transaction_id => null,
            scn            => null,
            old_values     => v_oldlist);
 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);

end;
/
commit;

select *
 from t1;

no rows selected

• User Defined INSERT LCR

declare
  v_name varchar2(256);
  lcr       sys.lcr$_row_record;
  v_newlist sys.lcr$_row_list;
  v_newid   sys.lcr$_row_unit;
  v_newtxt  sys.lcr$_row_unit;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 v_newid:=sys.lcr$_row_unit(
         'ID',
         anydata.convertnumber(1),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_newtxt:=sys.lcr$_row_unit(
         'TEXT',
         anydata.convertvarchar2('Text 9'),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_newlist:=sys.lcr$_row_list(v_newid,v_newtxt);
 lcr:=sys.lcr$_row_record.construct(
            source_database_name=>v_name,
            command_type => 'INSERT',
            object_owner=> user,
            object_name => 'T1',
            tag => null,
            transaction_id => null,
            scn            => null,
            new_values     => v_newlist);
 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);

end;
/
commit;

select *
 from t1;

ID TEXT
-- ------
1 Text 9

• User Defined UPDATE LCR

declare
  v_name varchar2(256);
  lcr       sys.lcr$_row_record;
  v_oldlist sys.lcr$_row_list;
  v_oldid   sys.lcr$_row_unit;
  v_oldtxt  sys.lcr$_row_unit;
  v_newlist sys.lcr$_row_list;
  v_newid   sys.lcr$_row_unit;
  v_newtxt  sys.lcr$_row_unit;
  enqopt    DBMS_AQ.ENQUEUE_OPTIONS_T;
  mprop     DBMS_AQ.MESSAGE_PROPERTIES_T;
  enq_msgid RAW(16);
begin
  -- Get DB Name
  select value into v_name
    from v$parameter
   where name='db_unique_name';

 mprop.SENDER_ID := SYS.AQ$_AGENT(user, null, null);

 v_oldid:=sys.lcr$_row_unit(
         'ID',
         anydata.convertnumber(1),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_newid:=sys.lcr$_row_unit(
         'ID',
         anydata.convertnumber(1),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_oldtxt:=sys.lcr$_row_unit(
         'TEXT',
         anydata.convertvarchar2('Text 9'),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_newtxt:=sys.lcr$_row_unit(
         'TEXT',
         anydata.convertvarchar2('Text 1'),
         DBMS_LCR.NOT_A_LOB,
         null,
         null);
 v_oldlist:=sys.lcr$_row_list(v_oldid,v_oldtxt);
 v_newlist:=sys.lcr$_row_list(v_newid,v_newtxt);
 lcr:=sys.lcr$_row_record.construct(
            source_database_name=>v_name,
            command_type => 'UPDATE',
            object_owner=> user,
            object_name => 'T1',
            tag => null,
            transaction_id => null,
            scn            => null,
            old_values     => v_oldlist,
            new_values     => v_newlist);
 DBMS_AQ.ENQUEUE(
   queue_name          =>  'custom_queue',
   enqueue_options     =>  enqopt,
   message_properties  =>  mprop,
   payload             =>  anydata.ConvertObject(lcr),
   msgid               =>  enq_msgid);

end;
/
commit;

select *
 from t1;

ID TEXT
-- ------
1 Text 1

Conclusion

As you can see, creating and enqueuing DML LCRs is very simple.

I'm hiting an issue with DDL LCRs on top of 11.1.0.7/Linux x86 and, for now, the only way I manage to create one is by using dbms_streams.convert_xml_to_lcr. I'll write a new post as soon as my issue is fixed.
Read more...

Oracle Streams Can Do Things, Mortals Can't!

In my last post, "Do Your Tables Need Keys Be Replicated With Streams?", I've introduced the "Manual Streams Simulator". The goal was to show how Oracle Streams works by using the simplest model we can. What I'll be showing now is that, like every tool or model, the "Manual Streams Simulator" has its limits. Actually I'll be pointing one of the things Streams can do, that you and your programs can NOT...

So, this post is about one of the many challenges Oracle Streams faces... because of its design. A very simple example will help you understand what I mean. Let's consider a table T1 and with a column COL1 and a unique constraint on COL1:

create table T1 (COL1 number,
                constraint T1_UK unique(COL1));

insert into T1(COL1) values(1);
insert into T1(COL1) values(2);
commit;

To identify the rows in the database, you can query their ROWID:

select ROWID, COL1
 from T1;

ROWID              COL1
------------------ ----
AAATF5AABAAAVxpAAA    1
AAATF5AABAAAVxpAAB    2

With Oracle and with a single update, you turn the value 1 into 2 and turn the value 2 into 1; here is how:

update T1 set COL1=decode(COL1,1,2,1)
where COL1 in (1,2);
commit;

Note:
That's not the case with all the RDBMS; but if you work with Oracle that's not something you

If you don't trust me, check the ROWID after the update:

select ROWID, COL1
 from T1;

ROWID              COL1
------------------ ----
AAATF5AABAAAVxpAAA    2
AAATF5AABAAAVxpAAB    1

Just think about it: the Capture process or the synchronous capture would turn that update into 2 LCRs that would be applied by 2 differents statements by the server apply process. If you trace the Apply server process, you'll see each statement actually look like that:

update /*+ streams restrict_all_ref_cons  */ "DESTINATION"."T1" p
   set "COL1"=decode(:1,'N',"COL1",:2)
 where (:3='N') and (:4="COL1")

So? Do you see what I mean? If you want to execute 2 separate update statements to change 1 into 2 and 2 into 1, you will need to defer the unique constraint, right? A Streams apply executes the 2 updates without any defer! The reason I'm so sure, is because it will work even if the constraint is not deferrable on the destination.

How does it manage to do it? Obviously that's an internal trick! What I know for sure is that the SCN of the capture is involved with that and I strongly suspect the /*+ streams */ hint is part of the deal. But unless you get your hand on the code, I doubt you'll manage to guess more.

The good news is that you can leverage that internal trick for your own; in my next post, I'll show you how you could use Streams to run an INSERT and a DELETE as if they where executed at the exact same SCN. It will be posted with my other posts in my section of The Pythian Group Blog. Just be patient, I'll keep you updated.
Read more...

Do Your Tables Need Keys Be Replicated With Streams?

There is a lot of misunderstanding regarding the "requirement" for unique or primary keys on tables you want to replicate with Oracle Streams. That's probably worth some explanations...

Check the questions below:

• Do you need a primary or a unique key on a table to replicate it with Streams?
• How would you replicate a table that contains duplicated rows?
• What if the primary key of the destination table is different from the primary key of the source table?

This post, not only answers these questions, but also explains WHY things are like they are.

The Manual Streams Simulator

I'm not 100% sure what is the best way to understand the challenges of a SQL based replication. A way that has worked pretty well with me was to model it ^[1]. I've called that approach the Manual Streams Simulator.

The Manual Streams Simulator is nothing more than a reproduction of a Streams configuration, in which YOU immitate with SQL what Streams would capture, propagate or apply. Among the many advantages of this Simulator, there are its extreme simplicity and the fact it is free of code thus bug-free! For now, the way the replicated environment is instantiated is not of any real interest. For this reason we'll just build 2 tables and immitate a 1-way Streams table replication between those:

• T1 is the source table; it contains 3 columns and 3 lines and doesn't have any index or key
• T2 is a copy of T1 and will be used as the destination for the replication

The script below creates the 2 tables described above:

create table t1(
        col1 number,
        col2 number,
        col3 number);

insert into t1 values (1,1,1);
insert into t1 values (1,2,2);
insert into t1 values (1,2,3);

commit;

col col1 format 9999
col col2 format 9999
col col3 format 9999
col rowid format a20

select col1, col2, col3, rowid
 from t1;

COL1 COL2 COL3 ROWID
---- ---- ---- ------------------
  1    1    1 AAASzVAABAAAVxpAAA
  1    2    2 AAASzVAABAAAVxpAAB
  1    2    3 AAASzVAABAAAVxpAAC
  
create table t2 as select * from t1;

select col1, col2, col3, rowid
 from t2;

COL1 COL2 COL3 ROWID
---- ---- ---- ------------------
  1    1    1 AAASzWAABAAAVxxAAA
  1    2    2 AAASzWAABAAAVxxAAB
  1    2    3 AAASzWAABAAAVxxAAC

Not all DML statements are equals

The first thing that comes to mind right after instantiating the tables is that not all DML statements are equals. The simplest DML statement you can get by far is insert ... values (); if it is executed on the source table:

insert into t1(col1, col2, col3)
 values (2,3,4);

commit;

All you have to do is execute the same statement on the destination; In this case, we just need to change the table name from T1 to T2:

insert into t1(col1, col2, col3)
 values (2,3,4);

commit;

With an update or a delete, it's different story. Try to transform the SQL statements below so they can run on the destination:

update T1 set col3=5 where rowid='AAASzVAABAAAVxpAAB';

commit;

delete from T1 where rowid='AAASzVAABAAAVxpAAC';

commit;

What you can guess with this simple example is probably the first challenge of a SQL based replication. In the case a row is modified or deleted on the source, the process that applies the changes on the detination needs to identify the corresponding row. In the case of new inserted rows, there is no such a need.

A consequence of the above is that, if you replicate a table that is an insert-only table, it will work in all the cases; unless it contains unsupported data types... It won't matter if you have duplicated rows in it, if you have not enabled any supplemental logging or if you don't have any way to identify rows, wether it's a key or something else. It will always work! Obviously it is a different story with deletes and updates.

Note:
In the case of a recovery or a physical standby, the changes including the ones generated by deletes or updates are re-applied to the exact same data, the exact same rows. For this reason those processes can use the rowid to regenerate the changes; there is no such the constraints like the constraints of the SQL-based replication. Database recovery and physical standby MRP can rely on the rowid to replicate the changes.

Value-Based and Row-Level Capture

Most of, not to say all, the applications have SQL statements like this one:

insert into mytable
  values(1, sysdate);

commit;

If this statement is captured like it is executed, the resulting data would differ in many cases on the source and on the destination. This is because there is no way to guaranty the same statement can be applied at the same time on both side. To insure data don't differ on both side of the replication, the 2 streams implicit captures, Synchronous and Asynchronous, don't rely on the formulas included in the original SQL: They capture the values of the changes!

A direct consequence of this, is that when SQL statement modifies several rows, Streams implicit captures track the values of the changes for each one of the impacted rows; In fact they generate a Logical Change Record (LCR) for each one of the changed rows. These approaches also guaranty changes captured by a statement like the one below will be the same on the source and the destination:

delete from mytable
 where col2<=sysdate-360;
commit;

Note
Nothing prevents you from creating a trigger on the source database, capturing the original DML statements with the ora_sql_text event attribute, managing it as user message and applying it on the destination database. However, the likelyhood it will result in different data on the destination database is very close to 100%. In addition, there is no easy or reliable way to detect what statement could lead to different data (and that's just for a 1-way replication!).

How to identify changed rows?

Lets get back to the Manual Streams Simulator and run a statement like the one below:

update T1 set col3=5 where rowid='AAASzVAABAAAVxpAAB';

commit;

How can you write a SQL statement that would guaranty the exact same row will be changed on the destination table?

• When the source and destination tables have the same primary key

If the 2 tables have the same primary or unique key, it is easy to manage: capture the value of the key with the update and build the WHERE clause condition with it, instead of the rowid; Lets say the key is (COL2, COL3), what you would write to make sure the change will be the same on the source and destination is :

update T2 set col3=5
where col2=2 and col3=2;

commit;

The 2 Streams implicit captures behave exactly like the Manual Streams Simulator: (1) If you use the synchronous capture, the OLD and NEW values for all the columns are captured anyway so Streams Apply can easily build the WHERE clause condition like you did; (2) If you use the asynchronous capture, all you need is to add supplemental logging for the primary key or the unique keys on the source table to get to the same result.

• When the source and destination tables have a row identifier that is not enforced by a key or when the keys on the 2 tables don't match

You won't make any difference between this second case and the previous one for the Manual Streams Simulator. If you know there is a unique identifier, even if it's not enforced by a key or if the key is different on the destination, you won't pay attention to that. Lets say the row identifier is (COL2, COL3) too, what you would write to make sure the change will be the same on the source and destination is identical to what you've written previously:

update T2 set col3=5
where col2=2 and col3=2;

commit;

Obviously, Streams cannot guess what a row identifier that is not enforced by a key is. You will have to help it. In the case of the synchronous capture, all the OLD and NEW values for all the columns are captured so Streams will be able to build the SQL that corresponds with the row change without changing anything on the capture side; You just need you tell it what the identifier is on the apply side; In the case of the asynchronous capture, columns values are not captured by default and you'll need is to add the supplemental logging for the columns that are the unique identifier; here is how to do it:

alter table T1
 add supplemental log group T1_UID(COL2, COL3) always;

In both cases, you also need to tell the Apply process what the unique identifier is. There are 2 ways to do it:

• you can use the dbms_apply_adm.set_key_columns procedure and define the identifier to use on the destination
• you can create the primary key corresponding to the unique identifier of the source on the destination table. You can do it, even if the identifier you use is not enforced by a key on the source. Obviously it assume there is no duplicate values for the key on the destination.

Note:
If you don't create the key on the destination table and don't specify it with the dbms_apply_adm.set_key_columns procedure, the apply process assumes the key is made of all the columns of the destination table, except the ones of type LONG, LONG RAW and LOB. In the above example, if the update is captured by an asynchronous capture, it will fail with a message like the one below:

ORA-26787: The row with key ("COL1","COL2","COL3")= (,2,2) does not exist in table T2
ORA-01403: no data found

In the case of a synchronous capture, the same update will succeed because all the column values are captured and the LCR will actually contain (1,2,2)

• When the source table doesn't have any unique identifier

In that case, Streams assumes all the columns, except the ones of type LONG, LONG RAW and LOB, make a key. If you use asynchronous capture, make sure you capture the data of all those columns by running the statement below:

alter table T1
  add supplemental log data (all) columns;

Note
That approach may not work for you. If there are duplicate values for what Streams considers as an identifier, the replication will work until one of those duplicate values is modified. The Apply process will then generate an error that will look like the one below:

ORA-01422: exact fetch returns more than requested number of rows

And you'll have to manage that error manually. In most cases, using the allow_duplicate_rows parameter to workaround this situation is a bad idea; however if your table doesn't have any column of type LONG, LONG RAW and LOB, a DML handler that would modify only one of the several duplicated rows can help to work around that issue.

Retrieving rows efficiently on the destination

Lets now imagine T1 is 1 million rows, 1GB in size and you don't use any of those exadata storage servers; What if you run update T1 set col3=col3+1; on the source database? I confess this is probably not a good idea, even without Streams, but who knows what developers can write ?

Capturing this change with one of the Streams implicit capture may indeed be a very bad idea. It will result in 1 million LCR and 1 update per row on the destination. If the row identifier is made of one column and is not indexed on the destination, the apply server process will perform a "table full scan" for every one of the 1 million captured changes. You can easily imagine that will just make the replication impossible in this case.

So, if it doesn't really matter wether the source table has a primary key or not, the way you index the destination table can be very important for Streams to be viable. Keep that in mind when you setup a Streams replication with source tables that don't have primary keys.

Conclusion

The existence of keys, unique or primary, on the capture side of a Streams configuration greatly simplifies the replication. On the other hand, missing a key on a table doesn't mean it cannot be replicated with Streams. In most cases, you can avoid to create surrogate keys with a sequence and a trigger on the source, that won't be supported by most application:

• If a table is an insert-only table, and doesn't have unsupported data types, it can always be replicated with the implicit captures,
• If you can find a combination of columns to be used to identify rows, even if it's not enforced by a key, you can use it to setup the replication
• If a table has duplicated rows but no LONG, LONG RAW and LOB, you can write a DML handler to manage the error associated with the change of one of these rows

One other important aspect to consider if you build a Streams configuration is its purpose. For example if you use Streams to migrate a database from AIX to Linux, or the other way around, managing 95% of the problem may be enough to meet your downtime requirements; You can then complete your Streams strategy with the method of your choice. And that's already another story...


[1] If I tend to agree with Moshez thoughts about "science is inductive reasoning" VS "deductive reasoning and (I would add) absurd reasoning are mathematics", I would like to emphasis that the Manual Streams Simulator doesn't have anything to do with science; it's engineering, or I should actually say reverse engineering.



Read more...