pg.ddx.io pgsql-hackers@postgresql.org mailing list archive help / color / mirror / Atom feed From: Álvaro Herrera <alvherre@kurilemu.de>
To: James Hunter <james.hunter.pg@gmail.com>
Cc: Jeff Davis <pgsql@j-davis.com>
Cc: pgsql-hackers@lists.postgresql.org <pgsql-hackers@lists.postgresql.org>
Subject: Re: Proposal: "query_work_mem" GUC, to distribute working memory to the query's individual operators
Date: Tue, 5 Aug 2025 14:15:51 +0200
Message-ID: <202508051215.x5niwfkbomdf@alvherre.pgsql> (raw )
In-Reply-To: <CAJVSvF7HaSz=-b1g5BCUML17=SdjbYV+pSFHH9WuRfRCGScGRw@mail.gmail.com >
Here's a rebased version of this patch. I didn't review it or touch it
in any way, just fixed conflicts from current master.
--
Álvaro Herrera Breisgau, Deutschland — https://www.EnterpriseDB.com/
"No deja de ser humillante para una persona de ingenio saber
que no hay tonto que no le pueda enseñar algo." (Jean B. Say)
Attachments: download all (4 files, .tar.gz)
0001-Store-working-memory-limit-per-Plan-SubPlan-rather-t.patch (text/x-diff, 54.5K) download
From 03df3a9feb6710cf34845fecfad3086389a45635 Mon Sep 17 00:00:00 2001
From: James Hunter <james.hunter.pg@gmail.com>
Date: Tue, 25 Feb 2025 22:44:01 +0000
Subject: [PATCH 1/4] Store working memory limit per Plan/SubPlan, rather than
in GUC
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
list, "workMemLimits", added to the PlannedStmt node. At runtimem an exec
node checks its limit by looking up the list element corresponding to its
plan->workmem_id field.
Indirecting the workMemLimit via a List index allows us to handle SubPlans
as well as Plans. It also allows a future extension to set limits on
individual Plans/SubPlans, without needing to re-traverse the Plan +
Expr tree.
To preserve backward, this commit also copies the "work_mem", etc., values
from the existing GUCs to the new field. This means that this commit is
just a refactoring, and doesn't change any behavior.
This "workmem_id" field is on the Plan node, instead of the corresponding
PlanState, because the workMemLimit needs to be set before we can call
ExecInitNode().
---
src/backend/executor/Makefile | 1 +
src/backend/executor/execGrouping.c | 10 +-
src/backend/executor/execMain.c | 6 +
src/backend/executor/execParallel.c | 2 +
src/backend/executor/execSRF.c | 5 +-
src/backend/executor/execWorkmem.c | 87 ++++++++++++
src/backend/executor/meson.build | 1 +
src/backend/executor/nodeAgg.c | 64 ++++++---
src/backend/executor/nodeBitmapIndexscan.c | 2 +-
src/backend/executor/nodeBitmapOr.c | 2 +-
src/backend/executor/nodeCtescan.c | 3 +-
src/backend/executor/nodeFunctionscan.c | 2 +
src/backend/executor/nodeHash.c | 22 +++-
src/backend/executor/nodeIncrementalSort.c | 4 +-
src/backend/executor/nodeMaterial.c | 3 +-
src/backend/executor/nodeMemoize.c | 2 +-
src/backend/executor/nodeRecursiveunion.c | 14 +-
src/backend/executor/nodeSetOp.c | 1 +
src/backend/executor/nodeSort.c | 4 +-
src/backend/executor/nodeSubplan.c | 16 +++
src/backend/executor/nodeTableFuncscan.c | 3 +-
src/backend/executor/nodeWindowAgg.c | 3 +-
src/backend/optimizer/path/costsize.c | 16 ++-
src/backend/optimizer/plan/createplan.c | 146 ++++++++++++++++++---
src/backend/optimizer/plan/planner.c | 5 +-
src/backend/optimizer/plan/subselect.c | 2 +-
src/include/executor/executor.h | 7 +
src/include/executor/hashjoin.h | 3 +-
src/include/executor/nodeAgg.h | 3 +-
src/include/executor/nodeHash.h | 3 +-
src/include/nodes/execnodes.h | 13 ++
src/include/nodes/pathnodes.h | 11 ++
src/include/nodes/plannodes.h | 27 +++-
src/include/nodes/primnodes.h | 3 +
src/include/optimizer/planmain.h | 4 +-
35 files changed, 434 insertions(+), 66 deletions(-)
create mode 100644 src/backend/executor/execWorkmem.c
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 b5400749353..24e8034e4ee 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,7 +176,6 @@ BuildTupleHashTable(PlanState *parent,
{
TupleHashTable hashtable;
Size entrysize;
- Size hash_mem_limit;
MemoryContext oldcontext;
bool allow_jit;
uint32 hash_iv = 0;
@@ -184,8 +184,12 @@ BuildTupleHashTable(PlanState *parent,
additionalsize = MAXALIGN(additionalsize);
entrysize = sizeof(TupleHashEntryData) + additionalsize;
- /* 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 0391798dd2c..6aa9dde5a80 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -949,6 +949,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/execParallel.c b/src/backend/executor/execParallel.c
index f098a5557cf..a8cb631963e 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -216,6 +216,8 @@ ExecSerializePlan(Plan *plan, EState *estate)
pstmt->utilityStmt = NULL;
pstmt->stmt_location = -1;
pstmt->stmt_len = -1;
+ pstmt->workMemCategories = estate->es_plannedstmt->workMemCategories;
+ pstmt->workMemLimits = estate->es_plannedstmt->workMemLimits;
/* Return serialized copy of our dummy PlannedStmt. */
return nodeToString(pstmt);
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..d8a19a58ebe
--- /dev/null
+++ b/src/backend/executor/execWorkmem.c
@@ -0,0 +1,87 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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
+ * its working-memory limit.
+ *
+ * Now, to allow different PlanState nodes to be restricted to different
+ * amounts of memory, each PlanState node reads this limit off the
+ * PlannedStmt's workMemLimits List, at the (1-based) position indicated
+ * by the PlanState's Plan node's "workmem_id" field.
+ *
+ * We assign the workmem_id and expand the workMemLimits List, when
+ * creating the Plan node; and then we set this limit by calling
+ * ExecAssignWorkMem(), from InitPlan(), before we initialize the PlanState
+ * nodes.
+ *
+ * The workMemLimit has always applied "per data structure," rather than
+ * "per PlanState". So a single SQL operator (e.g., RecursiveUnion) can
+ * use more than the workMemLimit, even though each of its data
+ * structures is restricted to it.
+ *
+ * We store the "workmem_id" 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 "workmem_id" fields on the Plan,
+ * but set the workMemLimit when we start execution, as part of
+ * InitPlan().
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/parallel.h"
+#include "executor/executor.h"
+#include "miscadmin.h"
+#include "nodes/plannodes.h"
+
+
+/* ------------------------------------------------------------------------
+ * ExecAssignWorkMem
+ *
+ * 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)
+{
+ ListCell *lc_category;
+ ListCell *lc_limit;
+
+ /*
+ * 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;
+
+ forboth(lc_category, plannedstmt->workMemCategories,
+ lc_limit, plannedstmt->workMemLimits)
+ {
+ lfirst_int(lc_limit) = lfirst_int(lc_category) == WORKMEM_HASH ?
+ get_hash_memory_limit() / 1024 : 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 377e016d732..bff143a8a8e 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 void hash_create_memory(AggState *aggstate);
@@ -412,6 +414,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,
@@ -434,7 +437,7 @@ static HashAggBatch *hashagg_batch_new(LogicalTape *input_tape, int setno,
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);
+ 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,
@@ -522,6 +525,15 @@ 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 =
+ workMemLimitFromId(aggstate,
+ aggstate->phases[1].aggnode->plan.workmem_id);
aggstate->sort_out = tuplesort_begin_heap(tupDesc,
sortnode->numCols,
@@ -529,7 +541,7 @@ initialize_phase(AggState *aggstate, int newphase)
sortnode->sortOperators,
sortnode->collations,
sortnode->nullsFirst,
- work_mem,
+ workmem_limit,
NULL, TUPLESORT_NONE);
}
@@ -585,6 +597,8 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
*/
if (pertrans->aggsortrequired)
{
+ int workmem_limit;
+
/*
* In case of rescan, maybe there could be an uncompleted sort
* operation? Clean it up if so.
@@ -592,6 +606,12 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
if (pertrans->sortstates[aggstate->current_set])
tuplesort_end(pertrans->sortstates[aggstate->current_set]);
+ /*
+ * Read the sort-input workmem limit off the first Agg node.
+ */
+ workmem_limit =
+ workMemLimitFromId(aggstate,
+ ((Agg *) aggstate->ss.ps.plan)->sortWorkMemId);
/*
* We use a plain Datum sorter when there's a single input column;
@@ -607,7 +627,7 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
pertrans->sortOperators[0],
pertrans->sortCollations[0],
pertrans->sortNullsFirst[0],
- work_mem, NULL, TUPLESORT_NONE);
+ workmem_limit, NULL, TUPLESORT_NONE);
}
else
pertrans->sortstates[aggstate->current_set] =
@@ -617,7 +637,7 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
pertrans->sortOperators,
pertrans->sortCollations,
pertrans->sortNullsFirst,
- work_mem, NULL, TUPLESORT_NONE);
+ workmem_limit, NULL, TUPLESORT_NONE);
}
/*
@@ -1496,7 +1516,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;
@@ -1506,7 +1526,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;
@@ -1535,6 +1556,7 @@ build_hash_table(AggState *aggstate, int setno, long nbuckets)
perhash->aggnode->grpCollations,
nbuckets,
additionalsize,
+ hash_mem_limit,
metacxt,
tablecxt,
tmpcxt,
@@ -1807,12 +1829,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)
@@ -1832,6 +1853,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;
@@ -1932,7 +1954,8 @@ hash_agg_enter_spill_mode(AggState *aggstate)
hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
perhash->aggnode->numGroups,
- aggstate->hashentrysize);
+ aggstate->hashentrysize,
+ (Size) workMemLimit(aggstate) * 1024);
}
}
}
@@ -2081,9 +2104,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;
@@ -2219,7 +2242,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) workMemLimit(aggstate) * 1024);
hashagg_spill_tuple(aggstate, spill, slot, hash);
pergroup[setno] = NULL;
@@ -2693,7 +2717,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) workMemLimit(aggstate) * 1024,
+ &aggstate->hash_mem_limit,
&aggstate->hash_ngroups_limit, NULL);
/*
@@ -2783,7 +2809,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) workMemLimit(aggstate) * 1024);
}
/* no memory for a new group, spill */
hashagg_spill_tuple(aggstate, &spill, spillslot, hash);
@@ -2982,13 +3009,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"))
@@ -3712,6 +3741,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
totalGroups += aggstate->perhash[k].aggnode->numGroups;
hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
+ (Size) workMemLimit(aggstate) * 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 abbb033881a..8bbf1d047c4 100644
--- a/src/backend/executor/nodeBitmapIndexscan.c
+++ b/src/backend/executor/nodeBitmapIndexscan.c
@@ -91,7 +91,7 @@ MultiExecBitmapIndexScan(BitmapIndexScanState *node)
else
{
/* XXX should we use less than work_mem for this? */
- tbm = tbm_create(work_mem * (Size) 1024,
+ tbm = tbm_create(workMemLimit(node) * (Size) 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..16d0a164292 100644
--- a/src/backend/executor/nodeBitmapOr.c
+++ b/src/backend/executor/nodeBitmapOr.c
@@ -143,7 +143,7 @@ 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,
+ result = tbm_create(workMemLimit(subnode) * (Size) 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..08f48f88e65 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, workMemLimit(scanstate));
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..fda42a278b8 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,
+ workMemLimit(node),
node->eflags & EXEC_FLAG_BACKWARD);
/*
@@ -154,6 +155,7 @@ FunctionNext(FunctionScanState *node)
node->ss.ps.ps_ExprContext,
node->argcontext,
fs->tupdesc,
+ workMemLimit(node),
node->eflags & EXEC_FLAG_BACKWARD);
/*
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 8d2201ab67f..bb9af08dc5d 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -448,6 +448,7 @@ ExecHashTableCreate(HashState *state)
Hash *node;
HashJoinTable hashtable;
Plan *outerNode;
+ size_t worker_space_allowed;
size_t space_allowed;
int nbuckets;
int nbatch;
@@ -471,11 +472,15 @@ ExecHashTableCreate(HashState *state)
*/
rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
+ worker_space_allowed = (size_t) workMemLimit(state) * 1024;
+ Assert(worker_space_allowed > 0);
+
ExecChooseHashTableSize(rows, outerNode->plan_width,
OidIsValid(node->skewTable),
state->parallel_state != NULL,
state->parallel_state != NULL ?
state->parallel_state->nparticipants - 1 : 0,
+ worker_space_allowed,
&space_allowed,
&nbuckets, &nbatch, &num_skew_mcvs);
@@ -599,6 +604,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,7 +664,8 @@ void
ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
bool try_combined_hash_mem,
int parallel_workers,
- size_t *space_allowed,
+ size_t worker_space_allowed,
+ size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
int *num_skew_mcvs)
@@ -687,9 +694,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
@@ -706,7 +713,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
@@ -808,7 +815,8 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
{
ExecChooseHashTableSize(ntuples, tupwidth, useskew,
false, parallel_workers,
- space_allowed,
+ worker_space_allowed,
+ total_space_allowed,
numbuckets,
numbatches,
num_skew_mcvs);
@@ -929,7 +937,7 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
nbatch /= 2;
nbuckets *= 2;
- *space_allowed = (*space_allowed) * 2;
+ *total_space_allowed = (*total_space_allowed) * 2;
}
Assert(nbuckets > 0);
@@ -1235,7 +1243,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..7a92c1eb2c0 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,
+ workMemLimit(pstate),
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,
+ workMemLimit(pstate),
NULL,
node->bounded ?
TUPLESORT_ALLOWBOUNDED :
diff --git a/src/backend/executor/nodeMaterial.c b/src/backend/executor/nodeMaterial.c
index 9798bb75365..bf5e921a205 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, workMemLimit(node));
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..4e3da4aab6b 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) workMemLimit(mstate) * 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..5ffffd327d2 100644
--- a/src/backend/executor/nodeRecursiveunion.c
+++ b/src/backend/executor/nodeRecursiveunion.c
@@ -33,6 +33,8 @@ build_hash_table(RecursiveUnionState *rustate)
{
RecursiveUnion *node = (RecursiveUnion *) rustate->ps.plan;
TupleDesc desc = ExecGetResultType(outerPlanState(rustate));
+ int workmem_limit = workMemLimitFromId(rustate,
+ node->hashWorkMemId);
Assert(node->numCols > 0);
Assert(node->numGroups > 0);
@@ -52,6 +54,7 @@ build_hash_table(RecursiveUnionState *rustate)
node->dupCollations,
node->numGroups,
0,
+ (Size) workmem_limit * 1024,
rustate->ps.state->es_query_cxt,
rustate->tableContext,
rustate->tempContext,
@@ -202,8 +205,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,
+ workMemLimit(rustate));
+ rustate->intermediate_table = tuplestore_begin_heap(false, false,
+ workMemLimit(rustate));
/*
* 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 4068481a523..0e2d02aa243 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) workMemLimit(setopstate) * 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..8ec939e25d7 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,
+ workMemLimit(pstate),
NULL,
tuplesortopts);
else
@@ -117,7 +117,7 @@ ExecSort(PlanState *pstate)
plannode->sortOperators,
plannode->collations,
plannode->nullsFirst,
- work_mem,
+ workMemLimit(pstate),
NULL,
tuplesortopts);
if (node->bounded)
diff --git a/src/backend/executor/nodeSubplan.c b/src/backend/executor/nodeSubplan.c
index f7f6fc2da0b..56036e79933 100644
--- a/src/backend/executor/nodeSubplan.c
+++ b/src/backend/executor/nodeSubplan.c
@@ -536,6 +536,12 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
if (node->hashtable)
ResetTupleHashTable(node->hashtable);
else
+ {
+ int workmem_limit;
+
+ workmem_limit = workMemLimitFromId(planstate,
+ subplan->hashtab_workmem_id);
+
node->hashtable = BuildTupleHashTable(node->parent,
node->descRight,
&TTSOpsVirtual,
@@ -546,10 +552,12 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
node->tab_collations,
nbuckets,
0,
+ (Size) workmem_limit * 1024,
node->planstate->state->es_query_cxt,
node->hashtablecxt,
node->hashtempcxt,
false);
+ }
if (!subplan->unknownEqFalse)
{
@@ -565,6 +573,12 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
if (node->hashnulls)
ResetTupleHashTable(node->hashnulls);
else
+ {
+ int workmem_limit;
+
+ workmem_limit = workMemLimitFromId(planstate,
+ subplan->hashnul_workmem_id);
+
node->hashnulls = BuildTupleHashTable(node->parent,
node->descRight,
&TTSOpsVirtual,
@@ -575,10 +589,12 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
node->tab_collations,
nbuckets,
0,
+ (Size) workmem_limit * 1024,
node->planstate->state->es_query_cxt,
node->hashtablecxt,
node->hashtempcxt,
false);
+ }
}
else
node->hashnulls = NULL;
diff --git a/src/backend/executor/nodeTableFuncscan.c b/src/backend/executor/nodeTableFuncscan.c
index 83ade3f9437..f679bd67bee 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,
+ workMemLimit(tstate));
/*
* 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..7660aa626b6 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,
+ workMemLimit(winstate));
/*
* 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 344a3188317..353f51fdff2 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -102,6 +102,7 @@
#include "optimizer/paths.h"
#include "optimizer/placeholder.h"
#include "optimizer/plancat.h"
+#include "optimizer/planmain.h"
#include "optimizer/restrictinfo.h"
#include "parser/parsetree.h"
#include "utils/lsyscache.h"
@@ -2833,7 +2834,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,
@@ -4256,6 +4258,7 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
true, /* useskew */
parallel_hash, /* try_combined_hash_mem */
outer_path->parallel_workers,
+ get_hash_memory_limit(),
&space_allowed,
&numbuckets,
&numbatches,
@@ -4583,6 +4586,17 @@ cost_subplan(PlannerInfo *root, SubPlan *subplan, Plan *plan)
sp_cost.startup += plan->total_cost +
cpu_operator_cost * plan->plan_rows;
+ /*
+ * Working memory needed for the hashtable (and hashnulls, if needed).
+ */
+ subplan->hashtab_workmem_id = add_hash_workmem(root->glob);
+
+ if (!subplan->unknownEqFalse)
+ {
+ /* Also needs a hashnulls table. */
+ subplan->hashnul_workmem_id = add_hash_workmem(root->glob);
+ }
+
/*
* The per-tuple costs include the cost of evaluating the lefthand
* expressions, plus the cost of probing the hashtable. We already
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index bfefc7dbea1..22834fe37f4 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -1706,6 +1706,8 @@ create_material_plan(PlannerInfo *root, MaterialPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_workmem(root->glob);
+
return plan;
}
@@ -1761,6 +1763,8 @@ create_memoize_plan(PlannerInfo *root, MemoizePath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_hash_workmem(root->glob);
+
return plan;
}
@@ -1907,6 +1911,8 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags)
best_path->path.rows,
0,
subplan);
+
+ plan->workmem_id = add_hash_workmem(root->glob);
}
else
{
@@ -2253,6 +2259,8 @@ create_sort_plan(PlannerInfo *root, SortPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_workmem(root->glob);
+
return plan;
}
@@ -2279,6 +2287,8 @@ create_incrementalsort_plan(PlannerInfo *root, IncrementalSortPath *best_path,
copy_generic_path_info(&plan->sort.plan, (Path *) best_path);
+ plan->sort.plan.workmem_id = add_workmem(root->glob);
+
return plan;
}
@@ -2390,6 +2400,12 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ if (plan->aggstrategy == AGG_HASHED)
+ plan->plan.workmem_id = add_hash_workmem(root->glob);
+
+ /* Also include working memory needed to sort the input: */
+ plan->sortWorkMemId = add_workmem(root->glob);
+
return plan;
}
@@ -2443,6 +2459,7 @@ static Plan *
create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
{
Agg *plan;
+ Agg *first_sort_agg = NULL;
Plan *subplan;
List *rollups = best_path->rollups;
AttrNumber *grouping_map;
@@ -2508,7 +2525,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
RollupData *rollup = lfirst(lc);
AttrNumber *new_grpColIdx;
Plan *sort_plan = NULL;
- Plan *agg_plan;
+ Agg *agg_plan;
AggStrategy strat;
new_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
@@ -2531,19 +2548,19 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
else
strat = AGG_SORTED;
- agg_plan = (Plan *) make_agg(NIL,
- NIL,
- strat,
- AGGSPLIT_SIMPLE,
- list_length((List *) linitial(rollup->gsets)),
- new_grpColIdx,
- extract_grouping_ops(rollup->groupClause),
- extract_grouping_collations(rollup->groupClause, subplan->targetlist),
- rollup->gsets,
- NIL,
- rollup->numGroups,
- best_path->transitionSpace,
- sort_plan);
+ agg_plan = make_agg(NIL,
+ NIL,
+ strat,
+ AGGSPLIT_SIMPLE,
+ list_length((List *) linitial(rollup->gsets)),
+ new_grpColIdx,
+ extract_grouping_ops(rollup->groupClause),
+ extract_grouping_collations(rollup->groupClause, subplan->targetlist),
+ rollup->gsets,
+ NIL,
+ rollup->numGroups,
+ best_path->transitionSpace,
+ sort_plan);
/*
* Remove stuff we don't need to avoid bloating debug output.
@@ -2554,6 +2571,12 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
sort_plan->lefttree = NULL;
}
+ if (agg_plan->aggstrategy == AGG_SORTED && !first_sort_agg)
+ {
+ /* This might be the first Sort agg. */
+ first_sort_agg = agg_plan;
+ }
+
chain = lappend(chain, agg_plan);
}
}
@@ -2586,6 +2609,29 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
/* Copy cost data from Path to Plan */
copy_generic_path_info(&plan->plan, &best_path->path);
+
+ /*
+ * NOTE: We will place the workmem needed to sort the input (if any)
+ * on the first agg, the Hash workmem on the first Hash agg, and the
+ * Sort workmem (if any) on the first Sort agg.
+ */
+ if (plan->aggstrategy == AGG_HASHED || plan->aggstrategy == AGG_MIXED)
+ {
+ /* All Hash Grouping Sets share the same workmem limit. */
+ plan->plan.workmem_id = add_hash_workmem(root->glob);
+ }
+ else if (plan->aggstrategy == AGG_SORTED)
+ {
+ /* Every Sort Grouping Set gets its own workmem limit. */
+ first_sort_agg = plan;
+ }
+
+ /* Store the workmem limit, for all Sorts, on the first Sort. */
+ if (first_sort_agg)
+ first_sort_agg->plan.workmem_id = add_workmem(root->glob);
+
+ /* Also include working memory needed to sort the input: */
+ plan->sortWorkMemId = add_workmem(root->glob);
}
return (Plan *) plan;
@@ -2750,6 +2796,8 @@ create_windowagg_plan(PlannerInfo *root, WindowAggPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_workmem(root->glob);
+
return plan;
}
@@ -2790,6 +2838,8 @@ create_setop_plan(PlannerInfo *root, SetOpPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_hash_workmem(root->glob);
+
return plan;
}
@@ -2826,6 +2876,12 @@ create_recursiveunion_plan(PlannerInfo *root, RecursiveUnionPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ plan->plan.workmem_id = add_workmem(root->glob);
+
+ /* Also include working memory for hash table. */
+ if (plan->numCols > 0)
+ plan->hashWorkMemId = add_hash_workmem(root->glob);
+
return plan;
}
@@ -3532,6 +3588,9 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
plan->plan_width = 0; /* meaningless */
plan->parallel_aware = false;
plan->parallel_safe = ipath->path.parallel_safe;
+
+ plan->workmem_id = add_workmem(root->glob);
+
/* Extract original index clauses, actual index quals, relevant ECs */
subquals = NIL;
subindexquals = NIL;
@@ -3839,6 +3898,8 @@ create_functionscan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
+ scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+
return scan_plan;
}
@@ -3882,6 +3943,8 @@ create_tablefuncscan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
+ scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+
return scan_plan;
}
@@ -4020,6 +4083,8 @@ create_ctescan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
+ scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+
return scan_plan;
}
@@ -4722,6 +4787,8 @@ create_mergejoin_plan(PlannerInfo *root,
copy_plan_costsize(matplan, inner_plan);
matplan->total_cost += cpu_operator_cost * matplan->plan_rows;
+ matplan->workmem_id = add_workmem(root->glob);
+
inner_plan = matplan;
}
@@ -5067,6 +5134,9 @@ create_hashjoin_plan(PlannerInfo *root,
copy_generic_path_info(&join_plan->join.plan, &best_path->jpath.path);
+ /* Assign workmem to the Hash subnode, not its parent HashJoin node. */
+ hash_plan->plan.workmem_id = add_hash_workmem(root->glob);
+
return join_plan;
}
@@ -5619,6 +5689,8 @@ label_sort_with_costsize(PlannerInfo *root, Sort *plan, double limit_tuples)
plan->plan.plan_width = lefttree->plan_width;
plan->plan.parallel_aware = false;
plan->plan.parallel_safe = lefttree->parallel_safe;
+
+ plan->plan.workmem_id = add_workmem(root->glob);
}
/*
@@ -5650,6 +5722,8 @@ label_incrementalsort_with_costsize(PlannerInfo *root, IncrementalSort *plan,
plan->sort.plan.plan_width = lefttree->plan_width;
plan->sort.plan.parallel_aware = false;
plan->sort.plan.parallel_safe = lefttree->parallel_safe;
+
+ plan->sort.plan.workmem_id = add_workmem(root->glob);
}
/*
@@ -6701,14 +6775,14 @@ make_material(Plan *lefttree)
/*
* materialize_finished_plan: stick a Material node atop a completed plan
- *
+ *r/
* There are a couple of places where we want to attach a Material node
* after completion of create_plan(), without any MaterialPath path.
* Those places should probably be refactored someday to do this on the
* Path representation, but it's not worth the trouble yet.
*/
Plan *
-materialize_finished_plan(Plan *subplan)
+materialize_finished_plan(PlannerGlobal *glob, Plan *subplan)
{
Plan *matplan;
Path matpath; /* dummy for result of cost_material */
@@ -6747,6 +6821,8 @@ materialize_finished_plan(Plan *subplan)
matplan->parallel_aware = false;
matplan->parallel_safe = subplan->parallel_safe;
+ matplan->workmem_id = add_workmem(glob);
+
return matplan;
}
@@ -7512,3 +7588,41 @@ is_projection_capable_plan(Plan *plan)
}
return true;
}
+
+static int
+add_workmem_internal(PlannerGlobal *glob, WorkMemCategory category)
+{
+ glob->workMemCategories = lappend_int(glob->workMemCategories, category);
+ /* the executor will fill this in later: */
+ glob->workMemLimits = lappend_int(glob->workMemLimits, 0);
+
+ Assert(list_length(glob->workMemCategories) ==
+ list_length(glob->workMemLimits));
+
+ return list_length(glob->workMemCategories);
+}
+
+/*
+ * add_workmem
+ * Add (non-hash) workmem info to the glob's lists
+ *
+ * This data structure will have its working-memory limit set to work_mem.
+ */
+int
+add_workmem(PlannerGlobal *glob)
+{
+ return add_workmem_internal(glob, WORKMEM_NORMAL);
+}
+
+/*
+ * add_hash_workmem
+ * Add hash workmem info to the glob's lists
+ *
+ * This data structure will have its working-memory limit set to work_mem *
+ * hash_mem_multiplier.
+ */
+int
+add_hash_workmem(PlannerGlobal *glob)
+{
+ return add_workmem_internal(glob, WORKMEM_HASH);
+}
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index d59d6e4c6a0..a431808be96 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -447,7 +447,7 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
if (cursorOptions & CURSOR_OPT_SCROLL)
{
if (!ExecSupportsBackwardScan(top_plan))
- top_plan = materialize_finished_plan(top_plan);
+ top_plan = materialize_finished_plan(glob, top_plan);
}
/*
@@ -584,6 +584,9 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
result->stmt_location = parse->stmt_location;
result->stmt_len = parse->stmt_len;
+ result->workMemCategories = glob->workMemCategories;
+ result->workMemLimits = glob->workMemLimits;
+
result->jitFlags = PGJIT_NONE;
if (jit_enabled && jit_above_cost >= 0 &&
top_plan->total_cost > jit_above_cost)
diff --git a/src/backend/optimizer/plan/subselect.c b/src/backend/optimizer/plan/subselect.c
index d71ed958e31..8bc99aa8bc1 100644
--- a/src/backend/optimizer/plan/subselect.c
+++ b/src/backend/optimizer/plan/subselect.c
@@ -533,7 +533,7 @@ build_subplan(PlannerInfo *root, Plan *plan, Path *path,
*/
else if (splan->parParam == NIL && enable_material &&
!ExecMaterializesOutput(nodeTag(plan)))
- plan = materialize_finished_plan(plan);
+ plan = materialize_finished_plan(root->glob, plan);
result = (Node *) splan;
isInitPlan = false;
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index a71502efeed..6008e3bc63c 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,
@@ -559,6 +560,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);
@@ -796,4 +798,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 6c4891bbaeb..fd8ed34178f 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -329,7 +329,8 @@ 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,
+ int used_bits,
+ Size hash_mem_limit, Size *mem_limit,
uint64 *ngroups_limit, int *num_partitions);
/* parallel instrumentation support */
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
index 3c1a09415aa..e4e9e0d1de1 100644
--- a/src/include/executor/nodeHash.h
+++ b/src/include/executor/nodeHash.h
@@ -59,7 +59,8 @@ extern void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
bool try_combined_hash_mem,
int parallel_workers,
- size_t *space_allowed,
+ size_t worker_space_allowed,
+ size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
int *num_skew_mcvs);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index e107d6e5f81..d543011d92a 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1276,6 +1276,19 @@ typedef struct PlanState
((PlanState *)(node))->instrument->nfiltered2 += (delta); \
} while(0)
+/* macros for fetching the workmem info associated with a PlanState */
+#define workMemFieldFromId(node, field, id) \
+ (list_nth_int(((PlanState *)(node))->state->es_plannedstmt->field, \
+ (id) - 1))
+#define workMemField(node, field) \
+ (workMemFieldFromId((node), field, ((PlanState *)(node))->plan->workmem_id))
+
+/* workmem limit: */
+#define workMemLimitFromId(node, id) \
+ (workMemFieldFromId(node, workMemLimits, id))
+#define workMemLimit(node) \
+ (workMemField(node, workMemLimits))
+
/*
* EPQState is state for executing an EvalPlanQual recheck on a candidate
* tuples e.g. in ModifyTable or LockRows.
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index ad2726f026f..181437ac933 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -182,6 +182,17 @@ typedef struct PlannerGlobal
/* hash table for NOT NULL attnums of relations */
struct HTAB *rel_notnullatts_hash pg_node_attr(read_write_ignore);
+
+ /*
+ * Working-memory info, for Plan and SubPlans. Any Plan or SubPlan that
+ * needs working memory for a data structure maintains a "workmem_id"
+ * index into the following lists (all kept in sync).
+ */
+
+ /* - IntList (of WorkMemCategory): is this a Hash or "normal" limit? */
+ List *workMemCategories;
+ /* - IntList: limit (in KB), after which data structure must spill */
+ List *workMemLimits;
} PlannerGlobal;
/* macro for fetching the Plan associated with a SubPlan node */
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 29d7732d6a0..ba8fdc2e6db 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -154,13 +154,23 @@ typedef struct PlannedStmt
ParseLoc stmt_location;
/* length in bytes; 0 means "rest of string" */
ParseLoc stmt_len;
+
+ /*
+ * Working-memory info, for Plan and SubPlans. Any Plan or SubPlan that
+ * needs working memory for a data structure maintains a "workmem_id"
+ * index into the following lists (all kept in sync).
+ */
+
+ /* - IntList (of WorkMemCategory): is this a Hash or "normal" limit? */
+ List *workMemCategories;
+ /* - IntList: limit (in KB), after which data structure must spill */
+ List *workMemLimits;
} PlannedStmt;
/* macro for fetching the Plan associated with a SubPlan node */
#define exec_subplan_get_plan(plannedstmt, subplan) \
((Plan *) list_nth((plannedstmt)->subplans, (subplan)->plan_id - 1))
-
/* ----------------
* Plan node
*
@@ -216,6 +226,8 @@ typedef struct Plan
*/
/* unique across entire final plan tree */
int plan_node_id;
+ /* 1-based id of workMem to use, or else zero */
+ int workmem_id;
/* target list to be computed at this node */
List *targetlist;
/* implicitly-ANDed qual conditions */
@@ -447,6 +459,9 @@ typedef struct RecursiveUnion
/* estimated number of groups in input */
long numGroups;
+
+ /* 1-based id of workMem to use for hash table, or else zero */
+ int hashWorkMemId;
} RecursiveUnion;
/* ----------------
@@ -1176,6 +1191,9 @@ typedef struct Agg
Oid *grpOperators pg_node_attr(array_size(numCols));
Oid *grpCollations pg_node_attr(array_size(numCols));
+ /* 1-based id of workMem to use to sort inputs, or else zero */
+ int sortWorkMemId;
+
/* estimated number of groups in input */
long numGroups;
@@ -1792,4 +1810,11 @@ typedef enum MonotonicFunction
MONOTONICFUNC_BOTH = MONOTONICFUNC_INCREASING | MONOTONICFUNC_DECREASING,
} MonotonicFunction;
+/* different data structures get different working-memory limits*/
+typedef enum WorkMemCategory
+{
+ WORKMEM_NORMAL, /* gets work_mem */
+ WORKMEM_HASH, /* gets hash_mem_multiplier * work_mem */
+} WorkMemCategory;
+
#endif /* PLANNODES_H */
diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h
index 6dfca3cb35b..c55b3cb356e 100644
--- a/src/include/nodes/primnodes.h
+++ b/src/include/nodes/primnodes.h
@@ -1111,6 +1111,9 @@ typedef struct SubPlan
/* Estimated execution costs: */
Cost startup_cost; /* one-time setup cost */
Cost per_call_cost; /* cost for each subplan evaluation */
+ /* 1-based id of workMem to use, or else zero: */
+ int hashtab_workmem_id; /* for hash table */
+ int hashnul_workmem_id; /* for NULLs hash table */
} SubPlan;
/*
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index 9d3debcab28..8436136026b 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -46,9 +46,11 @@ extern ForeignScan *make_foreignscan(List *qptlist, List *qpqual,
Plan *outer_plan);
extern Plan *change_plan_targetlist(Plan *subplan, List *tlist,
bool tlist_parallel_safe);
-extern Plan *materialize_finished_plan(Plan *subplan);
+extern Plan *materialize_finished_plan(PlannerGlobal *glob, Plan *subplan);
extern bool is_projection_capable_path(Path *path);
extern bool is_projection_capable_plan(Plan *plan);
+extern int add_workmem(PlannerGlobal *glob);
+extern int add_hash_workmem(PlannerGlobal *glob);
/* External use of these functions is deprecated: */
extern Sort *make_sort_from_sortclauses(List *sortcls, Plan *lefttree);
--
2.39.5
0002-Add-workmem-estimates-to-Path-node-and-PlannedStmt.patch (text/x-diff, 60.2K) download
From b4af98013dddfa8f56c14b3d6a0b667b9faefece Mon Sep 17 00:00:00 2001
From: James Hunter <james.hunter.pg@gmail.com>
Date: Tue, 4 Mar 2025 23:03:19 +0000
Subject: [PATCH 2/4] Add "workmem" estimates to Path node and PlannedStmt
To allow for future optimizers to make decisions at Path time, this commit
aggregates the Path's total working memory onto the Path's "workmem" field,
normalized to a minimum of 64 KB and rounded up to the next whole KB.
To allow future hooks to override ExecAssignWorkMem(), this commit then
breaks that total working memory into per-data structure working memory,
and stores it, next to the workMemLimit, on the PlannedStmt.
---
src/backend/executor/execParallel.c | 2 +
src/backend/executor/nodeHash.c | 32 +-
src/backend/nodes/tidbitmap.c | 18 ++
src/backend/optimizer/path/costsize.c | 406 ++++++++++++++++++++++--
src/backend/optimizer/plan/createplan.c | 267 +++++++++++++---
src/backend/optimizer/plan/planner.c | 2 +
src/backend/optimizer/prep/prepagg.c | 12 +
src/backend/optimizer/util/pathnode.c | 53 +++-
src/include/executor/nodeHash.h | 3 +-
src/include/nodes/execnodes.h | 12 +
src/include/nodes/pathnodes.h | 10 +-
src/include/nodes/plannodes.h | 7 +-
src/include/nodes/tidbitmap.h | 1 +
src/include/optimizer/cost.h | 13 +-
src/include/optimizer/planmain.h | 3 +-
15 files changed, 762 insertions(+), 79 deletions(-)
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index a8cb631963e..5c90a29d7d1 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -217,6 +217,8 @@ ExecSerializePlan(Plan *plan, EState *estate)
pstmt->stmt_location = -1;
pstmt->stmt_len = -1;
pstmt->workMemCategories = estate->es_plannedstmt->workMemCategories;
+ pstmt->workMemEstimates = estate->es_plannedstmt->workMemEstimates;
+ pstmt->workMemCounts = estate->es_plannedstmt->workMemCounts;
pstmt->workMemLimits = estate->es_plannedstmt->workMemLimits;
/* Return serialized copy of our dummy PlannedStmt. */
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index bb9af08dc5d..7d09ac8b5a3 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -35,6 +35,7 @@
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
+#include "optimizer/cost.h"
#include "port/pg_bitutils.h"
#include "utils/dynahash.h"
#include "utils/lsyscache.h"
@@ -453,6 +454,7 @@ ExecHashTableCreate(HashState *state)
int nbuckets;
int nbatch;
double rows;
+ int workmem; /* ignored */
int num_skew_mcvs;
int log2_nbuckets;
MemoryContext oldcxt;
@@ -482,7 +484,7 @@ ExecHashTableCreate(HashState *state)
state->parallel_state->nparticipants - 1 : 0,
worker_space_allowed,
&space_allowed,
- &nbuckets, &nbatch, &num_skew_mcvs);
+ &nbuckets, &nbatch, &num_skew_mcvs, &workmem);
/* nbuckets must be a power of 2 */
log2_nbuckets = my_log2(nbuckets);
@@ -668,7 +670,8 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
- int *num_skew_mcvs)
+ int *num_skew_mcvs,
+ int *workmem)
{
int tupsize;
double inner_rel_bytes;
@@ -769,6 +772,27 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
*num_skew_mcvs = 0;
/*
+ * Set "workmem" to the amount of memory needed to hold the inner rel in a
+ * single batch. So this calculation doesn't care about "max_pointers".
+ */
+ dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
+ nbuckets = (int) dbuckets;
+ /* don't let nbuckets be really small, though ... */
+ nbuckets = Max(nbuckets, 1024);
+ /* ... and force it to be a power of 2. */
+ nbuckets = pg_nextpower2_32(nbuckets);
+ bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
+
+ /* Don't forget the 2% overhead reserved for skew buckets! */
+ *workmem = useskew ?
+ normalize_work_bytes((inner_rel_bytes + bucket_bytes) *
+ 100.0 / (100.0 - SKEW_HASH_MEM_PERCENT)) :
+ normalize_work_bytes(inner_rel_bytes + bucket_bytes);
+
+ /*
+ * Now redo the nbuckets and bucket_bytes calculations, taking memory
+ * limits into account.
+ *
* Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
* memory is filled, assuming a single batch; but limit the value so that
* the pointer arrays we'll try to allocate do not exceed hash_table_bytes
@@ -799,6 +823,7 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
* the required bucket headers, we will need multiple batches.
*/
bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
+
if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
{
/* We'll need multiple batches */
@@ -819,7 +844,8 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
total_space_allowed,
numbuckets,
numbatches,
- num_skew_mcvs);
+ num_skew_mcvs,
+ workmem);
return;
}
diff --git a/src/backend/nodes/tidbitmap.c b/src/backend/nodes/tidbitmap.c
index 41031aa8f2f..425333b0218 100644
--- a/src/backend/nodes/tidbitmap.c
+++ b/src/backend/nodes/tidbitmap.c
@@ -1560,6 +1560,24 @@ tbm_calculate_entries(Size maxbytes)
return (int) nbuckets;
}
+/*
+ * tbm_calculate_bytes
+ *
+ * Estimate number of bytes needed to store maxentries hashtable entries.
+ *
+ * This function is the inverse of tbm_calculate_entries(), and is used to
+ * estimate a work_mem limit, based on cardinality.
+ */
+double
+tbm_calculate_bytes(double maxentries)
+{
+ maxentries = Min(maxentries, INT_MAX - 1); /* safety limit */
+ maxentries = Max(maxentries, 16); /* sanity limit */
+
+ return maxentries * (sizeof(PagetableEntry) + sizeof(Pointer) +
+ sizeof(Pointer));
+}
+
/*
* Create a shared or private bitmap iterator and start iteration.
*
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 353f51fdff2..27daa1966c2 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -105,6 +105,7 @@
#include "optimizer/planmain.h"
#include "optimizer/restrictinfo.h"
#include "parser/parsetree.h"
+#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/selfuncs.h"
#include "utils/spccache.h"
@@ -201,9 +202,14 @@ static Cost append_nonpartial_cost(List *subpaths, int numpaths,
int parallel_workers);
static void set_rel_width(PlannerInfo *root, RelOptInfo *rel);
static int32 get_expr_width(PlannerInfo *root, const Node *expr);
-static double relation_byte_size(double tuples, int width);
static double page_size(double tuples, int width);
static double get_parallel_divisor(Path *path);
+static void compute_sort_output_sizes(double input_tuples, int input_width,
+ double limit_tuples,
+ double *output_tuples,
+ double *output_bytes);
+static double compute_bitmap_workmem(RelOptInfo *baserel, Path *bitmapqual,
+ Cardinality max_ancestor_rows);
/*
@@ -1113,6 +1119,18 @@ cost_bitmap_heap_scan(Path *path, PlannerInfo *root, RelOptInfo *baserel,
path->disabled_nodes = enable_bitmapscan ? 0 : 1;
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.
+ *
+ * (When we convert this path into a BitmapHeapScan plan, we'll break this
+ * overall estimate down into per-node estimates, just as we do for
+ * AggPaths.)
+ */
+ path->workmem = compute_bitmap_workmem(baserel, bitmapqual,
+ 0.0 /* max_ancestor_rows */ );
}
/*
@@ -1588,6 +1606,16 @@ cost_functionscan(Path *path, PlannerInfo *root,
path->disabled_nodes = 0;
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+
+ /*
+ * Per "XXX" comment above, this workmem estimate is likely to be wrong,
+ * because the "rows" estimate is pretty phony. Report the estimate
+ * anyway, for completeness. (This is at least better than saying it won't
+ * use *any* working memory.)
+ */
+ path->workmem = list_length(rte->functions) *
+ normalize_work_bytes(relation_byte_size(path->rows,
+ path->pathtarget->width));
}
/*
@@ -1645,6 +1673,16 @@ cost_tablefuncscan(Path *path, PlannerInfo *root,
path->disabled_nodes = 0;
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+
+ /*
+ * Per "XXX" comment above, this workmem estimate is likely to be wrong,
+ * because the "rows" estimate is pretty phony. Report the estimate
+ * anyway, for completeness. (This is at least better than saying it won't
+ * use *any* working memory.)
+ */
+ path->workmem =
+ normalize_work_bytes(relation_byte_size(path->rows,
+ path->pathtarget->width));
}
/*
@@ -1741,6 +1779,9 @@ cost_ctescan(Path *path, PlannerInfo *root,
path->disabled_nodes = 0;
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+ path->workmem =
+ normalize_work_bytes(relation_byte_size(path->rows,
+ path->pathtarget->width));
}
/*
@@ -1824,7 +1865,7 @@ cost_resultscan(Path *path, PlannerInfo *root,
* We are given Paths for the nonrecursive and recursive terms.
*/
void
-cost_recursive_union(Path *runion, Path *nrterm, Path *rterm)
+cost_recursive_union(RecursiveUnionPath *runion, Path *nrterm, Path *rterm)
{
Cost startup_cost;
Cost total_cost;
@@ -1851,12 +1892,37 @@ cost_recursive_union(Path *runion, Path *nrterm, Path *rterm)
*/
total_cost += cpu_tuple_cost * total_rows;
- runion->disabled_nodes = nrterm->disabled_nodes + rterm->disabled_nodes;
- runion->startup_cost = startup_cost;
- runion->total_cost = total_cost;
- runion->rows = total_rows;
- runion->pathtarget->width = Max(nrterm->pathtarget->width,
- rterm->pathtarget->width);
+ runion->path.disabled_nodes = nrterm->disabled_nodes + rterm->disabled_nodes;
+ runion->path.startup_cost = startup_cost;
+ runion->path.total_cost = total_cost;
+ runion->path.rows = total_rows;
+ runion->path.pathtarget->width = Max(nrterm->pathtarget->width,
+ rterm->pathtarget->width);
+
+ /*
+ * Include memory for working and intermediate tables. Since we'll
+ * repeatedly swap the two tables, use 2x whichever is larger as our
+ * estimate.
+ */
+ runion->path.workmem =
+ normalize_work_bytes(
+ Max(relation_byte_size(nrterm->rows,
+ nrterm->pathtarget->width),
+ relation_byte_size(rterm->rows,
+ rterm->pathtarget->width))
+ * 2);
+
+ if (list_length(runion->distinctList) > 0)
+ {
+ /* Also include memory for hash table. */
+ Size hashentrysize;
+
+ hashentrysize = MAXALIGN(runion->path.pathtarget->width) +
+ MAXALIGN(SizeofMinimalTupleHeader);
+
+ runion->path.workmem +=
+ normalize_work_bytes(runion->numGroups * hashentrysize);
+ }
}
/*
@@ -1896,7 +1962,7 @@ cost_recursive_union(Path *runion, Path *nrterm, Path *rterm)
* 'limit_tuples' is the bound on the number of output tuples; -1 if no bound
*/
static void
-cost_tuplesort(Cost *startup_cost, Cost *run_cost,
+cost_tuplesort(Cost *startup_cost, Cost *run_cost, Cost *nbytes,
double tuples, int width,
Cost comparison_cost, int sort_mem,
double limit_tuples)
@@ -1916,17 +1982,8 @@ cost_tuplesort(Cost *startup_cost, Cost *run_cost,
/* Include the default cost-per-comparison */
comparison_cost += 2.0 * cpu_operator_cost;
- /* Do we have a useful LIMIT? */
- if (limit_tuples > 0 && limit_tuples < tuples)
- {
- output_tuples = limit_tuples;
- output_bytes = relation_byte_size(output_tuples, width);
- }
- else
- {
- output_tuples = tuples;
- output_bytes = input_bytes;
- }
+ compute_sort_output_sizes(tuples, width, limit_tuples,
+ &output_tuples, &output_bytes);
if (output_bytes > sort_mem_bytes)
{
@@ -1983,6 +2040,7 @@ cost_tuplesort(Cost *startup_cost, Cost *run_cost,
* counting the LIMIT otherwise.
*/
*run_cost = cpu_operator_cost * tuples;
+ *nbytes = output_bytes;
}
/*
@@ -2012,6 +2070,7 @@ cost_incremental_sort(Path *path,
input_groups;
Cost group_startup_cost,
group_run_cost,
+ group_nbytes,
group_input_run_cost;
List *presortedExprs = NIL;
ListCell *l;
@@ -2086,7 +2145,7 @@ cost_incremental_sort(Path *path,
* Estimate the average cost of sorting of one group where presorted keys
* are equal.
*/
- cost_tuplesort(&group_startup_cost, &group_run_cost,
+ cost_tuplesort(&group_startup_cost, &group_run_cost, &group_nbytes,
group_tuples, width, comparison_cost, sort_mem,
limit_tuples);
@@ -2127,6 +2186,14 @@ cost_incremental_sort(Path *path,
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+
+ /*
+ * Incremental sort switches between two Tuplesortstates: one that sorts
+ * all columns ("full"), and that sorts only suffix columns ("prefix").
+ * We'll assume they're both around the same size: large enough to hold
+ * one sort group.
+ */
+ path->workmem = normalize_work_bytes(group_nbytes * 2.0);
}
/*
@@ -2151,8 +2218,9 @@ cost_sort(Path *path, PlannerInfo *root,
{
Cost startup_cost;
Cost run_cost;
+ Cost nbytes;
- cost_tuplesort(&startup_cost, &run_cost,
+ cost_tuplesort(&startup_cost, &run_cost, &nbytes,
tuples, width,
comparison_cost, sort_mem,
limit_tuples);
@@ -2163,6 +2231,7 @@ cost_sort(Path *path, PlannerInfo *root,
path->disabled_nodes = input_disabled_nodes + (enable_sort ? 0 : 1);
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+ path->workmem = normalize_work_bytes(nbytes);
}
/*
@@ -2549,6 +2618,7 @@ cost_material(Path *path,
path->disabled_nodes = input_disabled_nodes + (enable_material ? 0 : 1);
path->startup_cost = startup_cost;
path->total_cost = startup_cost + run_cost;
+ path->workmem = normalize_work_bytes(nbytes);
}
/*
@@ -2622,6 +2692,9 @@ cost_memoize_rescan(PlannerInfo *root, MemoizePath *mpath,
/* Remember the ndistinct estimate for EXPLAIN */
mpath->est_unique_keys = ndistinct;
+ /* How much working memory would we need, to store every distinct tuple? */
+ mpath->path.workmem = normalize_work_bytes(ndistinct * est_entry_bytes);
+
/*
* Since we've already estimated the maximum number of entries we can
* store at once and know the estimated number of distinct values we'll be
@@ -2899,6 +2972,19 @@ cost_agg(Path *path, PlannerInfo *root,
path->disabled_nodes = disabled_nodes;
path->startup_cost = startup_cost;
path->total_cost = total_cost;
+
+ /* Include memory needed to produce output. */
+ path->workmem =
+ compute_agg_output_workmem(root, aggstrategy, numGroups,
+ aggcosts->transitionSpace, input_tuples,
+ input_width, false /* cost_sort */ );
+
+ /* Also include memory needed to sort inputs (if needed): */
+ if (aggcosts->numSortBuffers > 0)
+ {
+ path->workmem += (double) aggcosts->numSortBuffers *
+ compute_agg_input_workmem(input_tuples, input_width);
+ }
}
/*
@@ -3133,7 +3219,7 @@ cost_windowagg(Path *path, PlannerInfo *root,
List *windowFuncs, WindowClause *winclause,
int input_disabled_nodes,
Cost input_startup_cost, Cost input_total_cost,
- double input_tuples)
+ double input_tuples, int width)
{
Cost startup_cost;
Cost total_cost;
@@ -3215,6 +3301,11 @@ cost_windowagg(Path *path, PlannerInfo *root,
if (startup_tuples > 1.0)
path->startup_cost += (total_cost - startup_cost) / input_tuples *
(startup_tuples - 1.0);
+
+
+ /* We need to store a window of size "startup_tuples", in a Tuplestore. */
+ path->workmem =
+ normalize_work_bytes(relation_byte_size(startup_tuples, width));
}
/*
@@ -3369,6 +3460,7 @@ initial_cost_nestloop(PlannerInfo *root, JoinCostWorkspace *workspace,
workspace->total_cost = startup_cost + run_cost;
/* Save private data for final_cost_nestloop */
workspace->run_cost = run_cost;
+ workspace->workmem = 0;
}
/*
@@ -3833,6 +3925,14 @@ initial_cost_mergejoin(PlannerInfo *root, JoinCostWorkspace *workspace,
workspace->total_cost = startup_cost + run_cost + inner_run_cost;
/* Save private data for final_cost_mergejoin */
workspace->run_cost = run_cost;
+
+ /*
+ * By itself, Merge Join requires no working memory. If it adds one or
+ * more Sort or Material nodes, we'll track their working memory when we
+ * create them, inside createplan.c.
+ */
+ workspace->workmem = 0;
+
workspace->inner_run_cost = inner_run_cost;
workspace->outer_rows = outer_rows;
workspace->inner_rows = inner_rows;
@@ -4204,6 +4304,7 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
double outer_path_rows = outer_path->rows;
double inner_path_rows = inner_path->rows;
double inner_path_rows_total = inner_path_rows;
+ int workmem;
int num_hashclauses = list_length(hashclauses);
int numbuckets;
int numbatches;
@@ -4262,7 +4363,8 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
&space_allowed,
&numbuckets,
&numbatches,
- &num_skew_mcvs);
+ &num_skew_mcvs,
+ &workmem);
/*
* If inner relation is too big then we will need to "batch" the join,
@@ -4293,6 +4395,7 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
workspace->numbuckets = numbuckets;
workspace->numbatches = numbatches;
workspace->inner_rows_total = inner_path_rows_total;
+ workspace->workmem = workmem;
}
/*
@@ -4301,8 +4404,8 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
*
* Note: the numbatches estimate is also saved into 'path' for use later
*
- * 'path' is already filled in except for the rows and cost fields and
- * num_batches
+ * 'path' is already filled in except for the rows and cost fields,
+ * num_batches, and workmem
* 'workspace' is the result from initial_cost_hashjoin
* 'extra' contains miscellaneous information about the join
*/
@@ -4319,6 +4422,7 @@ final_cost_hashjoin(PlannerInfo *root, HashPath *path,
List *hashclauses = path->path_hashclauses;
Cost startup_cost = workspace->startup_cost;
Cost run_cost = workspace->run_cost;
+ int workmem = workspace->workmem;
int numbuckets = workspace->numbuckets;
int numbatches = workspace->numbatches;
Cost cpu_per_tuple;
@@ -4555,6 +4659,7 @@ final_cost_hashjoin(PlannerInfo *root, HashPath *path,
path->jpath.path.startup_cost = startup_cost;
path->jpath.path.total_cost = startup_cost + run_cost;
+ path->jpath.path.workmem = workmem;
}
@@ -4577,6 +4682,9 @@ cost_subplan(PlannerInfo *root, SubPlan *subplan, Plan *plan)
if (subplan->useHashTable)
{
+ long nbuckets;
+ Size hashentrysize;
+
/*
* If we are using a hash table for the subquery outputs, then the
* cost of evaluating the query is a one-time cost. We charge one
@@ -4588,13 +4696,37 @@ cost_subplan(PlannerInfo *root, SubPlan *subplan, Plan *plan)
/*
* Working memory needed for the hashtable (and hashnulls, if needed).
+ * The logic below MUST match the logic in buildSubPlanHash() and
+ * ExecInitSubPlan().
*/
- subplan->hashtab_workmem_id = add_hash_workmem(root->glob);
+ nbuckets = clamp_cardinality_to_long(plan->plan_rows);
+ if (nbuckets < 1)
+ nbuckets = 1;
+
+ hashentrysize = MAXALIGN(plan->plan_width) +
+ MAXALIGN(SizeofMinimalTupleHeader);
+
+ subplan->hashtab_workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_bytes((double) nbuckets *
+ hashentrysize));
if (!subplan->unknownEqFalse)
{
/* Also needs a hashnulls table. */
- subplan->hashnul_workmem_id = add_hash_workmem(root->glob);
+ if (IsA(subplan->testexpr, OpExpr))
+ nbuckets = 1; /* there can be only one entry */
+ else
+ {
+ nbuckets /= 16;
+ if (nbuckets < 1)
+ nbuckets = 1;
+ }
+
+ subplan->hashnul_workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_bytes((double) nbuckets *
+ hashentrysize));
}
/*
@@ -6481,7 +6613,7 @@ get_expr_width(PlannerInfo *root, const Node *expr)
* Estimate the storage space in bytes for a given number of tuples
* of a given width (size in bytes).
*/
-static double
+double
relation_byte_size(double tuples, int width)
{
return tuples * (MAXALIGN(width) + MAXALIGN(SizeofHeapTupleHeader));
@@ -6660,3 +6792,219 @@ compute_gather_rows(Path *path)
return clamp_row_est(path->rows * get_parallel_divisor(path));
}
+
+/*
+ * compute_sort_output_sizes
+ * Estimate amount of memory and rows needed to hold a Sort operator's output
+ */
+static void
+compute_sort_output_sizes(double input_tuples, int input_width,
+ double limit_tuples,
+ double *output_tuples, double *output_bytes)
+{
+ /*
+ * We want to be sure the cost of a sort is never estimated as zero, even
+ * if passed-in tuple count is zero. Besides, mustn't do log(0)...
+ */
+ if (input_tuples < 2.0)
+ input_tuples = 2.0;
+
+ /* Do we have a useful LIMIT? */
+ if (limit_tuples > 0 && limit_tuples < input_tuples)
+ *output_tuples = limit_tuples;
+ else
+ *output_tuples = input_tuples;
+
+ *output_bytes = relation_byte_size(*output_tuples, input_width);
+}
+
+/*
+ * compute_agg_input_workmem
+ * Estimate memory (in KB) needed to hold a sort buffer for aggregate's input
+ *
+ * Some aggregates involve DISTINCT or ORDER BY, so they need to sort their
+ * input, before they can process it. We need one sort buffer per such
+ * aggregate, and this function returns that sort buffer's (estimated) size (in
+ * KB).
+ */
+int
+compute_agg_input_workmem(double input_tuples, double input_width)
+{
+ double output_tuples; /* ignored */
+ double output_bytes;
+
+ /* Account for size of one buffer needed to sort the input. */
+ compute_sort_output_sizes(input_tuples, input_width,
+ 0.0 /* limit_tuples */ ,
+ &output_tuples, &output_bytes);
+ return normalize_work_bytes(output_bytes);
+}
+
+/*
+ * compute_agg_output_workmem
+ * Estimate amount of memory needed (in KB) to hold an aggregate's output
+ *
+ * In a Hash aggregate, we need space for the hash table that holds the
+ * aggregated data.
+ *
+ * Sort aggregates require output space only if they are part of a Grouping
+ * Sets chain: the first aggregate writes to its "sort_out" buffer, which the
+ * second aggregate uses as its "sort_in" buffer, and sorts.
+ *
+ * In the latter case, the "Path" code already costs the sort by calling
+ * cost_sort(), so it passes "cost_sort = false" to this function, to avoid
+ * double-counting.
+ */
+int
+compute_agg_output_workmem(PlannerInfo *root, AggStrategy aggstrategy,
+ double numGroups, uint64 transitionSpace,
+ double input_tuples, double input_width,
+ bool cost_sort)
+{
+ /* Account for size of hash table to hold the output. */
+ if (aggstrategy == AGG_HASHED || aggstrategy == AGG_MIXED)
+ {
+ double hashentrysize;
+
+ hashentrysize = hash_agg_entry_size(list_length(root->aggtransinfos),
+ input_width, transitionSpace);
+ return normalize_work_bytes(numGroups * hashentrysize);
+ }
+
+ /* Account for the size of the "sort_out" buffer. */
+ if (cost_sort && aggstrategy == AGG_SORTED)
+ {
+ double output_tuples; /* ignored */
+ double output_bytes;
+
+ Assert(aggstrategy == AGG_SORTED);
+
+ compute_sort_output_sizes(numGroups, input_width,
+ 0.0 /* limit_tuples */ ,
+ &output_tuples, &output_bytes);
+ return normalize_work_bytes(output_bytes);
+ }
+
+ return 0;
+}
+
+/*
+ * compute_bitmap_workmem
+ * Estimate total working memory (in KB) needed by bitmapqual
+ *
+ * Although we don't fill in the workmem_est or rows fields on the bitmapqual's
+ * paths, we fill them in on the owning BitmapHeapPath. This function estimates
+ * the total work_mem needed by all BitmapOrPaths and IndexPaths inside
+ * bitmapqual.
+ */
+static double
+compute_bitmap_workmem(RelOptInfo *baserel, Path *bitmapqual,
+ Cardinality max_ancestor_rows)
+{
+ double workmem = 0.0;
+ Cost cost; /* not used */
+ Selectivity selec;
+ Cardinality plan_rows;
+
+ /* How many rows will this node output? */
+ cost_bitmap_tree_node(bitmapqual, &cost, &selec);
+ plan_rows = clamp_row_est(selec * baserel->tuples);
+
+ /*
+ * At runtime, we'll reuse the left-most child's TID bitmap. Let that
+ * child that child know to request enough working memory to hold all its
+ * ancestors' results.
+ */
+ max_ancestor_rows = Max(max_ancestor_rows, plan_rows);
+
+ if (IsA(bitmapqual, BitmapAndPath))
+ {
+ BitmapAndPath *apath = (BitmapAndPath *) bitmapqual;
+ ListCell *l;
+
+ foreach(l, apath->bitmapquals)
+ {
+ workmem +=
+ compute_bitmap_workmem(baserel, (Path *) lfirst(l),
+ foreach_current_index(l) == 0 ?
+ max_ancestor_rows : 0.0);
+ }
+ }
+ else if (IsA(bitmapqual, BitmapOrPath))
+ {
+ BitmapOrPath *opath = (BitmapOrPath *) bitmapqual;
+ ListCell *l;
+
+ foreach(l, opath->bitmapquals)
+ {
+ workmem +=
+ compute_bitmap_workmem(baserel, (Path *) lfirst(l),
+ foreach_current_index(l) == 0 ?
+ max_ancestor_rows : 0.0);
+ }
+ }
+ else if (IsA(bitmapqual, IndexPath))
+ {
+ /* Working memory needed for 1 TID bitmap. */
+ workmem +=
+ normalize_work_bytes(tbm_calculate_bytes(max_ancestor_rows));
+ }
+
+ return workmem;
+}
+
+/*
+ * normalize_work_kb
+ * Convert a double, "KB" working-memory estimate to an int, "KB" value
+ *
+ * Normalizes non-zero input to a minimum of 64 (KB), rounding up to the
+ * nearest whole KB.
+ */
+int
+normalize_work_kb(double nkb)
+{
+ double workmem;
+
+ if (nkb == 0.0)
+ return 0; /* caller apparently doesn't need any workmem */
+
+ /*
+ * We'll assign working-memory to SQL operators in 1 KB increments, so
+ * round up to the next whole KB.
+ */
+ workmem = ceil(nkb);
+
+ /*
+ * Although some components can probably work with < 64 KB of working
+ * memory, PostgreSQL has imposed a hard minimum of 64 KB on the
+ * "work_mem" GUC, for a long time; so, by now, some components probably
+ * rely on this minimum, implicitly, and would fail if we tried to assign
+ * them < 64 KB.
+ *
+ * Perhaps this minimum can be relaxed, in the future; but memory sizes
+ * keep increasing, and right now the minimum of 64 KB = 1.6 percent of
+ * the default "work_mem" of 4 MB.
+ *
+ * So, even with this (overly?) cautious normalization, with the default
+ * GUC settings, we can still achieve a working-memory reduction of
+ * 64-to-1.
+ */
+ workmem = Max((double) 64, workmem);
+
+ /* And clamp to MAX_KILOBYTES. */
+ workmem = Min(workmem, (double) MAX_KILOBYTES);
+
+ return (int) workmem;
+}
+
+/*
+ * normalize_work_bytes
+ * Convert a double, "bytes" working-memory estimate to an int, "KB" value
+ *
+ * Same as above, but takes input in bytes rather than in KB.
+ */
+int
+normalize_work_bytes(double nbytes)
+{
+ return normalize_work_kb(nbytes / 1024.0);
+}
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 22834fe37f4..aba15d54fa1 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -130,6 +130,7 @@ static BitmapHeapScan *create_bitmap_scan_plan(PlannerInfo *root,
BitmapHeapPath *best_path,
List *tlist, List *scan_clauses);
static Plan *create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
+ Cardinality max_ancestor_rows,
List **qual, List **indexqual, List **indexECs);
static void bitmap_subplan_mark_shared(Plan *plan);
static TidScan *create_tidscan_plan(PlannerInfo *root, TidPath *best_path,
@@ -319,6 +320,8 @@ static ModifyTable *make_modifytable(PlannerInfo *root, Plan *subplan,
int epqParam);
static GatherMerge *create_gather_merge_plan(PlannerInfo *root,
GatherMergePath *best_path);
+static int add_workmem(PlannerGlobal *glob, int estimate);
+static int add_workmems(PlannerGlobal *glob, int estimate, int count);
/*
@@ -1706,7 +1709,8 @@ create_material_plan(PlannerInfo *root, MaterialPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(best_path->path.workmem));
return plan;
}
@@ -1763,7 +1767,9 @@ create_memoize_plan(PlannerInfo *root, MemoizePath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_hash_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_kb(best_path->path.workmem));
return plan;
}
@@ -1912,7 +1918,9 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags)
0,
subplan);
- plan->workmem_id = add_hash_workmem(root->glob);
+ plan->workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_kb(best_path->path.workmem));
}
else
{
@@ -2259,7 +2267,9 @@ create_sort_plan(PlannerInfo *root, SortPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_workmem(root->glob,
+ normalize_work_kb(best_path->path.workmem));
return plan;
}
@@ -2287,7 +2297,13 @@ create_incrementalsort_plan(PlannerInfo *root, IncrementalSortPath *best_path,
copy_generic_path_info(&plan->sort.plan, (Path *) best_path);
- plan->sort.plan.workmem_id = add_workmem(root->glob);
+ /*
+ * IncrementalSort creates two sort buffers, which the Path's "workmem"
+ * estimate combined into a single value. Split it into two now.
+ */
+ plan->sort.plan.workmem_id =
+ add_workmems(root->glob,
+ normalize_work_kb(best_path->spath.path.workmem / 2), 2);
return plan;
}
@@ -2400,11 +2416,32 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
+ /*
+ * Replace the AggPath's overall workmem estimate with finer-grained
+ * estimates.
+ */
if (plan->aggstrategy == AGG_HASHED)
- plan->plan.workmem_id = add_hash_workmem(root->glob);
+ {
+ int workmem =
+ compute_agg_output_workmem(root, AGG_HASHED,
+ plan->numGroups,
+ plan->transitionSpace,
+ subplan->plan_rows,
+ subplan->plan_width,
+ false /* cost_sort */ );
- /* Also include working memory needed to sort the input: */
- plan->sortWorkMemId = add_workmem(root->glob);
+ plan->plan.workmem_id = add_hash_workmem(root->glob, workmem);
+ }
+
+ /* Also include estimated memory needed to sort the input: */
+ if (best_path->numSortBuffers > 0)
+ {
+ int workmem = compute_agg_input_workmem(subplan->plan_rows,
+ subplan->plan_width);
+
+ plan->sortWorkMemId =
+ add_workmems(root->glob, workmem, best_path->numSortBuffers);
+ }
return plan;
}
@@ -2466,6 +2503,9 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
int maxref;
List *chain;
ListCell *lc;
+ int num_sort_aggs = 0;
+ int max_sort_agg_workmem = 0.0;
+ double sum_hash_agg_workmem = 0.0;
/* Shouldn't get here without grouping sets */
Assert(root->parse->groupingSets);
@@ -2527,6 +2567,8 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
Plan *sort_plan = NULL;
Agg *agg_plan;
AggStrategy strat;
+ bool cost_sort;
+ int workmem;
new_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
@@ -2577,6 +2619,33 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
first_sort_agg = agg_plan;
}
+ /*
+ * If we're an AGG_SORTED, but not the last, we need to cost
+ * working memory needed to produce our "sort_out" buffer.
+ */
+ cost_sort = foreach_current_index(lc) < list_length(rollups) - 1;
+
+ /* Estimated memory needed to hold the output: */
+ workmem =
+ compute_agg_output_workmem(root, agg_plan->aggstrategy,
+ agg_plan->numGroups,
+ agg_plan->transitionSpace,
+ subplan->plan_rows,
+ subplan->plan_width,
+ cost_sort);
+
+ if (agg_plan->aggstrategy == AGG_HASHED)
+ {
+ /* All Hash Grouping Sets share the same workmem limit. */
+ sum_hash_agg_workmem += workmem;
+ }
+ else if (agg_plan->aggstrategy == AGG_SORTED)
+ {
+ /* Every Sort Grouping Set gets its own workmem limit. */
+ max_sort_agg_workmem = Max(max_sort_agg_workmem, workmem);
+ ++num_sort_aggs;
+ }
+
chain = lappend(chain, agg_plan);
}
}
@@ -2588,6 +2657,8 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
RollupData *rollup = linitial(rollups);
AttrNumber *top_grpColIdx;
int numGroupCols;
+ bool cost_sort;
+ int workmem;
top_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
@@ -2610,6 +2681,27 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
/* Copy cost data from Path to Plan */
copy_generic_path_info(&plan->plan, &best_path->path);
+ /*
+ * If we're an AGG_SORTED, but not the last, we need to cost working
+ * memory needed to produce our "sort_out" buffer.
+ */
+ cost_sort = list_length(rollups) > 1;
+
+ /*
+ * Replace the overall workmem estimate that we copied from the Path
+ * with finer-grained estimates.
+ *
+ */
+
+ /* Estimated memory needed to hold the output: */
+ workmem =
+ compute_agg_output_workmem(root, plan->aggstrategy,
+ plan->numGroups,
+ plan->transitionSpace,
+ subplan->plan_rows,
+ subplan->plan_width,
+ cost_sort);
+
/*
* NOTE: We will place the workmem needed to sort the input (if any)
* on the first agg, the Hash workmem on the first Hash agg, and the
@@ -2618,20 +2710,37 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
if (plan->aggstrategy == AGG_HASHED || plan->aggstrategy == AGG_MIXED)
{
/* All Hash Grouping Sets share the same workmem limit. */
- plan->plan.workmem_id = add_hash_workmem(root->glob);
+ sum_hash_agg_workmem += workmem;
+ plan->plan.workmem_id = add_hash_workmem(root->glob,
+ sum_hash_agg_workmem);
}
else if (plan->aggstrategy == AGG_SORTED)
{
/* Every Sort Grouping Set gets its own workmem limit. */
+ max_sort_agg_workmem = Max(max_sort_agg_workmem, workmem);
+ ++num_sort_aggs;
+
first_sort_agg = plan;
}
/* Store the workmem limit, for all Sorts, on the first Sort. */
- if (first_sort_agg)
- first_sort_agg->plan.workmem_id = add_workmem(root->glob);
+ if (num_sort_aggs > 1)
+ {
+ first_sort_agg->plan.workmem_id =
+ add_workmems(root->glob, max_sort_agg_workmem,
+ num_sort_aggs > 2 ? 2 : 1);
+ }
/* Also include working memory needed to sort the input: */
- plan->sortWorkMemId = add_workmem(root->glob);
+ if (best_path->numSortBuffers > 0)
+ {
+ workmem = compute_agg_input_workmem(subplan->plan_rows,
+ subplan->plan_width);
+
+ plan->sortWorkMemId =
+ add_workmems(root->glob, workmem,
+ best_path->numSortBuffers * list_length(rollups));
+ }
}
return (Plan *) plan;
@@ -2796,7 +2905,8 @@ create_windowagg_plan(PlannerInfo *root, WindowAggPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(best_path->path.workmem));
return plan;
}
@@ -2838,7 +2948,9 @@ create_setop_plan(PlannerInfo *root, SetOpPath *best_path, int flags)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_hash_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_kb(best_path->path.workmem));
return plan;
}
@@ -2876,11 +2988,38 @@ create_recursiveunion_plan(PlannerInfo *root, RecursiveUnionPath *best_path)
copy_generic_path_info(&plan->plan, (Path *) best_path);
- plan->plan.workmem_id = add_workmem(root->glob);
+ /*
+ * Replace our overall "workmem" estimate with estimates at finer
+ * granularity.
+ */
+
+ /*
+ * Include memory for working and intermediate tables. Since we'll
+ * repeatedly swap the two tables, use the larger of the two as our
+ * working- memory estimate.
+ *
+ * NOTE: The Path's "workmem" estimate is for the whole Path, but the
+ * Plan's "workmem" estimates are *per data structure*. So, this value is
+ * half of the corresponding Path's value.
+ */
+ plan->plan.workmem_id =
+ add_workmems(root->glob,
+ normalize_work_bytes(Max(relation_byte_size(leftplan->plan_rows,
+ leftplan->plan_width),
+ relation_byte_size(rightplan->plan_rows,
+ rightplan->plan_width))),
+ 2);
/* Also include working memory for hash table. */
if (plan->numCols > 0)
- plan->hashWorkMemId = add_hash_workmem(root->glob);
+ {
+ Size entrysize =
+ sizeof(TupleHashEntryData) + plan->plan.plan_width;
+
+ plan->hashWorkMemId =
+ add_hash_workmem(root->glob,
+ normalize_work_bytes(plan->numGroups * entrysize));
+ }
return plan;
}
@@ -3322,6 +3461,7 @@ create_bitmap_scan_plan(PlannerInfo *root,
/* Process the bitmapqual tree into a Plan tree and qual lists */
bitmapqualplan = create_bitmap_subplan(root, best_path->bitmapqual,
+ 0.0 /* max_ancestor_rows */ ,
&bitmapqualorig, &indexquals,
&indexECs);
@@ -3433,9 +3573,24 @@ create_bitmap_scan_plan(PlannerInfo *root,
*/
static Plan *
create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
+ Cardinality max_ancestor_rows,
List **qual, List **indexqual, List **indexECs)
{
Plan *plan;
+ Cost cost; /* not used */
+ Selectivity selec;
+ Cardinality plan_rows;
+
+ /* How many rows will this node output? */
+ cost_bitmap_tree_node(bitmapqual, &cost, &selec);
+ plan_rows = clamp_row_est(selec * bitmapqual->parent->tuples);
+
+ /*
+ * At runtime, we'll reuse the left-most child's TID bitmap. Let that
+ * child that child know to request enough working memory to hold all its
+ * ancestors' results.
+ */
+ max_ancestor_rows = Max(max_ancestor_rows, plan_rows);
if (IsA(bitmapqual, BitmapAndPath))
{
@@ -3461,6 +3616,8 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
List *subindexEC;
subplan = create_bitmap_subplan(root, (Path *) lfirst(l),
+ foreach_current_index(l) == 0 ?
+ max_ancestor_rows : 0.0,
&subqual, &subindexqual,
&subindexEC);
subplans = lappend(subplans, subplan);
@@ -3472,8 +3629,7 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
plan = (Plan *) make_bitmap_and(subplans);
plan->startup_cost = apath->path.startup_cost;
plan->total_cost = apath->path.total_cost;
- plan->plan_rows =
- clamp_row_est(apath->bitmapselectivity * apath->path.parent->tuples);
+ plan->plan_rows = plan_rows;
plan->plan_width = 0; /* meaningless */
plan->parallel_aware = false;
plan->parallel_safe = apath->path.parallel_safe;
@@ -3508,6 +3664,8 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
List *subindexEC;
subplan = create_bitmap_subplan(root, (Path *) lfirst(l),
+ foreach_current_index(l) == 0 ?
+ max_ancestor_rows : 0.0,
&subqual, &subindexqual,
&subindexEC);
subplans = lappend(subplans, subplan);
@@ -3536,8 +3694,7 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
plan = (Plan *) make_bitmap_or(subplans);
plan->startup_cost = opath->path.startup_cost;
plan->total_cost = opath->path.total_cost;
- plan->plan_rows =
- clamp_row_est(opath->bitmapselectivity * opath->path.parent->tuples);
+ plan->plan_rows = plan_rows;
plan->plan_width = 0; /* meaningless */
plan->parallel_aware = false;
plan->parallel_safe = opath->path.parallel_safe;
@@ -3583,13 +3740,14 @@ create_bitmap_subplan(PlannerInfo *root, Path *bitmapqual,
/* and set its cost/width fields appropriately */
plan->startup_cost = 0.0;
plan->total_cost = ipath->indextotalcost;
- plan->plan_rows =
- clamp_row_est(ipath->indexselectivity * ipath->path.parent->tuples);
+ plan->plan_rows = plan_rows;
plan->plan_width = 0; /* meaningless */
plan->parallel_aware = false;
plan->parallel_safe = ipath->path.parallel_safe;
- plan->workmem_id = add_workmem(root->glob);
+ plan->workmem_id =
+ add_workmem(root->glob,
+ normalize_work_bytes(tbm_calculate_bytes(max_ancestor_rows)));
/* Extract original index clauses, actual index quals, relevant ECs */
subquals = NIL;
@@ -3898,7 +4056,15 @@ create_functionscan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
- scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+ /*
+ * Replace the path's total working-memory estimate with a per-function
+ * estimate.
+ */
+ scan_plan->scan.plan.workmem_id =
+ add_workmems(root->glob,
+ normalize_work_bytes(relation_byte_size(scan_plan->scan.plan.plan_rows,
+ scan_plan->scan.plan.plan_width)),
+ list_length(functions));
return scan_plan;
}
@@ -3943,7 +4109,8 @@ create_tablefuncscan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
- scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+ scan_plan->scan.plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(best_path->workmem));
return scan_plan;
}
@@ -4083,7 +4250,8 @@ create_ctescan_plan(PlannerInfo *root, Path *best_path,
copy_generic_path_info(&scan_plan->scan.plan, best_path);
- scan_plan->scan.plan.workmem_id = add_workmem(root->glob);
+ scan_plan->scan.plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(best_path->workmem));
return scan_plan;
}
@@ -4786,8 +4954,10 @@ create_mergejoin_plan(PlannerInfo *root,
*/
copy_plan_costsize(matplan, inner_plan);
matplan->total_cost += cpu_operator_cost * matplan->plan_rows;
-
- matplan->workmem_id = add_workmem(root->glob);
+ matplan->workmem_id =
+ add_workmem(root->glob,
+ normalize_work_bytes(relation_byte_size(matplan->plan_rows,
+ matplan->plan_width)));
inner_plan = matplan;
}
@@ -5135,7 +5305,9 @@ create_hashjoin_plan(PlannerInfo *root,
copy_generic_path_info(&join_plan->join.plan, &best_path->jpath.path);
/* Assign workmem to the Hash subnode, not its parent HashJoin node. */
- hash_plan->plan.workmem_id = add_hash_workmem(root->glob);
+ hash_plan->plan.workmem_id =
+ add_hash_workmem(root->glob,
+ normalize_work_kb(best_path->jpath.path.workmem));
return join_plan;
}
@@ -5690,7 +5862,8 @@ label_sort_with_costsize(PlannerInfo *root, Sort *plan, double limit_tuples)
plan->plan.parallel_aware = false;
plan->plan.parallel_safe = lefttree->parallel_safe;
- plan->plan.workmem_id = add_workmem(root->glob);
+ plan->plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(sort_path.workmem));
}
/*
@@ -5723,7 +5896,8 @@ label_incrementalsort_with_costsize(PlannerInfo *root, IncrementalSort *plan,
plan->sort.plan.parallel_aware = false;
plan->sort.plan.parallel_safe = lefttree->parallel_safe;
- plan->sort.plan.workmem_id = add_workmem(root->glob);
+ plan->sort.plan.workmem_id =
+ add_workmem(root->glob, normalize_work_kb(sort_path.workmem));
}
/*
@@ -6821,7 +6995,8 @@ materialize_finished_plan(PlannerGlobal *glob, Plan *subplan)
matplan->parallel_aware = false;
matplan->parallel_safe = subplan->parallel_safe;
- matplan->workmem_id = add_workmem(glob);
+ matplan->workmem_id =
+ add_workmem(glob, normalize_work_kb(matpath.workmem));
return matplan;
}
@@ -7590,12 +7765,22 @@ is_projection_capable_plan(Plan *plan)
}
static int
-add_workmem_internal(PlannerGlobal *glob, WorkMemCategory category)
+add_workmem_internal(PlannerGlobal *glob, WorkMemCategory category,
+ int estimate, int count)
{
+ if (estimate == 0 || count == 0)
+ return 0;
+
glob->workMemCategories = lappend_int(glob->workMemCategories, category);
+ glob->workMemEstimates = lappend_int(glob->workMemEstimates, estimate);
+ glob->workMemCounts = lappend_int(glob->workMemCounts, count);
/* the executor will fill this in later: */
glob->workMemLimits = lappend_int(glob->workMemLimits, 0);
+ Assert(list_length(glob->workMemCategories) ==
+ list_length(glob->workMemEstimates));
+ Assert(list_length(glob->workMemCategories) ==
+ list_length(glob->workMemCounts));
Assert(list_length(glob->workMemCategories) ==
list_length(glob->workMemLimits));
@@ -7608,10 +7793,10 @@ add_workmem_internal(PlannerGlobal *glob, WorkMemCategory category)
*
* This data structure will have its working-memory limit set to work_mem.
*/
-int
-add_workmem(PlannerGlobal *glob)
+static int
+add_workmem(PlannerGlobal *glob, int estimate)
{
- return add_workmem_internal(glob, WORKMEM_NORMAL);
+ return add_workmem_internal(glob, WORKMEM_NORMAL, estimate, 1);
}
/*
@@ -7622,7 +7807,13 @@ add_workmem(PlannerGlobal *glob)
* hash_mem_multiplier.
*/
int
-add_hash_workmem(PlannerGlobal *glob)
+add_hash_workmem(PlannerGlobal *glob, int estimate)
{
- return add_workmem_internal(glob, WORKMEM_HASH);
+ return add_workmem_internal(glob, WORKMEM_HASH, estimate, 1);
+}
+
+static int
+add_workmems(PlannerGlobal *glob, int estimate, int count)
+{
+ return add_workmem_internal(glob, WORKMEM_NORMAL, estimate, count);
}
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index a431808be96..007e298565a 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -585,6 +585,8 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
result->stmt_len = parse->stmt_len;
result->workMemCategories = glob->workMemCategories;
+ result->workMemEstimates = glob->workMemEstimates;
+ result->workMemCounts = glob->workMemCounts;
result->workMemLimits = glob->workMemLimits;
result->jitFlags = PGJIT_NONE;
diff --git a/src/backend/optimizer/prep/prepagg.c b/src/backend/optimizer/prep/prepagg.c
index c0a2f04a8c3..0d0fb5cf8ed 100644
--- a/src/backend/optimizer/prep/prepagg.c
+++ b/src/backend/optimizer/prep/prepagg.c
@@ -691,5 +691,17 @@ get_agg_clause_costs(PlannerInfo *root, AggSplit aggsplit, AggClauseCosts *costs
costs->finalCost.startup += argcosts.startup;
costs->finalCost.per_tuple += argcosts.per_tuple;
}
+
+ /*
+ * How many aggrefs need to sort their input? (Each such aggref gets
+ * its own sort buffer. The logic here MUST match the corresponding
+ * logic in function build_pertrans_for_aggref().)
+ */
+ if (!AGGKIND_IS_ORDERED_SET(aggref->aggkind) &&
+ !aggref->aggpresorted &&
+ (aggref->aggdistinct || aggref->aggorder))
+ {
+ ++costs->numSortBuffers;
+ }
}
}
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index a4c5867cdcb..070b86563b1 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -1737,6 +1737,13 @@ create_memoize_path(PlannerInfo *root, RelOptInfo *rel, Path *subpath,
pathnode->path.total_cost = subpath->total_cost + cpu_tuple_cost;
pathnode->path.rows = subpath->rows;
+ /*
+ * For now, set workmem at hash memory limit. Function
+ * cost_memoize_rescan() will adjust this field, same as it does for field
+ * "est_entries".
+ */
+ pathnode->path.workmem = normalize_work_bytes(get_hash_memory_limit());
+
return pathnode;
}
@@ -1965,12 +1972,14 @@ create_unique_path(PlannerInfo *root, RelOptInfo *rel, Path *subpath,
pathnode->path.disabled_nodes = agg_path.disabled_nodes;
pathnode->path.startup_cost = agg_path.startup_cost;
pathnode->path.total_cost = agg_path.total_cost;
+ pathnode->path.workmem = agg_path.workmem;
}
else
{
pathnode->path.disabled_nodes = sort_path.disabled_nodes;
pathnode->path.startup_cost = sort_path.startup_cost;
pathnode->path.total_cost = sort_path.total_cost;
+ pathnode->path.workmem = sort_path.workmem;
}
rel->cheapest_unique_path = (Path *) pathnode;
@@ -2317,6 +2326,13 @@ create_worktablescan_path(PlannerInfo *root, RelOptInfo *rel,
/* Cost is the same as for a regular CTE scan */
cost_ctescan(pathnode, root, rel, pathnode->param_info);
+ /*
+ * But working memory used is 0, since the worktable scan doesn't create a
+ * tuplestore -- it just reuses a tuplestore already created by a
+ * recursive union.
+ */
+ pathnode->workmem = 0;
+
return pathnode;
}
@@ -3314,6 +3330,7 @@ create_agg_path(PlannerInfo *root,
pathnode->aggstrategy = aggstrategy;
pathnode->aggsplit = aggsplit;
+ pathnode->numSortBuffers = aggcosts ? aggcosts->numSortBuffers : 0;
pathnode->numGroups = numGroups;
pathnode->transitionSpace = aggcosts ? aggcosts->transitionSpace : 0;
pathnode->groupClause = groupClause;
@@ -3364,6 +3381,8 @@ create_groupingsets_path(PlannerInfo *root,
ListCell *lc;
bool is_first = true;
bool is_first_sort = true;
+ int num_sort_nodes = 0;
+ double max_sort_workmem = 0.0;
/* The topmost generated Plan node will be an Agg */
pathnode->path.pathtype = T_Agg;
@@ -3400,6 +3419,7 @@ create_groupingsets_path(PlannerInfo *root,
pathnode->path.pathkeys = NIL;
pathnode->aggstrategy = aggstrategy;
+ pathnode->numSortBuffers = agg_costs ? agg_costs->numSortBuffers : 0;
pathnode->rollups = rollups;
pathnode->qual = having_qual;
pathnode->transitionSpace = agg_costs ? agg_costs->transitionSpace : 0;
@@ -3463,6 +3483,8 @@ create_groupingsets_path(PlannerInfo *root,
subpath->pathtarget->width);
if (!rollup->is_hashed)
is_first_sort = false;
+
+ pathnode->path.workmem += agg_path.workmem;
}
else
{
@@ -3475,6 +3497,12 @@ create_groupingsets_path(PlannerInfo *root,
work_mem,
-1.0);
+ /*
+ * We costed sorting the previous "sort" rollup's "sort_out"
+ * buffer. How much memory did it need?
+ */
+ max_sort_workmem = Max(max_sort_workmem, sort_path.workmem);
+
/* Account for cost of aggregation */
cost_agg(&agg_path, root,
@@ -3488,12 +3516,17 @@ create_groupingsets_path(PlannerInfo *root,
sort_path.total_cost,
sort_path.rows,
subpath->pathtarget->width);
+
+ pathnode->path.workmem += agg_path.workmem;
}
pathnode->path.disabled_nodes += agg_path.disabled_nodes;
pathnode->path.total_cost += agg_path.total_cost;
pathnode->path.rows += agg_path.rows;
}
+
+ if (!rollup->is_hashed)
+ ++num_sort_nodes;
}
/* add tlist eval cost for each output row */
@@ -3501,6 +3534,17 @@ create_groupingsets_path(PlannerInfo *root,
pathnode->path.total_cost += target->cost.startup +
target->cost.per_tuple * pathnode->path.rows;
+ /*
+ * Include working memory needed to sort agg output. If there's only 1
+ * sort rollup, then we don't need any memory. If there are 2 sort
+ * rollups, we need enough memory for 1 sort buffer. If there are >= 3
+ * sort rollups, we need only 2 sort buffers, since we're
+ * double-buffering.
+ */
+ pathnode->path.workmem += num_sort_nodes > 2 ?
+ max_sort_workmem * 2.0 :
+ max_sort_workmem;
+
return pathnode;
}
@@ -3650,7 +3694,8 @@ create_windowagg_path(PlannerInfo *root,
subpath->disabled_nodes,
subpath->startup_cost,
subpath->total_cost,
- subpath->rows);
+ subpath->rows,
+ subpath->pathtarget->width);
/* add tlist eval cost for each output row */
pathnode->path.startup_cost += target->cost.startup;
@@ -3775,7 +3820,11 @@ create_setop_path(PlannerInfo *root,
MAXALIGN(SizeofMinimalTupleHeader);
if (hashentrysize * numGroups > get_hash_memory_limit())
pathnode->path.disabled_nodes++;
+
+ pathnode->path.workmem =
+ normalize_work_bytes(numGroups * hashentrysize);
}
+
pathnode->path.rows = outputRows;
return pathnode;
@@ -3826,7 +3875,7 @@ create_recursiveunion_path(PlannerInfo *root,
pathnode->wtParam = wtParam;
pathnode->numGroups = numGroups;
- cost_recursive_union(&pathnode->path, leftpath, rightpath);
+ cost_recursive_union(pathnode, leftpath, rightpath);
return pathnode;
}
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
index e4e9e0d1de1..6cd9bffbee5 100644
--- a/src/include/executor/nodeHash.h
+++ b/src/include/executor/nodeHash.h
@@ -63,7 +63,8 @@ extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
- int *num_skew_mcvs);
+ int *num_skew_mcvs,
+ int *workmem);
extern int ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue);
extern void ExecHashEstimate(HashState *node, ParallelContext *pcxt);
extern void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index d543011d92a..e15c37608d1 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1283,6 +1283,18 @@ typedef struct PlanState
#define workMemField(node, field) \
(workMemFieldFromId((node), field, ((PlanState *)(node))->plan->workmem_id))
+/* workmem estimate: */
+#define workMemEstimateFromId(node, id) \
+ (workMemFieldFromId(node, workMemEstimates, id))
+#define workMemEstimate(node) \
+ (workMemField(node, workMemEstimates))
+
+/* workmem count: */
+#define workMemCountFromId(node, id) \
+ (workMemFieldFromId(node, workMemCounts, id))
+#define workMemCount(node) \
+ (workMemField(node, workMemCounts))
+
/* workmem limit: */
#define workMemLimitFromId(node, id) \
(workMemFieldFromId(node, workMemLimits, id))
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 181437ac933..779a56ede1a 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -60,6 +60,7 @@ typedef struct AggClauseCosts
QualCost transCost; /* total per-input-row execution costs */
QualCost finalCost; /* total per-aggregated-row costs */
Size transitionSpace; /* space for pass-by-ref transition data */
+ int numSortBuffers; /* # of required input-sort buffers */
} AggClauseCosts;
/*
@@ -188,9 +189,12 @@ typedef struct PlannerGlobal
* needs working memory for a data structure maintains a "workmem_id"
* index into the following lists (all kept in sync).
*/
-
/* - IntList (of WorkMemCategory): is this a Hash or "normal" limit? */
List *workMemCategories;
+ /* - IntList: estimate (in KB) of memory needed to avoid spilling */
+ List *workMemEstimates;
+ /* - IntList: how many data structures get a copy of this info */
+ List *workMemCounts;
/* - IntList: limit (in KB), after which data structure must spill */
List *workMemLimits;
} PlannerGlobal;
@@ -1807,6 +1811,7 @@ typedef struct Path
int disabled_nodes; /* count of disabled nodes */
Cost startup_cost; /* cost expended before fetching any tuples */
Cost total_cost; /* total cost (assuming all tuples fetched) */
+ Cost workmem; /* estimated work_mem (in KB) */
/* sort ordering of path's output; a List of PathKey nodes; see above */
List *pathkeys;
@@ -2411,6 +2416,7 @@ typedef struct AggPath
Path *subpath; /* path representing input source */
AggStrategy aggstrategy; /* basic strategy, see nodes.h */
AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
+ int numSortBuffers; /* number of inputs that require sorting */
Cardinality numGroups; /* estimated number of groups in input */
uint64 transitionSpace; /* for pass-by-ref transition data */
List *groupClause; /* a list of SortGroupClause's */
@@ -2452,6 +2458,7 @@ typedef struct GroupingSetsPath
Path path;
Path *subpath; /* path representing input source */
AggStrategy aggstrategy; /* basic strategy */
+ int numSortBuffers; /* number of inputs that require sorting */
List *rollups; /* list of RollupData */
List *qual; /* quals (HAVING quals), if any */
uint64 transitionSpace; /* for pass-by-ref transition data */
@@ -3495,6 +3502,7 @@ typedef struct JoinCostWorkspace
/* Fields below here should be treated as private to costsize.c */
Cost run_cost; /* non-startup cost components */
+ Cost workmem; /* estimated work_mem (in KB) */
/* private for cost_nestloop code */
Cost inner_run_cost; /* also used by cost_mergejoin code */
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index ba8fdc2e6db..2134b15f95f 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -160,9 +160,12 @@ typedef struct PlannedStmt
* needs working memory for a data structure maintains a "workmem_id"
* index into the following lists (all kept in sync).
*/
-
/* - IntList (of WorkMemCategory): is this a Hash or "normal" limit? */
List *workMemCategories;
+ /* - IntList: estimate (in KB) of memory needed to avoid spilling */
+ List *workMemEstimates;
+ /* - IntList: how many data structures get a copy of this info */
+ List *workMemCounts;
/* - IntList: limit (in KB), after which data structure must spill */
List *workMemLimits;
} PlannedStmt;
@@ -1191,6 +1194,8 @@ typedef struct Agg
Oid *grpOperators pg_node_attr(array_size(numCols));
Oid *grpCollations pg_node_attr(array_size(numCols));
+ /* number of inputs that require sorting */
+ int numSorts;
/* 1-based id of workMem to use to sort inputs, or else zero */
int sortWorkMemId;
diff --git a/src/include/nodes/tidbitmap.h b/src/include/nodes/tidbitmap.h
index 99f795ceab5..d89a0f71a72 100644
--- a/src/include/nodes/tidbitmap.h
+++ b/src/include/nodes/tidbitmap.h
@@ -108,6 +108,7 @@ extern void tbm_end_shared_iterate(TBMSharedIterator *iterator);
extern TBMSharedIterator *tbm_attach_shared_iterate(dsa_area *dsa,
dsa_pointer dp);
extern int tbm_calculate_entries(Size maxbytes);
+extern double tbm_calculate_bytes(double maxentries);
extern TBMIterator tbm_begin_iterate(TIDBitmap *tbm,
dsa_area *dsa, dsa_pointer dsp);
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index b523bcda8f3..ef80f6f9339 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -106,7 +106,7 @@ extern void cost_namedtuplestorescan(Path *path, PlannerInfo *root,
RelOptInfo *baserel, ParamPathInfo *param_info);
extern void cost_resultscan(Path *path, PlannerInfo *root,
RelOptInfo *baserel, ParamPathInfo *param_info);
-extern void cost_recursive_union(Path *runion, Path *nrterm, Path *rterm);
+extern void cost_recursive_union(RecursiveUnionPath *runion, Path *nrterm, Path *rterm);
extern void cost_sort(Path *path, PlannerInfo *root,
List *pathkeys, int input_disabled_nodes,
Cost input_cost, double tuples, int width,
@@ -139,7 +139,7 @@ extern void cost_windowagg(Path *path, PlannerInfo *root,
List *windowFuncs, WindowClause *winclause,
int input_disabled_nodes,
Cost input_startup_cost, Cost input_total_cost,
- double input_tuples);
+ double input_tuples, int width);
extern void cost_group(Path *path, PlannerInfo *root,
int numGroupCols, double numGroups,
List *quals,
@@ -218,9 +218,18 @@ extern void set_namedtuplestore_size_estimates(PlannerInfo *root, RelOptInfo *re
extern void set_result_size_estimates(PlannerInfo *root, RelOptInfo *rel);
extern void set_foreign_size_estimates(PlannerInfo *root, RelOptInfo *rel);
extern PathTarget *set_pathtarget_cost_width(PlannerInfo *root, PathTarget *target);
+extern double relation_byte_size(double tuples, int width);
extern double compute_bitmap_pages(PlannerInfo *root, RelOptInfo *baserel,
Path *bitmapqual, double loop_count,
Cost *cost_p, double *tuples_p);
extern double compute_gather_rows(Path *path);
+extern int compute_agg_input_workmem(double input_tuples, double input_width);
+extern int compute_agg_output_workmem(PlannerInfo *root,
+ AggStrategy aggstrategy,
+ double numGroups, uint64 transitionSpace,
+ double input_tuples, double input_width,
+ bool cost_sort);
+extern int normalize_work_kb(double nkb);
+extern int normalize_work_bytes(double nbytes);
#endif /* COST_H */
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index 8436136026b..21894adffcc 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -49,8 +49,7 @@ extern Plan *change_plan_targetlist(Plan *subplan, List *tlist,
extern Plan *materialize_finished_plan(PlannerGlobal *glob, Plan *subplan);
extern bool is_projection_capable_path(Path *path);
extern bool is_projection_capable_plan(Plan *plan);
-extern int add_workmem(PlannerGlobal *glob);
-extern int add_hash_workmem(PlannerGlobal *glob);
+extern int add_hash_workmem(PlannerGlobal *glob, int estimate);
/* External use of these functions is deprecated: */
extern Sort *make_sort_from_sortclauses(List *sortcls, Plan *lefttree);
--
2.39.5
0003-Add-EXPLAIN-work_mem-on-command-option.patch (text/x-diff, 47.9K) download
From f4824a23a2726059148de260d15b16dc411b19f1 Mon Sep 17 00:00:00 2001
From: James Hunter <james.hunter.pg@gmail.com>
Date: Wed, 26 Feb 2025 01:02:19 +0000
Subject: [PATCH 3/4] Add EXPLAIN (work_mem on) command option
So that users can see how much working memory a query is likely to use, as
well as how much memory it will be limited to, this commit adds an
EXPLAIN (work_mem on) command option that displays the workmem estimate
and limit, added in the previous two commits.
---
src/backend/commands/explain.c | 231 +++++++++
src/backend/commands/explain_state.c | 2 +
src/backend/executor/nodeHash.c | 7 +-
src/backend/optimizer/path/costsize.c | 4 +-
src/include/commands/explain_state.h | 3 +
src/include/executor/nodeHash.h | 2 +-
src/test/regress/expected/workmem.out | 653 ++++++++++++++++++++++++++
src/test/regress/parallel_schedule | 2 +-
src/test/regress/sql/workmem.sql | 307 ++++++++++++
9 files changed, 1205 insertions(+), 6 deletions(-)
create mode 100644 src/test/regress/expected/workmem.out
create mode 100644 src/test/regress/sql/workmem.sql
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 8345bc0264b..bb73ab8741e 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -22,6 +22,8 @@
#include "commands/explain_format.h"
#include "commands/explain_state.h"
#include "commands/prepare.h"
+#include "executor/hashjoin.h"
+#include "executor/nodeHash.h"
#include "foreign/fdwapi.h"
#include "jit/jit.h"
#include "libpq/pqformat.h"
@@ -29,6 +31,7 @@
#include "nodes/extensible.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
+#include "optimizer/cost.h"
#include "parser/analyze.h"
#include "parser/parsetree.h"
#include "rewrite/rewriteHandler.h"
@@ -165,6 +168,14 @@ static ExplainWorkersState *ExplainCreateWorkersState(int num_workers);
static void ExplainOpenWorker(int n, ExplainState *es);
static void ExplainCloseWorker(int n, ExplainState *es);
static void ExplainFlushWorkersState(ExplainState *es);
+static void compute_subplan_workmem(List *plans, double *sp_estimate,
+ double *sp_limit);
+static void compute_agg_workmem(PlanState *planstate, Agg *agg,
+ double *agg_estimate, double *agg_limit);
+static void compute_hash_workmem(PlanState *planstate, double *hash_estimate,
+ double *hash_limit);
+static void increment_workmem(PlanState *planstate, int workmem_id,
+ double *estimate, double *limit);
@@ -678,6 +689,14 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
ExplainPropertyFloat("Execution Time", "ms", 1000.0 * totaltime, 3,
es);
+ if (es->work_mem)
+ {
+ ExplainPropertyFloat("Total Working Memory Estimate", "kB",
+ es->total_workmem_estimate, 0, es);
+ ExplainPropertyFloat("Total Working Memory Limit", "kB",
+ es->total_workmem_limit, 0, es);
+ }
+
ExplainCloseGroup("Query", NULL, true, es);
}
@@ -1813,6 +1832,72 @@ ExplainNode(PlanState *planstate, List *ancestors,
}
}
+ if (es->work_mem)
+ {
+ double plan_estimate = 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_estimate,
+ &plan_limit);
+ compute_subplan_workmem(planstate->subPlan, &plan_estimate,
+ &plan_limit);
+
+ /* Include working memory used by this Plan, itself. */
+ switch (nodeTag(plan))
+ {
+ case T_Agg:
+ compute_agg_workmem(planstate, (Agg *) plan,
+ &plan_estimate, &plan_limit);
+ break;
+ case T_Hash:
+ compute_hash_workmem(planstate, &plan_estimate, &plan_limit);
+ break;
+ case T_RecursiveUnion:
+ {
+ RecursiveUnion *runion = (RecursiveUnion *) plan;
+
+ if (runion->hashWorkMemId > 0)
+ increment_workmem(planstate, runion->hashWorkMemId,
+ &plan_estimate, &plan_limit);
+ }
+ /* FALLTHROUGH */
+ default:
+ if (plan->workmem_id > 0)
+ increment_workmem(planstate, plan->workmem_id,
+ &plan_estimate, &plan_limit);
+ break;
+ }
+
+ /*
+ * Every parallel worker (plus the leader) gets its own copy of
+ * working memory.
+ */
+ plan_estimate *= (1 + es->num_workers);
+ plan_limit *= (1 + es->num_workers);
+
+ es->total_workmem_estimate += plan_estimate;
+ es->total_workmem_limit += plan_limit;
+
+ if (plan_estimate > 0.0 || plan_limit > 0.0)
+ {
+ if (es->format == EXPLAIN_FORMAT_TEXT)
+ appendStringInfo(es->str,
+ " (work_mem=%.0f kB) (limit=%.0f kB)",
+ plan_estimate, plan_limit);
+ else
+ {
+ ExplainPropertyFloat("Working Memory Estimate", "kB",
+ plan_estimate, 0, es);
+ ExplainPropertyFloat("Working Memory Limit", "kB",
+ plan_limit, 0, es);
+ }
+ }
+ }
+
/*
* We have to forcibly clean up the instrumentation state because we
* haven't done ExecutorEnd yet. This is pretty grotty ...
@@ -2366,6 +2451,24 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (planstate->initPlan)
ExplainSubPlans(planstate->initPlan, ancestors, "InitPlan", es);
+ if (nodeTag(plan) == T_Gather || nodeTag(plan) == T_GatherMerge)
+ {
+ /*
+ * Other than initPlan-s, every node below us gets the # of planned
+ * workers we specified.
+ */
+ Assert(es->num_workers == 0);
+
+ if (nodeTag(plan) == T_Gather)
+ es->num_workers = es->analyze ?
+ ((GatherState *) planstate)->nworkers_launched :
+ ((Gather *) plan)->num_workers;
+ else
+ es->num_workers = es->analyze ?
+ ((GatherMergeState *) planstate)->nworkers_launched :
+ ((GatherMerge *) plan)->num_workers;
+ }
+
/* lefttree */
if (outerPlanState(planstate))
ExplainNode(outerPlanState(planstate), ancestors,
@@ -2422,6 +2525,12 @@ ExplainNode(PlanState *planstate, List *ancestors,
ExplainCloseGroup("Plans", "Plans", false, es);
}
+ if (nodeTag(plan) == T_Gather || nodeTag(plan) == T_GatherMerge)
+ {
+ /* End of parallel sub-tree. */
+ es->num_workers = 0;
+ }
+
/* in text format, undo whatever indentation we added */
if (es->format == EXPLAIN_FORMAT_TEXT)
es->indent = save_indent;
@@ -4994,3 +5103,125 @@ ExplainFlushWorkersState(ExplainState *es)
pfree(wstate->worker_state_save);
pfree(wstate);
}
+
+/*
+ * compute_subplan_work_mem - compute total workmem for a SubPlan object
+ *
+ * If a SubPlan object uses a hash table, then that hash table needs working
+ * memory. We display that working memory on the owning Plan. This function
+ * increments work_mem counters to include the SubPlan's working-memory.
+ */
+static void
+compute_subplan_workmem(List *plans, double *sp_estimate, double *sp_limit)
+{
+ foreach_node(SubPlanState, sps, plans)
+ {
+ SubPlan *sp = sps->subplan;
+
+ if (sp->hashtab_workmem_id > 0)
+ increment_workmem(sps->planstate, sp->hashtab_workmem_id,
+ sp_estimate, sp_limit);
+
+ if (sp->hashnul_workmem_id > 0)
+ increment_workmem(sps->planstate, sp->hashnul_workmem_id,
+ sp_estimate, sp_limit);
+ }
+}
+
+static void
+compute_agg_workmem_node(PlanState *planstate, Agg *agg, double *agg_estimate,
+ double *agg_limit)
+{
+ /* Record memory used for output data structures. */
+ if (agg->plan.workmem_id > 0)
+ increment_workmem(planstate, agg->plan.workmem_id, agg_estimate,
+ agg_limit);
+
+ /* Record memory used for input sort buffers. */
+ if (agg->sortWorkMemId > 0)
+ increment_workmem(planstate, agg->sortWorkMemId, agg_estimate,
+ agg_limit);
+}
+
+/*
+ * compute_agg_workmem - compute Agg node's total workmem estimate and limit
+ *
+ * An Agg node might point to a chain of additional Agg nodes. When we explain
+ * the plan, we display only the first, "main" Agg node.
+ */
+static void
+compute_agg_workmem(PlanState *planstate, Agg *agg, double *agg_estimate,
+ double *agg_limit)
+{
+ compute_agg_workmem_node(planstate, agg, agg_estimate, agg_limit);
+
+ /* Also include the chain of GROUPING SETS aggs. */
+ foreach_node(Agg, aggnode, agg->chain)
+ compute_agg_workmem_node(planstate, aggnode, agg_estimate, agg_limit);
+}
+
+/*
+ * compute_hash_workmem - compute total workmem for a Hash node
+ *
+ * This function is complicated, because we currently can adjust workmem limits
+ * for Hash (Joins), at runtime; and because the memory a Hash (Join) needs
+ * per-batch is not currently counted against the workmem limit.
+ *
+ * Here, we try to give a more accurate accounting than we'd get from just
+ * displaying limit * count.
+ */
+static void
+compute_hash_workmem(PlanState *planstate, double *hash_estimate,
+ double *hash_limit)
+{
+ double count = workMemCount(planstate);
+ double estimate = workMemEstimate(planstate);
+ size_t limit = workMemLimit(planstate);
+ HashState *hstate = (HashState *) planstate;
+ Plan *plan = planstate->plan;
+ Hash *hash = (Hash *) plan;
+ Plan *outerNode = outerPlan(plan);
+ double rows;
+ size_t nbytes;
+ size_t total_space_allowed; /* ignored */
+ int nbuckets; /* ignored */
+ int nbatch;
+ int num_skew_mcvs; /* ignored */
+ int workmem_estimate; /* ignored */
+
+ /*
+ * For Hash Joins, we currently don't count per-batch memory against the
+ * "workmem_limit", but we can at least estimate it for display with the
+ * Plan.
+ */
+ rows = plan->parallel_aware ? hash->rows_total : outerNode->plan_rows;
+ nbytes = limit * 1024;
+
+ ExecChooseHashTableSize(rows, outerNode->plan_width,
+ OidIsValid(hash->skewTable),
+ hstate->parallel_state != NULL,
+ hstate->parallel_state != NULL ?
+ hstate->parallel_state->nparticipants - 1 : 0,
+ &nbytes, &total_space_allowed,
+ &nbuckets, &nbatch, &num_skew_mcvs,
+ &workmem_estimate);
+
+ /* Include space for per-batch memory, if any: 2 blocks per batch. */
+ if (nbatch > 1)
+ nbytes += nbatch * 2 * BLCKSZ;
+
+ Assert(nbytes >= limit * 1024);
+
+ *hash_estimate += estimate * count;
+ *hash_limit += (double) normalize_work_bytes(nbytes) * count;
+}
+
+static void
+increment_workmem(PlanState *planstate, int workmem_id, double *estimate,
+ double *limit)
+{
+ double count = workMemCountFromId(planstate, workmem_id);
+
+ *estimate += workMemEstimateFromId(planstate, workmem_id) * count;
+ *limit += workMemLimitFromId(planstate, workmem_id) * count;
+}
diff --git a/src/backend/commands/explain_state.c b/src/backend/commands/explain_state.c
index 60d98d63a62..eafa15b6795 100644
--- a/src/backend/commands/explain_state.c
+++ b/src/backend/commands/explain_state.c
@@ -115,6 +115,8 @@ ParseExplainOptionList(ExplainState *es, List *options, ParseState *pstate)
}
else if (strcmp(opt->defname, "memory") == 0)
es->memory = defGetBoolean(opt);
+ else if (strcmp(opt->defname, "work_mem") == 0)
+ es->work_mem = defGetBoolean(opt);
else if (strcmp(opt->defname, "serialize") == 0)
{
if (opt->arg)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 7d09ac8b5a3..6ae3d649be6 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -482,7 +482,7 @@ ExecHashTableCreate(HashState *state)
state->parallel_state != NULL,
state->parallel_state != NULL ?
state->parallel_state->nparticipants - 1 : 0,
- worker_space_allowed,
+ &worker_space_allowed,
&space_allowed,
&nbuckets, &nbatch, &num_skew_mcvs, &workmem);
@@ -666,7 +666,7 @@ void
ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
bool try_combined_hash_mem,
int parallel_workers,
- size_t worker_space_allowed,
+ size_t *worker_space_allowed,
size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
@@ -699,7 +699,7 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
/*
* Caller tells us our (per-worker) in-memory hashtable size limit.
*/
- hash_table_bytes = worker_space_allowed;
+ hash_table_bytes = *worker_space_allowed;
/*
* Parallel Hash tries to use the combined hash_mem of all workers to
@@ -963,6 +963,7 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
nbatch /= 2;
nbuckets *= 2;
+ *worker_space_allowed = (*worker_space_allowed) * 2;
*total_space_allowed = (*total_space_allowed) * 2;
}
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 27daa1966c2..d7fd4b214d8 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -4309,6 +4309,7 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
int numbuckets;
int numbatches;
int num_skew_mcvs;
+ size_t worker_space_allowed;
size_t space_allowed; /* unused */
/* Count up disabled nodes. */
@@ -4354,12 +4355,13 @@ initial_cost_hashjoin(PlannerInfo *root, JoinCostWorkspace *workspace,
* XXX at some point it might be interesting to try to account for skew
* optimization in the cost estimate, but for now, we don't.
*/
+ worker_space_allowed = get_hash_memory_limit();
ExecChooseHashTableSize(inner_path_rows_total,
inner_path->pathtarget->width,
true, /* useskew */
parallel_hash, /* try_combined_hash_mem */
outer_path->parallel_workers,
- get_hash_memory_limit(),
+ &worker_space_allowed,
&space_allowed,
&numbuckets,
&numbatches,
diff --git a/src/include/commands/explain_state.h b/src/include/commands/explain_state.h
index 32728f5d1a1..98639a32aec 100644
--- a/src/include/commands/explain_state.h
+++ b/src/include/commands/explain_state.h
@@ -69,6 +69,9 @@ typedef struct ExplainState
bool hide_workers; /* set if we find an invisible Gather */
int rtable_size; /* length of rtable excluding the RTE_GROUP
* entry */
+ int num_workers; /* # of worker processes *planned* to use */
+ double total_workmem_estimate; /* total working memory estimate */
+ double total_workmem_limit; /* total working memory limit */
/* state related to the current plan node */
ExplainWorkersState *workers_state; /* needed if parallel plan */
/* extensions */
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
index 6cd9bffbee5..b346a270b67 100644
--- a/src/include/executor/nodeHash.h
+++ b/src/include/executor/nodeHash.h
@@ -59,7 +59,7 @@ extern void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
bool try_combined_hash_mem,
int parallel_workers,
- size_t worker_space_allowed,
+ size_t *worker_space_allowed,
size_t *total_space_allowed,
int *numbuckets,
int *numbatches,
diff --git a/src/test/regress/expected/workmem.out b/src/test/regress/expected/workmem.out
new file mode 100644
index 00000000000..ca8edde6d5f
--- /dev/null
+++ b/src/test/regress/expected/workmem.out
@@ -0,0 +1,653 @@
+----
+-- Tests that show "work_mem" output to EXPLAIN plans.
+----
+-- Note: Function derived from file explain.sql. We can't use that other
+-- function, since we're run in parallel with explain.sql.
+create or replace function workmem_filter(text) returns setof text
+language plpgsql as
+$$
+declare
+ ln text;
+begin
+ for ln in execute $1
+ loop
+ -- Mask out work_mem estimate, since it might be brittle
+ ln := regexp_replace(ln, '\mwork_mem=\d+\M', 'work_mem=N', 'g');
+ ln := regexp_replace(ln, '\mMemory Estimate: \d+\M', 'Memory Estimate: N', 'g');
+ return next ln;
+ end loop;
+end;
+$$;
+-- Unique -> hash agg
+set enable_hashagg = on;
+select workmem_filter('
+explain (costs off, work_mem on)
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+');
+ workmem_filter
+-----------------------------------------------------------------
+ Sort (work_mem=N kB) (limit=4096 kB)
+ Sort Key: onek.unique1
+ -> Nested Loop
+ -> 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 Estimate: N kB
+ Total Working Memory Limit: 12288 kB
+(11 rows)
+
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 1 | 214 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 2 | 3 | BAAAAA | GIAAAA | OOOOxx
+ 20 | 306 | 0 | 0 | 0 | 0 | 0 | 20 | 20 | 20 | 20 | 0 | 1 | UAAAAA | ULAAAA | OOOOxx
+ 99 | 101 | 1 | 3 | 9 | 19 | 9 | 99 | 99 | 99 | 99 | 18 | 19 | VDAAAA | XDAAAA | HHHHxx
+(3 rows)
+
+reset enable_hashagg;
+-- Unique -> sort
+set enable_hashagg = off;
+select workmem_filter('
+explain (costs off, work_mem on)
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+');
+ workmem_filter
+----------------------------------------------------------------------
+ Sort (work_mem=N kB) (limit=4096 kB)
+ Sort Key: onek.unique1
+ -> Nested Loop
+ -> Unique
+ -> 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 Estimate: N kB
+ Total Working Memory Limit: 8192 kB
+(12 rows)
+
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 1 | 214 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 1 | 2 | 3 | BAAAAA | GIAAAA | OOOOxx
+ 20 | 306 | 0 | 0 | 0 | 0 | 0 | 20 | 20 | 20 | 20 | 0 | 1 | UAAAAA | ULAAAA | OOOOxx
+ 99 | 101 | 1 | 3 | 9 | 19 | 9 | 99 | 99 | 99 | 99 | 18 | 19 | VDAAAA | XDAAAA | HHHHxx
+(3 rows)
+
+reset enable_hashagg;
+-- Incremental Sort
+select workmem_filter('
+explain (costs off, work_mem on)
+select * from (select * from tenk1 order by four) t order by four, ten
+limit 1;
+');
+ workmem_filter
+---------------------------------------------------------
+ Limit
+ -> Incremental Sort (work_mem=N kB) (limit=8192 kB)
+ Sort Key: tenk1.four, tenk1.ten
+ Presorted Key: tenk1.four
+ -> Sort (work_mem=N kB) (limit=4096 kB)
+ Sort Key: tenk1.four
+ -> Seq Scan on tenk1
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 12288 kB
+(9 rows)
+
+select * from (select * from tenk1 order by four) t order by four, ten
+limit 1;
+ unique1 | unique2 | two | four | ten | twenty | hundred | thousand | twothousand | fivethous | tenthous | odd | even | stringu1 | stringu2 | string4
+---------+---------+-----+------+-----+--------+---------+----------+-------------+-----------+----------+-----+------+----------+----------+---------
+ 4220 | 5017 | 0 | 0 | 0 | 0 | 20 | 220 | 220 | 4220 | 4220 | 40 | 41 | IGAAAA | ZKHAAA | HHHHxx
+(1 row)
+
+-- Hash Join
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select t1.unique1, t2.hundred
+from onek t1, tenk1 t2
+where exists (select 1 from tenk1 t3
+ where t3.thousand = t1.unique1 and t3.tenthous = t2.hundred)
+ and t1.unique1 < 1
+) t;
+');
+ workmem_filter
+--------------------------------------------------------------------------------
+ Aggregate
+ -> Nested Loop
+ -> Hash Join
+ Hash Cond: (t3.thousand = t1.unique1)
+ -> 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) (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 Estimate: N kB
+ Total Working Memory Limit: 16384 kB
+(14 rows)
+
+select count(*) from (
+select t1.unique1, t2.hundred
+from onek t1, tenk1 t2
+where exists (select 1 from tenk1 t3
+ where t3.thousand = t1.unique1 and t3.tenthous = t2.hundred)
+ and t1.unique1 < 1
+) t;
+ count
+-------
+ 100
+(1 row)
+
+-- Materialize
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select t1.f1
+from int4_tbl t1, int4_tbl t2
+ left join int4_tbl t3 on t3.f1 > 0
+ left join int4_tbl t4 on t3.f1 > 1
+where t4.f1 is null
+) t;
+');
+ 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) (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) (limit=4096 kB)
+ -> Seq Scan on int4_tbl t4
+ -> Seq Scan on int4_tbl t1
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 8192 kB
+(15 rows)
+
+select count(*) from (
+select t1.f1
+from int4_tbl t1, int4_tbl t2
+ left join int4_tbl t3 on t3.f1 > 0
+ left join int4_tbl t4 on t3.f1 > 1
+where t4.f1 is null
+) t;
+ count
+-------
+ 0
+(1 row)
+
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ 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) (limit=8192 kB)
+ Hash Key: "*VALUES*".column1, "*VALUES*".column2
+ Hash Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ 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
+group by grouping sets((a, b), (a));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+(4 rows)
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ 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) (limit=8192 kB)
+ Group Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1
+ Sort Key: "*VALUES*".column2
+ Group Key: "*VALUES*".column2
+ Sort Key: "*VALUES*".column3
+ Group Key: "*VALUES*".column3
+ Sort Key: "*VALUES*".column4
+ Group Key: "*VALUES*".column4
+ -> Sort (work_mem=N kB) (limit=4096 kB)
+ Sort Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ 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
+group by grouping sets((a, b), (a), (b), (c), (d));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+ | | 5
+ | | 6
+ | | 7
+ | | 8
+ | 1 | 9
+ | 2 | 10
+(10 rows)
+
+reset enable_hashagg;
+-- Agg (hash, parallel)
+set parallel_setup_cost=0;
+set parallel_tuple_cost=0;
+set min_parallel_table_scan_size=0;
+set max_parallel_workers_per_gather=4;
+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) (limit=8192 kB)
+ Group Key: (length((stringu1)::text))
+ -> Gather
+ Workers Planned: 4
+ -> Partial HashAggregate (work_mem=N kB) (limit=40960 kB)
+ Group Key: length((stringu1)::text)
+ -> Parallel Seq Scan on tenk1
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 49152 kB
+(9 rows)
+
+select length(stringu1) from tenk1 group by length(stringu1);
+ length
+--------
+ 6
+(1 row)
+
+reset parallel_setup_cost;
+reset parallel_tuple_cost;
+reset min_parallel_table_scan_size;
+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
+-------------------------------------
+ Aggregate
+ -> Seq Scan on tenk1
+ Total Working Memory Estimate: 0 kB
+ Total Working Memory Limit: 0 kB
+(4 rows)
+
+select MAX(length(stringu1)) from tenk1;
+ max
+-----
+ 6
+(1 row)
+
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+ workmem_filter
+---------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=4096 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(4 rows)
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+ count
+-------
+ 2000
+(1 row)
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+ workmem_filter
+--------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series (work_mem=N kB) (limit=12288 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 12288 kB
+(4 rows)
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+ count
+-------
+ 12
+(1 row)
+
+-- Table Function Scan
+CREATE TABLE workmem_xmldata(data xml);
+select workmem_filter('
+EXPLAIN (COSTS OFF, work_mem on)
+SELECT xmltable.*
+ FROM (SELECT data FROM workmem_xmldata) x,
+ LATERAL XMLTABLE(''/ROWS/ROW''
+ PASSING data
+ COLUMNS id int PATH ''@id'',
+ _id FOR ORDINALITY,
+ country_name text PATH ''COUNTRY_NAME'' NOT NULL,
+ country_id text PATH ''COUNTRY_ID'',
+ region_id int PATH ''REGION_ID'',
+ size float PATH ''SIZE'',
+ unit text PATH ''SIZE/@unit'',
+ premier_name text PATH ''PREMIER_NAME'' DEFAULT ''not specified'');
+');
+ workmem_filter
+--------------------------------------------------------------------------
+ Nested Loop
+ -> Seq Scan on workmem_xmldata
+ -> Table Function Scan on "xmltable" (work_mem=N kB) (limit=4096 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(5 rows)
+
+SELECT xmltable.*
+ FROM (SELECT data FROM workmem_xmldata) x,
+ LATERAL XMLTABLE('/ROWS/ROW'
+ PASSING data
+ COLUMNS id int PATH '@id',
+ _id FOR ORDINALITY,
+ country_name text PATH 'COUNTRY_NAME' NOT NULL,
+ country_id text PATH 'COUNTRY_ID',
+ region_id int PATH 'REGION_ID',
+ size float PATH 'SIZE',
+ unit text PATH 'SIZE/@unit',
+ premier_name text PATH 'PREMIER_NAME' DEFAULT 'not specified');
+ id | _id | country_name | country_id | region_id | size | unit | premier_name
+----+-----+--------------+------------+-----------+------+------+--------------
+(0 rows)
+
+drop table workmem_xmldata;
+-- SetOp [no work_mem]
+explain (costs off, work_mem on)
+select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
+ QUERY PLAN
+------------------------------------------------------------
+ SetOp Except
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ -> Index Only Scan using tenk1_unique2 on tenk1 tenk1_1
+ Filter: (unique2 <> 10)
+ Total Working Memory Estimate: 0 kB
+ Total Working Memory Limit: 0 kB
+(6 rows)
+
+select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
+ unique1
+---------
+ 10
+(1 row)
+
+-- HashSetOp
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from
+ ( select unique1 from tenk1 intersect select fivethous from tenk1 ) ss;
+');
+ workmem_filter
+------------------------------------------------------------------
+ Aggregate
+ -> 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 Estimate: N kB
+ Total Working Memory Limit: 8192 kB
+(6 rows)
+
+select count(*) from
+ ( select unique1 from tenk1 intersect select fivethous from tenk1 ) ss;
+ count
+-------
+ 5000
+(1 row)
+
+-- RecursiveUnion and Memoize (also WorkTable Scan [no work_mem])
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(o.four), sum(ss.a) from onek o
+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
+-----------------------------------------------------------------------------
+ Aggregate
+ -> Nested Loop
+ -> Seq Scan on onek o
+ Filter: (ten = 1)
+ -> Memoize (work_mem=N kB) (limit=8192 kB)
+ Cache Key: o.four
+ Cache Mode: binary
+ -> CTE Scan on x (work_mem=N kB) (limit=4096 kB)
+ CTE x
+ -> Recursive Union (work_mem=N kB) (limit=16384 kB)
+ -> Result
+ -> WorkTable Scan on x x_1
+ Filter: (a < 10)
+ Total Working Memory Estimate: N kB
+ 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 (
+ select o.four as a union select a + 1 from x where a < 10)
+ select * from x) ss where o.ten = 1;
+ sum | sum
+------+------
+ 1700 | 5350
+(1 row)
+
+-- CTE Scan
+select workmem_filter('
+explain (costs off, work_mem on)
+WITH q1(x,y) AS (
+ SELECT hundred, sum(ten) FROM tenk1 GROUP BY hundred
+ )
+SELECT count(*) FROM q1 WHERE y > (SELECT sum(y)/100 FROM q1 qsub);
+');
+ workmem_filter
+--------------------------------------------------------------------
+ Aggregate
+ CTE q1
+ -> 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) (limit=4096 kB)
+ -> CTE Scan on q1 (work_mem=N kB) (limit=4096 kB)
+ Filter: ((y)::numeric > (InitPlan 2).col1)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 16384 kB
+(12 rows)
+
+WITH q1(x,y) AS (
+ SELECT hundred, sum(ten) FROM tenk1 GROUP BY hundred
+ )
+SELECT count(*) FROM q1 WHERE y > (SELECT sum(y)/100 FROM q1 qsub);
+ count
+-------
+ 50
+(1 row)
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+---------------------------------------------------------------------------------------
+ Limit
+ -> 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) (limit=4096 kB)
+ Total Working Memory Estimate: N kB
+ 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))
+limit 5;
+ sum
+---------
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+(5 rows)
+
+-- Bitmap Heap Scan
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select * from tenk1 a join tenk1 b on
+ (a.unique1 = 1 and b.unique1 = 2) or (a.unique2 = 3 and b.hundred = 4)
+);
+');
+ workmem_filter
+-------------------------------------------------------------------------------------------------------
+ Aggregate
+ -> Nested Loop
+ Join Filter: (((a.unique1 = 1) AND (b.unique1 = 2)) OR ((a.unique2 = 3) AND (b.hundred = 4)))
+ -> Bitmap Heap Scan on tenk1 b
+ Recheck Cond: ((hundred = 4) OR (unique1 = 2))
+ -> BitmapOr
+ -> 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) (limit=4096 kB)
+ Index Cond: (unique1 = 2)
+ -> 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) (limit=4096 kB)
+ Index Cond: (unique2 = 3)
+ -> Bitmap Index Scan on tenk1_unique1 (work_mem=N kB) (limit=4096 kB)
+ Index Cond: (unique1 = 1)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 20480 kB
+(20 rows)
+
+select count(*) from (
+select * from tenk1 a join tenk1 b on
+ (a.unique1 = 1 and b.unique1 = 2) or (a.unique2 = 3 and b.hundred = 4)
+);
+ count
+-------
+ 101
+(1 row)
+
+-- InitPlan with hash table ("IN SELECT")
+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) (limit=16384 kB)
+ SubPlan 1
+ -> Append
+ -> Result
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 16384 kB
+(7 rows)
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+ ?column?
+----------
+ f
+(1 row)
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+ workmem_filter
+----------------------------------------------------------------
+ Result (work_mem=N kB) (limit=16384 kB)
+ SubPlan 3
+ -> 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 Estimate: N kB
+ Total Working Memory Limit: 24576 kB
+(10 rows)
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+ ?column?
+----------
+ t
+(1 row)
+
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index fbffc67ae60..0d59d37bb35 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -123,7 +123,7 @@ test: plancache limit plpgsql copy2 temp domain rangefuncs prepare conversion tr
# The stats test resets stats, so nothing else needing stats access can be in
# this group.
# ----------
-test: partition_join partition_prune reloptions hash_part indexing partition_aggregate partition_info tuplesort explain compression compression_lz4 memoize stats predicate numa
+test: partition_join partition_prune reloptions hash_part indexing partition_aggregate partition_info tuplesort explain compression compression_lz4 memoize stats predicate numa workmem
# event_trigger depends on create_am and cannot run concurrently with
# any test that runs DDL
diff --git a/src/test/regress/sql/workmem.sql b/src/test/regress/sql/workmem.sql
new file mode 100644
index 00000000000..2de22be0427
--- /dev/null
+++ b/src/test/regress/sql/workmem.sql
@@ -0,0 +1,307 @@
+----
+-- Tests that show "work_mem" output to EXPLAIN plans.
+----
+
+-- Note: Function derived from file explain.sql. We can't use that other
+-- function, since we're run in parallel with explain.sql.
+create or replace function workmem_filter(text) returns setof text
+language plpgsql as
+$$
+declare
+ ln text;
+begin
+ for ln in execute $1
+ loop
+ -- Mask out work_mem estimate, since it might be brittle
+ ln := regexp_replace(ln, '\mwork_mem=\d+\M', 'work_mem=N', 'g');
+ ln := regexp_replace(ln, '\mMemory Estimate: \d+\M', 'Memory Estimate: N', 'g');
+ return next ln;
+ end loop;
+end;
+$$;
+
+-- Unique -> hash agg
+set enable_hashagg = on;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+');
+
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+
+reset enable_hashagg;
+
+-- Unique -> sort
+set enable_hashagg = off;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+');
+
+select *
+from onek
+where (unique1,ten) in (values (1,1), (20,0), (99,9), (17,99))
+order by unique1;
+
+reset enable_hashagg;
+
+-- Incremental Sort
+select workmem_filter('
+explain (costs off, work_mem on)
+select * from (select * from tenk1 order by four) t order by four, ten
+limit 1;
+');
+
+select * from (select * from tenk1 order by four) t order by four, ten
+limit 1;
+
+-- Hash Join
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select t1.unique1, t2.hundred
+from onek t1, tenk1 t2
+where exists (select 1 from tenk1 t3
+ where t3.thousand = t1.unique1 and t3.tenthous = t2.hundred)
+ and t1.unique1 < 1
+) t;
+');
+
+select count(*) from (
+select t1.unique1, t2.hundred
+from onek t1, tenk1 t2
+where exists (select 1 from tenk1 t3
+ where t3.thousand = t1.unique1 and t3.tenthous = t2.hundred)
+ and t1.unique1 < 1
+) t;
+
+-- Materialize
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select t1.f1
+from int4_tbl t1, int4_tbl t2
+ left join int4_tbl t3 on t3.f1 > 0
+ left join int4_tbl t4 on t3.f1 > 1
+where t4.f1 is null
+) t;
+');
+
+select count(*) from (
+select t1.f1
+from int4_tbl t1, int4_tbl t2
+ left join int4_tbl t3 on t3.f1 > 0
+ left join int4_tbl t4 on t3.f1 > 1
+where t4.f1 is null
+) t;
+
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+
+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
+group by grouping sets((a, b), (a));
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+
+reset enable_hashagg;
+
+-- Agg (hash, parallel)
+set parallel_setup_cost=0;
+set parallel_tuple_cost=0;
+set min_parallel_table_scan_size=0;
+set max_parallel_workers_per_gather=4;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+select length(stringu1) from tenk1 group by length(stringu1);
+');
+
+select length(stringu1) from tenk1 group by length(stringu1);
+
+reset parallel_setup_cost;
+reset parallel_tuple_cost;
+reset min_parallel_table_scan_size;
+reset max_parallel_workers_per_gather;
+
+-- Agg (simple) [no work_mem]
+explain (costs off, work_mem on)
+select MAX(length(stringu1)) from tenk1;
+
+select MAX(length(stringu1)) from tenk1;
+
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+
+-- Table Function Scan
+CREATE TABLE workmem_xmldata(data xml);
+
+select workmem_filter('
+EXPLAIN (COSTS OFF, work_mem on)
+SELECT xmltable.*
+ FROM (SELECT data FROM workmem_xmldata) x,
+ LATERAL XMLTABLE(''/ROWS/ROW''
+ PASSING data
+ COLUMNS id int PATH ''@id'',
+ _id FOR ORDINALITY,
+ country_name text PATH ''COUNTRY_NAME'' NOT NULL,
+ country_id text PATH ''COUNTRY_ID'',
+ region_id int PATH ''REGION_ID'',
+ size float PATH ''SIZE'',
+ unit text PATH ''SIZE/@unit'',
+ premier_name text PATH ''PREMIER_NAME'' DEFAULT ''not specified'');
+');
+
+SELECT xmltable.*
+ FROM (SELECT data FROM workmem_xmldata) x,
+ LATERAL XMLTABLE('/ROWS/ROW'
+ PASSING data
+ COLUMNS id int PATH '@id',
+ _id FOR ORDINALITY,
+ country_name text PATH 'COUNTRY_NAME' NOT NULL,
+ country_id text PATH 'COUNTRY_ID',
+ region_id int PATH 'REGION_ID',
+ size float PATH 'SIZE',
+ unit text PATH 'SIZE/@unit',
+ premier_name text PATH 'PREMIER_NAME' DEFAULT 'not specified');
+
+drop table workmem_xmldata;
+
+-- SetOp [no work_mem]
+explain (costs off, work_mem on)
+select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
+
+select unique1 from tenk1 except select unique2 from tenk1 where unique2 != 10;
+
+-- HashSetOp
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from
+ ( select unique1 from tenk1 intersect select fivethous from tenk1 ) ss;
+');
+
+select count(*) from
+ ( select unique1 from tenk1 intersect select fivethous from tenk1 ) ss;
+
+-- RecursiveUnion and Memoize (also WorkTable Scan [no work_mem])
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(o.four), sum(ss.a) from onek o
+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;
+');
+
+select sum(o.four), sum(ss.a) from onek o
+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;
+
+-- CTE Scan
+select workmem_filter('
+explain (costs off, work_mem on)
+WITH q1(x,y) AS (
+ SELECT hundred, sum(ten) FROM tenk1 GROUP BY hundred
+ )
+SELECT count(*) FROM q1 WHERE y > (SELECT sum(y)/100 FROM q1 qsub);
+');
+
+WITH q1(x,y) AS (
+ SELECT hundred, sum(ten) FROM tenk1 GROUP BY hundred
+ )
+SELECT count(*) FROM q1 WHERE y > (SELECT sum(y)/100 FROM q1 qsub);
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+');
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+
+-- Bitmap Heap Scan
+select workmem_filter('
+explain (costs off, work_mem on)
+select count(*) from (
+select * from tenk1 a join tenk1 b on
+ (a.unique1 = 1 and b.unique1 = 2) or (a.unique2 = 3 and b.hundred = 4)
+);
+');
+
+select count(*) from (
+select * from tenk1 a join tenk1 b on
+ (a.unique1 = 1 and b.unique1 = 2) or (a.unique2 = 3 and b.hundred = 4)
+);
+
+-- InitPlan with hash table ("IN SELECT")
+select workmem_filter('
+explain (costs off, work_mem on)
+select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
+');
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+
+select 1 = any (select (select 1) where 1 = any (select 1));
--
2.39.5
0004-Add-workmem_hook-to-allow-extensions-to-override-per.patch (text/x-diff, 48.4K) download
From f0e5f69ac14927bb057bd765a48af4de2023b7bd Mon Sep 17 00:00:00 2001
From: James Hunter <james.hunter.pg@gmail.com>
Date: Wed, 5 Mar 2025 01:21:20 +0000
Subject: [PATCH 4/4] Add "workmem_hook" to allow extensions to override
per-node work_mem
---
contrib/Makefile | 3 +-
contrib/meson.build | 1 +
contrib/workmem/Makefile | 20 +
contrib/workmem/expected/workmem.out | 676 +++++++++++++++++++++++++++
contrib/workmem/meson.build | 28 ++
contrib/workmem/sql/workmem.sql | 304 ++++++++++++
contrib/workmem/workmem.c | 409 ++++++++++++++++
src/backend/executor/execWorkmem.c | 40 +-
src/include/executor/executor.h | 4 +
9 files changed, 1474 insertions(+), 11 deletions(-)
create mode 100644 contrib/workmem/Makefile
create mode 100644 contrib/workmem/expected/workmem.out
create mode 100644 contrib/workmem/meson.build
create mode 100644 contrib/workmem/sql/workmem.sql
create mode 100644 contrib/workmem/workmem.c
diff --git a/contrib/Makefile b/contrib/Makefile
index 2f0a88d3f77..042f5128376 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -51,7 +51,8 @@ SUBDIRS = \
tsm_system_rows \
tsm_system_time \
unaccent \
- vacuumlo
+ vacuumlo \
+ workmem
ifeq ($(with_ssl),openssl)
SUBDIRS += pgcrypto sslinfo
diff --git a/contrib/meson.build b/contrib/meson.build
index ed30ee7d639..a1b283789b5 100644
--- a/contrib/meson.build
+++ b/contrib/meson.build
@@ -70,4 +70,5 @@ subdir('tsm_system_time')
subdir('unaccent')
subdir('uuid-ossp')
subdir('vacuumlo')
+subdir('workmem')
subdir('xml2')
diff --git a/contrib/workmem/Makefile b/contrib/workmem/Makefile
new file mode 100644
index 00000000000..f920cdb9964
--- /dev/null
+++ b/contrib/workmem/Makefile
@@ -0,0 +1,20 @@
+# contrib/workmem/Makefile
+
+MODULE_big = workmem
+OBJS = \
+ $(WIN32RES) \
+ workmem.o
+PGFILEDESC = "workmem - extension that adjusts PostgreSQL work_mem per node"
+
+REGRESS = workmem
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/workmem
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/workmem/expected/workmem.out b/contrib/workmem/expected/workmem.out
new file mode 100644
index 00000000000..f69883b0005
--- /dev/null
+++ b/contrib/workmem/expected/workmem.out
@@ -0,0 +1,676 @@
+load 'workmem';
+-- Note: Function derived from file explain.sql. We can't use that other
+-- function, since we're run in parallel with explain.sql.
+create or replace function workmem_filter(text) returns setof text
+language plpgsql as
+$$
+declare
+ ln text;
+begin
+ for ln in execute $1
+ loop
+ -- Mask out work_mem estimate, since it might be brittle
+ ln := regexp_replace(ln, '\mwork_mem=\d+\M', 'work_mem=N', 'g');
+ ln := regexp_replace(ln, '\mMemory Estimate: \d+\M', 'Memory Estimate: N', 'g');
+ return next ln;
+ end loop;
+end;
+$$;
+--====
+-- Test suite 1: default workmem.query_work_mem (= 100 MB)
+--====
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=25600 kB)
+ -> Sort (work_mem=N kB) (limit=25600 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> HashAggregate (work_mem=N kB) (limit=51200 kB)
+ Hash Key: "*VALUES*".column1, "*VALUES*".column2
+ Hash Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 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
+group by grouping sets((a, b), (a));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+(4 rows)
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=20480 kB)
+ -> Sort (work_mem=N kB) (limit=20480 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> GroupAggregate (work_mem=N kB) (limit=40960 kB)
+ Group Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1
+ Sort Key: "*VALUES*".column2
+ Group Key: "*VALUES*".column2
+ Sort Key: "*VALUES*".column3
+ Group Key: "*VALUES*".column3
+ Sort Key: "*VALUES*".column4
+ Group Key: "*VALUES*".column4
+ -> Sort (work_mem=N kB) (limit=20480 kB)
+ Sort Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 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
+group by grouping sets((a, b), (a), (b), (c), (d));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+ | | 5
+ | | 6
+ | | 7
+ | | 8
+ | 1 | 9
+ | 2 | 10
+(10 rows)
+
+reset enable_hashagg;
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+ workmem_filter
+-----------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=102400 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 kB
+(4 rows)
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+ count
+-------
+ 2000
+(1 row)
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+ workmem_filter
+---------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series (work_mem=N kB) (limit=102399 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102399 kB
+(4 rows)
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+ count
+-------
+ 12
+(1 row)
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+----------------------------------------------------------------------------------------
+ Limit
+ -> WindowAgg (work_mem=N kB) (limit=34134 kB)
+ -> Sort (work_mem=N kB) (limit=34133 kB)
+ Sort Key: ((a.n < 3))
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=34133 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 kB
+(7 rows)
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+ sum
+---------
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+(5 rows)
+
+-- InitPlan with hash table ("IN SELECT")
+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) (limit=102400 kB)
+ SubPlan 1
+ -> Append
+ -> Result
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 kB
+(7 rows)
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+ ?column?
+----------
+ f
+(1 row)
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+ workmem_filter
+----------------------------------------------------------------
+ Result (work_mem=N kB) (limit=68267 kB)
+ SubPlan 3
+ -> Result (work_mem=N kB) (limit=34133 kB)
+ One-Time Filter: (ANY (1 = (hashed SubPlan 2).col1))
+ InitPlan 1
+ -> Result
+ SubPlan 2
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 102400 kB
+(10 rows)
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+ ?column?
+----------
+ t
+(1 row)
+
+--====
+-- Test suite 2: set workmem.query_work_mem to 4 MB
+--====
+set workmem.query_work_mem = 4096;
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=1024 kB)
+ -> Sort (work_mem=N kB) (limit=1024 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> HashAggregate (work_mem=N kB) (limit=2048 kB)
+ Hash Key: "*VALUES*".column1, "*VALUES*".column2
+ Hash Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 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
+group by grouping sets((a, b), (a));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+(4 rows)
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=820 kB)
+ -> Sort (work_mem=N kB) (limit=819 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> GroupAggregate (work_mem=N kB) (limit=1638 kB)
+ Group Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1
+ Sort Key: "*VALUES*".column2
+ Group Key: "*VALUES*".column2
+ Sort Key: "*VALUES*".column3
+ Group Key: "*VALUES*".column3
+ Sort Key: "*VALUES*".column4
+ Group Key: "*VALUES*".column4
+ -> Sort (work_mem=N kB) (limit=819 kB)
+ Sort Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 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
+group by grouping sets((a, b), (a), (b), (c), (d));
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+ | | 5
+ | | 6
+ | | 7
+ | | 8
+ | 1 | 9
+ | 2 | 10
+(10 rows)
+
+reset enable_hashagg;
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+ workmem_filter
+---------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=4096 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(4 rows)
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+ count
+-------
+ 2000
+(1 row)
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+ workmem_filter
+-------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series (work_mem=N kB) (limit=4095 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4095 kB
+(4 rows)
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+ count
+-------
+ 12
+(1 row)
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+---------------------------------------------------------------------------------------
+ Limit
+ -> WindowAgg (work_mem=N kB) (limit=1366 kB)
+ -> Sort (work_mem=N kB) (limit=1365 kB)
+ Sort Key: ((a.n < 3))
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=1365 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(7 rows)
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+ sum
+---------
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+(5 rows)
+
+-- InitPlan with hash table ("IN SELECT")
+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) (limit=4096 kB)
+ SubPlan 1
+ -> Append
+ -> Result
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(7 rows)
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+ ?column?
+----------
+ f
+(1 row)
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+ workmem_filter
+----------------------------------------------------------------
+ Result (work_mem=N kB) (limit=2731 kB)
+ SubPlan 3
+ -> Result (work_mem=N kB) (limit=1365 kB)
+ One-Time Filter: (ANY (1 = (hashed SubPlan 2).col1))
+ InitPlan 1
+ -> Result
+ SubPlan 2
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 4096 kB
+(10 rows)
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+ ?column?
+----------
+ t
+(1 row)
+
+reset workmem.query_work_mem;
+--====
+-- Test suite 3: set workmem.query_work_mem to 80 KB
+--====
+set workmem.query_work_mem = 80;
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=20 kB)
+ -> Sort (work_mem=N kB) (limit=20 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> HashAggregate (work_mem=N kB) (limit=40 kB)
+ Hash Key: "*VALUES*".column1, "*VALUES*".column2
+ Hash Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 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
+group by grouping sets((a, b), (a));
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+(4 rows)
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+----------------------------------------------------------------------
+ WindowAgg (work_mem=N kB) (limit=16 kB)
+ -> Sort (work_mem=N kB) (limit=16 kB)
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2 NULLS FIRST
+ -> GroupAggregate (work_mem=N kB) (limit=32 kB)
+ Group Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1
+ Sort Key: "*VALUES*".column2
+ Group Key: "*VALUES*".column2
+ Sort Key: "*VALUES*".column3
+ Group Key: "*VALUES*".column3
+ Sort Key: "*VALUES*".column4
+ Group Key: "*VALUES*".column4
+ -> Sort (work_mem=N kB) (limit=16 kB)
+ Sort Key: "*VALUES*".column1
+ -> Values Scan on "*VALUES*"
+ Filter: (column1 = column2)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 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
+group by grouping sets((a, b), (a), (b), (c), (d));
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ a | b | row_number
+---+---+------------
+ 1 | | 1
+ 1 | 1 | 2
+ 2 | | 3
+ 2 | 2 | 4
+ | | 5
+ | | 6
+ | | 7
+ | | 8
+ | 1 | 9
+ | 2 | 10
+(10 rows)
+
+reset enable_hashagg;
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+ workmem_filter
+-------------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=80 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 kB
+(4 rows)
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+ count
+-------
+ 2000
+(1 row)
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+-----------------------------------------------------------------------
+ Aggregate
+ -> Function Scan on generate_series (work_mem=N kB) (limit=78 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 78 kB
+(4 rows)
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ count
+-------
+ 12
+(1 row)
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+-------------------------------------------------------------------------------------
+ Limit
+ -> WindowAgg (work_mem=N kB) (limit=27 kB)
+ -> Sort (work_mem=N kB) (limit=27 kB)
+ Sort Key: ((a.n < 3))
+ -> Function Scan on generate_series a (work_mem=N kB) (limit=26 kB)
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 kB
+(7 rows)
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ sum
+---------
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+ 2000997
+(5 rows)
+
+-- InitPlan with hash table ("IN SELECT")
+select workmem_filter('
+explain (costs off, work_mem on)
+select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+---------------------------------------
+ Result (work_mem=N kB) (limit=80 kB)
+ SubPlan 1
+ -> Append
+ -> Result
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 kB
+(7 rows)
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ ?column?
+----------
+ f
+(1 row)
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ workmem_filter
+----------------------------------------------------------------
+ Result (work_mem=N kB) (limit=54 kB)
+ SubPlan 3
+ -> Result (work_mem=N kB) (limit=26 kB)
+ One-Time Filter: (ANY (1 = (hashed SubPlan 2).col1))
+ InitPlan 1
+ -> Result
+ SubPlan 2
+ -> Result
+ Total Working Memory Estimate: N kB
+ Total Working Memory Limit: 80 kB
+(10 rows)
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+WARNING: not enough working memory for query: increase workmem.query_work_mem
+ ?column?
+----------
+ t
+(1 row)
+
+reset workmem.query_work_mem;
diff --git a/contrib/workmem/meson.build b/contrib/workmem/meson.build
new file mode 100644
index 00000000000..fce8030ba45
--- /dev/null
+++ b/contrib/workmem/meson.build
@@ -0,0 +1,28 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+workmem_sources = files(
+ 'workmem.c',
+)
+
+if host_system == 'windows'
+ workmem_sources += rc_lib_gen.process(win32ver_rc, extra_args: [
+ '--NAME', 'workmem',
+ '--FILEDESC', 'workmem - extension that adjusts PostgreSQL work_mem per node',])
+endif
+
+workmem = shared_module('workmem',
+ workmem_sources,
+ kwargs: contrib_mod_args,
+)
+contrib_targets += workmem
+
+tests += {
+ 'name': 'workmem',
+ 'sd': meson.current_source_dir(),
+ 'bd': meson.current_build_dir(),
+ 'regress': {
+ 'sql': [
+ 'workmem',
+ ],
+ },
+}
diff --git a/contrib/workmem/sql/workmem.sql b/contrib/workmem/sql/workmem.sql
new file mode 100644
index 00000000000..4e1ec056b80
--- /dev/null
+++ b/contrib/workmem/sql/workmem.sql
@@ -0,0 +1,304 @@
+load 'workmem';
+
+-- Note: Function derived from file explain.sql. We can't use that other
+-- function, since we're run in parallel with explain.sql.
+create or replace function workmem_filter(text) returns setof text
+language plpgsql as
+$$
+declare
+ ln text;
+begin
+ for ln in execute $1
+ loop
+ -- Mask out work_mem estimate, since it might be brittle
+ ln := regexp_replace(ln, '\mwork_mem=\d+\M', 'work_mem=N', 'g');
+ ln := regexp_replace(ln, '\mMemory Estimate: \d+\M', 'Memory Estimate: N', 'g');
+ return next ln;
+ end loop;
+end;
+$$;
+
+--====
+-- Test suite 1: default workmem.query_work_mem (= 100 MB)
+--====
+
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+
+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
+group by grouping sets((a, b), (a));
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+
+reset enable_hashagg;
+
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+');
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+
+-- InitPlan with hash table ("IN SELECT")
+select workmem_filter('
+explain (costs off, work_mem on)
+select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
+');
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+
+--====
+-- Test suite 2: set workmem.query_work_mem to 4 MB
+--====
+set workmem.query_work_mem = 4096;
+
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+
+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
+group by grouping sets((a, b), (a));
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+
+reset enable_hashagg;
+
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+');
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+
+-- InitPlan with hash table ("IN SELECT")
+select workmem_filter('
+explain (costs off, work_mem on)
+select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
+');
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+
+reset workmem.query_work_mem;
+
+--====
+-- Test suite 3: set workmem.query_work_mem to 80 KB
+--====
+set workmem.query_work_mem = 80;
+
+----
+-- Some tests from src/test/regress/sql/workmem.sql that don't require
+-- test_setup.sql, etc., to be run first.
+----
+
+-- Grouping Sets (Hash)
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a));
+');
+
+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
+group by grouping sets((a, b), (a));
+
+-- Grouping Sets (Sort)
+set enable_hashagg = off;
+
+select workmem_filter('
+explain (costs off, work_mem on)
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+');
+
+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
+group by grouping sets((a, b), (a), (b), (c), (d));
+
+reset enable_hashagg;
+
+-- Function Scan
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+');
+
+select count(*) from (
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+) t;
+
+-- Three Function Scans
+select workmem_filter('
+explain (work_mem on, costs off)
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+');
+
+select count(*)
+from rows from(generate_series(1, 5),
+ generate_series(2, 10),
+ generate_series(4, 15));
+
+-- WindowAgg
+select workmem_filter('
+explain (costs off, work_mem on)
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+');
+
+select sum(n) over(partition by m)
+from (SELECT n < 3 as m, n from generate_series(1,2000) a(n))
+limit 5;
+
+-- InitPlan with hash table ("IN SELECT")
+select workmem_filter('
+explain (costs off, work_mem on)
+select ''foo''::text in (select ''bar''::name union all select ''bar''::name);
+');
+
+select 'foo'::text in (select 'bar'::name union all select 'bar'::name);
+
+-- SubPlan with hash table
+select workmem_filter('
+explain (costs off, work_mem on)
+select 1 = any (select (select 1) where 1 = any (select 1));
+');
+
+select 1 = any (select (select 1) where 1 = any (select 1));
+
+reset workmem.query_work_mem;
diff --git a/contrib/workmem/workmem.c b/contrib/workmem/workmem.c
new file mode 100644
index 00000000000..d78f60c7d8d
--- /dev/null
+++ b/contrib/workmem/workmem.c
@@ -0,0 +1,409 @@
+/*-------------------------------------------------------------------------
+ *
+ * workmem.c
+ *
+ *
+ * Copyright (c) 2025, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * contrib/workmem/workmem.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/parallel.h"
+#include "common/int.h"
+#include "executor/executor.h"
+#include "miscadmin.h"
+#include "utils/guc.h"
+
+PG_MODULE_MAGIC;
+
+/* Local variables */
+
+/*
+ * A Target represents a collection of data structures, belonging to an
+ * execution node, that all share the same memory limit.
+ *
+ * For example, in parallel query, every parallel worker (plus the leader)
+ * gets a copy of the execution node, and therefore a copy of all of that
+ * node's work_mem limits. In this case, we'll track a single Target, but its
+ * count will include (1 + num_workers), because this Target gets "applied"
+ * to (1 + num_workers) execution nodes.
+ */
+typedef struct Target
+{
+ /* # of data structures to which target applies: */
+ int count;
+ /* workmem estimate for each of these data structures: */
+ int workmem;
+ /* (original) workmem limit for each of these data structures: */
+ int limit;
+ /* workmem estimate, but capped at (original) workmem limit: */
+ int priority;
+ /* ratio of (priority / limit); measure's Target's "greediness": */
+ double ratio;
+ /* link to target's actual limit, so we can set it: */
+ int *target_limit;
+} Target;
+
+typedef struct WorkMemStats
+{
+ /* total # of data structures that get working memory: */
+ uint64 count;
+ /* total working memory estimated for this query: */
+ uint64 workmem;
+ /* total working memory (currently) reserved for this query: */
+ uint64 limit;
+ /* total "capped" working memory estimate: */
+ uint64 priority;
+ /* list of Targets, used to update actual workmem limits: */
+ List *targets;
+} WorkMemStats;
+
+/* GUC variables */
+static int workmem_query_work_mem = 100 * 1024; /* kB */
+
+/* internal functions */
+static void workmem_fn(PlannedStmt *plannedstmt);
+
+static int clamp_priority(int workmem, int limit);
+static Target * make_target(int workmem, int *target_limit, int count);
+static void add_target(WorkMemStats * workmem_stats, Target * target);
+
+/* Sort comparators: sort by ratio, ascending or descending. */
+static int target_compare_asc(const ListCell *a, const ListCell *b);
+static int target_compare_desc(const ListCell *a, const ListCell *b);
+
+/*
+ * Module load callback
+ */
+void
+_PG_init(void)
+{
+ /* Define custom GUC variable. */
+ DefineCustomIntVariable("workmem.query_work_mem",
+ "Amount of working-memory (in kB) to provide each "
+ "query.",
+ NULL,
+ &workmem_query_work_mem,
+ 100 * 1024, /* default to 100 MB */
+ 64,
+ INT_MAX,
+ PGC_USERSET,
+ GUC_UNIT_KB,
+ NULL,
+ NULL,
+ NULL);
+
+ MarkGUCPrefixReserved("workmem");
+
+ /* Install hooks. */
+ ExecAssignWorkMem_hook = workmem_fn;
+}
+
+static void
+workmem_analyze(PlannedStmt *plannedstmt, WorkMemStats * workmem_stats)
+{
+ int idx;
+
+ for (idx = 0; idx < list_length(plannedstmt->workMemCategories); ++idx)
+ {
+ WorkMemCategory category;
+ int count;
+ int estimate;
+ ListCell *limit_cell;
+ int limit;
+ Target *target;
+
+ category =
+ (WorkMemCategory) list_nth_int(plannedstmt->workMemCategories, idx);
+ count = list_nth_int(plannedstmt->workMemCounts, idx);
+ estimate = list_nth_int(plannedstmt->workMemEstimates, idx);
+
+ limit = category == WORKMEM_HASH ?
+ get_hash_memory_limit() / 1024 : work_mem;
+ limit_cell = list_nth_cell(plannedstmt->workMemLimits, idx);
+ lfirst_int(limit_cell) = limit;
+
+ target = make_target(estimate, &lfirst_int(limit_cell), count);
+ add_target(workmem_stats, target);
+ }
+}
+
+static void
+workmem_set(PlannedStmt *plannedstmt, WorkMemStats * workmem_stats)
+{
+ int remaining = workmem_query_work_mem;
+
+ if (workmem_stats->limit <= remaining)
+ {
+ /*
+ * "High memory" case: we have more than enough query_work_mem; now
+ * hand out the excess.
+ */
+
+ /* This is memory that exceeds workmem limits. */
+ remaining -= workmem_stats->limit;
+
+ /*
+ * Sort targets from highest ratio to lowest. When we assign memory to
+ * a Target, we'll truncate fractional KB; so by going through the
+ * list from highest to lowest ratio, we ensure that the lowest ratios
+ * get the leftover fractional KBs.
+ */
+ list_sort(workmem_stats->targets, target_compare_desc);
+
+ foreach_ptr(Target, target, workmem_stats->targets)
+ {
+ double fraction;
+ int extra_workmem;
+
+ /* How much extra work mem should we assign to this target? */
+ fraction = (double) target->workmem / workmem_stats->workmem;
+
+ /* NOTE: This is extra workmem *per data structure*. */
+ extra_workmem = (int) (fraction * remaining);
+
+ *target->target_limit += extra_workmem;
+
+ /* OK, we've handled this target. */
+ workmem_stats->workmem -= (target->workmem * target->count);
+ remaining -= (extra_workmem * target->count);
+ }
+ }
+ else if (workmem_stats->priority <= remaining)
+ {
+ /*
+ * "Medium memory" case: we don't have enough query_work_mem to give
+ * every target its full allotment, but we do have enough to give it
+ * as much as (we estimate) it needs.
+ *
+ * So, just take some memory away from nodes that (we estimate) won't
+ * need it.
+ */
+
+ /* This is memory that exceeds workmem estimates. */
+ remaining -= workmem_stats->priority;
+
+ /*
+ * Sort targets from highest ratio to lowest. We'll skip any Target
+ * with ratio > 1.0, because (we estimate) they already need their
+ * full allotment. Also, once a target reaches its workmem limit,
+ * we'll stop giving it more workmem, leaving the surplus memory to be
+ * assigned to targets with smaller ratios.
+ */
+ list_sort(workmem_stats->targets, target_compare_desc);
+
+ foreach_ptr(Target, target, workmem_stats->targets)
+ {
+ double fraction;
+ int extra_workmem;
+
+ /* How much extra work mem should we assign to this target? */
+ fraction = (double) target->workmem / workmem_stats->workmem;
+
+ /*
+ * Don't give the target more than its (original) limit.
+ *
+ * NOTE: This is extra workmem *per data structure*.
+ */
+ extra_workmem = Min((int) (fraction * remaining),
+ target->limit - target->priority);
+
+ *target->target_limit = target->priority + extra_workmem;
+
+ /* OK, we've handled this target. */
+ workmem_stats->workmem -= (target->workmem * target->count);
+ remaining -= (extra_workmem * target->count);
+ }
+ }
+ else
+ {
+ uint64 limit = workmem_stats->limit;
+
+ /*
+ * "Low memory" case: we are severely memory constrained, and need to
+ * take "priority" memory away from targets that (we estimate)
+ * actually need it. We'll do this by (effectively) reducing the
+ * global "work_mem" limit, uniformly, for all targets, until we're
+ * under the query_work_mem limit.
+ */
+ elog(WARNING,
+ "not enough working memory for query: increase "
+ "workmem.query_work_mem");
+
+ /*
+ * Sort targets from lowest ratio to highest. For any target whose
+ * ratio is < the target_ratio, we'll just assign it its priority (=
+ * workmem) as limit, and return the excess workmem to our "limit",
+ * for use by subsequent, greedier, targets.
+ */
+ list_sort(workmem_stats->targets, target_compare_asc);
+
+ foreach_ptr(Target, target, workmem_stats->targets)
+ {
+ double target_ratio;
+ int target_limit;
+
+ /*
+ * If we restrict our targets to this ratio, we'll stay within the
+ * query_work_mem limit.
+ */
+ target_ratio = (double) remaining / limit;
+
+ /*
+ * Don't give this target more than its priority request (but we
+ * might give it less).
+ */
+ target_limit = Min(target->priority,
+ target_ratio * target->limit);
+ *target->target_limit = target_limit;
+
+ /* "Remaining" decreases by memory we actually assigned. */
+ remaining -= (target_limit * target->count);
+
+ /*
+ * "Limit" decreases by target's original memory limit.
+ *
+ * If target_limit <= target->priority, so we restricted this
+ * target to less memory than (we estimate) it needs, then the
+ * target_ratio will stay the same, since, letting A = remaining,
+ * B = limit, and R = ratio, we'll have:
+ *
+ * R=A/B <=> A=R*B <=> A-R*X = R*B - R*X <=> A-R*X = R * (B-X) <=>
+ * R = (A-R*X) / (B-X)
+ *
+ * -- which is what we wanted to prove.
+ *
+ * And if target_limit > target->priority, so we didn't need to
+ * restrict this target beyond its priority estimate, then the
+ * target_ratio will increase. This means more memory for the
+ * remaining, greedier, targets.
+ */
+ limit -= (target->limit * target->count);
+
+ target_ratio = (double) remaining / limit;
+ }
+ }
+}
+
+/*
+ * workmem_fn: updates the query plan's work_mem based on query_work_mem
+ */
+static void
+workmem_fn(PlannedStmt *plannedstmt)
+{
+ WorkMemStats workmem_stats;
+ MemoryContext context,
+ oldcontext;
+
+ /*
+ * We already assigned working-memory limits on the leader, and those
+ * limits were sent to the workers inside the serialized Plan.
+ *
+ * We could re-assign working-memory limits on the parallel worker, to
+ * only those Plan nodes that got sent to the worker, but for now we don't
+ * bother.
+ */
+ if (IsParallelWorker())
+ return;
+
+ if (workmem_query_work_mem == -1)
+ return; /* disabled */
+
+ memset(&workmem_stats, 0, sizeof(workmem_stats));
+
+ /*
+ * Set up our own memory context, so we can drop the metadata we generate,
+ * all at once.
+ */
+ context = AllocSetContextCreate(CurrentMemoryContext,
+ "workmem_fn context",
+ ALLOCSET_DEFAULT_SIZES);
+
+ oldcontext = MemoryContextSwitchTo(context);
+
+ /* Figure out how much total working memory this query wants/needs. */
+ workmem_analyze(plannedstmt, &workmem_stats);
+
+ /* Now restrict the query to workmem.query_work_mem. */
+ workmem_set(plannedstmt, &workmem_stats);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Drop all our metadata. */
+ MemoryContextDelete(context);
+}
+
+static int
+clamp_priority(int workmem, int limit)
+{
+ return Min(workmem, limit);
+}
+
+static Target *
+make_target(int workmem, int *target_limit, int count)
+{
+ Target *result = palloc_object(Target);
+
+ result->count = count;
+ result->workmem = workmem;
+ result->limit = *target_limit;
+ result->priority = clamp_priority(result->workmem, result->limit);
+ result->ratio = (double) result->priority / result->limit;
+ result->target_limit = target_limit;
+
+ return result;
+}
+
+static void
+add_target(WorkMemStats * workmem_stats, Target * target)
+{
+ workmem_stats->count += target->count;
+ workmem_stats->workmem += target->count * target->workmem;
+ workmem_stats->limit += target->count * target->limit;
+ workmem_stats->priority += target->count * target->priority;
+ workmem_stats->targets = lappend(workmem_stats->targets, target);
+}
+
+/* This "ascending" comparator sorts least-greedy Targets first. */
+static int
+target_compare_asc(const ListCell *a, const ListCell *b)
+{
+ double a_val = ((Target *) a->ptr_value)->ratio;
+ double b_val = ((Target *) b->ptr_value)->ratio;
+
+ /*
+ * Sort in ascending order: smallest ratio first, then (if ratios equal)
+ * smallest workmem.
+ */
+ if (a_val == b_val)
+ {
+ return pg_cmp_s32(((Target *) a->ptr_value)->workmem,
+ ((Target *) b->ptr_value)->workmem);
+ }
+ else
+ return a_val > b_val ? 1 : -1;
+}
+
+/* This "descending" comparator sorts most-greedy Targets first. */
+static int
+target_compare_desc(const ListCell *a, const ListCell *b)
+{
+ double a_val = ((Target *) a->ptr_value)->ratio;
+ double b_val = ((Target *) b->ptr_value)->ratio;
+
+ /*
+ * Sort in descending order: largest ratio first, then (if ratios equal)
+ * largest workmem.
+ */
+ if (a_val == b_val)
+ {
+ return pg_cmp_s32(((Target *) b->ptr_value)->workmem,
+ ((Target *) a->ptr_value)->workmem);
+ }
+ else
+ return b_val > a_val ? 1 : -1;
+}
diff --git a/src/backend/executor/execWorkmem.c b/src/backend/executor/execWorkmem.c
index d8a19a58ebe..37420666065 100644
--- a/src/backend/executor/execWorkmem.c
+++ b/src/backend/executor/execWorkmem.c
@@ -52,6 +52,10 @@
#include "nodes/plannodes.h"
+/* Hook for plugins to get control in ExecAssignWorkMem */
+ExecAssignWorkMem_hook_type ExecAssignWorkMem_hook = NULL;
+
+
/* ------------------------------------------------------------------------
* ExecAssignWorkMem
*
@@ -64,20 +68,36 @@
*/
void
ExecAssignWorkMem(PlannedStmt *plannedstmt)
+{
+ if (ExecAssignWorkMem_hook)
+ (*ExecAssignWorkMem_hook) (plannedstmt);
+ else
+ {
+ /*
+ * 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.
+ *
+ * We bail out here, in case the hook wants to re-assign memory on
+ * parallel workers, and maybe wants to call
+ * standard_ExecAssignWorkMem() first, as well.
+ */
+ if (IsParallelWorker())
+ return;
+
+ standard_ExecAssignWorkMem(plannedstmt);
+ }
+}
+
+void
+standard_ExecAssignWorkMem(PlannedStmt *plannedstmt)
{
ListCell *lc_category;
ListCell *lc_limit;
- /*
- * 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;
-
forboth(lc_category, plannedstmt->workMemCategories,
lc_limit, plannedstmt->workMemLimits)
{
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 6008e3bc63c..7a34fa47489 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -96,6 +96,9 @@ typedef bool (*ExecutorCheckPerms_hook_type) (List *rangeTable,
bool ereport_on_violation);
extern PGDLLIMPORT ExecutorCheckPerms_hook_type ExecutorCheckPerms_hook;
+/* Hook for plugins to get control in ExecAssignWorkMem() */
+typedef void (*ExecAssignWorkMem_hook_type) (PlannedStmt *plannedstmt);
+extern PGDLLIMPORT ExecAssignWorkMem_hook_type ExecAssignWorkMem_hook;
/*
* prototypes from functions in execAmi.c
@@ -802,5 +805,6 @@ extern ResultRelInfo *ExecLookupResultRelByOid(ModifyTableState *node,
* prototypes from functions in execWorkmem.c
*/
extern void ExecAssignWorkMem(PlannedStmt *plannedstmt);
+extern void standard_ExecAssignWorkMem(PlannedStmt *plannedstmt);
#endif /* EXECUTOR_H */
--
2.39.5
view thread (29+ messages) latest in thread
Message-ID: <202508051215.x5niwfkbomdf@alvherre.pgsql>
Permalink: ../202508051215.x5niwfkbomdf@alvherre.pgsql/
Also on: postgresql.org/message-id/202508051215.x5niwfkbomdf@alvherre.pgsql
copy link · copy postgr.es · copy git am
reply Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: alvherre@kurilemu.de, james.hunter.pg@gmail.com, pgsql@j-davis.com, pgsql-hackers@lists.postgresql.org
Subject: Re: Proposal: "query_work_mem" GUC, to distribute working memory to the query's individual operators
In-Reply-To: <202508051215.x5niwfkbomdf@alvherre.pgsql>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox