From a7a8eeeb2ccebd765b704ff2e86f7769cd359531 Mon Sep 17 00:00:00 2001
From: James Hunter <james.hunter.pg@gmail.com>
Date: Fri, 21 Feb 2025 00:16:22 +0000
Subject: [PATCH 3/4] EXPLAIN (WORK_MEM ON) now shows working memory limit

This commit moves the working-memory limit that an executor node checks, at
runtime, from the "work_mem" and "hash_mem_multiplier" GUCs, to a new
field, "work_mem", added to the Plan node. To preserve backward
compatibility, it also copies the "work_mem", etc., values from these GUCs
to the new field.

This field is on the Plan node, instead of the PlanState, because it needs
to be set before we can call ExecInitNode(). Many PlanStates look at their
working-memory limit when creating their data structures, during
initialization. So the field is on the Plan node, but set between planning
and execution phases.

Also modifies "EXPLAIN (WORK_MEM ON)" so that it also displays this
working-memory limit.
---
 src/backend/commands/explain.c             |  59 ++++-
 src/backend/executor/Makefile              |   1 +
 src/backend/executor/execGrouping.c        |  10 +-
 src/backend/executor/execMain.c            |   6 +
 src/backend/executor/execSRF.c             |   5 +-
 src/backend/executor/execWorkmem.c         | 281 +++++++++++++++++++++
 src/backend/executor/meson.build           |   1 +
 src/backend/executor/nodeAgg.c             |  69 +++--
 src/backend/executor/nodeBitmapIndexscan.c |   3 +-
 src/backend/executor/nodeBitmapOr.c        |   3 +-
 src/backend/executor/nodeCtescan.c         |   3 +-
 src/backend/executor/nodeFunctionscan.c    |   2 +
 src/backend/executor/nodeHash.c            |  23 +-
 src/backend/executor/nodeIncrementalSort.c |   4 +-
 src/backend/executor/nodeMaterial.c        |   3 +-
 src/backend/executor/nodeMemoize.c         |   2 +-
 src/backend/executor/nodeRecursiveunion.c  |  12 +-
 src/backend/executor/nodeSetOp.c           |   1 +
 src/backend/executor/nodeSort.c            |   4 +-
 src/backend/executor/nodeSubplan.c         |   2 +
 src/backend/executor/nodeTableFuncscan.c   |   3 +-
 src/backend/executor/nodeWindowAgg.c       |   3 +-
 src/backend/optimizer/path/costsize.c      |  15 +-
 src/include/commands/explain.h             |   1 +
 src/include/executor/executor.h            |   7 +
 src/include/executor/hashjoin.h            |   3 +-
 src/include/executor/nodeAgg.h             |   5 +-
 src/include/executor/nodeHash.h            |   3 +-
 src/include/nodes/plannodes.h              |   8 +-
 src/include/nodes/primnodes.h              |   2 +
 src/test/regress/expected/workmem.out      | 184 ++++++++------
 31 files changed, 577 insertions(+), 151 deletions(-)
 create mode 100644 src/backend/executor/execWorkmem.c

diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index e09d7f868c9..07c6d34764b 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -180,8 +180,8 @@ static void ExplainJSONLineEnding(ExplainState *es);
 static void ExplainYAMLLineStarting(ExplainState *es);
 static void escape_yaml(StringInfo buf, const char *str);
 static SerializeMetrics GetSerializationMetrics(DestReceiver *dest);
-static void compute_subplan_workmem(List *plans, double *workmem);
-static void compute_agg_workmem(Agg *agg, double *workmem);
+static void compute_subplan_workmem(List *plans, double *workmem, double *limit);
+static void compute_agg_workmem(Agg *agg, double *workmem, double *limit);
 
 
 
