execParallel.h File Reference

#include "access/parallel.h"
#include "nodes/execnodes.h"
#include "nodes/parsenodes.h"
#include "nodes/plannodes.h"
#include "utils/dsa.h"

Include dependency graph for execParallel.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ParallelExecutorInfo

Typedefs
typedef struct SharedExecutorInstrumentation	SharedExecutorInstrumentation
typedef struct ParallelExecutorInfo	ParallelExecutorInfo

Functions
ParallelExecutorInfo *	ExecInitParallelPlan (PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed)
void	ExecParallelCreateReaders (ParallelExecutorInfo *pei)
void	ExecParallelFinish (ParallelExecutorInfo *pei)
void	ExecParallelCleanup (ParallelExecutorInfo *pei)
void	ExecParallelReinitialize (PlanState *planstate, ParallelExecutorInfo *pei, Bitmapset *sendParams)
void	ParallelQueryMain (dsm_segment *seg, shm_toc *toc)

Typedef Documentation

ParallelExecutorInfo

typedef struct ParallelExecutorInfo ParallelExecutorInfo

SharedExecutorInstrumentation

typedef struct SharedExecutorInstrumentation SharedExecutorInstrumentation

Definition at line 1 of file execParallel.h.

Function Documentation

ExecInitParallelPlan()

ParallelExecutorInfo* ExecInitParallelPlan	(	PlanState *	planstate,
		EState *	estate,
		Bitmapset *	sendParams,
		int	nworkers,
		int64	tuples_needed
	)

Definition at line 590 of file execParallel.c.

