Isolate parallel leader from workers

contrib/pg_stat_statements/pg_stat_statements.c

@@ -362,11 +362,7 @@ static void pgss_ExecutorRun(QueryDesc *queryDesc,
 							 uint64 count);
 static void pgss_ExecutorFinish(QueryDesc *queryDesc);
 static void pgss_ExecutorEnd(QueryDesc *queryDesc);
-static void pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
-								bool readOnlyTree,
-								ProcessUtilityContext context, ParamListInfo params,
-								QueryEnvironment *queryEnv,
-								DestReceiver *dest, QueryCompletion *qc);
+static void pgss_ProcessUtility(UtilityStmtDesc *desc);
 static void pgss_store(const char *query, int64 queryId,
 					   int query_location, int query_len,
 					   bool toplevel,
@@ -1125,12 +1121,11 @@ pgss_ExecutorEnd(QueryDesc *queryDesc)
  * ProcessUtility hook
  */
 static void
-pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
-					bool readOnlyTree,
-					ProcessUtilityContext context,
-					ParamListInfo params, QueryEnvironment *queryEnv,
-					DestReceiver *dest, QueryCompletion *qc)
+pgss_ProcessUtility(UtilityStmtDesc *desc)
 {
+	PlannedStmt *pstmt = desc->pstmt;
+	const char *queryString = desc->queryString;
+	QueryCompletion *qc = desc->qc;
 	Node	   *parsetree = pstmt->utilityStmt;
 	int64		saved_queryId = pstmt->queryId;
 	int			saved_stmt_location = pstmt->stmt_location;
@@ -1172,29 +1167,22 @@ pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
 		!IsA(parsetree, ExecuteStmt) &&
 		!IsA(parsetree, PrepareStmt))
 	{
-		instr_time	start;
-		instr_time	duration;
 		uint64		rows;
-		BufferUsage bufusage_start,
-					bufusage;
-		WalUsage	walusage_start,
-					walusage;
 
-		bufusage_start = pgBufferUsage;
-		walusage_start = pgWalUsage;
-		INSTR_TIME_SET_CURRENT(start);
+		/*
+		 * Request statement-level instrumentation (timing, buffers and WAL)
+		 * so that we can read the command's resource usage from desc->instr
+		 * below, the same way we read query_instr for optimizable statements.
+		 */
+		desc->instr_options |= INSTRUMENT_ALL;
 
 		nesting_level++;
 		PG_TRY();
 		{
 			if (prev_ProcessUtility)
-				prev_ProcessUtility(pstmt, queryString, readOnlyTree,
-									context, params, queryEnv,
-									dest, qc);
+				prev_ProcessUtility(desc);
 			else
-				standard_ProcessUtility(pstmt, queryString, readOnlyTree,
-										context, params, queryEnv,
-										dest, qc);
+				standard_ProcessUtility(desc);
 		}
 		PG_FINALLY();
 		{
@@ -1213,9 +1201,6 @@ pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
 		 */
 		pstmt = NULL;
 
-		INSTR_TIME_SET_CURRENT(duration);
-		INSTR_TIME_SUBTRACT(duration, start);
-
 		/*
 		 * Track the total number of rows retrieved or affected by the utility
 		 * statements of COPY, FETCH, CREATE TABLE AS, CREATE MATERIALIZED
@@ -1227,29 +1212,27 @@ pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
 					   qc->commandTag == CMDTAG_REFRESH_MATERIALIZED_VIEW)) ?
 			qc->nprocessed : 0;
 
-		/* 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);
-
-		pgss_store(queryString,
-				   saved_queryId,
-				   saved_stmt_location,
-				   saved_stmt_len,
-				   nesting_level == 0,
-				   PGSS_EXEC,
-				   INSTR_TIME_GET_MILLISEC(duration),
-				   rows,
-				   &bufusage,
-				   &walusage,
-				   NULL,
-				   NULL,
-				   0,
-				   0,
-				   saved_planOrigin);
+		/*
+		 * standard_ProcessUtility allocates desc->instr when instrumentation
+		 * was requested; it stays NULL only if another hook took over and
+		 * skipped it, in which case we have nothing to record.
+		 */
+		if (desc->instr)
+			pgss_store(queryString,
+					   saved_queryId,
+					   saved_stmt_location,
+					   saved_stmt_len,
+					   nesting_level == 0,
+					   PGSS_EXEC,
+					   INSTR_TIME_GET_MILLISEC(desc->instr->total),
+					   rows,
+					   &desc->instr->bufusage,
+					   &desc->instr->walusage,
+					   NULL,
+					   NULL,
+					   0,
+					   0,
+					   saved_planOrigin);
 	}
 	else
 	{
@@ -1274,13 +1257,9 @@ pgss_ProcessUtility(PlannedStmt *pstmt, const char *queryString,
 		PG_TRY();
 		{
 			if (prev_ProcessUtility)
-				prev_ProcessUtility(pstmt, queryString, readOnlyTree,
-									context, params, queryEnv,
-									dest, qc);
+				prev_ProcessUtility(desc);
 			else
-				standard_ProcessUtility(pstmt, queryString, readOnlyTree,
-										context, params, queryEnv,
-										dest, qc);
+				standard_ProcessUtility(desc);
 		}
 		PG_FINALLY();
 		{

contrib/sepgsql/hooks.c

@@ -313,16 +313,9 @@ sepgsql_exec_check_perms(List *rangeTbls, List *rteperminfos, bool abort)
  * break whole of the things if nefarious user would use.
  */
 static void
-sepgsql_utility_command(PlannedStmt *pstmt,
-						const char *queryString,
-						bool readOnlyTree,
-						ProcessUtilityContext context,
-						ParamListInfo params,
-						QueryEnvironment *queryEnv,
-						DestReceiver *dest,
-						QueryCompletion *qc)
+sepgsql_utility_command(UtilityStmtDesc *desc)
 {
-	Node	   *parsetree = pstmt->utilityStmt;
+	Node	   *parsetree = desc->pstmt->utilityStmt;
 	sepgsql_context_info_t saved_context_info = sepgsql_context_info;
 	ListCell   *cell;
 
@@ -381,13 +374,9 @@ sepgsql_utility_command(PlannedStmt *pstmt,
 		}
 
 		if (next_ProcessUtility_hook)
-			(*next_ProcessUtility_hook) (pstmt, queryString, readOnlyTree,
-										 context, params, queryEnv,
-										 dest, qc);
+			(*next_ProcessUtility_hook) (desc);
 		else
-			standard_ProcessUtility(pstmt, queryString, readOnlyTree,
-									context, params, queryEnv,
-									dest, qc);
+			standard_ProcessUtility(desc);
 	}
 	PG_FINALLY();
 	{

src/backend/access/brin/brin.c

@@ -37,6 +37,7 @@
 #include "storage/freespace.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/acl.h"
 #include "utils/datum.h"
 #include "utils/fmgrprotos.h"
@@ -2572,8 +2573,10 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state)
 	 * Next, accumulate WAL usage.  (This must wait for the workers to finish,
 	 * or we might get incomplete data.)
 	 */
-	for (i = 0; i < brinleader->pcxt->nworkers_launched; i++)
-		InstrAccumParallelQuery(&brinleader->bufferusage[i], &brinleader->walusage[i]);
+	if (ActiveUtilityStmtDesc && ActiveUtilityStmtDesc->instr)
+		for (i = 0; i < brinleader->pcxt->nworkers_launched; i++)
+			InstrAccumParallelQuery(ActiveUtilityStmtDesc->instr,
+									&brinleader->bufferusage[i], &brinleader->walusage[i]);
 
 	/* Free last reference to MVCC snapshot, if one was used */
 	if (IsMVCCSnapshot(brinleader->snapshot))

src/backend/access/gin/gininsert.c

@@ -32,6 +32,7 @@
 #include "storage/proc.h"
 #include "storage/predicate.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/datum.h"
 #include "utils/memutils.h"
 #include "utils/builtins.h"
@@ -1124,8 +1125,10 @@ _gin_end_parallel(GinLeader *ginleader, GinBuildState *state)
 	 * Next, accumulate WAL usage.  (This must wait for the workers to finish,
 	 * or we might get incomplete data.)
 	 */
-	for (i = 0; i < ginleader->pcxt->nworkers_launched; i++)
-		InstrAccumParallelQuery(&ginleader->bufferusage[i], &ginleader->walusage[i]);
+	if (ActiveUtilityStmtDesc && ActiveUtilityStmtDesc->instr)
+		for (i = 0; i < ginleader->pcxt->nworkers_launched; i++)
+			InstrAccumParallelQuery(ActiveUtilityStmtDesc->instr,
+									&ginleader->bufferusage[i], &ginleader->walusage[i]);
 
 	/* Free last reference to MVCC snapshot, if one was used */
 	if (IsMVCCSnapshot(ginleader->snapshot))

src/backend/access/nbtree/nbtsort.c

@@ -55,6 +55,7 @@
 #include "storage/condition_variable.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/rel.h"
 #include "utils/sortsupport.h"
 #include "utils/tuplesort.h"
@@ -1619,8 +1620,10 @@ _bt_end_parallel(BTLeader *btleader)
 	 * Next, accumulate WAL usage.  (This must wait for the workers to finish,
 	 * or we might get incomplete data.)
 	 */
-	for (i = 0; i < btleader->pcxt->nworkers_launched; i++)
-		InstrAccumParallelQuery(&btleader->bufferusage[i], &btleader->walusage[i]);
+	if (ActiveUtilityStmtDesc && ActiveUtilityStmtDesc->instr)
+		for (i = 0; i < btleader->pcxt->nworkers_launched; i++)
+			InstrAccumParallelQuery(ActiveUtilityStmtDesc->instr,
+									&btleader->bufferusage[i], &btleader->walusage[i]);
 
 	/* Free last reference to MVCC snapshot, if one was used */
 	if (IsMVCCSnapshot(btleader->snapshot))

src/backend/commands/analyze.c

@@ -357,8 +357,6 @@ do_analyze_rel(Relation onerel, const VacuumParams *params,
 	WalUsage	startwalusage = pgWalUsage;
 	BufferUsage startbufferusage = pgBufferUsage;
 	BufferUsage bufferusage;
-	PgStat_Counter startreadtime = 0;
-	PgStat_Counter startwritetime = 0;
 
 	verbose = (params->options & VACOPT_VERBOSE) != 0;
 	instrument = (verbose || (AmAutoVacuumWorkerProcess() &&
@@ -396,18 +394,10 @@ do_analyze_rel(Relation onerel, const VacuumParams *params,
 
 	/*
 	 * When verbose or autovacuum logging is used, initialize a resource usage
-	 * snapshot and optionally track I/O timing.
+	 * snapshot.
 	 */
 	if (instrument)
-	{
-		if (track_io_timing)
-		{
-			startreadtime = pgStatBlockReadTime;
-			startwritetime = pgStatBlockWriteTime;
-		}
-
 		pg_rusage_init(&ru0);
-	}
 
 	/* Used for instrumentation and stats report */
 	starttime = GetCurrentTimestamp();
@@ -852,8 +842,10 @@ do_analyze_rel(Relation onerel, const VacuumParams *params,
 			}
 			if (track_io_timing)
 			{
-				double		read_ms = (double) (pgStatBlockReadTime - startreadtime) / 1000;
-				double		write_ms = (double) (pgStatBlockWriteTime - startwritetime) / 1000;
+				double		read_ms = INSTR_TIME_GET_MILLISEC(bufferusage.shared_blk_read_time) +
+					INSTR_TIME_GET_MILLISEC(bufferusage.local_blk_read_time);
+				double		write_ms = INSTR_TIME_GET_MILLISEC(bufferusage.shared_blk_write_time) +
+					INSTR_TIME_GET_MILLISEC(bufferusage.local_blk_write_time);
 
 				appendStringInfo(&buf, _("I/O timings: read: %.3f ms, write: %.3f ms\n"),
 								 read_ms, write_ms);

src/backend/commands/extension.c

@@ -1187,19 +1187,18 @@ execute_sql_string(const char *sql, const char *filename)
 			}
 			else
 			{
+				UtilityStmtDesc utilityDesc = {0};
+
 				if (IsA(stmt->utilityStmt, TransactionStmt))
 					ereport(ERROR,
 							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 							 errmsg("transaction control statements are not allowed within an extension script")));
 
-				ProcessUtility(stmt,
-							   sql,
-							   false,
-							   PROCESS_UTILITY_QUERY,
-							   NULL,
-							   NULL,
-							   dest,
-							   NULL);
+				utilityDesc.pstmt = stmt;
+				utilityDesc.queryString = sql;
+				utilityDesc.context = PROCESS_UTILITY_QUERY;
+				utilityDesc.dest = dest;
+				ProcessUtility(&utilityDesc);
 			}
 
 			PopActiveSnapshot();

src/backend/commands/foreigncmds.c

@@ -1656,6 +1656,7 @@ ImportForeignSchema(ImportForeignSchemaStmt *stmt)
 			RawStmt    *rs = lfirst_node(RawStmt, lc2);
 			CreateForeignTableStmt *cstmt = (CreateForeignTableStmt *) rs->stmt;
 			PlannedStmt *pstmt;
+			UtilityStmtDesc utilityDesc = {0};
 
 			/*
 			 * Because we only allow CreateForeignTableStmt, we can skip parse
@@ -1686,9 +1687,11 @@ ImportForeignSchema(ImportForeignSchemaStmt *stmt)
 			pstmt->planOrigin = PLAN_STMT_INTERNAL;
 
 			/* Execute statement */
-			ProcessUtility(pstmt, cmd, false,
-						   PROCESS_UTILITY_SUBCOMMAND, NULL, NULL,
-						   None_Receiver, NULL);
+			utilityDesc.pstmt = pstmt;
+			utilityDesc.queryString = cmd;
+			utilityDesc.context = PROCESS_UTILITY_SUBCOMMAND;
+			utilityDesc.dest = None_Receiver;
+			ProcessUtility(&utilityDesc);
 
 			/* Be sure to advance the command counter between subcommands */
 			CommandCounterIncrement();

src/backend/commands/schemacmds.c

@@ -208,6 +208,7 @@ CreateSchemaCommand(ParseState *pstate, CreateSchemaStmt *stmt,
 	{
 		Node	   *node = (Node *) lfirst(parsetree_item);
 		PlannedStmt *wrapper;
+		UtilityStmtDesc utilityDesc = {0};
 
 		/* need to make a wrapper PlannedStmt */
 		wrapper = makeNode(PlannedStmt);
@@ -219,14 +220,11 @@ CreateSchemaCommand(ParseState *pstate, CreateSchemaStmt *stmt,
 		wrapper->planOrigin = PLAN_STMT_INTERNAL;
 
 		/* do this step */
-		ProcessUtility(wrapper,
-					   pstate->p_sourcetext,
-					   false,
-					   PROCESS_UTILITY_SUBCOMMAND,
-					   NULL,
-					   NULL,
-					   None_Receiver,
-					   NULL);
+		utilityDesc.pstmt = wrapper;
+		utilityDesc.queryString = pstate->p_sourcetext;
+		utilityDesc.context = PROCESS_UTILITY_SUBCOMMAND;
+		utilityDesc.dest = None_Receiver;
+		ProcessUtility(&utilityDesc);
 
 		/* make sure later steps can see the object created here */
 		CommandCounterIncrement();

src/backend/commands/vacuumparallel.c

@@ -47,6 +47,7 @@
 #include "storage/bufmgr.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
+#include "tcop/utility.h"
 #include "utils/injection_point.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
@@ -989,8 +990,10 @@ parallel_vacuum_process_all_indexes(ParallelVacuumState *pvs, int num_index_scan
 		/* Wait for all vacuum workers to finish */
 		WaitForParallelWorkersToFinish(pvs->pcxt);
 
-		for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
-			InstrAccumParallelQuery(&pvs->buffer_usage[i], &pvs->wal_usage[i]);
+		if (ActiveUtilityStmtDesc && ActiveUtilityStmtDesc->instr)
+			for (int i = 0; i < pvs->pcxt->nworkers_launched; i++)
+				InstrAccumParallelQuery(ActiveUtilityStmtDesc->instr,
+										&pvs->buffer_usage[i], &pvs->wal_usage[i]);
 	}
 
 	/*

src/backend/executor/execMain.c

@@ -259,6 +259,9 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags)
 	if (queryDesc->query_instr_options)
 		queryDesc->query_instr = InstrAlloc(queryDesc->query_instr_options);
 
+	/* Let parallel leader code reach the query-level instrumentation */
+	estate->es_query_instr = queryDesc->query_instr;
+
 	/*
 	 * Set up an AFTER-trigger statement context, unless told not to, or
 	 * unless it's EXPLAIN-only mode (when ExecutorFinish won't be called).