@@ -843,6 +843,8 @@ ExplainOnePlan(PlannedStmt *plannedstmt, CachedPlan *cplan,
 	{
 		ExplainPropertyFloat("Total Working Memory", "kB",
 							 es->total_workmem, 0, es);
+		ExplainPropertyFloat("Total Working Memory Limit", "kB",
+							 es->total_workmem_limit, 0, es);
 	}
 
 	ExplainCloseGroup("Query", NULL, true, es);
@@ -1983,19 +1985,20 @@ ExplainNode(PlanState *planstate, List *ancestors,
 	if (es->work_mem)
 	{
 		double		plan_workmem = 0.0;
+		double		plan_limit = 0.0;
 
 		/*
 		 * Include working memory used by this Plan's SubPlan objects, whether
 		 * they are included on the Plan's initPlan or subPlan lists.
 		 */
-		compute_subplan_workmem(planstate->initPlan, &plan_workmem);
-		compute_subplan_workmem(planstate->subPlan, &plan_workmem);
+		compute_subplan_workmem(planstate->initPlan, &plan_workmem, &plan_limit);
+		compute_subplan_workmem(planstate->subPlan, &plan_workmem, &plan_limit);
 
 		/* Include working memory used by this Plan, itself. */
 		switch (nodeTag(plan))
 		{
 			case T_Agg:
-				compute_agg_workmem((Agg *) plan, &plan_workmem);
+				compute_agg_workmem((Agg *) plan, &plan_workmem, &plan_limit);
 				break;
 			case T_FunctionScan:
 				{
@@ -2003,6 +2006,8 @@ ExplainNode(PlanState *planstate, List *ancestors,
 
 					plan_workmem += (double) plan->workmem *
 						list_length(fscan->functions);
+					plan_limit += (double) plan->workmem_limit *
+						list_length(fscan->functions);
 					break;
 				}
 			case T_IncrementalSort:
@@ -2011,7 +2016,8 @@ ExplainNode(PlanState *planstate, List *ancestors,
 				 * IncrementalSort creates two Tuplestores, each of
 				 * (estimated) size workmem.
 				 */
-				plan_workmem = (double) plan->workmem * 2;
+				plan_workmem += (double) plan->workmem * 2;
+				plan_limit += (double) plan->workmem_limit * 2;
 				break;
 			case T_RecursiveUnion:
 				{
@@ -2024,11 +2030,15 @@ ExplainNode(PlanState *planstate, List *ancestors,
 					 */
 					plan_workmem += (double) plan->workmem * 2 +
 						runion->hashWorkMem;
+					plan_limit += (double) plan->workmem_limit * 2 +
+						runion->hashWorkMemLimit;
 					break;
 				}
 			default:
 				if (plan->workmem > 0)
 					plan_workmem += plan->workmem;
+				if (plan->workmem_limit > 0)
+					plan_limit += plan->workmem_limit;
 				break;
 		}
 
@@ -2037,17 +2047,23 @@ ExplainNode(PlanState *planstate, List *ancestors,
 		 * working memory.
 		 */
 		plan_workmem *= (1 + es->num_workers);
+		plan_limit *= (1 + es->num_workers);
 
 		es->total_workmem += plan_workmem;
+		es->total_workmem_limit += plan_limit;
 
-		if (plan_workmem > 0.0)
+		if (plan_workmem > 0.0 || plan_limit > 0.0)
 		{
 			if (es->format == EXPLAIN_FORMAT_TEXT)
-				appendStringInfo(es->str, "  (work_mem=%.0f kB)",
-								 plan_workmem);
+				appendStringInfo(es->str, "  (work_mem=%.0f kB limit=%.0f kB)",
+								 plan_workmem, plan_limit);
 			else
+			{
 				ExplainPropertyFloat("Working Memory", "kB",
 									 plan_workmem, 0, es);
+				ExplainPropertyFloat("Working Memory Limit", "kB",
+									 plan_limit, 0, es);
+			}
 		}
 	}
 
@@ -6062,29 +6078,39 @@ GetSerializationMetrics(DestReceiver *dest)
  * increments work_mem counters to include the SubPlan's working-memory.
  */
 static void
-compute_subplan_workmem(List *plans, double *workmem)
+compute_subplan_workmem(List *plans, double *workmem, double *limit)
 {
 	foreach_node(SubPlanState, sps, plans)
 	{
 		SubPlan    *sp = sps->subplan;
 
 		if (sp->hashtab_workmem > 0)
+		{
 			*workmem += sp->hashtab_workmem;
+			*limit += sp->hashtab_workmem_limit;
+		}
 
 		if (sp->hashnul_workmem > 0)
+		{
 			*workmem += sp->hashnul_workmem;
+			*limit += sp->hashnul_workmem_limit;
+		}
 	}
 }
 
-/* Compute an Agg's working memory estimate. */
+/* Compute an Agg's working memory estimate and limit. */
 typedef struct AggWorkMem
 {
 	double		input_sort_workmem;
+	double		input_sort_limit;
 
 	double		output_hash_workmem;
+	double		output_hash_limit;
 
 	int			num_sort_nodes;
+
 	double		max_output_sort_workmem;
+	double		output_sort_limit;
 }			AggWorkMem;
 
 static void
@@ -6092,6 +6118,7 @@ compute_agg_workmem_node(Agg *agg, AggWorkMem * mem)
 {
 	/* Record memory used for input sort buffers. */
 	mem->input_sort_workmem += (double) agg->numSorts * agg->sortWorkMem;
+	mem->input_sort_limit += (double) agg->numSorts * agg->sortWorkMemLimit;
 
 	/* Record memory used for output data structures. */
 	switch (agg->aggstrategy)
@@ -6102,6 +6129,9 @@ compute_agg_workmem_node(Agg *agg, AggWorkMem * mem)
 			mem->max_output_sort_workmem =
 				Max(mem->max_output_sort_workmem, agg->plan.workmem);
 
+			if (mem->output_sort_limit == 0)
+				mem->output_sort_limit = agg->plan.workmem_limit;
+
 			++mem->num_sort_nodes;
 			break;
 		case AGG_HASHED:
@@ -6112,6 +6142,7 @@ compute_agg_workmem_node(Agg *agg, AggWorkMem * mem)
 			 * lifetime of the Agg.
 			 */
 			mem->output_hash_workmem += agg->plan.workmem;
+			mem->output_hash_limit += agg->plan.workmem_limit;
 			break;
 		default:
 
@@ -6135,7 +6166,7 @@ compute_agg_workmem_node(Agg *agg, AggWorkMem * mem)
  * value on the main Agg node.
  */
 static void
-compute_agg_workmem(Agg *agg, double *workmem)
+compute_agg_workmem(Agg *agg, double *workmem, double *limit)
 {
 	AggWorkMem	mem;
 	ListCell   *lc;
@@ -6153,9 +6184,13 @@ compute_agg_workmem(Agg *agg, double *workmem)
 	}
 
 	*workmem = mem.input_sort_workmem + mem.output_hash_workmem;
+	*limit = mem.input_sort_limit + mem.output_hash_limit;;
 
 	/* We'll have at most two sort buffers alive, at any time. */
 	*workmem += mem.num_sort_nodes > 2 ?
 		mem.max_output_sort_workmem * 2.0 :
 		mem.max_output_sort_workmem;
+	*limit += mem.num_sort_nodes > 2 ?
+		mem.output_sort_limit * 2.0 :
+		mem.output_sort_limit;
 }
diff --git a/src/backend/executor/Makefile b/src/backend/executor/Makefile
index 11118d0ce02..8aa9580558f 100644
--- a/src/backend/executor/Makefile
+++ b/src/backend/executor/Makefile
@@ -30,6 +30,7 @@ OBJS = \
 	execScan.o \
 	execTuples.o \
 	execUtils.o \
+	execWorkmem.o \
 	functions.o \
 	instrument.o \
 	nodeAgg.o \
diff --git a/src/backend/executor/execGrouping.c b/src/backend/executor/execGrouping.c
index 33b124fbb0a..bcd1822da80 100644
--- a/src/backend/executor/execGrouping.c
+++ b/src/backend/executor/execGrouping.c
@@ -168,6 +168,7 @@ BuildTupleHashTable(PlanState *parent,
 					Oid *collations,
 					long nbuckets,
 					Size additionalsize,
+					Size hash_mem_limit,
 					MemoryContext metacxt,
 					MemoryContext tablecxt,
 					MemoryContext tempcxt,
@@ -175,15 +176,18 @@ BuildTupleHashTable(PlanState *parent,
 {
 	TupleHashTable hashtable;
 	Size		entrysize = sizeof(TupleHashEntryData) + additionalsize;
-	Size		hash_mem_limit;
 	MemoryContext oldcontext;
 	bool		allow_jit;
 	uint32		hash_iv = 0;
 
 	Assert(nbuckets > 0);
 
-	/* Limit initial table size request to not more than hash_mem */
-	hash_mem_limit = get_hash_memory_limit() / entrysize;
+	/*
+	 * Limit initial table size request to not more than hash_mem
+	 *
+	 * XXX - we should also limit the *maximum* table size to hash_mem.
+	 */
+	hash_mem_limit = hash_mem_limit / entrysize;
 	if (nbuckets > hash_mem_limit)
 		nbuckets = hash_mem_limit;
 
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 0493b7d5365..78fd887a84d 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1050,6 +1050,12 @@ InitPlan(QueryDesc *queryDesc, int eflags)
 	/* signal that this EState is not used for EPQ */
 	estate->es_epq_active = NULL;
 
+	/*
+	 * Assign working memory to SubPlan and Plan nodes, before initializing
+	 * their states.
+	 */
+	ExecAssignWorkMem(plannedstmt);
+
 	/*
 	 * Initialize private state information for each SubPlan.  We must do this
 	 * before running ExecInitNode on the main query tree, since
diff --git a/src/backend/executor/execSRF.c b/src/backend/executor/execSRF.c
index a03fe780a02..4b1e7e0ad1e 100644
--- a/src/backend/executor/execSRF.c
+++ b/src/backend/executor/execSRF.c
@@ -102,6 +102,7 @@ ExecMakeTableFunctionResult(SetExprState *setexpr,
 							ExprContext *econtext,
 							MemoryContext argContext,
 							TupleDesc expectedDesc,
+							int workMem,
 							bool randomAccess)
 {
 	Tuplestorestate *tupstore = NULL;
@@ -261,7 +262,7 @@ ExecMakeTableFunctionResult(SetExprState *setexpr,
 				MemoryContext oldcontext =
 					MemoryContextSwitchTo(econtext->ecxt_per_query_memory);
 
-				tupstore = tuplestore_begin_heap(randomAccess, false, work_mem);
+				tupstore = tuplestore_begin_heap(randomAccess, false, workMem);
 				rsinfo.setResult = tupstore;
 				if (!returnsTuple)
 				{
@@ -396,7 +397,7 @@ no_function_result:
 		MemoryContext oldcontext =
 			MemoryContextSwitchTo(econtext->ecxt_per_query_memory);
 
-		tupstore = tuplestore_begin_heap(randomAccess, false, work_mem);
+		tupstore = tuplestore_begin_heap(randomAccess, false, workMem);
 		rsinfo.setResult = tupstore;
 		MemoryContextSwitchTo(oldcontext);
 
diff --git a/src/backend/executor/execWorkmem.c b/src/backend/executor/execWorkmem.c
new file mode 100644
index 00000000000..c513b90fc77
--- /dev/null
+++ b/src/backend/executor/execWorkmem.c
@@ -0,0 +1,281 @@
+/*-------------------------------------------------------------------------
+ *
+ * execWorkmem.c
+ *	 routine to set the "workmem_limit" field(s) on Plan nodes that need
+ *   workimg memory.
+ *
+ * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/executor/execWorkmem.c
+ *
+ * INTERFACE ROUTINES
+ *		ExecAssignWorkMem	- assign working memory to Plan nodes
+ *
+ *	 NOTES
+ *		Historically, every PlanState node, during initialization, looked at
+ *		the "work_mem" (plus maybe "hash_mem_multiplier") GUC, to determine
+ *		what working-memory limit was imposed on it.
+ *
+ *		Now, to allow different PlanState nodes to be restricted to different
+ *		amounts of memory, each PlanState node reads this limit off its
+ *		corresponding Plan node's "workmem_limit" field. And we populate that
+ *		field by calling ExecAssignWorkMem(), from InitPlan(), before we
+ *		initialize the PlanState nodes.
+ *
+ * 		The "workmem_limit" field is a limit "per data structure," rather than
+ *		"per PlanState". This is needed because some SQL operators (e.g.,
+ *		RecursiveUnion and Agg) require multiple data structures, and sometimes
+ *		the data structures don't all share the same memory requirement. So we
+ *		cannot always just divide a "per PlanState" limit among individual data
+ *		structures. Instead, we maintain the limits on the data structures (and
+ *		EXPLAIN, for example, sums them up into a single, human-readable
+ *		number).
+ *
+ *		Note that the *Path's* "workmem" estimate is per SQL operator, but when
+ *		we convert that Path to a Plan we also break its "workmem" estimate
+ *		down into per-data structure estimates. Some operators therefore
+ *		require additional "limit" fields, which we add to the corresponding
+ *		Plan.
+ *
+ *		We store the "workmem_limit" field(s) on the Plan, instead of the
+ *		PlanState, even though it conceptually belongs to execution rather than
+ *		to planning, because we need it to be set before initializing the
+ *		corresponding PlanState. This is a chicken-and-egg problem. We could,
+ *		of course, make ExecInitNode() a two-phase operation, but that seems
+ *		like overkill. Instead, we store these "limit" fields on the Plan, but
+ *		set them when we start execution, as part of InitPlan().
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/parallel.h"
+#include "executor/executor.h"
+#include "miscadmin.h"
+#include "optimizer/cost.h"
+
+
+/* decls for local routines only used within this module */
+static void assign_workmem_subplan(SubPlan *subplan);
+static void assign_workmem_plan(Plan *plan);
+static void assign_workmem_agg(Agg *agg);
+static void assign_workmem_agg_node(Agg *agg, bool is_first, bool is_last,
+									bool *is_first_sort);
+
+/* end of local decls */
+
+
+/* ------------------------------------------------------------------------
+ *		ExecAssignWorkMem
+ *
+ *		Recursively assigns working memory to any Plans or SubPlans that need
+ *		it.
+ *
+ *		Inputs:
+ *		  'plannedstmt' is the statement to which we assign working memory
+ *
+ * ------------------------------------------------------------------------
+ */
+void
+ExecAssignWorkMem(PlannedStmt *plannedstmt)
+{
+	/*
+	 * No need to re-assign working memory on parallel workers, since workers
+	 * have the same work_mem and hash_mem_multiplier GUCs as the leader.
+	 *
+	 * We already assigned working-memory limits on the leader, and those
+	 * limits were sent to the workers inside the serialized Plan.
+	 */
+	if (IsParallelWorker())
+		return;
+
+	/* Assign working memory to the Plans referred to by SubPlan objects. */
+	foreach_ptr(Plan, plan, plannedstmt->subplans)
+	{
+		if (plan)
+			assign_workmem_plan(plan);
+	}
+
+	/* And assign working memory to the main Plan tree. */
+	assign_workmem_plan(plannedstmt->planTree);
+}
+
+static void
+assign_workmem_subplan(SubPlan *subplan)
+{
+	subplan->hashtab_workmem_limit = subplan->useHashTable ?
+		normalize_workmem(get_hash_memory_limit()) : 0;
+
+	subplan->hashnul_workmem_limit =
+		subplan->useHashTable && !subplan->unknownEqFalse ?
+		normalize_workmem(get_hash_memory_limit()) : 0;
+}
+
+static void
+assign_workmem_plan(Plan *plan)
+{
+	/* Make sure there's enough stack available. */
+	check_stack_depth();
+
+	/* Assign working memory to this node's (hashed) SubPlans. */
+	foreach_node(SubPlan, subplan, plan->initPlan)
+		assign_workmem_subplan(subplan);
+
+	foreach_node(SubPlan, subplan, plan->subPlan)
+		assign_workmem_subplan(subplan);
+
+	/* Assign working memory to this node. */
+	switch (nodeTag(plan))
+	{
+		case T_BitmapIndexScan:
+		case T_CteScan:
+		case T_FunctionScan:
+		case T_IncrementalSort:
+		case T_Material:
+		case T_Sort:
+		case T_TableFuncScan:
+		case T_WindowAgg:
+			if (plan->workmem > 0)
+				plan->workmem_limit = work_mem;
+			break;
+		case T_Hash:
+		case T_Memoize:
+		case T_SetOp:
+			if (plan->workmem > 0)
+				plan->workmem_limit =
+					normalize_workmem(get_hash_memory_limit());
+			break;
+		case T_Agg:
+			assign_workmem_agg((Agg *) plan);
+			break;
+		case T_RecursiveUnion:
+			{
+				RecursiveUnion *runion = (RecursiveUnion *) plan;
+
+				plan->workmem_limit = work_mem;
+
+				if (runion->numCols > 0)
+				{
+					/* Also include memory for hash table. */
+					runion->hashWorkMemLimit =
+						normalize_workmem(get_hash_memory_limit());
+				}
+
+				break;
+			}
+		default:
+			Assert(plan->workmem == 0);
+			plan->workmem_limit = 0;
+			break;
+	}
+
+	/*
+	 * Assign working memory to this node's children. (Logic copied from
+	 * ExplainNode().)
+	 */
+	if (outerPlan(plan))
+		assign_workmem_plan(outerPlan(plan));
+
+	if (innerPlan(plan))
+		assign_workmem_plan(innerPlan(plan));
+
+	switch (nodeTag(plan))
+	{
+		case T_Append:
+			foreach_ptr(Plan, child, ((Append *) plan)->appendplans)
+				assign_workmem_plan(child);
+			break;
+		case T_MergeAppend:
+			foreach_ptr(Plan, child, ((MergeAppend *) plan)->mergeplans)
+				assign_workmem_plan(child);
+			break;
+		case T_BitmapAnd:
+			foreach_ptr(Plan, child, ((BitmapAnd *) plan)->bitmapplans)
+				assign_workmem_plan(child);
+			break;
+		case T_BitmapOr:
+			foreach_ptr(Plan, child, ((BitmapOr *) plan)->bitmapplans)
+				assign_workmem_plan(child);
+			break;
+		case T_SubqueryScan:
+			assign_workmem_plan(((SubqueryScan *) plan)->subplan);
+			break;
+		case T_CustomScan:
+			foreach_ptr(Plan, child, ((CustomScan *) plan)->custom_plans)
+				assign_workmem_plan(child);
+			break;
+		default:
+			break;
+	}
+}
+
+static void
+assign_workmem_agg(Agg *agg)
+{
+	bool		is_first_sort = true;
+
+	/* Assign working memory to the main Agg node. */
+	assign_workmem_agg_node(agg,
+							true /* is_first */ ,
+							agg->chain == NULL /* is_last */ ,
+							&is_first_sort);
+
+	/* Assign working memory to any other grouping sets. */
+	foreach_node(Agg, aggnode, agg->chain)
+	{
+		assign_workmem_agg_node(aggnode,
+								false /* is_first */ ,
+								foreach_current_index(aggnode) ==
+								list_length(agg->chain) - 1 /* is_last */ ,
+								&is_first_sort);
+	}
+}
+
+static void
+assign_workmem_agg_node(Agg *agg, bool is_first, bool is_last,
+						bool *is_first_sort)
+{
+	switch (agg->aggstrategy)
+	{
+		case AGG_HASHED:
+		case AGG_MIXED:
+
+			/*
+			 * Because nodeAgg.c will combine all AGG_HASHED nodes into a
+			 * single phase, it's easier to store the hash working-memory
+			 * limit on the first AGG_{HASHED,MIXED} node, and set it to zero
+			 * for all subsequent AGG_HASHED nodes.
+			 */
+			agg->plan.workmem_limit = is_first ?
+				normalize_workmem(get_hash_memory_limit()) : 0;
+			break;
+		case AGG_SORTED:
+
+			/*
+			 * Also store the sort-output working-memory limit on the first
+			 * AGG_SORTED node, and set it to zero for all subsequent
+			 * AGG_SORTED nodes.
+			 *
+			 * We'll need working-memory to hold the "sort_out" only if this
+			 * isn't the last Agg node (in which case there's no one to sort
+			 * our output).
+			 */
+			agg->plan.workmem_limit = *is_first_sort && !is_last ?
+				work_mem : 0;
+
+			*is_first_sort = false;
+			break;
+		default:
+			break;
+	}
+
+	/* Also include memory needed to sort the input: */
+	if (agg->numSorts > 0)
+	{
+		Assert(agg->sortWorkMem > 0);
+
+		agg->sortWorkMemLimit = work_mem;
+	}
+}
diff --git a/src/backend/executor/meson.build b/src/backend/executor/meson.build
index 2cea41f8771..4e65974f5f3 100644
--- a/src/backend/executor/meson.build
+++ b/src/backend/executor/meson.build
@@ -18,6 +18,7 @@ backend_sources += files(
   'execScan.c',
   'execTuples.c',
   'execUtils.c',
+  'execWorkmem.c',
   'functions.c',
   'instrument.c',
   'nodeAgg.c',
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index ceb8c8a8039..9e5bcf7ada4 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -258,6 +258,7 @@
 #include "executor/execExpr.h"
 #include "executor/executor.h"
 #include "executor/nodeAgg.h"
+#include "executor/nodeHash.h"
 #include "lib/hyperloglog.h"
 #include "miscadmin.h"
 #include "nodes/nodeFuncs.h"
@@ -403,7 +404,8 @@ static void find_cols(AggState *aggstate, Bitmapset **aggregated,
 					  Bitmapset **unaggregated);
 static bool find_cols_walker(Node *node, FindColsContext *context);
 static void build_hash_tables(AggState *aggstate);
-static void build_hash_table(AggState *aggstate, int setno, long nbuckets);
+static void build_hash_table(AggState *aggstate, int setno, long nbuckets,
+							 Size hash_mem_limit);
 static void hashagg_recompile_expressions(AggState *aggstate, bool minslot,
 										  bool nullcheck);
 static long hash_choose_num_buckets(double hashentrysize,
@@ -411,6 +413,7 @@ static long hash_choose_num_buckets(double hashentrysize,
 static int	hash_choose_num_partitions(double input_groups,
 									   double hashentrysize,
 									   int used_bits,
+									   Size hash_mem_limit,
 									   int *log2_npartitions);
 static void initialize_hash_entry(AggState *aggstate,
 								  TupleHashTable hashtable,
@@ -431,9 +434,10 @@ static HashAggBatch *hashagg_batch_new(LogicalTape *input_tape, int setno,
 									   int64 input_tuples, double input_card,
 									   int used_bits);
 static MinimalTuple hashagg_batch_read(HashAggBatch *batch, uint32 *hashp);
-static void hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset,
-							   int used_bits, double input_groups,
-							   double hashentrysize);
+static void hashagg_spill_init(HashAggSpill *spill,
+							   LogicalTapeSet *tapeset, int used_bits,
+							   double input_groups, double hashentrysize,
+							   Size hash_mem_limit);
 static Size hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill,
 								TupleTableSlot *inputslot, uint32 hash);
 static void hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill,
@@ -521,6 +525,14 @@ initialize_phase(AggState *aggstate, int newphase)
 		Sort	   *sortnode = aggstate->phases[newphase + 1].sortnode;
 		PlanState  *outerNode = outerPlanState(aggstate);
 		TupleDesc	tupDesc = ExecGetResultType(outerNode);
+		int			workmem_limit;
+
+		/*
+		 * Read the sort-output workmem_limit off the first AGG_SORTED node.
+		 * Since phase 0 is always AGG_HASHED, this will always be phase 1.
+		 */
+		workmem_limit = aggstate->phases[1].aggnode->plan.workmem_limit;
+		Assert(workmem_limit > 0);
 
 		aggstate->sort_out = tuplesort_begin_heap(tupDesc,
 												  sortnode->numCols,
@@ -528,7 +540,7 @@ initialize_phase(AggState *aggstate, int newphase)
 												  sortnode->sortOperators,
 												  sortnode->collations,
 												  sortnode->nullsFirst,
-												  work_mem,
+												  workmem_limit,
 												  NULL, TUPLESORT_NONE);
 	}
 
@@ -577,7 +589,7 @@ fetch_input_tuple(AggState *aggstate)
  */
 static void
 initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
-					 AggStatePerGroup pergroupstate)
+					 AggStatePerGroup pergroupstate, size_t workMem)
 {
 	/*
 	 * Start a fresh sort operation for each DISTINCT/ORDER BY aggregate.
@@ -591,6 +603,7 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
 		if (pertrans->sortstates[aggstate->current_set])
 			tuplesort_end(pertrans->sortstates[aggstate->current_set]);
 
+		Assert(workMem > 0);
 
 		/*
 		 * We use a plain Datum sorter when there's a single input column;
@@ -606,7 +619,7 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
 									  pertrans->sortOperators[0],
 									  pertrans->sortCollations[0],
 									  pertrans->sortNullsFirst[0],
-									  work_mem, NULL, TUPLESORT_NONE);
+									  workMem, NULL, TUPLESORT_NONE);
 		}
 		else
 			pertrans->sortstates[aggstate->current_set] =
@@ -616,7 +629,7 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
 									 pertrans->sortOperators,
 									 pertrans->sortCollations,
 									 pertrans->sortNullsFirst,
-									 work_mem, NULL, TUPLESORT_NONE);
+									 workMem, NULL, TUPLESORT_NONE);
 	}
 
 	/*
@@ -687,7 +700,8 @@ initialize_aggregates(AggState *aggstate,
 			AggStatePerTrans pertrans = &transstates[transno];
 			AggStatePerGroup pergroupstate = &pergroup[transno];
 
-			initialize_aggregate(aggstate, pertrans, pergroupstate);
+			initialize_aggregate(aggstate, pertrans, pergroupstate,
+								 aggstate->phase->aggnode->sortWorkMemLimit);
 		}
 	}
 }
@@ -1498,7 +1512,7 @@ build_hash_tables(AggState *aggstate)
 		}
 #endif
 
-		build_hash_table(aggstate, setno, nbuckets);
+		build_hash_table(aggstate, setno, nbuckets, memory);
 	}
 
 	aggstate->hash_ngroups_current = 0;
@@ -1508,7 +1522,8 @@ build_hash_tables(AggState *aggstate)
  * Build a single hashtable for this grouping set.
  */
 static void
-build_hash_table(AggState *aggstate, int setno, long nbuckets)
+build_hash_table(AggState *aggstate, int setno, long nbuckets,
+				 Size hash_mem_limit)
 {
 	AggStatePerHash perhash = &aggstate->perhash[setno];
 	MemoryContext metacxt = aggstate->hash_metacxt;
@@ -1537,6 +1552,7 @@ build_hash_table(AggState *aggstate, int setno, long nbuckets)
 											 perhash->aggnode->grpCollations,
 											 nbuckets,
 											 additionalsize,
+											 hash_mem_limit,
 											 metacxt,
 											 hashcxt,
 											 tmpcxt,
@@ -1805,12 +1821,11 @@ hashagg_recompile_expressions(AggState *aggstate, bool minslot, bool nullcheck)
  */
 void
 hash_agg_set_limits(double hashentrysize, double input_groups, int used_bits,
-					Size *mem_limit, uint64 *ngroups_limit,
+					Size hash_mem_limit, Size *mem_limit, uint64 *ngroups_limit,
 					int *num_partitions)
 {
 	int			npartitions;
 	Size		partition_mem;
-	Size		hash_mem_limit = get_hash_memory_limit();
 
 	/* if not expected to spill, use all of hash_mem */
 	if (input_groups * hashentrysize <= hash_mem_limit)
@@ -1830,6 +1845,7 @@ hash_agg_set_limits(double hashentrysize, double input_groups, int used_bits,
 	npartitions = hash_choose_num_partitions(input_groups,
 											 hashentrysize,
 											 used_bits,
+											 hash_mem_limit,
 											 NULL);
 	if (num_partitions != NULL)
 		*num_partitions = npartitions;
@@ -1927,7 +1943,8 @@ hash_agg_enter_spill_mode(AggState *aggstate)
 
 			hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
 							   perhash->aggnode->numGroups,
-							   aggstate->hashentrysize);
+							   aggstate->hashentrysize,
+							   (Size) aggstate->ss.ps.plan->workmem_limit * 1024);
 		}
 	}
 }
@@ -2014,9 +2031,9 @@ hash_choose_num_buckets(double hashentrysize, long ngroups, Size memory)
  */
 static int
 hash_choose_num_partitions(double input_groups, double hashentrysize,
-						   int used_bits, int *log2_npartitions)
+						   int used_bits, Size hash_mem_limit,
+						   int *log2_npartitions)
 {
-	Size		hash_mem_limit = get_hash_memory_limit();
 	double		partition_limit;
 	double		mem_wanted;
 	double		dpartitions;
@@ -2095,7 +2112,8 @@ initialize_hash_entry(AggState *aggstate, TupleHashTable hashtable,
 		AggStatePerTrans pertrans = &aggstate->pertrans[transno];
 		AggStatePerGroup pergroupstate = &pergroup[transno];
 
-		initialize_aggregate(aggstate, pertrans, pergroupstate);
+		initialize_aggregate(aggstate, pertrans, pergroupstate,
+							 aggstate->phase->aggnode->sortWorkMemLimit);
 	}
 }
 
@@ -2156,7 +2174,8 @@ lookup_hash_entries(AggState *aggstate)
 			if (spill->partitions == NULL)
 				hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
 								   perhash->aggnode->numGroups,
-								   aggstate->hashentrysize);
+								   aggstate->hashentrysize,
+								   (Size) aggstate->ss.ps.plan->workmem_limit * 1024);
 
 			hashagg_spill_tuple(aggstate, spill, slot, hash);
 			pergroup[setno] = NULL;
@@ -2630,7 +2649,9 @@ agg_refill_hash_table(AggState *aggstate)
 	aggstate->hash_batches = list_delete_last(aggstate->hash_batches);
 
 	hash_agg_set_limits(aggstate->hashentrysize, batch->input_card,
-						batch->used_bits, &aggstate->hash_mem_limit,
+						batch->used_bits,
+						(Size) aggstate->ss.ps.plan->workmem_limit * 1024,
+						&aggstate->hash_mem_limit,
 						&aggstate->hash_ngroups_limit, NULL);
 
 	/*
@@ -2718,7 +2739,8 @@ agg_refill_hash_table(AggState *aggstate)
 				 */
 				spill_initialized = true;
 				hashagg_spill_init(&spill, tapeset, batch->used_bits,
-								   batch->input_card, aggstate->hashentrysize);
+								   batch->input_card, aggstate->hashentrysize,
+								   (Size) aggstate->ss.ps.plan->workmem_limit * 1024);
 			}
 			/* no memory for a new group, spill */
 			hashagg_spill_tuple(aggstate, &spill, spillslot, hash);
@@ -2916,13 +2938,15 @@ agg_retrieve_hash_table_in_memory(AggState *aggstate)
  */
 static void
 hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset, int used_bits,
-				   double input_groups, double hashentrysize)
+				   double input_groups, double hashentrysize,
+				   Size hash_mem_limit)
 {
 	int			npartitions;
 	int			partition_bits;
 
 	npartitions = hash_choose_num_partitions(input_groups, hashentrysize,
-											 used_bits, &partition_bits);
+											 used_bits, hash_mem_limit,
+											 &partition_bits);
 
 #ifdef USE_INJECTION_POINTS
 	if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-single-partition"))
