all params checked && execute with params

Author: Vladimir Ershov

pgpro_scheduler--2.0.sql

@@ -195,6 +195,10 @@ BEGIN
 		last_avail := NULL;
 	END IF;
 
+	IF node IS NULL THEN
+		node := 'master';
+	END IF;
+
 	IF run_as IS NOT NULL AND run_as <> session_user THEN
 		executor := run_as;
 		BEGIN
@@ -211,10 +215,10 @@ BEGIN
 
 	INSERT INTO at_jobs_submitted
 		(node, at, do_sql, owner, executor, name, comments, max_run_time,
-		 postpone, last_start_available, depends_on)
+		 postpone, last_start_available, depends_on, params)
 	VALUES
 		(node, run_after, query, session_user, executor,  name, comments,
-		 max_duration, max_wait_interval, last_avail, depends_on)
+		 max_duration, max_wait_interval, last_avail, depends_on, params)
 	RETURNING id INTO job_id;
 
 	RETURN job_id;

src/scheduler_executor.c

@@ -150,7 +150,14 @@ void executor_worker_main(Datum arg)
 		{
 			START_SPI_SNAP();
 		}
-		ret = execute_spi(job->dosql[i], &error);
+		if(job->type == AtJob && i == 0 && job->sql_params_n > 0)
+		{
+			ret = execute_spi_params_prepared(job->dosql[i], job->sql_params_n, job->sql_params, &error);
+		}
+		else
+		{
+			ret = execute_spi(job->dosql[i], &error);
+		}
 		if(ret < 0)
 		{
 			/* success = false; */

src/scheduler_job.c

@@ -62,6 +62,7 @@ job_t *get_at_job(int cron_id, char *nodename, char **perror)
 		}
 
 		j = init_scheduler_job(NULL, AtJob);
+		j->cron_id = cron_id;
 		j->node = _copy_string(nodename);
 		j->dosql = get_textarray_from_spi(0, 2, &j->dosql_n);
 		j->executor = get_text_from_spi(0, 3);
@@ -163,7 +164,7 @@ job_t *_at_get_jobs_to_do(char *nodename, int *n, int *is_error, int limit)
 	int ret, got, i;
 	Oid argtypes[2] = { TEXTOID, INT4OID };
 	Datum values[2];
-	const char *get_job_sql = "select id, at, last_start_available, max_run_time, max_instances, executor from ONLY at_jobs_submitted where at <= 'now' and (last_start_available is NULL OR last_start_available > 'now') AND node = $1 order by at,  submit_time limit $2";
+	const char *get_job_sql = "select id, at, last_start_available, max_run_time,  executor from ONLY at_jobs_submitted where at <= 'now' and (last_start_available is NULL OR last_start_available > 'now') AND node = $1 order by at,  submit_time limit $2";
 
 	*is_error = *n = 0;
 	START_SPI_SNAP();
@@ -184,9 +185,8 @@ job_t *_at_get_jobs_to_do(char *nodename, int *n, int *is_error, int limit)
 				jobs[i].start_at = get_timestamp_from_spi(i, 2, 0);
 				jobs[i].last_start_avail = get_timestamp_from_spi(i, 3, 0);
 				jobs[i].timelimit = get_interval_seconds_from_spi(i, 4, 0);
-				jobs[i].max_instances = get_int_from_spi(i, 5, 1);
 				jobs[i].node = _copy_string(nodename);
-				jobs[i].executor = get_text_from_spi(i, 6);
+				jobs[i].executor = get_text_from_spi(i, 5);
 			}
 		}
 	}
@@ -323,19 +323,23 @@ job_t *set_job_error(job_t *j, const char *fmt, ...)
 	return j;
 }
 
-int move_job_to_log(job_t *j, bool status)
+int move_job_to_log(job_t *j, bool status, bool process)
 {
 	if(j->type == CronJob) _cron_move_job_to_log(j, status);
-	return _at_move_job_to_log(j, status);
+	return _at_move_job_to_log(j, status, process);
 }
 
