Parallel query: Refactor instrumentation to reduce repetition, report directly into shmem

src/backend/access/brin/brin.c

@@ -2845,7 +2845,7 @@ _brin_parallel_scan_and_build(BrinBuildState *state,
 
 	scan = table_beginscan_parallel(heap,
 									ParallelTableScanFromBrinShared(brinshared),
-									SO_NONE);
+									NULL, SO_NONE);
 
 	reltuples = table_index_build_scan(heap, index, indexInfo, true, true,
 									   brinbuildCallbackParallel, state, scan);

src/backend/access/gin/gininsert.c

@@ -2069,7 +2069,7 @@ _gin_parallel_scan_and_build(GinBuildState *state,
 
 	scan = table_beginscan_parallel(heap,
 									ParallelTableScanFromGinBuildShared(ginshared),
-									SO_NONE);
+									NULL, SO_NONE);
 
 	reltuples = table_index_build_scan(heap, index, indexInfo, true, progress,
 									   ginBuildCallbackParallel, state, scan);

src/backend/access/heap/heapam.c

@@ -1167,6 +1167,7 @@ TableScanDesc
 heap_beginscan(Relation relation, Snapshot snapshot,
 			   int nkeys, ScanKey key,
 			   ParallelTableScanDesc parallel_scan,
+			   struct TableScanInstrumentation *instrument,
 			   uint32 flags)
 {
 	HeapScanDesc scan;
@@ -1317,9 +1318,15 @@ heap_beginscan(Relation relation, Snapshot snapshot,
 	/* enable read stream instrumentation */
 	if ((flags & SO_SCAN_INSTRUMENT) && (scan->rs_read_stream != NULL))
 	{
-		scan->rs_base.rs_instrument = palloc0_object(TableScanInstrumentation);
-		read_stream_enable_stats(scan->rs_read_stream,
-								 &scan->rs_base.rs_instrument->io);
+		/*
+		 * The caller (the executor) always supplies the location where the
+		 * statistics should be accumulated -- the node's own storage, or a
+		 * parallel worker's slot in shared memory -- so the scan never owns
+		 * the instrumentation and never frees it.
+		 */
+		Assert(instrument != NULL);
+		scan->rs_base.rs_instrument = instrument;
+		read_stream_enable_stats(scan->rs_read_stream, &instrument->io);
 	}
 
 	scan->rs_vmbuffer = InvalidBuffer;
@@ -1425,9 +1432,6 @@ heap_endscan(TableScanDesc sscan)
 	if (scan->rs_base.rs_flags & SO_TEMP_SNAPSHOT)
 		UnregisterSnapshot(scan->rs_base.rs_snapshot);
 
-	if (scan->rs_base.rs_instrument)
-		pfree(scan->rs_base.rs_instrument);
-
 	pfree(scan);
 }
 

src/backend/access/nbtree/nbtsort.c

@@ -1929,7 +1929,7 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2,
 	indexInfo->ii_Concurrent = btshared->isconcurrent;
 	scan = table_beginscan_parallel(btspool->heap,
 									ParallelTableScanFromBTShared(btshared),
-									SO_NONE);
+									NULL, SO_NONE);
 	reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo,
 									   true, progress, _bt_build_callback,
 									   &buildstate, scan);

src/backend/access/table/tableam.c

@@ -118,7 +118,7 @@ table_beginscan_catalog(Relation relation, int nkeys, ScanKeyData *key)
 	Snapshot	snapshot = RegisterSnapshot(GetCatalogSnapshot(relid));
 
 	return table_beginscan_common(relation, snapshot, nkeys, key,
-								  NULL, flags, SO_NONE);
+								  NULL, NULL, flags, SO_NONE);
 }
 
 
@@ -164,6 +164,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,
 
 TableScanDesc
 table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan,
+						 struct TableScanInstrumentation *instrument,
 						 uint32 flags)
 {
 	Snapshot	snapshot;
@@ -186,12 +187,13 @@ table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan,
 	}
 
 	return table_beginscan_common(relation, snapshot, 0, NULL,
-								  pscan, internal_flags, flags);
+								  pscan, instrument, internal_flags, flags);
 }
 
 TableScanDesc
 table_beginscan_parallel_tidrange(Relation relation,
 								  ParallelTableScanDesc pscan,
+								  struct TableScanInstrumentation *instrument,
 								  uint32 flags)
 {
 	Snapshot	snapshot;
@@ -217,7 +219,7 @@ table_beginscan_parallel_tidrange(Relation relation,
 	}
 
 	sscan = table_beginscan_common(relation, snapshot, 0, NULL,
-								   pscan, internal_flags, flags);
+								   pscan, instrument, internal_flags, flags);
 	return sscan;
 }
 

