Oracle10gR2 Streams(四)全库复制之二

STRMADM@ning>create directory dir_source as 'f:\oracle\source';

Directory created.

STRMADM@dest>create directory dir_dest as 'f:\oracle\dest';

Directory created.

alter system set log_archive_dest_2='service=dest arch noregister';

STRMADM@dest>begin
  2  DBMS_STREAMS_ADM.MAINTAIN_GLOBAL(
  3     source_directory_object      => 'DIR_SOURCE',
  4     destination_directory_object => 'DIR_DEST',
  5     source_database              => 'ning.test',
  6     destination_database         => 'dest.test',
  7     perform_actions              => true,
  8     dump_file_name               => 'streams_rep.dmp',
  9     bi_directional               => false,
 10     include_ddl                  => true,
 11     instantiation                => DBMS_STREAMS_ADM.INSTANTIATION_FULL);
 12  end;
 13  /

PL/SQL procedure successfully completed.

STRMADM@ning>create table ning.teststrm(i int);

Table created.

STRMADM@ning>insert into ning.teststrm values(1);

1 row created.

STRMADM@ning>commit;

Commit complete.

STRMADM@ning>alter system archive log current;

System altered.

STRMADM@dest>select * from ning.teststrm;

         I
----------
         1

STRMADM@dest>begin
  2  DBMS_STREAMS_ADM.MAINTAIN_GLOBAL(
  3     source_directory_object      => 'DIR_SOURCE',
  4     destination_directory_object => 'DIR_DEST',
  5     source_database              => 'ning.test',
  6     destination_database         => 'dest.test',
  7     perform_actions              => false,
  8     script_directory_object      =>'DIR_SCRIPT',
  9     script_name                  => 'global_script.sql',
 10     dump_file_name               => 'streams_rep.dmp',
 11     bi_directional               => false,
 12     include_ddl                  => true,
 13     instantiation                => DBMS_STREAMS_ADM.INSTANTIATION_NONE);
 14  end;
 15  /

PL/SQL procedure successfully completed.

SET ECHO ON
SET VERIFY OFF
WHENEVER SQLERROR EXIT SQL.SQLCODE;

-------------------------------------------------------------------
-- get TNSNAME and streams admin user details for both the databases
--------------------------------------------------------------------
PROMPT
PROMPT 'Enter TNS Name of site 1 as parameter 1:'
DEFINE db1                 = &1
PROMPT
PROMPT 'Enter streams admin username for site 1 as parameter 2:'
DEFINE strm_adm_db1        = &2
PROMPT
PROMPT 'Enter streams admin password for site 1 as parameter 3:'
DEFINE strm_adm_pwd_db1    = &3
PROMPT
PROMPT 'Enter TNS Name of site 2 as parameter 4:'
DEFINE db2                 = &4
PROMPT
PROMPT 'Enter streams admin username for site 2 as parameter 5:'
DEFINE strm_adm_db2        = &5
PROMPT
PROMPT 'Enter streams admin password for site 2 as parameter 6:'
DEFINE strm_adm_pwd_db2    = &6
PROMPT
PROMPT 'Enter TNS Name of site 3 as parameter 7:'
DEFINE db3                 = &7
PROMPT
PROMPT 'Enter streams admin username for site 3 as parameter 8:'
DEFINE strm_adm_db3        = &8
PROMPT
PROMPT 'Enter streams admin password for site 3 as parameter 9:'
DEFINE strm_adm_pwd_db3    = &9

-- connect as streams administrator to site 1
PROMPT Connecting as streams administrator to site 1
CONNECT &strm_adm_db1/&strm_adm_pwd_db1@&db1
--
-- Add database level supplemental logging
--
BEGIN
  EXECUTE IMMEDIATE 'ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY, UNIQUE INDEX, FOREIGN KEY) COLUMNS';
END;
/
--
-- Dictionary build
--
BEGIN
  dbms_capture_adm.build;
END;
/
--
-- Prepare global instantiation
--
BEGIN
  dbms_capture_adm.prepare_global_instantiation;