{
    ParallelExecutorInfo *pei;
    ParallelContext *pcxt;
    ExecParallelEstimateContext e;
    ExecParallelInitializeDSMContext d;
    FixedParallelExecutorState *fpes;
    char       *pstmt_data;
    char       *pstmt_space;
    char       *paramlistinfo_space;
    BufferUsage *bufusage_space;
    WalUsage   *walusage_space;
    SharedExecutorInstrumentation *instrumentation = NULL;
    SharedJitInstrumentation *jit_instrumentation = NULL;
    int         pstmt_len;
    int         paramlistinfo_len;
    int         instrumentation_len = 0;
    int         jit_instrumentation_len = 0;
    int         instrument_offset = 0;
    Size        dsa_minsize = dsa_minimum_size();
    char       *query_string;
    int         query_len;
 
    /*
     * Force any initplan outputs that we're going to pass to workers to be
     * evaluated, if they weren't already.
     *
     * For simplicity, we use the EState's per-output-tuple ExprContext here.
     * That risks intra-query memory leakage, since we might pass through here
     * many times before that ExprContext gets reset; but ExecSetParamPlan
     * doesn't normally leak any memory in the context (see its comments), so
     * it doesn't seem worth complicating this function's API to pass it a
     * shorter-lived ExprContext.  This might need to change someday.
     */
    ExecSetParamPlanMulti(sendParams, GetPerTupleExprContext(estate));
 
    /* Allocate object for return value. */
    pei = palloc0(sizeof(ParallelExecutorInfo));
    pei->finished = false;
    pei->planstate = planstate;
 
    /* Fix up and serialize plan to be sent to workers. */
    pstmt_data = ExecSerializePlan(planstate->plan, estate);
 
    /* Create a parallel context. */
    pcxt = CreateParallelContext("postgres", "ParallelQueryMain", nworkers);
    pei->pcxt = pcxt;
 
    /*
     * Before telling the parallel context to create a dynamic shared memory
     * segment, we need to figure out how big it should be.  Estimate space
     * for the various things we need to store.
     */
 
    /* Estimate space for fixed-size state. */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           sizeof(FixedParallelExecutorState));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Estimate space for query text. */
    query_len = strlen(estate->es_sourceText);
    shm_toc_estimate_chunk(&pcxt->estimator, query_len + 1);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Estimate space for serialized PlannedStmt. */
    pstmt_len = strlen(pstmt_data) + 1;
    shm_toc_estimate_chunk(&pcxt->estimator, pstmt_len);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Estimate space for serialized ParamListInfo. */
    paramlistinfo_len = EstimateParamListSpace(estate->es_param_list_info);
    shm_toc_estimate_chunk(&pcxt->estimator, paramlistinfo_len);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /*
     * Estimate space for BufferUsage.
     *
     * If EXPLAIN is not in use and there are no extensions loaded that care,
     * we could skip this.  But we have no way of knowing whether anyone's
     * looking at pgBufferUsage, so do it unconditionally.
     */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /*
     * Same thing for WalUsage.
     */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Estimate space for tuple queues. */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(PARALLEL_TUPLE_QUEUE_SIZE, pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /*
     * Give parallel-aware nodes a chance to add to the estimates, and get a
     * count of how many PlanState nodes there are.
     */
    e.pcxt = pcxt;
    e.nnodes = 0;
    ExecParallelEstimate(planstate, &e);
 
    /* Estimate space for instrumentation, if required. */
    if (estate->es_instrument)
    {
        instrumentation_len =
            offsetof(SharedExecutorInstrumentation, plan_node_id) +
            sizeof(int) * e.nnodes;
        instrumentation_len = MAXALIGN(instrumentation_len);
        instrument_offset = instrumentation_len;
        instrumentation_len +=
            mul_size(sizeof(Instrumentation),
                     mul_size(e.nnodes, nworkers));
        shm_toc_estimate_chunk(&pcxt->estimator, instrumentation_len);
        shm_toc_estimate_keys(&pcxt->estimator, 1);
 
        /* Estimate space for JIT instrumentation, if required. */
        if (estate->es_jit_flags != PGJIT_NONE)
        {
            jit_instrumentation_len =
                offsetof(SharedJitInstrumentation, jit_instr) +
                sizeof(JitInstrumentation) * nworkers;
            shm_toc_estimate_chunk(&pcxt->estimator, jit_instrumentation_len);
            shm_toc_estimate_keys(&pcxt->estimator, 1);
        }
    }
 
    /* Estimate space for DSA area. */
    shm_toc_estimate_chunk(&pcxt->estimator, dsa_minsize);
    shm_toc_estimate_keys(&pcxt->estimator, 1);
 
    /* Everyone's had a chance to ask for space, so now create the DSM. */
    InitializeParallelDSM(pcxt);
 
    /*
     * OK, now we have a dynamic shared memory segment, and it should be big
     * enough to store all of the data we estimated we would want to put into
     * it, plus whatever general stuff (not specifically executor-related) the
     * ParallelContext itself needs to store there.  None of the space we
     * asked for has been allocated or initialized yet, though, so do that.
     */
 
    /* Store fixed-size state. */
    fpes = shm_toc_allocate(pcxt->toc, sizeof(FixedParallelExecutorState));
    fpes->tuples_needed = tuples_needed;
    fpes->param_exec = InvalidDsaPointer;
    fpes->eflags = estate->es_top_eflags;
    fpes->jit_flags = estate->es_jit_flags;
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fpes);
 
    /* Store query string */
    query_string = shm_toc_allocate(pcxt->toc, query_len + 1);
    memcpy(query_string, estate->es_sourceText, query_len + 1);
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, query_string);
 
    /* Store serialized PlannedStmt. */
    pstmt_space = shm_toc_allocate(pcxt->toc, pstmt_len);
    memcpy(pstmt_space, pstmt_data, pstmt_len);
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_PLANNEDSTMT, pstmt_space);
 
    /* Store serialized ParamListInfo. */
    paramlistinfo_space = shm_toc_allocate(pcxt->toc, paramlistinfo_len);
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_PARAMLISTINFO, paramlistinfo_space);
    SerializeParamList(estate->es_param_list_info, &paramlistinfo_space);
 
    /* Allocate space for each worker's BufferUsage; no need to initialize. */
    bufusage_space = shm_toc_allocate(pcxt->toc,
                                      mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, bufusage_space);
    pei->buffer_usage = bufusage_space;
 
    /* Same for WalUsage. */
    walusage_space = shm_toc_allocate(pcxt->toc,
                                      mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_KEY_WAL_USAGE, walusage_space);
    pei->wal_usage = walusage_space;
 
    /* Set up the tuple queues that the workers will write into. */
    pei->tqueue = ExecParallelSetupTupleQueues(pcxt, false);
 
    /* We don't need the TupleQueueReaders yet, though. */
    pei->reader = NULL;
 
    /*
     * If instrumentation options were supplied, allocate space for the data.
     * It only gets partially initialized here; the rest happens during
     * ExecParallelInitializeDSM.
     */
    if (estate->es_instrument)
    {
        Instrumentation *instrument;
        int         i;
 
        instrumentation = shm_toc_allocate(pcxt->toc, instrumentation_len);
        instrumentation->instrument_options = estate->es_instrument;
        instrumentation->instrument_offset = instrument_offset;
        instrumentation->num_workers = nworkers;
        instrumentation->num_plan_nodes = e.nnodes;
        instrument = GetInstrumentationArray(instrumentation);
        for (i = 0; i < nworkers * e.nnodes; ++i)
            InstrInit(&instrument[i], estate->es_instrument);
        shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION,
                       instrumentation);
        pei->instrumentation = instrumentation;
 
        if (estate->es_jit_flags != PGJIT_NONE)
        {
            jit_instrumentation = shm_toc_allocate(pcxt->toc,
                                                   jit_instrumentation_len);
            jit_instrumentation->num_workers = nworkers;
            memset(jit_instrumentation->jit_instr, 0,
                   sizeof(JitInstrumentation) * nworkers);
            shm_toc_insert(pcxt->toc, PARALLEL_KEY_JIT_INSTRUMENTATION,
                           jit_instrumentation);
            pei->jit_instrumentation = jit_instrumentation;
        }
    }
 
    /*
     * Create a DSA area that can be used by the leader and all workers.
     * (However, if we failed to create a DSM and are using private memory
     * instead, then skip this.)
     */
    if (pcxt->seg != NULL)
    {
        char       *area_space;
 
        area_space = shm_toc_allocate(pcxt->toc, dsa_minsize);
        shm_toc_insert(pcxt->toc, PARALLEL_KEY_DSA, area_space);
        pei->area = dsa_create_in_place(area_space, dsa_minsize,
                                        LWTRANCHE_PARALLEL_QUERY_DSA,
                                        pcxt->seg);
 
        /*
         * Serialize parameters, if any, using DSA storage.  We don't dare use
         * the main parallel query DSM for this because we might relaunch
         * workers after the values have changed (and thus the amount of
         * storage required has changed).
         */
        if (!bms_is_empty(sendParams))
        {
            pei->param_exec = SerializeParamExecParams(estate, sendParams,
                                                       pei->area);
            fpes->param_exec = pei->param_exec;
        }
    }
 
    /*
     * Give parallel-aware nodes a chance to initialize their shared data.
     * This also initializes the elements of instrumentation->ps_instrument,
     * if it exists.
     */
    d.pcxt = pcxt;
    d.instrumentation = instrumentation;
    d.nnodes = 0;
 
    /* Install our DSA area while initializing the plan. */
    estate->es_query_dsa = pei->area;
    ExecParallelInitializeDSM(planstate, &d);
    estate->es_query_dsa = NULL;
 
    /*
     * Make sure that the world hasn't shifted under our feet.  This could
     * probably just be an Assert(), but let's be conservative for now.
     */
    if (e.nnodes != d.nnodes)
        elog(ERROR, "inconsistent count of PlanState nodes");
 
    /* OK, we're ready to rock and roll. */
    return pei;
}

