Apply的属性

ddl_handler

DDL操作触发，可以用来记录DDL历史

CREATE TABLE strmadmin.history_ddl_lcrs(

timestamp DATE,

source_database_name VARCHAR2(128),

command_type VARCHAR2(30),

object_owner VARCHAR2(32),

object_name VARCHAR2(32),

object_type VARCHAR2(18),

ddl_text CLOB,

logon_user VARCHAR2(32),

current_schema VARCHAR2(32),

base_table_owner VARCHAR2(32),

base_table_name VARCHAR2(32),

tag RAW(10),

transaction_id VARCHAR2(10),

scn NUMBER);

CREATE OR REPLACE PROCEDURE history_ddl(in_any IN ANYDATA)

IS

lcr SYS.LCR$_DDL_RECORD;

rc PLS_INTEGER;

ddl_text CLOB;

BEGIN

-- Access the LCR

rc := in_any.GETOBJECT(lcr);

DBMS_LOB.CREATETEMPORARY(ddl_text, TRUE);

lcr.GET_DDL_TEXT(ddl_text);

-- Insert DDL LCR information into history_ddl_lcrs table

INSERT INTO strmadmin.history_ddl_lcrs VALUES(

SYSDATE, lcr.GET_SOURCE_DATABASE_NAME(), lcr.GET_COMMAND_TYPE(),

lcr.GET_OBJECT_OWNER(), lcr.GET_OBJECT_NAME(), lcr.GET_OBJECT_TYPE(),

ddl_text, lcr.GET_LOGON_USER(), lcr.GET_CURRENT_SCHEMA(),

lcr.GET_BASE_TABLE_OWNER(), lcr.GET_BASE_TABLE_NAME(), lcr.GET_TAG(),

lcr.GET_TRANSACTION_ID(), lcr.GET_SCN());

-- Apply DDL LCR

lcr.EXECUTE();

-- Free temporary LOB space

DBMS_LOB.FREETEMPORARY(ddl_text);

END;

/

BEGIN

DBMS_APPLY_ADM.ALTER_APPLY(

apply_name => 'strep01_apply',

ddl_handler => 'strmadmin.history_ddl');

END;

/

precommit_handler

commit触发,配合DBMS_APPLY_ADM.SET_DML_HANDLER使用

DML Handler用来记录DML语句，其记录不了commit标签

Precommit handler用来记录Commit标签

例子参考

Configuring a Precommit Handler for the Streams Apply Process in Oracle 10.1 onwards [ID 252042.1]

message_handler

处理persistent user messages，自定义apply过程。

一般做Streams不需要用这个handler

参考

Oracle® Streams Advanced Queuing User's Guide 11g Release 2 (11.2)

-- Oracle Streams Messaging Examples

-- Configuring an Apply Process

http://docs.oracle.com/cd/E11882_01/server.112/e11013/aq_demo.htm#ADQUE3487

DBMS_APPLY_ADM

ADD_STMT_HANDLER

DML lcr触发，基于一个表的一个DML操作触发

将指定的handle/stmt添加给apply。handle通过DBMS_STREAMS_HANDLER_ADM来定义

提取LCR记录，使用SQL语句加工，插入的本来的表。表结构变化较大可用

DECLARE

stmt CLOB;

BEGIN

stmt := 'INSERT INTO oe.orders(order_id, order_date, order_mode, customer_id, order_status, order_total, sales_rep_id, promotion_id)

VALUES(:new.order_id,:new.order_date,:new.order_mode,:new.customer_id,DECODE(:new.order_status, 1, 2, :new.order_status), :new.order_total,:new.sales_rep_id,:new.promotion_id)';

DBMS_APPLY_ADM.ADD_STMT_HANDLER(

object_name => 'oe.orders',

operation_name => 'INSERT',

handler_name => 'modify_orders',

statement => stmt,

apply_name => 'apply$_sta_2',

comment => 'Modifies inserts into the orders table'

);

END;

/

创建一个handler，名位track_jobs。

该handler会执行两组操作

1，原语句

2，审计语句

ALTER TABLE hr.jobs ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

CREATE SEQUENCE hr.track_jobs_seq START WITH 1 INCREMENT BY 1;

CREATE TABLE hr.track_jobs(

change_id NUMBER CONSTRAINT track_jobs_pk PRIMARY KEY,

job_id VARCHAR2(10),

job_title VARCHAR2(35),

min_salary_old NUMBER(6),

min_salary_new NUMBER(6),

max_salary_old NUMBER(6),

max_salary_new NUMBER(6),

timestamp TIMESTAMP);

BEGIN

DBMS_STREAMS_HANDLER_ADM.CREATE_STMT_HANDLER(

handler_name => 'track_jobs',

comment => 'Tracks updates to the jobs table');