END;
/
-- connect as streams administrator to site 2
PROMPT Connecting as streams administrator to site 2
CONNECT &strm_adm_db2/&strm_adm_pwd_db2@&db2
--
-- Set up queue "STRMADM"."NING$CAPQ"
--
BEGIN
  dbms_streams_adm.set_up_queue(
    queue_table => '"STRMADM"."NING$CAPQT"',
    storage_clause => NULL,
    queue_name => '"STRMADM"."NING$CAPQ"',
    queue_user => '');
END;
/
--
-- PROPAGATE changes for all objects in the database
--
DECLARE
  version_num            NUMBER := 0;
  release_num            NUMBER := 0;
  pos                    NUMBER;
  initpos                NUMBER;
  q2q                    BOOLEAN;
  stmt                   VARCHAR2(100);
  ver                    VARCHAR2(30);
  compat                 VARCHAR2(30); 

BEGIN
  BEGIN
    stmt := 'BEGIN dbms_utility.db_version@DEST.TEST(:ver, :compat); END;';
    EXECUTE IMMEDIATE stmt USING OUT ver, OUT compat;
    -- Extract version number
    initpos := 1;
    pos := INSTR(compat, '.', initpos, 1);
    IF pos > 0 THEN
      version_num := TO_NUMBER(SUBSTR(compat, initpos, pos - initpos));
      initpos := pos + 1;

      -- Extract release number
      pos := INSTR(compat, '.', initpos, 1);
      IF pos > 0 THEN
        release_num := TO_NUMBER(SUBSTR(compat, initpos,
                                   pos - initpos));
        initpos := pos + 1;
      ELSE
        release_num := TO_NUMBER(SUBSTR(compat, initpos));
      END IF;
    ELSE
      version_num := TO_NUMBER(SUBSTR(compat, initpos));
    END IF;

    -- use q2q propagation if compatibility >= 10.2
    IF version_num > 10 OR
       (version_num = 10 AND release_num >=2) THEN
      q2q := TRUE;
    ELSE
      q2q := FALSE;
    END IF;

  EXCEPTION WHEN OTHERS THEN
    q2q := FALSE;
  END;


  dbms_streams_adm.add_global_propagation_rules(
    streams_name => '',
    source_queue_name => '"STRMADM"."NING$CAPQ"',
    destination_queue_name => '"STRMADM"."NING$APPQ"',
    include_dml => TRUE,
    include_ddl => TRUE,
    include_tagged_lcr => TRUE,
    source_database => 'NING.TEST',
    inclusion_rule => TRUE,
    and_condition => NULL,
    queue_to_queue => q2q);
END;
/
--
-- Disable propagation. Enable after destination has been setup
--
DECLARE
  q2q       VARCHAR2(10);
  destn_q   VARCHAR2(65);
BEGIN
  SELECT queue_to_queue INTO q2q
  FROM dba_propagation
  WHERE source_queue_owner = 'STRMADM' AND
        source_queue_name = 'NING$CAPQ' AND
        destination_queue_owner = 'STRMADM' AND
        destination_queue_name = 'NING$APPQ' AND
        destination_dblink = 'AQ$_LOCAL';

  IF q2q = 'TRUE' THEN
    destn_q := '"STRMADM"."NING$APPQ"';
  ELSE
    destn_q := NULL;
  END IF;

  dbms_aqadm.disable_propagation_schedule(
    queue_name => '"STRMADM"."NING$CAPQ"',
    destination => '',
    destination_queue => destn_q);
EXCEPTION WHEN OTHERS THEN
  IF sqlcode = -24065 THEN NULL;  -- propagation already disabled
  ELSE RAISE;
  END IF;
END;
/
--
-- Create downstream capture NING$CAP
--
DECLARE
  cnt       NUMBER;
  first_scn NUMBER;
BEGIN
  SELECT COUNT(*) INTO cnt
  FROM dba_capture
  WHERE capture_name = 'NING$CAP';

  -- Create capture only if the process does not exist.
  IF cnt = 0 THEN
    SELECT MAX(first_change#) INTO first_scn
    FROM v$archived_log@NING.TEST
    WHERE dictionary_begin = 'YES' AND
          first_change# <= (SELECT MAX(first_change#)
                            FROM v$archived_log@NING.TEST
                            WHERE dictionary_end  = 'YES');

    dbms_capture_adm.create_capture(
      queue_name=> '"STRMADM"."NING$CAPQ"',
      capture_name=> '"NING$CAP"',
      use_database_link => FALSE,
      first_scn => first_scn,
      source_database=> 'NING.TEST');
  END IF;
