nodeGather.c File Reference

#include "postgres.h"
#include "executor/execParallel.h"
#include "executor/executor.h"
#include "executor/nodeGather.h"
#include "executor/tqueue.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "utils/wait_event.h"

Include dependency graph for nodeGather.c:

Go to the source code of this file.

Functions
static TupleTableSlot *	ExecGather (PlanState *pstate)
static TupleTableSlot *	gather_getnext (GatherState *gatherstate)
static MinimalTuple	gather_readnext (GatherState *gatherstate)
static void	ExecShutdownGatherWorkers (GatherState *node)
GatherState *	ExecInitGather (Gather *node, EState *estate, int eflags)
void	ExecEndGather (GatherState *node)
void	ExecShutdownGather (GatherState *node)
void	ExecReScanGather (GatherState *node)

Function Documentation

ExecEndGather()

void ExecEndGather

(

GatherState *

node

)

Definition at line 251 of file nodeGather.c.

{
    ExecEndNode(outerPlanState(node));  /* let children clean up first */
    ExecShutdownGather(node);
}

References ExecEndNode(), ExecShutdownGather(), and outerPlanState.

Referenced by ExecEndNode().

ExecGather()

static TupleTableSlot * ExecGather ( PlanState *  pstate )

static

Definition at line 137 of file nodeGather.c.

{
    GatherState *node = castNode(GatherState, pstate);
    TupleTableSlot *slot;
    ExprContext *econtext;
 
    CHECK_FOR_INTERRUPTS();
 
    /*
     * Initialize the parallel context and workers on first execution. We do
     * this on first execution rather than during node initialization, as it
     * needs to allocate a large dynamic segment, so it is better to do it
     * only if it is really needed.
     */
    if (!node->initialized)
    {
        EState     *estate = node->ps.state;
        Gather     *gather = (Gather *) node->ps.plan;
 
        /*
         * Sometimes we might have to run without parallelism; but if parallel
         * mode is active then we can try to fire up some workers.
         */
        if (gather->num_workers > 0 && estate->es_use_parallel_mode)
        {
            ParallelContext *pcxt;
 
            /* Initialize, or re-initialize, shared state needed by workers. */
            if (!node->pei)
                node->pei = ExecInitParallelPlan(outerPlanState(node),
                                                 estate,
                                                 gather->initParam,
                                                 gather->num_workers,
                                                 node->tuples_needed);
            else
                ExecParallelReinitialize(outerPlanState(node),
                                         node->pei,
                                         gather->initParam);
 
            /*
             * Register backend workers. We might not get as many as we
             * requested, or indeed any at all.
             */
            pcxt = node->pei->pcxt;
            LaunchParallelWorkers(pcxt);
            /* We save # workers launched for the benefit of EXPLAIN */
            node->nworkers_launched = pcxt->nworkers_launched;
 
            /*
             * Count number of workers originally wanted and actually
             * launched.
             */
            estate->es_parallel_workers_to_launch += pcxt->nworkers_to_launch;
            estate->es_parallel_workers_launched += pcxt->nworkers_launched;
 
            /* Set up tuple queue readers to read the results. */
            if (pcxt->nworkers_launched > 0)
            {
                ExecParallelCreateReaders(node->pei);
                /* Make a working array showing the active readers */
                node->nreaders = pcxt->nworkers_launched;
                node->reader = (TupleQueueReader **)
                    palloc(node->nreaders * sizeof(TupleQueueReader *));
                memcpy(node->reader, node->pei->reader,
                       node->nreaders * sizeof(TupleQueueReader *));
            }
            else
            {
                /* No workers?  Then never mind. */
                node->nreaders = 0;
                node->reader = NULL;
            }
            node->nextreader = 0;
        }
 
        /* Run plan locally if no workers or enabled and not single-copy. */
        node->need_to_scan_locally = (node->nreaders == 0)
            || (!gather->single_copy && parallel_leader_participation);
        node->initialized = true;
    }
 
    /*
     * Reset per-tuple memory context to free any expression evaluation
     * storage allocated in the previous tuple cycle.
     */
    econtext = node->ps.ps_ExprContext;
    ResetExprContext(econtext);
 
    /*
     * Get next tuple, either from one of our workers, or by running the plan
     * ourselves.
     */
    slot = gather_getnext(node);
    if (TupIsNull(slot))
        return NULL;
 
    /* If no projection is required, we're done. */
    if (node->ps.ps_ProjInfo == NULL)
        return slot;
 
    /*
     * Form the result tuple using ExecProject(), and return it.
     */
    econtext->ecxt_outertuple = slot;
    return ExecProject(node->ps.ps_ProjInfo);
}

