pg_stat_monitor.c

/*-------------------------------------------------------------------------
 *
 * pg_stat_monitor.c
 *	  Track statement execution times across a whole database cluster.
 *
 * Portions Copyright © 2018-2025, Percona LLC and/or its affiliates
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 *
 * Portions Copyright (c) 1994, The Regents of the University of California
 *
 *-------------------------------------------------------------------------
 */

#include <postgres.h>

#include <arpa/inet.h>
#include <math.h>
#include <sys/resource.h>
#include <sys/time.h>

#include <access/hash.h>
#include <access/parallel.h>
#include <access/xact.h>
#include <catalog/pg_authid.h>
#include <commands/dbcommands.h>
#include <commands/explain.h>
#include <common/ip.h>
#include <funcapi.h>
#include <jit/jit.h>
#include <mb/pg_wchar.h>
#include <miscadmin.h>
#include <nodes/pg_list.h>
#include <optimizer/planner.h>
#include <parser/analyze.h>
#include <parser/parsetree.h>
#include <parser/scanner.h>
#include <parser/scansup.h>
#include <pgstat.h>
#include <storage/ipc.h>
#include <storage/proc.h>
#include <storage/shmem.h>
#include <tcop/utility.h>
#include <utils/acl.h>
#include <utils/builtins.h>
#include <utils/lsyscache.h>
#include <utils/memutils.h>

#if PG_VERSION_NUM >= 180000
#include <commands/explain_state.h>
#include <commands/explain_format.h>
#endif

#include "pg_stat_monitor.h"

 /*
  * Extension version number, for supporting older extension versions' objects
  */
typedef enum pgsmVersion
{
	PGSM_V1_0 = 0,
	PGSM_V2_0,
	PGSM_V2_1,
	PGSM_V2_3,
} pgsmVersion;

PG_MODULE_MAGIC;

#define BUILD_VERSION                   "2.4.0"

/* Number of output arguments (columns) for various API versions */
#define PG_STAT_MONITOR_COLS_V1_0    52
#define PG_STAT_MONITOR_COLS_V2_0    64
#define PG_STAT_MONITOR_COLS_V2_1    70
#define PG_STAT_MONITOR_COLS_V2_3    73
#define PG_STAT_MONITOR_COLS         PG_STAT_MONITOR_COLS_V2_3	/* maximum of above */

#define PGUNSIXBIT(val) (((val) & 0x3F) + '0')

#define pgsm_enabled(level) \
    (!IsParallelWorker() && \
    (pgsm_track == PGSM_TRACK_ALL || \
    (pgsm_track == PGSM_TRACK_TOP && (level) == 0)))

#define PGSM_INVALID_IP_MASK	0xFFFFFFFF

#define pgsm_client_ip_is_valid() \
	(pgsm_client_ip != PGSM_INVALID_IP_MASK)

/*---- Initialization Function Declarations ----*/
void		_PG_init(void);

/* Current nesting depth of planner/ExecutorRun/ProcessUtility calls */
static int	nesting_level = 0;

#if PG_VERSION_NUM < 170000
/* Before planner nesting level was counted separately */
static int	plan_nested_level = 0;
#endif

/* Histogram bucket variables */
static double hist_bucket_min;
static double hist_bucket_max;
static double hist_bucket_timings[MAX_RESPONSE_BUCKET + 2][2];	/* Start and end timings */
static int	hist_bucket_count_user;
static int	hist_bucket_count_total;

static uint32 pgsm_client_ip = PGSM_INVALID_IP_MASK;

/* The array to store outer layer query id */
static int64 *nested_queryids;
static char **nested_query_txts;
static List *lentries = NIL;

static char relations[REL_LST][REL_LEN];

static int	num_relations;		/* Number of relation in the query */
static bool system_init = false;
static struct rusage rusage_start;
static struct rusage rusage_end;

/* Application name and length; set each time when an entry is created locally */
static char app_name[APPLICATIONNAME_LEN];
static int	app_name_len;


/* Query buffer, store queries' text. */
static char *pgsm_explain(QueryDesc *queryDesc);

static void extract_query_comments(const char *query, char *comments, size_t max_len);
static void set_histogram_bucket_timings(void);
static void histogram_bucket_timings(int index, double *b_start, double *b_end);
static int	get_histogram_bucket(double q_time);

static bool IsSystemInitialized(void);
static double time_diff(struct timeval end, struct timeval start);
static void request_additional_shared_resources(void);


/* Saved hook values in case of unload */

#if PG_VERSION_NUM >= 150000
static void pgsm_shmem_request(void);
static shmem_request_hook_type prev_shmem_request_hook = NULL;
#endif
static planner_hook_type planner_hook_next = NULL;
static post_parse_analyze_hook_type prev_post_parse_analyze_hook = NULL;
static ExecutorStart_hook_type prev_ExecutorStart = NULL;
static ExecutorRun_hook_type prev_ExecutorRun = NULL;
static ExecutorFinish_hook_type prev_ExecutorFinish = NULL;
static ExecutorEnd_hook_type prev_ExecutorEnd = NULL;
static ProcessUtility_hook_type prev_ProcessUtility = NULL;
static emit_log_hook_type prev_emit_log_hook = NULL;

DECLARE_HOOK(void pgsm_emit_log_hook, ErrorData *edata);
static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
static ExecutorCheckPerms_hook_type prev_ExecutorCheckPerms_hook = NULL;

PG_FUNCTION_INFO_V1(pg_stat_monitor_version);
PG_FUNCTION_INFO_V1(pg_stat_monitor_reset);
PG_FUNCTION_INFO_V1(pg_stat_monitor_1_0);
PG_FUNCTION_INFO_V1(pg_stat_monitor_2_0);
PG_FUNCTION_INFO_V1(pg_stat_monitor_2_1);
PG_FUNCTION_INFO_V1(pg_stat_monitor_2_3);
PG_FUNCTION_INFO_V1(pg_stat_monitor);
PG_FUNCTION_INFO_V1(get_histogram_timings);
PG_FUNCTION_INFO_V1(pg_stat_monitor_hook_stats);

static uint pg_get_client_addr(bool *ok);
static int	pg_get_application_name(char *name, int buff_size);
static PgBackendStatus *pg_get_backend_status(void);
static Datum intarray_get_datum(int32 arr[], int len);

DECLARE_HOOK(void pgsm_post_parse_analyze, ParseState *pstate, Query *query, JumbleState *jstate);
DECLARE_HOOK(void pgsm_ExecutorStart, QueryDesc *queryDesc, int eflags);
#if PG_VERSION_NUM < 180000
DECLARE_HOOK(void pgsm_ExecutorRun, QueryDesc *queryDesc, ScanDirection direction, uint64 count, bool execute_once);
#else
DECLARE_HOOK(void pgsm_ExecutorRun, QueryDesc *queryDesc, ScanDirection direction, uint64 count);
#endif
DECLARE_HOOK(void pgsm_ExecutorFinish, QueryDesc *queryDesc);
DECLARE_HOOK(void pgsm_ExecutorEnd, QueryDesc *queryDesc);
#if PG_VERSION_NUM < 160000
DECLARE_HOOK(bool pgsm_ExecutorCheckPerms, List *rt, bool abort);
#else
DECLARE_HOOK(bool pgsm_ExecutorCheckPerms, List *rt, List *rp, bool abort);
#endif

DECLARE_HOOK(PlannedStmt *pgsm_planner_hook, Query *parse, const char *query_string, int cursorOptions, ParamListInfo boundParams);
DECLARE_HOOK(void pgsm_ProcessUtility, PlannedStmt *pstmt, const char *queryString,
			 bool readOnlyTree,
			 ProcessUtilityContext context,
			 ParamListInfo params, QueryEnvironment *queryEnv,
			 DestReceiver *dest,
			 QueryCompletion *qc);

static int64 pgsm_hash_string(const char *str, int len);
char	   *unpack_sql_state(int sql_state);

static pgsmEntry *pgsm_create_hash_entry(uint64 bucket_id, int64 queryid, PlanInfo *plan_info);
static void pgsm_add_to_list(pgsmEntry *entry, char *query_text, int query_len);
static void pgsm_delete_entry(uint64 queryid);
static pgsmEntry *pgsm_get_entry_for_query(int64 queryid, PlanInfo *plan_info, const char *query_text, int query_len, bool create, CmdType cmd_type);
static int64 get_pgsm_query_id_hash(const char *norm_query, int len);

static void pgsm_cleanup_callback(void *arg);
static void pgsm_store_error(const char *query, ErrorData *edata);

/*---- Local variables ----*/
static MemoryContextCallback mem_cxt_reset_callback =
{
	.func = pgsm_cleanup_callback,
	.arg = NULL
};
static volatile bool callback_setup = false;

static void pgsm_update_entry(pgsmEntry *entry,
							  const char *query,
							  char *comments,
							  int comments_len,
							  PlanInfo *plan_info,
							  SysInfo *sys_info,
							  ErrorInfo *error_info,
							  double plan_total_time,
							  double exec_total_time,
							  uint64 rows,
							  BufferUsage *bufusage,
							  WalUsage *walusage,
							  const struct JitInstrumentation *jitusage,
							  int parallel_workers_to_launch,
							  int parallel_workers_launched,
							  bool reset,
							  pgsmStoreKind kind);
static void pgsm_store(pgsmEntry *entry);

static void pg_stat_monitor_internal(FunctionCallInfo fcinfo,
									 pgsmVersion api_version,
									 bool showtext);

static char *generate_normalized_query(JumbleState *jstate, const char *query,
									   int query_loc, int *query_len_p, int encoding);
static void fill_in_constant_lengths(JumbleState *jstate, const char *query, int query_loc);
static int	comp_location(const void *a, const void *b);

static uint64 get_next_wbucket(pgsmSharedState *pgsm);

/*
 * To prevent deadlocks against our own backend we need to disable error
 * capture while holding the LWLock. The error capture hook is responsible
 * itself for re-enabling data capture when called on ERROR or above since
 * then we may not have been able to call pgsm_lock_release() due to the
 * statement being aborted.
 */
static bool disable_error_capture = false;

static void pgsm_lock_aquire(pgsmSharedState *pgsm, LWLockMode mode);
static void pgsm_lock_release(pgsmSharedState *pgsm);

/*
 * Module load callback
 */
/*  cppcheck-suppress unusedFunction */
void
_PG_init(void)
{
	elog(DEBUG2, "[pg_stat_monitor] pg_stat_monitor: %s().", __FUNCTION__);

	/*
	 * In order to create our shared memory area, we have to be loaded via
	 * shared_preload_libraries.  If not, fall out without hooking into any of
	 * the main system.  (We don't throw error here because it seems useful to
	 * allow the pg_stat_monitor functions to be created even when the module
	 * isn't active.  The functions must protect themselves against being
	 * called then, however.)
	 */
	if (!process_shared_preload_libraries_in_progress)
		return;

	/* Initialize the GUC variables */
	init_guc();

	set_histogram_bucket_timings();

	/*
	 * Inform the postmaster that we want to enable query_id calculation if
	 * compute_query_id is set to auto.
	 */
	EnableQueryId();

	EmitWarningsOnPlaceholders("pg_stat_monitor");

	/*
	 * Install hooks.
	 */
#if PG_VERSION_NUM >= 150000
	prev_shmem_request_hook = shmem_request_hook;
	shmem_request_hook = pgsm_shmem_request;
#else
	request_additional_shared_resources();
#endif
	prev_shmem_startup_hook = shmem_startup_hook;
	shmem_startup_hook = pgsm_shmem_startup;
	prev_post_parse_analyze_hook = post_parse_analyze_hook;
	post_parse_analyze_hook = HOOK(pgsm_post_parse_analyze);
	prev_ExecutorStart = ExecutorStart_hook;
	ExecutorStart_hook = HOOK(pgsm_ExecutorStart);
	prev_ExecutorRun = ExecutorRun_hook;
	ExecutorRun_hook = HOOK(pgsm_ExecutorRun);
	prev_ExecutorFinish = ExecutorFinish_hook;
	ExecutorFinish_hook = HOOK(pgsm_ExecutorFinish);
	prev_ExecutorEnd = ExecutorEnd_hook;
	ExecutorEnd_hook = HOOK(pgsm_ExecutorEnd);
	prev_ProcessUtility = ProcessUtility_hook;
	ProcessUtility_hook = HOOK(pgsm_ProcessUtility);
	planner_hook_next = planner_hook;
	planner_hook = HOOK(pgsm_planner_hook);
	prev_emit_log_hook = emit_log_hook;
	emit_log_hook = HOOK(pgsm_emit_log_hook);
	prev_ExecutorCheckPerms_hook = ExecutorCheckPerms_hook;
	ExecutorCheckPerms_hook = HOOK(pgsm_ExecutorCheckPerms);

	nested_queryids = (int64 *) malloc(sizeof(int64) * max_stack_depth);
	nested_query_txts = (char **) calloc(max_stack_depth, sizeof(char *));

	system_init = true;
}

/*
 * shmem_startup hook: allocate or attach to shared memory,
 * then load any pre-existing statistics from file.
 * Also create and load the query-texts file, which is expected to exist
 * (even if empty) while the module is enabled.
 */
void
pgsm_shmem_startup(void)
{
	if (prev_shmem_startup_hook)
		prev_shmem_startup_hook();

	pgsm_startup();
}

static void
request_additional_shared_resources(void)
{
	/*
	 * Request additional shared resources.  (These are no-ops if we're not in
	 * the postmaster process.)  We'll allocate or attach to the shared
	 * resources in pgsm_shmem_startup().
	 */
	RequestAddinShmemSpace(pgsm_ShmemSize() + HOOK_STATS_SIZE);
	RequestNamedLWLockTranche("pg_stat_monitor", 1);
}

/*
 * Select the version of pg_stat_monitor.
 */
Datum
pg_stat_monitor_version(PG_FUNCTION_ARGS)
{
	PG_RETURN_TEXT_P(cstring_to_text(BUILD_VERSION));
}

