Report work_mem wanted by a sort that spilled to disk

src/backend/executor/nodeIncrementalSort.c

@@ -105,12 +105,14 @@
 				Assert(IsParallelWorker()); \
 				Assert(ParallelWorkerNumber <= (node)->shared_info->num_workers); \
 				instrumentSortedGroup(&(node)->shared_info->sinfo[ParallelWorkerNumber].groupName##GroupInfo, \
-									  (node)->groupName##_state); \
+									  (node)->groupName##_state, \
+									  (node)->ss.ps.instrument); \
 			} \
 			else \
 			{ \
 				instrumentSortedGroup(&(node)->incsort_info.groupName##GroupInfo, \
-									  (node)->groupName##_state); \
+									  (node)->groupName##_state, \
+									  (node)->ss.ps.instrument); \
 			} \
 		} \
 	} while (0)
@@ -126,7 +128,8 @@
  */
 static void
 instrumentSortedGroup(IncrementalSortGroupInfo *groupInfo,
-					  Tuplesortstate *sortState)
+					  Tuplesortstate *sortState,
+					  Instrumentation *instr)
 {
 	TuplesortInstrumentation sort_instr;
 
@@ -151,6 +154,26 @@ instrumentSortedGroup(IncrementalSortGroupInfo *groupInfo,
 			break;
 	}
 
+	/*
+	 * GPDB: remember how much memory the batches of this node used, and the
+	 * largest work_mem any single batch would have needed to stay in memory,
+	 * so that EXPLAIN ANALYZE can report "Memory used" and "Memory wanted".
+	 */
+	if ((int64) sort_instr.workmemused > groupInfo->maxWorkmemUsed)
+		groupInfo->maxWorkmemUsed = sort_instr.workmemused;
+	if ((int64) sort_instr.workmemwanted > groupInfo->maxWorkmemWanted)
+		groupInfo->maxWorkmemWanted = sort_instr.workmemwanted;
+
+	if (instr != NULL)
+	{
+		if (instr->workmemused < groupInfo->maxWorkmemUsed)
+			instr->workmemused = groupInfo->maxWorkmemUsed;
+		if (instr->workmemwanted < groupInfo->maxWorkmemWanted)
+			instr->workmemwanted = groupInfo->maxWorkmemWanted;
+		if (sort_instr.spaceType == SORT_SPACE_TYPE_DISK)
+			instr->workfileCreated = true;
+	}
+
 	/* Track each sort method we've used. */
 	groupInfo->sortMethods |= sort_instr.sortMethod;
 }
@@ -313,7 +336,7 @@ switchToPresortedPrefixMode(PlanState *pstate)
 												&(plannode->sort.sortOperators[nPresortedCols]),
 												&(plannode->sort.collations[nPresortedCols]),
 												&(plannode->sort.nullsFirst[nPresortedCols]),
-												work_mem,
+												PlanStateOperatorMemKB((PlanState *) node),
 												NULL,
 												false);
 		node->prefixsort_state = prefixsort_state;
@@ -614,7 +637,7 @@ ExecIncrementalSort(PlanState *pstate)
 												  plannode->sort.sortOperators,
 												  plannode->sort.collations,
 												  plannode->sort.nullsFirst,
-												  work_mem,
+												  PlanStateOperatorMemKB((PlanState *) node),
 												  NULL,
 												  false);
 			node->fullsort_state = fullsort_state;
@@ -1253,4 +1276,24 @@ ExecIncrementalSortRetrieveInstrumentation(IncrementalSortState *node)
 	si = palloc(size);
 	memcpy(si, node->shared_info, size);
 	node->shared_info = si;