References castNode, CHECK_FOR_INTERRUPTS, ExprContext::ecxt_outertuple, EState::es_parallel_workers_launched, EState::es_parallel_workers_to_launch, EState::es_use_parallel_mode, ExecInitParallelPlan(), ExecParallelCreateReaders(), ExecParallelReinitialize(), ExecProject(), gather_getnext(), if(), GatherState::initialized, Gather::initParam, LaunchParallelWorkers(), GatherState::need_to_scan_locally, GatherState::nextreader, GatherState::nreaders, Gather::num_workers, ParallelContext::nworkers_launched, GatherState::nworkers_launched, ParallelContext::nworkers_to_launch, outerPlanState, palloc(), parallel_leader_participation, ParallelExecutorInfo::pcxt, GatherState::pei, PlanState::plan, GatherState::ps, PlanState::ps_ExprContext, PlanState::ps_ProjInfo, ParallelExecutorInfo::reader, GatherState::reader, ResetExprContext, Gather::single_copy, PlanState::state, TupIsNull, and GatherState::tuples_needed.

Referenced by ExecInitGather().

ExecInitGather()

GatherState * ExecInitGather	(	Gather *	node,
		EState *	estate,
		int	eflags
	)

Definition at line 53 of file nodeGather.c.

{
    GatherState *gatherstate;
    Plan       *outerNode;
    TupleDesc   tupDesc;
 
    /* Gather node doesn't have innerPlan node. */
    Assert(innerPlan(node) == NULL);
 
    /*
     * create state structure
     */
    gatherstate = makeNode(GatherState);
    gatherstate->ps.plan = (Plan *) node;
    gatherstate->ps.state = estate;
    gatherstate->ps.ExecProcNode = ExecGather;
 
    gatherstate->initialized = false;
    gatherstate->need_to_scan_locally =
        !node->single_copy && parallel_leader_participation;
    gatherstate->tuples_needed = -1;
 
    /*
     * Miscellaneous initialization
     *
     * create expression context for node
     */
    ExecAssignExprContext(estate, &gatherstate->ps);
 
    /*
     * now initialize outer plan
     */
    outerNode = outerPlan(node);
    outerPlanState(gatherstate) = ExecInitNode(outerNode, estate, eflags);
    tupDesc = ExecGetResultType(outerPlanState(gatherstate));
 
    /*
     * Leader may access ExecProcNode result directly (if
     * need_to_scan_locally), or from workers via tuple queue.  So we can't
     * trivially rely on the slot type being fixed for expressions evaluated
     * within this node.
     */
    gatherstate->ps.outeropsset = true;
    gatherstate->ps.outeropsfixed = false;
 
    /*
     * Initialize result type and projection.
     */
    ExecInitResultTypeTL(&gatherstate->ps);
    ExecConditionalAssignProjectionInfo(&gatherstate->ps, tupDesc, OUTER_VAR);
 
    /*
     * Without projections result slot type is not trivially known, see
     * comment above.
     */
    if (gatherstate->ps.ps_ProjInfo == NULL)
    {
        gatherstate->ps.resultopsset = true;
        gatherstate->ps.resultopsfixed = false;
    }
 
    /*
     * Initialize funnel slot to same tuple descriptor as outer plan.
     */
    gatherstate->funnel_slot = ExecInitExtraTupleSlot(estate, tupDesc,
                                                      &TTSOpsMinimalTuple);
 
    /*
     * Gather doesn't support checking a qual (it's always more efficient to
     * do it in the child node).
     */
    Assert(!node->plan.qual);
 
    return gatherstate;
}