References ParallelExecutorInfo::area, bms_is_empty, ParallelExecutorInfo::buffer_usage, CreateParallelContext(), dsa_create_in_place(), dsa_minimum_size(), FixedParallelExecutorState::eflags, elog(), ERROR, EState::es_instrument, EState::es_jit_flags, EState::es_param_list_info, EState::es_query_dsa, EState::es_sourceText, EState::es_top_eflags, EstimateParamListSpace(), ParallelContext::estimator, ExecParallelEstimate(), ExecParallelInitializeDSM(), ExecParallelSetupTupleQueues(), ExecSerializePlan(), ExecSetParamPlanMulti(), ParallelExecutorInfo::finished, GetInstrumentationArray, GetPerTupleExprContext, i, InitializeParallelDSM(), InstrInit(), SharedExecutorInstrumentation::instrument_offset, SharedExecutorInstrumentation::instrument_options, ExecParallelInitializeDSMContext::instrumentation, ParallelExecutorInfo::instrumentation, InvalidDsaPointer, FixedParallelExecutorState::jit_flags, SharedJitInstrumentation::jit_instr, ParallelExecutorInfo::jit_instrumentation, LWTRANCHE_PARALLEL_QUERY_DSA, MAXALIGN, mul_size(), ExecParallelInitializeDSMContext::nnodes, SharedExecutorInstrumentation::num_plan_nodes, SharedExecutorInstrumentation::num_workers, SharedJitInstrumentation::num_workers, ParallelContext::nworkers, palloc0(), PARALLEL_KEY_BUFFER_USAGE, PARALLEL_KEY_DSA, PARALLEL_KEY_EXECUTOR_FIXED, PARALLEL_KEY_INSTRUMENTATION, PARALLEL_KEY_JIT_INSTRUMENTATION, PARALLEL_KEY_PARAMLISTINFO, PARALLEL_KEY_PLANNEDSTMT, PARALLEL_KEY_QUERY_TEXT, PARALLEL_KEY_WAL_USAGE, PARALLEL_TUPLE_QUEUE_SIZE, FixedParallelExecutorState::param_exec, ParallelExecutorInfo::param_exec, ExecParallelInitializeDSMContext::pcxt, ParallelExecutorInfo::pcxt, PGJIT_NONE, PlanState::plan, ParallelExecutorInfo::planstate, ParallelExecutorInfo::reader, ParallelContext::seg, SerializeParamExecParams(), SerializeParamList(), shm_toc_allocate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_insert(), ParallelContext::toc, ParallelExecutorInfo::tqueue, FixedParallelExecutorState::tuples_needed, and ParallelExecutorInfo::wal_usage.

