🌐 AI搜索 & 代理 主页
Skip to content

Commit 9bdb300

Browse files
david-rowleydavid-rowley
committed
Fix EXPLAIN ANALYZE for parallel HashAgg plans
Since 1f39bce, HashAgg nodes have had the ability to spill to disk when memory consumption exceeds work_mem. That commit added new properties to EXPLAIN ANALYZE to show the maximum memory usage and disk usage, however, it didn't quite go as far as showing that information for parallel workers. Since workers may have experienced something very different from the main process, we should show this information per worker, as is done in Sort. Reviewed-by: Justin Pryzby Reviewed-by: Jeff Davis Discussion: https://postgr.es/m/CAApHDvpEKbfZa18mM1TD7qV6PG+w97pwCWq5tVD0dX7e11gRJw@mail.gmail.com Backpatch-through: 13, where the hashagg spilling code was added.
1 parent f219167 commit 9bdb300

File tree

5 files changed

+244
-17
lines changed

5 files changed

+244
-17
lines changed

‎src/backend/commands/explain.c‎

Lines changed: 96 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -3051,29 +3051,111 @@ show_hashagg_info(AggState *aggstate, ExplainState *es)
30513051
Agg *agg = (Agg *) aggstate->ss.ps.plan;
30523052
int64 memPeakKb = (aggstate->hash_mem_peak + 1023) / 1024;
30533053

3054-
Assert(IsA(aggstate, AggState));
3055-
30563054
if (agg->aggstrategy != AGG_HASHED &&
30573055
agg->aggstrategy != AGG_MIXED)
30583056
return;
30593057

3060-
if (es->costs && aggstate->hash_planned_partitions > 0)
3058+
if (es->format != EXPLAIN_FORMAT_TEXT)
30613059
{
3062-
ExplainPropertyInteger("Planned Partitions", NULL,
3063-
aggstate->hash_planned_partitions, es);
3060+
3061+
if (es->costs && aggstate->hash_planned_partitions > 0)
3062+
{
3063+
ExplainPropertyInteger("Planned Partitions", NULL,
3064+
aggstate->hash_planned_partitions, es);
3065+
}
3066+
3067+
if (!es->analyze)
3068+
return;
3069+
3070+
/* EXPLAIN ANALYZE */
3071+
ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb, es);
3072+
if (aggstate->hash_batches_used > 0)
3073+
{
3074+
ExplainPropertyInteger("Disk Usage", "kB",
3075+
aggstate->hash_disk_used, es);
3076+
ExplainPropertyInteger("HashAgg Batches", NULL,
3077+
aggstate->hash_batches_used, es);
3078+
}
30643079
}
3080+
else
3081+
{
3082+
bool gotone = false;
30653083

3066-
if (!es->analyze)
3067-
return;
3084+
if (es->costs && aggstate->hash_planned_partitions > 0)
3085+
{
3086+
ExplainIndentText(es);
3087+
appendStringInfo(es->str, "Planned Partitions: %d",
3088+
aggstate->hash_planned_partitions);
3089+
gotone = true;
3090+
}
3091+
3092+
if (!es->analyze)
3093+
{
3094+
if (gotone)
3095+
appendStringInfoChar(es->str, '\n');
3096+
return;
3097+
}
3098+
3099+
if (!gotone)
3100+
ExplainIndentText(es);
3101+
else
3102+
appendStringInfoString(es->str, " ");
3103+
3104+
appendStringInfo(es->str, "Peak Memory Usage: " INT64_FORMAT " kB",
3105+
memPeakKb);
30683106

3069-
/* EXPLAIN ANALYZE */
3070-
ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb, es);
3071-
if (aggstate->hash_batches_used > 0)
3107+
if (aggstate->hash_batches_used > 0)
3108+
appendStringInfo(es->str, " Disk Usage: " UINT64_FORMAT " kB HashAgg Batches: %d",
3109+
aggstate->hash_disk_used,
3110+
aggstate->hash_batches_used);
3111+
appendStringInfoChar(es->str, '\n');
3112+
}
3113+
3114+
/* Display stats for each parallel worker */
3115+
if (es->analyze && aggstate->shared_info != NULL)
30723116
{
3073-
ExplainPropertyInteger("Disk Usage", "kB",
3074-
aggstate->hash_disk_used, es);
3075-
ExplainPropertyInteger("HashAgg Batches", NULL,
3076-
aggstate->hash_batches_used, es);
3117+
for (int n = 0; n < aggstate->shared_info->num_workers; n++)
3118+
{
3119+
AggregateInstrumentation *sinstrument;
3120+
uint64 hash_disk_used;
3121+
int hash_batches_used;
3122+
3123+
sinstrument = &aggstate->shared_info->sinstrument[n];
3124+
hash_disk_used = sinstrument->hash_disk_used;
3125+
hash_batches_used = sinstrument->hash_batches_used;
3126+
memPeakKb = (sinstrument->hash_mem_peak + 1023) / 1024;
3127+
3128+
if (es->workers_state)
3129+
ExplainOpenWorker(n, es);
3130+
3131+
if (es->format == EXPLAIN_FORMAT_TEXT)
3132+
{
3133+
ExplainIndentText(es);
3134+
3135+
appendStringInfo(es->str, "Peak Memory Usage: " INT64_FORMAT " kB",
3136+
memPeakKb);
3137+
3138+
if (hash_batches_used > 0)
3139+
appendStringInfo(es->str, " Disk Usage: " UINT64_FORMAT " kB HashAgg Batches: %d",
3140+
hash_disk_used, hash_batches_used);
3141+
appendStringInfoChar(es->str, '\n');
3142+
}
3143+
else
3144+
{
3145+
ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb,
3146+
es);
3147+
if (hash_batches_used > 0)
3148+
{
3149+
ExplainPropertyInteger("Disk Usage", "kB", hash_disk_used,
3150+
es);
3151+
ExplainPropertyInteger("HashAgg Batches", NULL,
3152+
hash_batches_used, es);
3153+
}
3154+
}
3155+
3156+
if (es->workers_state)
3157+
ExplainCloseWorker(n, es);
3158+
}
30773159
}
30783160
}
30793161

