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

Commit7df2c1f

Browse files
committed
Force rescanning of parallel-aware scan nodes below a Gather[Merge].
The ExecReScan machinery contains various optimizations for postponingor skipping rescans of plan subtrees; for example a HashAgg node mayconclude that it can re-use the table it built before, instead ofre-reading its input subtree. But that is wrong if the input containsa parallel-aware table scan node, since the portion of the table scannedby the leader process is likely to vary from one rescan to the next.This explains the timing-dependent buildfarm failures we saw aftercommita2b70c8.The established mechanism for showing that a plan node's output ispotentially variable is to mark it as depending on some runtime Param.Hence, to fix this, invent a dummy Param (one that has a PARAM_EXECparameter number, but carries no actual value) associated with each Gatheror GatherMerge node, mark parallel-aware nodes below that node as dependenton that Param, and arrange for ExecReScanGather[Merge] to flag that Paramas changed whenever the Gather[Merge] node is rescanned.This solution breaks an undocumented assumption made by the parallelexecutor logic, namely that all rescans of nodes below a Gather[Merge]will happen synchronously during the ReScan of the top node itself.But that's fundamentally contrary to the design of the ExecReScan code,and so was doomed to fail someday anyway (even if you want to arguethat the bug being fixed here wasn't a failure of that assumption).A follow-on patch will address that issue. In the meantime, the worstthat's expected to happen is that given very bad timing luck, the leadermight have to do all the work during a rescan, because workers thinkthey have nothing to do, if they are able to start up before the eventualReScan of the leader's parallel-aware table scan node has reset theshared scan state.Although this problem exists in 9.6, there does not seem to be any wayfor it to manifest there. Without GatherMerge, it seems that a plan treethat has a rescan-short-circuiting node below Gather will always alsohave one above it that will short-circuit in the same cases, preventingthe Gather from being rescanned. Hence we won't take the risk ofback-patching this change into 9.6. But v10 needs it.Discussion:https://postgr.es/m/CAA4eK1JkByysFJNh9M349u_nNjqETuEnY_y1VUc_kJiU0bxtaQ@mail.gmail.com
1 parent00f6d5c commit7df2c1f

File tree

11 files changed

+155
-24
lines changed

11 files changed

+155
-24
lines changed

‎src/backend/executor/nodeGather.c

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -432,6 +432,9 @@ ExecShutdownGather(GatherState *node)
432432
void
433433
ExecReScanGather(GatherState*node)
434434
{
435+
Gather*gather= (Gather*)node->ps.plan;
436+
PlanState*outerPlan=outerPlanState(node);
437+
435438
/*
436439
* Re-initialize the parallel workers to perform rescan of relation. We
437440
* want to gracefully shutdown all the workers so that they should be able
@@ -445,5 +448,22 @@ ExecReScanGather(GatherState *node)
445448
if (node->pei)
446449
ExecParallelReinitialize(node->pei);
447450

448-
ExecReScan(node->ps.lefttree);
451+
/*
452+
* Set child node's chgParam to tell it that the next scan might deliver a
453+
* different set of rows within the leader process. (The overall rowset
454+
* shouldn't change, but the leader process's subset might; hence nodes
455+
* between here and the parallel table scan node mustn't optimize on the
456+
* assumption of an unchanging rowset.)
457+
*/
458+
if (gather->rescan_param >=0)
459+
outerPlan->chgParam=bms_add_member(outerPlan->chgParam,
460+
gather->rescan_param);
461+
462+
463+
/*
464+
* if chgParam of subnode is not null then plan will be re-scanned by
465+
* first ExecProcNode.
466+
*/
467+
if (outerPlan->chgParam==NULL)
468+
ExecReScan(outerPlan);
449469
}

‎src/backend/executor/nodeGatherMerge.c

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -327,6 +327,9 @@ ExecShutdownGatherMergeWorkers(GatherMergeState *node)
327327
void
328328
ExecReScanGatherMerge(GatherMergeState*node)
329329
{
330+
GatherMerge*gm= (GatherMerge*)node->ps.plan;
331+
PlanState*outerPlan=outerPlanState(node);
332+
330333
/*
331334
* Re-initialize the parallel workers to perform rescan of relation. We
332335
* want to gracefully shutdown all the workers so that they should be able
@@ -341,7 +344,24 @@ ExecReScanGatherMerge(GatherMergeState *node)
341344
if (node->pei)
342345
ExecParallelReinitialize(node->pei);
343346

344-
ExecReScan(node->ps.lefttree);
347+
/*
348+
* Set child node's chgParam to tell it that the next scan might deliver a
349+
* different set of rows within the leader process. (The overall rowset
350+
* shouldn't change, but the leader process's subset might; hence nodes
351+
* between here and the parallel table scan node mustn't optimize on the
352+
* assumption of an unchanging rowset.)
353+
*/
354+
if (gm->rescan_param >=0)
355+
outerPlan->chgParam=bms_add_member(outerPlan->chgParam,
356+
gm->rescan_param);
357+
358+
359+
/*
360+
* if chgParam of subnode is not null then plan will be re-scanned by
361+
* first ExecProcNode.
362+
*/
363+
if (outerPlan->chgParam==NULL)
364+
ExecReScan(outerPlan);
345365
}
346366

347367
/*

‎src/backend/nodes/copyfuncs.c

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -361,6 +361,7 @@ _copyGather(const Gather *from)
361361
* copy remainder of node
362362
*/
363363
COPY_SCALAR_FIELD(num_workers);
364+
COPY_SCALAR_FIELD(rescan_param);
364365
COPY_SCALAR_FIELD(single_copy);
365366
COPY_SCALAR_FIELD(invisible);
366367

@@ -384,6 +385,7 @@ _copyGatherMerge(const GatherMerge *from)
384385
* copy remainder of node
385386
*/
386387
COPY_SCALAR_FIELD(num_workers);
388+
COPY_SCALAR_FIELD(rescan_param);
387389
COPY_SCALAR_FIELD(numCols);
388390
COPY_POINTER_FIELD(sortColIdx,from->numCols*sizeof(AttrNumber));
389391
COPY_POINTER_FIELD(sortOperators,from->numCols*sizeof(Oid));

‎src/backend/nodes/outfuncs.c

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -479,6 +479,7 @@ _outGather(StringInfo str, const Gather *node)
479479
_outPlanInfo(str, (constPlan*)node);
480480

481481
WRITE_INT_FIELD(num_workers);
482+
WRITE_INT_FIELD(rescan_param);
482483
WRITE_BOOL_FIELD(single_copy);
483484
WRITE_BOOL_FIELD(invisible);
484485
}
@@ -493,6 +494,7 @@ _outGatherMerge(StringInfo str, const GatherMerge *node)
493494
_outPlanInfo(str, (constPlan*)node);
494495