+
+	/*
+	 * GPDB: the workers' memory figures are not aggregated by the generic
+	 * instrumentation code, so fold them into this node's instrumentation.
+	 */
+	if (node->ss.ps.instrument != NULL)
+	{
+		for (int n = 0; n < si->num_workers; n++)
+		{
+			int64		used = Max(si->sinfo[n].fullsortGroupInfo.maxWorkmemUsed,
+								   si->sinfo[n].prefixsortGroupInfo.maxWorkmemUsed);
+			int64		wanted = Max(si->sinfo[n].fullsortGroupInfo.maxWorkmemWanted,
+									 si->sinfo[n].prefixsortGroupInfo.maxWorkmemWanted);
+
+			if (node->ss.ps.instrument->workmemused < used)
+				node->ss.ps.instrument->workmemused = used;
+			if (node->ss.ps.instrument->workmemwanted < wanted)
+				node->ss.ps.instrument->workmemwanted = wanted;
+		}
+	}
 }

src/backend/executor/nodeSort.c

@@ -122,13 +122,10 @@ ExecSort(PlanState *pstate)
 		/* CDB */
 
 		/* If EXPLAIN ANALYZE, share our Instrumentation object with sort. */
-		/* GPDB_12_MERGE_FIXME: broken */
-#if 0
 		if (node->ss.ps.instrument && node->ss.ps.instrument->need_cdb)
 			tuplesort_set_instrument(tuplesortstate,
 									 node->ss.ps.instrument,
 									 node->ss.ps.cdbexplainbuf);
-#endif
 		/*
 		 * Scan the subplan and feed all the tuples to tuplesort.
 		 */