@@ -3649,6 +3673,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 			totalGroups += aggstate->perhash[k].aggnode->numGroups;
 
 		hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
+							(Size) aggstate->ss.ps.plan->workmem_limit * 1024,
 							&aggstate->hash_mem_limit,
 							&aggstate->hash_ngroups_limit,
 							&aggstate->hash_planned_partitions);
diff --git a/src/backend/executor/nodeBitmapIndexscan.c b/src/backend/executor/nodeBitmapIndexscan.c
index 0b32c3a022f..5e006baa88d 100644
--- a/src/backend/executor/nodeBitmapIndexscan.c
+++ b/src/backend/executor/nodeBitmapIndexscan.c
@@ -91,7 +91,8 @@ MultiExecBitmapIndexScan(BitmapIndexScanState *node)
 	else
 	{
 		/* XXX should we use less than work_mem for this? */
-		tbm = tbm_create(work_mem * (Size) 1024,
+		Assert(node->ss.ps.plan->workmem_limit > 0);
+		tbm = tbm_create((Size) node->ss.ps.plan->workmem_limit * 1024,
 						 ((BitmapIndexScan *) node->ss.ps.plan)->isshared ?
 						 node->ss.ps.state->es_query_dsa : NULL);
 	}
diff --git a/src/backend/executor/nodeBitmapOr.c b/src/backend/executor/nodeBitmapOr.c
index 231760ec93d..4ba32639f7d 100644
--- a/src/backend/executor/nodeBitmapOr.c
+++ b/src/backend/executor/nodeBitmapOr.c
@@ -143,7 +143,8 @@ MultiExecBitmapOr(BitmapOrState *node)
 			if (result == NULL) /* first subplan */
 			{
 				/* XXX should we use less than work_mem for this? */
-				result = tbm_create(work_mem * (Size) 1024,
+				Assert(subnode->plan->workmem_limit > 0);
+				result = tbm_create((Size) subnode->plan->workmem_limit * 1024,
 									((BitmapOr *) node->ps.plan)->isshared ?
 									node->ps.state->es_query_dsa : NULL);
 			}
diff --git a/src/backend/executor/nodeCtescan.c b/src/backend/executor/nodeCtescan.c
index e1675f66b43..2272185dce7 100644
--- a/src/backend/executor/nodeCtescan.c
+++ b/src/backend/executor/nodeCtescan.c
@@ -232,7 +232,8 @@ ExecInitCteScan(CteScan *node, EState *estate, int eflags)
 		/* I am the leader */
 		prmdata->value = PointerGetDatum(scanstate);
 		scanstate->leader = scanstate;
-		scanstate->cte_table = tuplestore_begin_heap(true, false, work_mem);
+		scanstate->cte_table =
+			tuplestore_begin_heap(true, false, node->scan.plan.workmem_limit);
 		tuplestore_set_eflags(scanstate->cte_table, scanstate->eflags);
 		scanstate->readptr = 0;
 	}
diff --git a/src/backend/executor/nodeFunctionscan.c b/src/backend/executor/nodeFunctionscan.c
index 644363582d9..bbb93a8dd58 100644
--- a/src/backend/executor/nodeFunctionscan.c
+++ b/src/backend/executor/nodeFunctionscan.c
@@ -95,6 +95,7 @@ FunctionNext(FunctionScanState *node)
 											node->ss.ps.ps_ExprContext,
 											node->argcontext,
 											node->funcstates[0].tupdesc,
+											node->ss.ps.plan->workmem_limit,
 											node->eflags & EXEC_FLAG_BACKWARD);
 
 			/*
@@ -154,6 +155,7 @@ FunctionNext(FunctionScanState *node)
 											node->ss.ps.ps_ExprContext,
 											node->argcontext,
 											fs->tupdesc,
+											node->ss.ps.plan->workmem_limit,
 											node->eflags & EXEC_FLAG_BACKWARD);
 
 			/*
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index d54cfe5fdbe..60afda04069 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -38,6 +38,7 @@
 #include "optimizer/cost.h"
 #include "port/pg_bitutils.h"
 #include "utils/dynahash.h"
+#include "utils/guc.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/syscache.h"
@@ -449,6 +450,7 @@ ExecHashTableCreate(HashState *state)
 	Hash	   *node;
 	HashJoinTable hashtable;
 	Plan	   *outerNode;
+	size_t		worker_space_allowed;
 	size_t		space_allowed;
 	int			nbuckets;
 	int			nbatch;
@@ -473,8 +475,12 @@ ExecHashTableCreate(HashState *state)
 	 */
 	rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
 
