Oracle Experience: Passing Parameters to Scheduler Programs in Chains

One of my previous post(Parallel Processing with Oracle Scheduler Chains), i have given some information about Scheduler Chains. As i mentioned in the post, it is not possible to set Scheduler Programs Arguments in Scheduler Chains on-the-fly. There is a restriction of Oracle that does not allow you to pass parameters via a pre-step before program execution by using DBMS_SCHEDULER.set_job_argument_value routine. If you do this, you will get “ORA-27457”. I will show how to get rid of tihs error with a simple workaround. The workaround is using DBMS_ALERT package. A communication channel will be established between scheduler programs(sessions) in order send/receive messages. Following demonstration will show you, how to serialize scheduler programs.

I will use following scenairo:

Business Scenario : We are in the telco sector and we are implementing something that should be run before montly billing process. Our telco operator has over 30M subs. So, we need to implement the something enough fast…

Implementaion : We have one process that should be run first in order to find the subscirbers who we are interested in. After the subs have been found, we can run some business rules over them.

Design : We will create a table(CUTOFF_INTERIM_TABLE) in order to hold interested-in-subscribers. A simple routine(FillInterimTable procedure) will fill this table. Then we will use a routine(ProcessInterimTable procedure) to get the subscribers and do something. First procedure(FillInterimTable) should take an input parameter(pin_TotalProcessCount) and use a simple MODULO operation(MOD(i, pin_TotalProcessCount) + 1) to seperate the subscirbers into chunks. Second procedure(ProcessInterimTable) can be run (pin_TotalProcessCount) times simultaneously, and every process will take only their chunks(…WHERE PROCESS_NUMBER = pin_ProcessNumber…). The important thing is that, second procedures should run if and only if the first procedure succeeded. Otherwise there is no need to run second processes because of there is no subscribers to be processed. Both of these two routines takes an additional input parameter(pid_Cutoff) that shows billing process cutoff date. Of course every billing time, this cutoff date will change. So, we need this parameter somehow configurable by passing parameters. As i said before, it is not possible to set job argument values on-the-fly. Because of this restriction, we will do some tricks . We will create an additional routine(SetJobCutoffParameters procedure) in order to set these cutoff values by communicating between sessions. What we will do in this procedure is, we will send this cutoff parameter from this routine(DBMS_ALERT package is used for this manner). In order to communicate between sessions, every session should register themselves to DBMS_ALERT package. FillInterimTable and ProcessInterimTable procedures should be started (for registering) before SetJobCutoffParameters procedure. And these two procedures should wait in a predefined interval to get the cutoff message. After FillInterimTable procedure receives the cutoff parameter, it will start to fill interim table. But ProcessInterimTable procedure will not start when it receives cutoff message. It will also wait FillInterimTable procedure to finish its job. Again, we will send a message from FillInterimTable to ProcessInterimTable procedure to start its job. We will use 5 parallel processes.

CREATE TABLE CUTOFF_INTERIM_TABLE

(

CUTOFF DATE,

CO_ID NUMBER,

STATUS VARCHAR2(1),

CREATE_DATE DATE,

REMARK VARCHAR2(1024),

PROCESS_NUMBER NUMBER

);

CREATE TABLE CUTOFF_PROCESS_EXECUTION_LOG

(

CUTOFF DATE,

PROCESS_NUMBER NUMBER,

LOG_MESSAGE VARCHAR2(1024),

LOG_TIMESTAMP TIMESTAMP

);

--create proceduress

CREATE OR REPLACE PROCEDURE CreateProcessExecutionLog

(

pid_Cutoff IN CUTOFF_PROCESS_EXECUTION_LOG.CUTOFF%TYPE,

pin_ProcessNumber IN CUTOFF_PROCESS_EXECUTION_LOG.PROCESS_NUMBER%TYPE,

pis_LogMessage IN CUTOFF_PROCESS_EXECUTION_LOG.LOG_MESSAGE%TYPE

) IS

PRAGMA AUTONOMOUS_TRANSACTION;

BEGIN

INSERT INTO CUTOFF_PROCESS_EXECUTION_LOG

(CUTOFF, PROCESS_NUMBER, LOG_MESSAGE, LOG_TIMESTAMP)

