Process all

CREATE TABLE NIC_EDW_STG.JOB_MASTER

(
        JOB_ID NUMBER,

JOB_NAME VARCHAR2(100),

PKG_NAME VARCHAR2(200),

JOB_TYPE_ID NUMBER,

JOB_RUN_DAY NUMBER(2,0),

JOB_RUN_INTERVAL NUMBER,

JOB_RUN_TIME TIMESTAMP (6),

LAST_SCHED_TIME TIMESTAMP (6),

WAIT_TIME TIMESTAMP (6),

JOB_ACTIVE NUMBER(2,0),

JOB_FATAL NUMBER(2,0),

JOB_LOAD_TYPE VARCHAR2(20),

SCENARIO_VER VARCHAR2(10) DEFAULT 001

)

CREATE TABLE NIC_EDW_STG.JOB_QUEUE
(         
QUEUE_ID NUMBER,

SEQ_ID NUMBER,

JOB_ID NUMBER,

JOB_START_TIME TIMESTAMP (6),

JOB_END_TIME TIMESTAMP (6),

JOB_STATUS NUMBER(2,0)
)

CREATE TABLE NIC_EDW_STG.JOB_EVENTS
(         
EVENT_ID NUMBER,

SEQ_NO NUMBER,

QUEUE_ID NUMBER,

STEP_ID NUMBER,

LOG_TIME TIMESTAMP (6),

STEP_NAME VARCHAR2(100),

STEP_DESC VARCHAR2(100),

STEP_ERROR VARCHAR2(250),

TOTAL_AFFECTED_ROWS NUMBER,

STEP_STATUS NUMBER(2,0)
)



CREATE TABLE NIC_EDW_STG.JOB_DEPENDENCY
(         
JOB_ID NUMBER,

JOB_PARENT_ID NUMBER,

PARENT_OLTP_F CHAR(1)

)



/* ODI work repository table */



CREATE TABLE DWPROD_ODIWORK_EXEC.SNP_SESSION

(           SESS_NO NUMBER(19,0) NOT NULL ENABLE,

SESS_NAME VARCHAR2(436 CHAR) NOT NULL ENABLE,

SCEN_VERSION VARCHAR2(35 CHAR),

LOG_LEVEL NUMBER(4,0) NOT NULL ENABLE,

THREAD_ID VARCHAR2(35 CHAR),

THREAD_CREATION VARCHAR2(35 CHAR),

SESS_BEG DATE,

SESS_END DATE,

SESS_DUR NUMBER(10,2),

SESS_STATUS VARCHAR2(2 CHAR) NOT NULL ENABLE,

SESS_RC VARCHAR2(35 CHAR),

SESS_MESS VARCHAR2(250 CHAR),

SESS_PARAMS VARCHAR2(250 CHAR),

SESS_KEYWORDS VARCHAR2(250 CHAR),

MASTER_AGENT_NAME VARCHAR2(35 CHAR),

SYNCHRO VARCHAR2(1 CHAR),

AGENT_NAME VARCHAR2(35 CHAR),

CONTEXT_CODE VARCHAR2(35 CHAR) NOT NULL ENABLE,

PARENT_SESS_NO NUMBER(19,0),

NB_CHILD_RUN NUMBER(10,0),

NB_CHILD_OK NUMBER(10,0),

NB_CHILD_KO NUMBER(10,0),

SCEN_NAME VARCHAR2(400 CHAR),

USER_NAME VARCHAR2(35 CHAR),

I_TXT_SESS_MESS NUMBER(19,0),

I_TXT_SESS_PARAMS NUMBER(19,0)

)
**************************************************************************

/* PROCESS_STATUS_PROC-->Procedure-->PROCESS_STATUS_PROC-->Update Status */
 DECLARE