Referenced by ExecGather(), and ExecGatherMerge().

ExecParallelCleanup()

void ExecParallelCleanup

(

ParallelExecutorInfo *

pei

)

Definition at line 1180 of file execParallel.c.

{
    /* Accumulate instrumentation, if any. */
    if (pei->instrumentation)
        ExecParallelRetrieveInstrumentation(pei->planstate,
                                            pei->instrumentation);
 
    /* Accumulate JIT instrumentation, if any. */
    if (pei->jit_instrumentation)
        ExecParallelRetrieveJitInstrumentation(pei->planstate,
                                               pei->jit_instrumentation);
 
    /* Free any serialized parameters. */
    if (DsaPointerIsValid(pei->param_exec))
    {
        dsa_free(pei->area, pei->param_exec);
        pei->param_exec = InvalidDsaPointer;
    }
    if (pei->area != NULL)
    {
        dsa_detach(pei->area);
        pei->area = NULL;
    }
    if (pei->pcxt != NULL)
    {
        DestroyParallelContext(pei->pcxt);
        pei->pcxt = NULL;
    }
    pfree(pei);
}

References ParallelExecutorInfo::area, DestroyParallelContext(), dsa_detach(), dsa_free(), DsaPointerIsValid, ExecParallelRetrieveInstrumentation(), ExecParallelRetrieveJitInstrumentation(), ParallelExecutorInfo::instrumentation, InvalidDsaPointer, ParallelExecutorInfo::jit_instrumentation, ParallelExecutorInfo::param_exec, ParallelExecutorInfo::pcxt, pfree(), and ParallelExecutorInfo::planstate.