VALUES

(pid_Cutoff, pin_ProcessNumber, pis_LogMessage, SYSTIMESTAMP);

COMMIT;

END CreateProcessExecutionLog;

CREATE OR REPLACE PROCEDURE FillInterimTable

(

pid_Cutoff IN CUTOFF_INTERIM_TABLE.CUTOFF%TYPE,

pin_TotalProcessCount IN NUMBER

) IS

vn_ProcessIndependent CUTOFF_INTERIM_TABLE.PROCESS_NUMBER%TYPE;

vd_Cutoff CUTOFF_INTERIM_TABLE.CUTOFF%TYPE;

vs_AlertMessage_SetParams VARCHAR2(100);

vs_AlertMessage_FillTable VARCHAR2(100);

vs_AlertStatus PLS_INTEGER;

vs_AlertName_SetParams VARCHAR2(100);

vs_AlertName_FillTable VARCHAR2(100);

vs_LogMessage CUTOFF_PROCESS_EXECUTION_LOG.LOG_MESSAGE%TYPE;

vn_AlertTimeout_SetParams PLS_INTEGER;

BEGIN

vn_ProcessIndependent := 0;

vs_AlertName_SetParams := 'CUTOFF__FILL_PRG__PARAM1';

vs_AlertName_FillTable := 'CUTOFF__FILL_PRG__FINISHED';

vn_AlertTimeout_SetParams := 30;

vs_AlertMessage_FillTable := 'FINISHED';

vs_LogMessage := 'FillInterimTable started. Total process count : ' || pin_TotalProcessCount;

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent, vs_LogMessage);

IF pid_Cutoff IS NULL THEN

vs_LogMessage := 'pid_Cutoff parameter is NULL.';

CreateProcessExecutionLog(NULL, vn_ProcessIndependent,vs_LogMessage );

vs_LogMessage := 'Registering DBMS_ALERT to be informed about ' || vs_AlertName_SetParams || ' message...';

CreateProcessExecutionLog(NULL, vn_ProcessIndependent,vs_LogMessage );

DBMS_ALERT.REGISTER(vs_AlertName_SetParams);

vs_LogMessage := 'Waiting for DBMS_ALERT ' || vs_AlertName_SetParams || ' message...';

CreateProcessExecutionLog(NULL, vn_ProcessIndependent,vs_LogMessage );

DBMS_ALERT.WAITONE(vs_AlertName_SetParams, vs_AlertMessage_SetParams, vs_AlertStatus, vn_AlertTimeout_SetParams);

IF vs_AlertStatus = 0 THEN

vd_Cutoff := TO_DATE(vs_AlertMessage_SetParams, 'DD.MM.YYYY');

vs_LogMessage := vs_AlertName_SetParams || ' is taken from alert message:'|| vs_AlertMessage_SetParams;

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent,vs_LogMessage );

ELSE

vs_LogMessage := 'Unable to get ' || vs_AlertName_SetParams || ' alert message:' || vs_AlertStatus;

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent, vs_LogMessage);

RAISE_APPLICATION_ERROR( -20011, vs_LogMessage );

END IF;

vs_LogMessage := 'Removing DBMS_ALERT for ' || vs_AlertName_SetParams;

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent,vs_LogMessage );

DBMS_ALERT.remove(vs_AlertName_SetParams);

ELSE

vd_Cutoff := pid_Cutoff;

END IF;

--some business checks etc

---

----

EXECUTE IMMEDIATE 'TRUNCATE TABLE CUTOFF_INTERIM_TABLE';

FOR i IN 1 .. 100000 LOOP

INSERT INTO CUTOFF_INTERIM_TABLE

(CUTOFF, CO_ID, STATUS, CREATE_DATE, REMARK, PROCESS_NUMBER )

VALUES

(vd_Cutoff, i, 'N', SYSDATE, 'Successfully created.', MOD(i, pin_TotalProcessCount) + 1);

END LOOP;

COMMIT;

vs_LogMessage := 'CUTOFF_INTERIM_TABLE insertion is finished.';

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent, vs_LogMessage);