CURSOR SESS IS SELECT sess_status FROM odi_exc_repository.snp_session WHERE
SESS_NO=96073565 ;
CURSOR STAT_CUR IS   SELECT QUEUE_ID,SEQ_ID,JOB_ID,JOB_START_TIME,DIFF,WAIT_TIME,JOB_STATUS,STEP_STATUS,STEP_STATUS_C,EVENT_D_C FROM (
 SELECT DISTINCT a.QUEUE_ID
   ,a.SEQ_ID
   ,a.JOB_ID
   ,a.JOB_START_TIME
   ,ROUND((TO_DATE(TO_CHAR(SYSDATE,'DD-MM-YYYY HH24:MI:SS'),'DD-MM-YYYY HH24:MI:SS')
   - TO_DATE(TO_CHAR(a.job_start_time,'DD-MM-YYYY HH24:MI:SS'),'DD-MM-YYYY HH24:MI:SS'))
   *1440) "DIFF"
   ,d.WAIT_TIME
   ,a.JOB_STATUS
   ,NVL(
   (SELECT STEP_STATUS FROM job_events WHERE (event_id,seq_no)=( SELECT MAX(event_id),MAX(seq_no)  FROM job_events WHERE event_id=
   (SELECT MAX(event_id) FROM job_events WHERE queue_id=a.QUEUE_ID))),0)  "STEP_STATUS"
    ,DECODE((SELECT NVL(STEP_STATUS,0) FROM job_events WHERE (event_id,seq_no)=( SELECT MAX(event_id),MAX(seq_no) FROM job_events WHERE event_id=
   (SELECT MAX(event_id) FROM job_events WHERE queue_id=a.QUEUE_ID))),1,'Running',2,'Succeeded',3,'Failed','NR') "STEP_STATUS_C"
  ,(SELECT MAX(event_id) FROM job_events WHERE queue_id=a.QUEUE_ID) "EVENT_D_C"
FROM    job_queue a
,job_dependency b
,job_events c
,job_master d
WHERE a.job_id=b.job_id(+) AND a.job_id=d.job_id AND a.queue_id=c.QUEUE_ID(+)  -- and a.job_status <=1
GROUP BY a.QUEUE_ID
   ,a.SEQ_ID
   ,a.JOB_ID
   ,a.JOB_START_TIME
       ,d.WAIT_TIME
   ,a.JOB_STATUS
ORDER BY queue_id
) WHERE JOB_STATUS <> STEP_STATUS OR job_status=0;

*********************************************************************

 