495496
WRITE_INT_FIELD(num_workers);
497+
WRITE_INT_FIELD(rescan_param);
496498
WRITE_INT_FIELD(numCols);
497499

498500
appendStringInfoString(str," :sortColIdx");

‎src/backend/nodes/readfuncs.c

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2163,6 +2163,7 @@ _readGather(void)
21632163
ReadCommonPlan(&local_node->plan);
21642164

21652165
READ_INT_FIELD(num_workers);
2166+
READ_INT_FIELD(rescan_param);
21662167
READ_BOOL_FIELD(single_copy);
21672168
READ_BOOL_FIELD(invisible);
21682169

@@ -2180,6 +2181,7 @@ _readGatherMerge(void)
21802181
ReadCommonPlan(&local_node->plan);
21812182

21822183
READ_INT_FIELD(num_workers);
2184+
READ_INT_FIELD(rescan_param);
21832185
READ_INT_FIELD(numCols);
21842186
READ_ATTRNUMBER_ARRAY(sortColIdx,local_node->numCols);
21852187
READ_OID_ARRAY(sortOperators,local_node->numCols);

‎src/backend/optimizer/README

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -374,6 +374,7 @@ RelOptInfo - a relation or joined relations
374374
MaterialPath - a Material plan node
375375
UniquePath - remove duplicate rows (either by hashing or sorting)
376376
GatherPath - collect the results of parallel workers
377+
GatherMergePath - collect parallel results, preserving their common sort order
377378
ProjectionPath - a Result plan node with child (used for projection)
378379
ProjectSetPath - a ProjectSet plan node applied to some sub-path
379380
SortPath - a Sort plan node applied to some sub-path
@@ -1030,7 +1031,7 @@ either by an entire query or some portion of the query in such a way that
10301031
some of that work can be done by one or more worker processes, which are
10311032
called parallel workers. Parallel workers are a subtype of dynamic
10321033
background workers; see src/backend/access/transam/README.parallel for a
1033-
fuller description.Academicliterature on parallel query suggests that
1034+
fuller description.The academicliterature on parallel query suggests
10341035
that parallel execution strategies can be divided into essentially two
10351036
categories: pipelined parallelism, where the execution of the query is
10361037
divided into multiple stages and each stage is handled by a separate
@@ -1046,16 +1047,14 @@ that the underlying table be partitioned. It only requires that (1)
10461047
there is some method of dividing the data from at least one of the base
10471048
tables involved in the relation across multiple processes, (2) allowing
10481049
each process to handle its own portion of the data, and then (3)
1049-
collecting the results. Requirements (2) and (3) is satisfied by the
1050-
executor node Gather, which launches any number of worker processes and
1051-
executes its single child plan in all of them (and perhaps in the leader
1052-
also, if the children aren't generating enough data to keep the leader
1053-
busy). Requirement (1) is handled by the SeqScan node: when invoked
1054-
with parallel_aware = true, this node will, in effect, partition the
1055-
table on a block by block basis, returning a subset of the tuples from
1056-
the relation in each worker where that SeqScan is executed. A similar
1057-
scheme could be (and probably should be) implemented for bitmap heap
1058-
scans.
1050+
collecting the results. Requirements (2) and (3) are satisfied by the
1051+
executor node Gather (or GatherMerge), which launches any number of worker
1052+
processes and executes its single child plan in all of them, and perhaps
1053+
in the leader also, if the children aren't generating enough data to keep
1054+
the leader busy. Requirement (1) is handled by the table scan node: when
1055+
invoked with parallel_aware = true, this node will, in effect, partition
1056+
the table on a block by block basis, returning a subset of the tuples from
1057+
the relation in each worker where that scan node is executed.
10591058