IF NOT( pin_TotalProcessCount = 1 ) THEN

vs_LogMessage := 'Signalling DBMS_ALERT for ' || vs_AlertName_FillTable;

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent, vs_LogMessage);

DBMS_ALERT.SIGNAL(vs_AlertName_FillTable, vs_AlertMessage_FillTable);

COMMIT;

END IF;

vs_LogMessage := 'FillInterimTable finished...';

CreateProcessExecutionLog(vd_Cutoff, vn_ProcessIndependent, vs_LogMessage);

COMMIT;

END FillInterimTable;

CREATE OR REPLACE PROCEDURE ProcessInterimTable

(

pid_Cutoff IN CUTOFF_INTERIM_TABLE.CUTOFF%TYPE,

pin_ProcessNumber IN CUTOFF_INTERIM_TABLE.PROCESS_NUMBER%TYPE,

pin_TotalProcessCount IN NUMBER

) IS

vd_Cutoff CUTOFF_INTERIM_TABLE.CUTOFF%TYPE;

vs_AlertMessage_SetParams VARCHAR2(100);

vs_AlertMessage_FillTable VARCHAR2(100);

vs_AlertStatus PLS_INTEGER;

vs_AlertName_SetParams VARCHAR2(100);

vs_AlertName_FillTable VARCHAR2(100);

vs_LogMessage CUTOFF_PROCESS_EXECUTION_LOG.LOG_MESSAGE%TYPE;

vn_AlertTimeout_SetParams PLS_INTEGER;

vn_AlertTimeout_FillTable PLS_INTEGER;

BEGIN

vs_LogMessage := 'ProcessInterimTable started for process ' || pin_ProcessNumber;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber, vs_LogMessage );

vs_AlertName_SetParams := 'CUTOFF__EXEC' || pin_ProcessNumber || '__PARAM1';

vs_AlertName_FillTable := 'CUTOFF__FILL_PRG__FINISHED';

vn_AlertTimeout_SetParams := 30;

vn_AlertTimeout_FillTable := 60 * 60 * 4;--4 hours

IF NOT(pin_TotalProcessCount = 1 AND pin_ProcessNumber = 1) THEN

vs_LogMessage := 'Registering DBMS_ALERT to be informed about ' || vs_AlertName_FillTable || ' message...';

CreateProcessExecutionLog(NULL, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.REGISTER(vs_AlertName_FillTable);

END IF;

IF pid_Cutoff IS NULL THEN

vs_LogMessage := 'pid_Cutoff parameter is NULL.';

CreateProcessExecutionLog(NULL, pin_ProcessNumber,vs_LogMessage );

vs_LogMessage := 'Registering DBMS_ALERT to be informed about ' || vs_AlertName_SetParams || ' message...';

CreateProcessExecutionLog(NULL, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.REGISTER(vs_AlertName_SetParams);

vs_LogMessage := 'Waiting for DBMS_ALERT ' || vs_AlertName_SetParams || ' message...';

CreateProcessExecutionLog(NULL, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.WAITONE(vs_AlertName_SetParams, vs_AlertMessage_SetParams, vs_AlertStatus, vn_AlertTimeout_SetParams);

IF vs_AlertStatus = 0 THEN

vd_Cutoff := TO_DATE(vs_AlertMessage_SetParams, 'DD.MM.YYYY');

vs_LogMessage := vs_AlertName_SetParams || ' is taken from alert message:'|| vs_AlertMessage_SetParams;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber, vs_LogMessage);

ELSE

vs_LogMessage := 'Unable to get ' || vs_AlertName_SetParams || ' from alert message:' || vs_AlertStatus;

CreateProcessExecutionLog(NULL, pin_ProcessNumber, vs_LogMessage);

RAISE_APPLICATION_ERROR( -20011, vs_LogMessage );

END IF;

vs_LogMessage := 'Removing DBMS_ALERT for ' || vs_AlertName_SetParams;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.remove(vs_AlertName_SetParams);

ELSE

vd_Cutoff := pid_Cutoff;

END IF;

IF pin_TotalProcessCount = 1 AND pin_ProcessNumber = 1 THEN

-- there is only one process, no need parallel processing

FillInterimTable(vd_Cutoff, pin_TotalProcessCount);

ELSE

vs_LogMessage := 'Waiting for DBMS_ALERT ' || vs_AlertName_FillTable || ' message in order to start processing...';

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.WAITONE(vs_AlertName_FillTable, vs_AlertMessage_FillTable, vs_AlertStatus, vn_AlertTimeout_FillTable);

IF vs_AlertStatus = 0 THEN

vs_LogMessage := vs_AlertName_FillTable || ' is taken from alert message:'|| vs_AlertMessage_FillTable;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber, vs_LogMessage);

ELSE

vs_LogMessage := 'Unable to get ' || vs_AlertName_FillTable || ' from alert message:' || vs_AlertStatus;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber, vs_LogMessage);

