diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index a1289e5..41d807c 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/nodeBitmapHeapscan.h" #include "executor/nodeCustom.h" #include "executor/nodeForeignscan.h" @@ -214,6 +215,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); @@ -278,6 +283,10 @@ ExecParallelInitializeDSM(PlanState *planstate, ExecForeignScanInitializeDSM((ForeignScanState *) planstate, d->pcxt); break; + case T_AppendState: + ExecAppendInitializeDSM((AppendState *) planstate, + d->pcxt); + break; case T_CustomScanState: ExecCustomScanInitializeDSM((CustomScanState *) planstate, d->pcxt); @@ -781,6 +790,9 @@ ExecParallelInitializeWorker(PlanState *planstate, shm_toc *toc) ExecForeignScanInitializeWorker((ForeignScanState *) planstate, toc); break; + case T_AppendState: + ExecAppendInitializeWorker((AppendState *) planstate, toc); + break; case T_CustomScanState: ExecCustomScanInitializeWorker((CustomScanState *) planstate, toc); diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index 6986cae..254e5ed 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -59,10 +59,48 @@ #include "executor/execdebug.h" #include "executor/nodeAppend.h" +#include "miscadmin.h" +#include "optimizer/cost.h" +#include "storage/spin.h" -static bool exec_append_initialize_next(AppendState *appendstate); +/* + * 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 ParallelAppendDescData +{ + slock_t pa_mutex; /* mutual exclusion to choose next subplan */ + int pa_first_plan; /* plan to choose while wrapping around plans */ + int pa_next_plan; /* next plan to choose by any worker */ + + /* + * pa_finished : workers currently executing the subplan. A worker which + * finishes a subplan should set pa_finished to true, so that no new + * worker picks this subplan. For non-partial subplan, a worker which picks + * up that subplan should immediately set to true, so as to make sure + * there are no more than 1 worker assigned to this subplan. + */ + bool pa_finished[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 bool exec_append_initialize_next(AppendState *appendstate); +static void set_finished(ParallelAppendDesc padesc, int whichplan); +static bool parallel_append_next(AppendState *state); +static bool leader_next(AppendState *state); +static int goto_next_plan(int curplan, int first_plan, int last_plan); + /* ---------------------------------------------------------------- * exec_append_initialize_next * @@ -77,6 +115,27 @@ 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) + { + /* Backward scan is not supported by parallel-aware plans */ + Assert(ScanDirectionIsForward(appendstate->ps.state->es_direction)); + + 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; @@ -176,10 +235,9 @@ ExecInitAppend(Append *node, EState *estate, int eflags) appendstate->ps.ps_ProjInfo = NULL; /* - * initialize to scan first subplan + * In case it's a sequential Append, initialize to scan first subplan. */ appendstate->as_whichplan = 0; - exec_append_initialize_next(appendstate); return appendstate; } @@ -199,6 +257,14 @@ ExecAppend(AppendState *node) TupleTableSlot *result; /* + * Check if we are already finished plans from parallel append. This + * can happen if all the subplans are finished when this worker + * has not even started returning tuples. + */ + if (node->as_padesc && node->as_whichplan == PA_INVALID_PLAN) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); + + /* * figure out which subplan we are currently processing */ subnode = node->appendplans[node->as_whichplan]; @@ -219,14 +285,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); @@ -266,6 +336,7 @@ void ExecReScanAppend(AppendState *node) { int i; + ParallelAppendDesc padesc = node->as_padesc; for (i = 0; i < node->as_nplans; i++) { @@ -285,6 +356,284 @@ ExecReScanAppend(AppendState *node) if (subnode->chgParam == NULL) ExecReScan(subnode); } + + if (padesc) + { + padesc->pa_first_plan = padesc->pa_next_plan = 0; + memset(padesc->pa_finished, 0, sizeof(bool) * node->as_nplans); + } + node->as_whichplan = 0; - exec_append_initialize_next(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_finished), + sizeof(bool) * 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; + + padesc = shm_toc_allocate(pcxt->toc, node->pappend_len); + SpinLockInit(&padesc->pa_mutex); + + /* + * Just setting all the number of workers to 0 is enough. The logic + * of choosing the next plan in workers will take care of everything + * else. + */ + memset(padesc->pa_finished, 0, sizeof(bool) * node->as_nplans); + + 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) +{ + SpinLockAcquire(&padesc->pa_mutex); + padesc->pa_finished[whichplan] = true; + SpinLockRelease(&padesc->pa_mutex); +} + +/* ---------------------------------------------------------------- + * parallel_append_next + * + * Determine the next subplan that should be executed. Each worker uses a + * shared next_subplan counter index to start looking for unfinished plan, + * executes the subplan, then shifts ahead this counter to the next + * subplan, so that other workers know which next plan to choose. This + * way, workers choose the subplans in round robin order, and thus they + * get evenly distributed among the subplans. + * + * 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 initial_plan; + int first_partial_plan = ((Append *)state->ps.plan)->first_partial_plan; + bool found; + + Assert(padesc != NULL); + + /* The parallel leader chooses its next subplan differently */ + if (!IsParallelWorker()) + return leader_next(state); + + SpinLockAcquire(&padesc->pa_mutex); + + /* Make a note of which subplan we have started with */ + initial_plan = padesc->pa_next_plan; + + /* + * Keep going to the next plan until we find an unfinished one. In the + * process, also keep track of the first unfinished non-partial subplan. As + * the non-partial subplans are taken one by one, the first unfinished + * subplan index will shift ahead, so that we don't have to visit the + * finished non-partial ones anymore. + */ + + found = false; + for (whichplan = initial_plan; whichplan != PA_INVALID_PLAN;) + { + /* + * Ignore plans that are already done processing. These also include + * non-partial subplans which have already been taken by a worker. + */ + if (!padesc->pa_finished[whichplan]) + { + found = true; + break; + } + + /* Either go to the next plan, or wrap around to the first one */ + whichplan = goto_next_plan(whichplan, padesc->pa_first_plan, + state->as_nplans - 1); + + /* + * If we have wrapped around and returned to the same index again, we + * are done scanning. + */ + if (whichplan == initial_plan) + break; + } + + /* If we didn't find any plan to execute, stop executing. */ + if (!found) + padesc->pa_next_plan = state->as_whichplan = PA_INVALID_PLAN; + else + { + /* + * If this a non-partial plan, immediately mark it finished, and shift + * ahead pa_first_plan. + */ + if (whichplan < first_partial_plan) + { + padesc->pa_finished[whichplan] = true; + padesc->pa_first_plan = whichplan + 1; + } + + /* + * Set the chosen plan, and the next plan to be picked by other + * workers. + */ + state->as_whichplan = whichplan; + padesc->pa_next_plan = goto_next_plan(whichplan, padesc->pa_first_plan, + state->as_nplans - 1); + } + + SpinLockRelease(&padesc->pa_mutex); + + /* + * 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. + */ + + return found; +} + +/* ---------------------------------------------------------------- + * leader_next + * + * To be used only if it's a parallel leader. The backend should scan + * backwards from the last plan. This is to prevent it from taking up + * the most expensive non-partial plan, i.e. the first subplan. + * ---------------------------------------------------------------- + */ +static bool +leader_next(AppendState *state) +{ + ParallelAppendDesc padesc = state->as_padesc; + int first_plan; + int whichplan; + int first_partial_plan = ((Append *)state->ps.plan)->first_partial_plan; + + /* The parallel leader should start from the last subplan. */ + + SpinLockAcquire(&padesc->pa_mutex); + first_plan = padesc->pa_first_plan; + + for (whichplan = state->as_nplans - 1; whichplan >= first_plan; + whichplan--) + { + if (!padesc->pa_finished[whichplan]) + { + /* If this a non-partial plan, immediately mark it finished */ + if (whichplan < first_partial_plan) + padesc->pa_finished[whichplan] = true; + + break; + } + } + + SpinLockRelease(&padesc->pa_mutex); + + /* Return false only if we didn't find any plan to execute */ + if (whichplan < first_plan) + { + state->as_whichplan = PA_INVALID_PLAN; + return false; + } + else + { + state->as_whichplan = whichplan; + return true; + } +} + +/* ---------------------------------------------------------------- + * goto_next_plan + * + * Either go to the next index, or wrap around to the first unfinished one. + * ---------------------------------------------------------------- + */ +static int goto_next_plan(int curplan, int first_plan, int last_plan) +{ + Assert(curplan <= last_plan); + + if (curplan < last_plan) + return curplan + 1; + else + { + /* + * We are already at the last plan. If the first_plan itsef is the last + * plan or if it is past the last plan, that means there is no next + * plan remaining. Return Invalid. + */ + if (first_plan >= last_plan) + return PA_INVALID_PLAN; + + return first_plan; + } } diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 25fd051..7ee0bb8 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_SCALAR_FIELD(first_partial_plan); return newnode; } diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index 7418fbe..38ade5f 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_INT_FIELD(first_partial_plan); } static void diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c index d3bbc02..fa1487a 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1565,6 +1565,7 @@ _readAppend(void) ReadCommonPlan(&local_node->plan); READ_NODE_FIELD(appendplans); + READ_INT_FIELD(first_partial_plan); READ_DONE(); } diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c index 43bfd23..094809d 100644 --- a/src/backend/optimizer/path/allpaths.c +++ b/src/backend/optimizer/path/allpaths.c @@ -99,7 +99,11 @@ static void generate_mergeappend_paths(PlannerInfo *root, RelOptInfo *rel, static Path *get_cheapest_parameterized_child_path(PlannerInfo *root, RelOptInfo *rel, Relids required_outer); +static List *paths_insert_sorted_by_cost(List *paths, List *insert_paths); static List *accumulate_append_subpath(List *subpaths, Path *path); +static List *accumulate_partialappend_subpath(List *partial_subpaths, + Path *subpath, bool is_partial, + List **nonpartial_subpaths); static void set_subquery_pathlist(PlannerInfo *root, RelOptInfo *rel, Index rti, RangeTblEntry *rte); static void set_function_pathlist(PlannerInfo *root, RelOptInfo *rel, @@ -1255,6 +1259,7 @@ add_paths_to_append_rel(PlannerInfo *root, RelOptInfo *rel, List *subpaths = NIL; bool subpaths_valid = true; List *partial_subpaths = NIL; + List *nonpartial_subpaths = NIL; bool partial_subpaths_valid = true; List *all_child_pathkeys = NIL; List *all_child_outers = NIL; @@ -1277,14 +1282,42 @@ add_paths_to_append_rel(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)); + partial_subpaths = accumulate_partialappend_subpath( + partial_subpaths, + linitial(childrel->partial_pathlist), + true, &nonpartial_subpaths); + else if (enable_parallelappend) + { + /* + * Extract the cheapest unparameterized, parallel-safe one among + * the child paths. + */ + Path *parallel_safe_path = + get_cheapest_parallel_safe_total_inner(childrel->pathlist); + + /* If we got one parallel-safe path, add it */ + if (parallel_safe_path) + { + partial_subpaths = + accumulate_partialappend_subpath(partial_subpaths, + parallel_safe_path, false, + &nonpartial_subpaths); + } + 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; @@ -1359,7 +1392,8 @@ add_paths_to_append_rel(PlannerInfo *root, RelOptInfo *rel, * if we have zero or one live subpath due to constraint exclusion.) */ if (subpaths_valid) - add_path(rel, (Path *) create_append_path(rel, subpaths, NULL, 0)); + add_path(rel, (Path *) create_append_path(rel, subpaths, NIL, + NULL, 0)); /* * Consider an append of partial unordered, unparameterized partial paths. @@ -1367,26 +1401,14 @@ add_paths_to_append_rel(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, + nonpartial_subpaths); - /* Generate a partial append path. */ - appendpath = create_append_path(rel, partial_subpaths, NULL, - parallel_workers); + appendpath = create_append_path(rel, nonpartial_subpaths, + partial_subpaths, + NULL, parallel_workers); add_partial_path(rel, (Path *) appendpath); } @@ -1438,7 +1460,8 @@ add_paths_to_append_rel(PlannerInfo *root, RelOptInfo *rel, if (subpaths_valid) add_path(rel, (Path *) - create_append_path(rel, subpaths, required_outer, 0)); + create_append_path(rel, subpaths, NIL, + required_outer, 0)); } } @@ -1652,6 +1675,123 @@ accumulate_append_subpath(List *subpaths, Path *path) } /* + * paths_insert_sorted_by_cost + * + * Insert each of the paths of 'insert_paths' into the right position in + * 'paths' so that 'paths' remains sorted by descending order of total cost. + * + * Return a possibly modified 'paths'. + */ +static List * +paths_insert_sorted_by_cost(List *paths, List *insert_paths) +{ + ListCell *lci; + + foreach(lci, insert_paths) + { + Path *insert_path = (Path *) lfirst(lci); + ListCell *lcp; + ListCell *prev; + + /* Insert this 'insert_path' into 'paths' */ + prev = NULL; + foreach(lcp, paths) + { + Path *path = (Path *) lfirst(lcp); + + /* found the right position ? */ + if (path->total_cost < insert_path->total_cost) + break; + prev = lcp; + } + + /* Inserting before the first element ? */ + if (prev == NULL) + paths = lcons(insert_path, paths); + else + (void) lappend_cell(paths, prev, insert_path); + } + + return paths; +} + +/* + * accumulate_partialappend_subpath: + * Add a subpath to the list being built for a partial Append. + * + * This is same as accumulate_append_subpath, except that two separate lists + * are created, one containing only partial subpaths, and the other containing + * only non-partial subpaths. Also, the non-partial paths are kept ordered + * by descending total cost. + * + * is_partial is true if the subpath being added is a partial subpath. + */ +static List * +accumulate_partialappend_subpath(List *partial_subpaths, + Path *subpath, bool is_partial, + List **nonpartial_subpaths) +{ + /* list_copy is important here to avoid sharing list substructure */ + + if (IsA(subpath, AppendPath)) + { + AppendPath *apath = (AppendPath *) subpath; + List *apath_partial_paths; + List *apath_nonpartial_paths; + + /* Split the Append subpaths into partial and non-partial paths */ + apath_nonpartial_paths = list_truncate(list_copy(apath->subpaths), + apath->first_partial_path); + apath_partial_paths = list_copy_tail(apath->subpaths, + apath->first_partial_path); + + /* Add non-partial subpaths, if any. */ + *nonpartial_subpaths = + paths_insert_sorted_by_cost(*nonpartial_subpaths, + apath_nonpartial_paths); + + /* Add partial subpaths, if any. */ + return list_concat(partial_subpaths, apath_partial_paths); + } + else if (IsA(subpath, MergeAppendPath)) + { + MergeAppendPath *mpath = (MergeAppendPath *) subpath; + + /* + * If at all MergeAppend is partial, all its child plans have to be + * partial : we don't currently support a mix of partial and + * non-partial MergeAppend subpaths. + */ + if (is_partial) + return list_concat(partial_subpaths, list_copy(mpath->subpaths)); + else + { + /* + * Since MergePath itself is non-partial, treat all its subpaths + * non-partial. + */ + *nonpartial_subpaths = + paths_insert_sorted_by_cost(*nonpartial_subpaths, + mpath->subpaths); + return partial_subpaths; + } + } + else + { + /* Just add it to the right list depending upon whether it's partial */ + if (is_partial) + return lappend(partial_subpaths, subpath); + else + { + *nonpartial_subpaths = + paths_insert_sorted_by_cost(*nonpartial_subpaths, + list_make1(subpath)); + return partial_subpaths; + } + } +} + +/* * set_dummy_rel_pathlist * Build a dummy path for a relation that's been excluded by constraints * @@ -1671,7 +1811,7 @@ set_dummy_rel_pathlist(RelOptInfo *rel) rel->pathlist = NIL; rel->partial_pathlist = NIL; - add_path(rel, (Path *) create_append_path(rel, NIL, NULL, 0)); + add_path(rel, (Path *) create_append_path(rel, NIL, NIL, NULL, 0)); /* * We set the cheapest path immediately, to ensure that IS_DUMMY_REL() diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c index a129d1e..e8df075 100644 --- a/src/backend/optimizer/path/costsize.c +++ b/src/backend/optimizer/path/costsize.c @@ -127,6 +127,7 @@ bool enable_material = true; bool enable_mergejoin = true; bool enable_hashjoin = true; bool enable_gathermerge = true; +bool enable_parallelappend = true; typedef struct { @@ -1704,6 +1705,98 @@ 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, int num_nonpartial_subpaths) +{ + ListCell *l; + + path->rows = 0; + path->startup_cost = 0; + path->total_cost = 0; + + if (path->parallel_aware) + { + int parallel_divisor; + Cost highest_nonpartial_cost = 0; + int worker; + + /* + * Make a note of the cost of first non-partial subpath, i.e. the first + * one in the list, if at all there are any non-partial subpaths. + */ + if (num_nonpartial_subpaths > 0) + highest_nonpartial_cost = ((Path *) linitial(subpaths))->total_cost; + + worker = 1; + 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. We don't consider non-partial paths separately. The + * parallel_divisor for non-partial paths is 1, and so overall we + * get a good approximation of per-worker cost. + */ + 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. We consider only the + * first few subplans that immediately get a worker assigned. + */ + if (worker <= path->parallel_workers) + { + path->startup_cost = Min(path->startup_cost, + subpath->startup_cost); + worker++; + } + } + + /* 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; + + /* + * No matter how fast the partial plans finish, the Append path is + * going to take at least the time needed for the costliest non-partial + * path to finish. This is actually an approximation. We can even + * consider all the other non-partial plans and how workers would get + * scheduled to determine the cost of finishing the non-partial paths. + * But we anyway can't calculate the total cost exactly, especially + * because we can't determine exactly when some of the workers would + * start executing partial plans. + */ + path->total_cost = Max(highest_nonpartial_cost, path->total_cost); + } + 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; + } + } +} + +/* * cost_merge_append * Determines and returns the cost of a MergeAppend node. * diff --git a/src/backend/optimizer/path/joinrels.c b/src/backend/optimizer/path/joinrels.c index 0551668..401d95a 100644 --- a/src/backend/optimizer/path/joinrels.c +++ b/src/backend/optimizer/path/joinrels.c @@ -1217,7 +1217,7 @@ mark_dummy_rel(RelOptInfo *rel) rel->partial_pathlist = NIL; /* Set up the dummy path */ - add_path(rel, (Path *) create_append_path(rel, NIL, NULL, 0)); + add_path(rel, (Path *) create_append_path(rel, NIL, NIL, NULL, 0)); /* Set or update cheapest_total_path and related fields */ set_cheapest(rel); diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c index 89e1946..1702015 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -199,7 +199,8 @@ 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, int first_partial_plan, + List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -1026,7 +1027,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, best_path->first_partial_path, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -5161,7 +5162,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, List *tlist) +make_append(List *appendplans, int first_partial_plan, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -5171,6 +5172,7 @@ make_append(List *appendplans, List *tlist) plan->lefttree = NULL; plan->righttree = NULL; node->appendplans = appendplans; + node->first_partial_plan = first_partial_plan; return node; } diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index 02286d9..529d91f 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -3345,10 +3345,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, NIL, NULL, 0); path->pathtarget = target; } else diff --git a/src/backend/optimizer/prep/prepunion.c b/src/backend/optimizer/prep/prepunion.c index 1389db1..e1d70a8 100644 --- a/src/backend/optimizer/prep/prepunion.c +++ b/src/backend/optimizer/prep/prepunion.c @@ -566,7 +566,7 @@ generate_union_path(SetOperationStmt *op, PlannerInfo *root, /* * Append the child results together. */ - path = (Path *) create_append_path(result_rel, pathlist, NULL, 0); + path = (Path *) create_append_path(result_rel, pathlist, NIL, NULL, 0); /* We have to manually jam the right tlist into the path; ick */ path->pathtarget = create_pathtarget(root, tlist); @@ -678,7 +678,7 @@ generate_nonunion_path(SetOperationStmt *op, PlannerInfo *root, /* * Append the child results together. */ - path = (Path *) create_append_path(result_rel, pathlist, NULL, 0); + path = (Path *) create_append_path(result_rel, pathlist, NIL, NULL, 0); /* We have to manually jam the right tlist into the path; ick */ path->pathtarget = create_pathtarget(root, tlist); diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c index 8ce772d..6475e23 100644 --- a/src/backend/optimizer/util/pathnode.c +++ b/src/backend/optimizer/util/pathnode.c @@ -1193,6 +1193,70 @@ create_tidscan_path(PlannerInfo *root, RelOptInfo *rel, List *tidquals, } /* + * get_append_num_workers + * Return the number of workers to request for partial append path. + */ +int +get_append_num_workers(List *partial_subpaths, List *nonpartial_subpaths) +{ + ListCell *lc; + double log2w; + int num_workers; + int max_per_plan_workers; + + /* + * log2(number_of_subpaths)+1 formula seems to give an appropriate number of + * workers for Append path either having high number of children (> 100) or + * having all non-partial subpaths or subpaths with 1-2 parallel_workers. + * Whereas, if the subpaths->parallel_workers is high, this formula is not + * suitable, because it does not take into account per-subpath workers. + * For e.g., with workers (2, 8, 8), the Append workers should be at least + * 8, whereas the formula gives 2. In this case, it seems better to follow + * the method used for calculating parallel_workers of an unpartitioned + * table : log3(table_size). So we treat the UNION query as if the data + * belongs to a single unpartitioned table, and then derive its workers. So + * it will be : logb(b^w1 + b^w2 + b^w3) where w1, w2.. are per-subplan + * workers and b is some logarithmic base such as 2 or 3. It turns out that + * this evaluates to a value just a bit greater than max(w1,w2, w3). So, we + * just use the maximum of workers formula. But this formula gives too few + * workers when all paths have single worker (meaning they are non-partial) + * For e.g. with workers : (1, 1, 1, 1, 1, 1), it is better to allocate 3 + * workers, whereas this method allocates only 1. + * So we use whichever method that gives higher number of workers. + */ + + /* Get log2(num_subpaths) i.e. ln(num_subpaths) / ln(2) */ + log2w = log(list_length(partial_subpaths) + + list_length(nonpartial_subpaths)) + / 0.693 ; + + /* Avoid further calculations if we already crossed max workers limit */ + if (max_parallel_workers_per_gather <= log2w + 1) + return max_parallel_workers_per_gather; + + + /* + * Get the parallel_workers value of the partial subpath having the highest + * parallel_workers. + */ + max_per_plan_workers = 1; + foreach(lc, partial_subpaths) + { + Path *subpath = lfirst(lc); + max_per_plan_workers = Max(max_per_plan_workers, + subpath->parallel_workers); + } + + /* Choose the higher of the results of the two formulae */ + num_workers = rint(Max(log2w, max_per_plan_workers) + 1); + + /* In no case use more than max_parallel_workers_per_gather workers. */ + num_workers = Min(num_workers, max_parallel_workers_per_gather); + + return num_workers; +} + +/* * create_append_path * Creates a path corresponding to an Append plan, returning the * pathnode. @@ -1200,8 +1264,9 @@ create_tidscan_path(PlannerInfo *root, RelOptInfo *rel, List *tidquals, * Note that we must handle subpaths = NIL, representing a dummy access path. */ AppendPath * -create_append_path(RelOptInfo *rel, List *subpaths, Relids required_outer, - int parallel_workers) +create_append_path(RelOptInfo *rel, + List *subpaths, List *partial_subpaths, + Relids required_outer, int parallel_workers) { AppendPath *pathnode = makeNode(AppendPath); ListCell *l; @@ -1211,40 +1276,29 @@ create_append_path(RelOptInfo *rel, List *subpaths, Relids required_outer, 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; + pathnode->first_partial_path = list_length(subpaths); + pathnode->subpaths = list_concat(subpaths, partial_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; + subpath->parallel_safe; /* All child paths must have same parameterization */ Assert(bms_equal(PATH_REQ_OUTER(subpath), required_outer)); } + cost_append(&pathnode->path, pathnode->subpaths, + pathnode->first_partial_path); + return pathnode; } diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 4feb26a..4f21c2e 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -911,6 +911,15 @@ 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 f856f60..c822cf2 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" @@ -1187,12 +1188,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 b880dc1..79048af 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -228,6 +228,7 @@ typedef struct Append { Plan plan; List *appendplans; + int first_partial_plan; } Append; /* ---------------- diff --git a/src/include/nodes/relation.h b/src/include/nodes/relation.h index 05d6f07..eea8c72 100644 --- a/src/include/nodes/relation.h +++ b/src/include/nodes/relation.h @@ -1108,15 +1108,22 @@ typedef struct CustomPath * AppendPath represents an Append plan, ie, successive execution of * several member plans. * + * For partial Append, 'subpaths' contains non-partial subpaths followed by + * partial subpaths. + * * Note: it is possible for "subpaths" to contain only one, or even no, * elements. These cases are optimized during create_append_plan. * In particular, an AppendPath with no subpaths is a "dummy" path that * is created to represent the case that a relation is provably empty. + * */ typedef struct AppendPath { Path path; List *subpaths; /* list of component Paths */ + + /* Index of first partial path in subpaths */ + int first_partial_path; } AppendPath; #define IS_DUMMY_PATH(p) \ diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h index d9a9b12..43dc72f 100644 --- a/src/include/optimizer/cost.h +++ b/src/include/optimizer/cost.h @@ -67,6 +67,7 @@ extern bool enable_material; extern bool enable_mergejoin; extern bool enable_hashjoin; extern bool enable_gathermerge; +extern bool enable_parallelappend; extern int constraint_exclusion; extern double clamp_row_est(double nrows); @@ -103,6 +104,8 @@ 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, + int num_nonpartial_subpaths); 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 373c722..9622d2f 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" @@ -63,8 +64,11 @@ 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 *partial_subpaths, + List *nonpartial_subpaths); +extern AppendPath *create_append_path(RelOptInfo *rel, + List *subpaths, List *partial_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 6494b20..36be3a7 100644 --- a/src/test/regress/expected/inherit.out +++ b/src/test/regress/expected/inherit.out @@ -1343,6 +1343,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 ------------------------------------------------------------------ @@ -1409,6 +1410,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 038a62e..6ffe23d 100644 --- a/src/test/regress/expected/select_parallel.out +++ b/src/test/regress/expected/select_parallel.out @@ -11,15 +11,16 @@ set parallel_setup_cost=0; set parallel_tuple_cost=0; set min_parallel_table_scan_size=0; set max_parallel_workers_per_gather=4; +-- test Parallel Append. explain (costs off) - select count(*) from a_star; + select round(avg(aa)), sum(aa) from a_star; QUERY PLAN ----------------------------------------------------- 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 @@ -28,12 +29,40 @@ explain (costs off) -> Parallel Seq Scan on f_star (11 rows) -select count(*) from a_star; - count -------- - 50 +select round(avg(aa)), sum(aa) from a_star; + round | sum +-------+----- + 14 | 355 +(1 row) + +-- Mix of partial and non-partial subplans. +alter table c_star set (parallel_workers = 0); +alter table d_star set (parallel_workers = 0); +explain (costs off) + select round(avg(aa)), sum(aa) from a_star; + QUERY PLAN +----------------------------------------------------- + Finalize Aggregate + -> Gather + Workers Planned: 4 + -> Partial Aggregate + -> Parallel Append + -> Seq Scan on d_star + -> Seq Scan on c_star + -> Parallel Seq Scan on a_star + -> Parallel Seq Scan on b_star + -> Parallel Seq Scan on e_star + -> Parallel Seq Scan on f_star +(11 rows) + +select round(avg(aa)), sum(aa) from a_star; + round | sum +-------+----- + 14 | 355 (1 row) +alter table c_star reset (parallel_workers); +alter table d_star reset (parallel_workers); -- test that parallel_restricted function doesn't run in worker alter table tenk1 set (parallel_workers = 4); explain (verbose, costs off) diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out index 568b783..97a9843 100644 --- a/src/test/regress/expected/sysviews.out +++ b/src/test/regress/expected/sysviews.out @@ -70,21 +70,22 @@ 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_gathermerge | 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 -(12 rows) + name | setting +-----------------------+--------- + enable_bitmapscan | on + enable_gathermerge | 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 +(13 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 e3e9e34..810070a 100644 --- a/src/test/regress/sql/inherit.sql +++ b/src/test/regress/sql/inherit.sql @@ -463,11 +463,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; diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql index 9311a77..0623319 100644 --- a/src/test/regress/sql/select_parallel.sql +++ b/src/test/regress/sql/select_parallel.sql @@ -15,9 +15,18 @@ set parallel_tuple_cost=0; set min_parallel_table_scan_size=0; set max_parallel_workers_per_gather=4; +-- test Parallel Append. explain (costs off) - select count(*) from a_star; -select count(*) from a_star; + select round(avg(aa)), sum(aa) from a_star; +select round(avg(aa)), sum(aa) from a_star; +-- Mix of partial and non-partial subplans. +alter table c_star set (parallel_workers = 0); +alter table d_star set (parallel_workers = 0); +explain (costs off) + select round(avg(aa)), sum(aa) from a_star; +select round(avg(aa)), sum(aa) from a_star; +alter table c_star reset (parallel_workers); +alter table d_star reset (parallel_workers); -- test that parallel_restricted function doesn't run in worker alter table tenk1 set (parallel_workers = 4);