src/backend/commands/explain.c

@@ -3150,7 +3150,8 @@ show_sort_info(SortState *sortstate, ExplainState *es)
 			const char *spaceType;
 			int64		spaceUsed;
 
-			sinstrument = &sortstate->shared_info->sinstrument[n];
+			sinstrument = GetWorkerInstr(sortstate->shared_info,
+										 TuplesortInstrumentation, n);
 			if (sinstrument->sortMethod == SORT_TYPE_STILL_IN_PROGRESS)
 				continue;		/* ignore any unfilled slots */
 			sortMethod = tuplesort_method_name(sinstrument->sortMethod);
@@ -3345,7 +3346,7 @@ show_incremental_sort_info(IncrementalSortState *incrsortstate,
 		for (n = 0; n < incrsortstate->shared_info->num_workers; n++)
 		{
 			IncrementalSortInfo *incsort_info =
-				&incrsortstate->shared_info->sinfo[n];
+				GetWorkerInstr(incrsortstate->shared_info, IncrementalSortInfo, n);
 
 			/*
 			 * If a worker hasn't processed any sort groups at all, then
@@ -3416,12 +3417,13 @@ show_hash_info(HashState *hashstate, ExplainState *es)
 	 */
 	if (hashstate->shared_info)
 	{
-		SharedHashInfo *shared_info = hashstate->shared_info;
+		SharedWorkerInstrumentation *shared_info = hashstate->shared_info;
 		int			i;
 
 		for (i = 0; i < shared_info->num_workers; ++i)
 		{
-			HashInstrumentation *worker_hi = &shared_info->hinstrument[i];
+			HashInstrumentation *worker_hi = GetWorkerInstr(shared_info,
+															HashInstrumentation, i);
 
 			hinstrument.nbuckets = Max(hinstrument.nbuckets,
 									   worker_hi->nbuckets);
@@ -3697,7 +3699,7 @@ show_memoize_info(MemoizeState *mstate, List *ancestors, ExplainState *es)
 	{
 		MemoizeInstrumentation *si;
 
-		si = &mstate->shared_info->sinstrument[n];
+		si = GetWorkerInstr(mstate->shared_info, MemoizeInstrumentation, n);
 
 		/*
 		 * Skip workers that didn't do any work.  We needn't bother checking
@@ -3752,7 +3754,7 @@ static void
 show_hashagg_info(AggState *aggstate, ExplainState *es)
 {
 	Agg		   *agg = (Agg *) aggstate->ss.ps.plan;
-	int64		memPeakKb = BYTES_TO_KILOBYTES(aggstate->hash_mem_peak);
+	int64		memPeakKb = BYTES_TO_KILOBYTES(aggstate->stats.hash_mem_peak);
 
 	if (agg->aggstrategy != AGG_HASHED &&
 		agg->aggstrategy != AGG_MIXED)
@@ -3769,13 +3771,13 @@ show_hashagg_info(AggState *aggstate, ExplainState *es)
 		 * detect this by checking how much memory it used.  If we find it
 		 * didn't do any work then we don't show its properties.
 		 */
-		if (es->analyze && aggstate->hash_mem_peak > 0)
+		if (es->analyze && aggstate->stats.hash_mem_peak > 0)
 		{
 			ExplainPropertyInteger("HashAgg Batches", NULL,
-								   aggstate->hash_batches_used, es);
+								   aggstate->stats.hash_batches_used, es);
 			ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb, es);
 			ExplainPropertyInteger("Disk Usage", "kB",
-								   aggstate->hash_disk_used, es);
+								   aggstate->stats.hash_disk_used, es);
 		}
 	}
 	else
@@ -3795,22 +3797,22 @@ show_hashagg_info(AggState *aggstate, ExplainState *es)
 		 * detect this by checking how much memory it used.  If we find it
 		 * didn't do any work then we don't show its properties.
 		 */
-		if (es->analyze && aggstate->hash_mem_peak > 0)
+		if (es->analyze && aggstate->stats.hash_mem_peak > 0)
 		{
 			if (!gotone)
 				ExplainIndentText(es);
 			else
 				appendStringInfoSpaces(es->str, 2);
 
 			appendStringInfo(es->str, "Batches: %d  Memory Usage: " INT64_FORMAT "kB",
-							 aggstate->hash_batches_used, memPeakKb);
+							 aggstate->stats.hash_batches_used, memPeakKb);
 			gotone = true;
 
 			/* Only display disk usage if we spilled to disk */
-			if (aggstate->hash_batches_used > 1)
+			if (aggstate->stats.hash_batches_used > 1)
 			{
 				appendStringInfo(es->str, "  Disk Usage: " UINT64_FORMAT "kB",
-								 aggstate->hash_disk_used);
+								 aggstate->stats.hash_disk_used);
 			}
 		}
 