RAISE_APPLICATION_ERROR( -20012, vs_LogMessage );

END IF;

vs_LogMessage := 'Removing DBMS_ALERT for ' || vs_AlertName_FillTable;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber,vs_LogMessage );

DBMS_ALERT.remove(vs_AlertName_FillTable);

END IF;

FOR rec_InterimTable IN (

SELECT T.CO_ID, T.REMARK, T.ROWID AS RD

FROM CUTOFF_INTERIM_TABLE T

WHERE STATUS = 'N'

AND PROCESS_NUMBER = pin_ProcessNumber

) LOOP

--some business processes

---

----

UPDATE CUTOFF_INTERIM_TABLE

SET STATUS = 'S',

REMARK = 'Successfully processed..'

WHERE ROWID = rec_InterimTable.RD;

COMMIT;

END LOOP;

vs_LogMessage := 'ProcessInterimTable finished for process ' || pin_ProcessNumber;

CreateProcessExecutionLog(vd_Cutoff, pin_ProcessNumber, vs_LogMessage );

END ProcessInterimTable;

CREATE OR REPLACE PROCEDURE SetJobCutoffParameters IS

vs_CutoffString VARCHAR2(100);

vd_Cutoff DATE;

vs_LogMessage CUTOFF_PROCESS_EXECUTION_LOG.LOG_MESSAGE%TYPE;

vs_AlertName VARCHAR2(100);

BEGIN

vs_CutoffString := '20.09.2011';

vd_Cutoff := TO_DATE(vs_CutoffString, 'DD.MM.YYYY');

vs_LogMessage := 'SetJobCutoffParameters started for cutoff ' || vs_CutoffString;

CreateProcessExecutionLog(vd_Cutoff, -1, vs_LogMessage);

vs_AlertName := 'CUTOFF__FILL_PRG__PARAM1';

vs_LogMessage := 'Signalling DBMS_ALERT for ' || vs_AlertName;

CreateProcessExecutionLog(vd_Cutoff, -1, vs_LogMessage);

DBMS_ALERT.SIGNAL(vs_AlertName, vs_CutoffString);

COMMIT;

FOR vn_ProcessNumber IN 1 .. 5 LOOP

vs_AlertName := 'CUTOFF__EXEC' || vn_ProcessNumber || '__PARAM1';

vs_LogMessage := 'Signalling DBMS_ALERT for ' || vs_AlertName;

CreateProcessExecutionLog(vd_Cutoff, -1, vs_LogMessage);

DBMS_ALERT.SIGNAL( vs_AlertName, vs_CutoffString);

COMMIT;

END LOOP;

vs_LogMessage := 'SetJobCutoffParameters finished for cutoff ' || vs_CutoffString;

CreateProcessExecutionLog(vd_Cutoff, -1, vs_LogMessage);

COMMIT;

END SetJobCutoffParameters;

DECLARE

vn_TotalProcessCount PLS_INTEGER;

BEGIN

vn_TotalProcessCount := 5;

DBMS_SCHEDULER.create_program(program_name => 'CUTOFF__FILL_PRG',

program_type => 'STORED_PROCEDURE',

program_action => 'FillInterimTable',

number_of_arguments => 2,

enabled => FALSE,

comments => 'Program for FillInterimTable(5 parallel processes)');

DBMS_SCHEDULER.define_program_argument(program_name => 'CUTOFF__FILL_PRG',

argument_position => 1,

argument_type => 'DATE',

default_value => NULL

);