#if PG_VERSION_NUM >= 150000
/*
 * shmem_request hook: request additional shared resources.  We'll allocate or
 * attach to the shared resources in pgsm_shmem_startup().
 */
static void
pgsm_shmem_request(void)
{
	if (prev_shmem_request_hook)
		prev_shmem_request_hook();
	request_additional_shared_resources();
}
#endif

static void
pgsm_post_parse_analyze_internal(ParseState *pstate, Query *query, JumbleState *jstate)
{
	pgsmEntry  *entry;
	const char *query_text;
	char	   *norm_query = NULL;
	int			norm_query_len;
	int			location;
	int			query_len;

	/* Safety check... */
	if (!IsSystemInitialized())
		return;

	if (callback_setup == false)
	{
		/*
		 * If MessageContext is valid setup a callback to cleanup our local
		 * stats list when the MessagContext gets reset
		 */
		if (MemoryContextIsValid(MessageContext))
		{
			MemoryContextRegisterResetCallback(MessageContext, &mem_cxt_reset_callback);
			callback_setup = true;
		}
	}

	if (!pgsm_enabled(nesting_level))
		return;

	/*
	 * If it's EXECUTE, clear the queryId so that stats will accumulate for
	 * the underlying PREPARE.  But don't do this if we're not tracking
	 * utility statements, to avoid messing up another extension that might be
	 * tracking them.
	 */
	if (query->utilityStmt)
	{
		if (pgsm_track_utility && IsA(query->utilityStmt, ExecuteStmt))
			query->queryId = INT64CONST(0);

		return;
	}

	/*
	 * If we are unlucky enough to get a hash of zero, use 1 instead, to
	 * prevent confusion with the utility-statement case.
	 */
	if (query->queryId == INT64CONST(0))
		query->queryId = INT64CONST(1);

	/*
	 * Let's save the normalized query so that we can save the data without in
	 * hash later on without the need of jstate which wouldn't be available.
	 */
	query_text = pstate->p_sourcetext;
	location = query->stmt_location;
	query_len = query->stmt_len;

	/* We should always have a valid query. */
	query_text = CleanQuerytext(query_text, &location, &query_len);
	Assert(query_text);

	norm_query_len = query_len;

	/* Generate a normalized query */
	if (jstate && jstate->clocations_count > 0 && (pgsm_enable_pgsm_query_id || pgsm_normalized_query))
	{
		norm_query = generate_normalized_query(jstate,
											   query_text,	/* query */
											   location,	/* query location */
											   &norm_query_len,
											   GetDatabaseEncoding());

		Assert(norm_query);
	}

	/*
	 * At this point, we don't know which bucket this query will land in, so
	 * passing 0. The store function MUST later update it based on the current
	 * bucket value. The correct bucket value will be needed then to search
	 * the hash table, or create the appropriate entry.
	 */
	entry = pgsm_create_hash_entry(0, query->queryId, NULL);

	/*
	 * Update other member that are not counters, so that we don't have to
	 * worry about these.
	 */
	entry->pgsm_query_id = get_pgsm_query_id_hash(norm_query ? norm_query : query_text, norm_query_len);
	entry->counters.info.cmd_type = query->commandType;

	/*
	 * Add the query text and entry to the local list.
	 *
	 * Preserve the normalized query if needed and we got a valid one.
	 * Otherwise, store the actual query so that we don't have to check what
	 * query to store when saving into the hash.
	 *
	 * In case of query_text, request the function to duplicate it so that it
	 * is put in the relevant memory context.
	 */
	if (pgsm_normalized_query && norm_query)
		pgsm_add_to_list(entry, norm_query, norm_query_len);
	else
	{
		pgsm_add_to_list(entry, (char *) query_text, query_len);
	}

	/* Check that we've not exceeded max_stack_depth */
	Assert(list_length(lentries) <= max_stack_depth);

	if (norm_query)
		pfree(norm_query);
}

/*
 * Post-parse-analysis hook: mark query with a queryId
 */
static void
pgsm_post_parse_analyze(ParseState *pstate, Query *query, JumbleState *jstate)
{
	if (prev_post_parse_analyze_hook)
		prev_post_parse_analyze_hook(pstate, query, jstate);

	pgsm_post_parse_analyze_internal(pstate, query, jstate);
}

/*
 * ExecutorStart hook: start up tracking if needed
 */
static void
pgsm_ExecutorStart(QueryDesc *queryDesc, int eflags)
{
	if (getrusage(RUSAGE_SELF, &rusage_start) != 0)
		elog(DEBUG1, "[pg_stat_monitor] pgsm_ExecutorStart: failed to execute getrusage.");

	if (prev_ExecutorStart)
		prev_ExecutorStart(queryDesc, eflags);
	else
		standard_ExecutorStart(queryDesc, eflags);

	/*
	 * If query has queryId zero, don't track it.  This prevents double
	 * counting of optimizable statements that are directly contained in
	 * utility statements.
	 */
	if (pgsm_enabled(nesting_level) &&
		queryDesc->plannedstmt->queryId != INT64CONST(0))
	{
		/*
		 * Set up to track total elapsed time in ExecutorRun.  Make sure the
		 * space is allocated in the per-query context so it will go away at
		 * ExecutorEnd.
		 */
		if (queryDesc->totaltime == NULL)
		{
			MemoryContext oldcxt;

			oldcxt = MemoryContextSwitchTo(queryDesc->estate->es_query_cxt);
			queryDesc->totaltime = InstrAlloc(1, INSTRUMENT_ALL, false);
			MemoryContextSwitchTo(oldcxt);
		}
	}
}


/*
 * ExecutorRun hook: all we need do is track nesting depth
 */
static void
#if PG_VERSION_NUM < 180000
pgsm_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count,
				 bool execute_once)