END;

/

DECLARE

stmt CLOB;

BEGIN

stmt := ':lcr.execute TRUE';

DBMS_STREAMS_HANDLER_ADM.ADD_STMT_TO_HANDLER(

handler_name => 'track_jobs',

statement => stmt,

execution_sequence => 10

);

END;

/

DECLARE

stmt CLOB;

BEGIN

stmt := 'INSERT INTO hr.track_jobs(change_id, job_id, job_title, min_salary_old, min_salary_new, max_salary_old, max_salary_new, timestamp)

VALUES( hr.track_jobs_seq.NEXTVAL, :new.job_id, :new.job_title, :old.min_salary, :new.min_salary, :old.max_salary, :new.max_salary, :source_time)';

DBMS_STREAMS_HANDLER_ADM.ADD_STMT_TO_HANDLER(

handler_name => 'track_jobs',

statement => stmt,

execution_sequence => 20);

END;

/

BEGIN

DBMS_APPLY_ADM.ADD_STMT_HANDLER(

object_name => 'hr.jobs',

operation_name => 'UPDATE',

handler_name => 'track_jobs',

apply_name => 'apply$_sta_2');

END;

/

SET_DML_HANDLER

将一个存过用于处理，基于一个表的一个DML操作触发，两种触发条件

1. 每条DML lcr触发

2. 当DML报错时触发

通过参数error_handler => true 区分

将语句审计到文本

CREATE DIRECTORY SQL_GEN_DIR AS '/usr/sql_gen';

CREATE OR REPLACE PROCEDURE strmadmin.sql_gen_dep(lcr_anydata IN SYS.ANYDATA) IS

lcr SYS.LCR$_ROW_RECORD;

int PLS_INTEGER;

row_txt_clob CLOB;

fp UTL_FILE.FILE_TYPE;

BEGIN

int := lcr_anydata.GETOBJECT(lcr);

DBMS_LOB.CREATETEMPORARY(row_txt_clob, TRUE);

-- Generate SQL from row LCR and save to file

lcr.GET_ROW_TEXT(row_txt_clob);

fp := UTL_FILE.FOPEN (

location => 'SQL_GEN_DIR',

filename => 'sql_gen_file.txt',

open_mode => 'a',

max_linesize => 5000);

UTL_FILE.PUT_LINE(

file => fp,

buffer => row_txt_clob,

autoflush => TRUE);

DBMS_LOB.TRIM(row_txt_clob, 0);

UTL_FILE.FCLOSE(fp);

-- Apply row LCR

lcr.EXECUTE(TRUE);

END;

/

BEGIN

DBMS_APPLY_ADM.SET_DML_HANDLER(

object_name => 'hr.departments',

object_type => 'TABLE',

operation_name => 'UPDATE',

error_handler => FALSE,

user_procedure => 'strmadmin.sql_gen_dep',

apply_database_link => NULL,

apply_name => NULL);

END;

/

添加一个error handler

CREATE TABLE strmadmin.errorlog(

logdate DATE,

apply_name VARCHAR2(30),

sender VARCHAR2(100),

object_name VARCHAR2(32),

command_type VARCHAR2(30),

errnum NUMBER,

errmsg VARCHAR2(2000),

text VARCHAR2(2000),

lcr SYS.LCR$_ROW_RECORD);

CREATE OR REPLACE PACKAGE errors_pkg

AS

TYPE emsg_array IS TABLE OF VARCHAR2(2000) INDEX BY BINARY_INTEGER;

PROCEDURE regions_pk_error(

message IN ANYDATA,

error_stack_depth IN NUMBER,

error_numbers IN DBMS_UTILITY.NUMBER_ARRAY,

error_messages IN EMSG_ARRAY);

END errors_pkg ;

/

CREATE OR REPLACE PACKAGE BODY errors_pkg AS

PROCEDURE regions_pk_error (

message IN ANYDATA,

error_stack_depth IN NUMBER,

error_numbers IN DBMS_UTILITY.NUMBER_ARRAY,

error_messages IN EMSG_ARRAY )

IS

reg_id NUMBER;

ad ANYDATA;

lcr SYS.LCR$_ROW_RECORD;

ret PLS_INTEGER;

vc VARCHAR2(30);

apply_name VARCHAR2(30);

errlog_rec errorlog%ROWTYPE ;

ov2 SYS.LCR$_ROW_LIST;

BEGIN

-- Access the error number from the top of the stack.

-- In case of check constraint violation,

-- get the name of the constraint violated.

IF error_numbers(1) IN ( 1 , 2290 ) THEN

ad := DBMS_STREAMS.GET_INFORMATION('CONSTRAINT_NAME');