DBMS_SCHEDULER.define_program_argument(program_name => 'CUTOFF__FILL_PRG',

argument_position => 2,

argument_type => 'NUMBER',

default_value => vn_TotalProcessCount);

DBMS_SCHEDULER.create_job(job_name => 'CUTOFF__FILL_JOB',

program_name => 'CUTOFF__FILL_PRG',

start_date => NULL,

enabled => FALSE,

auto_drop => FALSE,

comments => 'Job for FillInterimTable(5 parallel processes)');

END;

BEGIN

DBMS_SCHEDULER.enable(name => 'CUTOFF__FILL_PRG' );

DBMS_SCHEDULER.enable(name => 'CUTOFF__FILL_JOB' );

END;

DECLARE

vn_TotalProcessCount PLS_INTEGER;

BEGIN

vn_TotalProcessCount := 5;

FOR vn_ProcessNumber IN 1..vn_TotalProcessCount LOOP

DBMS_SCHEDULER.create_program(program_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG',

program_type => 'STORED_PROCEDURE',

program_action => 'ProcessInterimTable',

number_of_arguments => 3,

enabled => FALSE,

comments => 'Program ProcessInterimTable ' || vn_ProcessNumber || ' of 5 process');

DBMS_SCHEDULER.define_program_argument(program_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG',

argument_position => 1,

argument_type => 'DATE',

default_value => NULL);

DBMS_SCHEDULER.define_program_argument(program_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG',

argument_position => 2,

argument_type => 'NUMBER',

default_value => vn_ProcessNumber);

DBMS_SCHEDULER.define_program_argument(program_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG',

argument_position => 3,

argument_type => 'NUMBER',

default_value => vn_TotalProcessCount);

DBMS_SCHEDULER.create_job(job_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_JOB',

program_name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG',

start_date => NULL,

enabled => FALSE,

auto_drop => FALSE,

comments => 'Job for ProcessInterimTable ' || vn_ProcessNumber || ' of 5 process');

END LOOP;

END;

DECLARE

vn_TotalProcessCount PLS_INTEGER;

BEGIN

vn_TotalProcessCount := 5;

FOR vn_ProcessNumber IN 1..vn_TotalProcessCount LOOP

DBMS_SCHEDULER.enable(name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_JOB');

DBMS_SCHEDULER.enable(name => 'CUTOFF__EXEC' || vn_ProcessNumber || '_PRG');

END LOOP;

END;

BEGIN

DBMS_SCHEDULER.create_program(program_name => 'CUTOFF__SETPARAMS_PRG',

program_type => 'STORED_PROCEDURE',

program_action => 'SetJobCutoffParameters',

enabled => TRUE,

comments => 'Program for SetJobCutoffParameters');

DBMS_SCHEDULER.create_job(job_name => 'CUTOFF__SETPARAMS_JOB',

program_name => 'CUTOFF__SETPARAMS_PRG',

start_date => NULL,

enabled => TRUE,

auto_drop => FALSE,

comments => 'Job for SetJobCutoffParameters');

END;

BEGIN

DBMS_SCHEDULER.create_chain(chain_name => 'CUTOFF__CHAIN',

rule_set_name => NULL,

evaluation_interval => NULL,

comments => 'Chain for CUTOFF( 5 processes)');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'SET_PARAMS_STEP',

program_name => 'CUTOFF__SETPARAMS_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'FILL_INTERIM_TABLE_STEP',

program_name => 'CUTOFF__FILL_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'EXECUTE_1OF5_STEP',

program_name => 'CUTOFF__EXEC1_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'EXECUTE_2OF5_STEP',

program_name => 'CUTOFF__EXEC2_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'EXECUTE_3OF5_STEP',

program_name => 'CUTOFF__EXEC3_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'EXECUTE_4OF5_STEP',

program_name => 'CUTOFF__EXEC4_PRG');

DBMS_SCHEDULER.define_chain_step(chain_name => 'CUTOFF__CHAIN',

step_name => 'EXECUTE_5OF5_STEP',

program_name => 'CUTOFF__EXEC5_PRG');

END;

BEGIN