‎src/backend/executor/execParallel.c‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525

2626
#include "executor/execParallel.h"
2727
#include "executor/executor.h"
28+
#include "executor/nodeAgg.h"
2829
#include "executor/nodeAppend.h"
2930
#include "executor/nodeBitmapHeapscan.h"
3031
#include "executor/nodeCustom.h"
@@ -288,7 +289,10 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e)
288289
/* even when not parallel-aware, for EXPLAIN ANALYZE */
289290
ExecIncrementalSortEstimate((IncrementalSortState *) planstate, e->pcxt);
290291
break;
291-
292+
case T_AggState:
293+
/* even when not parallel-aware, for EXPLAIN ANALYZE */
294+
ExecAggEstimate((AggState *) planstate, e->pcxt);
295+
break;
292296
default:
293297
break;
294298
}
@@ -505,7 +509,10 @@ ExecParallelInitializeDSM(PlanState *planstate,
505509
/* even when not parallel-aware, for EXPLAIN ANALYZE */
506510
ExecIncrementalSortInitializeDSM((IncrementalSortState *) planstate, d->pcxt);
507511
break;
508-
512+
case T_AggState:
513+
/* even when not parallel-aware, for EXPLAIN ANALYZE */
514+
ExecAggInitializeDSM((AggState *) planstate, d->pcxt);
515+
break;
509516
default:
510517
break;
511518
}
@@ -1048,6 +1055,9 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
10481055
case T_HashState:
10491056
ExecHashRetrieveInstrumentation((HashState *) planstate);
10501057
break;
1058+
case T_AggState:
1059+
ExecAggRetrieveInstrumentation((AggState *) planstate);
1060+
break;
10511061
default:
10521062
break;
10531063
}
@@ -1336,7 +1346,10 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
13361346
ExecIncrementalSortInitializeWorker((IncrementalSortState *) planstate,
13371347
pwcxt);
13381348
break;
1339-
1349+
case T_AggState:
1350+
/* even when not parallel-aware, for EXPLAIN ANALYZE */
1351+
ExecAggInitializeWorker((AggState *) planstate, pwcxt);
1352+
break;
13401353
default:
13411354
break;
13421355
}

‎src/backend/executor/nodeAgg.c‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -240,6 +240,7 @@
240240
#include "postgres.h"
241241

242242
#include "access/htup_details.h"
243+
#include "access/parallel.h"
243244
#include "catalog/objectaccess.h"
244245
#include "catalog/pg_aggregate.h"
245246
#include "catalog/pg_proc.h"
@@ -4483,6 +4484,22 @@ ExecEndAgg(AggState *node)
44834484
int numGroupingSets = Max(node->maxsets, 1);
44844485
int setno;
44854486

4487+
/*
4488+
* When ending a parallel worker, copy the statistics gathered by the
4489+
* worker back into shared memory so that it can be picked up by the main
4490+
* process to report in EXPLAIN ANALYZE.
4491+
*/
4492+
if (node->shared_info && IsParallelWorker())
4493+
{
4494+
AggregateInstrumentation *si;
4495+
4496+
Assert(ParallelWorkerNumber <= node->shared_info->num_workers);
4497+
si = &node->shared_info->sinstrument[ParallelWorkerNumber];
4498+
si->hash_batches_used = node->hash_batches_used;
4499+
si->hash_disk_used = node->hash_disk_used;
4500+
si->hash_mem_peak = node->hash_mem_peak;
4501+
}
4502+
44864503
/* Make sure we have closed any open tuplesorts */
44874504

