SELECT supplemental_log_data_min,        
       supplemental_log_data_pk,        
       supplemental_log_data_ui,         
       supplemental_log_data_fk,        
       supplemental_log_data_all 
FROM v$database; 

alter database add supplemental log data;

alter table MonSchema.MaTable add supplemental log data (PRIMARY KEY) columns ; 

select 'alter table '||owner||'.'||table_name||' drop supplemental log group '||log_group_name||';' 
from dba_log_groups 
where owner = 'MonSchema' 
and table_name = ?MaTable';

select table_owner, table_name,supplemental_log_data_all, supplemental_log_data_FK, supplemental_log_data_ui, supplemental_log_data_pk 
from dba_capture_prepared_tables 
where supplemental_log_data_all <> 'NO' 
OR supplemental_log_data_PK<>'NO' 
OR supplemental_log_data_FK<>'NO' 
OR supplemental_log_data_UI<>'NO' ; 

CONNECT SYS/******@ORAOLTP AS SYSDBA
CREATE USER STREAMSADM IDENTIFIED BY ******** ACCOUNT UNLOCK ;
GRANT DBA TO STREAMSADM ;
GRANT RESOURCES TO STREAMSADM ;

CONNECT SYS/******@ORADWH AS SYSDBA
CREATE TABLESPACE STREAMS_D01 DATAFILE '/VotreLocalisation/STREAMSADM_D01.dat' SIZE=200M AUTOEXTEND ON NEXT 100M MAXSIZE 2G ;
CREATE USER STREAMSADM IDENTIFIED BY ********  DEFAULT TABLESPACE STREAMS_D01 ACCOUNT UNLOCK;
GRANT DBA TO STREAMSADM ;
GRANT RESOURCE TO STREAMSADM ;

GRANT EXECUTE ON DBMS_FLASHBACK TO STREAMSADM ; 
GRANT EXECUTE ON  dbms_streams_adm TO STREAMSADM ; 
GRANT EXECUTE ON  dbms_apply_adm TO STREAMSADM ; 
GRANT EXECUTE ON  dbms_streams_adm TO STREAMSADM ;

BEGIN
  DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee          => 'STREAMSADM',    
    grant_privileges => true);
END;
/

CONNECT STREAMSADM/*******@ORADWH
-- création
CREATE DATABASE LINK ORAOLTP CONNECT TO CURRENT_USER USING 'ORAOLTP.DEVELOPPEZ.COM';

-- Test
SELECT * FROM DUAL@ORAOLTP ;

PROMPT Création de la queue de capture
begin
    dbms_streams_adm.set_up_queue(
	queue_table  => 'streamsadm.stream_queue_cpt',
	queue_name   => 'stream_queue_cpt',
	queue_user   => 'streamsadm');
end;
/

-- Test
SELECT * FROM streamsadm.stream_queue_cpt ;

PROMPT Création de la queue d'application

begin
     dbms_streams_adm.set_up_queue(
     	queue_table  => 'streamsadm.stream_queue_appl',
	queue_name   => 'stream_queue_appl',
	queue_user   => 'streamsadm');
end;
/

-- Test
SELECT * FROM streamsadm.stream_queue_appl ;

BEGIN
  DBMS_CAPTURE_ADM.CREATE_CAPTURE(
    queue_name         => 'streamsadm.stream_queue_cpt',
    capture_name       => 'strm_capture',
    rule_set_name      => NULL,
    start_scn          => NULL,
    source_database    => 'ORAOLTP',
    use_database_link  => true,
    first_scn          => NULL,
    logfile_assignment => 'implicit');
END;
/

-- Test

SELECT CAPTURE_NAME, STATUS, STATUS_CHANGE_TIME FROM DBA_CAPTURE ;

DECLARE
   cStreamUser CONSTANT VARCHAR2(30)           := 'STREAMSADM';
   cStreamPropName CONSTANT VARCHAR2(30)	   := 'STRM_PROP';
   cStreamApplyName CONSTANT VARCHAR2(30)      := 'STRM_APPLY';
   cStreamCaptureName CONSTANT VARCHAR2(30)	   := 'STRM_CAPTURE';
   cSourceQueueName CONSTANT VARCHAR2(30)      := cStreamUser || '.' || 'STREAM_QUEUE_CPT';
   cDestinationQueueName CONSTANT VARCHAR2(30) := cStreamUser || '.' || 'STREAM_QUEUE_APPL';
   
   pOwner VARCHAR2(30) := 'STREAM';
   pTableName VARCHAR2(30) := 'STREAM_LASTCOMMIT';
   pSourceInstance VARCHAR2(30) := 'ORAOLTP.DEVELOPPEZ.COM';
   iScn NUMBER := 0 ;
   vSql varchar2(500) ;
 
 BEGIN
  
 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    				 table_name			=>  Upper(pOwner)||'.'||Upper(pTableName),
    				 streams_type		=>  'capture',
    				 streams_name		=>  cStreamCaptureName,
    				 queue_name			=>  cSourceQueueName,
    				 include_dml		=>  true,
    				 include_ddl		=>  false,
    				 include_tagged_lcr	=>  false,
    				 source_database	=>  pSourceInstance,
    				 inclusion_rule		=>  true);


dbms_streams_adm.add_table_rules( 
		table_name         			=> pOwner||'.'||pTableName,
		streams_type       			=> 'APPLY',
		streams_name       			=> cStreamApplyName,
		queue_name         			=> cSourceQueueName,
		include_dml        			=>  true,
		include_ddl        			=>  FALSE,
		include_tagged_lcr 			=>  false,
		inclusion_rule     			=>  true,
		source_database    			=> pSourceInstance);

vSql := 'SELECT  DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER() into iSCN FROM DUAL@'||pSourceInstance ;
execute immediate (vSql) ;

vSql := 'truncate table '|| pOwner||'.'||pTableName ;
execute immediate (vSql) ;

vSql := 'insert into '|| pOwner||'.'||pTableName
            ||' select * from '||pOwner||'.'||pTableName||'@'||pSourceInstance ;
execute immediate (vSql) ;

dbms_apply_adm.set_table_instantiation_scn (
        source_object_name			=> Upper(pOwner)||'.'||Upper(pTableName),
        source_database_name		=> pSourceInstance,
		instantiation_scn     		=> iSCN 
		) ;
END;
/

SELECT 'CAPTURE', C.CAPTURE_NAME, C.STATUS, C.STATUS_CHANGE_TIME 
FROM DBA_CAPTURE C
UNION ALL
SELECT 'APPLY', a.APPLY_NAME, a.STATUS, a.STATUS_CHANGE_TIME 
from DBA_APPLY a

SELECT * from DBA_APPLY_ERROR ;

SET SERVEROUTPUT ON SIZE 1000000

EXEC print_transaction('1.12.3455')

select  substr(rule_condition, instr(rule_condition,'''')+1, instr(substr(rule_condition, instr(rule_condition,'''')+1),'''')-1) SCHEMAS, 
        substr(rule_condition, instr(rule_condition, 'get_object_name()')+21, 
        instr(rule_condition, ')) and :dml.is_null_tag()') - instr(rule_condition, 'get_object_name()')-22) TABLES,
        source_database 
from sys.STREAMS$_RULES 
where streams_name='STRM_CAPTURE';

BEGIN
DBMS_STREAMS_ADM.REMOVE_STREAMS_CONFIGURATION ;

FOR maqueue IN (select name from dba_queues where owner = 'STREAMSADM') 
LOOP
    DBMS_STREAMS_ADM.REMOVE_QUEUE(maqueue.name,TRUE);
END LOOP;

FOR maregle IN (select rule_set_name from dba_rule_sets where rule_set_owner = 'STREAMSADM')
LOOP
   dbms_rule_adm.drop_rule_set(maregle.rule_set_name,TRUE);
END LOOP;
END ;
/