END;
/
--
-- CAPTURE changes for all objects in the database
--
DECLARE
  compat         VARCHAR2(512);
  initpos        NUMBER;
  pos            NUMBER;
  version_num    NUMBER;
  release_num    NUMBER;
  compat_func    VARCHAR2(65);
  get_compatible VARCHAR2(4000);
BEGIN
  SELECT value INTO compat
  FROM v$parameter
  WHERE name = 'compatible';

  -- Extract version number
  initpos := 1;
  pos := INSTR(compat, '.', initpos, 1);
  IF pos > 0 THEN
    version_num := TO_NUMBER(SUBSTR(compat, initpos, pos - initpos));
    initpos := pos + 1;

    -- Extract release number
    pos := INSTR(compat, '.', initpos, 1);
    IF pos > 0 THEN
      release_num := TO_NUMBER(SUBSTR(compat, initpos, pos - initpos));
      initpos := pos + 1;
    ELSE
      release_num := TO_NUMBER(SUBSTR(compat, initpos));
    END IF;
  END IF;

  IF version_num < 10 THEN
    compat_func := 'dbms_streams.compatible_9_2';
  ELSIF version_num = 10 THEN
    IF release_num < 2 THEN
      compat_func := 'dbms_streams.compatible_10_1';
    ELSE
      compat_func := 'dbms_streams.compatible_10_2';
    END IF;
  ELSE
    compat_func := 'dbms_streams.compatible_10_2';
  END IF;

  get_compatible := ':lcr.get_compatible() <= '||compat_func;


  dbms_streams_adm.add_global_rules(
    streams_type => 'CAPTURE',
    streams_name => '"NING$CAP"',
    queue_name => '"STRMADM"."NING$CAPQ"',
    include_dml => TRUE,
    include_ddl => TRUE,
    include_tagged_lcr => TRUE,
    source_database => 'NING.TEST',
    inclusion_rule => TRUE,
    and_condition => get_compatible);
END;
/
--
-- EXPORT: If export/import is the chosen instantiation method
-- do the export here.
-- Once export is complete, tranfer the export dump file(s) to the
-- appropriate location on the system hosting the destination
-- database.
--

--
-- Start capture process NING$CAP
--
BEGIN
  dbms_capture_adm.start_capture(
    capture_name => '"NING$CAP"');
EXCEPTION WHEN OTHERS THEN
  IF sqlcode = -26666 THEN NULL;  -- CAPTURE process already running
  ELSE RAISE;
  END IF;
END;
/
--
-- IMPORT: If export/import or datapump network mode import
-- is the chosen instantiation method, do the import here.
--

--
-- RMAN: If RMAN is the chosen instantiation method, do the
-- instantiation here.  Once the instantiation is complete,
-- do the necessary setup on the instantiated database, which
-- includes renaming global name and setting up database links.
-- Also, choose an appropriate instantiation scn at the source
-- and set it as the global instantiation scn at the destination
--

--
-- Set up queue "STRMADM"."NING$APPQ"
--
BEGIN
  dbms_streams_adm.set_up_queue(
    queue_table => '"STRMADM"."NING$APPQT"',
    storage_clause => NULL,
    queue_name => '"STRMADM"."NING$APPQ"',
    queue_user => '');
END;
/
--
-- APPLY changes for all objects in the database
--
DECLARE
  compat         VARCHAR2(512);
  initpos        NUMBER;
  pos            NUMBER;
  version_num    NUMBER;
  release_num    NUMBER;
  compat_func    VARCHAR2(65);
  get_compatible VARCHAR2(4000);