#else
pgsm_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count)
#endif
{
	if (nesting_level >= 0 && nesting_level < max_stack_depth)
	{
		nested_queryids[nesting_level] = queryDesc->plannedstmt->queryId;
		nested_query_txts[nesting_level] = strdup(queryDesc->sourceText);
	}

	nesting_level++;
	PG_TRY();
	{
		if (prev_ExecutorRun)
		{
#if PG_VERSION_NUM < 180000
			prev_ExecutorRun(queryDesc, direction, count, execute_once);
#else
			prev_ExecutorRun(queryDesc, direction, count);
#endif
		}
		else
		{
#if PG_VERSION_NUM < 180000
			standard_ExecutorRun(queryDesc, direction, count, execute_once);
#else
			standard_ExecutorRun(queryDesc, direction, count);
#endif
		}
		nesting_level--;
		if (nesting_level >= 0 && nesting_level < max_stack_depth)
		{
			nested_queryids[nesting_level] = INT64CONST(0);
			if (nested_query_txts[nesting_level])
				free(nested_query_txts[nesting_level]);
			nested_query_txts[nesting_level] = NULL;
		}
	}
	PG_CATCH();
	{
		nesting_level--;
		if (nesting_level >= 0 && nesting_level < max_stack_depth)
		{
			nested_queryids[nesting_level] = INT64CONST(0);
			if (nested_query_txts[nesting_level])
				free(nested_query_txts[nesting_level]);
			nested_query_txts[nesting_level] = NULL;
		}
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * ExecutorFinish hook: all we need do is track nesting depth
 */
static void
pgsm_ExecutorFinish(QueryDesc *queryDesc)
{
	nesting_level++;

	PG_TRY();
	{
		if (prev_ExecutorFinish)
			prev_ExecutorFinish(queryDesc);
		else
			standard_ExecutorFinish(queryDesc);
		nesting_level--;
	}
	PG_CATCH();
	{
		nesting_level--;
		PG_RE_THROW();
	}
	PG_END_TRY();

}

static char *
pgsm_explain(QueryDesc *queryDesc)
{
	ExplainState *es = NewExplainState();

	es->buffers = false;
	es->analyze = false;
	es->verbose = false;
	es->costs = false;
	es->format = EXPLAIN_FORMAT_TEXT;

	ExplainBeginOutput(es);
	ExplainPrintPlan(es, queryDesc);
	ExplainEndOutput(es);

	if (es->str->len > 0 && es->str->data[es->str->len - 1] == '\n')
		es->str->data[--es->str->len] = '\0';
	return es->str->data;
}

/*
 * ExecutorEnd hook: store results if needed
 */
static void
pgsm_ExecutorEnd(QueryDesc *queryDesc)
{
	int64		queryId = queryDesc->plannedstmt->queryId;
	SysInfo		sys_info;
	PlanInfo	plan_info;
	PlanInfo   *plan_ptr = NULL;
	pgsmEntry  *entry = NULL;

	/* Extract the plan information in case of SELECT statement */
	if (queryDesc->operation == CMD_SELECT && pgsm_enable_query_plan)
	{
		int			rv;
		MemoryContext oldctx;

		/*
		 * Making sure it is a per query context so that there's no memory
		 * leak when executor ends.
		 */
		oldctx = MemoryContextSwitchTo(queryDesc->estate->es_query_cxt);

		rv = snprintf(plan_info.plan_text, PLAN_TEXT_LEN, "%s", pgsm_explain(queryDesc));

		/*
		 * If snprint didn't write anything or there was an error, let's keep
		 * planinfo as NULL.
		 */
		if (rv > 0)
		{
			plan_info.plan_len = (rv < PLAN_TEXT_LEN) ? rv : PLAN_TEXT_LEN - 1;
			plan_info.planid = pgsm_hash_string(plan_info.plan_text, plan_info.plan_len);
			plan_ptr = &plan_info;
		}

		/* Switch back to old context */
		MemoryContextSwitchTo(oldctx);
	}

	if (queryId != INT64CONST(0) && queryDesc->totaltime && pgsm_enabled(nesting_level))
	{
		entry = pgsm_get_entry_for_query(queryId, plan_ptr, (char *) queryDesc->sourceText, strlen(queryDesc->sourceText), true, queryDesc->operation);
		if (!entry)
		{
			elog(DEBUG2, "[pg_stat_monitor] pgsm_ExecutorEnd: Failed to find entry for [" INT64_FORMAT "] %s.", queryId, queryDesc->sourceText);
			return;
		}

		if (entry->key.planid == 0)
			entry->key.planid = (plan_ptr) ? plan_ptr->planid : 0;

		/*
		 * Make sure stats accumulation is done.  (Note: it's okay if several
		 * levels of hook all do this.)
		 */
		InstrEndLoop(queryDesc->totaltime);

		sys_info.utime = 0;
		sys_info.stime = 0;

		if (getrusage(RUSAGE_SELF, &rusage_end) != 0)
			elog(DEBUG1, "[pg_stat_monitor] pgsm_ExecutorEnd: Failed to execute getrusage.");
		else
		{
			sys_info.utime = time_diff(rusage_end.ru_utime, rusage_start.ru_utime);
			sys_info.stime = time_diff(rusage_end.ru_stime, rusage_start.ru_stime);
		}

		entry->counters.info.cmd_type = queryDesc->operation;

		pgsm_update_entry(entry,	/* entry */
						  NULL, /* query */
						  NULL, /* comments */
						  0,	/* comments length */
						  plan_ptr, /* PlanInfo */
						  &sys_info,	/* SysInfo */
						  NULL, /* ErrorInfo */
						  0,	/* plan_total_time */
						  queryDesc->totaltime->total * 1000.0, /* exec_total_time */
						  queryDesc->estate->es_processed,	/* rows */
						  &queryDesc->totaltime->bufusage,	/* bufusage */
						  &queryDesc->totaltime->walusage,	/* walusage */
#if PG_VERSION_NUM >= 150000
						  queryDesc->estate->es_jit ? &queryDesc->estate->es_jit->instr : NULL, /* jitusage */
#else
						  NULL,
#endif
#if PG_VERSION_NUM >= 180000
						  queryDesc->estate->es_parallel_workers_to_launch, /* parallel_workers_to_launch */
						  queryDesc->estate->es_parallel_workers_launched,	/* parallel_workers_launched */
#else
						  0,	/* parallel_workers_to_launch */
						  0,	/* parallel_workers_launched */
#endif
						  false,	/* reset */
						  PGSM_EXEC);	/* kind */

		pgsm_store(entry);
	}

	if (prev_ExecutorEnd)
		prev_ExecutorEnd(queryDesc);
	else
		standard_ExecutorEnd(queryDesc);

	pgsm_delete_entry(queryDesc->plannedstmt->queryId);

	num_relations = 0;
}

static bool
#if PG_VERSION_NUM < 160000
pgsm_ExecutorCheckPerms(List *rt, bool abort)
#else
pgsm_ExecutorCheckPerms(List *rt, List *rp, bool abort)
#endif
{
	ListCell   *lr = NULL;
	int			i = 0;
	int			j = 0;
	Oid			list_oid[20];

	num_relations = 0;

	foreach(lr, rt)
	{
		RangeTblEntry *rte = lfirst(lr);

		if (rte->rtekind != RTE_RELATION
#if PG_VERSION_NUM >= 160000
			&& (rte->rtekind != RTE_SUBQUERY && rte->relkind != 'v')
#endif
			)
			continue;

		if (i < REL_LST)
		{
			bool		found = false;

			for (j = 0; j < i; j++)
			{
				if (list_oid[j] == rte->relid)
					found = true;
			}

			if (!found)
			{
				char	   *namespace_name;
				char	   *relation_name;

				list_oid[j] = rte->relid;
				namespace_name = get_namespace_name(get_rel_namespace(rte->relid));
				relation_name = get_rel_name(rte->relid);
				if (rte->relkind == 'v')
					snprintf(relations[i++], REL_LEN, "%s.%s*", namespace_name, relation_name);
				else
					snprintf(relations[i++], REL_LEN, "%s.%s", namespace_name, relation_name);
			}
		}
	}
	num_relations = i;

	if (prev_ExecutorCheckPerms_hook)
#if PG_VERSION_NUM < 160000
		return prev_ExecutorCheckPerms_hook(rt, abort);
#else
		return prev_ExecutorCheckPerms_hook(rt, rp, abort);
#endif

	return true;
}

static PlannedStmt *
pgsm_planner_hook(Query *parse, const char *query_string, int cursorOptions, ParamListInfo boundParams)
{
	PlannedStmt *result;
	int64		queryId = parse->queryId;

	/*
	 * We can't process the query if no query_string is provided, as
	 * pgsm_store needs it.  We also ignore query without queryid, as it would
	 * be treated as a utility statement, which may not be the case.
	 *
	 * Note that planner_hook can be called from the planner itself, so we
	 * have a specific nesting level for the planner.  However, utility
	 * commands containing optimizable statements can also call the planner,
	 * same for regular DML (for instance for underlying foreign key queries).
	 * So testing the planner nesting level only is not enough to detect real
	 * top level planner call.
	 */

	bool		enabled;
#if PG_VERSION_NUM >= 170000
	enabled = pgsm_enabled(nesting_level);
#else
	enabled = pgsm_enabled(plan_nested_level + nesting_level);
#endif

	if (enabled && pgsm_track_planning && query_string && queryId != INT64CONST(0))
	{
		pgsmEntry  *entry = NULL;
		instr_time	start;
		instr_time	duration;
		BufferUsage bufusage_start;
		BufferUsage bufusage;
		WalUsage	walusage_start;
		WalUsage	walusage;

		/* We need to track buffer usage as the planner can access them. */
		bufusage_start = pgBufferUsage;

		/*
		 * Similarly the planner could write some WAL records in some cases
		 * (e.g. setting a hint bit with those being WAL-logged)
		 */
		walusage_start = pgWalUsage;
		INSTR_TIME_SET_CURRENT(start);

		if (MemoryContextIsValid(MessageContext))
			entry = pgsm_get_entry_for_query(queryId, NULL, query_string, strlen(query_string), true, parse->commandType);

#if PG_VERSION_NUM >= 170000
		nesting_level++;
#else
		plan_nested_level++;
#endif
		PG_TRY();
		{
			/*
			 * If there is a previous installed hook, then assume it's going
			 * to call standard_planner() function, otherwise we call the
			 * function here. This is to avoid calling standard_planner()
			 * function twice, since it modifies the first argument (Query *),
			 * the second call would trigger an assertion failure.
			 */
			if (planner_hook_next)
				result = planner_hook_next(parse, query_string, cursorOptions, boundParams);
			else
				result = standard_planner(parse, query_string, cursorOptions, boundParams);
		}
		PG_FINALLY();
		{
#if PG_VERSION_NUM >= 170000
			nesting_level--;
#else
			plan_nested_level--;
#endif
		}
		PG_END_TRY();

		INSTR_TIME_SET_CURRENT(duration);
		INSTR_TIME_SUBTRACT(duration, start);

		/* calc differences of buffer counters. */
		memset(&bufusage, 0, sizeof(BufferUsage));
		BufferUsageAccumDiff(&bufusage, &pgBufferUsage, &bufusage_start);

		/* calc differences of WAL counters. */
		memset(&walusage, 0, sizeof(WalUsage));
		WalUsageAccumDiff(&walusage, &pgWalUsage, &walusage_start);

		/* The plan details are captured when the query finishes */
		if (entry)
			pgsm_update_entry(entry,	/* entry */
							  NULL, /* query */
							  NULL, /* comments */
							  0,	/* comments length */
							  NULL, /* PlanInfo */
							  NULL, /* SysInfo */
							  NULL, /* ErrorInfo */
							  INSTR_TIME_GET_MILLISEC(duration),	/* plan_total_time */
							  0,	/* exec_total_time */
							  0,	/* rows */
							  &bufusage,	/* bufusage */
							  &walusage,	/* walusage */
							  NULL, /* jitusage */
							  0,	/* parallel_workers_to_launch */
							  0,	/* parallel_workers_launched */
							  false,	/* reset */
							  PGSM_PLAN);	/* kind */
	}
	else
	{
		/*
		 * Even though we're not tracking plan time for this statement, we
		 * must still increment the nesting level, to ensure that functions
		 * evaluated during planning are not seen as top-level calls.
		 *
		 * If there is a previous installed hook, then assume it's going to
		 * call standard_planner() function, otherwise we call the function
		 * here. This is to avoid calling standard_planner() function twice,
		 * since it modifies the first argument (Query *), the second call
		 * would trigger an assertion failure.
		 */

#if PG_VERSION_NUM >= 170000
		nesting_level++;
#else
		plan_nested_level++;
#endif
		PG_TRY();
		{
			if (planner_hook_next)
				result = planner_hook_next(parse, query_string, cursorOptions, boundParams);
			else
				result = standard_planner(parse, query_string, cursorOptions, boundParams);
		}
		PG_FINALLY();
		{
#if PG_VERSION_NUM >= 170000
			nesting_level--;
#else
			plan_nested_level--;
#endif
		}
		PG_END_TRY();

	}
	return result;
}

/*
 * ProcessUtility hook
 */
static void
pgsm_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
					bool readOnlyTree,
					ProcessUtilityContext context,
					ParamListInfo params, QueryEnvironment *queryEnv,
					DestReceiver *dest,
					QueryCompletion *qc)

{
	Node	   *parsetree = pstmt->utilityStmt;
	int64		queryId = pstmt->queryId;
	bool		enabled = pgsm_track_utility && pgsm_enabled(nesting_level);

	/*
	 * Force utility statements to get queryId zero.  We do this even in cases
	 * where the statement contains an optimizable statement for which a
	 * queryId could be derived (such as EXPLAIN or DECLARE CURSOR).  For such
	 * cases, runtime control will first go through ProcessUtility and then
	 * the executor, and we don't want the executor hooks to do anything,
	 * since we are already measuring the statement's costs at the utility
	 * level.
	 */
	if (enabled)
		pstmt->queryId = INT64CONST(0);

	/*
	 * If it's an EXECUTE statement, we don't track it and don't increment the
	 * nesting level.  This allows the cycles to be charged to the underlying
	 * PREPARE instead (by the Executor hooks), which is much more useful.
	 *
	 * We also don't track execution of PREPARE.  If we did, we would get one
	 * hash table entry for the PREPARE (with hash calculated from the query
	 * string), and then a different one with the same query string (but hash
	 * calculated from the query tree) would be used to accumulate costs of
	 * ensuing EXECUTEs.  This would be confusing.  Since PREPARE doesn't
	 * actually run the planner (only parse+rewrite), its costs are generally
	 * pretty negligible and it seems okay to just ignore it.
	 *
	 * Likewise, we don't track execution of DEALLOCATE.
	 */
	if (enabled &&
		!IsA(parsetree, ExecuteStmt) &&
		!IsA(parsetree, PrepareStmt) &&
		!IsA(parsetree, DeallocateStmt))
	{
		char	   *query_text;
		int			location;
		int			query_len;
		instr_time	start;
		instr_time	duration;
		uint64		rows;
		SysInfo		sys_info;
		BufferUsage bufusage;
		BufferUsage bufusage_start = pgBufferUsage;
		WalUsage	walusage;
		WalUsage	walusage_start = pgWalUsage;
		pgsmEntry  *entry = pgsm_create_hash_entry(0, queryId, NULL);

		if (getrusage(RUSAGE_SELF, &rusage_start) != 0)
			elog(DEBUG1, "[pg_stat_monitor] pgsm_ProcessUtility: Failed to execute getrusage.");

		INSTR_TIME_SET_CURRENT(start);
		nesting_level++;

		PG_TRY();
		{
			if (prev_ProcessUtility)
				prev_ProcessUtility(pstmt, queryString,
									readOnlyTree,
									context, params, queryEnv,
									dest,
									qc);
			else
				standard_ProcessUtility(pstmt, queryString,
										readOnlyTree,
										context, params, queryEnv,
										dest,
										qc);
			nesting_level--;
		}
		PG_CATCH();
		{
			nesting_level--;
			PG_RE_THROW();
		}

		sys_info.utime = 0;
		sys_info.stime = 0;

		PG_END_TRY();

		if (getrusage(RUSAGE_SELF, &rusage_end) != 0)
			elog(DEBUG1, "[pg_stat_monitor] pgsm_ProcessUtility: Failed to execute getrusage.");
		else
		{
			sys_info.utime = time_diff(rusage_end.ru_utime, rusage_start.ru_utime);
			sys_info.stime = time_diff(rusage_end.ru_stime, rusage_start.ru_stime);
		}

		INSTR_TIME_SET_CURRENT(duration);
		INSTR_TIME_SUBTRACT(duration, start);

		rows = (qc && (qc->commandTag == CMDTAG_COPY ||
					   qc->commandTag == CMDTAG_FETCH ||
					   qc->commandTag == CMDTAG_SELECT ||
					   qc->commandTag == CMDTAG_REFRESH_MATERIALIZED_VIEW))
			? qc->nprocessed
			: 0;

		/* calc differences of WAL counters. */
		memset(&walusage, 0, sizeof(WalUsage));
		WalUsageAccumDiff(&walusage, &pgWalUsage, &walusage_start);

		/* calc differences of buffer counters. */
		memset(&bufusage, 0, sizeof(BufferUsage));
		BufferUsageAccumDiff(&bufusage, &pgBufferUsage, &bufusage_start);

		location = pstmt->stmt_location;
		query_len = pstmt->stmt_len;
		query_text = (char *) CleanQuerytext(queryString, &location, &query_len);

		entry->pgsm_query_id = get_pgsm_query_id_hash(query_text, query_len);
		entry->counters.info.cmd_type = pstmt->commandType;

		pgsm_add_to_list(entry, query_text, query_len);

		/* Check that we've not exceeded max_stack_depth */
		Assert(list_length(lentries) <= max_stack_depth);

		/* The plan details are captured when the query finishes */
		pgsm_update_entry(entry,	/* entry */
						  (char *) query_text,	/* query */
						  NULL, /* comments */
						  0,	/* comments length */
						  NULL, /* PlanInfo */
						  &sys_info,	/* SysInfo */
						  NULL, /* ErrorInfo */
						  0,	/* plan_total_time */
						  INSTR_TIME_GET_MILLISEC(duration),	/* exec_total_time */
						  rows, /* rows */
						  &bufusage,	/* bufusage */
						  &walusage,	/* walusage */
						  NULL, /* jitusage */
						  0,	/* parallel_workers_to_launch */
						  0,	/* parallel_workers_launched */
						  false,	/* reset */
						  PGSM_EXEC);	/* kind */

		pgsm_store(entry);
	}
	else
	{
		/*
		 * Even though we're not tracking execution time for this statement,
		 * we must still increment the nesting level, to ensure that functions
		 * evaluated within it are not seen as top-level calls.  But don't do
		 * so for EXECUTE; that way, when control reaches pgss_planner or
		 * pgss_ExecutorStart, we will treat the costs as top-level if
		 * appropriate.  Likewise, don't bump for PREPARE, so that parse
		 * analysis will treat the statement as top-level if appropriate.
		 *
		 * Likewise, we don't track execution of DEALLOCATE.
		 *
		 * To be absolutely certain we don't mess up the nesting level,
		 * evaluate the bump_level condition just once.
		 */

#if PG_VERSION_NUM >= 170000
		bool		bump_level =
			!IsA(parsetree, ExecuteStmt) &&
			!IsA(parsetree, PrepareStmt) &&
			!IsA(parsetree, DeallocateStmt);

		if (bump_level)
			nesting_level++;

		PG_TRY();
		{
#endif
			if (prev_ProcessUtility)
				prev_ProcessUtility(pstmt, queryString,
									readOnlyTree,
									context, params, queryEnv,
									dest,
									qc);
			else
				standard_ProcessUtility(pstmt, queryString,
										readOnlyTree,
										context, params, queryEnv,
										dest,
										qc);

#if PG_VERSION_NUM >= 170000
			if (bump_level)
				nesting_level--;
		}
		PG_CATCH();
		{
			if (bump_level)
				nesting_level--;
			PG_RE_THROW();
		}
		PG_END_TRY();
#endif
	}
	pgsm_delete_entry(pstmt->queryId);
}

/*
 * Given an arbitrarily long query string, produce a hash for the purposes of
 * identifying the query, without normalizing constants.  Used when hashing
 * utility statements.
 */
static int64
pgsm_hash_string(const char *str, int len)
{
	return DatumGetInt64(hash_any_extended((const unsigned char *) str, len, 0));
}

static PgBackendStatus *
pg_get_backend_status(void)
{

#if PG_VERSION_NUM >= 170000
	return pgstat_get_beentry_by_proc_number(MyProcPid);
#elif PG_VERSION_NUM >= 160000
	return &(pgstat_get_local_beentry_by_backend_id(MyBackendId)->backendStatus);
#else
	LocalPgBackendStatus *local_beentry;
	int			num_backends = pgstat_fetch_stat_numbackends();
	int			i;

	for (i = 1; i <= num_backends; i++)
	{
		PgBackendStatus *beentry;

		local_beentry = pgstat_fetch_stat_local_beentry(i);
		if (!local_beentry)
			continue;

		beentry = &local_beentry->backendStatus;

		if (beentry->st_procpid == MyProcPid)
			return beentry;
	}
	return NULL;

#endif
}

/*
 * The caller should allocate max_len memory to name including terminating null.
 * The function returns the length of the string.
 */
static int
pg_get_application_name(char *name, int buff_size)
{
	PgBackendStatus *beentry;

	/* Try to read application name from GUC directly */
	if (application_name && *application_name)
		snprintf(name, buff_size, "%s", application_name);
	else
	{
		beentry = pg_get_backend_status();

		if (!beentry)
			snprintf(name, buff_size, "%s", "unknown");
		else
			snprintf(name, buff_size, "%s", beentry->st_appname);
	}

	/* Return length so that others don't have to calculate */
	return strlen(name);
}

static uint
pg_get_client_addr(bool *ok)
{
	PgBackendStatus *beentry = pg_get_backend_status();
	char		remote_host[NI_MAXHOST];
	int			ret;

	remote_host[0] = '\0';

	if (!beentry)
		return ntohl(inet_addr("127.0.0.1"));

	*ok = true;

	ret = pg_getnameinfo_all(&beentry->st_clientaddr.addr,
							 beentry->st_clientaddr.salen,
							 remote_host, sizeof(remote_host),
							 NULL, 0,
							 NI_NUMERICHOST | NI_NUMERICSERV);
	if (ret != 0)
		return ntohl(inet_addr("127.0.0.1"));

	if (strcmp(remote_host, "[local]") == 0)
		return ntohl(inet_addr("127.0.0.1"));

	return ntohl(inet_addr(remote_host));
}

static void
pgsm_update_entry(pgsmEntry *entry,
				  const char *query,
				  char *comments,
				  int comments_len,
				  PlanInfo *plan_info,
				  SysInfo *sys_info,
				  ErrorInfo *error_info,
				  double plan_total_time,
				  double exec_total_time,
				  uint64 rows,
				  BufferUsage *bufusage,
				  WalUsage *walusage,
				  const struct JitInstrumentation *jitusage,
				  int parallel_workers_to_launch,
				  int parallel_workers_launched,
				  bool reset,
				  pgsmStoreKind kind)
{
	int			index;
	double		old_mean;
	int			plan_text_len = plan_info ? plan_info->plan_len : 0;

	/*
	 * Start collecting data for next bucket and reset all counters and
	 * timestamps
	 */
	if (reset)
	{
		memset(&entry->counters, 0, sizeof(Counters));
		entry->stats_since = GetCurrentTimestamp();
		entry->minmax_stats_since = entry->stats_since;
	}

	if (kind == PGSM_STORE)
		SpinLockAcquire(&entry->mutex);

	/*
	 * Extract comments if enabled and only when the query has completed with
	 * or without error
	 */
	if (pgsm_extract_comments && kind == PGSM_STORE
		&& !entry->counters.info.comments[0] && comments_len > 0)
		strlcpy(entry->counters.info.comments, comments, COMMENTS_LEN);

	if (kind == PGSM_PLAN || kind == PGSM_STORE)
	{
		if (entry->counters.plancalls.calls == 0)
			entry->counters.plancalls.usage = USAGE_INIT;

		entry->counters.plancalls.calls += 1;
		entry->counters.plantime.total_time += plan_total_time;

		if (entry->counters.plancalls.calls == 1)
		{
			entry->counters.plantime.min_time = plan_total_time;
			entry->counters.plantime.max_time = plan_total_time;
			entry->counters.plantime.mean_time = plan_total_time;
		}
		else
		{
			/* Increment the counts, except when jstate is not NULL */
			old_mean = entry->counters.plantime.mean_time;

			entry->counters.plantime.mean_time += (plan_total_time - old_mean) / entry->counters.plancalls.calls;
			entry->counters.plantime.sum_var_time += (plan_total_time - old_mean) * (plan_total_time - entry->counters.plantime.mean_time);

			/* calculate min and max time */
			if (entry->counters.plantime.min_time > plan_total_time)
				entry->counters.plantime.min_time = plan_total_time;

			if (entry->counters.plantime.max_time < plan_total_time)
				entry->counters.plantime.max_time = plan_total_time;
		}
	}

	if (kind == PGSM_EXEC || kind == PGSM_STORE)
	{
		if (entry->counters.calls.calls == 0)
			entry->counters.calls.usage = USAGE_INIT;

		entry->counters.calls.calls += 1;
		entry->counters.time.total_time += exec_total_time;

		if (entry->counters.calls.calls == 1)
		{
			entry->counters.time.min_time = exec_total_time;
			entry->counters.time.max_time = exec_total_time;
			entry->counters.time.mean_time = exec_total_time;
		}
		else
		{
			/* Increment the counts, except when jstate is not NULL */
			old_mean = entry->counters.time.mean_time;
			entry->counters.time.mean_time += (exec_total_time - old_mean) / entry->counters.calls.calls;
			entry->counters.time.sum_var_time += (exec_total_time - old_mean) * (exec_total_time - entry->counters.time.mean_time);

			/* calculate min and max time */
			if (entry->counters.time.min_time > exec_total_time)
				entry->counters.time.min_time = exec_total_time;

			if (entry->counters.time.max_time < exec_total_time)
				entry->counters.time.max_time = exec_total_time;
		}

		index = get_histogram_bucket(exec_total_time);
		entry->counters.resp_calls[index]++;
	}

	if (plan_text_len > 0 && !entry->counters.planinfo.plan_text[0])
	{
		entry->counters.planinfo.planid = plan_info->planid;
		entry->counters.planinfo.plan_len = plan_text_len;
		strlcpy(entry->counters.planinfo.plan_text, plan_info->plan_text, PLAN_TEXT_LEN);
	}

	/* Only should process this once when storing the data */
	if (kind == PGSM_STORE)
	{
		if (pgsm_track_application_names && app_name_len > 0 && !entry->counters.info.application_name[0])
			strlcpy(entry->counters.info.application_name, app_name, APPLICATIONNAME_LEN);

		entry->counters.info.num_relations = num_relations;
		for (int i = 0; i < num_relations; i++)
			strlcpy(entry->counters.info.relations[i], relations[i], REL_LEN);

		if (nesting_level > 0 && nesting_level < max_stack_depth && entry->key.parentid != 0 && pgsm_track == PGSM_TRACK_ALL)
		{
			if (!DsaPointerIsValid(entry->counters.info.parent_query))
			{
				int			parent_query_len = nested_query_txts[nesting_level - 1] ?
					strlen(nested_query_txts[nesting_level - 1]) : 0;

				/* If we have a parent query, store it in the raw dsa area */
				if (parent_query_len > 0)
				{
					char	   *qry_buff;
					dsa_area   *query_dsa_area = get_dsa_area_for_query_text();

					/*
					 * Use dsa_allocate_extended with DSA_ALLOC_NO_OOM flag,
					 * as we don't want to get an error if memory allocation
					 * fails.
					 */
					dsa_pointer qry = dsa_allocate_extended(query_dsa_area, parent_query_len + 1, DSA_ALLOC_NO_OOM | DSA_ALLOC_ZERO);

					if (DsaPointerIsValid(qry))
					{
						qry_buff = dsa_get_address(query_dsa_area, qry);
						memcpy(qry_buff, nested_query_txts[nesting_level - 1], parent_query_len);
						qry_buff[parent_query_len] = 0;
						/* store the dsa pointer for parent query text */
						entry->counters.info.parent_query = qry;
					}
				}
			}
		}
		else
		{
			Assert(!DsaPointerIsValid(entry->counters.info.parent_query));
		}
	}

	if (error_info)
	{
		entry->counters.error.elevel = error_info->elevel;
		strlcpy(entry->counters.error.sqlcode, error_info->sqlcode, SQLCODE_LEN);
		strlcpy(entry->counters.error.message, error_info->message, ERROR_MESSAGE_LEN);
	}

	entry->counters.calls.rows += rows;

	if (bufusage)
	{
		entry->counters.blocks.shared_blks_hit += bufusage->shared_blks_hit;
		entry->counters.blocks.shared_blks_read += bufusage->shared_blks_read;
		entry->counters.blocks.shared_blks_dirtied += bufusage->shared_blks_dirtied;
		entry->counters.blocks.shared_blks_written += bufusage->shared_blks_written;
		entry->counters.blocks.local_blks_hit += bufusage->local_blks_hit;
		entry->counters.blocks.local_blks_read += bufusage->local_blks_read;
		entry->counters.blocks.local_blks_dirtied += bufusage->local_blks_dirtied;
		entry->counters.blocks.local_blks_written += bufusage->local_blks_written;
		entry->counters.blocks.temp_blks_read += bufusage->temp_blks_read;
		entry->counters.blocks.temp_blks_written += bufusage->temp_blks_written;

#if PG_VERSION_NUM < 170000
		entry->counters.blocks.shared_blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage->blk_read_time);
		entry->counters.blocks.shared_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->blk_write_time);
#else
		entry->counters.blocks.shared_blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage->shared_blk_read_time);
		entry->counters.blocks.shared_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->shared_blk_write_time);
		entry->counters.blocks.local_blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage->local_blk_read_time);
		entry->counters.blocks.local_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->local_blk_write_time);