Referenced by ExecShutdownGather(), and ExecShutdownGatherMerge().

ExecParallelCreateReaders()

void ExecParallelCreateReaders

(

ParallelExecutorInfo *

pei

)

Definition at line 874 of file execParallel.c.

{
    int         nworkers = pei->pcxt->nworkers_launched;
    int         i;
 
    Assert(pei->reader == NULL);
 
    if (nworkers > 0)
    {
        pei->reader = (TupleQueueReader **)
            palloc(nworkers * sizeof(TupleQueueReader *));
 
        for (i = 0; i < nworkers; i++)
        {
            shm_mq_set_handle(pei->tqueue[i],
                              pei->pcxt->worker[i].bgwhandle);
            pei->reader[i] = CreateTupleQueueReader(pei->tqueue[i]);
        }
    }
}

References Assert(), ParallelWorkerInfo::bgwhandle, CreateTupleQueueReader(), i, ParallelContext::nworkers_launched, palloc(), ParallelExecutorInfo::pcxt, ParallelExecutorInfo::reader, shm_mq_set_handle(), ParallelExecutorInfo::tqueue, and ParallelContext::worker.

Referenced by ExecGather(), and ExecGatherMerge().

ExecParallelFinish()

void ExecParallelFinish

(

ParallelExecutorInfo *

pei

)

Definition at line 1127 of file execParallel.c.

{
    int         nworkers = pei->pcxt->nworkers_launched;
    int         i;
 
    /* Make this be a no-op if called twice in a row. */
    if (pei->finished)
        return;
 
    /*
     * Detach from tuple queues ASAP, so that any still-active workers will
     * notice that no further results are wanted.
     */
    if (pei->tqueue != NULL)
    {
        for (i = 0; i < nworkers; i++)
            shm_mq_detach(pei->tqueue[i]);
        pfree(pei->tqueue);
        pei->tqueue = NULL;
    }
 
    /*
     * While we're waiting for the workers to finish, let's get rid of the
     * tuple queue readers.  (Any other local cleanup could be done here too.)
     */
    if (pei->reader != NULL)
    {
        for (i = 0; i < nworkers; i++)
            DestroyTupleQueueReader(pei->reader[i]);
        pfree(pei->reader);
        pei->reader = NULL;
    }
 
    /* Now wait for the workers to finish. */
    WaitForParallelWorkersToFinish(pei->pcxt);
 
    /*
     * Next, accumulate buffer/WAL usage.  (This must wait for the workers to
     * finish, or we might get incomplete data.)
     */
    for (i = 0; i < nworkers; i++)
        InstrAccumParallelQuery(&pei->buffer_usage[i], &pei->wal_usage[i]);
 
    pei->finished = true;
}

References ParallelExecutorInfo::buffer_usage, DestroyTupleQueueReader(), ParallelExecutorInfo::finished, i, InstrAccumParallelQuery(), ParallelContext::nworkers_launched, ParallelExecutorInfo::pcxt, pfree(), ParallelExecutorInfo::reader, shm_mq_detach(), ParallelExecutorInfo::tqueue, WaitForParallelWorkersToFinish(), and ParallelExecutorInfo::wal_usage.

Referenced by ExecShutdownGatherMergeWorkers(), and ExecShutdownGatherWorkers().

ExecParallelReinitialize()

void ExecParallelReinitialize	(	PlanState *	planstate,
		ParallelExecutorInfo *	pei,
		Bitmapset *	sendParams
	)

Definition at line 900 of file execParallel.c.

