IT 研究室 ( 前IT DBA's 資訊站): More about Oracle Streams on 10g+

以往Oracle 9i 所知道的Oracle streams, 都必須要在本機source作分析redo logs, 再傳遞資料流到異機去

而Oracle 10g R2 有很大的進展, 可以把捕捉與異動都放在Target端去做, 稱之為Staging database(Stage Site)

Ref:

http://www.udpwork.com/item/6559.html

CDC的全稱是Change Data Capture，主要用在資料倉庫中，對原資料庫的資料進行抽取、傳輸到資料倉庫中，用於進行分析和統計。CDC有同步模式和非同步模式：

1. CDC同步模式：
通過trigger來實現。

2. CDC非同步模式：
2.1 非同步HotLog模式
2.2 非同步分散式HotLog模式
2.3 非同步AutoLog模式。
2.3.1 非同步Autolog模式-online redo log
2.3.2 非同步Autolog模式-archive log

今天我們來配置的是非同步AutoLog模式-online redo log(澳洲電訊的某應用的資料庫就是用的該架構)，該模式的好處是日誌傳輸到備機（stage site）後，後續的publish，subscribe，capture操作都是在備機（stage site），不是在生產資料庫（source site），因此能大大減輕生產資料庫的壓力。

上圖為online redo的傳輸，另外還有archive log的傳輸：

與online redo傳輸不同的是：前者RFS進程傳輸的是redo，到standby redo log，downstream capture進程處理的是standby redo log；後者RFS進程傳輸的是archive log，downstream capture進程處理的是archive log。

注意：Autolog CDC需要與downstream配合使用。
CDC的capture有兩種：local capture和downstream capture。
可以通過

SQL> SELECT CAPTURE_NAME,status, CAPTURE_TYPE from dba_capture;

CAPTURE_NAME STATUS CAPTURE_TY
------------------------------ -------- ----------
CDC$C_CS_USM ENABLED LOCAL
STRM_CAPTURE ENABLED DOWNSTREAM

SQL>

來查看。

capture進程負責創建LCRs（logical change record），CDC能"看到"DDL LCRs，但是不做處理，CDC只處理DML的LCRs。

下面我們就開始來配置CDC。

CDC的生產資料庫，我們成為Source site，目的機器，用於做資料倉庫的機器，我們稱作Stage site。
做CDC的要求2邊的資料庫版本都是10g以上，且stage site的資料庫版本必須等於或者大於source site的資料庫版本。

1. 配置source site：
1.1 配置初始化參數，以sysdba登錄：

1.2 重啟資料庫，檢查

SQL> shutdown immediate;
Database closed.
Database dismounted.
ORACLE instance shut down.
SQL>
SQL> startup mount;
ORACLE instance started.

Total System Global Area 448790528 bytes
Fixed Size 1261332 bytes
Variable Size 339738860 bytes
Database Buffers 100663296 bytes
Redo Buffers 7127040 bytes
Database mounted.
SQL>
SQL> alter database archivelog;

Database altered.

SQL> alter database force logging;

Database altered.

SQL> alter database add supplemental log data;


Database altered.

SQL> SQL>
SQL> alter database open;

Database altered.

SQL> archive log list;
Database log mode Archive Mode
Automatic archival Enabled
Archive destination /oracle/app/oracle/arch/cdc01
Oldest online log sequence 1
Next log sequence to archive 3
Current log sequence 3
SQL>
SQL> alter system switch logfile;

System altered.

SQL> archive log list;
Database log mode Archive Mode
Automatic archival Enabled
Archive destination /oracle/app/oracle/arch/cdc01
Oldest online log sequence 2
Next log sequence to archive 3
Current log sequence 4
SQL>
SQL>
SQL>
SQL>
SQL> col log_min format a7
SQL> col log_pk format a6
SQL> col log_pk format a6
SQL> col log_ui format a6
SQL> col log_fk format a6
SQL> col log_all format a7
SQL> col force_log format a9
SQL>
SQL> SELECT supplemental_log_data_min LOG_MIN, supplemental_log_data_pk LOG_PK, supplemental_log_data_ui LOG_UI, supplemental_log_data_fk LOG_FK,
2 supplemental_log_data_all LOG_ALL, force_logging FORCE_LOG
3 FROM v$database;

LOG_MIN LOG_PK LOG_UI LOG_FK LOG_ALL FORCE_LOG
------- ------ ------ ------ ------- ---------
YES NO NO NO NO YES

SQL>
SQL>
SQL>
SQL> SELECT tablespace_name, force_logging
2 FROM dba_tablespaces;

TABLESPACE_NAME FORCE_
------------------------------------------------------------ ------
SYSTEM NO
UNDOTBS1 NO
SYSAUX NO
TEMP NO
USERS NO

SQL> SELECT table_name
2 FROM dba_tables
3 WHERE owner = 'SYS'
4 AND table_name LIKE 'CDC%$';

TABLE_NAME
------------------------------------------------------------
CDC_SYSTEM$
CDC_SUBSCRIBERS$
CDC_SUBSCRIBED_TABLES$
CDC_SUBSCRIBED_COLUMNS$
CDC_PROPAGATIONS$
CDC_PROPAGATED_SETS$
CDC_CHANGE_TABLES$
CDC_CHANGE_SOURCES$
CDC_CHANGE_SETS$
CDC_CHANGE_COLUMNS$

10 rows selected.

SQL> desc cdc_system$
Name Null? Type
----------------------------------------- -------- ----------------------------
MAJOR_VERSION NOT NULL NUMBER
MINOR_VERSION NOT NULL NUMBER

SQL>
SQL> SELECT * FROM cdc_system$;

MAJOR_VERSION MINOR_VERSION
------------- -------------
   1 0

SQL> SELECT * FROM global_name;

GLOBAL_NAME
--------------------------------------------------------------------------------
CDC01.REGRESS.RDBMS.DEV.US.ORACLE.COM

SQL>
SQL>
SQL> desc v$log
Name Null? Type
----------------------------------------- -------- ----------------------------
GROUP# NUMBER
THREAD# NUMBER
SEQUENCE# NUMBER
BYTES NUMBER
MEMBERS NUMBER
ARCHIVED VARCHAR2(3)
STATUS VARCHAR2(16)
FIRST_CHANGE# NUMBER
FIRST_TIME DATE

SQL>
SQL> SELECT group#, bytes
2 FROM v$log;

   GROUP# BYTES
---------- ----------
   1 52428800
   2 52428800
   3 52428800

SQL>

2. 配置stage site.
2.1 配置初始化參數，用sysdba登錄：

SQL> alter system set global_names=TRUE scope=BOTH;

System altered.

SQL> alter system set streams_pool_size=200M scope=BOTH;

System altered.

SQL> alter system set undo_retention=3600 scope=BOTH;

System altered.

SQL> alter system set log_archive_dest_1 ="location=/oracle/app/oracle/arch/cdc01 mandatory reopen=2";

System altered.

SQL>

SQL> alter system set log_archive_dest_1="location=/oracle/app/oracle/arch/cdc02 mandatory reopen=2 valid_for=(online_logfile,primary_role)";

System altered.

SQL>
SQL> alter system set log_archive_dest_2="location=/oracle/app/oracle/arch/cdc02_dest2 mandatory valid_for=(standby_logfile,primary_role)";

System altered.

SQL>

SQL> alter system set log_archive_dest_state_1 = enable;

System altered.

SQL> alter system set log_archive_dest_state_2 = enable;

System altered.

SQL> alter system set log_archive_format="cdc02_%R_%T_%S.arc" scope=spfile;

System altered.

SQL>

2.2 重啟資料庫，檢查。

3. 創建stream adminstrator（在stage site執行：），從上面文章開始處的第一個圖，我們看出，當log（無論是online還是archive）到stage site之後，由downstream capture進程進行挖掘。因此我們要在stage site先創建stream administrator。

downstream capture用來產生stage site的change source（注意，除了非同步HotLog模式，其他模式的change source都在stage site），stage change set，和stage change table。
注：在本例中，change source只存在stage site； change set在source site和stage site都有；change table在stage site，對應在source site叫source table。

SQL> CREATE TABLESPACE cdc_tbsp
2 datafile '/oracle/app/oracle/oradata/cdc02/cdctbsp01.dbf' SIZE 50M
3 AUTOEXTEND OFF
4 BLOCKSIZE 8192
5 EXTENT MANAGEMENT LOCAL UNIFORM SIZE 64K;

Tablespace created.

SQL>
SQL>
SQL>
SQL>
SQL>
SQL> CREATE USER cdcadmin
2 IDENTIFIED by cdcadmin
3 DEFAULT TABLESPACE cdc_tbsp
4 TEMPORARY TABLESPACE temp
5 QUOTA UNLIMITED ON cdc_tbsp;

User created.

SQL>
SQL>
SQL> GRANT CREATE SESSION TO cdcadmin;

Grant succeeded.

SQL> GRANT CREATE SEQUENCE TO cdcadmin;

Grant succeeded.

SQL> GRANT CREATE TABLE TO cdcadmin;

Grant succeeded.

SQL>
SQL>
SQL> GRANT SELECT_CATALOG_ROLE TO cdcadmin;

Grant succeeded.

SQL> GRANT EXECUTE_CATALOG_ROLE TO cdcadmin;

Grant succeeded.

SQL> GRANT execute ON dbms_cdc_publish TO cdcadmin;

Grant succeeded.

SQL> exec dbms_streams_auth.grant_admin_privilege('CDCADMIN');

PL/SQL procedure successfully completed.

SQL>
SQL> grant dba to cdcadmin;

Grant succeeded.

SQL

4. 準備用來做cdc複製的表（，即source table，操作是在source site）。

我們的source database的instance name叫cdc01，裡面有一個用戶叫app_user，我們選取了該用戶下的cdc_demo3表，作為用來測試的source table，同時，為了演示方便，我們只取一個表作為source table，因此source change set中，就只有一個表。所以，該表也是source change set。（上面說了，change set在source site和stage site都存在，為了區別，我把在source site的change set叫source change set，把在stage site的change set 叫stage change set。）

SQL> conn app_user/app_user
Connected.
SQL> SELECT table_name, reason
2 FROM all_streams_unsupported
3 WHERE owner = 'APP_USER'
4 ORDER BY 1;

no rows selected

SQL>
SQL> select OBJECT_NAME,OBJECT_TYPE,OBJECT_ID from t1
2 /

OBJECT_NAME OBJECT_TYPE OBJECT_ID
-------------------- -------------------------------------- ----------
T1 TABLE 51414

SQL> CREATE TABLE cdc_demo3 AS
2 SELECT * FROM T1;

Table created.

SQL>
SQL> ALTER TABLE cdc_demo3
2 ADD CONSTRAINT pk_cdc_demo3
3 PRIMARY KEY (OBJECT_ID)
4 USING INDEX;

Table altered.

SQL>
SQL> ALTER TABLE app_user.cdc_demo3 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Table altered.

SQL>

SQL> SELECT * FROM user_log_groups;

OWNER LOG_GROUP_NAME TABLE_NAME LOG_GROUP_TYPE ALWAYS GENERATED
---------- -------------------- --------------- ------------------------------ ---------------------- ----------------------------
APP_USER SYS_C005144 CDC_DEMO3 ALL COLUMN LOGGING ALWAYS GENERATED NAME

SQL>

5. 創建standby redo。
5.1 在stage site建立路徑，注意和在source site的dest2一致。

[oracle@stream2 ~]$ mkdir -p /oracle/app/oracle/arch/stdbylogs
[oracle@stream2 ~]$

5.2 在stage site建立standby log：

SQL> ALTER DATABASE ADD STANDBY LOGFILE GROUP 4
2 ('/oracle/app/oracle/arch/stdbylogs/slog04.log') SIZE 50M;

Database altered.

SQL> ALTER DATABASE ADD STANDBY LOGFILE GROUP 5
2 ('/oracle/app/oracle/arch/stdbylogs/slog05.log') SIZE 50M;

Database altered.

SQL> ALTER DATABASE ADD STANDBY LOGFILE GROUP 6
2 ('/oracle/app/oracle/arch/stdbylogs/slog06.log') SIZE 50M;

Database altered.

SQL> ALTER DATABASE ADD STANDBY LOGFILE GROUP 7
2 ('/oracle/app/oracle/arch/stdbylogs/slog07.log') SIZE 50M;

Database altered.

SQL> --注意group數要比source site的log的group數多1.原來有3組，所以要建4組standby redo log.
SQL>
SQL> SELECT group#, bytes, status
2 FROM v$standby_log;

   GROUP# BYTES STATUS
---------- ---------- --------------------
   4 52428800 UNASSIGNED
   5 52428800 UNASSIGNED
   6 52428800 UNASSIGNED
   7 52428800 UNASSIGNED

SQL>

6. 產生實體source site的資料字典。
取source site的scn，是為了創建logmnr資料字典的時候，需要知道scn。

SQL> DECLARE
2 f_scn NUMBER;
3 BEGIN
4 dbms_capture_adm.build ( f_scn );
5 DBMS_OUTPUT.PUT_LINE('The first_scn value is: '|| f_scn);
6 END;
7 /
The first_scn value is: 524561

PL/SQL procedure successfully completed.

SQL>

我們記下該scn號：524561

7. 準備soruce table的產生實體。

注意在source site的每一個source table都必須準備產生實體，如果不準備產生實體，在stage site的capture會出現問題，無法capture source table的變化。

SQL> conn / as sysdba
Connected.
SQL> desc dba_capture_prepared_tables
Name Null? Type
----------------------------------------- -------- ----------------------------
TABLE_OWNER NOT NULL VARCHAR2(30)
TABLE_NAME NOT NULL VARCHAR2(30)
SCN NOT NULL NUMBER
TIMESTAMP DATE
SUPPLEMENTAL_LOG_DATA_PK VARCHAR2(8)
SUPPLEMENTAL_LOG_DATA_UI VARCHAR2(8)
SUPPLEMENTAL_LOG_DATA_FK VARCHAR2(8)
SUPPLEMENTAL_LOG_DATA_ALL VARCHAR2(8)

SQL>
SQL>
SQL> SELECT table_name, scn, supplemental_log_data_pk PK, supplemental_log_data_ui UI,
2 supplemental_log_data_fk FK, supplemental_log_data_all "ALL"
3 FROM dba_capture_prepared_tables;

no rows selected

SQL>

SQL> exec dbms_capture_adm.prepare_table_instantiation('app_user.t1');

PL/SQL procedure successfully completed.

SQL>
SQL>
SQL>
SQL> SELECT table_name, scn, supplemental_log_data_pk PK, supplemental_log_data_ui UI,
2 supplemental_log_data_fk FK, supplemental_log_data_all "ALL"
3* FROM dba_capture_prepared_tables
SQL> /

TABLE_NAME SCN PK UI FK ALL
---------- ---------- ---------------- ---------------- ---------------- ----------------
T1 525556 IMPLICIT IMPLICIT IMPLICIT NO

SQL>

8. 確定各個source database和創建change source。

在我們的例子中，source database就只有一個cdc01，change source，再說一次，是出現在stage中的。我們把change source取名叫做CS_DEMO3。
在stage site：

SQL> exec dbms_cdc_publish.create_autolog_change_source('CS_DEMO3', 'AutoLog Demo', 'CDC01', 524561, 'Y');

PL/SQL procedure successfully completed.

SQL> --注意，如果是非同步Autolog模式-archive log，最後的一個參數可以用'N'。

9. 創建stage change set。我們這裡把stage change set取名叫做OMEGA_CSET。

注意當CDC創建了change set的時候，stream的capture和apply進程也被同時創建了，但是僅僅是創建，還未啟動。

在stage site：

SQL> exec dbms_cdc_publish.create_change_set('OMEGA_CSET', 'change set info', 'CS_DEMO3', 'Y');

PL/SQL procedure successfully completed.

SQL>

10. 確定用於查詢的用戶，即subscriber，我們假定該用戶叫為app_dev不存在，我們新建一個。

注：該使用者在change table創建完成之後，需要授權該使用者能讀取change table。

SQL> create user app_dev identified by app_dev default tablespace USERS;

User created.

SQL> grant connect,resource to app_dev;

Grant succeeded.

SQL>

11. 在stage site上，創建change table。

change table是publisher做的最後一個事情，從目標到最後的過程為：source table->source change set->change source->stage change set->change table.當然，這期間可以過濾某些欄位，或者某些表。

我們開始創建change table，在stage site：

SQL> conn cdcadmin/cdcadmin
Connected.
SQL> SELECT object_name, object_type
2 FROM user_objects
3 ORDER BY 2,1;

no rows selected

SQL>
SQL> exec dbms_cdc_publish.create_change_table('CDCADMIN', 'T1_CTAB', 'OMEGA_CSET', 'APP_USER', 'CDC_DEMO3', 'OBJECT_NAME VARCHAR2(128), SUBOBJECT_NAME VARCHAR2(30), OBJECT_ID NUMBER, DATA_OBJECT_ID NUMBER, OBJECT_TYPE VARCHAR2(19), CREATED DATE, LAST_DDL_TIME DATE, TIMESTAMP VARCHAR2(19), STATUS VARCHAR2(7), TEMPORARY VARCHAR2(1), GENERATED VARCHAR2(1), SECONDARY VARCHAR2(1)', 'BOTH', 'Y', 'N', 'N', 'N', 'N', 'N', 'Y', 'TABLESPACE CDC_TBSP');

PL/SQL procedure successfully completed.

SQL> SELECT object_name, object_type
2 FROM user_objects
3 ORDER BY 2,1;

OBJECT_NAME OBJECT_TYPE
-------------------- --------------------------------------
T1_CTAB TABLE
T1_CTAB TABLE PARTITION

SQL>
SQL> SELECT table_name, composite, partition_name, high_value
2* FROM user_tab_partitions
SQL> /

TABLE_NAME COMPOS PARTITION_NAME HIGH_VALUE
-------------------- ------ ------------------------------------------------------------ ---------------
T1_CTAB NO P1 281474976710656

SQL>
SQL> conn cdcadmin/cdcadmin
Connected.
SQL> GRANT select ON t1_ctab TO app_dev;

Grant succeeded.

SQL>
SQL> conn / as sysdba
Connected.
SQL>
SQL> desc cdc_change_tables$
Name Null? Type
----------------------------------------------------- -------- ------------------------------------
OBJ# NOT NULL NUMBER
CHANGE_SET_NAME NOT NULL VARCHAR2(30)
SOURCE_SCHEMA_NAME NOT NULL VARCHAR2(30)
SOURCE_TABLE_NAME NOT NULL VARCHAR2(30)
CHANGE_TABLE_SCHEMA NOT NULL VARCHAR2(30)
CHANGE_TABLE_NAME NOT NULL VARCHAR2(30)
CREATED NOT NULL DATE
CREATED_SCN NUMBER
MVL_FLAG NUMBER
CAPTURED_VALUES NOT NULL VARCHAR2(1)
MVL_TEMP_LOG VARCHAR2(30)
MVL_V7TRIGGER VARCHAR2(30)
LAST_ALTERED DATE
LOWEST_SCN NOT NULL NUMBER
MVL_OLDEST_RID NUMBER
MVL_OLDEST_PK NUMBER
MVL_OLDEST_SEQ NUMBER
MVL_OLDEST_OID NUMBER
MVL_OLDEST_NEW NUMBER
MVL_OLDEST_RID_TIME DATE
MVL_OLDEST_PK_TIME DATE
MVL_OLDEST_SEQ_TIME DATE
MVL_OLDEST_OID_TIME DATE
MVL_OLDEST_NEW_TIME DATE
MVL_BACKCOMPAT_VIEW VARCHAR2(30)
MVL_PHYSMVL VARCHAR2(30)
HIGHEST_SCN NUMBER
HIGHEST_TIMESTAMP DATE
CHANGE_TABLE_TYPE NOT NULL NUMBER
MAJOR_VERSION NOT NULL NUMBER
MINOR_VERSION NOT NULL NUMBER
SOURCE_TABLE_OBJ# NUMBER
SOURCE_TABLE_VER NUMBER

SQL>
SQL> l
1 SELECT change_set_name, source_schema_name, source_table_name
2* FROM cdc_change_tables$
SQL> /

CHANGE_SET_NAME SOURCE_SCHEMA_NAME SOURCE_TABLE_NAME
-------------------- -------------------- ------------------------------------------------------------
OMEGA_CSET APP_USER CDC_DEMO3

SQL>

12. enable change set:
注意當change set is enabled，Streams capture進程和apply進程將啟動。

SQL> conn / as sysdba
Connected.
SQL> SELECT set_name, change_source_name, capture_enabled
2 FROM cdc_change_sets$;

SET_NAME CHANGE_SOURCE_NAME CA
------------------------------------------------------------ ------------------------------------------------------------ --
SYNC_SET SYNC_SOURCE Y
OMEGA_CSET CS_DEMO3 N

SQL>
SQL> conn cdcadmin/cdcadmin
Connected.
SQL>
SQL> exec dbms_cdc_publish.alter_change_set('OMEGA_CSET', enable_capture => 'Y');

PL/SQL procedure successfully completed.

SQL>
SQL> conn / as sysdba
Connected.
SQL> SELECT set_name, change_source_name, capture_enabled
2 FROM cdc_change_sets$;

SET_NAME CHANGE_SOURCE_NAME CA
------------------------------------------------------------ ------------------------------------------------------------ --
SYNC_SET SYNC_SOURCE Y
OMEGA_CSET CS_DEMO3 Y

SQL>

13. 在source site做switch logfile.

SQL> conn / as sysdba
Connected.
SQL> alter system switch logfile;

System altered.

SQL>

好了，上述publisher的配置就完成了。但是我們還要繼續配置subscriber。

14. 檢查source table對subscriber是否有access許可權：
只需在stage site檢查all_source_name即可，不需去source site：

SQL> SELECT * FROM ALL_SOURCE_TABLES;

SOURCE_SCHEMA_NAME SOURCE_TABLE_NAME
-------------------- --------------------
APP_USER CDC_DEMO3

SQL>

15. 找到subscriber有許可權訪問的change set的名字和change set的每一列：
在stage site：

SQL> SELECT UNIQUE CHANGE_SET_NAME, COLUMN_NAME, PUB_ID
2 FROM ALL_PUBLISHED_COLUMNS
3 WHERE SOURCE_SCHEMA_NAME ='APP_USER' AND SOURCE_TABLE_NAME = 'CDC_DEMO3';

CHANGE_SET_NAME COLUMN_NAME PUB_ID
-------------------- -------------------- ----------
OMEGA_CSET CREATED 51473
OMEGA_CSET DATA_OBJECT_ID 51473
OMEGA_CSET GENERATED 51473
OMEGA_CSET LAST_DDL_TIME 51473
OMEGA_CSET OBJECT_ID 51473
OMEGA_CSET OBJECT_NAME 51473
OMEGA_CSET OBJECT_TYPE 51473
OMEGA_CSET SECONDARY 51473
OMEGA_CSET STATUS 51473
OMEGA_CSET SUBOBJECT_NAME 51473
OMEGA_CSET TEMPORARY 51473
OMEGA_CSET TIMESTAMP 51473

12 rows selected.

SQL>

16. 創建subscription。
在stage site：

SQL> BEGIN
2 DBMS_CDC_SUBSCRIBE.CREATE_SUBSCRIPTION(
3 change_set_name => 'OMEGA_CSET',
4 description => 'Change data for cdc_demo3',
5 subscription_name => 'cdc_demo3_sub');
6 END;
7 /

PL/SQL procedure successfully completed.

SQL>

17. 訂閱source table和source table中的列，你可以訂閱所有的列，可以選擇幾列來訂閱：

SQL> BEGIN
2 DBMS_CDC_SUBSCRIBE.SUBSCRIBE(
3 subscription_name => 'cdc_demo3_sub',
4 source_schema => 'APP_USER',
5 source_table => 'CDC_DEMO3',
6 column_list => 'OBJECT_NAME, OBJECT_ID, STATUS',
7 subscriber_view => 'CDC_DEMO3_SUB_VIEW');
8 END;
9 /

PL/SQL procedure successfully completed.

SQL>

注意，如果在第15步，查到的各個列的pub_id是不一樣的，那麼在訂閱的時候，就要針對不同的列指定不同的pub id，詳情可見online document上的Database Data Warehousing Guide

18. 啟動訂閱：

SQL> BEGIN
2 DBMS_CDC_SUBSCRIBE.ACTIVATE_SUBSCRIPTION(
3 subscription_name => 'cdc_demo3_sub');
4 END;
5 /

PL/SQL procedure successfully completed.

SQL>

19. 獲得下一個可用的change data。

SQL> BEGIN
2 DBMS_CDC_SUBSCRIBE.EXTEND_WINDOW(
3 subscription_name => 'cdc_demo3_sub');
4 END;
5 /

PL/SQL procedure successfully completed.

SQL>

20. 完成，此時你就可以在CDC_DEMO3_SUB_VIEW中查資料了。

參考文檔：
Metalink：Doc ID 972876.1
Oracle Streams and Change Data Capture(CDC) 10gR2
Online Document-Database Data Warehousing Guide
Oracle Change Data Capture Asynchronous Autolog Demo