References Assert(), ExecAssignExprContext(), ExecConditionalAssignProjectionInfo(), ExecGather(), ExecGetResultType(), ExecInitExtraTupleSlot(), ExecInitNode(), ExecInitResultTypeTL(), PlanState::ExecProcNode, GatherState::funnel_slot, GatherState::initialized, innerPlan, makeNode, GatherState::need_to_scan_locally, OUTER_VAR, PlanState::outeropsfixed, PlanState::outeropsset, outerPlan, outerPlanState, parallel_leader_participation, PlanState::plan, Gather::plan, GatherState::ps, PlanState::ps_ProjInfo, Plan::qual, PlanState::resultopsfixed, PlanState::resultopsset, Gather::single_copy, PlanState::state, TTSOpsMinimalTuple, and GatherState::tuples_needed.

Referenced by ExecInitNode().

ExecReScanGather()

void ExecReScanGather

(

GatherState *

node

)

Definition at line 442 of file nodeGather.c.

{
    Gather     *gather = (Gather *) node->ps.plan;
    PlanState  *outerPlan = outerPlanState(node);
 
    /* Make sure any existing workers are gracefully shut down */
    ExecShutdownGatherWorkers(node);
 
    /* Mark node so that shared state will be rebuilt at next call */
    node->initialized = false;
 
    /*
     * Set child node's chgParam to tell it that the next scan might deliver a
     * different set of rows within the leader process.  (The overall rowset
     * shouldn't change, but the leader process's subset might; hence nodes
     * between here and the parallel table scan node mustn't optimize on the
     * assumption of an unchanging rowset.)
     */
    if (gather->rescan_param >= 0)
        outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
                                             gather->rescan_param);
 
    /*
     * If chgParam of subnode is not null then plan will be re-scanned by
     * first ExecProcNode.  Note: because this does nothing if we have a
     * rescan_param, it's currently guaranteed that parallel-aware child nodes
     * will not see a ReScan call until after they get a ReInitializeDSM call.
     * That ordering might not be something to rely on, though.  A good rule
     * of thumb is that ReInitializeDSM should reset only shared state, ReScan
     * should reset only local state, and anything that depends on both of
     * those steps being finished must wait until the first ExecProcNode call.
     */
    if (outerPlan->chgParam == NULL)
        ExecReScan(outerPlan);
}

References bms_add_member(), ExecReScan(), ExecShutdownGatherWorkers(), GatherState::initialized, outerPlan, outerPlanState, PlanState::plan, GatherState::ps, and Gather::rescan_param.

Referenced by ExecReScan().

ExecShutdownGather()

void ExecShutdownGather

(

GatherState *

node

)

Definition at line 418 of file nodeGather.c.

{
    ExecShutdownGatherWorkers(node);
 
    /* Now destroy the parallel context. */
    if (node->pei != NULL)
    {
        ExecParallelCleanup(node->pei);
        node->pei = NULL;
    }
}

References ExecParallelCleanup(), ExecShutdownGatherWorkers(), and GatherState::pei.

Referenced by ExecEndGather(), and ExecShutdownNode_walker().

ExecShutdownGatherWorkers()

static void ExecShutdownGatherWorkers ( GatherState *  node )

static

Definition at line 400 of file nodeGather.c.

{
    if (node->pei != NULL)
        ExecParallelFinish(node->pei);
 
    /* Flush local copy of reader array */
    if (node->reader)
        pfree(node->reader);
    node->reader = NULL;
}

References ExecParallelFinish(), GatherState::pei, pfree(), and GatherState::reader.

Referenced by ExecReScanGather(), ExecShutdownGather(), and gather_readnext().

gather_getnext()

static TupleTableSlot * gather_getnext ( GatherState *  gatherstate )

static

Definition at line 263 of file nodeGather.c.