10601059
Just as we do for non-parallel access methods, we build Paths to
10611060
represent access strategies that can be used in a parallel plan. These

‎src/backend/optimizer/plan/createplan.c

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -267,7 +267,7 @@ static Unique *make_unique_from_sortclauses(Plan *lefttree, List *distinctList);
267267
staticUnique*make_unique_from_pathkeys(Plan*lefttree,
268268
List*pathkeys,intnumCols);
269269
staticGather*make_gather(List*qptlist,List*qpqual,
270-
intnworkers,boolsingle_copy,Plan*subplan);
270+
intnworkers,intrescan_param,boolsingle_copy,Plan*subplan);
271271
staticSetOp*make_setop(SetOpCmdcmd,SetOpStrategystrategy,Plan*lefttree,
272272
List*distinctList,AttrNumberflagColIdx,intfirstFlag,
273273
longnumGroups);
@@ -1471,6 +1471,7 @@ create_gather_plan(PlannerInfo *root, GatherPath *best_path)
14711471
gather_plan=make_gather(tlist,
14721472
NIL,
14731473
best_path->num_workers,
1474+
SS_assign_special_param(root),
14741475
best_path->single_copy,
14751476
subplan);
14761477

@@ -1505,6 +1506,9 @@ create_gather_merge_plan(PlannerInfo *root, GatherMergePath *best_path)
15051506
gm_plan->num_workers=best_path->num_workers;
15061507
copy_generic_path_info(&gm_plan->plan,&best_path->path);
15071508

1509+
/* Assign the rescan Param. */
1510+
gm_plan->rescan_param=SS_assign_special_param(root);
1511+
15081512
/* Gather Merge is pointless with no pathkeys; use Gather instead. */
15091513
Assert(pathkeys!=NIL);
15101514