#endif

#if PG_VERSION_NUM >= 150000
		entry->counters.blocks.temp_blk_read_time += INSTR_TIME_GET_MILLISEC(bufusage->temp_blk_read_time);
		entry->counters.blocks.temp_blk_write_time += INSTR_TIME_GET_MILLISEC(bufusage->temp_blk_write_time);
#endif

#if PG_VERSION_NUM < 170000
		memcpy((void *) &entry->counters.blocks.instr_shared_blk_read_time, &bufusage->blk_read_time, sizeof(instr_time));
		memcpy((void *) &entry->counters.blocks.instr_shared_blk_write_time, &bufusage->blk_write_time, sizeof(instr_time));
#else
		memcpy((void *) &entry->counters.blocks.instr_shared_blk_read_time, &bufusage->shared_blk_read_time, sizeof(instr_time));
		memcpy((void *) &entry->counters.blocks.instr_shared_blk_write_time, &bufusage->shared_blk_write_time, sizeof(instr_time));
		memcpy((void *) &entry->counters.blocks.instr_local_blk_write_time, &bufusage->local_blk_write_time, sizeof(instr_time));
		memcpy((void *) &entry->counters.blocks.instr_local_blk_write_time, &bufusage->local_blk_write_time, sizeof(instr_time));
#endif


#if PG_VERSION_NUM >= 150000
		memcpy((void *) &entry->counters.blocks.instr_temp_blk_read_time, &bufusage->temp_blk_read_time, sizeof(bufusage->temp_blk_read_time));
		memcpy((void *) &entry->counters.blocks.instr_temp_blk_write_time, &bufusage->temp_blk_write_time, sizeof(bufusage->temp_blk_write_time));
#endif
	}

	entry->counters.calls.usage += USAGE_EXEC(exec_total_time + plan_total_time);

	if (sys_info)
	{
		entry->counters.sysinfo.utime += sys_info->utime;
		entry->counters.sysinfo.stime += sys_info->stime;
	}
	if (walusage)
	{
		entry->counters.walusage.wal_records += walusage->wal_records;
		entry->counters.walusage.wal_fpi += walusage->wal_fpi;
		entry->counters.walusage.wal_bytes += walusage->wal_bytes;
#if PG_VERSION_NUM >= 180000
		entry->counters.walusage.wal_buffers_full += walusage->wal_buffers_full;
#endif
	}
	if (jitusage)
	{
		entry->counters.jitinfo.jit_functions += jitusage->created_functions;
		entry->counters.jitinfo.jit_generation_time += INSTR_TIME_GET_MILLISEC(jitusage->generation_counter);

		if (INSTR_TIME_GET_MILLISEC(jitusage->inlining_counter))
			entry->counters.jitinfo.jit_inlining_count++;
		entry->counters.jitinfo.jit_inlining_time += INSTR_TIME_GET_MILLISEC(jitusage->inlining_counter);

		if (INSTR_TIME_GET_MILLISEC(jitusage->optimization_counter))
			entry->counters.jitinfo.jit_optimization_count++;
		entry->counters.jitinfo.jit_optimization_time += INSTR_TIME_GET_MILLISEC(jitusage->optimization_counter);

		if (INSTR_TIME_GET_MILLISEC(jitusage->emission_counter))
			entry->counters.jitinfo.jit_emission_count++;
		entry->counters.jitinfo.jit_emission_time += INSTR_TIME_GET_MILLISEC(jitusage->emission_counter);

#if PG_VERSION_NUM >= 170000
		if (INSTR_TIME_GET_MILLISEC(jitusage->deform_counter))
			entry->counters.jitinfo.jit_deform_count++;
		entry->counters.jitinfo.jit_deform_time += INSTR_TIME_GET_MILLISEC(jitusage->deform_counter);
#endif

		/* Only do this for local storage scenarios */
		if (kind != PGSM_STORE)
		{
			memcpy((void *) &entry->counters.jitinfo.instr_generation_counter, &jitusage->generation_counter, sizeof(instr_time));
			memcpy((void *) &entry->counters.jitinfo.instr_inlining_counter, &jitusage->inlining_counter, sizeof(instr_time));
			memcpy((void *) &entry->counters.jitinfo.instr_optimization_counter, &jitusage->optimization_counter, sizeof(instr_time));
			memcpy((void *) &entry->counters.jitinfo.instr_emission_counter, &jitusage->emission_counter, sizeof(instr_time));

#if PG_VERSION_NUM >= 170000
			memcpy((void *) &entry->counters.jitinfo.instr_deform_counter, &jitusage->deform_counter, sizeof(instr_time));
#endif
		}
	}

	/* parallel worker counters */
	entry->counters.parallel_workers_to_launch += parallel_workers_to_launch;
	entry->counters.parallel_workers_launched += parallel_workers_launched;

	if (kind == PGSM_STORE)
		SpinLockRelease(&entry->mutex);
}

static void
pgsm_store_error(const char *query, ErrorData *edata)
{
	pgsmEntry  *entry;
	int64		queryid = 0;
	int			len = strlen(query);

	if (!query || len == 0)
		return;

	len = strlen(query);

	queryid = pgsm_hash_string(query, len);

	entry = pgsm_create_hash_entry(0, queryid, NULL);
	entry->query_text.query_pointer = pnstrdup(query, len);

	entry->pgsm_query_id = get_pgsm_query_id_hash(query, len);

	entry->counters.error.elevel = edata->elevel;
	snprintf(entry->counters.error.message, ERROR_MESSAGE_LEN, "%s", edata->message);
	snprintf(entry->counters.error.sqlcode, SQLCODE_LEN, "%s", unpack_sql_state(edata->sqlerrcode));

	pgsm_store(entry);
}

static void
pgsm_add_to_list(pgsmEntry *entry, char *query_text, int query_len)
{
	/* Switch to pgsm memory context */
	MemoryContext oldctx = MemoryContextSwitchTo(GetPgsmMemoryContext());

	entry->query_text.query_pointer = pnstrdup(query_text, query_len);
	lentries = lappend(lentries, entry);
	MemoryContextSwitchTo(oldctx);
}