ret := ad.GetVarchar2(errlog_rec.text);

ELSE

errlog_rec.text := NULL ;

END IF ;

-- Get the name of the sender and the name of the apply process.

ad := DBMS_STREAMS.GET_INFORMATION('SENDER');

ret := ad.GETVARCHAR2(errlog_rec.sender);

apply_name := DBMS_STREAMS.GET_STREAMS_NAME();

-- Try to access the LCR.

ret := message.GETOBJECT(lcr);

errlog_rec.object_name := lcr.GET_OBJECT_NAME() ;

errlog_rec.command_type := lcr.GET_COMMAND_TYPE() ;

errlog_rec.errnum := error_numbers(1) ;

errlog_rec.errmsg := error_messages(1) ;

INSERT INTO strmadmin.errorlog VALUES (SYSDATE, apply_name,

errlog_rec.sender, errlog_rec.object_name, errlog_rec.command_type,

errlog_rec.errnum, errlog_rec.errmsg, errlog_rec.text, lcr);

-- Add the logic to change the contents of LCR with correct values.

-- In this example, get a new region_id number

-- from the hr.reg_exception_s sequence.

ov2 := lcr.GET_VALUES('new', 'n');

FOR i IN 1 .. ov2.count

LOOP

IF ov2(i).column_name = 'REGION_ID' THEN

SELECT hr.reg_exception_s.NEXTVAL INTO reg_id FROM DUAL;

ov2(i).data := ANYDATA.ConvertNumber(reg_id) ;

END IF ;

END LOOP ;

-- Set the NEW values in the LCR.

lcr.SET_VALUES(value_type => 'NEW', value_list => ov2);

-- Execute the modified LCR to apply it.

lcr.EXECUTE(TRUE);

END regions_pk_error;

END errors_pkg;

/

BEGIN

DBMS_APPLY_ADM.SET_DML_HANDLER(

object_name => 'hr.regions',

object_type => 'TABLE',

operation_name => 'INSERT',

error_handler => TRUE,

user_procedure => 'strmadmin.errors_pkg.regions_pk_error',

apply_database_link => NULL,

apply_name => NULL);

END;

/

SET_CHANGE_HANDLER

change handler是一种特殊的DML Handler

记录表的修改到变化表,使用DBMS_STREAMS_ADM.MAINTAIN_CHANGE_TABLE 配置更简单

该过程只是一个设置过程，change handler记录在 DBA_APPLY_CHANGE_HANDLERS，查询出来可以直接设置

Oracle® Streams Concepts and Administration 11g Release 2 (11.2)

-- 20 Using Oracle Streams to Record Table Changes

http://docs.oracle.com/cd/E11882_01/server.112/e17069/strms_change_table.htm#STRMS1520

SET_UPDATE_CONFLICT_HANDLER

UPDATE发生问题触发，如何处理问题

DECLARE

cols DBMS_UTILITY.NAME_ARRAY;

BEGIN

cols(1) := 'dcxm_mc';

cols(2) := 'swws_dm';

cols(3) := 'ywhj_dm';

cols(4) := 'xybz';

cols(5) := 'yxbz';

DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(object_name => 'ctais2.dm_dcxm',

method_name => 'OVERWRITE',

resolution_column => 'swws_dm',

column_list => cols);

END;

/

method_name -> MAXIMUM / MINIMUM / OVERWRITE / DISCARD

SET_RULE_TRANSFORM_FUNCTION

自定义的转换，弥补像RENAME_SCHEMA这样的不足

How to Setup Custom Rule Based Transformation [ID 783203.1]

ADD_COLUMN & DELETE_COLUMN & KEEP_COLUMNS & RENAME_COLUMNS & RENAME_TABLE & RENAME_SCHEMA

How to ADD/REMOVE a column from a Streams Replicated Table using Declarative Rule Transformations [ID 781625.1]

Setup Streams Replication Between Different Source and Target Schemas with Different Table Structures [ID 784899.1]

ITPub博客

STREAMS笔记(7) rule - handle & TRANSFORM

Apply的属性

ddl_handler

precommit_handler

message_handler

DBMS_APPLY_ADM

ADD_STMT_HANDLER

SET_DML_HANDLER

SET_CHANGE_HANDLER

SET_UPDATE_CONFLICT_HANDLER

SET_RULE_TRANSFORM_FUNCTION

ADD_COLUMN & DELETE_COLUMN & KEEP_COLUMNS & RENAME_COLUMNS & RENAME_TABLE & RENAME_SCHEMA

DBMS_STREAMS_HANDLER_ADM

CREATE_STMT_HANDLER

ADD_STMT_TO_HANDLER

最新文章