{
    EState     *estate = planstate->state;
    FixedParallelExecutorState *fpes;
 
    /* Old workers must already be shut down */
    Assert(pei->finished);
 
    /*
     * Force any initplan outputs that we're going to pass to workers to be
     * evaluated, if they weren't already (see comments in
     * ExecInitParallelPlan).
     */
    ExecSetParamPlanMulti(sendParams, GetPerTupleExprContext(estate));
 
    ReinitializeParallelDSM(pei->pcxt);
    pei->tqueue = ExecParallelSetupTupleQueues(pei->pcxt, true);
    pei->reader = NULL;
    pei->finished = false;
 
    fpes = shm_toc_lookup(pei->pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
 
    /* Free any serialized parameters from the last round. */
    if (DsaPointerIsValid(fpes->param_exec))
    {
        dsa_free(pei->area, fpes->param_exec);
        fpes->param_exec = InvalidDsaPointer;
    }
 
    /* Serialize current parameter values if required. */
    if (!bms_is_empty(sendParams))
    {
        pei->param_exec = SerializeParamExecParams(estate, sendParams,
                                                   pei->area);
        fpes->param_exec = pei->param_exec;
    }
 
    /* Traverse plan tree and let each child node reset associated state. */
    estate->es_query_dsa = pei->area;
    ExecParallelReInitializeDSM(planstate, pei->pcxt);
    estate->es_query_dsa = NULL;
}

References ParallelExecutorInfo::area, Assert(), bms_is_empty, dsa_free(), DsaPointerIsValid, EState::es_query_dsa, ExecParallelReInitializeDSM(), ExecParallelSetupTupleQueues(), ExecSetParamPlanMulti(), ParallelExecutorInfo::finished, GetPerTupleExprContext, InvalidDsaPointer, PARALLEL_KEY_EXECUTOR_FIXED, FixedParallelExecutorState::param_exec, ParallelExecutorInfo::param_exec, ParallelExecutorInfo::pcxt, ParallelExecutorInfo::reader, ReinitializeParallelDSM(), SerializeParamExecParams(), shm_toc_lookup(), PlanState::state, ParallelContext::toc, and ParallelExecutorInfo::tqueue.

Referenced by ExecGather(), and ExecGatherMerge().

ParallelQueryMain()

void ParallelQueryMain	(	dsm_segment *	seg,
		shm_toc *	toc
	)

Definition at line 1396 of file execParallel.c.

{
    FixedParallelExecutorState *fpes;
    BufferUsage *buffer_usage;
    WalUsage   *wal_usage;
    DestReceiver *receiver;
    QueryDesc  *queryDesc;
    SharedExecutorInstrumentation *instrumentation;
    SharedJitInstrumentation *jit_instrumentation;
    int         instrument_options = 0;
    void       *area_space;
    dsa_area   *area;
    ParallelWorkerContext pwcxt;
 
    /* Get fixed-size state. */
    fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
 
    /* Set up DestReceiver, SharedExecutorInstrumentation, and QueryDesc. */
    receiver = ExecParallelGetReceiver(seg, toc);
    instrumentation = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, true);
    if (instrumentation != NULL)
        instrument_options = instrumentation->instrument_options;
    jit_instrumentation = shm_toc_lookup(toc, PARALLEL_KEY_JIT_INSTRUMENTATION,
                                         true);
    queryDesc = ExecParallelGetQueryDesc(toc, receiver, instrument_options);
 
    /* Setting debug_query_string for individual workers */
    debug_query_string = queryDesc->sourceText;
 
    /* Report workers' query for monitoring purposes */
    pgstat_report_activity(STATE_RUNNING, debug_query_string);
 
    /* Attach to the dynamic shared memory area. */
    area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
    area = dsa_attach_in_place(area_space, seg);
 
    /* Start up the executor */
    queryDesc->plannedstmt->jitFlags = fpes->jit_flags;
    ExecutorStart(queryDesc, fpes->eflags);
 
    /* Special executor initialization steps for parallel workers */
    queryDesc->planstate->state->es_query_dsa = area;
    if (DsaPointerIsValid(fpes->param_exec))
    {
        char       *paramexec_space;
 
        paramexec_space = dsa_get_address(area, fpes->param_exec);
        RestoreParamExecParams(paramexec_space, queryDesc->estate);
    }
    pwcxt.toc = toc;
    pwcxt.seg = seg;
    ExecParallelInitializeWorker(queryDesc->planstate, &pwcxt);
 
    /* Pass down any tuple bound */
    ExecSetTupleBound(fpes->tuples_needed, queryDesc->planstate);
 
    /*
     * Prepare to track buffer/WAL usage during query execution.
     *
     * We do this after starting up the executor to match what happens in the
     * leader, which also doesn't count buffer accesses and WAL activity that
     * occur during executor startup.
     */
    InstrStartParallelQuery();
 
    /*
     * Run the plan.  If we specified a tuple bound, be careful not to demand
     * more tuples than that.
     */
    ExecutorRun(queryDesc,
                ForwardScanDirection,
                fpes->tuples_needed < 0 ? (int64) 0 : fpes->tuples_needed,
                true);
 
    /* Shut down the executor */
    ExecutorFinish(queryDesc);
 
    /* Report buffer/WAL usage during parallel execution. */
    buffer_usage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
    wal_usage = shm_toc_lookup(toc, PARALLEL_KEY_WAL_USAGE, false);
    InstrEndParallelQuery(&buffer_usage[ParallelWorkerNumber],
                          &wal_usage[ParallelWorkerNumber]);
 
    /* Report instrumentation data if any instrumentation options are set. */
    if (instrumentation != NULL)
        ExecParallelReportInstrumentation(queryDesc->planstate,
                                          instrumentation);
 
    /* Report JIT instrumentation data if any */
    if (queryDesc->estate->es_jit && jit_instrumentation != NULL)
    {
        Assert(ParallelWorkerNumber < jit_instrumentation->num_workers);
        jit_instrumentation->jit_instr[ParallelWorkerNumber] =
            queryDesc->estate->es_jit->instr;
    }
 
    /* Must do this after capturing instrumentation. */
    ExecutorEnd(queryDesc);
 
    /* Cleanup. */
    dsa_detach(area);
    FreeQueryDesc(queryDesc);
    receiver->rDestroy(receiver);
}

