/*
*
*  Author  : Vishal Gupta
*  Purpose : Merge statspack from multiple sources into single database
*
*  Revision History:
*  ===================
*  Date       Author        Description
*  ---------  ------------  -----------------------------------------
*  05-Aug-07  Vishal Gupta  First Draft
*/

set serveroutput on size 1000000
set verify off

DECLARE
  sql_text                 VARCHAR2(32767);
  v_env_type               VARCHAR2(30) := 'PROD';
  v_db_version             VARCHAR2(30) := '10.2';
  v_instance               VARCHAR2(30)  := 'TRIP01PD';
  v_db_link                VARCHAR2(4000) ;
  v_perfstat_schema_owner  VARCHAR2(30) ;
  v_update_cols_count      NUMBER;
  v_scn                    NUMBER;
  v_dbid                   NUMBER;
  v_snap_id_exists         VARCHAR2(1);
  v_days                   NUMBER := 3;
  v_sqlcode                NUMBER := 0;

 
  CURSOR cur_constraint_cols(p_owner DBA_TAB_COLUMNS.OWNER%TYPE
                           , p_table_name DBA_TAB_COLUMNS.TABLE_NAME%TYPE)  
  is
              SELECT cc.COLUMN_NAME, tc.NULLABLE
                FROM DBA_CONSTRAINTS C, DBA_CONS_COLUMNS CC , DBA_TAB_COLUMNS tc
               WHERE C.TABLE_NAME = p_table_name
               AND   C.CONSTRAINT_TYPE in ( 'P' ,'U') 
               AND   C.OWNER = p_owner
               AND   C.OWNER = CC.OWNER
               AND   C.CONSTRAINT_NAME = CC.CONSTRAINT_NAME
               AND   C.TABLE_NAME = CC.TABLE_NAME
               AND   cc.OWNER = tc.OWNER
               AND   cc.TABLE_NAME = tc.TABLE_NAME
               AND   cc.COLUMN_NAME = tc.COLUMN_NAME
               order by POSITION;
  rec_constraint_cols  cur_constraint_cols%ROWTYPE;              

  CURSOR cur_tab_cols(p_owner DBA_TAB_COLUMNS.OWNER%TYPE
                    , p_table_name DBA_TAB_COLUMNS.TABLE_NAME%TYPE)  
  is 
              SELECT column_name 
                FROM DBA_TAB_COLUMNS 
               WHERE TABLE_NAME = p_table_name 
                 AND OWNER = p_owner
               order by COLUMN_ID;
  rec_tab_cols  cur_tab_cols%ROWTYPE;              

  CURSOR cur_is_table_partitioned(p_owner DBA_TAB_COLUMNS.OWNER%TYPE
                    , p_table_name DBA_TAB_COLUMNS.TABLE_NAME%TYPE)
  is
              SELECT partitioned
                FROM dba_tables
               WHERE TABLE_NAME = p_table_name
                 AND OWNER = p_owner;
  rec_is_table_partitioned cur_is_table_partitioned%ROWTYPE;
   
  CURSOR cur_update_cols(p_owner DBA_TAB_COLUMNS.OWNER%TYPE
                       , p_table_name DBA_TAB_COLUMNS.TABLE_NAME%TYPE)  
  is 
              SELECT column_name 
                FROM DBA_TAB_COLUMNS t
               WHERE TABLE_NAME = p_table_name 
                 AND OWNER = p_owner
                 AND t.column_name NOT IN ( SELECT COLUMN_NAME FROM DBA_CONSTRAINTS C, DBA_CONS_COLUMNS CC 
                                   WHERE C.TABLE_NAME = t.table_name
                                   AND   C.CONSTRAINT_TYPE in ( 'P' ,'U') 
                                   AND   C.OWNER = t.OWNER
                                   AND   C.OWNER = CC.OWNER
                                   AND   C.CONSTRAINT_NAME = CC.CONSTRAINT_NAME
                                   AND   C.TABLE_NAME = CC.TABLE_NAME)
               order by COLUMN_ID;
  rec_update_cols  cur_update_cols%ROWTYPE;              

  CURSOR cur_check_snapid_exists(p_owner DBA_TAB_COLUMNS.OWNER%TYPE, p_table_name DBA_TAB_COLUMNS.TABLE_NAME%TYPE)  
  is 
              SELECT 'Y' 
                FROM DBA_CONSTRAINTS a
               WHERE OWNER = p_owner
                 AND TABLE_NAME = p_table_name 
                 AND CONSTRAINT_TYPE = 'R'
                 AND R_CONSTRAINT_NAME = 'STATS$SNAPSHOT_PK';

