add GUC schema var - what scheme in use now

Author: Vladimir Ershov

README.md

@@ -41,6 +41,11 @@ pgpro_scheduler это расширение PostgreSQL и не требует н
 * 	**schedule.database** - строковая переменная, указывает с какими базам может
 	работать планировщик. Что бы указать несколько баз, нужно перечислить их
 	имена через запятую. По умолчанию - пустая строка.
+*	**schedule.scheme** - строковая переменная, указывает в какой `scheme` 
+	находятся служебные таблицы планировщика. Для изменения требуется
+	перезагрузка. Обычно ее не надо менять. Может использоваться для работы 
+	на реплике, если используется foreign data wrapper. По умолчанию - 
+	schedule.
 *	**schedule.nodename** - строковая переменная, содержит название узла.
 	По умолчанию - master. Если расширение используется на одной машине,
 	то переменная не имеет смысла.

src/pgpro_scheduler.c

@@ -40,11 +40,12 @@ volatile sig_atomic_t got_sighup = false;
 volatile sig_atomic_t got_sigterm = false;
 
 /* Custom GUC variables */
-static char *scheduler_databases = "";
-static char *scheduler_nodename = "master";
-static char *scheduler_transaction_state = "undefined";
+static char *scheduler_databases = NULL;
+static char *scheduler_nodename = NULL;
+static char *scheduler_transaction_state = NULL;
 static int  scheduler_max_workers = 2;
 static bool scheduler_service_enabled = false;
+static char *scheduler_schema = NULL;
 /* Custom GUC done */
 
 extern void