References Assert(), debug_query_string, dsa_attach_in_place(), dsa_detach(), dsa_get_address(), DsaPointerIsValid, FixedParallelExecutorState::eflags, EState::es_jit, EState::es_query_dsa, QueryDesc::estate, ExecParallelGetQueryDesc(), ExecParallelGetReceiver(), ExecParallelInitializeWorker(), ExecParallelReportInstrumentation(), ExecSetTupleBound(), ExecutorEnd(), ExecutorFinish(), ExecutorRun(), ExecutorStart(), ForwardScanDirection, FreeQueryDesc(), JitContext::instr, InstrEndParallelQuery(), InstrStartParallelQuery(), SharedExecutorInstrumentation::instrument_options, FixedParallelExecutorState::jit_flags, SharedJitInstrumentation::jit_instr, PlannedStmt::jitFlags, PARALLEL_KEY_BUFFER_USAGE, PARALLEL_KEY_DSA, PARALLEL_KEY_EXECUTOR_FIXED, PARALLEL_KEY_INSTRUMENTATION, PARALLEL_KEY_JIT_INSTRUMENTATION, PARALLEL_KEY_WAL_USAGE, ParallelWorkerNumber, FixedParallelExecutorState::param_exec, pgstat_report_activity(), QueryDesc::plannedstmt, QueryDesc::planstate, _DestReceiver::rDestroy, RestoreParamExecParams(), ParallelWorkerContext::seg, shm_toc_lookup(), QueryDesc::sourceText, PlanState::state, STATE_RUNNING, ParallelWorkerContext::toc, and FixedParallelExecutorState::tuples_needed.