BEGIN

 DBMS_OUTPUT.PUT_LINE('###########################################');
 DBMS_OUTPUT.PUT_LINE('Job Started at ' || to_char(sysdate,'DD-MON-YYYY hh24:MI:SS') );

 FOR i in (SELECT database
                , active
                , version
                , environment
                , schema_name 
             from perfstat_control.STATSPACK_MERGE_CONTROL 
            where active = 'Y' 
              AND version = v_db_version
              AND environment = v_env_type
              AND database    = NVL(v_instance,database) 
            ORDER BY database)
   LOOP
         v_db_link := i.database;
         v_perfstat_schema_owner := i.schema_name;
         
         DBMS_OUTPUT.PUT_LINE('------------------------------------------------');
         DBMS_OUTPUT.PUT_LINE(v_db_link);
         DBMS_OUTPUT.PUT_LINE(v_perfstat_schema_owner);

         -- New data might get inserted while we are merging all the tables of statspack
         -- corresponding snap_id would not exists for data inserted after merge of stats$snapshot
         -- So, we need to use flashback query "as os SCN 1234".
         v_scn := 0;

         BEGIN           
                  
               EXECUTE IMMEDIATE 'SELECT sys.dbms_flashback.get_system_change_number() FROM dual@' || v_db_link INTO v_scn;
               --dbms_output.put_line('SELECT dbid from FROM v$database@' || v_db_link);
               EXECUTE IMMEDIATE 'select dbid from stats$database_instance@' || v_db_link || ' where startup_time = (select max(startup_time) from stats$database_instance@' || v_db_link || ')' INTO v_dbid;
               --dbms_output.put_line('SELECT dbid from FROM v$database@' || v_db_link);
      
               FOR tbl in (
               select 'STATS$DATABASE_INSTANCE' table_name from dual
                       UNION ALL
                          select 'STATS$LEVEL_DESCRIPTION' from dual
                       UNION ALL
                          select 'STATS$STATSPACK_PARAMETER' from dual
                       UNION ALL
                          select 'STATS$SNAPSHOT' from dual
                       UNION ALL
                          select * 
                          from  (SELECT table_name 
                                   FROM DBA_TABLES 
                                  WHERE OWNER = v_perfstat_schema_owner
                                    and  table_name not in ('STATS$DATABASE_INSTANCE'
                                                           ,'STATS$LEVEL_DESCRIPTION'
                                                           ,'STATS$STATSPACK_PARAMETER'
                                                           ,'STATS$SNAPSHOT'
                                                           ,'STATS$IDLE_EVENT') 
                                  order by 1
                                  )
                          )
               loop
            
                /*   Sample Merge statement
                --    
                --    Limit the historical data to be sampled from source database to x days old.
                --    Old data in central repository is not deleted.
                --    
                --    MERGE
                --    INTO table_l l 
                --    USING (select a.* from stats$sql_summary@FIRE02PD a as of SCN 1234, stats$snapshot@FIRE02PD b as of SCN 1234 
                --           where a.dbid = b.dbid and a.instance_number = b.instance_number and a.snap_id = b.snap_id
                --           and and b.snap_time > sysdate - 3 ) r 
                --    ON( l.primary_key_col1 = r.primary_key_col1
                --        AND l.primary_key_col2 = r.primary_key_col2  -- For nullable columns use NVL
                --       )
                --    WHEN matched THEN
                --        UPDATE
                --        SET l.col1 = r.col2,
                --            l.col2 = r.col2
                --        WHERE l.col1 != r.col1
                --           OR l.col2 != r.col2
                --           OR l.col3 != r.col3
                --    WHEN NOT matched THEN
                --        INSERT(col1, col2)
                --        VALUES(r.col1, r.col2)
                --    ;


                */
            
               v_snap_id_exists := 'N';
               OPEN cur_check_snapid_exists(v_perfstat_schema_owner, tbl.table_name);
               FETCH cur_check_snapid_exists INTO v_snap_id_exists;
               CLOSE cur_check_snapid_exists;

               OPEN cur_is_table_partitioned(v_perfstat_schema_owner, tbl.table_name);
               FETCH cur_is_table_partitioned INTO rec_is_table_partitioned;
               CLOSE cur_is_table_partitioned;
            
               sql_text := 'merge /*+ index(L) index(R) */ into '  ; 
               sql_text := sql_text || v_perfstat_schema_owner || '.' || tbl.table_name ;

               IF v_snap_id_exists = 'Y' AND rec_is_table_partitioned.partitioned = 'YES' THEN
                 sql_text := sql_text || ' PARTITION (P' || v_dbid || ') ';
               END IF;

               sql_text := sql_text || ' l using ( SELECT a.* from ' || tbl.table_name || '@' || v_db_link || ' as of SCN ' || v_scn || ' a ';
                           
               IF v_snap_id_exists = 'Y' THEN
                 sql_text := sql_text || ' , stats$snapshot@' || v_db_link || ' as of SCN ' || v_scn || ' b ';
                 sql_text := sql_text || ' where a.dbid = b.dbid ';
                 sql_text := sql_text || ' and a.instance_number = b.instance_number ';
                 sql_text := sql_text || ' and a.snap_id = b.snap_id ';
                 sql_text := sql_text || ' and b.snap_time > sysdate - ' || v_days ;
               END IF;            
               
               sql_text := sql_text || ' ) r ON ( ';
            
               OPEN cur_constraint_cols(v_perfstat_schema_owner, tbl.table_name);
               LOOP
                  FETCH cur_constraint_cols INTO rec_constraint_cols;
                  EXIT WHEN cur_constraint_cols%NOTFOUND;
                  IF rec_constraint_cols.nullable = 'Y' THEN
                     sql_text := sql_text || 'nvl(l.' || rec_constraint_cols.COLUMN_NAME || ',''0'') = nvl(r.' || rec_constraint_cols.column_name||',''0'') AND ';
                  ELSE
                     sql_text := sql_text || 'l.' || rec_constraint_cols.COLUMN_NAME || ' = r.' || rec_constraint_cols.column_name||' AND ';
                  END IF;            
               END LOOP;
               CLOSE cur_constraint_cols;
               sql_text := SUBSTR(sql_text,1, LENGTH(sql_text) -4);
               sql_text := sql_text || ' )  ';
            
               OPEN cur_update_cols(v_perfstat_schema_owner, tbl.table_name);
               LOOP
                  FETCH cur_update_cols INTO rec_update_cols;
                  EXIT WHEN cur_update_cols%NOTFOUND;
               END LOOP;   
               v_update_cols_count := cur_update_cols%ROWCOUNT;
               CLOSE cur_update_cols;
              
               IF v_update_cols_count > 0 THEN
                   sql_text := sql_text || ' WHEN MATCHED THEN UPDATE SET ';
                   OPEN cur_update_cols(v_perfstat_schema_owner, tbl.table_name);
                   LOOP
                      FETCH cur_update_cols INTO rec_update_cols;
                      EXIT WHEN cur_update_cols%NOTFOUND;
                      sql_text := sql_text || 'l.' || rec_update_cols.COLUMN_NAME || ' = r.' || rec_update_cols.column_name||',';
                   END LOOP;
                   CLOSE cur_update_cols;
                   sql_text := SUBSTR(sql_text,1, LENGTH(sql_text) - 1);
                
                   sql_text := sql_text || ' WHERE ' ;
                   OPEN cur_update_cols(v_perfstat_schema_owner, tbl.table_name);
                   LOOP
                      FETCH cur_update_cols INTO rec_update_cols;
                      EXIT WHEN cur_update_cols%NOTFOUND;
                      sql_text := sql_text || 'l.' || rec_update_cols.COLUMN_NAME || ' != r.' || rec_update_cols.column_name||' OR ';
                   END LOOP;
                   CLOSE cur_update_cols;
                   sql_text := SUBSTR(sql_text,1, LENGTH(sql_text) -3);
      
               END IF;
               
               sql_text := sql_text || ' WHEN NOT MATCHED THEN INSERT (' ;
               
               OPEN cur_tab_cols(v_perfstat_schema_owner, tbl.table_name);
               LOOP
                  FETCH cur_tab_cols INTO rec_tab_cols;
                  EXIT WHEN cur_tab_cols%NOTFOUND;
                  sql_text := sql_text || rec_tab_cols.COLUMN_NAME || ',';
               END LOOP;
               CLOSE cur_tab_cols;
               sql_text := SUBSTR(sql_text,1, LENGTH(sql_text) -1);
            
               sql_text := sql_text || ' ) VALUES (';
            
               OPEN cur_tab_cols(v_perfstat_schema_owner, tbl.table_name);
               LOOP
                  FETCH cur_tab_cols INTO rec_tab_cols;
                  EXIT WHEN cur_tab_cols%NOTFOUND;
                  sql_text := sql_text ||  'r.' || rec_tab_cols.COLUMN_NAME || ',';
               END LOOP;
               CLOSE cur_tab_cols;
               sql_text := SUBSTR(sql_text,1, LENGTH(sql_text) -1);
               sql_text := sql_text || ' ) ' ;
               DBMS_OUTPUT.PUT_LINE (tbl.table_name);   
                IF tbl.table_name = 'STATS$SGASTAT' THEN
                  null;
                  DBMS_OUTPUT.PUT_LINE (SQL_TEXT);   
                  EXIT;
                END IF;
                --DBMS_OUTPUT.PUT_LINE (SQL_TEXT);   
               --execute immediate sql_text ;  
               --commit;
               end loop;

               --EXECUTE immediate 'alter session close database link ' || v_db_link ;

          EXCEPTION 
              WHEN others THEN
                  DBMS_OUTPUT.PUT_LINE(sqlerrm);
                  IF v_sqlcode = 0 THEN
                     v_sqlcode := SQLCODE;
                  END IF;
               FOR i in (select db_link from v$dblink)
               LOOP
                 commit;
                 EXECUTE immediate 'alter session close database link ' || i.db_link ;
               END LOOP; 
          END;     
          
    END LOOP;     
    DBMS_OUTPUT.PUT_LINE('###########################################');
    DBMS_OUTPUT.PUT_LINE('Job Finished at ' || to_char(sysdate,'DD-MON-YYYY hh24:MI:SS') );
   
    IF v_sqlcode <> 0 THEN
       raise_application_error(-20001,'There were some errors in statspack consolidation'); 
    END IF;
END;
/