LCOV - code coverage report
Current view: top level - src/backend/executor - nodeGather.c (source / functions) Hit Total Coverage
Test: PostgreSQL 17devel Lines: 131 134 97.8 %
Date: 2023-12-02 16:11:01 Functions: 8 8 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * nodeGather.c
       4             :  *    Support routines for scanning a plan via multiple workers.
       5             :  *
       6             :  * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  * A Gather executor launches parallel workers to run multiple copies of a
      10             :  * plan.  It can also run the plan itself, if the workers are not available
      11             :  * or have not started up yet.  It then merges all of the results it produces
      12             :  * and the results from the workers into a single output stream.  Therefore,
      13             :  * it will normally be used with a plan where running multiple copies of the
      14             :  * same plan does not produce duplicate output, such as parallel-aware
      15             :  * SeqScan.
      16             :  *
      17             :  * Alternatively, a Gather node can be configured to use just one worker
      18             :  * and the single-copy flag can be set.  In this case, the Gather node will
      19             :  * run the plan in one worker and will not execute the plan itself.  In
      20             :  * this case, it simply returns whatever tuples were returned by the worker.
      21             :  * If a worker cannot be obtained, then it will run the plan itself and
      22             :  * return the results.  Therefore, a plan used with a single-copy Gather
      23             :  * node need not be parallel-aware.
      24             :  *
      25             :  * IDENTIFICATION
      26             :  *    src/backend/executor/nodeGather.c
      27             :  *
      28             :  *-------------------------------------------------------------------------
      29             :  */
      30             : 
      31             : #include "postgres.h"
      32             : 
      33             : #include "access/relscan.h"
      34             : #include "access/xact.h"
      35             : #include "executor/execdebug.h"
      36             : #include "executor/execParallel.h"
      37             : #include "executor/nodeGather.h"
      38             : #include "executor/nodeSubplan.h"
      39             : #include "executor/tqueue.h"
      40             : #include "miscadmin.h"
      41             : #include "optimizer/optimizer.h"
      42             : #include "pgstat.h"
      43             : #include "utils/memutils.h"
      44             : #include "utils/rel.h"
      45             : 
      46             : 
      47             : static TupleTableSlot *ExecGather(PlanState *pstate);
      48             : static TupleTableSlot *gather_getnext(GatherState *gatherstate);
      49             : static MinimalTuple gather_readnext(GatherState *gatherstate);
      50             : static void ExecShutdownGatherWorkers(GatherState *node);
      51             : 
      52             : 
      53             : /* ----------------------------------------------------------------
      54             :  *      ExecInitGather
      55             :  * ----------------------------------------------------------------
      56             :  */
      57             : GatherState *
      58         950 : ExecInitGather(Gather *node, EState *estate, int eflags)
      59             : {
      60             :     GatherState *gatherstate;
      61             :     Plan       *outerNode;
      62             :     TupleDesc   tupDesc;
      63             : 
      64             :     /* Gather node doesn't have innerPlan node. */
      65             :     Assert(innerPlan(node) == NULL);
      66             : 
      67             :     /*
      68             :      * create state structure
      69             :      */
      70         950 :     gatherstate = makeNode(GatherState);
      71         950 :     gatherstate->ps.plan = (Plan *) node;
      72         950 :     gatherstate->ps.state = estate;
      73         950 :     gatherstate->ps.ExecProcNode = ExecGather;
      74             : 
      75         950 :     gatherstate->initialized = false;
      76         950 :     gatherstate->need_to_scan_locally =
      77         950 :         !node->single_copy && parallel_leader_participation;
      78         950 :     gatherstate->tuples_needed = -1;
      79             : 
      80             :     /*
      81             :      * Miscellaneous initialization
      82             :      *
      83             :      * create expression context for node
      84             :      */
      85         950 :     ExecAssignExprContext(estate, &gatherstate->ps);
      86             : 
      87             :     /*
      88             :      * now initialize outer plan
      89             :      */
      90         950 :     outerNode = outerPlan(node);
      91         950 :     outerPlanState(gatherstate) = ExecInitNode(outerNode, estate, eflags);
      92         950 :     tupDesc = ExecGetResultType(outerPlanState(gatherstate));
      93             : 
      94             :     /*
      95             :      * Leader may access ExecProcNode result directly (if
      96             :      * need_to_scan_locally), or from workers via tuple queue.  So we can't
      97             :      * trivially rely on the slot type being fixed for expressions evaluated
      98             :      * within this node.
      99             :      */
     100         950 :     gatherstate->ps.outeropsset = true;
     101         950 :     gatherstate->ps.outeropsfixed = false;
     102             : 
     103             :     /*
     104             :      * Initialize result type and projection.
     105             :      */
     106         950 :     ExecInitResultTypeTL(&gatherstate->ps);
     107         950 :     ExecConditionalAssignProjectionInfo(&gatherstate->ps, tupDesc, OUTER_VAR);
     108             : 
     109             :     /*
     110             :      * Without projections result slot type is not trivially known, see
     111             :      * comment above.
     112             :      */
     113         950 :     if (gatherstate->ps.ps_ProjInfo == NULL)
     114             :     {
     115         908 :         gatherstate->ps.resultopsset = true;
     116         908 :         gatherstate->ps.resultopsfixed = false;
     117             :     }
     118             : 
     119             :     /*
     120             :      * Initialize funnel slot to same tuple descriptor as outer plan.
     121             :      */
     122         950 :     gatherstate->funnel_slot = ExecInitExtraTupleSlot(estate, tupDesc,
     123             :                                                       &TTSOpsMinimalTuple);
     124             : 
     125             :     /*
     126             :      * Gather doesn't support checking a qual (it's always more efficient to
     127             :      * do it in the child node).
     128             :      */
     129             :     Assert(!node->plan.qual);
     130             : 
     131         950 :     return gatherstate;
     132             : }
     133             : 
     134             : /* ----------------------------------------------------------------
     135             :  *      ExecGather(node)
     136             :  *
     137             :  *      Scans the relation via multiple workers and returns
     138             :  *      the next qualifying tuple.
     139             :  * ----------------------------------------------------------------
     140             :  */
     141             : static TupleTableSlot *
     142     2918300 : ExecGather(PlanState *pstate)
     143             : {
     144     2918300 :     GatherState *node = castNode(GatherState, pstate);
     145             :     TupleTableSlot *slot;
     146             :     ExprContext *econtext;
     147             : 
     148     2918300 :     CHECK_FOR_INTERRUPTS();
     149             : 
     150             :     /*
     151             :      * Initialize the parallel context and workers on first execution. We do
     152             :      * this on first execution rather than during node initialization, as it
     153             :      * needs to allocate a large dynamic segment, so it is better to do it
     154             :      * only if it is really needed.
     155             :      */
     156     2918300 :     if (!node->initialized)
     157             :     {
     158         758 :         EState     *estate = node->ps.state;
     159         758 :         Gather     *gather = (Gather *) node->ps.plan;
     160             : 
     161             :         /*
     162             :          * Sometimes we might have to run without parallelism; but if parallel
     163             :          * mode is active then we can try to fire up some workers.
     164             :          */
     165         758 :         if (gather->num_workers > 0 && estate->es_use_parallel_mode)
     166             :         {
     167             :             ParallelContext *pcxt;
     168             : 
     169             :             /* Initialize, or re-initialize, shared state needed by workers. */
     170         752 :             if (!node->pei)
     171         524 :                 node->pei = ExecInitParallelPlan(outerPlanState(node),
     172             :                                                  estate,
     173             :                                                  gather->initParam,
     174             :                                                  gather->num_workers,
     175             :                                                  node->tuples_needed);
     176             :             else
     177         228 :                 ExecParallelReinitialize(outerPlanState(node),
     178         228 :                                          node->pei,
     179             :                                          gather->initParam);
     180             : 
     181             :             /*
     182             :              * Register backend workers. We might not get as many as we
     183             :              * requested, or indeed any at all.
     184             :              */
     185         752 :             pcxt = node->pei->pcxt;
     186         752 :             LaunchParallelWorkers(pcxt);
     187             :             /* We save # workers launched for the benefit of EXPLAIN */
     188         752 :             node->nworkers_launched = pcxt->nworkers_launched;
     189             : 
     190             :             /* Set up tuple queue readers to read the results. */
     191         752 :             if (pcxt->nworkers_launched > 0)
     192             :             {
     193         746 :                 ExecParallelCreateReaders(node->pei);
     194             :                 /* Make a working array showing the active readers */
     195         746 :                 node->nreaders = pcxt->nworkers_launched;
     196         746 :                 node->reader = (TupleQueueReader **)
     197         746 :                     palloc(node->nreaders * sizeof(TupleQueueReader *));
     198         746 :                 memcpy(node->reader, node->pei->reader,
     199         746 :                        node->nreaders * sizeof(TupleQueueReader *));
     200             :             }
     201             :             else
     202             :             {
     203             :                 /* No workers?  Then never mind. */
     204           6 :                 node->nreaders = 0;
     205           6 :                 node->reader = NULL;
     206             :             }
     207         752 :             node->nextreader = 0;
     208             :         }
     209             : 
     210             :         /* Run plan locally if no workers or enabled and not single-copy. */
     211        1516 :         node->need_to_scan_locally = (node->nreaders == 0)
     212         758 :             || (!gather->single_copy && parallel_leader_participation);
     213         758 :         node->initialized = true;
     214             :     }
     215             : 
     216             :     /*
     217             :      * Reset per-tuple memory context to free any expression evaluation
     218             :      * storage allocated in the previous tuple cycle.
     219             :      */
     220     2918300 :     econtext = node->ps.ps_ExprContext;
     221     2918300 :     ResetExprContext(econtext);
     222             : 
     223             :     /*
     224             :      * Get next tuple, either from one of our workers, or by running the plan
     225             :      * ourselves.
     226             :      */
     227     2918300 :     slot = gather_getnext(node);
     228     2918294 :     if (TupIsNull(slot))
     229         752 :         return NULL;
     230             : 
     231             :     /* If no projection is required, we're done. */
     232     2917542 :     if (node->ps.ps_ProjInfo == NULL)
     233     2917542 :         return slot;
     234             : 
     235             :     /*
     236             :      * Form the result tuple using ExecProject(), and return it.
     237             :      */
     238           0 :     econtext->ecxt_outertuple = slot;
     239           0 :     return ExecProject(node->ps.ps_ProjInfo);
     240             : }
     241             : 
     242             : /* ----------------------------------------------------------------
     243             :  *      ExecEndGather
     244             :  *
     245             :  *      frees any storage allocated through C routines.
     246             :  * ----------------------------------------------------------------
     247             :  */
     248             : void
     249         944 : ExecEndGather(GatherState *node)
     250             : {
     251         944 :     ExecEndNode(outerPlanState(node));  /* let children clean up first */
     252         944 :     ExecShutdownGather(node);
     253         944 : }
     254             : 
     255             : /*
     256             :  * Read the next tuple.  We might fetch a tuple from one of the tuple queues
     257             :  * using gather_readnext, or if no tuple queue contains a tuple and the
     258             :  * single_copy flag is not set, we might generate one locally instead.
     259             :  */
     260             : static TupleTableSlot *
     261     2918300 : gather_getnext(GatherState *gatherstate)
     262             : {
     263     2918300 :     PlanState  *outerPlan = outerPlanState(gatherstate);
     264             :     TupleTableSlot *outerTupleSlot;
     265     2918300 :     TupleTableSlot *fslot = gatherstate->funnel_slot;
     266             :     MinimalTuple tup;
     267             : 
     268     2919572 :     while (gatherstate->nreaders > 0 || gatherstate->need_to_scan_locally)
     269             :     {
     270     2918820 :         CHECK_FOR_INTERRUPTS();
     271             : 
     272     2918820 :         if (gatherstate->nreaders > 0)
     273             :         {
     274     2884850 :             tup = gather_readnext(gatherstate);
     275             : 
     276     2884844 :             if (HeapTupleIsValid(tup))
     277             :             {
     278     1441002 :                 ExecStoreMinimalTuple(tup,  /* tuple to store */
     279             :                                       fslot,    /* slot to store the tuple */
     280             :                                       false);   /* don't pfree tuple  */
     281     1441002 :                 return fslot;
     282             :             }
     283             :         }
     284             : 
     285     1477812 :         if (gatherstate->need_to_scan_locally)
     286             :         {
     287     1477098 :             EState     *estate = gatherstate->ps.state;
     288             : 
     289             :             /* Install our DSA area while executing the plan. */
     290     1477098 :             estate->es_query_dsa =
     291     1477098 :                 gatherstate->pei ? gatherstate->pei->area : NULL;
     292     1477098 :             outerTupleSlot = ExecProcNode(outerPlan);
     293     1477098 :             estate->es_query_dsa = NULL;
     294             : 
     295     1477098 :             if (!TupIsNull(outerTupleSlot))
     296     1476540 :                 return outerTupleSlot;
     297             : 
     298         558 :             gatherstate->need_to_scan_locally = false;
     299             :         }
     300             :     }
     301             : 
     302         752 :     return ExecClearTuple(fslot);
     303             : }
     304             : 
     305             : /*
     306             :  * Attempt to read a tuple from one of our parallel workers.
     307             :  */
     308             : static MinimalTuple
     309     2884850 : gather_readnext(GatherState *gatherstate)
     310             : {
     311     2884850 :     int         nvisited = 0;
     312             : 
     313             :     for (;;)
     314     3862454 :     {
     315             :         TupleQueueReader *reader;
     316             :         MinimalTuple tup;
     317             :         bool        readerdone;
     318             : 
     319             :         /* Check for async events, particularly messages from workers. */
     320     6747304 :         CHECK_FOR_INTERRUPTS();
     321             : 
     322             :         /*
     323             :          * Attempt to read a tuple, but don't block if none is available.
     324             :          *
     325             :          * Note that TupleQueueReaderNext will just return NULL for a worker
     326             :          * which fails to initialize.  We'll treat that worker as having
     327             :          * produced no tuples; WaitForParallelWorkersToFinish will error out
     328             :          * when we get there.
     329             :          */
     330             :         Assert(gatherstate->nextreader < gatherstate->nreaders);
     331     6747298 :         reader = gatherstate->reader[gatherstate->nextreader];
     332     6747298 :         tup = TupleQueueReaderNext(reader, true, &readerdone);
     333             : 
     334             :         /*
     335             :          * If this reader is done, remove it from our working array of active
     336             :          * readers.  If all readers are done, we're outta here.
     337             :          */
     338     6747298 :         if (readerdone)
     339             :         {
     340             :             Assert(!tup);
     341        2016 :             --gatherstate->nreaders;
     342        2016 :             if (gatherstate->nreaders == 0)
     343             :             {
     344         740 :                 ExecShutdownGatherWorkers(gatherstate);
     345     2884844 :                 return NULL;
     346             :             }
     347        1276 :             memmove(&gatherstate->reader[gatherstate->nextreader],
     348        1276 :                     &gatherstate->reader[gatherstate->nextreader + 1],
     349             :                     sizeof(TupleQueueReader *)
     350        1276 :                     * (gatherstate->nreaders - gatherstate->nextreader));
     351        1276 :             if (gatherstate->nextreader >= gatherstate->nreaders)
     352        1062 :                 gatherstate->nextreader = 0;
     353        1276 :             continue;
     354             :         }
     355             : 
     356             :         /* If we got a tuple, return it. */
     357     6745282 :         if (tup)
     358     1441002 :             return tup;
     359             : 
     360             :         /*
     361             :          * Advance nextreader pointer in round-robin fashion.  Note that we
     362             :          * only reach this code if we weren't able to get a tuple from the
     363             :          * current worker.  We used to advance the nextreader pointer after
     364             :          * every tuple, but it turns out to be much more efficient to keep
     365             :          * reading from the same queue until that would require blocking.
     366             :          */
     367     5304280 :         gatherstate->nextreader++;
     368     5304280 :         if (gatherstate->nextreader >= gatherstate->nreaders)
     369     1459028 :             gatherstate->nextreader = 0;
     370             : 
     371             :         /* Have we visited every (surviving) TupleQueueReader? */
     372     5304280 :         nvisited++;
     373     5304280 :         if (nvisited >= gatherstate->nreaders)
     374             :         {
     375             :             /*
     376             :              * If (still) running plan locally, return NULL so caller can
     377             :              * generate another tuple from the local copy of the plan.
     378             :              */
     379     1459032 :             if (gatherstate->need_to_scan_locally)
     380     1443102 :                 return NULL;
     381             : 
     382             :             /* Nothing to do except wait for developments. */
     383       15930 :             (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
     384             :                              WAIT_EVENT_EXECUTE_GATHER);
     385       15930 :             ResetLatch(MyLatch);
     386       15930 :             nvisited = 0;
     387             :         }
     388             :     }
     389             : }
     390             : 
     391             : /* ----------------------------------------------------------------
     392             :  *      ExecShutdownGatherWorkers
     393             :  *
     394             :  *      Stop all the parallel workers.
     395             :  * ----------------------------------------------------------------
     396             :  */
     397             : static void
     398        2520 : ExecShutdownGatherWorkers(GatherState *node)
     399             : {
     400        2520 :     if (node->pei != NULL)
     401        1486 :         ExecParallelFinish(node->pei);
     402             : 
     403             :     /* Flush local copy of reader array */
     404        2520 :     if (node->reader)
     405         740 :         pfree(node->reader);
     406        2520 :     node->reader = NULL;
     407        2520 : }
     408             : 
     409             : /* ----------------------------------------------------------------
     410             :  *      ExecShutdownGather
     411             :  *
     412             :  *      Destroy the setup for parallel workers including parallel context.
     413             :  * ----------------------------------------------------------------
     414             :  */
     415             : void
     416        1480 : ExecShutdownGather(GatherState *node)
     417             : {
     418        1480 :     ExecShutdownGatherWorkers(node);
     419             : 
     420             :     /* Now destroy the parallel context. */
     421        1480 :     if (node->pei != NULL)
     422             :     {
     423         518 :         ExecParallelCleanup(node->pei);
     424         518 :         node->pei = NULL;
     425             :     }
     426        1480 : }
     427             : 
     428             : /* ----------------------------------------------------------------
     429             :  *                      Join Support
     430             :  * ----------------------------------------------------------------
     431             :  */
     432             : 
     433             : /* ----------------------------------------------------------------
     434             :  *      ExecReScanGather
     435             :  *
     436             :  *      Prepare to re-scan the result of a Gather.
     437             :  * ----------------------------------------------------------------
     438             :  */
     439             : void
     440         300 : ExecReScanGather(GatherState *node)
     441             : {
     442         300 :     Gather     *gather = (Gather *) node->ps.plan;
     443         300 :     PlanState  *outerPlan = outerPlanState(node);
     444             : 
     445             :     /* Make sure any existing workers are gracefully shut down */
     446         300 :     ExecShutdownGatherWorkers(node);
     447             : 
     448             :     /* Mark node so that shared state will be rebuilt at next call */
     449         300 :     node->initialized = false;
     450             : 
     451             :     /*
     452             :      * Set child node's chgParam to tell it that the next scan might deliver a
     453             :      * different set of rows within the leader process.  (The overall rowset
     454             :      * shouldn't change, but the leader process's subset might; hence nodes
     455             :      * between here and the parallel table scan node mustn't optimize on the
     456             :      * assumption of an unchanging rowset.)
     457             :      */
     458         300 :     if (gather->rescan_param >= 0)
     459         300 :         outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
     460             :                                              gather->rescan_param);
     461             : 
     462             :     /*
     463             :      * If chgParam of subnode is not null then plan will be re-scanned by
     464             :      * first ExecProcNode.  Note: because this does nothing if we have a
     465             :      * rescan_param, it's currently guaranteed that parallel-aware child nodes
     466             :      * will not see a ReScan call until after they get a ReInitializeDSM call.
     467             :      * That ordering might not be something to rely on, though.  A good rule
     468             :      * of thumb is that ReInitializeDSM should reset only shared state, ReScan
     469             :      * should reset only local state, and anything that depends on both of
     470             :      * those steps being finished must wait until the first ExecProcNode call.
     471             :      */
     472         300 :     if (outerPlan->chgParam == NULL)
     473           0 :         ExecReScan(outerPlan);
     474         300 : }

Generated by: LCOV version 1.14