Re: Parallel Append implementation

Hi,

On 2017-03-24 21:32:57 +0530, Amit Khandekar wrote:
> diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
> index a107545..e9e8676 100644
> --- a/src/backend/executor/nodeAppend.c
> +++ b/src/backend/executor/nodeAppend.c
> @@ -59,9 +59,47 @@
>
> #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 ParallelAppendDescData
> +{
> + LWLock pa_lock; /* 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;

Pointer hiding typedefs make this Andres sad.

> @@ -291,6 +362,276 @@ 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);
> + }
> +

Is it actually guaranteed that none of the parallel workers are doing
something at that point?

> +/* ----------------------------------------------------------------
> + * exec_append_parallel_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
> +exec_append_parallel_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);
> +
> + /* Backward scan is not supported by parallel-aware plans */
> + Assert(ScanDirectionIsForward(state->ps.state->es_direction));
> +
> + /* The parallel leader chooses its next subplan differently */
> + if (!IsParallelWorker())
> + return exec_append_leader_next(state);

It's a bit weird that the leader's case does is so separate, and does
it's own lock acquisition.

> + 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;
> + }
> +
> + /*
> + * Note: There is a chance that just after the child plan node is
> + * chosen above, some other worker finishes this node and sets
> + * pa_finished to true. 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.
> + */

IIRC not all node types are safe against being executed again when
they've previously returned NULL. That's why e.g. nodeMaterial.c
contains the following blurb:
/*
* If necessary, try to fetch another row from the subplan.
*
* Note: the eof_underlying state variable exists to short-circuit further
* subplan calls. It's not optional, unfortunately, because some plan
* node types are not robust about being called again when they've already
* returned NULL.
*/

> + 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.
> + */

Why is that?

> +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),

That's the per-subplan workers for three subplans? That's not
necessarily clear.

> 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

Which "UNION query"?

> + * 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) */
> + log2w = fls(list_length(partial_subpaths) +
> + list_length(nonpartial_subpaths));
> +
> + /* 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;
> +}

Hm. I'm not really convinced by the logic here. Wouldn't it be better
to try to compute the minimum total cost across all workers for
1..#max_workers for the plans in an iterative manner? I.e. try to map
each of the subplans to 1 (if non-partial) or N workers (partial) using
some fitting algorith (e.g. always choosing the worker(s) that currently
have the least work assigned). I think the current algorithm doesn't
lead to useful #workers for e.g. cases with a lot of non-partial,
high-startup plans - imo a quite reasonable scenario.

I'm afraid this is too late for v10 - do you agree?

- Andres