rename_schema procedure of DBMS_STREAMS_ADM in the documentation, you'll find the following note:Declarative transformations can transform row LCRs only. These row LCRs can be captured by a capture process, captured by a synchronous capture, or constructed and enqueued by an application. Therefore, a DML rule must be specified when you run this procedure. If a DDL is specified, then the procedure raises an error.And actually, if you try to apply that rule-based transformation to DDL LCR rule, you get the following error:
ORA-26746: DDL rule "STRMADMIN"."SOURCE30" not allowed for this operationThis post provides an example of a custom rule-based transformation for that renames a schema in the even of certain type of DDL statement; it is built without specifying any action context (see note 309575.1), even if obviously an action context is implicitly created by the
dbms_streams_adm.set_rule_transform_function procedure of oracle 10g. The basic idea is help you build an exemple of DDL replication in a single database in order to introduce some of the limits of DDL...This post contains in the following sections:
- Prepare the
SOURCEandDESTINATIONschemes - Configure supplemental logging and the LogMiner data dictionary
- Create the Streams administrator and a Streams queue
- Create the Streams capture
- Create the apply process and set the schema instantiation SCN
- Associate the rule-based transformations with the DML and DDL apply rule
- Start the capture and apply processes
- Test the replication
- Suppress the Oracle Streams Configuration
Prepare the SOURCE and DESTINATION schemes
In this example, we'll capture DDL and DML changes made to the SOURCE schema and apply them to the DESTINATION schema; We'll start by creating those 2 empty schemas:connect / as sysdba
create user source
identified by source
default tablespace users
temporary tablespace temp;
grant connect,resource to source;
create user destination
identified by destination
default tablespace users
temporary tablespace temp;
grant connect,resource to destination;
Configure supplemental logging and the LogMiner data dictionary
In this example we will force supplemental logging for all the primary and uniques keys of the database. If we don't do it, we would have to add supplemental logging to the tables after we create each one of them in theSOURCE schema:connect / as sysdbaWe'll also store the dictionary in the redo logs for LogMiner:
alter database add supplemental log
data (primary key, unique index) columns;
select SUPPLEMENTAL_LOG_DATA_MIN,
SUPPLEMENTAL_LOG_DATA_PK,
SUPPLEMENTAL_LOG_DATA_UI
from gv$database;
SUPPLEME SUP SUP
-------- --- ---
YES YES YES
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;
/
First SCN Value = 2480147
Note:
In this case, theSOURCEschema is empty and we don't need to execute theprepare_schema_instantiationprocedure ofdbms_capture_adm. That's a very unlikely scenario and if have to instantiate existing objects, don't forget to execute this procedure.
Create the Streams administrator and a Streams queue
Create a Streams administrator (STRMADMIN) and a queue to be use to stage the messages (streams_queue):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;
/
Create the Streams capture
Like in the first blog example about Streams, the next step consists in creating a capture process and associating a declarative rule-based transformation to the DML capture rule. Make sure you are connected asSTRMADMIN to perform these operations:connect strmadmin/strmadmin
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'source',
streams_type => 'capture',
streams_name => 'streams_capture',
queue_name => 'strmadmin.streams_queue',
include_dml => true,
include_ddl => true,
include_tagged_lcr => false,
source_database => 'BLACK',
inclusion_rule => true);
END;
/
set lines 120
col streams_name format a16
col streams_type format a9
col schema_name format a10
col rule_type format a8
col rule_name format a10
col rule_owner format a15
select STREAMS_NAME
, STREAMS_TYPE
, SCHEMA_NAME
, RULE_TYPE
, RULE_NAME
, RULE_OWNER
from DBA_STREAMS_SCHEMA_RULES;
STREAMS_NAME STREAMS_T SCHEMA_NAM RULE_TYP RULE_NAME RULE_OWNER
---------------- --------- ---------- -------- ---------- ----------
STREAMS_CAPTURE CAPTURE SOURCE DML SOURCE42 STRMADMIN
STREAMS_CAPTURE CAPTURE SOURCE DDL SOURCE43 STRMADMIN
Create the apply process and set the schema instantiation SCN
We now need to create an apply process, add rules so that it applies the captured LCR and define the schema instantiation SCN so that it knows at what SCN it should start its work. The script below perform these operations:connect strmadmin/strmadmin
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name => 'source',
streams_type => 'apply',
streams_name => 'streams_apply',
queue_name => 'strmadmin.streams_queue',
include_dml => true,
include_ddl => true,
include_tagged_lcr => false,
source_database => 'BLACK',
inclusion_rule => true);
END;
/
select streams_name,
streams_type,
schema_name,
rule_type,
rule_name,
rule_owner
from dba_streams_schema_rules
where streams_name='STREAMS_APPLY';
STREAMS_NAME STREAMS_TY SCHEMA_NAME RULE_ RULE_NAME RULE_OWNER
------------- ---------- ------------ ----- --------------- ------------
STREAMS_APPLY APPLY SOURCE DML SOURCE60 STRMADMIN
STREAMS_APPLY APPLY SOURCE DDL SOURCE61 STRMADMIN
col APPLY_SCN format 999999999999 new_value instantiation_scn
select dbms_flashback.get_system_change_number apply_scn
from dual;
APPLY_SCN
---------
2565304
select &&instantiation_scn apply_scn
from dual;
APPLY_SCN
---------
2565304
begin
dbms_apply_adm.set_schema_instantiation_scn(
source_schema_name => 'source',
source_database_name => 'BLACK',
instantiation_scn => &&instantiation_scn);
end;
/
col source_database format a10
col source_schema format a15
col instantiation_scn format 999999999999
select source_database, source_schema, instantiation_scn
from dba_apply_instantiated_schemas;
SOURCE_DAT SOURCE_SCHEMA INSTANTIATION_SCN
---------- --------------- -----------------
BLACK DESTINATION 2565304
Associate the rule-based transformations with the DML and DDL apply rule
Before we associate the transformations with the apply rules, we'll write the a function to transform the current_schema property of the DDL LCR. For the purpose of this example, I've added addl_logging table and a sequence ddl_logging_seq in the STRMADMIN schema so that we can store some logs about the transformed LCR. Below is the script that creates the function:connect strmadmin/strmadminOnce the function created, we can use it as a custom rule-based transformation for the DDL LCR. We'll add a declarative rule-based transformation to rename the schema of the DML LCR:
create sequence ddl_logging_seq;
create table ddl_logging(
ddl_logging_seq number,
text clob);
create or replace function source2destination(in_any in anydata)
return anydata
IS
lcr sys.lcr$_ddl_record;
rc number;
seq number;
ddl CLOB;
begin
-- Check if the object type is SYS.LCR$_DDL_RECORD
if in_any.GETTYPENAME='SYS.LCR$_DDL_RECORD' THEN
-- Put the row LCR into lcr
rc := in_any.GETOBJECT(lcr);
-- Log DDL and other properties in ddl_logging
seq:=ddl_logging_seq.nextval;
insert into ddl_logging(ddl_logging_seq, text)
values (seq, 'CURRENT SCHEMA : '||lcr.get_current_schema());
DBMS_LOB.CREATETEMPORARY(ddl, true);
lcr.GET_DDL_TEXT(ddl);
insert into ddl_logging(ddl_logging_seq, text)
values (seq, 'CAPTURED DDL : '||ddl);
dbms_lob.freetemporary(ddl);
commit;
-- Transform the LCR
lcr.set_current_schema('DESTINATION');
-- Return the transformed LCR
return anydata.convertobject(lcr);
end if;
end;
/
set lines 120
col streams_name format a16
col streams_type format a9
col schema_name format a10
col rule_type format a8
col rule_name format a10
col rule_owner format a15
select STREAMS_NAME
, STREAMS_TYPE
, SCHEMA_NAME
, RULE_TYPE
, RULE_NAME
, RULE_OWNER
from DBA_STREAMS_SCHEMA_RULES;
STREAMS_NAME STREAMS_T SCHEMA_NAM RULE_TYP RULE_NAME RULE_OWNER
---------------- --------- ---------- -------- ---------- ----------
STREAMS_CAPTURE CAPTURE SOURCE DML SOURCE42 STRMADMIN
STREAMS_CAPTURE CAPTURE SOURCE DDL SOURCE43 STRMADMIN
STREAMS_APPLY APPLY SOURCE DML SOURCE60 STRMADMIN
STREAMS_APPLY APPLY SOURCE DDL SOURCE61 STRMADMIN
accept rulename prompt "Enter the Rule Name: "
Enter the Rule Name: SOURCE61
begin
dbms_streams_adm.set_rule_transform_function(
rule_name => '&&rulename',
transform_function => 'source2destination');
end;
/
col rule_owner format a10
col rule_name format a10
col transform_function_name format a32
select rule_owner,rule_name,transform_function_name,custom_type
from dba_streams_transform_function;
RULE_OWNER RULE_NAME TRANSFORM_FUNCTION_NAME CUSTOM_TYPE
---------- ---------- -------------------------------- -----------
STRMADMIN SOURCE61 "STRMADMIN"."SOURCE2DESTINATION" ONE TO ONE
select RULE_ACTION_CONTEXT
from DBA_RULES
where RULE_NAME='&&rulename'
and rule_owner='STRMADMIN';
RULE_ACTION_CONTEXT(ACTX_LIST(NVN_NAME, NVN_VALUE()))
-----------------------------------------------------------------------------
RE$NV_LIST(RE$NV_ARRAY(RE$NV_NODE('STREAMS$_TRANSFORM_FUNCTION', ANYDATA())))
select STREAMS_NAME
, STREAMS_TYPE
, SCHEMA_NAME
, RULE_TYPE
, RULE_NAME
, RULE_OWNER
from DBA_STREAMS_SCHEMA_RULES;
STREAMS_NAME STREAMS_T SCHEMA_NAM RULE_TYP RULE_NAME RULE_OWNER
---------------- --------- ---------- -------- ---------- ----------
STREAMS_CAPTURE CAPTURE SOURCE DML SOURCE42 STRMADMIN
STREAMS_CAPTURE CAPTURE SOURCE DDL SOURCE43 STRMADMIN
STREAMS_APPLY APPLY SOURCE DML SOURCE60 STRMADMIN
STREAMS_APPLY APPLY SOURCE DDL SOURCE61 STRMADMIN
accept rulename prompt "Enter the Rule Name: "
Enter the Rule Name: SOURCE60
begin
dbms_streams_adm.rename_schema(
rule_name => '&&rulename' ,
from_schema_name => 'SOURCE',
to_schema_name => 'DESTINATION',
step_number => 0,
operation => 'add');
end;
/
Start the capture and apply processes
Start the capture and the apply processes:exec dbms_capture_adm.start_capture('streams_capture');
exec dbms_apply_adm.start_apply('streams_apply');
Test the replication
NoteCheck the replication is working as expected:
The DDL commands are captured as it and the name of the source schema must not be included in it otherwise the apply process will fail. To avoid this problem you can usealter session set current_schema=source;
connect / as sysdba
alter session set current_schema=source;
create table t1(id number,
text varchar2(100),
constraint t1_pk primary key(id));
col owner format a12
select owner, table_name
from dba_tables
where owner in ('SOURCE','DESTINATION');
OWNER TABLE_NAME
------------ ----------
SOURCE T1
DESTINATION T1
select owner, constraint_name
from dba_constraints
where owner in ('SOURCE','DESTINATION');
OWNER CONSTRAINT_NAME
------------ ---------------
SOURCE T1_PK
DESTINATION T1_PK
select owner, index_name
from dba_indexes
where owner in ('SOURCE','DESTINATION');
OWNER INDEX_NAME
------------ ----------
DESTINATION T1_PK
SOURCE T1_PK
insert into t1 values(1,'Text 1');
commit;
col id format 99
col text format a10
select id, text
from destination.t1;
ID TEXT
-- ------
1 Text 1
update t1 set text='Text 2'
where id=1;
commit;
ID TEXT
-- ------
1 Text 2
drop table t1;
select owner, table_name
from dba_tables
where owner in ('SOURCE','DESTINATION');
no rows selected
select owner,
object_name,
original_name
from dba_recyclebin
where owner in ('SOURCE','DESTINATION')
order by 2,3;
OWNER OBJECT_NAME ORIGINAL_NAME
------------ ------------------------------ -------------
SOURCE BIN$ZKFLC8wxhgngQAB/AQAw5g==$0 T1_PK
DESTINATION BIN$ZKFLC8wyhgngQAB/AQAw5g==$0 T1
SOURCE BIN$ZKFLC8wyhgngQAB/AQAw5g==$0 T1
DESTINATION BIN$ZKFLDgByejLgQAB/AQAwqA==$0 T1_PK
set long 1000
set longchuncksize 1000
col text format a100 wor wra
select text from strmadmin.ddl_logging;
TEXT
-----------------------------------------------------------------
CURRENT SCHEMA : SOURCE
BASE TABLE NAME: SOURCE
OBJECT OWNER: SOURCE
CAPTURED DDL : create table t1(id number,
text varchar2(100),
constraint t1_pk primary key(id))
CURRENT SCHEMA : SOURCE
BASE TABLE NAME: SOURCE
OBJECT OWNER: SOURCE
CAPTURED DDL : drop table t1 AS "BIN$ZKFLC8wyhgngQAB/AQAw5g==$0"
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:connect / as sysdba
exec dbms_capture_adm.stop_capture('streams_capture');
exec dbms_apply_adm.stop_apply('streams_apply');
exec dbms_capture_adm.drop_capture('streams_capture',true);
exec dbms_apply_adm.drop_apply('streams_apply',true);
begin
for i in (select source_schema name
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 => 'BLACK',
instantiation_scn => null);
end loop;
for i in (select source_object_owner||'.'||
source_object_name name
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 => 'BLACK',
instantiation_scn => null);
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 destination cascade;
drop user source cascade;
alter database drop supplemental log
data (primary key, unique index) columns;
select SUPPLEMENTAL_LOG_DATA_MIN,
SUPPLEMENTAL_LOG_DATA_PK,
SUPPLEMENTAL_LOG_DATA_UI
from gv$database;
SUPPLEME SUP SUP
-------- --- ---
YES NO NO
Posts
No comments:
Post a Comment