@@ -6238,6 +6242,7 @@ static Gather *
62386242
make_gather(List*qptlist,
62396243
List*qpqual,
62406244
intnworkers,
6245+
intrescan_param,
62416246
boolsingle_copy,
62426247
Plan*subplan)
62436248
{
@@ -6249,6 +6254,7 @@ make_gather(List *qptlist,
62496254
plan->lefttree=subplan;
62506255
plan->righttree=NULL;
62516256
node->num_workers=nworkers;
6257+
node->rescan_param=rescan_param;
62526258
node->single_copy=single_copy;
62536259
node->invisible= false;
62546260

‎src/backend/optimizer/plan/planner.c

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -374,6 +374,12 @@ standard_planner(Query *parse, int cursorOptions, ParamListInfo boundParams)
374374
gather->single_copy= true;
375375
gather->invisible= (force_parallel_mode==FORCE_PARALLEL_REGRESS);
376376

377+
/*
378+
* Since this Gather has no parallel-aware descendants to signal to,
379+
* we don't need a rescan Param.
380+
*/
381+
gather->rescan_param=-1;
382+
377383
/*
378384
* Ideally we'd use cost_gather here, but setting up dummy path data
379385
* to satisfy it doesn't seem much cleaner than knowing what it does.

‎src/backend/optimizer/plan/subselect.c

Lines changed: 69 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ static Node *process_sublinks_mutator(Node *node,
7979
process_sublinks_context*context);
8080
staticBitmapset*finalize_plan(PlannerInfo*root,
8181
Plan*plan,
82+
intgather_param,
8283
Bitmapset*valid_params,
8384
Bitmapset*scan_params);
8485
staticboolfinalize_primnode(Node*node,finalize_primnode_context*context);
@@ -2217,12 +2218,15 @@ void
22172218
SS_finalize_plan(PlannerInfo*root,Plan*plan)
22182219
{
22192220
/* No setup needed, just recurse through plan tree. */
2220-
(void)finalize_plan(root,plan,root->outer_params,NULL);
2221+
(void)finalize_plan(root,plan,-1,root->outer_params,NULL);
22212222
}
22222223

22232224
/*
22242225
* Recursive processing of all nodes in the plan tree
22252226
*
2227+
* gather_param is the rescan_param of an ancestral Gather/GatherMerge,
2228+
* or -1 if there is none.
2229+
*
22262230
* valid_params is the set of param IDs supplied by outer plan levels
22272231
* that are valid to reference in this plan node or its children.
22282232
*
@@ -2249,7 +2253,9 @@ SS_finalize_plan(PlannerInfo *root, Plan *plan)
22492253
* can be handled more cleanly.
22502254
*/
22512255
staticBitmapset*
2252-
finalize_plan(PlannerInfo*root,Plan*plan,Bitmapset*valid_params,
2256+
finalize_plan(PlannerInfo*root,Plan*plan,
2257+
intgather_param,
2258+
Bitmapset*valid_params,
22532259
Bitmapset*scan_params)
22542260
{
22552261
finalize_primnode_contextcontext;
@@ -2302,6 +2308,18 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
23022308
finalize_primnode((Node*)plan->targetlist,&context);
23032309
finalize_primnode((Node*)plan->qual,&context);
23042310

2311+
/*
2312+
* If it's a parallel-aware scan node, mark it as dependent on the parent
2313+
* Gather/GatherMerge's rescan Param.
2314+
*/
2315+
if (plan->parallel_aware)
2316+
{
2317+
if (gather_param<0)
2318+
elog(ERROR,"parallel-aware plan node is not below a Gather");
2319+
context.paramids=
2320+
bms_add_member(context.paramids,gather_param);
2321+
}
2322+
23052323
/* Check additional node-type-specific fields */
23062324
switch (nodeTag(plan))
23072325
{
@@ -2512,6 +2530,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
25122530
bms_add_members(context.paramids,
25132531
finalize_plan(root,
25142532
(Plan*)lfirst(lc),
2533+
gather_param,
25152534
valid_params,
25162535
scan_params));
25172536
}
@@ -2542,6 +2561,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
25422561
bms_add_members(context.paramids,
25432562
finalize_plan(root,
25442563
(Plan*)lfirst(l),
2564+
gather_param,
25452565
valid_params,
25462566
scan_params));
25472567
}
@@ -2558,6 +2578,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
25582578
bms_add_members(context.paramids,
25592579
finalize_plan(root,
25602580
(Plan*)lfirst(l),
2581+
gather_param,
25612582
valid_params,
25622583
scan_params));
25632584
}
@@ -2574,6 +2595,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
25742595
bms_add_members(context.paramids,
25752596
finalize_plan(root,
25762597
(Plan*)lfirst(l),
2598+
gather_param,
25772599
valid_params,
25782600
scan_params));
25792601
}
@@ -2590,6 +2612,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
25902612
bms_add_members(context.paramids,
25912613
finalize_plan(root,
25922614
(Plan*)lfirst(l),
2615+
gather_param,
25932616
valid_params,
25942617
scan_params));
25952618
}
@@ -2606,6 +2629,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
26062629
bms_add_members(context.paramids,
26072630
finalize_plan(root,
26082631
(Plan*)lfirst(l),
2632+
gather_param,
26092633
valid_params,
26102634
scan_params));
26112635
}
@@ -2697,13 +2721,51 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
26972721
&context);
26982722
break;
26992723

