Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit8526bcb

Browse files
committed
Try again to fix accumulation of parallel worker instrumentation.
When a Gather or Gather Merge node is started and stopped multipletimes, accumulate instrumentation data only once, at the end, insteadof after each execution, to avoid recording inflated totals.Commit778e78a, the previous attemptat a fix, instead reset the state after every execution, which workedfor the general instrumentation data but had problems for the additionalinstrumentation specific to Sort and Hash nodes.Report by hubert depesz lubaczewski. Analysis and fix by Amit Kapila,following a design proposal from Thomas Munro, with a comment tweakby me.Discussion:http://postgr.es/m/20171127175631.GA405@depesz.com
1 parent38fc547 commit8526bcb

File tree

7 files changed

+96
-49
lines changed

7 files changed

+96
-49
lines changed

‎src/backend/executor/execParallel.c

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -899,12 +899,8 @@ ExecParallelReInitializeDSM(PlanState *planstate,
899899
pcxt);
900900
break;
901901
caseT_HashState:
902-
/* even when not parallel-aware, for EXPLAIN ANALYZE */
903-
ExecHashReInitializeDSM((HashState*)planstate,pcxt);
904-
break;
905902
caseT_SortState:
906-
/* even when not parallel-aware, for EXPLAIN ANALYZE */
907-
ExecSortReInitializeDSM((SortState*)planstate,pcxt);
903+
/* these nodes have DSM state, but no reinitialization is required */
908904
break;
909905

910906
default:
@@ -977,7 +973,7 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
977973

978974
/*
979975
* Finish parallel execution. We wait for parallel workers to finish, and
980-
* accumulate their buffer usage and instrumentation.
976+
* accumulate their buffer usage.
981977
*/
982978
void
983979
ExecParallelFinish(ParallelExecutorInfo*pei)
@@ -1023,23 +1019,23 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
10231019
for (i=0;i<nworkers;i++)
10241020
InstrAccumParallelQuery(&pei->buffer_usage[i]);
10251021

1026-
/* Finally, accumulate instrumentation, if any. */
1027-
if (pei->instrumentation)
1028-
ExecParallelRetrieveInstrumentation(pei->planstate,
1029-
pei->instrumentation);
1030-
10311022
pei->finished= true;
10321023
}
10331024

10341025
/*
1035-
*Clean up whatever ParallelExecutorInfo resources still exist after
1036-
* ExecParallelFinish. We separate these routines because someone might
1037-
* want to examine the contents of the DSM after ExecParallelFinish and
1038-
* before calling this routine.
1026+
*Accumulate instrumentation, and then clean up whatever ParallelExecutorInfo
1027+
*resources still exist afterExecParallelFinish. We separate these
1028+
*routines because someone mightwant to examine the contents of the DSM
1029+
*after ExecParallelFinish andbefore calling this routine.
10391030
*/
10401031
void
10411032
ExecParallelCleanup(ParallelExecutorInfo*pei)
10421033
{
1034+
/* Accumulate instrumentation, if any. */
1035+
if (pei->instrumentation)
1036+
ExecParallelRetrieveInstrumentation(pei->planstate,
1037+
pei->instrumentation);
1038+
10431039
/* Free any serialized parameters. */
10441040
if (DsaPointerIsValid(pei->param_exec))
10451041
{

‎src/backend/executor/nodeHash.c

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1669,19 +1669,6 @@ ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt)
16691669
node->shared_info);
16701670
}
16711671

1672-
/*
1673-
* Reset shared state before beginning a fresh scan.
1674-
*/
1675-
void
1676-
ExecHashReInitializeDSM(HashState*node,ParallelContext*pcxt)
1677-
{
1678-
if (node->shared_info!=NULL)
1679-
{
1680-
memset(node->shared_info->hinstrument,0,
1681-
node->shared_info->num_workers*sizeof(HashInstrumentation));
1682-
}
1683-
}
1684-
16851672
/*
16861673
* Locate the DSM space for hash table instrumentation data that we'll write
16871674
* to at shutdown time.

‎src/backend/executor/nodeSort.c

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -396,23 +396,6 @@ ExecSortInitializeDSM(SortState *node, ParallelContext *pcxt)
396396
node->shared_info);
397397
}
398398

399-
/* ----------------------------------------------------------------
400-
*ExecSortReInitializeDSM
401-
*
402-
*Reset shared state before beginning a fresh scan.
403-
* ----------------------------------------------------------------
404-
*/
405-
void
406-
ExecSortReInitializeDSM(SortState*node,ParallelContext*pcxt)
407-
{
408-
/* If there's any instrumentation space, clear it for next time */
409-
if (node->shared_info!=NULL)
410-
{
411-
memset(node->shared_info->sinstrument,0,
412-
node->shared_info->num_workers*sizeof(TuplesortInstrumentation));
413-
}
414-
}
415-
416399
/* ----------------------------------------------------------------
417400
*ExecSortInitializeWorker
418401
*

‎src/include/executor/nodeHash.h

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@ extern intExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue);
5252
externvoidExecHashEstimate(HashState*node,ParallelContext*pcxt);
5353
externvoidExecHashInitializeDSM(HashState*node,ParallelContext*pcxt);
5454
externvoidExecHashInitializeWorker(HashState*node,ParallelWorkerContext*pwcxt);
55-
externvoidExecHashReInitializeDSM(HashState*node,ParallelContext*pcxt);
5655
externvoidExecHashRetrieveInstrumentation(HashState*node);
5756
externvoidExecShutdownHash(HashState*node);
5857
externvoidExecHashGetInstrumentation(HashInstrumentation*instrument,

‎src/include/executor/nodeSort.h

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ extern void ExecReScanSort(SortState *node);
2626
/* parallel instrumentation support */
2727
externvoidExecSortEstimate(SortState*node,ParallelContext*pcxt);
2828
externvoidExecSortInitializeDSM(SortState*node,ParallelContext*pcxt);
29-
externvoidExecSortReInitializeDSM(SortState*node,ParallelContext*pcxt);
3029
externvoidExecSortInitializeWorker(SortState*node,ParallelWorkerContext*pwcxt);
3130
externvoidExecSortRetrieveInstrumentation(SortState*node);
3231

