diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 98d4f1e..6b34dab 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -25,6 +25,7 @@ #include "executor/execParallel.h" #include "executor/executor.h" +#include "executor/nodeAppend.h" #include "executor/nodeCustom.h" #include "executor/nodeForeignscan.h" #include "executor/nodeSeqscan.h" @@ -206,6 +207,10 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e) ExecForeignScanEstimate((ForeignScanState *) planstate, e->pcxt); break; + case T_AppendState: + ExecAppendEstimate((AppendState *) planstate, + e->pcxt); + break; case T_CustomScanState: ExecCustomScanEstimate((CustomScanState *) planstate, e->pcxt); @@ -258,6 +263,10 @@ ExecParallelInitializeDSM(PlanState *planstate, ExecIndexScanInitializeDSM((IndexScanState *) planstate, d->pcxt); break; + case T_AppendState: + ExecAppendInitializeDSM((AppendState *) planstate, + d->pcxt); + break; case T_ForeignScanState: ExecForeignScanInitializeDSM((ForeignScanState *) planstate, d->pcxt); @@ -737,6 +746,9 @@ ExecParallelInitializeWorker(PlanState *planstate, shm_toc *toc) case T_IndexScanState: ExecIndexScanInitializeWorker((IndexScanState *) planstate, toc); break; + case T_AppendState: + ExecAppendInitializeWorker((AppendState *) planstate, toc); + break; case T_ForeignScanState: ExecForeignScanInitializeWorker((ForeignScanState *) planstate, toc); diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index 6986cae..97bfc89 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -59,9 +59,48 @@ #include "executor/execdebug.h" #include "executor/nodeAppend.h" +#include "miscadmin.h" +#include "optimizer/cost.h" +#include "storage/spin.h" +/* + * Shared state for Parallel Append. + * + * Each backend participating in a Parallel Append has its own + * descriptor in backend-private memory, and those objects all contain + * a pointer to this structure. + */ +typedef struct ParallelAppendInfo +{ + int pa_num_workers; /* workers currently executing the subplan */ + int pa_max_workers; /* max workers that should run the subplan */ +} ParallelAppendInfo; + +typedef struct ParallelAppendDescData +{ + slock_t pa_mutex; /* mutual exclusion to choose next subplan */ + ParallelAppendInfo pa_info[FLEXIBLE_ARRAY_MEMBER]; +} ParallelAppendDescData; + +typedef ParallelAppendDescData *ParallelAppendDesc; + +/* + * Special value of AppendState->as_whichplan for Parallel Append, which + * indicates there are no plans left to be executed. + */ +#define PA_INVALID_PLAN -1 + + +static void exec_append_scan_first(AppendState *appendstate); static bool exec_append_initialize_next(AppendState *appendstate); +static void set_finished(ParallelAppendDesc padesc, int whichplan); +static bool parallel_append_next(AppendState *state); +static inline void +exec_append_scan_first(AppendState *appendstate) +{ + appendstate->as_whichplan = 0; +} /* ---------------------------------------------------------------- * exec_append_initialize_next @@ -77,6 +116,22 @@ exec_append_initialize_next(AppendState *appendstate) int whichplan; /* + * In case it's parallel-aware, follow it's own logic of choosing the next + * subplan. + */ + if (appendstate->as_padesc) + return parallel_append_next(appendstate); + + /* + * Not parallel-aware. Fine, just go on to the next subplan in the + * appropriate direction. + */ + if (ScanDirectionIsForward(appendstate->ps.state->es_direction)) + appendstate->as_whichplan++; + else + appendstate->as_whichplan--; + + /* * get information from the append node */ whichplan = appendstate->as_whichplan; @@ -178,8 +233,7 @@ ExecInitAppend(Append *node, EState *estate, int eflags) /* * initialize to scan first subplan */ - appendstate->as_whichplan = 0; - exec_append_initialize_next(appendstate); + exec_append_scan_first(appendstate); return appendstate; } @@ -198,6 +252,14 @@ ExecAppend(AppendState *node) PlanState *subnode; TupleTableSlot *result; + /* Check if we are already finished plans from parallel append */ + if (node->as_padesc && node->as_whichplan == PA_INVALID_PLAN) + { + elog(DEBUG2, "ParallelAppend : pid %d : all plans already finished", + MyProcPid); + return ExecClearTuple(node->ps.ps_ResultTupleSlot); + } + /* * figure out which subplan we are currently processing */ @@ -219,14 +281,18 @@ ExecAppend(AppendState *node) } /* - * Go on to the "next" subplan in the appropriate direction. If no - * more subplans, return the empty slot set up for us by - * ExecInitAppend. + * We are done with this subplan. There might be other workers still + * processing the last chunk of rows for this same subplan, but there's + * no point for new workers to run this subplan, so mark this subplan + * as finished. + */ + if (node->as_padesc) + set_finished(node->as_padesc, node->as_whichplan); + + /* + * Go on to the "next" subplan. If no more subplans, return the empty + * slot set up for us by ExecInitAppend. */ - if (ScanDirectionIsForward(node->ps.state->es_direction)) - node->as_whichplan++; - else - node->as_whichplan--; if (!exec_append_initialize_next(node)) return ExecClearTuple(node->ps.ps_ResultTupleSlot); @@ -270,6 +336,7 @@ ExecReScanAppend(AppendState *node) for (i = 0; i < node->as_nplans; i++) { PlanState *subnode = node->appendplans[i]; + ParallelAppendDesc padesc = node->as_padesc; /* * ExecReScan doesn't know about my subplans, so I have to do @@ -284,7 +351,223 @@ ExecReScanAppend(AppendState *node) */ if (subnode->chgParam == NULL) ExecReScan(subnode); + + if (padesc) + { + /* + * Just setting all the number of workers to 0 is enough. The logic + * of choosing the next plan will take care of everything else. + * pa_max_workers is already set initially. + */ + padesc->pa_info[i].pa_num_workers = 0; + } } - node->as_whichplan = 0; - exec_append_initialize_next(node); + + exec_append_scan_first(node); +} + +/* ---------------------------------------------------------------- + * Parallel Append Support + * ---------------------------------------------------------------- + */ + +/* ---------------------------------------------------------------- + * ExecAppendEstimate + * + * estimates the space required to serialize Append node. + * ---------------------------------------------------------------- + */ +void +ExecAppendEstimate(AppendState *node, + ParallelContext *pcxt) +{ + node->pappend_len = + add_size(offsetof(struct ParallelAppendDescData, pa_info), + sizeof(*node->as_padesc->pa_info) * node->as_nplans); + + shm_toc_estimate_chunk(&pcxt->estimator, node->pappend_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); +} + + +/* ---------------------------------------------------------------- + * ExecAppendInitializeDSM + * + * Set up a Parallel Append descriptor. + * ---------------------------------------------------------------- + */ +void +ExecAppendInitializeDSM(AppendState *node, + ParallelContext *pcxt) +{ + ParallelAppendDesc padesc; + List *num_workers_list = ((Append*)node->ps.plan)->num_workers; + ListCell *lc; + int i; + + padesc = shm_toc_allocate(pcxt->toc, node->pappend_len); + SpinLockInit(&padesc->pa_mutex); + + Assert(node->as_nplans == list_length(num_workers_list)); + + i = 0; + foreach(lc, num_workers_list) + { + /* Initialize the max workers count for each subplan. */ + padesc->pa_info[i].pa_max_workers = lfirst_int(lc); + + /* + * Also, initialize current number of workers. Just setting all the + * number of workers to 0 is enough. The logic of choosing the next + * plan in workers will take care of initializing everything else. + */ + padesc->pa_info[i].pa_num_workers = 0; + + i++; + } + + shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id, padesc); + node->as_padesc = padesc; + + /* Choose the optimal subplan to be executed. */ + (void) parallel_append_next(node); +} + +/* ---------------------------------------------------------------- + * ExecAppendInitializeWorker + * + * Copy relevant information from TOC into planstate, and initialize + * whatever is required to choose and execute the optimal subplan. + * ---------------------------------------------------------------- + */ +void +ExecAppendInitializeWorker(AppendState *node, shm_toc *toc) +{ + node->as_padesc = shm_toc_lookup(toc, node->ps.plan->plan_node_id); + + /* Choose the optimal subplan to be executed. */ + (void) parallel_append_next(node); +} + +/* ---------------------------------------------------------------- + * set_finished + * + * Indicate that this child plan node is about to be finished, so no other + * workers should take up this node. Workers who are already executing + * this node will continue to do so, but workers looking for next nodes to + * pick up would skip this node after this function is called. It is + * possible that multiple workers call this function for the same node at + * the same time, because these workers were executing the same node and + * they finished with it at the same time. The spinlock is not for this + * purpose. The spinlock is used so that it does not change the + * pa_num_workers field while workers are choosing the next node. + * ---------------------------------------------------------------- + */ +static void +set_finished(ParallelAppendDesc padesc, int whichplan) +{ + elog(DEBUG2, "Parallelappend : pid %d : finishing plan %d", + MyProcPid, whichplan); + + SpinLockAcquire(&padesc->pa_mutex); + padesc->pa_info[whichplan].pa_num_workers = -1; + SpinLockRelease(&padesc->pa_mutex); +} + +/* ---------------------------------------------------------------- + * parallel_append_next + * + * Determine the optimal subplan that should be executed. The logic is to + * choose the subplan that is being executed by the least number of + * workers. + * + * Returns false if and only if all subplans are already finished + * processing. + * ---------------------------------------------------------------- + */ +static bool +parallel_append_next(AppendState *state) +{ + ParallelAppendDesc padesc = state->as_padesc; + int whichplan; + int min_whichplan = PA_INVALID_PLAN; + int min_workers = -1; /* Keep compiler quiet */ + + Assert(padesc != NULL); + + SpinLockAcquire(&padesc->pa_mutex); + + /* Choose the plan with the least number of workers */ + for (whichplan = 0; whichplan < state->as_nplans; whichplan++) + { + ParallelAppendInfo *painfo = &padesc->pa_info[whichplan]; + + /* Ignore plans that are already done processing */ + if (painfo->pa_num_workers == -1) + { + elog(DEBUG2, "ParallelAppend : pid %d : ignoring plan %d" + " since pa_num_workers is -1", + MyProcPid, whichplan); + continue; + } + + /* Ignore plans that are already being processed by max_workers */ + if (painfo->pa_num_workers == painfo->pa_max_workers) + { + elog(DEBUG2, "ParallelAppend : pid %d : ignoring plan %d," + " since reached max_worker count %d", + MyProcPid, whichplan, painfo->pa_max_workers); + continue; + } + + /* + * Keep track of the node with the least workers so far. For the very + * first plan, choose that one as the least-workers node. + */ + if (min_whichplan == PA_INVALID_PLAN || + painfo->pa_num_workers < min_workers) + { + min_whichplan = whichplan; + min_workers = painfo->pa_num_workers; + } + } + + /* Increment worker count for the chosen node, if at all we found one. */ + if (min_whichplan != PA_INVALID_PLAN) + padesc->pa_info[min_whichplan].pa_num_workers++; + + /* + * Save the chosen plan index. It can be PA_INVALID_PLAN, which means we + * are done with all nodes (Note : this meaning applies only to *parallel* + * append). + */ + state->as_whichplan = min_whichplan; + + /* + * Note: There is a chance that just after the child plan node is chosen + * here and spinlock released, some other worker finishes this node and + * calls set_finished(). In that case, this worker will go ahead and call + * ExecProcNode(child_node), which will return NULL tuple since it is + * already finished, and then once again this worker will try to choose + * next subplan; but this is ok : it's just an extra "choose_next_subplan" + * operation. + */ + SpinLockRelease(&padesc->pa_mutex); + elog(DEBUG2, "ParallelAppend : pid %d : Chosen plan : %d", + MyProcPid, min_whichplan); + + /* + * If we didn't find any node to work on, it means each subplan is either + * finished or has reached it's pa_max_workers. In such case, should this + * worker wait for some subplan to have its worker count drop below its + * pa_max_workers so that it can choose that subplan ? It turns out that + * it's not worth again finding a subplan to work on. Non-partial subplan + * anyway can have only one worker, and that worker will execute it to + * completion. For a partial subplan, if at all it reaches pa_max_workers, + * it's worker count will reduce only when it's workers find that there is + * nothing more to be executed, so there is no point taking up such node if + * it's worker count reduces. In conclusion, just stop executing once we + * don't find nodes to work on. Indicate the same by returning false. + */ + return (min_whichplan == PA_INVALID_PLAN ? false : true); } diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 05d8538..160860e 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -236,6 +236,7 @@ _copyAppend(const Append *from) * copy remainder of node */ COPY_NODE_FIELD(appendplans); + COPY_NODE_FIELD(num_workers); return newnode; } diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index b3802b4..a007f69 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -369,6 +369,7 @@ _outAppend(StringInfo str, const Append *node) _outPlanInfo(str, (const Plan *) node); WRITE_NODE_FIELD(appendplans); + WRITE_NODE_FIELD(num_workers); } static void diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c index d2f69fe..6142488 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1539,6 +1539,7 @@ _readAppend(void) ReadCommonPlan(&local_node->plan); READ_NODE_FIELD(appendplans); + READ_NODE_FIELD(num_workers); READ_DONE(); } diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c index eeacf81..648d04a 100644 --- a/src/backend/optimizer/path/allpaths.c +++ b/src/backend/optimizer/path/allpaths.c @@ -1232,14 +1232,50 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel, */ if (childrel->cheapest_total_path->param_info == NULL) subpaths = accumulate_append_subpath(subpaths, - childrel->cheapest_total_path); + childrel->cheapest_total_path); else subpaths_valid = false; /* Same idea, but for a partial plan. */ if (childrel->partial_pathlist != NIL) + { partial_subpaths = accumulate_append_subpath(partial_subpaths, linitial(childrel->partial_pathlist)); + } + else if (enable_parallelappend) + { + /* + * Extract the first unparameterized, parallel-safe one among the + * child paths. + */ + Path *parallel_safe_path = NULL; + foreach(lcp, childrel->pathlist) + { + Path *child_path = (Path *) lfirst(lcp); + if (child_path->parallel_safe && + child_path->param_info == NULL) + { + parallel_safe_path = child_path; + break; + } + } + + /* If we got one parallel-safe path, add it */ + if (parallel_safe_path) + { + partial_subpaths = + accumulate_append_subpath(partial_subpaths, + parallel_safe_path); + } + else + { + /* + * This child rel neither has a partial path, nor has a + * parallel-safe path. So drop the idea for partial append path. + */ + partial_subpaths_valid = false; + } + } else partial_subpaths_valid = false; @@ -1322,24 +1358,10 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel, if (partial_subpaths_valid) { AppendPath *appendpath; - ListCell *lc; - int parallel_workers = 0; - - /* - * Decide on the number of workers to request for this append path. - * For now, we just use the maximum value from among the members. It - * might be useful to use a higher number if the Append node were - * smart enough to spread out the workers, but it currently isn't. - */ - foreach(lc, partial_subpaths) - { - Path *path = lfirst(lc); + int parallel_workers; - parallel_workers = Max(parallel_workers, path->parallel_workers); - } - Assert(parallel_workers > 0); + parallel_workers = get_append_num_workers(partial_subpaths, NULL); - /* Generate a partial append path. */ appendpath = create_append_path(rel, partial_subpaths, NULL, parallel_workers); add_partial_path(rel, (Path *) appendpath); diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c index d01630f..0ac1feb 100644 --- a/src/backend/optimizer/path/costsize.c +++ b/src/backend/optimizer/path/costsize.c @@ -126,6 +126,7 @@ bool enable_nestloop = true; bool enable_material = true; bool enable_mergejoin = true; bool enable_hashjoin = true; +bool enable_parallelappend = true; typedef struct { @@ -1560,6 +1561,82 @@ cost_sort(Path *path, PlannerInfo *root, } /* + * cost_append + * Determines and returns the cost of an Append node. + * + * We charge nothing extra for the Append itself, which perhaps is too + * optimistic, but since it doesn't do any selection or projection, it is a + * pretty cheap node. + */ +void +cost_append(Path *path, List *subpaths, Relids required_outer) +{ + ListCell *l; + + path->rows = 0; + path->startup_cost = 0; + path->total_cost = 0; + + if (path->parallel_aware) + { + int parallel_divisor; + + foreach(l, subpaths) + { + Path *subpath = (Path *) lfirst(l); + + /* + * The subpath rows and cost is per worker. We need total count + * of each of the subpaths, so that we can determine the total cost + * of Append. + */ + parallel_divisor = get_parallel_divisor(subpath); + path->rows += (subpath->rows * parallel_divisor); + path->total_cost += (subpath->total_cost * parallel_divisor); + + /* + * Append would start returning tuples when the child node having + * lowest startup cost is done setting up. + */ + path->startup_cost = Min(path->startup_cost, + subpath->startup_cost); + + path->parallel_safe = path->parallel_safe && + subpath->parallel_safe; + + /* All child paths must have same parameterization */ + Assert(bms_equal(PATH_REQ_OUTER(subpath), required_outer)); + } + + /* The row count and cost should represent per-worker figures. */ + parallel_divisor = get_parallel_divisor(path); + path->rows = clamp_row_est(path->rows / parallel_divisor); + path->total_cost /= parallel_divisor; + + } + else + { + /* Compute rows and costs as sums of subplan rows and costs. */ + foreach(l, subpaths) + { + Path *subpath = (Path *) lfirst(l); + + path->rows += subpath->rows; + + path->total_cost += subpath->total_cost; + if (l == list_head(subpaths)) /* first node? */ + path->startup_cost = subpath->startup_cost; + + path->parallel_safe = path->parallel_safe && + subpath->parallel_safe; + + /* All child paths must have same parameterization */ + Assert(bms_equal(PATH_REQ_OUTER(subpath), required_outer)); + } + } +} + +/* * cost_merge_append * Determines and returns the cost of a MergeAppend node. * diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c index 997bdcf..d36808b 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -29,6 +29,7 @@ #include "nodes/nodeFuncs.h" #include "optimizer/clauses.h" #include "optimizer/cost.h" +#include "optimizer/pathnode.h" #include "optimizer/paths.h" #include "optimizer/placeholder.h" #include "optimizer/plancat.h" @@ -194,7 +195,7 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *appendplans, List *tlist); +static Append *make_append(List *appendplans, List *num_workers, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -962,6 +963,7 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) Append *plan; List *tlist = build_path_tlist(root, &best_path->path); List *subplans = NIL; + List *num_workers_list; ListCell *subpaths; /* @@ -1000,6 +1002,10 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) subplans = lappend(subplans, subplan); } + /* Get a list of number of workers for each of the subplans */ + (void) get_append_num_workers(best_path->subpaths, + &num_workers_list); + /* * XXX ideally, if there's just one child, we'd not bother to generate an * Append node but just return the single child. At the moment this does @@ -1007,7 +1013,7 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(subplans, tlist); + plan = make_append(subplans, num_workers_list, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -5009,7 +5015,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, List *tlist) +make_append(List *appendplans, List *num_workers, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -5019,6 +5025,7 @@ make_append(List *appendplans, List *tlist) plan->lefttree = NULL; plan->righttree = NULL; node->appendplans = appendplans; + node->num_workers = num_workers; return node; } diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index 3d33d46..6f4ab23 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -3334,10 +3334,7 @@ create_grouping_paths(PlannerInfo *root, paths = lappend(paths, path); } path = (Path *) - create_append_path(grouped_rel, - paths, - NULL, - 0); + create_append_path(grouped_rel, paths, NULL, 0); path->pathtarget = target; } else diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c index 3248296..7c244d9 100644 --- a/src/backend/optimizer/util/pathnode.c +++ b/src/backend/optimizer/util/pathnode.c @@ -1192,6 +1192,55 @@ create_tidscan_path(PlannerInfo *root, RelOptInfo *rel, List *tidquals, } /* + * get_append_num_workers + * Return the number of workers to request for partial append path. + * Optionally return the list of per-subplan worker count through + * 'per_subplan_workers' + */ +int +get_append_num_workers(List *subpaths, List **per_subplan_workers) +{ + ListCell *lc; + int total_workers = 0; + int subplan_workers; + int i = 0; + + if (per_subplan_workers) + *per_subplan_workers = NIL; + + foreach(lc, subpaths) + { + Path *subpath = lfirst(lc); + RelOptInfo *rel = subpath->parent; + + /* + * If this subpath is actually the cheapest partial path, take into + * account its parallel workers, else consider one worker since it's + * non-partial. + */ + if (rel->partial_pathlist != NIL && + (Path *) linitial(rel->partial_pathlist) == subpath) + subplan_workers = subpath->parallel_workers; + else + subplan_workers = 1; + + if (per_subplan_workers) + { + *per_subplan_workers = + lappend_int(*per_subplan_workers, subplan_workers); + } + total_workers += subplan_workers; + i++; + } + + /* In no case use more than max_parallel_workers_per_gather. */ + total_workers = Min(total_workers, + max_parallel_workers_per_gather); + + return total_workers; +} + +/* * create_append_path * Creates a path corresponding to an Append plan, returning the * pathnode. @@ -1203,46 +1252,21 @@ create_append_path(RelOptInfo *rel, List *subpaths, Relids required_outer, int parallel_workers) { AppendPath *pathnode = makeNode(AppendPath); - ListCell *l; pathnode->path.pathtype = T_Append; pathnode->path.parent = rel; pathnode->path.pathtarget = rel->reltarget; pathnode->path.param_info = get_appendrel_parampathinfo(rel, required_outer); - pathnode->path.parallel_aware = false; + pathnode->path.parallel_aware = + (enable_parallelappend && parallel_workers > 0); pathnode->path.parallel_safe = rel->consider_parallel; pathnode->path.parallel_workers = parallel_workers; pathnode->path.pathkeys = NIL; /* result is always considered * unsorted */ pathnode->subpaths = subpaths; - /* - * We don't bother with inventing a cost_append(), but just do it here. - * - * Compute rows and costs as sums of subplan rows and costs. We charge - * nothing extra for the Append itself, which perhaps is too optimistic, - * but since it doesn't do any selection or projection, it is a pretty - * cheap node. - */ - pathnode->path.rows = 0; - pathnode->path.startup_cost = 0; - pathnode->path.total_cost = 0; - foreach(l, subpaths) - { - Path *subpath = (Path *) lfirst(l); - - pathnode->path.rows += subpath->rows; - - if (l == list_head(subpaths)) /* first node? */ - pathnode->path.startup_cost = subpath->startup_cost; - pathnode->path.total_cost += subpath->total_cost; - pathnode->path.parallel_safe = pathnode->path.parallel_safe && - subpath->parallel_safe; - - /* All child paths must have same parameterization */ - Assert(bms_equal(PATH_REQ_OUTER(subpath), required_outer)); - } + cost_append(&pathnode->path, subpaths, required_outer); return pathnode; } diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 5d8fb2e..0efb5cf 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -901,6 +901,16 @@ static struct config_bool ConfigureNamesBool[] = true, NULL, NULL, NULL }, + { + {"enable_parallelappend", PGC_USERSET, QUERY_TUNING_METHOD, + gettext_noop("Enables the planner's use of parallel append plans."), + NULL + }, + &enable_parallelappend, + true, + NULL, NULL, NULL + }, + { {"geqo", PGC_USERSET, QUERY_TUNING_GEQO, diff --git a/src/include/executor/nodeAppend.h b/src/include/executor/nodeAppend.h index 6fb4662..e76027f 100644 --- a/src/include/executor/nodeAppend.h +++ b/src/include/executor/nodeAppend.h @@ -14,11 +14,15 @@ #ifndef NODEAPPEND_H #define NODEAPPEND_H +#include "access/parallel.h" #include "nodes/execnodes.h" extern AppendState *ExecInitAppend(Append *node, EState *estate, int eflags); extern TupleTableSlot *ExecAppend(AppendState *node); extern void ExecEndAppend(AppendState *node); extern void ExecReScanAppend(AppendState *node); +extern void ExecAppendEstimate(AppendState *node, ParallelContext *pcxt); +extern void ExecAppendInitializeDSM(AppendState *node, ParallelContext *pcxt); +extern void ExecAppendInitializeWorker(AppendState *node, shm_toc *toc); #endif /* NODEAPPEND_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 9f41bab..df5cfd5 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -21,6 +21,7 @@ #include "lib/pairingheap.h" #include "nodes/params.h" #include "nodes/plannodes.h" +#include "storage/spin.h" #include "utils/hsearch.h" #include "utils/reltrigger.h" #include "utils/sortsupport.h" @@ -1184,12 +1185,15 @@ typedef struct ModifyTableState * whichplan which plan is being executed (0 .. n-1) * ---------------- */ +struct ParallelAppendDescData; typedef struct AppendState { PlanState ps; /* its first field is NodeTag */ PlanState **appendplans; /* array of PlanStates for my inputs */ int as_nplans; int as_whichplan; + struct ParallelAppendDescData *as_padesc; /* parallel coordination info */ + Size pappend_len; /* size of parallel coordination info */ } AppendState; /* ---------------- diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index f72f7a8..8c06ee0 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -228,6 +228,7 @@ typedef struct Append { Plan plan; List *appendplans; + List *num_workers; } Append; /* ---------------- diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h index 72200fa..1c47b4b 100644 --- a/src/include/optimizer/cost.h +++ b/src/include/optimizer/cost.h @@ -66,6 +66,7 @@ extern bool enable_nestloop; extern bool enable_material; extern bool enable_mergejoin; extern bool enable_hashjoin; +extern bool enable_parallelappend; extern int constraint_exclusion; extern double clamp_row_est(double nrows); @@ -98,6 +99,7 @@ extern void cost_sort(Path *path, PlannerInfo *root, List *pathkeys, Cost input_cost, double tuples, int width, Cost comparison_cost, int sort_mem, double limit_tuples); +extern void cost_append(Path *path, List *subpaths, Relids required_outer); extern void cost_merge_append(Path *path, PlannerInfo *root, List *pathkeys, int n_streams, Cost input_startup_cost, Cost input_total_cost, diff --git a/src/include/optimizer/pathnode.h b/src/include/optimizer/pathnode.h index 53cad24..d5429c8 100644 --- a/src/include/optimizer/pathnode.h +++ b/src/include/optimizer/pathnode.h @@ -14,6 +14,7 @@ #ifndef PATHNODE_H #define PATHNODE_H +#include "nodes/bitmapset.h" #include "nodes/relation.h" @@ -62,8 +63,10 @@ extern BitmapOrPath *create_bitmap_or_path(PlannerInfo *root, List *bitmapquals); extern TidPath *create_tidscan_path(PlannerInfo *root, RelOptInfo *rel, List *tidquals, Relids required_outer); -extern AppendPath *create_append_path(RelOptInfo *rel, List *subpaths, - Relids required_outer, int parallel_workers); +extern int get_append_num_workers(List *subpaths, List **per_subplan_workers); +extern AppendPath *create_append_path(RelOptInfo *rel, + List *subpaths, Relids required_outer, + int parallel_workers); extern MergeAppendPath *create_merge_append_path(PlannerInfo *root, RelOptInfo *rel, List *subpaths, diff --git a/src/test/regress/expected/inherit.out b/src/test/regress/expected/inherit.out index a8c8b28..1ef8c4d 100644 --- a/src/test/regress/expected/inherit.out +++ b/src/test/regress/expected/inherit.out @@ -1337,6 +1337,7 @@ select min(1-id) from matest0; reset enable_indexscan; set enable_seqscan = off; -- plan with fewest seqscans should be merge +set enable_parallelappend = off; -- Don't let parallel-append interfere explain (verbose, costs off) select * from matest0 order by 1-id; QUERY PLAN ------------------------------------------------------------------ @@ -1403,6 +1404,7 @@ select min(1-id) from matest0; (1 row) reset enable_seqscan; +reset enable_parallelappend; drop table matest0 cascade; NOTICE: drop cascades to 3 other objects DETAIL: drop cascades to table matest1 diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out index 48fb80e..75c6103 100644 --- a/src/test/regress/expected/select_parallel.out +++ b/src/test/regress/expected/select_parallel.out @@ -17,9 +17,9 @@ explain (costs off) ----------------------------------------------------- Finalize Aggregate -> Gather - Workers Planned: 1 + Workers Planned: 4 -> Partial Aggregate - -> Append + -> Parallel Append -> Parallel Seq Scan on a_star -> Parallel Seq Scan on b_star -> Parallel Seq Scan on c_star diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out index d48abd7..7a303fa 100644 --- a/src/test/regress/expected/sysviews.out +++ b/src/test/regress/expected/sysviews.out @@ -70,20 +70,21 @@ select count(*) >= 0 as ok from pg_prepared_xacts; -- This is to record the prevailing planner enable_foo settings during -- a regression test run. select name, setting from pg_settings where name like 'enable%'; - name | setting -----------------------+--------- - enable_bitmapscan | on - enable_hashagg | on - enable_hashjoin | on - enable_indexonlyscan | on - enable_indexscan | on - enable_material | on - enable_mergejoin | on - enable_nestloop | on - enable_seqscan | on - enable_sort | on - enable_tidscan | on -(11 rows) + name | setting +-----------------------+--------- + enable_bitmapscan | on + enable_hashagg | on + enable_hashjoin | on + enable_indexonlyscan | on + enable_indexscan | on + enable_material | on + enable_mergejoin | on + enable_nestloop | on + enable_parallelappend | on + enable_seqscan | on + enable_sort | on + enable_tidscan | on +(12 rows) -- Test that the pg_timezone_names and pg_timezone_abbrevs views are -- more-or-less working. We can't test their contents in any great detail diff --git a/src/test/regress/sql/inherit.sql b/src/test/regress/sql/inherit.sql index a8b7eb1..e9203a1 100644 --- a/src/test/regress/sql/inherit.sql +++ b/src/test/regress/sql/inherit.sql @@ -462,11 +462,13 @@ select min(1-id) from matest0; reset enable_indexscan; set enable_seqscan = off; -- plan with fewest seqscans should be merge +set enable_parallelappend = off; -- Don't let parallel-append interfere explain (verbose, costs off) select * from matest0 order by 1-id; select * from matest0 order by 1-id; explain (verbose, costs off) select min(1-id) from matest0; select min(1-id) from matest0; reset enable_seqscan; +reset enable_parallelappend; drop table matest0 cascade;