@@ -398,7 +399,18 @@ pg_scheduler_startup(void)
 
 void _PG_init(void)
 {
-	RequestAddinShmemSpace(1000);
+	DefineCustomStringVariable(
+		"schedule.schema",
+		"The name of scheduler schema",
+		NULL,
+		&scheduler_schema,
+		"schedule",
+		PGC_POSTMASTER,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
 	DefineCustomStringVariable(
 		"schedule.database",
 		"On which databases scheduler could be run",

src/scheduler_job.c

@@ -25,7 +25,7 @@ job_t *get_jobs_to_do(char *nodename, int *n, int *is_error)
 	int ret, got, i;
 	Oid argtypes[1] = { TEXTOID };
 	Datum values[1];
-	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances, cron.executor, cron.next_time_statement from schedule.at at, schedule.cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1 order by at.start_at";
+	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances, cron.executor, cron.next_time_statement from at at, cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1 order by at.start_at";
 
 	*is_error = *n = 0;
 	START_SPI_SNAP();
@@ -68,7 +68,7 @@ job_t *get_expired_jobs(char *nodename, int *n, int *is_error)
 
 	*n = *is_error = 0;
 	initStringInfo(&sql);
-	appendStringInfo(&sql, "select start_at, last_start_available, cron, started, active from schedule.at where last_start_available < 'now' and not active and node = '%s'", nodename);
+	appendStringInfo(&sql, "select start_at, last_start_available, cron, started, active from at where last_start_available < 'now' and not active and node = '%s'", nodename);
 	ret = SPI_execute(sql.data, true, 0);
 	if(ret == SPI_OK_SELECT)
 	{
@@ -115,8 +115,8 @@ int move_job_to_log(job_t *j, bool status)
 	char  nulls[4] = { ' ', ' ', ' ', ' ' };	
 	Oid argtypes[4] = { BOOLOID, TEXTOID, INT4OID, TIMESTAMPTZOID };
 	int ret;
-	const char *del_sql = "delete from schedule.at where start_at = $1 and cron = $2";
-	const char *sql = "insert into schedule.log (start_at,  last_start_available, retry, cron, node, started, status, finished, message)  SELECT start_at, last_start_available, retry, cron, node, started, $1 as status, 'now'::timestamp as finished, $2 as message from schedule.at where cron = $3 and start_at = $4";
+	const char *del_sql = "delete from at where start_at = $1 and cron = $2";
+	const char *sql = "insert into log (start_at,  last_start_available, retry, cron, node, started, status, finished, message)  SELECT start_at, last_start_available, retry, cron, node, started, $1 as status, 'now'::timestamp as finished, $2 as message from at where cron = $3 and start_at = $4";
 
 	/* in perl was this at first $status = 0 if $job->{spoiled}; skip so far */
 

src/scheduler_manager.c

@@ -44,55 +44,51 @@ extern volatile sig_atomic_t got_sigterm;
 
 int checkSchedulerNamespace(void)
 {
-	const char *sql = "select count(*) from pg_namespace where nspname = 'schedule'";
-	int found  = 0;
-	int ret;
-	int ntup;
-	bool isnull;
+	const char *sql = "select count(*) from pg_catalog.pg_namespace where nspname = $1";
+	int count  = 0;
+	const char *schema;
+	Oid argtypes[1] = { TEXTOID };
+	Datum values[1];
 
-	pgstat_report_activity(STATE_RUNNING, "initialize: check namespace");
 	SetCurrentStatementStartTimestamp();
+	pgstat_report_activity(STATE_RUNNING, "initialize: check namespace");
+
+	schema = GetConfigOption("schedule.schema", false, true);
+
 	StartTransactionCommand();
 	SPI_connect();
 	PushActiveSnapshot(GetTransactionSnapshot());
 
-	ret = SPI_execute(sql, true, 0);
-	if(ret == SPI_OK_SELECT && SPI_processed == 1)
+	values[0] = CStringGetTextDatum(schema);
+	count = select_count_with_args(sql, 1, argtypes, values, NULL);
+
+	if(count == -1)
 	{
-		ntup = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0],
-					SPI_tuptable->tupdesc, 1, &isnull)
-		);
-		if(!isnull && ntup == 1)
-		{
-			found =  1;
-		}
-		else if(isnull)
-		{
-			elog(LOG, "Scheduler manager: %s: cannot check namespace: count return null",
-				MyBgworkerEntry->bgw_name);
-		}
-		else if(ntup > 1)
-		{
-			elog(LOG, "Scheduler manager: %s: cannot check namespace: found %d namespaces",
-				MyBgworkerEntry->bgw_name, ntup);
-		}
+		elog(ERROR, "Scheduler manager: %s: cannot check namespace: sql error",
+													MyBgworkerEntry->bgw_name); 
 	}
-	else if(ret != SPI_OK_SELECT)
+	else if(count > 1 || count == 0 )
 	{
-		elog(LOG, "Scheduler manager: %s: cannot check namespace: error code %d",
-				MyBgworkerEntry->bgw_name, ret);
+		elog(LOG, "Scheduler manager: %s: cannot check namespace: found %d namespaces",
+											MyBgworkerEntry->bgw_name, count);
 	}
-	else if(SPI_processed != 1)
+	else if(count == -2)
 	{
-		elog(LOG, "Scheduler manager: %s: cannot check namespace: count return %ud tups",
-				MyBgworkerEntry->bgw_name,
-				(unsigned)SPI_processed);
+		elog(LOG, "Scheduler manager: %s: cannot check namespace: count return null",
+											MyBgworkerEntry->bgw_name);
 	}
+	else if(count != 1)
+	{
+		elog(ERROR, "Scheduler manager: %s: cannot check namespace: unknown error %d",
+											MyBgworkerEntry->bgw_name, count); 
+	}
+
 	SPI_finish();
 	PopActiveSnapshot();
 	CommitTransactionCommand();
+	if(count) SetConfigOption("search_path", "schedule", PGC_USERSET, PGC_S_SESSION);
 
-	return found;
+	return count;
 }
 
 int get_scheduler_maxworkers(void)
@@ -269,7 +265,7 @@ scheduler_task_t *scheduler_get_active_tasks(scheduler_manager_ctx_t *ctx, int *
 
 	*nt = 0;
 	initStringInfo(&sql);
-	appendStringInfo(&sql, "select id, rule, postpone, _next_exec_time, next_time_statement from schedule.cron where active and not broken and (start_date <= 'now' or start_date is null) and (end_date <= 'now' or end_date is null) and node = '%s'", ctx->nodename);
+	appendStringInfo(&sql, "select id, rule, postpone, _next_exec_time, next_time_statement from cron where active and not broken and (start_date <= 'now' or start_date is null) and (end_date <= 'now' or end_date is null) and node = '%s'", ctx->nodename);
 
 	pgstat_report_activity(STATE_RUNNING, "select 'at' tasks");
 
@@ -590,7 +586,7 @@ int how_many_instances_on_work(scheduler_manager_ctx_t *ctx, int cron_id)
 int set_job_on_free_slot(scheduler_manager_ctx_t *ctx, job_t *job)
 {
 	scheduler_manager_slot_t *item;
-	const char *sql = "update schedule.at set started = 'now'::timestamp with time zone, active = true where cron = $1 and start_at = $2";
+	const char *sql = "update at set started = 'now'::timestamp with time zone, active = true where cron = $1 and start_at = $2";
 	Datum values[2];
 	Oid argtypes[2] = {INT4OID, TIMESTAMPTZOID};
 	int ret;
@@ -977,7 +973,7 @@ int mark_job_broken(scheduler_manager_ctx_t *ctx, int cron_id, char *reason)
 	Oid types[2] = { INT4OID, TEXTOID };
 	Datum values[2];
 	char *error;
-	char *sql = "update schedule.cron set reason = $2, broken = true where id = $1";
+	char *sql = "update cron set reason = $2, broken = true where id = $1";
 	int ret;
 
 	values[0] = Int32GetDatum(cron_id);
@@ -997,7 +993,7 @@ int update_cron_texttime(scheduler_manager_ctx_t *ctx, int cron_id, TimestampTz
 	bool nulls[2] = { ' ', ' ' };
 	char *error;
 	int ret;
-	char *sql = "update schedule.cron set _next_exec_time = $2 where id = $1";
+	char *sql = "update cron set _next_exec_time = $2 where id = $1";
 
 	values[0] = Int32GetDatum(cron_id);
 	if(next > 0)
@@ -1073,9 +1069,9 @@ int insert_at_record(char *nodename, int cron_id, TimestampTz start_at, Timestam
 	Datum values[4];
 	char  nulls[4] = { ' ', ' ', ' ', ' ' };
 	Oid argtypes[4];
-	char *insert_sql = "insert into schedule.at (start_at, last_start_available, node, retry, cron, active) values ($1, $2, $3, 0, $4, false)";
-	char *at_sql = "select count(start_at) from schedule.at where cron = $1 and start_at = $2";
-	char *log_sql = "select count(start_at) from schedule.log where cron = $1 and start_at = $2";
+	char *insert_sql = "insert into at (start_at, last_start_available, node, retry, cron, active) values ($1, $2, $3, 0, $4, false)";
+	char *at_sql = "select count(start_at) from at where cron = $1 and start_at = $2";
+	char *log_sql = "select count(start_at) from log where cron = $1 and start_at = $2";
 	int count, ret;
 
 	argtypes[0] = INT4OID;
@@ -1213,11 +1209,11 @@ void clean_at_table(scheduler_manager_ctx_t *ctx)
 	char *error = NULL;
 
 	START_SPI_SNAP();
-	if(execute_spi("truncate schedule.at", &error) < 0)
+	if(execute_spi("truncate at", &error) < 0)
 	{
 		manager_fatal_error(ctx, 0, "Cannot clean 'at' table: %s", error);
 	}
-	if(execute_spi("update schedule.cron set _next_exec_time = NULL where _next_exec_time is not NULL", &error) < 0)
+	if(execute_spi("update cron set _next_exec_time = NULL where _next_exec_time is not NULL", &error) < 0)
 	{
 		manager_fatal_error(ctx, 0, "Cannot clean cron _next time: %s", error);
 	}