-int _at_move_job_to_log(job_t *j, bool status)
+int _at_move_job_to_log(job_t *j, bool status, bool process)
 {
 	Datum values[3];	
 	char  nulls[3] = { ' ', ' ', ' ' };	
 	Oid argtypes[4] = { INT4OID, BOOLOID, TEXTOID };
 	int ret;
-	const char *sql = "WITH moved_rows AS (DELETE from ONLY at_jobs_process a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_done SELECT *, $2 as status, $3 as reason FROM moved_rows";
+	const char *sql_process = "WITH moved_rows AS (DELETE from ONLY at_jobs_process a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_done SELECT *, $2 as status, $3 as reason FROM moved_rows";
+	const char *sql_submitted = "WITH moved_rows AS (DELETE from ONLY at_jobs_submitted a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_done SELECT *, NULL as start_time, $2 as status, $3 as reason FROM moved_rows";
+	const char *sql;
+
+	sql = process ? sql_process: sql_submitted;
 
 	values[0] = Int32GetDatum(j->cron_id);
 	values[1] = BoolGetDatum(status);

src/scheduler_job.h

@@ -46,12 +46,12 @@ job_t *_cron_get_jobs_to_do(char *nodename, int *n, int *is_error, int limit);
 job_t *_at_get_jobs_to_do(char *nodename, int *n, int *is_error, int limit);
 job_t *get_jobs_to_do(char *nodename, task_type_t type, int *n, int *is_error, int limit);
 job_t *set_job_error(job_t *j, const char *fmt, ...) pg_attribute_printf(2, 3);
-int move_job_to_log(job_t *j, bool status);
+int move_job_to_log(job_t *j, bool status, bool processed);
 void destroy_job(job_t *j, int selfdestroy);
 job_t *get_at_job(int cron_id, char *nodename, char **perror);
 job_t *get_cron_job(int cron_id, TimestampTz start_at, char *nodename, char **perror);
 int _cron_move_job_to_log(job_t *j, bool status);
-int _at_move_job_to_log(job_t *j, bool status);
+int _at_move_job_to_log(job_t *j, bool status, bool processed);
 
 #endif
 

src/scheduler_manager.c

@@ -840,12 +840,13 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 
 		for(i = start_i; i < N + start_i; i++)
 		{
-			ni = how_many_instances_on_work(ctx, &(jobs[i]));
-			if(ni >= jobs[i].max_instances)
+			ni = type == CronJob ?
+				how_many_instances_on_work(ctx, &(jobs[i])): 100000;
+			if(type == CronJob && ni >= jobs[i].max_instances)
 			{
 				START_SPI_SNAP();
 				set_job_error(&jobs[i], "max instances limit reached");
-				move_job_to_log(&jobs[i], false);
+				move_job_to_log(&jobs[i], false, false);
 				destroy_job(&jobs[i], 0);
 				STOP_SPI_SNAP();
 				jobs[i].cron_id = -1;
@@ -867,9 +868,11 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 						set_job_error(&jobs[i],
 								"Cannot set at job %d to worker",
 								 			jobs[i].cron_id);
+						elog(ERROR, "Cannot set job to free slot type=%d, id=%d", 
+									jobs[i].type, jobs[i].cron_id);
 					}
 					START_SPI_SNAP();
-					move_job_to_log(&jobs[i], false);
+					move_job_to_log(&jobs[i], false, false);
 					destroy_job(&jobs[i], 0);
 					jobs[i].cron_id = -1;
 					STOP_SPI_SNAP();
@@ -1063,7 +1066,7 @@ int scheduler_check_slots(scheduler_manager_ctx_t *ctx, scheduler_manager_pool_t
 						}
 					}
 				}
-				move_job_to_log(item->job, job_status);
+				move_job_to_log(item->job, job_status, true);
 				STOP_SPI_SNAP();
 
 				last  = p->len - p->free - 1;
@@ -1179,7 +1182,7 @@ int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type
 				set_job_error(&expired[i], "job start time %s expired", ts);
 			}
 
-			move_ret  = move_job_to_log(&expired[i], 0);
+			move_ret  = move_job_to_log(&expired[i], 0, false);
 			if(move_ret < 0)
 			{
 				elog(LOG, "Scheduler manager %s: cannot move %s job %d@%s%s to log",

src/scheduler_spi_utils.c

@@ -351,4 +351,87 @@ int execute_spi(const char *sql, char **error)
 	return execute_spi_sql_with_args(sql, 0, NULL, NULL, NULL, error);
 }
 
+int execute_spi_params_prepared(const char *sql, int nparams, char **params, char **error)
+{
+	int ret = -100;
+	ErrorData *edata;
+	MemoryContext old;
+	int errorSet = 0;
+	char other[100];
+	SPIPlanPtr plan;
+	Oid *paramtypes;
+	Datum *values;
+	int i;
+
+	*error = NULL;
+
+	paramtypes = worker_alloc(sizeof(Oid) * nparams);
+	values = worker_alloc(sizeof(Datum) * nparams);
+	for(i=0; i < nparams; i++)
+	{
+		paramtypes[i] = TEXTOID;
+		values[i] = CStringGetTextDatum(params[i]);
+	}
+
+	PG_TRY();
+	{
+		plan = SPI_prepare(sql, nparams, paramtypes);
+		if(plan)
+		{
+			ret = SPI_execute_plan(plan, values, NULL, false, 0);
+		}
+	}
+	PG_CATCH();
+	{
+		old = switch_to_worker_context();
+
+		edata = CopyErrorData();
+		if(edata->message)
+		{
+			*error = _copy_string(edata->message);
+		}
+		else if(edata->detail)
+		{
+			*error = _copy_string(edata->detail);
+		}
+		else
+		{
+			*error = _copy_string("unknown error");
+		}
+		errorSet = 1;
+		FreeErrorData(edata);
+		MemoryContextSwitchTo(old);
+		FlushErrorState();
+	}
+	PG_END_TRY();
+
+	pfree(values);
+	pfree(paramtypes);
+
+	if(!errorSet && ret < 0)
+	{
+		if(ret == SPI_ERROR_CONNECT)
+		{
+			*error = _copy_string("Connection error");
+		}
+		else if(ret == SPI_ERROR_COPY)
+		{
+			*error = _copy_string("COPY error");
+		}
+		else if(ret == SPI_ERROR_OPUNKNOWN)
+		{
+			*error = _copy_string("SPI_ERROR_OPUNKNOWN");
+		}
+		else if(ret == SPI_ERROR_UNCONNECTED)
+		{
+			*error = _copy_string("Unconnected call");
+		}
+		else
+		{
+			sprintf(other, "error number: %d", ret);
+			*error = _copy_string(other);
+		}
+	}
 
+	return ret;
+}

src/scheduler_spi_utils.h

@@ -29,5 +29,6 @@ char *get_text_from_spi(int row_n, int pos);
 Oid get_oid_from_spi(int row_n, int pos, Oid def);
 int execute_spi_sql_with_args(const char *sql, int n, Oid *argtypes, Datum *values, char *nulls, char **error);
 int execute_spi(const char *sql, char **error);
+int execute_spi_params_prepared(const char *sql, int nparams, char **params, char **error);
 
 #endif