2724+
caseT_Gather:
2725+
/* child nodes are allowed to reference rescan_param, if any */
2726+
locally_added_param= ((Gather*)plan)->rescan_param;
2727+
if (locally_added_param >=0)
2728+
{
2729+
valid_params=bms_add_member(bms_copy(valid_params),
2730+
locally_added_param);
2731+
2732+
/*
2733+
* We currently don't support nested Gathers. The issue so
2734+
* far as this function is concerned would be how to identify
2735+
* which child nodes depend on which Gather.
2736+
*/
2737+
Assert(gather_param<0);
2738+
/* Pass down rescan_param to child parallel-aware nodes */
2739+
gather_param=locally_added_param;
2740+
}
2741+
/* rescan_param does *not* get added to scan_params */
2742+
break;
2743+
2744+
caseT_GatherMerge:
2745+
/* child nodes are allowed to reference rescan_param, if any */
2746+
locally_added_param= ((GatherMerge*)plan)->rescan_param;
2747+
if (locally_added_param >=0)
2748+
{
2749+
valid_params=bms_add_member(bms_copy(valid_params),
2750+
locally_added_param);
2751+
2752+
/*
2753+
* We currently don't support nested Gathers. The issue so
2754+
* far as this function is concerned would be how to identify
2755+
* which child nodes depend on which Gather.
2756+
*/
2757+
Assert(gather_param<0);
2758+
/* Pass down rescan_param to child parallel-aware nodes */
2759+
gather_param=locally_added_param;
2760+
}
2761+
/* rescan_param does *not* get added to scan_params */
2762+
break;
2763+
27002764
caseT_ProjectSet:
27012765
caseT_Hash:
27022766
caseT_Material:
27032767
caseT_Sort:
27042768
caseT_Unique:
2705-
caseT_Gather:
2706-
caseT_GatherMerge:
27072769
caseT_SetOp:
27082770
caseT_Group:
27092771
/* no node-type-specific fields need fixing */
@@ -2717,6 +2779,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
27172779
/* Process left and right child plans, if any */
27182780
child_params=finalize_plan(root,
27192781
plan->lefttree,
2782+
gather_param,
27202783
valid_params,
27212784
scan_params);
27222785
context.paramids=bms_add_members(context.paramids,child_params);
@@ -2726,6 +2789,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
27262789
/* right child can reference nestloop_params as well as valid_params */
27272790
child_params=finalize_plan(root,
27282791
plan->righttree,
2792+
gather_param,
27292793
bms_union(nestloop_params,valid_params),
27302794
scan_params);
27312795
/* ... and they don't count as parameters used at my level */
@@ -2737,6 +2801,7 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
27372801
/* easy case */
27382802
child_params=finalize_plan(root,
27392803
plan->righttree,
2804+
gather_param,
27402805
valid_params,
27412806
scan_params);
27422807
}

0 commit comments

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