44884505
if (node->sort_in)
@@ -4854,3 +4871,89 @@ aggregate_dummy(PG_FUNCTION_ARGS)
48544871
fcinfo->flinfo->fn_oid);
48554872
return (Datum) 0; /* keep compiler quiet */
48564873
}
4874+
4875+
/* ----------------------------------------------------------------
4876+
* Parallel Query Support
4877+
* ----------------------------------------------------------------
4878+
*/
4879+
4880+
/* ----------------------------------------------------------------
4881+
* ExecAggEstimate
4882+
*
4883+
* Estimate space required to propagate aggregate statistics.
4884+
* ----------------------------------------------------------------
4885+
*/
4886+
void
4887+
ExecAggEstimate(AggState *node, ParallelContext *pcxt)
4888+
{
4889+
Size size;
4890+
4891+
/* don't need this if not instrumenting or no workers */
4892+
if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4893+
return;
4894+
4895+
size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation));
4896+
size = add_size(size, offsetof(SharedAggInfo, sinstrument));
4897+
shm_toc_estimate_chunk(&pcxt->estimator, size);
4898+
shm_toc_estimate_keys(&pcxt->estimator, 1);
4899+
}
4900+
4901+
/* ----------------------------------------------------------------
4902+
* ExecAggInitializeDSM
4903+
*
4904+
* Initialize DSM space for aggregate statistics.
4905+
* ----------------------------------------------------------------
4906+
*/
4907+
void
4908+
ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt)
4909+
{
4910+
Size size;
4911+
4912+
/* don't need this if not instrumenting or no workers */
4913+
if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4914+
return;
4915+
4916+
size = offsetof(SharedAggInfo, sinstrument)
4917+
+ pcxt->nworkers * sizeof(AggregateInstrumentation);
4918+
node->shared_info = shm_toc_allocate(pcxt->toc, size);
4919+
/* ensure any unfilled slots will contain zeroes */
4920+
memset(node->shared_info, 0, size);
4921+
node->shared_info->num_workers = pcxt->nworkers;
4922+
shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
4923+
node->shared_info);
4924+
}
4925+
4926+
/* ----------------------------------------------------------------
4927+
* ExecAggInitializeWorker
4928+
*
4929+
* Attach worker to DSM space for aggregate statistics.
4930+
* ----------------------------------------------------------------
4931+
*/
4932+
void
4933+
ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt)
4934+
{
4935+
node->shared_info =
4936+
shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
4937+
}
4938+
4939+
/* ----------------------------------------------------------------
4940+
* ExecAggRetrieveInstrumentation
4941+
*
4942+
* Transfer aggregate statistics from DSM to private memory.
4943+
* ----------------------------------------------------------------
4944+
*/
4945+
void
4946+
ExecAggRetrieveInstrumentation(AggState *node)
4947+
{
4948+
Size size;
4949+
SharedAggInfo *si;
4950+
4951+
if (node->shared_info == NULL)
4952+
return;
4953+
4954+
size = offsetof(SharedAggInfo, sinstrument)
4955+
+ node->shared_info->num_workers * sizeof(AggregateInstrumentation);
4956+
si = palloc(size);
4957+
memcpy(si, node->shared_info, size);
4958+
node->shared_info = si;
4959+
}

‎src/include/executor/nodeAgg.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
#ifndef NODEAGG_H
1515
#define NODEAGG_H
1616

17+
#include "access/parallel.h"
1718
#include "nodes/execnodes.h"
1819

1920

@@ -323,4 +324,10 @@ extern void hash_agg_set_limits(double hashentrysize, uint64 input_groups,
323324
int used_bits, Size *mem_limit,
324325
uint64 *ngroups_limit, int *num_partitions);
325326

327+
/* parallel instrumentation support */
328+
extern void ExecAggEstimate(AggState *node, ParallelContext *pcxt);
329+
extern void ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt);
330+
extern void ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt);
331+
extern void ExecAggRetrieveInstrumentation(AggState *node);
332+
326333
#endif /* NODEAGG_H */

‎src/include/nodes/execnodes.h‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2101,6 +2101,27 @@ typedef struct GroupState
21012101
bool grp_done; /* indicates completion of Group scan */
21022102
} GroupState;
21032103

2104+
/* ---------------------
2105+
* per-worker aggregate information
2106+
* ---------------------
2107+
*/
2108+
typedef struct AggregateInstrumentation
2109+
{
2110+
Size hash_mem_peak; /* peak hash table memory usage */
2111+
uint64 hash_disk_used; /* kB of disk space used */
2112+
int hash_batches_used; /* batches used during entire execution */
2113+
} AggregateInstrumentation;
2114+
2115+
/* ----------------
2116+
* Shared memory container for per-worker aggregate information
2117+
* ----------------
2118+
*/
2119+
typedef struct SharedAggInfo
2120+
{
2121+
int num_workers;
2122+
AggregateInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER];
2123+
} SharedAggInfo;
2124+
21042125
/* ---------------------
21052126
* AggState information
21062127
*
@@ -2190,6 +2211,7 @@ typedef struct AggState
21902211
AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than
21912212
* ->hash_pergroup */
21922213
ProjectionInfo *combinedproj; /* projection machinery */
2214+
SharedAggInfo *shared_info; /* one entry per worker */
21932215
} AggState;
21942216

21952217
/* ----------------

0 commit comments

Comments
 (0)