static void
pgsm_delete_entry(uint64 queryid)
{
	pgsmEntry  *entry = NULL;
	ListCell   *lc = NULL;

	if (lentries == NIL)
		return;

	entry = (pgsmEntry *) llast(lentries);
	if (entry->key.queryid == queryid)
	{
		pfree(entry->query_text.query_pointer);
		entry->query_text.query_pointer = NULL;
		lentries = list_delete_last(lentries);
		return;
	}

	/*
	 * The rest of the code is just paranoia. In theory this list is a stack,
	 * and we always want to remove the last item. Similarly, in the getter
	 * method we are always looking for the last item.
	 */

	foreach(lc, lentries)
	{
		entry = lfirst(lc);
		if (entry->key.queryid == queryid)
		{
			pfree(entry->query_text.query_pointer);
			entry->query_text.query_pointer = NULL;
			lentries = list_delete_cell(lentries, lc);
			return;
		}
	}
}

static pgsmEntry *
pgsm_get_entry_for_query(int64 queryid, PlanInfo *plan_info, const char *query_text, int query_len, bool create, CmdType cmd_type)
{
	pgsmEntry  *entry = NULL;
	ListCell   *lc = NULL;

	/* First bet is on the last entry */
	if (lentries == NIL && !create)
		return NULL;

	if (lentries)
	{
		entry = (pgsmEntry *) llast(lentries);
		if (entry->key.queryid == queryid)
			return entry;

		foreach(lc, lentries)
		{
			entry = lfirst(lc);
			if (entry->key.queryid == queryid)
				return entry;
		}
	}
	if (create && query_text)
	{
		/*
		 * At this point, we don't know which bucket this query will land in,
		 * so passing 0. The store function MUST later update it based on the
		 * current bucket value. The correct bucket value will be needed then
		 * to search the hash table, or create the appropriate entry.
		 */
		entry = pgsm_create_hash_entry(0, queryid, plan_info);

		/*
		 * Update other member that are not counters, so that we don't have to
		 * worry about these.
		 */
		entry->pgsm_query_id = get_pgsm_query_id_hash(query_text, query_len);
		entry->counters.info.cmd_type = cmd_type;
		pgsm_add_to_list(entry, (char *) query_text, query_len);
	}

	return entry;
}

static void
pgsm_cleanup_callback(void *arg)
{
	/* Reset the memory context holding the list */
	MemoryContextReset(GetPgsmMemoryContext());

	lentries = NIL;
	callback_setup = false;
}

/*
 * Function encapsulating some external calls for filling up the hash key data structure.
 * The bucket_id may not be known at this stage. So pass any value that you may wish.
 */
static pgsmEntry *
pgsm_create_hash_entry(uint64 bucket_id, int64 queryid, PlanInfo *plan_info)
{
	pgsmEntry  *entry;
	int			sec_ctx;
	bool		found_client_addr = false;
	MemoryContext oldctx;
	char	   *datname = NULL;
	char	   *username = NULL;

	/* Create an entry in the pgsm memory context */
	oldctx = MemoryContextSwitchTo(GetPgsmMemoryContext());
	entry = palloc0(sizeof(pgsmEntry));

	/*
	 * Get the user ID. Let's use this instead of GetUserID as this won't
	 * throw an assertion in case of an error.
	 */
	GetUserIdAndSecContext((Oid *) &entry->key.userid, &sec_ctx);

	if (pgsm_track_application_names)
	{
		/* Get the application name and set appid */
		app_name_len = pg_get_application_name(app_name, APPLICATIONNAME_LEN);
		entry->key.appid = pgsm_hash_string((const char *) app_name, app_name_len);
	}

	/* client address */
	if (!pgsm_client_ip_is_valid())
		pgsm_client_ip = pg_get_client_addr(&found_client_addr);

	entry->key.ip = pgsm_client_ip;

	/* PlanID, if there is one */
	entry->key.planid = plan_info ? plan_info->planid : 0;

	/* Set remaining data */
	entry->key.dbid = MyDatabaseId;
	entry->key.queryid = queryid;
	entry->key.bucket_id = bucket_id;
	entry->key.parentid = 0;

#if PG_VERSION_NUM >= 170000
	entry->key.toplevel = ((nesting_level) == 0);
#else
	entry->key.toplevel = ((nesting_level + plan_nested_level) == 0);
#endif

	if (IsTransactionState())
	{
		datname = get_database_name(entry->key.dbid);
		username = GetUserNameFromId(entry->key.userid, true);

		snprintf(entry->datname, sizeof(entry->datname), "%s", datname);
		snprintf(entry->username, sizeof(entry->username), "%s", username);

		if (datname)
			pfree(datname);

		if (username)
			pfree(username);
	}

	MemoryContextSwitchTo(oldctx);

	return entry;
}


/*
 * Store some statistics for a statement.
 *
 * If queryId is 0 then this is a utility statement and we should compute
 * a suitable queryId internally.
 *
 * If jstate is not NULL then we're trying to create an entry for which
 * we have no statistics as yet; we just want to record the normalized
 * query string.  total_time, rows, bufusage are ignored in this case.
 */
static void
pgsm_store(pgsmEntry *entry)
{
	pgsmEntry  *shared_hash_entry;
	pgsmSharedState *pgsm;
	bool		found;
	uint64		bucketid;
	uint64		prev_bucket_id;
	bool		reset = false;	/* Only used in update function - HAMID */
	char	   *query;
	int			query_len;
	BufferUsage bufusage;
	WalUsage	walusage;
	JitInstrumentation jitusage;
	char		comments[COMMENTS_LEN] = {0};
	int			comments_len;

	/* Safety check... */
	if (!IsSystemInitialized())
		return;

	pgsm = pgsm_get_ss();

	prev_bucket_id = pg_atomic_read_u64(&pgsm->current_wbucket);
	bucketid = get_next_wbucket(pgsm);

	if (bucketid != prev_bucket_id)
		reset = true;

	entry->key.bucket_id = bucketid;
	query = entry->query_text.query_pointer;
	query_len = strlen(query);

	/* Let's do all the leg work here before we acquire any locks */
	extract_query_comments(query, comments, sizeof(comments));
	comments_len = strlen(comments);

	/* bufusage */
	bufusage.shared_blks_hit = entry->counters.blocks.shared_blks_hit;
	bufusage.shared_blks_read = entry->counters.blocks.shared_blks_read;
	bufusage.shared_blks_dirtied = entry->counters.blocks.shared_blks_dirtied;
	bufusage.shared_blks_written = entry->counters.blocks.shared_blks_written;
	bufusage.local_blks_hit = entry->counters.blocks.local_blks_hit;
	bufusage.local_blks_read = entry->counters.blocks.local_blks_read;
	bufusage.local_blks_dirtied = entry->counters.blocks.local_blks_dirtied;
	bufusage.local_blks_written = entry->counters.blocks.local_blks_written;
	bufusage.temp_blks_read = entry->counters.blocks.temp_blks_read;
	bufusage.temp_blks_written = entry->counters.blocks.temp_blks_written;

#if PG_VERSION_NUM < 170000
	memcpy(&bufusage.blk_read_time, &entry->counters.blocks.instr_shared_blk_read_time, sizeof(instr_time));
	memcpy(&bufusage.blk_write_time, &entry->counters.blocks.instr_shared_blk_write_time, sizeof(instr_time));
#else
	memcpy(&bufusage.shared_blk_read_time, &entry->counters.blocks.instr_shared_blk_read_time, sizeof(instr_time));
	memcpy(&bufusage.shared_blk_write_time, &entry->counters.blocks.instr_shared_blk_write_time, sizeof(instr_time));
	memcpy(&bufusage.local_blk_read_time, &entry->counters.blocks.instr_local_blk_read_time, sizeof(instr_time));
	memcpy(&bufusage.local_blk_write_time, &entry->counters.blocks.instr_local_blk_write_time, sizeof(instr_time));
#endif

#if PG_VERSION_NUM >= 150000
	memcpy(&bufusage.temp_blk_read_time, &entry->counters.blocks.instr_temp_blk_read_time, sizeof(instr_time));
	memcpy(&bufusage.temp_blk_write_time, &entry->counters.blocks.instr_temp_blk_write_time, sizeof(instr_time));
#endif

	/* walusage */
	walusage.wal_records = entry->counters.walusage.wal_records;
	walusage.wal_fpi = entry->counters.walusage.wal_fpi;
	walusage.wal_bytes = entry->counters.walusage.wal_bytes;

#if PG_VERSION_NUM >= 180000
	walusage.wal_buffers_full = entry->counters.walusage.wal_buffers_full;
#endif

	/* jit */
	jitusage.created_functions = entry->counters.jitinfo.jit_functions;
	memcpy(&jitusage.generation_counter, &entry->counters.jitinfo.instr_generation_counter, sizeof(instr_time));
	memcpy(&jitusage.inlining_counter, &entry->counters.jitinfo.instr_inlining_counter, sizeof(instr_time));
	memcpy(&jitusage.optimization_counter, &entry->counters.jitinfo.instr_optimization_counter, sizeof(instr_time));
	memcpy(&jitusage.emission_counter, &entry->counters.jitinfo.instr_emission_counter, sizeof(instr_time));


	/* Update parent id if needed */
	if (pgsm_track == PGSM_TRACK_ALL && nesting_level > 0 && nesting_level < max_stack_depth)
	{
		entry->key.parentid = nested_queryids[nesting_level - 1];
	}
	else
	{
		entry->key.parentid = INT64CONST(0);
	}

#if PG_VERSION_NUM >= 170000
	memcpy(&jitusage.deform_counter, &entry->counters.jitinfo.instr_deform_counter, sizeof(instr_time));
#endif

	/*
	 * Acquire a share lock to start with. We'd have to acquire exclusive if
	 * we need to create the entry.
	 */
	pgsm_lock_aquire(pgsm, LW_SHARED);
	shared_hash_entry = (pgsmEntry *) pgsm_hash_find(get_pgsmHash(), &entry->key, &found);

	if (!shared_hash_entry)
	{
		dsa_pointer dsa_query_pointer;
		dsa_area   *query_dsa_area;
		char	   *query_buff;

		/* New query, truncate length if necessary. */
		if (query_len > pgsm_query_max_len)
			query_len = pg_mbcliplen(query, query_len, pgsm_query_max_len);

		/* Save the query text in raw dsa area */
		query_dsa_area = get_dsa_area_for_query_text();
		dsa_query_pointer = dsa_allocate_extended(query_dsa_area, query_len + 1, DSA_ALLOC_NO_OOM | DSA_ALLOC_ZERO);
		if (!DsaPointerIsValid(dsa_query_pointer))
		{
			pgsm_lock_release(pgsm);
			return;
		}

		/*
		 * Get the memory address from DSA pointer and copy the query text in
		 * local variable
		 */
		query_buff = dsa_get_address(query_dsa_area, dsa_query_pointer);
		memcpy(query_buff, query, query_len);

		pgsm_lock_release(pgsm);
		pgsm_lock_aquire(pgsm, LW_EXCLUSIVE);

		/* OK to create a new hashtable entry */
		PG_TRY();
		{
			shared_hash_entry = hash_entry_alloc(pgsm, &entry->key, GetDatabaseEncoding());
		}
		PG_CATCH();
		{
			if (DsaPointerIsValid(dsa_query_pointer))
				dsa_free(query_dsa_area, dsa_query_pointer);
			PG_RE_THROW();
		}
		PG_END_TRY();

		if (shared_hash_entry == NULL)
		{
			pgsm_lock_release(pgsm);

			if (DsaPointerIsValid(dsa_query_pointer))
				dsa_free(query_dsa_area, dsa_query_pointer);

			/*
			 * Out of memory; report only if the state has changed now.
			 * Otherwise we risk filling up the log file with these message.
			 */
			if (!IsSystemOOM())
			{
				pgsm->pgsm_oom = true;

				disable_error_capture = true;
				ereport(WARNING,
						(errcode(ERRCODE_OUT_OF_MEMORY),
						 errmsg("[pg_stat_monitor] pgsm_store: Hash table is out of memory and can no longer store queries!"),
						 errdetail("You may reset the view or when the buckets are deallocated, pg_stat_monitor will resume saving " \
								   "queries. Alternatively, try increasing the value of pg_stat_monitor.pgsm_max.")));
				disable_error_capture = false;
			}

			return;
		}
		else
		{
			/* If we got a new entry, reset the oom value false */
			pgsm->pgsm_oom = false;
		}

		/* If we already have the pointer set, free this one */
		if (DsaPointerIsValid(shared_hash_entry->query_text.query_pos))
			dsa_free(query_dsa_area, dsa_query_pointer);
		else
			shared_hash_entry->query_text.query_pos = dsa_query_pointer;

		shared_hash_entry->pgsm_query_id = entry->pgsm_query_id;
		shared_hash_entry->encoding = entry->encoding;
		shared_hash_entry->counters.info.cmd_type = entry->counters.info.cmd_type;
		shared_hash_entry->counters.info.parent_query = InvalidDsaPointer;

		snprintf(shared_hash_entry->datname, sizeof(shared_hash_entry->datname), "%s", entry->datname);
		snprintf(shared_hash_entry->username, sizeof(shared_hash_entry->username), "%s", entry->username);
	}

	pgsm_update_entry(shared_hash_entry,	/* entry */
					  query,	/* query */
					  comments, /* comments */
					  comments_len, /* comments length */
					  &entry->counters.planinfo,	/* PlanInfo */
					  &entry->counters.sysinfo, /* SysInfo */
					  &entry->counters.error,	/* ErrorInfo */
					  entry->counters.plantime.total_time,	/* plan_total_time */
					  entry->counters.time.total_time,	/* exec_total_time */
					  entry->counters.calls.rows,	/* rows */
					  &bufusage,	/* bufusage */
					  &walusage,	/* walusage */
					  &jitusage,	/* jitusage */
					  entry->counters.parallel_workers_to_launch,	/* parallel_workers_to_launch */
					  entry->counters.parallel_workers_launched,	/* parallel_workers_launched */
					  reset,	/* reset */
					  PGSM_STORE);

	if (reset)
	{
		shared_hash_entry->counters.info.cmd_type = entry->counters.info.cmd_type;
	}

	memset(&entry->counters, 0, sizeof(entry->counters));
	pgsm_lock_release(pgsm);
}