BEGIN
  SELECT value INTO compat
  FROM v$parameter
  WHERE name = 'compatible';

  -- Extract version number
  initpos := 1;
  pos := INSTR(compat, '.', initpos, 1);
  IF pos > 0 THEN
    version_num := TO_NUMBER(SUBSTR(compat, initpos, pos - initpos));
    initpos := pos + 1;

    -- Extract release number
    pos := INSTR(compat, '.', initpos, 1);
    IF pos > 0 THEN
      release_num := TO_NUMBER(SUBSTR(compat, initpos, pos - initpos));
      initpos := pos + 1;
    ELSE
      release_num := TO_NUMBER(SUBSTR(compat, initpos));
    END IF;
  END IF;

  IF version_num < 10 THEN
    compat_func := 'dbms_streams.compatible_9_2';
  ELSIF version_num = 10 THEN
    IF release_num < 2 THEN
      compat_func := 'dbms_streams.compatible_10_1';
    ELSE
      compat_func := 'dbms_streams.compatible_10_2';
    END IF;
  ELSE
    compat_func := 'dbms_streams.compatible_10_2';
  END IF;

  get_compatible := ':lcr.get_compatible() <= '||compat_func;


  dbms_streams_adm.add_global_rules(
    streams_type => 'APPLY',
    streams_name => '',
    queue_name => '"STRMADM"."NING$APPQ"',
    include_dml => TRUE,
    include_ddl => TRUE,
    include_tagged_lcr => TRUE,
    source_database => 'NING.TEST',
    inclusion_rule => TRUE,
    and_condition => get_compatible);
END;
/
--
-- Get tag value to be used for Apply
--
DECLARE
  found            BINARY_INTEGER := 0;
  tag_num          NUMBER;
  apply_nm         VARCHAR2(30);
  apply_nm_dqt     VARCHAR2(32);
BEGIN
  SELECT apply_name INTO apply_nm
  FROM dba_apply_progress
  WHERE source_database = 'NING.TEST';

  apply_nm_dqt := '"' || apply_nm || '"';
  -- Use the apply object id as the tag
  SELECT o.object_id INTO tag_num
  FROM dba_objects o
  WHERE o.object_name= apply_nm AND
        o.object_type='APPLY';
  LOOP
    BEGIN
      found := 0;
      SELECT 1 INTO found FROM dba_apply
      WHERE apply_name != apply_nm AND
            apply_tag = hextoraw(tag_num);
    EXCEPTION WHEN no_data_found THEN
      EXIT;
    END;
    EXIT WHEN (found = 0);
    tag_num := tag_num + 1;
  END LOOP;
  -- alter apply
  dbms_apply_adm.alter_apply(
    apply_name => apply_nm_dqt,
    apply_tag => hextoraw(tag_num));
END;
/
--
-- Start apply process applying changes from NING.TEST
--
DECLARE
  apply_nm VARCHAR2(32);
  apply_nm_dqt VARCHAR2(32);
BEGIN
  SELECT apply_name INTO apply_nm
  FROM dba_apply_progress
  WHERE source_database = 'NING.TEST';

  apply_nm_dqt := '"' || apply_nm || '"';
  dbms_apply_adm.start_apply(
    apply_name => apply_nm_dqt);
EXCEPTION WHEN OTHERS THEN
  IF sqlcode = -26666 THEN NULL;  -- APPLY process already running
  ELSE RAISE;
  END IF;
END;
/
--
-- Enable propagation schedule for "STRMADM"."NING$CAPQ"
-- to DEST.TEST
--
DECLARE
  q2q       VARCHAR2(10);
  destn_q   VARCHAR2(65);
BEGIN
  SELECT queue_to_queue INTO q2q
  FROM dba_propagation
  WHERE source_queue_owner = 'STRMADM' AND
        source_queue_name = 'NING$CAPQ' AND
        destination_queue_owner = 'STRMADM' AND
        destination_queue_name = 'NING$APPQ' AND
        destination_dblink = 'AQ$_LOCAL';

  IF q2q = 'TRUE' THEN
    destn_q := '"STRMADM"."NING$APPQ"';
  ELSE
    destn_q := NULL;
  END IF;

  dbms_aqadm.enable_propagation_schedule(
    queue_name => '"STRMADM"."NING$CAPQ"',
    destination => '',
    destination_queue => destn_q);
EXCEPTION WHEN OTHERS THEN
  IF sqlcode = -24064 THEN NULL; -- propagation already enabled
  ELSE RAISE;
  END IF;
END;
/