/* PROCESS_CONTROL-->Variable-->ETL_PROCESS_CNTL_START--> */ 

 SELECT NVL2(WM_CONCAT(SID||':'||SERIAL#), 

'PREVIUOS RUN OF PROCESS_CONTROL IS RUNNING FOR SID:SERIAL# '||WM_CONCAT(SID||':'||SERIAL#),

999) STATUS FROM GV$SESSION WHERE USERNAME='PROC_CNTL' AND STATUS<>'KILLED' 

 

/* PROCESS_CONTROL-->Procedure-->JOB_UPDATE_JOB_MASTER-->CHECK PREV LOAD */ 

 DECLARE

CNT  NUMBER(2);

ETL_REF NUMBER(10);

NEW_ERROR EXCEPTION;

BEGIN

SELECT ETL_REF INTO ETL_REF FROM ETL_REF;

SELECT COUNT(DISTINCT A.TABLE_TYPE) INTO CNT FROM ETL_SRCSTG_EXT_STATUS A, ETL_REF B WHERE A.ETL_REF=B.ETL_REF;

IF CNT = 2 THEN

DBMS_OUTPUT.PUT_LINE('PREV LOAD COMPLETED');

ELSE

RAISE NEW_ERROR;

END IF;

EXCEPTION

WHEN NEW_ERROR THEN

RAISE_APPLICATION_ERROR(-20001,'PREVIOUS LOAD FOR THE ETL_REF '||ETL_REF||' NOT COMPLETED');

END; 

 

/* PROCESS_CONTROL-->Procedure-->JOB_UPDATE_JOB_MASTER-->TRUNC_JOB_QUEUE */ 

 TRUNCATE TABLE  JOB_QUEUE 

 

/* PROCESS_CONTROL-->Procedure-->JOB_UPDATE_JOB_MASTER-->ETL_REF_INCREMENT */ 

 DECLARE

CUR_ETL NUMBER(10);

TOT NUMBER(2);

CNT NUMBER(1);

BEGIN

SELECT TO_NUMBER(TO_CHAR(SYSDATE-1,'YYYYMMDD'))  INTO CUR_ETL FROM DUAL;

DBMS_OUTPUT.PUT_LINE('Todays Etl Ref'||CUR_ETL);

SELECT COUNT(1) INTO TOT  FROM ETL_REF WHERE SUBSTR(ETL_REF,1,8)=CUR_ETL;

DBMS_OUTPUT.PUT_LINE('Available Values '||TOT);

SELECT COUNT(1) INTO CNT FROM ETL_REF;

CUR_ETL:=CUR_ETL||'00';

IF CNT =0 THEN

   INSERT INTO ETL_REF VALUES(CUR_ETL,SYSDATE);

   COMMIT; 

ELSE

  IF TOT = 0 THEN

    UPDATE ETL_REF SET ETL_REF=CUR_ETL,START_TIME=SYSDATE;

    COMMIT;

    ELSE

    DBMS_OUTPUT.PUT_LINE('Already etl_Ref exists');

    SELECT ETL_REF INTO CUR_ETL FROM ETL_REF;

    CUR_ETL:=CUR_ETL+1;

    DBMS_OUTPUT.PUT_LINE('The Next value will be '||CUR_ETL);

    UPDATE ETL_REF SET ETL_REF=CUR_ETL,START_TIME=SYSDATE;

    COMMIT;

    END IF;

END IF;    

    INSERT INTO ETL_REF_HIS_VALUE 

    SELECT A.ETL_REF,SYSTIMESTAMP FROM ETL_REF A WHERE NOT EXISTS ( SELECT 'X' FROM ETL_REF_HIS_VALUE B WHERE A.ETL_REF=B.ETL_REF);COMMIT;

END; 

 

/* PROCESS_CONTROL-->Procedure-->JOB_UPDATE_JOB_MASTER-->UPDATE JOB MASTER */ 

 BEGIN

UPDATE JOB_MASTER SET JOB_RUN_TIME=SYSDATE+1/(24*60) WHERE JOB_ACTIVE=1;

COMMIT;

END; 

 

/* PROCESS_CONTROL-->Procedure-->JOB_UPDATE_JOB_MASTER-->TRUNC_JOB_EVENTS */ 

 TRUNCATE TABLE JOB_EVENTS 

 

/* PROCESS_CONTROL-->Procedure-->PROCESS_ETL_REF_LOAD_PROC-->PROCESS_ETL_REF_LOAD_PROC */ 

 BEGIN

/*EXECUTE IMMEDIATE 'UPDATE ETL_REF SET ETL_REF=(SELECT ETL_REF+1 FROM ETL_REF),START_TIME=SYSTIMESTAMP';  TO BE INCLUDEDE LATER */ 

EXECUTE IMMEDIATE 'TRUNCATE TABLE ETL_REF_LOAD';

EXECUTE IMMEDIATE 'INSERT INTO ETL_REF_LOAD SELECT B.SCENARIO_NAME,B.TABLE_NAME,B.CAT_VAL,A.ETL_REF ETL_REF,D.START_TIME FROM ETL_REF_HIS_VALUE A,

        ETL_DATE B, ETL_DATE C, ETL_REF D

  WHERE 

      B.TABLE_NAME=C.TABLE_NAME

      AND B.SCENARIO_NAME=C.SCENARIO_NAME

      AND NVL(B.CAT_VAL,'||'''1'''||')=NVL(C.CAT_VAL,'||'''1'''||')

      AND A.ETL_REF > NVL(B.ETL_REF,0)

      AND B.SCENARIO_NAME NOT LIKE '||'''SRCSTG%'''||'

      ORDER BY C.TABLE_NAME, C.SCENARIO_NAME, C.CAT_VAL,A.ETL_REF';

END; 

***********************************************************************

 

 

/* PROCESS_SCHEDULING-->Procedure-->PROCESS_SCHEDULING_PROC-->Process Scheduling */ 

 declare

CURSOR que_cur IS  

SELECT job_id,RUN_TIME,FLAG,job_run_interval FROM (

SELECT job_id,TO_DATE(TO_CHAR(SYSDATE,'dd/mm/yyyy')||TO_CHAR(JOB_RUN_TIME,' 

HH24:MI:ss'),'dd/mm/yyyy HH24:MI:ss') RUN_TIME,'X' FLAG,job_run_interval

FROM job_masterWHERE job_active=1  AND (

CASE job_run_day 

WHEN 1 THEN TO_DATE(TO_CHAR(SYSDATE,'dd/mm/yyyy')||TO_CHAR(JOB_RUN_TIME,' 

HH24:MI:ss'),'dd/mm/yyyy HH24:MI:ss')

WHEN 2 THEN TO_DATE(TO_CHAR(NEXT_DAY(SYSDATE,'SAT'),'dd/mm/yyyy')||TO_CHAR(JOB_RUN_TIME,' 

HH24:MI:ss'),'dd/mm/yyyy HH24:MI:ss')

WHEN 3 THEN TO_DATE(TO_CHAR(LAST_DAY(SYSDATE),'dd/mm/yyyy')||TO_CHAR(JOB_RUN_TIME,' 

HH24:MI:ss'),'dd/mm/yyyy HH24:MI:ss')

END )

BETWEEN SYSDATE  AND SYSDATE+(10/(24*60)) 

UNION

SELECT job_id,(CASE  WHENJOB_RUN_TIME >= LAST_SCHED_TIME THEN  JOB_RUN_TIME ELSE 

LAST_SCHED_TIME  END ) RUN_TIME,'I' FLAG,job_run_interval

FROM ( SELECT job_id,TO_DATE(TO_CHAR(SYSDATE,'dd/mm/yyyy')||TO_CHAR(JOB_RUN_TIME,' 

HH24:MI:ss'),'dd/mm/yyyy HH24:MI:ss') JOB_RUN_TIME,

NVL(LAST_SCHED_TIME,TO_DATE('01/01/1907','dd/mm/yyyy')) LAST_SCHED_TIME 

,job_run_interval

FROM job_master 

WHERE job_run_day=0 AND job_active=1)

WHERE ( CASE WHEN  JOB_RUN_TIME >= NVL(LAST_SCHED_TIME,TO_DATE('01/01/1907','dd/mm/yyyy')) 

THEN  JOB_RUN_TIME  ELSE LAST_SCHED_TIME END ) 

BETWEEN SYSDATE AND SYSDATE+(10/(24*60)));

QUEUE_JOB_ID INTEGER:=#NIC_APPLICATION.SCH_QUEUE_ID;

prc_job_id INTEGER:=#NIC_APPLICATION.SCH_JOB_ID;

queue_id_inc INTEGER;

seq_id_inc INTEGER;

src_count INTEGER :=0;

eve_id_inc integer;

no_of_job integer :=0;

err_flag integer :=0;

SESS_NO varchar2(60);

BEGIN

insert into job_events (EVENT_ID,SEQ_NO, QUEUE_ID, STEP_ID, LOG_TIME, STEP_NAME, STEP_DESC, STEP_ERROR, 

TOTAL_AFFECTED_ROWS, STEP_STATUS)values(96071565,(select nvl(max(seq_no),0)+1 from job_events where event_id=96071565) ,QUEUE_JOB_ID,10,systimestamp,'Job identification','Collect the job to put in job queue',null,null,1);

COMMIT;

SELECT NVL(MAX(queue_id),0),NVL(MAX(seq_id),0) INTO queue_id_inc,seq_id_inc FROM job_queue;

seq_id_inc:=seq_id_inc+1;

FOR cur IN que_cur LOOP

queue_id_inc:=queue_id_inc+1;

INSERT INTO job_queue (QUEUE_ID, SEQ_ID, JOB_ID, JOB_START_TIME, 

JOB_END_TIME, JOB_STATUS) 

VALUES (queue_id_inc,seq_id_inc,cur.job_id,cur.RUN_TIME,NULL,0);

no_of_job:=no_of_job+1;

IF cur.flag='I' AND CUR.job_run_interval>0 THEN

UPDATE job_master SET 

LAST_SCHED_TIME=cur.run_time+(cur.job_run_interval/(24*60)) WHERE job_id = cur.job_id ;

END IF;

END LOOP;

COMMIT;

insert into job_events (EVENT_ID,SEQ_NO, QUEUE_ID, STEP_ID, LOG_TIME, STEP_NAME, STEP_DESC, STEP_ERROR, 

TOTAL_AFFECTED_ROWS, STEP_STATUS)values(96071565,(select nvl(max(seq_no),0)+1 from job_events where event_id=96071565 ),QUEUE_JOB_ID,99,systimestamp,'Job Completed','Completed',null,null,2);

COMMIT;

END;

****************************************************************************

 

/* PROCESS_CONTROL-->Procedure-->PROCESS_CONTROL_PROC-->Ins into JOB_QUEUE */ 

 declare

sed_id_c integer:=0;

QUEUE_ID_C integer:=0;

begin

select nvl(max(seq_id),0) into sed_id_c from job_queue;

select nvl(max(queue_id),0) into QUEUE_ID_C from job_queue;

sed_id_c:=sed_id_c+1;

QUEUE_ID_C:=QUEUE_ID_C+1;

INSERT INTO JOB_QUEUE ( QUEUE_ID, SEQ_ID, JOB_ID, JOB_START_TIME, JOB_END_TIME,JOB_STATUS ) VALUES

(QUEUE_ID_C, sed_id_c, 1, SYSDATE, NULL, 0); 

COMMIT;

end;