/*
 * Reset all statement statistics.
 */
Datum
pg_stat_monitor_reset(PG_FUNCTION_ARGS)
{
	pgsmSharedState *pgsm;

	/* Safety check... */
	if (!IsSystemInitialized())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("pg_stat_monitor: must be loaded via shared_preload_libraries")));

	pgsm = pgsm_get_ss();
	pgsm_lock_aquire(pgsm, LW_EXCLUSIVE);
	hash_entry_dealloc(INVALID_BUCKET_ID, INVALID_BUCKET_ID, NULL);

	pgsm_lock_release(pgsm);
	PG_RETURN_VOID();
}

Datum
pg_stat_monitor_1_0(PG_FUNCTION_ARGS)
{
	pg_stat_monitor_internal(fcinfo, PGSM_V1_0, true);
	return (Datum) 0;
}

Datum
pg_stat_monitor_2_0(PG_FUNCTION_ARGS)
{
	pg_stat_monitor_internal(fcinfo, PGSM_V2_0, true);
	return (Datum) 0;
}

Datum
pg_stat_monitor_2_1(PG_FUNCTION_ARGS)
{
	pg_stat_monitor_internal(fcinfo, PGSM_V2_1, true);
	return (Datum) 0;
}

Datum
pg_stat_monitor_2_3(PG_FUNCTION_ARGS)
{
	pg_stat_monitor_internal(fcinfo, PGSM_V2_3, true);
	return (Datum) 0;
}

/*
  * Legacy entry point for pg_stat_monitor() API versions 1.0
  */
Datum
pg_stat_monitor(PG_FUNCTION_ARGS)
{
	pg_stat_monitor_internal(fcinfo, PGSM_V1_0, true);
	return (Datum) 0;
}

static bool
IsBucketValid(uint64 bucketid)
{
	long		secs;
	int			microsecs;
	TimestampTz current_tz = GetCurrentTimestamp();
	pgsmSharedState *pgsm = pgsm_get_ss();

	TimestampDifference(pgsm->bucket_start_time[bucketid], current_tz, &secs, &microsecs);

	if (secs > ((int64) pgsm_bucket_time * pgsm_max_buckets))
		return false;
	return true;
}

/* Common code for all versions of pg_stat_monitor() */
static void
pg_stat_monitor_internal(FunctionCallInfo fcinfo,
						 pgsmVersion api_version,
						 bool showtext)
{
	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	TupleDesc	tupdesc;
	Tuplestorestate *tupstore;
	MemoryContext per_query_ctx;
	MemoryContext oldcontext;
	PGSM_HASH_SEQ_STATUS hstat;
	pgsmEntry  *entry;
	pgsmSharedState *pgsm;

	int			expected_columns;

	switch (api_version)
	{
		case PGSM_V1_0:
			expected_columns = PG_STAT_MONITOR_COLS_V1_0;
			break;
		case PGSM_V2_0:
			expected_columns = PG_STAT_MONITOR_COLS_V2_0;
			break;
		case PGSM_V2_1:
			expected_columns = PG_STAT_MONITOR_COLS_V2_1;
			break;
		case PGSM_V2_3:
			expected_columns = PG_STAT_MONITOR_COLS_V2_3;
			break;
		default:
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: Unknown API version")));
	}

	/* Disallow old api usage */
	if (api_version < PGSM_V2_0)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: API version not supported."),
				 errhint("Upgrade pg_stat_monitor extension")));
	/* Safety check... */
	if (!IsSystemInitialized())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: Must be loaded via shared_preload_libraries.")));

	/* Out of memory? */
	if (IsSystemOOM())
		ereport(WARNING,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: Hash table is out of memory and can no longer store queries!"),
				 errdetail("You may reset the view or when the buckets are deallocated, pg_stat_monitor will resume saving " \
						   "queries. Alternatively, try increasing the value of pg_stat_monitor.pgsm_max.")));

	/* check to see if caller supports us returning a tuplestore */
	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: Set-valued function called in context that cannot accept a set.")));
	if (!(rsinfo->allowedModes & SFRM_Materialize))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("[pg_stat_monitor] pg_stat_monitor_internal: Materialize mode required, but it is not " \
						"allowed in this context.")));

	/* Switch into long-lived context to construct returned data structures */
	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
	oldcontext = MemoryContextSwitchTo(per_query_ctx);

	/* Build a tuple descriptor for our result type */
	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "[pg_stat_monitor] pg_stat_monitor_internal: Return type must be a row type.");

	if (tupdesc->natts != expected_columns)
		elog(ERROR, "[pg_stat_monitor] pg_stat_monitor_internal: Incorrect number of output arguments, received %d, required %d.", tupdesc->natts, expected_columns);

	tupstore = tuplestore_begin_heap(true, false, work_mem);
	rsinfo->returnMode = SFRM_Materialize;
	rsinfo->setResult = tupstore;
	rsinfo->setDesc = tupdesc;

	MemoryContextSwitchTo(oldcontext);

	pgsm = pgsm_get_ss();
	pgsm_lock_aquire(pgsm, LW_SHARED);
	pgsm_hash_seq_init(&hstat, get_pgsmHash(), false);

	while ((entry = pgsm_hash_seq_next(&hstat)) != NULL)
	{
		Datum		values[PG_STAT_MONITOR_COLS] = {0};
		bool		nulls[PG_STAT_MONITOR_COLS] = {0};
		int			i = 0;
		Counters	tmp;
		pgsmHashKey tmpkey;
		double		stddev;
		int64		queryid = entry->key.queryid;
		int64		bucketid = entry->key.bucket_id;
		Oid			dbid = entry->key.dbid;
		Oid			userid = entry->key.userid;
		uint32		ip = entry->key.ip;
		int64		planid = entry->key.planid;
		int64		pgsm_query_id = entry->pgsm_query_id;
		dsa_area   *query_dsa_area;
		char	   *query_ptr;
		char	   *query_txt = NULL;
		char	   *parent_query_txt = NULL;

		bool		toplevel = entry->key.toplevel;
		bool		is_allowed_role = is_member_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS);

		/* Load the query text from dsa area */
		if (DsaPointerIsValid(entry->query_text.query_pos))
		{
			query_dsa_area = get_dsa_area_for_query_text();
			query_ptr = dsa_get_address(query_dsa_area, entry->query_text.query_pos);
			query_txt = pstrdup(query_ptr);
		}
		else
			query_txt = pstrdup("Query string not available");	/* Should never happen.
																 * Just a safety check */

		/* copy counters to a local variable to keep locking time short */
		{
			SpinLockAcquire(&entry->mutex);
			tmp = entry->counters;
			tmpkey = entry->key;
			SpinLockRelease(&entry->mutex);
		}

		/*
		 * In case that query plan is enabled, there is no need to show 0
		 * planid query
		 */
		if (tmp.info.cmd_type == CMD_SELECT && pgsm_enable_query_plan && planid == 0)
			continue;

		if (!IsBucketValid(bucketid))
		{
			continue;
		}

		/* read the parent query text if any */
		if (tmpkey.parentid != INT64CONST(0))
		{
			if (DsaPointerIsValid(tmp.info.parent_query))
			{
				query_dsa_area = get_dsa_area_for_query_text();
				query_ptr = dsa_get_address(query_dsa_area, tmp.info.parent_query);
				parent_query_txt = pstrdup(query_ptr);
			}
			else
				parent_query_txt = pstrdup("parent query text not available");
		}
		/* bucketid at column number 0 */
		values[i++] = Int64GetDatumFast(bucketid);

		/* userid at column number 1 */
		values[i++] = ObjectIdGetDatum(userid);

		/* username at column number 2 */
		values[i++] = CStringGetTextDatum(entry->username);

		/* dbid at column number 3 */
		values[i++] = ObjectIdGetDatum(dbid);

		/* datname at column number 4 */
		values[i++] = CStringGetTextDatum(entry->datname);

		/*
		 * ip address at column number 5, Superusers or members of
		 * pg_read_all_stats members are allowed
		 */
		if (is_allowed_role || userid == GetUserId())
			values[i++] = UInt32GetDatum(ip);
		else
			nulls[i++] = true;

		/* queryid at column number 6 */
		values[i++] = Int64GetDatum(queryid);

		/* planid at column number 7 */
		if (planid)
		{
			values[i++] = Int64GetDatum(planid);
		}
		else
		{
			nulls[i++] = true;
		}
		if (is_allowed_role || userid == GetUserId())
		{
			if (showtext)
			{
				char	   *enc;

				/* query at column number 8 */
				enc = pg_any_to_server(query_txt, strlen(query_txt), GetDatabaseEncoding());
				values[i++] = CStringGetTextDatum(enc);
				if (enc != query_txt)
					pfree(enc);
				/* plan at column number 9 */
				if (planid && tmp.planinfo.plan_text[0])
					values[i++] = CStringGetTextDatum(tmp.planinfo.plan_text);
				else
					nulls[i++] = true;
			}
			else
			{
				/* query at column number 8 */
				nulls[i++] = true;
				/* plan at column number 9 */
				nulls[i++] = true;
			}
		}
		else
		{
			/* query text and plan at column number 8 and 9 */
			values[i++] = CStringGetTextDatum("<insufficient privilege>");
			values[i++] = CStringGetTextDatum("<insufficient privilege>");
		}

		/* pgsm_query_id at column number 10 */
		if (pgsm_query_id)
			values[i++] = Int64GetDatum(pgsm_query_id);
		else
			nulls[i++] = true;

		/* parentid at column number 11 */
		if (tmpkey.parentid != INT64CONST(0))
		{
			values[i++] = Int64GetDatum(tmpkey.parentid);
			values[i++] = CStringGetTextDatum(parent_query_txt);
		}
		else
		{
			nulls[i++] = true;
			nulls[i++] = true;
		}

		/* application_name at column number 13 */
		if (strlen(tmp.info.application_name) > 0)
			values[i++] = CStringGetTextDatum(tmp.info.application_name);
		else
			nulls[i++] = true;

		/* relations at column number 14 */
		if (tmp.info.num_relations > 0)
		{
			int			j;
			char	   *text_str = palloc0(TOTAL_RELS_LENGTH);
			char	   *tmp_str = palloc0(TOTAL_RELS_LENGTH);
			bool		first = true;

			/*
			 * Need to calculate the actual size, and avoid unnessary memory
			 * usage
			 */
			for (j = 0; j < tmp.info.num_relations; j++)
			{
				if (first)
				{
					snprintf(text_str, 1024, "%s", tmp.info.relations[j]);
					first = false;
					continue;
				}
				snprintf(tmp_str, 1024, "%s,%s", text_str, tmp.info.relations[j]);
				snprintf(text_str, 1024, "%s", tmp_str);
			}
			pfree(tmp_str);
			values[i++] = CStringGetTextDatum(text_str);
		}
		else
			nulls[i++] = true;

		/* cmd_type at column number 15 */
		if (tmp.info.cmd_type == CMD_NOTHING)
			nulls[i++] = true;
		else
			values[i++] = Int64GetDatumFast((int64) tmp.info.cmd_type);

		/* elevel at column number 16 */
		values[i++] = Int64GetDatumFast(tmp.error.elevel);

		/* sqlcode at column number 17 */
		if (strlen(tmp.error.sqlcode) == 0)
			nulls[i++] = true;
		else
			values[i++] = CStringGetTextDatum(tmp.error.sqlcode);

		/* message at column number 18 */
		if (strlen(tmp.error.message) == 0)
			nulls[i++] = true;
		else
			values[i++] = CStringGetTextDatum(tmp.error.message);

		/* bucket_start_time at column number 19 */
		values[i++] = TimestampTzGetDatum(pgsm->bucket_start_time[entry->key.bucket_id]);

		if (tmp.calls.calls == 0)
		{
			/* Query of pg_stat_monitor itself started from zero count */
			tmp.calls.calls++;
			tmp.resp_calls[0]++;
		}

		/* calls at column number 20 */
		values[i++] = Int64GetDatumFast(tmp.calls.calls);

		/* total_time at column number 21 */
		values[i++] = Float8GetDatumFast(tmp.time.total_time);

		/* min_time at column number 22 */
		values[i++] = Float8GetDatumFast(tmp.time.min_time);

		/* max_time at column number 23 */
		values[i++] = Float8GetDatumFast(tmp.time.max_time);

		/* mean_time at column number 24 */
		values[i++] = Float8GetDatumFast(tmp.time.mean_time);
		if (tmp.calls.calls > 1)
			stddev = sqrt(tmp.time.sum_var_time / tmp.calls.calls);
		else
			stddev = 0.0;

		/* stddev_exec_time at column number 25 */
		values[i++] = Float8GetDatumFast(stddev);

		/* rows at column number 26 */
		values[i++] = Int64GetDatumFast(tmp.calls.rows);

		if (tmp.calls.calls == 0)
		{
			/* Query of pg_stat_monitor itslef started from zero count */
			tmp.calls.calls++;
			tmp.resp_calls[0]++;
		}

		/* plans at column number 27 */
		values[i++] = Int64GetDatumFast(tmp.plancalls.calls);

		/* total_plan_time at column number 28 */
		values[i++] = Float8GetDatumFast(tmp.plantime.total_time);

		/* min_plan_time at column number 29 */
		values[i++] = Float8GetDatumFast(tmp.plantime.min_time);

		/* max_plan_time at column number 30 */
		values[i++] = Float8GetDatumFast(tmp.plantime.max_time);

		/* mean_plan_time at column number 31 */
		values[i++] = Float8GetDatumFast(tmp.plantime.mean_time);
		if (tmp.plancalls.calls > 1)
			stddev = sqrt(tmp.plantime.sum_var_time / tmp.plancalls.calls);
		else
			stddev = 0.0;

		/* stddev_plan_time at column number 32 */
		values[i++] = Float8GetDatumFast(stddev);

		/* blocks are from column number 33 - 48 */
		values[i++] = Int64GetDatumFast(tmp.blocks.shared_blks_hit);
		values[i++] = Int64GetDatumFast(tmp.blocks.shared_blks_read);
		values[i++] = Int64GetDatumFast(tmp.blocks.shared_blks_dirtied);
		values[i++] = Int64GetDatumFast(tmp.blocks.shared_blks_written);
		values[i++] = Int64GetDatumFast(tmp.blocks.local_blks_hit);
		values[i++] = Int64GetDatumFast(tmp.blocks.local_blks_read);
		values[i++] = Int64GetDatumFast(tmp.blocks.local_blks_dirtied);
		values[i++] = Int64GetDatumFast(tmp.blocks.local_blks_written);
		values[i++] = Int64GetDatumFast(tmp.blocks.temp_blks_read);
		values[i++] = Int64GetDatumFast(tmp.blocks.temp_blks_written);
		values[i++] = Float8GetDatumFast(tmp.blocks.shared_blk_read_time);
		values[i++] = Float8GetDatumFast(tmp.blocks.shared_blk_write_time);
		if (api_version >= PGSM_V2_1)
		{
			values[i++] = Float8GetDatumFast(tmp.blocks.local_blk_read_time);
			values[i++] = Float8GetDatumFast(tmp.blocks.local_blk_write_time);
		}
		values[i++] = Float8GetDatumFast(tmp.blocks.temp_blk_read_time);
		values[i++] = Float8GetDatumFast(tmp.blocks.temp_blk_write_time);

		/* resp_calls at column number 49 */
		values[i++] = intarray_get_datum(tmp.resp_calls, hist_bucket_count_total);

		/* cpu_user_time at column number 50 */
		values[i++] = Float8GetDatumFast(tmp.sysinfo.utime);

		/* cpu_sys_time at column number 51 */
		values[i++] = Float8GetDatumFast(tmp.sysinfo.stime);
		{
			char		buf[256];
			Datum		wal_bytes;

			/* wal_records at column number 52 */
			values[i++] = Int64GetDatumFast(tmp.walusage.wal_records);

			/* wal_fpi at column number 53 */
			values[i++] = Int64GetDatumFast(tmp.walusage.wal_fpi);

			snprintf(buf, sizeof buf, UINT64_FORMAT, tmp.walusage.wal_bytes);

			/* Convert to numeric */
			wal_bytes = DirectFunctionCall3(numeric_in,
											CStringGetDatum(buf),
											ObjectIdGetDatum(0),
											Int32GetDatum(-1));
			/* wal_bytes at column number 54 */
			values[i++] = wal_bytes;

			if (api_version >= PGSM_V2_3)
			{
				/* wal_buffers_full at column number 55 */
				values[i++] = Int64GetDatumFast(tmp.walusage.wal_buffers_full);
			}

			/* application_name at column number 56 */
			if (strlen(tmp.info.comments) > 0)
				values[i++] = CStringGetTextDatum(tmp.info.comments);
			else
				nulls[i++] = true;

			/* blocks are from column number 57 - 64 */
			values[i++] = Int64GetDatumFast(tmp.jitinfo.jit_functions);
			values[i++] = Float8GetDatumFast(tmp.jitinfo.jit_generation_time);
			values[i++] = Int64GetDatumFast(tmp.jitinfo.jit_inlining_count);
			values[i++] = Float8GetDatumFast(tmp.jitinfo.jit_inlining_time);
			values[i++] = Int64GetDatumFast(tmp.jitinfo.jit_optimization_count);
			values[i++] = Float8GetDatumFast(tmp.jitinfo.jit_optimization_time);
			values[i++] = Int64GetDatumFast(tmp.jitinfo.jit_emission_count);
			values[i++] = Float8GetDatumFast(tmp.jitinfo.jit_emission_time);
			if (api_version >= PGSM_V2_1)
			{
				/* at column number 65 */
				values[i++] = Int64GetDatumFast(tmp.jitinfo.jit_deform_count);
				values[i++] = Float8GetDatumFast(tmp.jitinfo.jit_deform_time);
			}
		}

		if (api_version >= PGSM_V2_3)
		{
			/* at column number 67 */
			values[i++] = Int64GetDatumFast(tmp.parallel_workers_to_launch);
			values[i++] = Int64GetDatumFast(tmp.parallel_workers_launched);
		}

		if (api_version >= PGSM_V2_1)
		{
			/* at column number 69 */
			values[i++] = TimestampTzGetDatum(entry->stats_since);
			values[i++] = TimestampTzGetDatum(entry->minmax_stats_since);
		}

		/* toplevel at column number 71 */
		values[i++] = BoolGetDatum(toplevel);

		/* bucket_done at column number 72 */
		values[i++] = BoolGetDatum(pg_atomic_read_u64(&pgsm->current_wbucket) != bucketid);

		/* clean up and return the tuplestore */
		tuplestore_putvalues(tupstore, tupdesc, values, nulls);

		if (query_txt)
			pfree(query_txt);
		if (parent_query_txt)
			pfree(parent_query_txt);
	}
	/* clean up and return the tuplestore */
	pgsm_hash_seq_term(&hstat);
	pgsm_lock_release(pgsm);
}