‎src/test/regress/expected/select_parallel.out

Lines changed: 58 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -465,14 +465,71 @@ select count(*) from bmscantest where a>1;
465465
99999
466466
(1 row)
467467

468+
-- test accumulation of stats for parallel nodes
468469
reset enable_seqscan;
470+
alter table tenk2 set (parallel_workers = 0);
471+
explain (analyze, timing off, summary off, costs off)
472+
select count(*) from tenk1, tenk2 where tenk1.hundred > 1
473+
and tenk2.thousand=0;
474+
QUERY PLAN
475+
--------------------------------------------------------------------------
476+
Aggregate (actual rows=1 loops=1)
477+
-> Nested Loop (actual rows=98000 loops=1)
478+
-> Seq Scan on tenk2 (actual rows=10 loops=1)
479+
Filter: (thousand = 0)
480+
Rows Removed by Filter: 9990
481+
-> Gather (actual rows=9800 loops=10)
482+
Workers Planned: 4
483+
Workers Launched: 4
484+
-> Parallel Seq Scan on tenk1 (actual rows=1960 loops=50)
485+
Filter: (hundred > 1)
486+
Rows Removed by Filter: 40
487+
(11 rows)
488+
489+
alter table tenk2 reset (parallel_workers);
490+
reset work_mem;
491+
create function explain_parallel_sort_stats() returns setof text
492+
language plpgsql as
493+
$$
494+
declare ln text;
495+
begin
496+
for ln in
497+
explain (analyze, timing off, summary off, costs off)
498+
select * from
499+
(select ten from tenk1 where ten < 100 order by ten) ss
500+
right join (values (1),(2),(3)) v(x) on true
501+
loop
502+
ln := regexp_replace(ln, 'Memory: \S*', 'Memory: xxx');
503+
return next ln;
504+
end loop;
505+
end;
506+
$$;
507+
select * from explain_parallel_sort_stats();
508+
explain_parallel_sort_stats
509+
--------------------------------------------------------------------------
510+
Nested Loop Left Join (actual rows=30000 loops=1)
511+
-> Values Scan on "*VALUES*" (actual rows=3 loops=1)
512+
-> Gather Merge (actual rows=10000 loops=3)
513+
Workers Planned: 4
514+
Workers Launched: 4
515+
-> Sort (actual rows=2000 loops=15)
516+
Sort Key: tenk1.ten
517+
Sort Method: quicksort Memory: xxx
518+
Worker 0: Sort Method: quicksort Memory: xxx
519+
Worker 1: Sort Method: quicksort Memory: xxx
520+
Worker 2: Sort Method: quicksort Memory: xxx
521+
Worker 3: Sort Method: quicksort Memory: xxx
522+
-> Parallel Seq Scan on tenk1 (actual rows=2000 loops=15)
523+
Filter: (ten < 100)
524+
(14 rows)
525+
469526
reset enable_indexscan;
470527
reset enable_hashjoin;
471528
reset enable_mergejoin;
472529
reset enable_material;
473530
reset effective_io_concurrency;
474-
reset work_mem;
475531
drop table bmscantest;
532+
drop function explain_parallel_sort_stats();
476533
-- test parallel merge join path.
477534
set enable_hashjoin to off;
478535
set enable_nestloop to off;

‎src/test/regress/sql/select_parallel.sql

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,14 +179,40 @@ insert into bmscantest select r, 'fooooooooooooooooooooooooooooooooooooooooooooo
179179
createindexi_bmtestON bmscantest(a);
180180
selectcount(*)from bmscantestwhere a>1;
181181

182+
-- test accumulation of stats for parallel nodes
182183
reset enable_seqscan;
184+
altertable tenk2set (parallel_workers=0);
185+
explain (analyze, timing off, summary off, costs off)
186+
selectcount(*)from tenk1, tenk2wheretenk1.hundred>1
187+
andtenk2.thousand=0;
188+
altertable tenk2 reset (parallel_workers);
189+
190+
reset work_mem;
191+
createfunctionexplain_parallel_sort_stats() returns setoftext
192+
language plpgsqlas
193+
$$
194+
declare lntext;
195+
begin
196+
for lnin
197+
explain (analyze, timing off, summary off, costs off)
198+
select*from
199+
(select tenfrom tenk1where ten<100order by ten) ss
200+
right join (values (1),(2),(3)) v(x)on true
201+
loop
202+
ln := regexp_replace(ln,'Memory:\S*','Memory: xxx');
203+
return next ln;
204+
end loop;
205+
end;
206+
$$;
207+
select*from explain_parallel_sort_stats();
208+
183209
reset enable_indexscan;
184210
reset enable_hashjoin;
185211
reset enable_mergejoin;
186212
reset enable_material;
187213
reset effective_io_concurrency;
188-
reset work_mem;
189214
droptable bmscantest;
215+
dropfunction explain_parallel_sort_stats();
190216

191217
-- test parallel merge join path.
192218
set enable_hashjoin to off;

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp