91 elog(
FATAL,
"SPI_execute failed: error code %d", ret);
107 "CREATE SCHEMA \"%s\" "
108 "CREATE TABLE \"%s\" ("
109 " type text CHECK (type IN ('total', 'delta')), "
111 "CREATE UNIQUE INDEX \"%s_unique_total\" ON \"%s\" (type) "
112 "WHERE type = 'total'",
122 elog(
FATAL,
"failed to create my schema");
157 elog(
LOG,
"%s initialized with %s.%s",
173 "WITH deleted AS (DELETE "
175 "WHERE type = 'delta' RETURNING value), "
176 "total AS (SELECT coalesce(sum(value), 0) as sum "
179 "SET value = %s.value + total.sum "
180 "FROM total WHERE type = 'total' "
181 "RETURNING %s.value",
249 elog(
FATAL,
"cannot select from table %s.%s: error code %d",
261 elog(
LOG,
"%s: count in %s.%s is now %d",
298 "Duration between each check (in seconds).",
311 "Database to connect to.",
323 "Number of workers.",
338 memset(&worker, 0,
sizeof(worker));
372 memset(&worker, 0,
sizeof(worker));
392 (
errcode(ERRCODE_INSUFFICIENT_RESOURCES),
393 errmsg(
"could not start background process"),
394 errhint(
"More details may be available in the server log.")));
397 (
errcode(ERRCODE_INSUFFICIENT_RESOURCES),
398 errmsg(
"cannot start background processes without postmaster"),
399 errhint(
"Kill all remaining database processes and restart the database.")));
void pgstat_report_activity(BackendState state, const char *cmd_str)
void RegisterBackgroundWorker(BackgroundWorker *worker)
BgwHandleStatus WaitForBackgroundWorkerStartup(BackgroundWorkerHandle *handle, pid_t *pidp)
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
#define BGW_NEVER_RESTART
@ BgWorkerStart_RecoveryFinished
#define BGWORKER_BACKEND_DATABASE_CONNECTION
#define BGWORKER_SHMEM_ACCESS
#define pg_attribute_noreturn()
elog(ERROR, "%s: %s", p2, msg)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
#define PG_RETURN_INT32(x)
#define PG_GETARG_INT32(n)
void DefineCustomStringVariable(const char *name, const char *short_desc, const char *long_desc, char **valueAddr, const char *bootValue, GucContext context, int flags, GucStringCheckHook check_hook, GucStringAssignHook assign_hook, GucShowHook show_hook)
void MarkGUCPrefixReserved(const char *className)
void DefineCustomIntVariable(const char *name, const char *short_desc, const char *long_desc, int *valueAddr, int bootValue, int minValue, int maxValue, GucContext context, int flags, GucIntCheckHook check_hook, GucIntAssignHook assign_hook, GucShowHook show_hook)
void ProcessConfigFile(GucContext context)
volatile sig_atomic_t ConfigReloadPending
void SignalHandlerForConfigReload(SIGNAL_ARGS)
void ResetLatch(Latch *latch)
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
#define WL_EXIT_ON_PM_DEATH
Assert(fmt[strlen(fmt) - 1] !='\n')
char * pstrdup(const char *in)
#define CHECK_FOR_INTERRUPTS()
bool process_shared_preload_libraries_in_progress
long pgstat_report_stat(bool force)
pqsigfunc pqsignal(int signo, pqsigfunc func)
const char * debug_query_string
static int64 DatumGetInt64(Datum X)
static Datum Int32GetDatum(int32 X)
static int32 DatumGetInt32(Datum X)
void BackgroundWorkerInitializeConnection(const char *dbname, const char *username, uint32 flags)
void BackgroundWorkerUnblockSignals(void)
BackgroundWorker * MyBgworkerEntry
const char * quote_identifier(const char *ident)
Snapshot GetTransactionSnapshot(void)
void PushActiveSnapshot(Snapshot snapshot)
void PopActiveSnapshot(void)
SPITupleTable * SPI_tuptable
int SPI_execute(const char *src, bool read_only, long tcount)
Datum SPI_getbinval(HeapTuple tuple, TupleDesc tupdesc, int fnumber, bool *isnull)
#define SPI_OK_UPDATE_RETURNING
void resetStringInfo(StringInfo str)
void appendStringInfo(StringInfo str, const char *fmt,...)
void initStringInfo(StringInfo str)
char bgw_function_name[BGW_MAXLEN]
char bgw_name[BGW_MAXLEN]
char bgw_type[BGW_MAXLEN]
BgWorkerStartTime bgw_start_time
char bgw_library_name[MAXPGPATH]
uint32 WaitEventExtensionNew(const char *wait_event_name)
static int worker_spi_naptime
static void initialize_worker_spi(worktable *table)
PGDLLEXPORT void worker_spi_main(Datum main_arg) pg_attribute_noreturn()
Datum worker_spi_launch(PG_FUNCTION_ARGS)
static uint32 worker_spi_wait_event_main
static char * worker_spi_database
static int worker_spi_total_workers
struct worktable worktable
PG_FUNCTION_INFO_V1(worker_spi_launch)
void StartTransactionCommand(void)
void SetCurrentStatementStartTimestamp(void)
void CommitTransactionCommand(void)