+	worker_space_allowed = (size_t) node->plan.workmem_limit * 1024;
+	Assert(worker_space_allowed > 0);
+
 	ExecChooseHashTableSize(rows, outerNode->plan_width,
 							OidIsValid(node->skewTable),
+							worker_space_allowed,
 							state->parallel_state != NULL,
 							state->parallel_state != NULL ?
 							state->parallel_state->nparticipants - 1 : 0,
@@ -601,6 +607,7 @@ ExecHashTableCreate(HashState *state)
 		{
 			pstate->nbatch = nbatch;
 			pstate->space_allowed = space_allowed;
+			pstate->worker_space_allowed = worker_space_allowed;
 			pstate->growth = PHJ_GROWTH_OK;
 
 			/* Set up the shared state for coordinating batches. */
@@ -658,9 +665,10 @@ ExecHashTableCreate(HashState *state)
 
 void
 ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
+						size_t worker_space_allowed,
 						bool try_combined_hash_mem,
 						int parallel_workers,
-						size_t *space_allowed,
+						size_t *total_space_allowed,
 						int *numbuckets,
 						int *numbatches,
 						int *num_skew_mcvs,
@@ -690,9 +698,9 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
 	inner_rel_bytes = ntuples * tupsize;
 
 	/*
-	 * Compute in-memory hashtable size limit from GUCs.
+	 * Caller tells us our (per-worker) in-memory hashtable size limit.
 	 */
-	hash_table_bytes = get_hash_memory_limit();
+	hash_table_bytes = worker_space_allowed;
 
 	/*
 	 * Parallel Hash tries to use the combined hash_mem of all workers to
@@ -709,7 +717,7 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
 		hash_table_bytes = (size_t) newlimit;
 	}
 
-	*space_allowed = hash_table_bytes;
+	*total_space_allowed = hash_table_bytes;
 
 	/*
 	 * If skew optimization is possible, estimate the number of skew buckets
@@ -813,8 +821,9 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
 		if (try_combined_hash_mem)
 		{
 			ExecChooseHashTableSize(ntuples, tupwidth, useskew,
-									false, parallel_workers,
-									space_allowed,
+									worker_space_allowed, false,
+									parallel_workers,
+									total_space_allowed,
 									numbuckets,
 									numbatches,
 									num_skew_mcvs,
@@ -1242,7 +1251,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
 					 * to switch from one large combined memory budget to the
 					 * regular hash_mem budget.
 					 */
-					pstate->space_allowed = get_hash_memory_limit();
+					pstate->space_allowed = pstate->worker_space_allowed;
 
 					/*
 					 * The combined hash_mem of all participants wasn't
diff --git a/src/backend/executor/nodeIncrementalSort.c b/src/backend/executor/nodeIncrementalSort.c
index 975b0397e7a..503d75e364b 100644
--- a/src/backend/executor/nodeIncrementalSort.c
+++ b/src/backend/executor/nodeIncrementalSort.c
@@ -312,7 +312,7 @@ switchToPresortedPrefixMode(PlanState *pstate)
 												&(plannode->sort.sortOperators[nPresortedCols]),
 												&(plannode->sort.collations[nPresortedCols]),
 												&(plannode->sort.nullsFirst[nPresortedCols]),
-												work_mem,
+												plannode->sort.plan.workmem_limit,
 												NULL,
 												node->bounded ? TUPLESORT_ALLOWBOUNDED : TUPLESORT_NONE);
 		node->prefixsort_state = prefixsort_state;
@@ -613,7 +613,7 @@ ExecIncrementalSort(PlanState *pstate)
 												  plannode->sort.sortOperators,
 												  plannode->sort.collations,
 												  plannode->sort.nullsFirst,
-												  work_mem,
+												  plannode->sort.plan.workmem_limit,
 												  NULL,
 												  node->bounded ?
 												  TUPLESORT_ALLOWBOUNDED :
diff --git a/src/backend/executor/nodeMaterial.c b/src/backend/executor/nodeMaterial.c
index 9798bb75365..10f764c1bd5 100644
--- a/src/backend/executor/nodeMaterial.c
+++ b/src/backend/executor/nodeMaterial.c
@@ -61,7 +61,8 @@ ExecMaterial(PlanState *pstate)
 	 */
 	if (tuplestorestate == NULL && node->eflags != 0)
 	{
-		tuplestorestate = tuplestore_begin_heap(true, false, work_mem);
+		tuplestorestate =
+			tuplestore_begin_heap(true, false, node->ss.ps.plan->workmem_limit);
 		tuplestore_set_eflags(tuplestorestate, node->eflags);
 		if (node->eflags & EXEC_FLAG_MARK)
 		{
diff --git a/src/backend/executor/nodeMemoize.c b/src/backend/executor/nodeMemoize.c
index 609deb12afb..a3fc37745ca 100644
--- a/src/backend/executor/nodeMemoize.c
+++ b/src/backend/executor/nodeMemoize.c
@@ -1036,7 +1036,7 @@ ExecInitMemoize(Memoize *node, EState *estate, int eflags)
 	mstate->mem_used = 0;
 
 	/* Limit the total memory consumed by the cache to this */
-	mstate->mem_limit = get_hash_memory_limit();
+	mstate->mem_limit = (Size) node->plan.workmem_limit * 1024;
 
 	/* A memory context dedicated for the cache */
 	mstate->tableContext = AllocSetContextCreate(CurrentMemoryContext,
diff --git a/src/backend/executor/nodeRecursiveunion.c b/src/backend/executor/nodeRecursiveunion.c
index 40f66fd0680..96dc8d53db3 100644
--- a/src/backend/executor/nodeRecursiveunion.c
+++ b/src/backend/executor/nodeRecursiveunion.c
@@ -52,6 +52,7 @@ build_hash_table(RecursiveUnionState *rustate)
 											 node->dupCollations,
 											 node->numGroups,
 											 0,
+											 (Size) node->hashWorkMemLimit * 1024,
 											 rustate->ps.state->es_query_cxt,
 											 rustate->tableContext,
 											 rustate->tempContext,
@@ -202,8 +203,15 @@ ExecInitRecursiveUnion(RecursiveUnion *node, EState *estate, int eflags)
 	/* initialize processing state */
 	rustate->recursing = false;
 	rustate->intermediate_empty = true;
-	rustate->working_table = tuplestore_begin_heap(false, false, work_mem);
-	rustate->intermediate_table = tuplestore_begin_heap(false, false, work_mem);
+
+	/*
+	 * NOTE: each of our working tables gets the same workmem_limit, since
+	 * we're going to swap them repeatedly.
+	 */
+	rustate->working_table =
+		tuplestore_begin_heap(false, false, node->plan.workmem_limit);
+	rustate->intermediate_table =
+		tuplestore_begin_heap(false, false, node->plan.workmem_limit);
 
 	/*
 	 * If hashing, we need a per-tuple memory context for comparisons, and a
diff --git a/src/backend/executor/nodeSetOp.c b/src/backend/executor/nodeSetOp.c
index 5b7ff9c3748..7b71adf05dc 100644
--- a/src/backend/executor/nodeSetOp.c
+++ b/src/backend/executor/nodeSetOp.c
@@ -105,6 +105,7 @@ build_hash_table(SetOpState *setopstate)
 												node->cmpCollations,
 												node->numGroups,
 												sizeof(SetOpStatePerGroupData),
+												(Size) node->plan.workmem_limit * 1024,
 												setopstate->ps.state->es_query_cxt,
 												setopstate->tableContext,
 												econtext->ecxt_per_tuple_memory,
diff --git a/src/backend/executor/nodeSort.c b/src/backend/executor/nodeSort.c
index f603337ecd3..1da77ab1d6a 100644
--- a/src/backend/executor/nodeSort.c
+++ b/src/backend/executor/nodeSort.c
@@ -107,7 +107,7 @@ ExecSort(PlanState *pstate)
 												   plannode->sortOperators[0],
 												   plannode->collations[0],
 												   plannode->nullsFirst[0],
-												   work_mem,
+												   plannode->plan.workmem_limit,
 												   NULL,
 												   tuplesortopts);
 		else
@@ -117,7 +117,7 @@ ExecSort(PlanState *pstate)
 												  plannode->sortOperators,
 												  plannode->collations,
 												  plannode->nullsFirst,
-												  work_mem,
+												  plannode->plan.workmem_limit,
 												  NULL,
 												  tuplesortopts);
 		if (node->bounded)
diff --git a/src/backend/executor/nodeSubplan.c b/src/backend/executor/nodeSubplan.c
index 49767ed6a52..73214501238 100644
--- a/src/backend/executor/nodeSubplan.c
+++ b/src/backend/executor/nodeSubplan.c
@@ -546,6 +546,7 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
 											  node->tab_collations,
 											  nbuckets,
 											  0,
+											  (Size) subplan->hashtab_workmem_limit * 1024,
 											  node->planstate->state->es_query_cxt,
 											  node->hashtablecxt,
 											  node->hashtempcxt,
@@ -575,6 +576,7 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
 												  node->tab_collations,
 												  nbuckets,
 												  0,
+												  (Size) subplan->hashnul_workmem_limit * 1024,
 												  node->planstate->state->es_query_cxt,
 												  node->hashtablecxt,
 												  node->hashtempcxt,
diff --git a/src/backend/executor/nodeTableFuncscan.c b/src/backend/executor/nodeTableFuncscan.c
index 83ade3f9437..8a9e534a743 100644
--- a/src/backend/executor/nodeTableFuncscan.c
+++ b/src/backend/executor/nodeTableFuncscan.c
@@ -276,7 +276,8 @@ tfuncFetchRows(TableFuncScanState *tstate, ExprContext *econtext)
 
 	/* build tuplestore for the result */
 	oldcxt = MemoryContextSwitchTo(econtext->ecxt_per_query_memory);
-	tstate->tupstore = tuplestore_begin_heap(false, false, work_mem);
+	tstate->tupstore = tuplestore_begin_heap(false, false,
+											 tstate->ss.ps.plan->workmem_limit);
 
 	/*
 	 * Each call to fetch a new set of rows - of which there may be very many
diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c
index 9a1acce2b5d..76819d140ba 100644
--- a/src/backend/executor/nodeWindowAgg.c
+++ b/src/backend/executor/nodeWindowAgg.c
@@ -1092,7 +1092,8 @@ prepare_tuplestore(WindowAggState *winstate)
 	Assert(winstate->buffer == NULL);
 
 	/* Create new tuplestore */
-	winstate->buffer = tuplestore_begin_heap(false, false, work_mem);
+	winstate->buffer = tuplestore_begin_heap(false, false,
+											 node->plan.workmem_limit);
 
 	/*
 	 * Set up read pointers for the tuplestore.  The current pointer doesn't
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 7c1fdde842b..fecea810b6e 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -1119,7 +1119,6 @@ cost_bitmap_heap_scan(Path *path, PlannerInfo *root, RelOptInfo *baserel,
 	path->startup_cost = startup_cost;
 	path->total_cost = startup_cost + run_cost;
 
-
 	/*
 	 * Set an overall working-memory estimate for the entire BitmapHeapPath --
 	 * including all of the IndexPaths and BitmapOrPaths in its bitmapqual.
@@ -2875,7 +2874,8 @@ cost_agg(Path *path, PlannerInfo *root,
 		hashentrysize = hash_agg_entry_size(list_length(root->aggtransinfos),
 											input_width,
 											aggcosts->transitionSpace);
-		hash_agg_set_limits(hashentrysize, numGroups, 0, &mem_limit,
+		hash_agg_set_limits(hashentrysize, numGroups, 0,
+							get_hash_memory_limit(), &mem_limit,
 							&ngroups_limit, &num_partitions);
 
 		nbatches = Max((numGroups * hashentrysize) / mem_limit,
@@ -4323,6 +4323,7 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
 	ExecChooseHashTableSize(inner_path_rows_total,
 							inner_path->pathtarget->width,
 							true,	/* useskew */
+							get_hash_memory_limit(),
 							parallel_hash,	/* try_combined_hash_mem */
 							outer_path->parallel_workers,
 							&space_allowed,
@@ -4651,15 +4652,19 @@ cost_subplan(PlannerInfo *root, SubPlan *subplan, Plan *plan)
 
 		/*
 		 * Estimate working memory needed for the hashtable (and hashnulls, if
-		 * needed). The logic below MUST match the logic in buildSubPlanHash()
-		 * and ExecInitSubPlan().
+		 * needed). The "nbuckets" estimate must match the logic in
+		 * buildSubPlanHash() and ExecInitSubPlan().
 		 */
 		nbuckets = clamp_cardinality_to_long(plan->plan_rows);
 		if (nbuckets < 1)
 			nbuckets = 1;
 
+		/*
+		 * This estimate must match the logic in subpath_is_hashable() (and
+		 * see comments there).
+		 */
 		hashentrysize = MAXALIGN(plan->plan_width) +
-			MAXALIGN(SizeofMinimalTupleHeader);
+			MAXALIGN(SizeofHeapTupleHeader);
 
 		subplan->hashtab_workmem =
 			normalize_workmem((double) nbuckets * hashentrysize);
diff --git a/src/include/commands/explain.h b/src/include/commands/explain.h
index 50454952eb2..498a1a3a4b6 100644
--- a/src/include/commands/explain.h
+++ b/src/include/commands/explain.h
@@ -72,6 +72,7 @@ typedef struct ExplainState
 								 * entry */
 	int			num_workers;	/* # of worker processes planned to use */
 	double		total_workmem;	/* total working memory estimate (in bytes) */
+	double		total_workmem_limit;	/* total working-memory limit (in kB) */
 	/* state related to the current plan node */
 	ExplainWorkersState *workers_state; /* needed if parallel plan */
 } ExplainState;
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index d12e3f451d2..c4147876d55 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -140,6 +140,7 @@ extern TupleHashTable BuildTupleHashTable(PlanState *parent,
 										  Oid *collations,
 										  long nbuckets,
 										  Size additionalsize,
+										  Size hash_mem_limit,
 										  MemoryContext metacxt,
 										  MemoryContext tablecxt,
 										  MemoryContext tempcxt,
@@ -499,6 +500,7 @@ extern Tuplestorestate *ExecMakeTableFunctionResult(SetExprState *setexpr,
 													ExprContext *econtext,
 													MemoryContext argContext,
 													TupleDesc expectedDesc,
+													int workMem,
 													bool randomAccess);
 extern SetExprState *ExecInitFunctionResultSet(Expr *expr,
 											   ExprContext *econtext, PlanState *parent);
@@ -724,4 +726,9 @@ extern ResultRelInfo *ExecLookupResultRelByOid(ModifyTableState *node,
 											   bool missing_ok,
 											   bool update_cache);
 
+/*
+ * prototypes from functions in execWorkmem.c
+ */
+extern void ExecAssignWorkMem(PlannedStmt *plannedstmt);
+
 #endif							/* EXECUTOR_H  */
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index ecff4842fd3..9b184c47322 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -253,7 +253,8 @@ typedef struct ParallelHashJoinState
 	ParallelHashGrowth growth;	/* control batch/bucket growth */
 	dsa_pointer chunk_work_queue;	/* chunk work queue */
 	int			nparticipants;
-	size_t		space_allowed;
+	size_t		space_allowed;	/* -- might be shared with other workers */
+	size_t		worker_space_allowed;	/* -- exclusive to this worker */
 	size_t		total_tuples;	/* total number of inner tuples */
 	LWLock		lock;			/* lock protecting the above */
 
diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h
index 34b82d0f5d1..728006b3ff5 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -329,8 +329,9 @@ extern void ExecReScanAgg(AggState *node);
 extern Size hash_agg_entry_size(int numTrans, Size tupleWidth,
 								Size transitionSpace);
 extern void hash_agg_set_limits(double hashentrysize, double input_groups,
-								int used_bits, Size *mem_limit,
-								uint64 *ngroups_limit, int *num_partitions);
+								int used_bits, Size hash_mem_limit,
+								Size *mem_limit, uint64 *ngroups_limit,
+								int *num_partitions);
 
 /* parallel instrumentation support */
 extern void ExecAggEstimate(AggState *node, ParallelContext *pcxt);
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
index fc5b20994dd..6a40730c065 100644
--- a/src/include/executor/nodeHash.h
+++ b/src/include/executor/nodeHash.h
@@ -57,9 +57,10 @@ extern bool ExecParallelScanHashTableForUnmatched(HashJoinState *hjstate,
 extern void ExecHashTableReset(HashJoinTable hashtable);
 extern void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
 extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
+									size_t worker_space_allowed,
 									bool try_combined_hash_mem,
 									int parallel_workers,
-									size_t *space_allowed,
+									size_t *total_space_allowed,
 									int *numbuckets,
 									int *numbatches,
 									int *num_skew_mcvs,
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index d3f8fd7bd6c..445953c77d3 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -169,6 +169,7 @@ typedef struct Plan
 	Cost		total_cost;
 
 	int			workmem;		/* estimated work_mem (in KB) */
+	int			workmem_limit;	/* work_mem limit per parallel worker (in KB) */
 
 	/*
 	 * planner's estimate of result size of this plan step
@@ -237,7 +238,7 @@ typedef struct Plan
 
 /* ----------------
  *	 Result node -
- *		If no outer plan, evaluate a variable-free targetlist.
+z *		If no outer plan, evaluate a variable-free targetlist.
  *		If outer plan, return tuples from outer plan (after a level of
  *		projection as shown by targetlist).
  *
@@ -433,6 +434,8 @@ typedef struct RecursiveUnion
 
 	/* estimated work_mem for hash table (in KB) */
 	int			hashWorkMem;
+	/* work_mem reserved for hash table */
+	int			hashWorkMemLimit;
 } RecursiveUnion;
 
 /* ----------------
@@ -1158,6 +1161,9 @@ typedef struct Agg
 	/* estimated work_mem needed to sort each input (in KB) */
 	int			sortWorkMem;
 
+	/* work_mem limit to sort one input (in KB) */
+	int			sortWorkMemLimit;
+
 	/* estimated number of groups in input */
 	long		numGroups;
 
diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h
index b7d6b0fe7dc..7232d07e8b8 100644
--- a/src/include/nodes/primnodes.h
+++ b/src/include/nodes/primnodes.h
@@ -1111,6 +1111,8 @@ typedef struct SubPlan
 	Cost		per_call_cost;	/* cost for each subplan evaluation */
 	int			hashtab_workmem;	/* estimated hashtable work_mem (in KB) */
 	int			hashnul_workmem;	/* estimated hashnulls work_mem (in KB) */
+	int			hashtab_workmem_limit;	/* hashtable work_mem limit (in kB) */
+	int			hashnul_workmem_limit;	/* hashnulls work_mem limit (in kB) */
 } SubPlan;
 
 /*
diff --git a/src/test/regress/expected/workmem.out b/src/test/regress/expected/workmem.out
index 215180808f4..c1a3bdd93d2 100644
--- a/src/test/regress/expected/workmem.out
+++ b/src/test/regress/expected/workmem.out
@@ -29,17 +29,18 @@ order by unique1;
 ');
                          workmem_filter                          
 -----------------------------------------------------------------
- Sort  (work_mem=N kB)
+ Sort  (work_mem=N kB limit=4096 kB)
    Sort Key: onek.unique1
    ->  Nested Loop
-         ->  HashAggregate  (work_mem=N kB)
+         ->  HashAggregate  (work_mem=N kB limit=8192 kB)
                Group Key: "*VALUES*".column1, "*VALUES*".column2
                ->  Values Scan on "*VALUES*"
          ->  Index Scan using onek_unique1 on onek
                Index Cond: (unique1 = "*VALUES*".column1)
                Filter: ("*VALUES*".column2 = ten)
  Total Working Memory: N kB
-(10 rows)
+ Total Working Memory Limit: 12288 kB
+(11 rows)
 
 select *
 from onek
@@ -64,18 +65,19 @@ order by unique1;
 ');
                             workmem_filter                            
 ----------------------------------------------------------------------
- Sort  (work_mem=N kB)
+ Sort  (work_mem=N kB limit=4096 kB)
    Sort Key: onek.unique1
    ->  Nested Loop
          ->  Unique
-               ->  Sort  (work_mem=N kB)
+               ->  Sort  (work_mem=N kB limit=4096 kB)
                      Sort Key: "*VALUES*".column1, "*VALUES*".column2
                      ->  Values Scan on "*VALUES*"
          ->  Index Scan using onek_unique1 on onek
                Index Cond: (unique1 = "*VALUES*".column1)
                Filter: ("*VALUES*".column2 = ten)
  Total Working Memory: N kB
-(11 rows)
+ Total Working Memory Limit: 8192 kB
+(12 rows)
 
 select *
 from onek
@@ -95,17 +97,18 @@ explain (costs off, work_mem on)
 select * from (select * from tenk1 order by four) t order by four, ten
 limit 1;
 ');
-             workmem_filter              
------------------------------------------
+                    workmem_filter                     
+-------------------------------------------------------
  Limit
-   ->  Incremental Sort  (work_mem=N kB)
+   ->  Incremental Sort  (work_mem=N kB limit=8192 kB)
          Sort Key: tenk1.four, tenk1.ten
          Presorted Key: tenk1.four
-         ->  Sort  (work_mem=N kB)
+         ->  Sort  (work_mem=N kB limit=4096 kB)
                Sort Key: tenk1.four
                ->  Seq Scan on tenk1
  Total Working Memory: N kB
-(8 rows)
+ Total Working Memory Limit: 12288 kB
+(9 rows)
 
 select * from (select * from tenk1 order by four) t order by four, ten
 limit 1;
@@ -131,16 +134,17 @@ where exists (select 1 from tenk1 t3
    ->  Nested Loop
          ->  Hash Join
                Hash Cond: (t3.thousand = t1.unique1)
-               ->  HashAggregate  (work_mem=N kB)
+               ->  HashAggregate  (work_mem=N kB limit=8192 kB)
                      Group Key: t3.thousand, t3.tenthous
                      ->  Index Only Scan using tenk1_thous_tenthous on tenk1 t3
-               ->  Hash  (work_mem=N kB)
+               ->  Hash  (work_mem=N kB limit=8192 kB)
                      ->  Index Only Scan using onek_unique1 on onek t1
                            Index Cond: (unique1 < 1)
          ->  Index Only Scan using tenk1_hundred on tenk1 t2
                Index Cond: (hundred = t3.tenthous)
  Total Working Memory: N kB
-(13 rows)
+ Total Working Memory Limit: 16384 kB
+(14 rows)
 
 select count(*) from (
 select t1.unique1, t2.hundred
@@ -165,23 +169,24 @@ from int4_tbl t1, int4_tbl t2
 where t4.f1 is null
 ) t;
 ');
-                       workmem_filter                        
--------------------------------------------------------------
+                              workmem_filter                              
+--------------------------------------------------------------------------
  Aggregate
    ->  Nested Loop
          ->  Nested Loop Left Join
                Filter: (t4.f1 IS NULL)
                ->  Seq Scan on int4_tbl t2
-               ->  Materialize  (work_mem=N kB)
+               ->  Materialize  (work_mem=N kB limit=4096 kB)
                      ->  Nested Loop Left Join
                            Join Filter: (t3.f1 > 1)
                            ->  Seq Scan on int4_tbl t3
                                  Filter: (f1 > 0)
-                           ->  Materialize  (work_mem=N kB)
+                           ->  Materialize  (work_mem=N kB limit=4096 kB)
                                  ->  Seq Scan on int4_tbl t4
          ->  Seq Scan on int4_tbl t1
  Total Working Memory: N kB
-(14 rows)
+ Total Working Memory Limit: 8192 kB
+(15 rows)
 
 select count(*) from (
 select t1.f1
@@ -204,16 +209,17 @@ group by grouping sets((a, b), (a));
 ');
                             workmem_filter                            
 ----------------------------------------------------------------------
- WindowAgg  (work_mem=N kB)
-   ->  Sort  (work_mem=N kB)
+ WindowAgg  (work_mem=N kB limit=4096 kB)
+   ->  Sort  (work_mem=N kB limit=4096 kB)
          Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
-         ->  HashAggregate  (work_mem=N kB)
+         ->  HashAggregate  (work_mem=N kB limit=8192 kB)
                Hash Key: "*VALUES*".column1, "*VALUES*".column2
                Hash Key: "*VALUES*".column1
                ->  Values Scan on "*VALUES*"
                      Filter: (column1 = column2)
  Total Working Memory: N kB
-(9 rows)
+ Total Working Memory Limit: 16384 kB
+(10 rows)
 
 select a, b, row_number() over (order by a, b nulls first)
 from (values (1, 1), (2, 2)) as t (a, b) where a = b
@@ -236,10 +242,10 @@ group by grouping sets((a, b), (a), (b), (c), (d));
 ');
                             workmem_filter                            
 ----------------------------------------------------------------------
- WindowAgg  (work_mem=N kB)
-   ->  Sort  (work_mem=N kB)
+ WindowAgg  (work_mem=N kB limit=4096 kB)
+   ->  Sort  (work_mem=N kB limit=4096 kB)
          Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
-         ->  GroupAggregate  (work_mem=N kB)
+         ->  GroupAggregate  (work_mem=N kB limit=8192 kB)
                Group Key: "*VALUES*".column1, "*VALUES*".column2
                Group Key: "*VALUES*".column1
                Sort Key: "*VALUES*".column2
@@ -248,12 +254,13 @@ group by grouping sets((a, b), (a), (b), (c), (d));
                  Group Key: "*VALUES*".column3
                Sort Key: "*VALUES*".column4
                  Group Key: "*VALUES*".column4
-               ->  Sort  (work_mem=N kB)
+               ->  Sort  (work_mem=N kB limit=4096 kB)
                      Sort Key: "*VALUES*".column1
                      ->  Values Scan on "*VALUES*"
                            Filter: (column1 = column2)
  Total Working Memory: N kB
-(17 rows)
+ Total Working Memory Limit: 20480 kB
+(18 rows)
 
 select a, b, row_number() over (order by a, b nulls first)
 from (values (1, 1, 1, 1), (2, 2, 2, 2)) as t (a, b, c, d) where a = b
@@ -282,17 +289,18 @@ select workmem_filter('
 explain (costs off, work_mem on)
 select length(stringu1) from tenk1 group by length(stringu1);
 ');
-                   workmem_filter                   
-----------------------------------------------------
- Finalize HashAggregate  (work_mem=N kB)
+                          workmem_filter                           
+-------------------------------------------------------------------
+ Finalize HashAggregate  (work_mem=N kB limit=8192 kB)
    Group Key: (length((stringu1)::text))
    ->  Gather
          Workers Planned: 4
-         ->  Partial HashAggregate  (work_mem=N kB)
+         ->  Partial HashAggregate  (work_mem=N kB limit=40960 kB)
                Group Key: length((stringu1)::text)
                ->  Parallel Seq Scan on tenk1
  Total Working Memory: N kB
-(8 rows)
+ Total Working Memory Limit: 49152 kB
+(9 rows)
 
 select length(stringu1) from tenk1 group by length(stringu1);
  length 
@@ -307,12 +315,13 @@ reset max_parallel_workers_per_gather;
 -- Agg (simple) [no work_mem]
 explain (costs off, work_mem on)
 select MAX(length(stringu1)) from tenk1;
-         QUERY PLAN         
-----------------------------
+            QUERY PLAN            
+----------------------------------
  Aggregate
    ->  Seq Scan on tenk1
  Total Working Memory: 0 kB
-(3 rows)
+ Total Working Memory Limit: 0 kB
+(4 rows)
 
 select MAX(length(stringu1)) from tenk1;
  max 
@@ -328,12 +337,13 @@ select sum(n) over(partition by m)
 from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
 ) t;
 ');
-                      workmem_filter                       
------------------------------------------------------------
+                             workmem_filter                              
+-------------------------------------------------------------------------
  Aggregate
-   ->  Function Scan on generate_series a  (work_mem=N kB)
+   ->  Function Scan on generate_series a  (work_mem=N kB limit=4096 kB)
  Total Working Memory: N kB
-(3 rows)
+ Total Working Memory Limit: 4096 kB
+(4 rows)
 
 select count(*) from (
 select sum(n) over(partition by m)
@@ -352,12 +362,13 @@ from rows from(generate_series(1, 5),
                generate_series(2, 10),
                generate_series(4, 15));
 ');
-                     workmem_filter                      
----------------------------------------------------------
+                             workmem_filter                             
+------------------------------------------------------------------------
  Aggregate
-   ->  Function Scan on generate_series  (work_mem=N kB)
+   ->  Function Scan on generate_series  (work_mem=N kB limit=12288 kB)
  Total Working Memory: N kB
-(3 rows)
+ Total Working Memory Limit: 12288 kB
+(4 rows)
 
 select count(*)
 from rows from(generate_series(1, 5),
@@ -384,13 +395,14 @@ SELECT  xmltable.*
                                   unit text PATH ''SIZE/@unit'',
                                   premier_name text PATH ''PREMIER_NAME'' DEFAULT ''not specified'');
 ');
-                      workmem_filter                      
-----------------------------------------------------------
+                             workmem_filter                             
+------------------------------------------------------------------------
  Nested Loop
    ->  Seq Scan on xmldata
-   ->  Table Function Scan on "xmltable"  (work_mem=N kB)
+   ->  Table Function Scan on "xmltable"  (work_mem=N kB limit=4096 kB)
  Total Working Memory: N kB
-(4 rows)
+ Total Working Memory Limit: 4096 kB
+(5 rows)
 
 SELECT  xmltable.*
    FROM (SELECT data FROM xmldata) x,
@@ -418,7 +430,8 @@ select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
    ->  Index Only Scan using tenk1_unique2 on tenk1 tenk1_1
          Filter: (unique2 <> 10)
  Total Working Memory: 0 kB
-(5 rows)
+ Total Working Memory Limit: 0 kB
+(6 rows)
 
 select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
  unique1 
@@ -435,11 +448,12 @@ select count(*) from
                           workmem_filter                          
 ------------------------------------------------------------------
  Aggregate
-   ->  HashSetOp Intersect  (work_mem=N kB)
+   ->  HashSetOp Intersect  (work_mem=N kB limit=8192 kB)
          ->  Seq Scan on tenk1
          ->  Index Only Scan using tenk1_unique1 on tenk1 tenk1_1
  Total Working Memory: N kB
-(5 rows)
+ Total Working Memory Limit: 8192 kB
+(6 rows)
 
 select count(*) from
   ( select unique1 from tenk1 intersect select fivethous from tenk1 ) ss;
@@ -456,23 +470,24 @@ cross join lateral (with recursive x(a) as (
           select o.four as a union select a + 1 from x where a < 10)
     select * from x) ss where o.ten = 1;
 ');
-                       workmem_filter                       
-------------------------------------------------------------
+                              workmem_filter                               
+---------------------------------------------------------------------------
  Aggregate
    ->  Nested Loop
          ->  Seq Scan on onek o
                Filter: (ten = 1)
-         ->  Memoize  (work_mem=N kB)
+         ->  Memoize  (work_mem=N kB limit=8192 kB)
                Cache Key: o.four
                Cache Mode: binary
-               ->  CTE Scan on x  (work_mem=N kB)
+               ->  CTE Scan on x  (work_mem=N kB limit=4096 kB)
                      CTE x
-                       ->  Recursive Union  (work_mem=N kB)
+                       ->  Recursive Union  (work_mem=N kB limit=16384 kB)
                              ->  Result
                              ->  WorkTable Scan on x x_1
                                    Filter: (a < 10)
  Total Working Memory: N kB
-(14 rows)
+ Total Working Memory Limit: 28672 kB
+(15 rows)
 
 select sum(o.four), sum(ss.a) from onek o
 cross join lateral (with recursive x(a) as (
@@ -491,20 +506,21 @@ WITH q1(x,y) AS (
   )
 SELECT count(*) FROM q1 WHERE y > (SELECT sum(y)/100 FROM q1 qsub);
 ');
-                   workmem_filter                   
-----------------------------------------------------
+                          workmem_filter                          
+------------------------------------------------------------------
  Aggregate
    CTE q1
-     ->  HashAggregate  (work_mem=N kB)
+     ->  HashAggregate  (work_mem=N kB limit=8192 kB)
            Group Key: tenk1.hundred
            ->  Seq Scan on tenk1
    InitPlan 2
      ->  Aggregate
-           ->  CTE Scan on q1 qsub  (work_mem=N kB)
-   ->  CTE Scan on q1  (work_mem=N kB)
+           ->  CTE Scan on q1 qsub  (work_mem=N kB limit=4096 kB)
+   ->  CTE Scan on q1  (work_mem=N kB limit=4096 kB)
          Filter: ((y)::numeric > (InitPlan 2).col1)
  Total Working Memory: N kB
-(11 rows)
+ Total Working Memory Limit: 16384 kB
+(12 rows)
 
 WITH q1(x,y) AS (
     SELECT hundred, sum(ten) FROM tenk1 GROUP BY hundred
@@ -522,15 +538,16 @@ select sum(n) over(partition by m)
 from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
 limit 5;
 ');
-                            workmem_filter                             
------------------------------------------------------------------------
+                                   workmem_filter                                    
+-------------------------------------------------------------------------------------
  Limit
-   ->  WindowAgg  (work_mem=N kB)
-         ->  Sort  (work_mem=N kB)
+   ->  WindowAgg  (work_mem=N kB limit=4096 kB)
+         ->  Sort  (work_mem=N kB limit=4096 kB)
                Sort Key: ((a.n < 3))
-               ->  Function Scan on generate_series a  (work_mem=N kB)
+               ->  Function Scan on generate_series a  (work_mem=N kB limit=4096 kB)
  Total Working Memory: N kB
-(6 rows)
+ Total Working Memory Limit: 12288 kB
+(7 rows)
 
 select sum(n) over(partition by m)
 from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
@@ -560,20 +577,21 @@ select * from tenk1 a join tenk1 b on
          ->  Bitmap Heap Scan on tenk1 b
                Recheck Cond: ((hundred = 4) OR (unique1 = 2))
                ->  BitmapOr
-                     ->  Bitmap Index Scan on tenk1_hundred  (work_mem=N kB)
+                     ->  Bitmap Index Scan on tenk1_hundred  (work_mem=N kB limit=4096 kB)
                            Index Cond: (hundred = 4)
-                     ->  Bitmap Index Scan on tenk1_unique1  (work_mem=N kB)
+                     ->  Bitmap Index Scan on tenk1_unique1  (work_mem=N kB limit=4096 kB)
                            Index Cond: (unique1 = 2)
-         ->  Materialize  (work_mem=N kB)
+         ->  Materialize  (work_mem=N kB limit=4096 kB)
                ->  Bitmap Heap Scan on tenk1 a
                      Recheck Cond: ((unique2 = 3) OR (unique1 = 1))
                      ->  BitmapOr
-                           ->  Bitmap Index Scan on tenk1_unique2  (work_mem=N kB)
+                           ->  Bitmap Index Scan on tenk1_unique2  (work_mem=N kB limit=4096 kB)
                                  Index Cond: (unique2 = 3)
-                           ->  Bitmap Index Scan on tenk1_unique1  (work_mem=N kB)
+                           ->  Bitmap Index Scan on tenk1_unique1  (work_mem=N kB limit=4096 kB)
                                  Index Cond: (unique1 = 1)
  Total Working Memory: N kB
-(19 rows)
+ Total Working Memory Limit: 20480 kB
+(20 rows)
 
 select count(*) from (
 select * from tenk1 a join tenk1 b on
@@ -589,15 +607,16 @@ select workmem_filter('
 explain (costs off, work_mem on)
 select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
 ');
-       workmem_filter       
-----------------------------
- Result  (work_mem=N kB)
+             workmem_filter             
+----------------------------------------
+ Result  (work_mem=N kB limit=16384 kB)
    SubPlan 1
      ->  Append
            ->  Result
            ->  Result
  Total Working Memory: N kB
-(6 rows)
+ Total Working Memory Limit: 16384 kB
+(7 rows)
 
 select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
  ?column? 
@@ -612,16 +631,17 @@ select 1 = any (select (select 1) where 1 = any (select 1));
 ');
                          workmem_filter                         
 ----------------------------------------------------------------
- Result  (work_mem=N kB)
+ Result  (work_mem=N kB limit=16384 kB)
    SubPlan 3
-     ->  Result  (work_mem=N kB)
+     ->  Result  (work_mem=N kB limit=8192 kB)
            One-Time Filter: (ANY (1 = (hashed SubPlan 2).col1))
            InitPlan 1
              ->  Result
            SubPlan 2
              ->  Result
  Total Working Memory: N kB
-(9 rows)
+ Total Working Memory Limit: 24576 kB
+(10 rows)
 
 select 1 = any (select (select 1) where 1 = any (select 1));
  ?column? 
-- 
2.47.1