@@ -3827,7 +3829,8 @@ show_hashagg_info(AggState *aggstate, ExplainState *es)
 			uint64		hash_disk_used;
 			int			hash_batches_used;
 
-			sinstrument = &aggstate->shared_info->sinstrument[n];
+			sinstrument = GetWorkerInstr(aggstate->shared_info,
+										 AggregateInstrumentation, n);
 			/* Skip workers that didn't do anything */
 			if (sinstrument->hash_mem_peak == 0)
 				continue;
@@ -3874,7 +3877,7 @@ static void
 show_indexsearches_info(PlanState *planstate, ExplainState *es)
 {
 	Plan	   *plan = planstate->plan;
-	SharedIndexScanInstrumentation *SharedInfo = NULL;
+	SharedWorkerInstrumentation *SharedInfo = NULL;
 	uint64		nsearches = 0;
 
 	if (!es->analyze)
@@ -3916,7 +3919,8 @@ show_indexsearches_info(PlanState *planstate, ExplainState *es)
 	{
 		for (int i = 0; i < SharedInfo->num_workers; ++i)
 		{
-			IndexScanInstrumentation *winstrument = &SharedInfo->winstrument[i];
+			IndexScanInstrumentation *winstrument =
+				GetWorkerInstr(SharedInfo, IndexScanInstrumentation, i);
 
 			nsearches += winstrument->nsearches;
 		}
@@ -3960,7 +3964,9 @@ show_tidbitmap_info(BitmapHeapScanState *planstate, ExplainState *es)
 	{
 		for (int n = 0; n < planstate->sinstrument->num_workers; n++)
 		{
-			BitmapHeapScanInstrumentation *si = &planstate->sinstrument->sinstrument[n];
+			BitmapHeapScanInstrumentation *si =
+				GetWorkerInstr(planstate->sinstrument,
+							   BitmapHeapScanInstrumentation, n);
 
 			if (si->exact_pages == 0 && si->lossy_pages == 0)
 				continue;
@@ -4084,14 +4090,16 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es)
 	{
 		case T_BitmapHeapScan:
 			{
-				SharedBitmapHeapInstrumentation *sinstrument
+				SharedWorkerInstrumentation *sinstrument
 				= ((BitmapHeapScanState *) planstate)->sinstrument;
 
 				if (sinstrument)
 				{
 					for (int i = 0; i < sinstrument->num_workers; ++i)
 					{
-						BitmapHeapScanInstrumentation *winstrument = &sinstrument->sinstrument[i];
+						BitmapHeapScanInstrumentation *winstrument =
+							GetWorkerInstr(sinstrument,
+										   BitmapHeapScanInstrumentation, i);
 
 						AccumulateIOStats(&stats, &winstrument->stats.io);
 
@@ -4108,14 +4116,15 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es)
 			}
 		case T_SeqScan:
 			{
-				SharedSeqScanInstrumentation *sinstrument
+				SharedWorkerInstrumentation *sinstrument
 				= ((SeqScanState *) planstate)->sinstrument;
 
 				if (sinstrument)
 				{
 					for (int i = 0; i < sinstrument->num_workers; ++i)
 					{
-						SeqScanInstrumentation *winstrument = &sinstrument->sinstrument[i];
+						SeqScanInstrumentation *winstrument =
+							GetWorkerInstr(sinstrument, SeqScanInstrumentation, i);
 
 						AccumulateIOStats(&stats, &winstrument->stats.io);
 
@@ -4132,14 +4141,16 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es)
 			}
 		case T_TidRangeScan:
 			{
-				SharedTidRangeScanInstrumentation *sinstrument
+				SharedWorkerInstrumentation *sinstrument
 				= ((TidRangeScanState *) planstate)->trss_sinstrument;
 
 				if (sinstrument)
 				{
 					for (int i = 0; i < sinstrument->num_workers; ++i)
 					{
-						TidRangeScanInstrumentation *winstrument = &sinstrument->sinstrument[i];
+						TidRangeScanInstrumentation *winstrument =
+							GetWorkerInstr(sinstrument,
+										   TidRangeScanInstrumentation, i);
 
 						AccumulateIOStats(&stats, &winstrument->stats.io);