static uint64
get_next_wbucket(pgsmSharedState *pgsm)
{
	struct timeval tv;
	uint64		current_bucket_sec;
	uint64		new_bucket_id;
	uint64		prev_bucket_id;
	bool		update_bucket = false;

	gettimeofday(&tv, NULL);
	current_bucket_sec = pg_atomic_read_u64(&pgsm->prev_bucket_sec);

	/*
	 * If current bucket expired we loop attempting to update prev_bucket_sec.
	 *
	 * pg_atomic_compare_exchange_u64 may fail in two possible ways: 1.
	 * Another thread/process updated the variable before us. 2. A spurious
	 * failure / hardware event.
	 *
	 * In both failure cases we read prev_bucket_sec from memory again, if it
	 * was a spurious failure then the value of prev_bucket_sec must be the
	 * same as before, which will cause the while loop to execute again.
	 *
	 * If another thread updated prev_bucket_sec, then its current value will
	 * definitely make the while condition to fail, we can stop the loop as
	 * another thread has already updated prev_bucket_sec.
	 */
	while ((tv.tv_sec - (uint) current_bucket_sec) >= ((uint) pgsm_bucket_time))
	{
		if (pg_atomic_compare_exchange_u64(&pgsm->prev_bucket_sec, &current_bucket_sec, (uint64) tv.tv_sec))
		{
			update_bucket = true;
			break;
		}

		current_bucket_sec = pg_atomic_read_u64(&pgsm->prev_bucket_sec);
	}

	if (update_bucket)
	{

		new_bucket_id = (tv.tv_sec / pgsm_bucket_time) % pgsm_max_buckets;

		/* Update bucket id and retrieve the previous one. */
		prev_bucket_id = pg_atomic_exchange_u64(&pgsm->current_wbucket, new_bucket_id);

		pgsm_lock_aquire(pgsm, LW_EXCLUSIVE);
		hash_entry_dealloc(new_bucket_id, prev_bucket_id, NULL);

		pgsm_lock_release(pgsm);

		/* Align the value in prev_bucket_sec to the bucket start time */
		tv.tv_sec = (tv.tv_sec) - (tv.tv_sec % pgsm_bucket_time);

		pg_atomic_exchange_u64(&pgsm->prev_bucket_sec, (uint64) tv.tv_sec);

		pgsm->bucket_start_time[new_bucket_id] = (TimestampTz) tv.tv_sec -
			((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
		pgsm->bucket_start_time[new_bucket_id] = pgsm->bucket_start_time[new_bucket_id] * USECS_PER_SEC;
		return new_bucket_id;
	}

	return pg_atomic_read_u64(&pgsm->current_wbucket);
}

/*
 * This function expects a NORMALIZED query as the input.
 * It iterates over the normalized query skipping comments and
 * multiple spaces. All spaces are converted to ' ' so that we
 * the calculation is independent of the space type whether
 * newline, tab, or any other type. Trailing and leading spaces
 * are also removed before calculating the hash.
 */
int64
get_pgsm_query_id_hash(const char *norm_query, int norm_len)
{
	char	   *query;
	char	   *q_iter;
	char	   *norm_q_iter = (char *) norm_query;
	int64		pgsm_query_id = 0;

	if (!pgsm_enable_pgsm_query_id)
		return 0;

	query = palloc(norm_len + 1);
	q_iter = query;

	while (norm_q_iter && *norm_q_iter && norm_q_iter < (norm_query + norm_len))
	{
		/*
		 * Skip multiline comments, + 1 is safe even if we've reach end of
		 * string
		 */
		if (*norm_q_iter == '/' && *(norm_q_iter + 1) == '*')
		{
			while (*norm_q_iter && *(norm_q_iter + 1) && (*norm_q_iter != '*' || *(norm_q_iter + 1) != '/'))
				norm_q_iter++;

			/*
			 * Skip the end if the current character is valid. norm_q_iter
			 * points to the *, we have to skip 2 characters
			 */
			if (*norm_q_iter)
				norm_q_iter++;
			if (*norm_q_iter)
				norm_q_iter++;

			continue;
		}

		/*
		 * Skip single line comments, + 1 is safe even if we've reach end of
		 * string
		 */
		if (*norm_q_iter == '-' && *(norm_q_iter + 1) == '-')
		{
			while (*norm_q_iter && *norm_q_iter != '\n')
				norm_q_iter++;
		}

		/* Skip white spaces */
		if (scanner_isspace(*norm_q_iter))
		{
			while (scanner_isspace(*++norm_q_iter));

			/*
			 * Let's replace it with a simple space. -1 is safe as we are
			 * making sure we are not at the start of the string.
			 */
			if (q_iter != query && !scanner_isspace(*(q_iter - 1)))
				*q_iter++ = ' ';

			continue;
		}

		*q_iter++ = *norm_q_iter++;
	}

	/* Ensure we have a terminating zero at the end */
	*q_iter = '\0';

	/* Get rid of trailing spaces */
	while (q_iter > query && *q_iter == '\0')
	{
		q_iter--;

		/* Continue reducing the string size if space is found. */
		if (scanner_isspace(*q_iter))
			*q_iter = '\0';
	}

	/* Calculate the hash. */
	pgsm_query_id = pgsm_hash_string(query, strlen(query));

	pfree(query);
	return pgsm_query_id;
}

/*
 * Generate a normalized version of the query string that will be used to
 * represent all similar queries.
 *
 * Note that the normalized representation may well vary depending on
 * just which "equivalent" query is used to create the hashtable entry.
 * We assume this is OK.
 *
 * If query_loc > 0, then "query" has been advanced by that much compared to
 * the original string start, so we need to translate the provided locations
 * to compensate.  (This lets us avoid re-scanning statements before the one
 * of interest, so it's worth doing.)
 *
 * *query_len_p contains the input string length, and is updated with
 * the result string length on exit.  The resulting string might be longer
 * or shorter depending on what happens with replacement of constants.
 *
 * Returns a palloc'd string.
 */
static char *
generate_normalized_query(JumbleState *jstate, const char *query,
						  int query_loc, int *query_len_p, int encoding)
{
	char	   *norm_query;
	int			query_len = *query_len_p;
	int			norm_query_buflen,	/* Space allowed for norm_query */
				len_to_wrt,		/* Length (in bytes) to write */
				quer_loc = 0,	/* Source query byte location */
				n_quer_loc = 0, /* Normalized query byte location */
				last_off = 0,	/* Offset from start for previous tok */
				last_tok_len = 0;	/* Length (in bytes) of that tok */
#if PG_VERSION_NUM >= 180000
	int			num_constants_replaced = 0;
#endif

	/*
	 * Get constants' lengths (core system only gives us locations).  Note
	 * this also ensures the items are sorted by location.
	 */
	fill_in_constant_lengths(jstate, query, query_loc);

	/*
	 * Allow for $n symbols to be longer than the constants they replace.
	 * Constants must take at least one byte in text form, while a $n symbol
	 * certainly isn't more than 11 bytes, even if n reaches INT_MAX.  We
	 * could refine that limit based on the max value of n for the current
	 * query, but it hardly seems worth any extra effort to do so.
	 */
	norm_query_buflen = query_len + jstate->clocations_count * 10;

	/* Allocate result buffer */
	norm_query = palloc(norm_query_buflen + 1);

	for (int i = 0; i < jstate->clocations_count; i++)
	{
		int			off,		/* Offset from start for cur tok */
					tok_len;	/* Length (in bytes) of that tok */

#if PG_VERSION_NUM >= 180000

		/*
		 * If we have an external param at this location, but no lists are
		 * being squashed across the query, then we skip here; this will make
		 * us print the characters found in the original query that represent
		 * the parameter in the next iteration (or after the loop is done),
		 * which is a bit odd but seems to work okay in most cases.
		 */
		if (jstate->clocations[i].extern_param && !jstate->has_squashed_lists)
			continue;
#endif

		off = jstate->clocations[i].location;
		/* Adjust recorded location if we're dealing with partial string */
		off -= query_loc;

		tok_len = jstate->clocations[i].length;

		if (tok_len < 0)
			continue;			/* ignore any duplicates */

		/* Copy next chunk (what precedes the next constant) */
		len_to_wrt = off - last_off;
		len_to_wrt -= last_tok_len;

		Assert(len_to_wrt >= 0);
		memcpy(norm_query + n_quer_loc, query + quer_loc, len_to_wrt);
		n_quer_loc += len_to_wrt;

#if PG_VERSION_NUM >= 180000

		/*
		 * And insert a param symbol in place of the constant token; and, if
		 * we have a squashable list, insert a placeholder comment starting
		 * from the list's second value.
		 */
		n_quer_loc += sprintf(norm_query + n_quer_loc, "$%d%s",
							  num_constants_replaced + 1 + jstate->highest_extern_param_id,
							  jstate->clocations[i].squashed ? " /*, ... */" : "");
		num_constants_replaced++;
#else
		/* And insert a param symbol in place of the constant token */
		n_quer_loc += sprintf(norm_query + n_quer_loc, "$%d",
							  i + 1 + jstate->highest_extern_param_id);
#endif

		/* move forward */
		quer_loc = off + tok_len;
		last_off = off;
		last_tok_len = tok_len;
	}

	/*
	 * We've copied up until the last ignorable constant.  Copy over the
	 * remaining bytes of the original query string.
	 */
	len_to_wrt = query_len - quer_loc;

	Assert(len_to_wrt >= 0);
	memcpy(norm_query + n_quer_loc, query + quer_loc, len_to_wrt);
	n_quer_loc += len_to_wrt;

	Assert(n_quer_loc <= norm_query_buflen);
	norm_query[n_quer_loc] = '\0';

	*query_len_p = n_quer_loc;
	return norm_query;
}

/*
 * Given a valid SQL string and an array of constant-location records,
 * fill in the textual lengths of those constants.
 *
 * The constants may use any allowed constant syntax, such as float literals,
 * bit-strings, single-quoted strings and dollar-quoted strings.  This is
 * accomplished by using the public API for the core scanner.
 *
 * It is the caller's job to ensure that the string is a valid SQL statement
 * with constants at the indicated locations.  Since in practice the string
 * has already been parsed, and the locations that the caller provides will
 * have originated from within the authoritative parser, this should not be
 * a problem.
 *
 * Duplicate constant pointers are possible, and will have their lengths
 * marked as '-1', so that they are later ignored.  (Actually, we assume the
 * lengths were initialized as -1 to start with, and don't change them here.)
 *
 * If query_loc > 0, then "query" has been advanced by that much compared to
 * the original string start, so we need to translate the provided locations
 * to compensate.  (This lets us avoid re-scanning statements before the one
 * of interest, so it's worth doing.)
 *
 * N.B. There is an assumption that a '-' character at a Const location begins
 * a negative numeric constant.  This precludes there ever being another
 * reason for a constant to start with a '-'.
 */
static void
fill_in_constant_lengths(JumbleState *jstate, const char *query,
						 int query_loc)
{
	LocationLen *locs;
	core_yyscan_t yyscanner;
	core_yy_extra_type yyextra;
	core_YYSTYPE yylval;
	YYLTYPE		yylloc;
	int			last_loc = -1;
	int			i;

	/*
	 * Sort the records by location so that we can process them in order while
	 * scanning the query text.
	 */
	if (jstate->clocations_count > 1)
		qsort(jstate->clocations, jstate->clocations_count,
			  sizeof(LocationLen), comp_location);
	locs = jstate->clocations;

	/* initialize the flex scanner --- should match raw_parser() */
	yyscanner = scanner_init(query,
							 &yyextra,
							 &ScanKeywords,
							 ScanKeywordTokens);

	/* we don't want to re-emit any escape string warnings */
	yyextra.escape_string_warning = false;

	/* Search for each constant, in sequence */
	for (i = 0; i < jstate->clocations_count; i++)
	{
		int			loc = locs[i].location;
		int			tok;

		/* Adjust recorded location if we're dealing with partial string */
		loc -= query_loc;

		Assert(loc >= 0);

#if PG_VERSION_NUM >= 180000
		if (locs[i].squashed)
			continue;			/* squashable list, ignore */
#endif

		if (loc <= last_loc)
			continue;			/* Duplicate constant, ignore */

		/* Lex tokens until we find the desired constant */
		for (;;)
		{
			tok = core_yylex(&yylval, &yylloc, yyscanner);

			/* We should not hit end-of-string, but if we do, behave sanely */
			if (tok == 0)
				break;			/* out of inner for-loop */

			/*
			 * We should find the token position exactly, but if we somehow
			 * run past it, work with that.
			 */
			if (yylloc >= loc)
			{
				if (query[loc] == '-')
				{
					/*
					 * It's a negative value - this is the one and only case
					 * where we replace more than a single token.
					 *
					 * Do not compensate for the core system's special-case
					 * adjustment of location to that of the leading '-'
					 * operator in the event of a negative constant.  It is
					 * also useful for our purposes to start from the minus
					 * symbol.  In this way, queries like "select * from foo
					 * where bar = 1" and "select * from foo where bar = -2"
					 * will have identical normalized query strings.
					 */
					tok = core_yylex(&yylval, &yylloc, yyscanner);
					if (tok == 0)
						break;	/* out of inner for-loop */
				}

				/*
				 * We now rely on the assumption that flex has placed a zero
				 * byte after the text of the current token in scanbuf.
				 */
				locs[i].length = strlen(yyextra.scanbuf + loc);
				break;			/* out of inner for-loop */
			}
		}

		/* If we hit end-of-string, give up, leaving remaining lengths -1 */
		if (tok == 0)
			break;

		last_loc = loc;
	}

	scanner_finish(yyscanner);
}

/*
 * comp_location: comparator for qsorting LocationLen structs by location
 */
static int
comp_location(const void *a, const void *b)
{
	int			l = ((const LocationLen *) a)->location;
	int			r = ((const LocationLen *) b)->location;

	if (l < r)
		return -1;
	else if (l > r)
		return +1;
	else
		return 0;
}

#define MAX_STRING_LEN	1024

/* Convert array of integers into Text datum */
static Datum
intarray_get_datum(int32 arr[], int len)
{
	StringInfoData str;
	Datum		datum;

	if (len < 1)
		return CStringGetTextDatum("");

	initStringInfo(&str);
	appendStringInfo(&str, "%d", arr[0]);

	for (int i = 1; i < len; i++)
		appendStringInfo(&str, ",%d", arr[i]);

	datum = CStringGetTextDatum(str.data);
	pfree(str.data);

	return datum;
}


Datum
pg_stat_monitor_hook_stats(PG_FUNCTION_ARGS)
{
	return (Datum) 0;
}


void
pgsm_emit_log_hook(ErrorData *edata)
{
	if (!IsSystemInitialized() || edata == NULL)
		goto exit;

	if (IsParallelWorker())
		goto exit;

	/* Check if PostgreSQL has finished its own bootstrapping code. */
	if (MyProc == NULL)
		goto exit;

	if (edata->elevel >= WARNING && !disable_error_capture && IsSystemOOM() == false)
		pgsm_store_error(debug_query_string ? debug_query_string : "", edata);

	/* We need to make sure we re-enable error capture if query was aborted */
	if (edata->elevel >= ERROR)
		disable_error_capture = false;
exit:
	if (prev_emit_log_hook)
		prev_emit_log_hook(edata);
}

bool
IsSystemInitialized(void)
{
	return (system_init && IsHashInitialize());
}

static double
time_diff(struct timeval end, struct timeval start)
{
	double		mstart;
	double		mend;

	mend = ((double) end.tv_sec * 1000.0 + (double) end.tv_usec / 1000.0);
	mstart = ((double) start.tv_sec * 1000.0 + (double) start.tv_usec / 1000.0);
	return mend - mstart;
}

char *
unpack_sql_state(int sql_state)
{
	static char buf[12];
	int			i;

	for (i = 0; i < 5; i++)
	{
		buf[i] = PGUNSIXBIT(sql_state);
		sql_state >>= 6;
	}

	buf[i] = '\0';
	return buf;
}

/* Validate histogram values and find the max number of histogram buckets that can be created */
static void
set_histogram_bucket_timings(void)
{
	double		b2_start;
	double		b2_end;
	int			b_count;

	hist_bucket_min = pgsm_histogram_min;
	hist_bucket_max = pgsm_histogram_max;
	hist_bucket_count_user = pgsm_histogram_buckets;
	b_count = hist_bucket_count_user;

	if (pgsm_histogram_buckets >= 2)
	{
		for (; hist_bucket_count_user > 0; hist_bucket_count_user--)
		{
			histogram_bucket_timings(2, &b2_start, &b2_end);

			/*
			 * The first bucket size will always be one or greater as we're
			 * doing min value + e^0; and e^0 = 1. Checking if histograms
			 * buckets overlap. That can only happen if the second bucket size
			 * is zero as we using exponential bucket sizes. Therefore, if the
			 * second bucket size is greater than 1, we'll never have
			 * overlapping buckets.
			 */
			if (b2_start != b2_end)
			{
				break;
			}
		}

		if (b_count != hist_bucket_count_user)
			ereport(WARNING,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("pg_stat_monitor: Histogram buckets are overlapping."),
					 errdetail("Histogram bucket size is set to %d [not including outlier buckets].", hist_bucket_count_user)));
	}

	/*
	 * Important that we keep user bucket count separate for calculations, but
	 * must add 1 for max outlier queries. However, for min, bucket should
	 * only be added if the minimum value provided by user is greater than 0
	 */
	hist_bucket_count_total = (hist_bucket_count_user + (int) (hist_bucket_max < HISTOGRAM_MAX_TIME) + (int) (hist_bucket_min > 0));

	for (b_count = 0; b_count < hist_bucket_count_total; b_count++)
	{
		histogram_bucket_timings(b_count, &hist_bucket_timings[b_count][HISTOGRAM_START], &hist_bucket_timings[b_count][HISTOGRAM_END]);
	}
}

/*
 * Given an index, return the histogram start and end times.
 */
static void
histogram_bucket_timings(int index, double *b_start, double *b_end)
{
	double		q_min = hist_bucket_min;
	double		q_max = hist_bucket_max;
	int			b_count = hist_bucket_count_total;
	int			b_count_user = hist_bucket_count_user;
	double		bucket_size;

	/*
	 * We must not skip any queries that fall outside the user defined
	 * histogram buckets. So capturing min/max outliers.
	 */
	if (index == 0 && q_min > 0)
	{
		*b_start = 0;
		*b_end = q_min;
		return;
	}
	else if (index == (b_count - 1) && q_max < HISTOGRAM_MAX_TIME)
	{
		*b_start = q_max;
		*b_end = -1;
		return;
	}

	/*
	 * Equisized logarithmic values will yield exponential values as required.
	 * For calculating logarithmic value, we MUST use the number of bucket
	 * provided by the user.
	 */
	bucket_size = log(q_max - q_min) / (double) b_count_user;

	/*
	 * Can't do exp(0) as that returns 1. So handling the case of first entry
	 * specifically
	 */
	*b_start = q_min + ((index == 0 || (index == 1 && q_min > 0)) ? 0 : exp(bucket_size * (index - 1 + (q_min == 0))));
	*b_end = q_min + exp(bucket_size * (index + (q_min == 0)));
}

/*
 * Get the histogram bucket index for a given query time.
 */
static int
get_histogram_bucket(double q_time)
{
	int			index = 0;
	double		exec_time = q_time;

	for (index = 0; index < hist_bucket_count_total; index++)
	{
		if (exec_time >= hist_bucket_timings[index][HISTOGRAM_START] && exec_time <= hist_bucket_timings[index][HISTOGRAM_END])
			return index;
	}

	/*
	 * So haven't found a histogram bucket for this query. That's only
	 * possible for the last bucket as its end time is less than 0.
	 */
	return (hist_bucket_count_total - 1);
}

/*
 * Get the timings of the histogram as a single string. The last bucket
 * has ellipses as the end value indication infinity.
 */
Datum
get_histogram_timings(PG_FUNCTION_ARGS)
{
	double		b_start;
	double		b_end;
	int			b_count = hist_bucket_count_total;
	int			index = 0;
	char	   *tmp_str = palloc0(MAX_STRING_LEN);
	char	   *text_str = palloc0(MAX_STRING_LEN);

	for (index = 0; index < b_count; index++)
	{
		histogram_bucket_timings(index, &b_start, &b_end);

		if (index == 0)
		{
			snprintf(text_str, MAX_STRING_LEN, "{{%.3lf - %.3lf}", b_start, b_end);
		}
		else if (index == (b_count - 1))
		{
			snprintf(tmp_str, MAX_STRING_LEN, "%s, (%.3lf - ...}}", text_str, b_start);
			snprintf(text_str, MAX_STRING_LEN, "%s", tmp_str);
		}
		else
		{
			snprintf(tmp_str, MAX_STRING_LEN, "%s, (%.3lf - %.3lf}", text_str, b_start, b_end);
			snprintf(text_str, MAX_STRING_LEN, "%s", tmp_str);
		}
	}

	pfree(tmp_str);

	return CStringGetTextDatum(text_str);
}


static bool
append_comment_char(char *comments, size_t max_len, size_t *idx, char c)
{
	if (*idx >= max_len)
		return false;

	comments[*idx] = c;
	(*idx)++;

	return true;
}

static void
extract_query_comments(const char *query, char *comments, size_t max_len)
{
	size_t		curr_len = 0;
	const char *q_iter = query;

	if (!pgsm_extract_comments)
	{
		comments[0] = 0;
		return;
	}

	while (q_iter && *q_iter)
	{
		/*
		 * multiline comments, + 1 is safe even if we've reach end of string
		 */
		if (*q_iter == '/' && *(q_iter + 1) == '*')
		{
			if (curr_len != 0)
			{
				if (!append_comment_char(comments, max_len, &curr_len, ','))
					return;
				if (!append_comment_char(comments, max_len, &curr_len, ' '))
					return;
			}
			while (*q_iter && *(q_iter + 1) && (*q_iter != '*' || *(q_iter + 1) != '/'))
			{
				if (!append_comment_char(comments, max_len, &curr_len, *q_iter))
					return;
				q_iter++;
			}

			if (*q_iter)
			{
				if (!append_comment_char(comments, max_len, &curr_len, *q_iter))
					return;
				q_iter++;
			}
			if (*q_iter)
			{
				if (!append_comment_char(comments, max_len, &curr_len, *q_iter))
					return;
				q_iter++;
			}
		}

		q_iter++;
	}

	comments[curr_len] = 0;
}

static void
pgsm_lock_aquire(pgsmSharedState *pgsm, LWLockMode mode)
{
	/* Disable error capturing while holding the lock to avoid deadlocks */
	LWLockAcquire(pgsm->lock, mode);
	disable_error_capture = true;
}

static void
pgsm_lock_release(pgsmSharedState *pgsm)
{
	disable_error_capture = false;
	LWLockRelease(pgsm->lock);
}