@@ -444,8 +441,8 @@ ExecSortExplainEnd(PlanState *planstate, struct StringInfoData *buf)
 
 	if (sortstate->tuplesortstate)
 	{
-		tuplesort_get_stats(sortstate->tuplesortstate,
-							&sortstate->sortstats);
+		tuplesort_finalize_stats(sortstate->tuplesortstate,
+								 &sortstate->sortstats);
 
 		if (planstate->instrument)
 		{
@@ -470,8 +467,8 @@ ExecEagerFreeSort(SortState *node)
 		 * Save stats like in ExecSortExplainEnd, so that we can display
 		 * them later in EXPLAIN ANALYZE.
 		 */
-		tuplesort_get_stats(node->tuplesortstate,
-							&node->sortstats);
+		tuplesort_finalize_stats(node->tuplesortstate,
+								 &node->sortstats);
 		if (node->ss.ps.instrument)
 		{
 			node->ss.ps.instrument->workfileCreated = (node->sortstats.spaceType == SORT_SPACE_TYPE_DISK);
@@ -579,4 +576,23 @@ ExecSortRetrieveInstrumentation(SortState *node)
 	si = palloc(size);
 	memcpy(si, node->shared_info, size);
 	node->shared_info = si;
+
+	/*
+	 * GPDB: the workers sorted their own share of the rows, so let
+	 * "Memory wanted" account for them too, not just for the leader's sort.
+	 */
+	if (node->ss.ps.instrument)
+	{
+		int			n;
+
+		for (n = 0; n < si->num_workers; n++)
+		{
+			if (si->sinstrument[n].sortMethod == SORT_TYPE_STILL_IN_PROGRESS)
+				continue;
+
+			node->ss.ps.instrument->workmemwanted =
+				Max(node->ss.ps.instrument->workmemwanted,
+					si->sinstrument[n].workmemwanted);
+		}
+	}
 }

src/backend/utils/resource_manager/memquota.c

@@ -243,6 +243,7 @@ IsMemoryIntensiveOperator(Node *node, PlannedStmt *stmt)
 	{
 		case T_Material:
 		case T_Sort:
+		case T_IncrementalSort:
 		case T_ShareInputScan:
 		case T_Hash:
 		case T_BitmapIndexScan:

src/backend/utils/sort/tuplesort.c

@@ -113,6 +113,7 @@
 #include "utils/rel.h"
 #include "utils/sortsupport.h"
 #include "utils/tuplesort.h"
+#include "utils/dynahash.h"
 
 #include "utils/faultinjector.h"
 
@@ -314,6 +315,7 @@ struct Tuplesortstate
 	int			memtupcount;	/* number of tuples currently present */
 	int			memtupsize;		/* allocated length of memtuples array */
 	bool		growmemtuples;	/* memtuples' growth still underway? */
+	int64		totalNumTuples; /* count of all input tuples */ /*CDB*/
 
 	/*
 	 * Memory for tuples is sometimes allocated using a simple slab allocator,
@@ -476,6 +478,14 @@ struct Tuplesortstate
 	/* we need typelen in order to know how to copy the Datums. */
 	int			datumTypeLen;
 
+	/*
+	 * CDB: EXPLAIN ANALYZE reporting interface and statistics.
+	 */
+	struct Instrumentation *instrument;
+	struct StringInfoData  *explainbuf;
+	uint64 spilledBytes;
+	int64		workmemwanted;	/* work_mem for an in-memory sort */
+
 	/*
 	 * Resource snapshot for time of sort start.
 	 */
@@ -868,6 +878,9 @@ tuplesort_begin_batch(Tuplesortstate *state)
 	 * see comments in grow_memtuples().
 	 */
 	state->growmemtuples = true;
+	state->totalNumTuples = 0; /*CDB*/
+	state->spilledBytes = 0; /*CDB*/
+	state->workmemwanted = 0; /*CDB*/
 	state->slabAllocatorUsed = false;
 	if (state->memtuples != NULL && state->memtupsize != INITIAL_MEMTUPSIZE)
 	{
@@ -1894,6 +1907,8 @@ puttuple_common(Tuplesortstate *state, SortTuple *tuple)
 {
 	Assert(!LEADER(state));
 
+	state->totalNumTuples++;
+
 	switch (state->status)
 	{
 		case TSS_INITIAL:
@@ -2118,6 +2133,27 @@ tuplesort_performsort(Tuplesortstate *state)
 			 * Note that mergeruns sets the correct state->status.
 			 */
 			dumptuples(state, true);
+
+			/*
+			 * CDB: How much work_mem would be enough for in-memory sort?
+			 *
+			 * The workmemwanted is summed up of the following:
+			 * (1) metadata: Tuplesortstate, tuple array
+			 * (2) the total bytes for all tuples.
+			 *
+			 * It is kept in the state as well, so that tuplesort_get_stats()
+			 * can hand it over from a parallel worker, which has no
+			 * Instrumentation of its own.
+			 */
+			state->workmemwanted =
+				sizeof(Tuplesortstate) +
+				((uint64) 1 << my_log2(state->totalNumTuples)) * sizeof(SortTuple) +
+				state->spilledBytes;
+
+			if (state->instrument && state->instrument->need_cdb)
+				state->instrument->workmemwanted =
+					Max(state->instrument->workmemwanted, state->workmemwanted);
+
 			mergeruns(state);
 			state->eof_reached = false;
 			state->markpos_block = 0L;
@@ -3223,6 +3259,7 @@ dumptuples(Tuplesortstate *state, bool alltuples)
 {
 	int			memtupwrite;
 	int			i;
+	long		prevAvailMem = state->availMem;
 
 	/*
 	 * Nothing to do if we still fit in available memory and have array slots,
@@ -3340,6 +3377,12 @@ dumptuples(Tuplesortstate *state, bool alltuples)
 			 pg_rusage_show(&state->ru_start));
 #endif
 
+	/* CDB: Accumulate total size of spilled tuples. */
+	if (state->availMem > prevAvailMem)
+	{
+		state->spilledBytes += state->availMem - prevAvailMem;
+	}
+
 	if (!alltuples)
 		selectnewtape(state);
 }
@@ -3471,7 +3514,11 @@ tuplesort_get_stats(Tuplesortstate *state,
 	else
 		stats->spaceType = SORT_SPACE_TYPE_MEMORY;
 	stats->spaceUsed = (state->maxSpace + 1023) / 1024;
-	stats->workmemused = MemoryContextGetPeakSpace(state->sortcontext);
+	if (state->instrument)
+		stats->workmemused = state->instrument->workmemused;
+	else
+		stats->workmemused = MemoryContextGetPeakSpace(state->sortcontext);
+	stats->workmemwanted = state->workmemwanted;
 
 	switch (state->maxSpaceStatus)
 	{
@@ -4877,3 +4924,44 @@ free_sort_tuple(Tuplesortstate *state, SortTuple *stup)
 		stup->tuple = NULL;
 	}
 }
+
+/*
+ * tuplesort_set_instrument
+ *
+ * May be called after tuplesort_begin_xxx() to enable reporting of
+ * statistics and events for EXPLAIN ANALYZE.
+ *
+ * The 'instr' and 'explainbuf' ptrs are retained in the 'state' object for
+ * possible use anytime during the sort, up to and including tuplesort_end().
+ * The caller must ensure that the referenced objects remain allocated and
+ * valid for the life of the Tuplesortstate object; or if they are to be
+ * freed early, disconnect them by calling again with NULL pointers.
+ */
+void
+tuplesort_set_instrument(Tuplesortstate            *state,
+						 struct Instrumentation    *instrument,
+						 struct StringInfoData     *explainbuf)
+{
+	state->instrument = instrument;
+	state->explainbuf = explainbuf;
+}
+
+/*
+ * tuplesort_finalize_stats
+ *
+ * Finalize the EXPLAIN ANALYZE stats.
+ */
+void
+tuplesort_finalize_stats(Tuplesortstate *state,
+					TuplesortInstrumentation *stats)
+{
+	if (state->instrument)
+	{
+		double  workmemused;
+
+		workmemused = MemoryContextGetPeakSpace(state->sortcontext);
+		if (state->instrument->workmemused < workmemused)
+			state->instrument->workmemused = workmemused;
+	}
+	tuplesort_get_stats(state, stats);
+}

src/include/nodes/execnodes.h

@@ -2649,6 +2649,8 @@ typedef struct IncrementalSortGroupInfo
 	int64		totalDiskSpaceUsed;
 	int64		maxMemorySpaceUsed;
 	int64		totalMemorySpaceUsed;
+	int64		maxWorkmemUsed;		/* CDB: work_mem actually used */
+	int64		maxWorkmemWanted;	/* CDB: work_mem to avoid scratch i/o */
 	bits32		sortMethods;	/* bitmask of TuplesortMethod */
 } IncrementalSortGroupInfo;
 

src/include/utils/tuplesort.h

@@ -97,6 +97,7 @@ typedef struct TuplesortInstrumentation
 	long		spaceUsed;		/* space consumption, in kB */
 
 	Size		workmemused;
+	Size		workmemwanted;	/* GPDB: work_mem for an in-memory sort */
 } TuplesortInstrumentation;
 
 
@@ -263,6 +264,8 @@ extern void tuplesort_reset(Tuplesortstate *state);
 
 extern void tuplesort_get_stats(Tuplesortstate *state,
 								TuplesortInstrumentation *stats);
+extern void tuplesort_finalize_stats(Tuplesortstate *state,
+								TuplesortInstrumentation *stats);
 extern const char *tuplesort_method_name(TuplesortMethod m);
 extern const char *tuplesort_space_type_name(TuplesortSpaceType t);
 
@@ -273,6 +276,10 @@ extern void tuplesort_initialize_shared(Sharedsort *shared, int nWorkers,
 										dsm_segment *seg);
 extern void tuplesort_attach_shared(Sharedsort *shared, dsm_segment *seg);
 
+extern void tuplesort_set_instrument(Tuplesortstate            *state,
+									 struct Instrumentation    *instrument,
+									 struct StringInfoData     *explainbuf);
+
 /*
  * These routines may only be called if randomAccess was specified 'true'.
  * Likewise, backwards scan in gettuple/getdatum is only allowed if