{
    PlanState  *outerPlan = outerPlanState(gatherstate);
    TupleTableSlot *outerTupleSlot;
    TupleTableSlot *fslot = gatherstate->funnel_slot;
    MinimalTuple tup;
 
    while (gatherstate->nreaders > 0 || gatherstate->need_to_scan_locally)
    {
        CHECK_FOR_INTERRUPTS();
 
        if (gatherstate->nreaders > 0)
        {
            tup = gather_readnext(gatherstate);
 
            if (HeapTupleIsValid(tup))
            {
                ExecStoreMinimalTuple(tup,  /* tuple to store */
                                      fslot,    /* slot to store the tuple */
                                      false);   /* don't pfree tuple  */
                return fslot;
            }
        }
 
        if (gatherstate->need_to_scan_locally)
        {
            EState     *estate = gatherstate->ps.state;
 
            /* Install our DSA area while executing the plan. */
            estate->es_query_dsa =
                gatherstate->pei ? gatherstate->pei->area : NULL;
            outerTupleSlot = ExecProcNode(outerPlan);
            estate->es_query_dsa = NULL;
 
            if (!TupIsNull(outerTupleSlot))
                return outerTupleSlot;
 
            gatherstate->need_to_scan_locally = false;
        }
    }
 
    return ExecClearTuple(fslot);
}

References ParallelExecutorInfo::area, CHECK_FOR_INTERRUPTS, EState::es_query_dsa, ExecClearTuple(), ExecProcNode(), ExecStoreMinimalTuple(), GatherState::funnel_slot, gather_readnext(), HeapTupleIsValid, GatherState::need_to_scan_locally, GatherState::nreaders, outerPlan, outerPlanState, GatherState::pei, GatherState::ps, PlanState::state, and TupIsNull.

Referenced by ExecGather().

gather_readnext()

static MinimalTuple gather_readnext ( GatherState *  gatherstate )

static

Definition at line 311 of file nodeGather.c.

{
    int         nvisited = 0;
 
    for (;;)
    {
        TupleQueueReader *reader;
        MinimalTuple tup;
        bool        readerdone;
 
        /* Check for async events, particularly messages from workers. */
        CHECK_FOR_INTERRUPTS();
 
        /*
         * Attempt to read a tuple, but don't block if none is available.
         *
         * Note that TupleQueueReaderNext will just return NULL for a worker
         * which fails to initialize.  We'll treat that worker as having
         * produced no tuples; WaitForParallelWorkersToFinish will error out
         * when we get there.
         */
        Assert(gatherstate->nextreader < gatherstate->nreaders);
        reader = gatherstate->reader[gatherstate->nextreader];
        tup = TupleQueueReaderNext(reader, true, &readerdone);
 
        /*
         * If this reader is done, remove it from our working array of active
         * readers.  If all readers are done, we're outta here.
         */
        if (readerdone)
        {
            Assert(!tup);
            --gatherstate->nreaders;
            if (gatherstate->nreaders == 0)
            {
                ExecShutdownGatherWorkers(gatherstate);
                return NULL;
            }
            memmove(&gatherstate->reader[gatherstate->nextreader],
                    &gatherstate->reader[gatherstate->nextreader + 1],
                    sizeof(TupleQueueReader *)
                    * (gatherstate->nreaders - gatherstate->nextreader));
            if (gatherstate->nextreader >= gatherstate->nreaders)
                gatherstate->nextreader = 0;
            continue;
        }
 
        /* If we got a tuple, return it. */
        if (tup)
            return tup;
 
        /*
         * Advance nextreader pointer in round-robin fashion.  Note that we
         * only reach this code if we weren't able to get a tuple from the
         * current worker.  We used to advance the nextreader pointer after
         * every tuple, but it turns out to be much more efficient to keep
         * reading from the same queue until that would require blocking.
         */
        gatherstate->nextreader++;
        if (gatherstate->nextreader >= gatherstate->nreaders)
            gatherstate->nextreader = 0;
 
        /* Have we visited every (surviving) TupleQueueReader? */
        nvisited++;
        if (nvisited >= gatherstate->nreaders)
        {
            /*
             * If (still) running plan locally, return NULL so caller can
             * generate another tuple from the local copy of the plan.
             */
            if (gatherstate->need_to_scan_locally)
                return NULL;
 
            /* Nothing to do except wait for developments. */
            (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                             WAIT_EVENT_EXECUTE_GATHER);
            ResetLatch(MyLatch);
            nvisited = 0;
        }
    }
}

References Assert(), CHECK_FOR_INTERRUPTS, ExecShutdownGatherWorkers(), MyLatch, GatherState::need_to_scan_locally, GatherState::nextreader, GatherState::nreaders, GatherState::reader, ResetLatch(), TupleQueueReaderNext(), WaitLatch(), WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

Referenced by gather_getnext().