LCOV - code coverage report
Current view: top level - src/backend/executor - nodeGatherMerge.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18devel Lines: 214 218 98.2 %
Date: 2024-12-02 19:14:57 Functions: 14 14 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * nodeGatherMerge.c
       4             :  *      Scan a plan in multiple workers, and do order-preserving merge.
       5             :  *
       6             :  * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  * IDENTIFICATION
      10             :  *    src/backend/executor/nodeGatherMerge.c
      11             :  *
      12             :  *-------------------------------------------------------------------------
      13             :  */
      14             : 
      15             : #include "postgres.h"
      16             : 
      17             : #include "executor/executor.h"
      18             : #include "executor/execParallel.h"
      19             : #include "executor/nodeGatherMerge.h"
      20             : #include "executor/tqueue.h"
      21             : #include "lib/binaryheap.h"
      22             : #include "miscadmin.h"
      23             : #include "optimizer/optimizer.h"
      24             : 
      25             : /*
      26             :  * When we read tuples from workers, it's a good idea to read several at once
      27             :  * for efficiency when possible: this minimizes context-switching overhead.
      28             :  * But reading too many at a time wastes memory without improving performance.
      29             :  * We'll read up to MAX_TUPLE_STORE tuples (in addition to the first one).
      30             :  */
      31             : #define MAX_TUPLE_STORE 10
      32             : 
      33             : /*
      34             :  * Pending-tuple array for each worker.  This holds additional tuples that
      35             :  * we were able to fetch from the worker, but can't process yet.  In addition,
      36             :  * this struct holds the "done" flag indicating the worker is known to have
      37             :  * no more tuples.  (We do not use this struct for the leader; we don't keep
      38             :  * any pending tuples for the leader, and the need_to_scan_locally flag serves
      39             :  * as its "done" indicator.)
      40             :  */
      41             : typedef struct GMReaderTupleBuffer
      42             : {
      43             :     MinimalTuple *tuple;        /* array of length MAX_TUPLE_STORE */
      44             :     int         nTuples;        /* number of tuples currently stored */
      45             :     int         readCounter;    /* index of next tuple to extract */
      46             :     bool        done;           /* true if reader is known exhausted */
      47             : } GMReaderTupleBuffer;
      48             : 
      49             : static TupleTableSlot *ExecGatherMerge(PlanState *pstate);
      50             : static int32 heap_compare_slots(Datum a, Datum b, void *arg);
      51             : static TupleTableSlot *gather_merge_getnext(GatherMergeState *gm_state);
      52             : static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader,
      53             :                                       bool nowait, bool *done);
      54             : static void ExecShutdownGatherMergeWorkers(GatherMergeState *node);
      55             : static void gather_merge_setup(GatherMergeState *gm_state);
      56             : static void gather_merge_init(GatherMergeState *gm_state);
      57             : static void gather_merge_clear_tuples(GatherMergeState *gm_state);
      58             : static bool gather_merge_readnext(GatherMergeState *gm_state, int reader,
      59             :                                   bool nowait);
      60             : static void load_tuple_array(GatherMergeState *gm_state, int reader);
      61             : 
      62             : /* ----------------------------------------------------------------
      63             :  *      ExecInitGather
      64             :  * ----------------------------------------------------------------
      65             :  */
      66             : GatherMergeState *
      67         330 : ExecInitGatherMerge(GatherMerge *node, EState *estate, int eflags)
      68             : {
      69             :     GatherMergeState *gm_state;
      70             :     Plan       *outerNode;
      71             :     TupleDesc   tupDesc;
      72             : 
      73             :     /* Gather merge node doesn't have innerPlan node. */
      74             :     Assert(innerPlan(node) == NULL);
      75             : 
      76             :     /*
      77             :      * create state structure
      78             :      */
      79         330 :     gm_state = makeNode(GatherMergeState);
      80         330 :     gm_state->ps.plan = (Plan *) node;
      81         330 :     gm_state->ps.state = estate;
      82         330 :     gm_state->ps.ExecProcNode = ExecGatherMerge;
      83             : 
      84         330 :     gm_state->initialized = false;
      85         330 :     gm_state->gm_initialized = false;
      86         330 :     gm_state->tuples_needed = -1;
      87             : 
      88             :     /*
      89             :      * Miscellaneous initialization
      90             :      *
      91             :      * create expression context for node
      92             :      */
      93         330 :     ExecAssignExprContext(estate, &gm_state->ps);
      94             : 
      95             :     /*
      96             :      * GatherMerge doesn't support checking a qual (it's always more efficient
      97             :      * to do it in the child node).
      98             :      */
      99             :     Assert(!node->plan.qual);
     100             : 
     101             :     /*
     102             :      * now initialize outer plan
     103             :      */
     104         330 :     outerNode = outerPlan(node);
     105         330 :     outerPlanState(gm_state) = ExecInitNode(outerNode, estate, eflags);
     106             : 
     107             :     /*
     108             :      * Leader may access ExecProcNode result directly (if
     109             :      * need_to_scan_locally), or from workers via tuple queue.  So we can't
     110             :      * trivially rely on the slot type being fixed for expressions evaluated
     111             :      * within this node.
     112             :      */
     113         330 :     gm_state->ps.outeropsset = true;
     114         330 :     gm_state->ps.outeropsfixed = false;
     115             : 
     116             :     /*
     117             :      * Store the tuple descriptor into gather merge state, so we can use it
     118             :      * while initializing the gather merge slots.
     119             :      */
     120         330 :     tupDesc = ExecGetResultType(outerPlanState(gm_state));
     121         330 :     gm_state->tupDesc = tupDesc;
     122             : 
     123             :     /*
     124             :      * Initialize result type and projection.
     125             :      */
     126         330 :     ExecInitResultTypeTL(&gm_state->ps);
     127         330 :     ExecConditionalAssignProjectionInfo(&gm_state->ps, tupDesc, OUTER_VAR);
     128             : 
     129             :     /*
     130             :      * Without projections result slot type is not trivially known, see
     131             :      * comment above.
     132             :      */
     133         330 :     if (gm_state->ps.ps_ProjInfo == NULL)
     134             :     {
     135         318 :         gm_state->ps.resultopsset = true;
     136         318 :         gm_state->ps.resultopsfixed = false;
     137             :     }
     138             : 
     139             :     /*
     140             :      * initialize sort-key information
     141             :      */
     142         330 :     if (node->numCols)
     143             :     {
     144             :         int         i;
     145             : 
     146         330 :         gm_state->gm_nkeys = node->numCols;
     147         330 :         gm_state->gm_sortkeys =
     148         330 :             palloc0(sizeof(SortSupportData) * node->numCols);
     149             : 
     150         774 :         for (i = 0; i < node->numCols; i++)
     151             :         {
     152         444 :             SortSupport sortKey = gm_state->gm_sortkeys + i;
     153             : 
     154         444 :             sortKey->ssup_cxt = CurrentMemoryContext;
     155         444 :             sortKey->ssup_collation = node->collations[i];
     156         444 :             sortKey->ssup_nulls_first = node->nullsFirst[i];
     157         444 :             sortKey->ssup_attno = node->sortColIdx[i];
     158             : 
     159             :             /*
     160             :              * We don't perform abbreviated key conversion here, for the same
     161             :              * reasons that it isn't used in MergeAppend
     162             :              */
     163         444 :             sortKey->abbreviate = false;
     164             : 
     165         444 :             PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
     166             :         }
     167             :     }
     168             : 
     169             :     /* Now allocate the workspace for gather merge */
     170         330 :     gather_merge_setup(gm_state);
     171             : 
     172         330 :     return gm_state;
     173             : }
     174             : 
     175             : /* ----------------------------------------------------------------
     176             :  *      ExecGatherMerge(node)
     177             :  *
     178             :  *      Scans the relation via multiple workers and returns
     179             :  *      the next qualifying tuple.
     180             :  * ----------------------------------------------------------------
     181             :  */
     182             : static TupleTableSlot *
     183      254186 : ExecGatherMerge(PlanState *pstate)
     184             : {
     185      254186 :     GatherMergeState *node = castNode(GatherMergeState, pstate);
     186             :     TupleTableSlot *slot;
     187             :     ExprContext *econtext;
     188             : 
     189      254186 :     CHECK_FOR_INTERRUPTS();
     190             : 
     191             :     /*
     192             :      * As with Gather, we don't launch workers until this node is actually
     193             :      * executed.
     194             :      */
     195      254186 :     if (!node->initialized)
     196             :     {
     197         162 :         EState     *estate = node->ps.state;
     198         162 :         GatherMerge *gm = castNode(GatherMerge, node->ps.plan);
     199             : 
     200             :         /*
     201             :          * Sometimes we might have to run without parallelism; but if parallel
     202             :          * mode is active then we can try to fire up some workers.
     203             :          */
     204         162 :         if (gm->num_workers > 0 && estate->es_use_parallel_mode)
     205             :         {
     206             :             ParallelContext *pcxt;
     207             : 
     208             :             /* Initialize, or re-initialize, shared state needed by workers. */
     209         162 :             if (!node->pei)
     210         132 :                 node->pei = ExecInitParallelPlan(outerPlanState(node),
     211             :                                                  estate,
     212             :                                                  gm->initParam,
     213             :                                                  gm->num_workers,
     214             :                                                  node->tuples_needed);
     215             :             else
     216          30 :                 ExecParallelReinitialize(outerPlanState(node),
     217          30 :                                          node->pei,
     218             :                                          gm->initParam);
     219             : 
     220             :             /* Try to launch workers. */
     221         162 :             pcxt = node->pei->pcxt;
     222         162 :             LaunchParallelWorkers(pcxt);
     223             :             /* We save # workers launched for the benefit of EXPLAIN */
     224         162 :             node->nworkers_launched = pcxt->nworkers_launched;
     225             : 
     226             :             /*
     227             :              * Count number of workers originally wanted and actually
     228             :              * launched.
     229             :              */
     230         162 :             estate->es_parallel_workers_to_launch += pcxt->nworkers_to_launch;
     231         162 :             estate->es_parallel_workers_launched += pcxt->nworkers_launched;
     232             : 
     233             :             /* Set up tuple queue readers to read the results. */
     234         162 :             if (pcxt->nworkers_launched > 0)
     235             :             {
     236         150 :                 ExecParallelCreateReaders(node->pei);
     237             :                 /* Make a working array showing the active readers */
     238         150 :                 node->nreaders = pcxt->nworkers_launched;
     239         150 :                 node->reader = (TupleQueueReader **)
     240         150 :                     palloc(node->nreaders * sizeof(TupleQueueReader *));
     241         150 :                 memcpy(node->reader, node->pei->reader,
     242         150 :                        node->nreaders * sizeof(TupleQueueReader *));
     243             :             }
     244             :             else
     245             :             {
     246             :                 /* No workers?  Then never mind. */
     247          12 :                 node->nreaders = 0;
     248          12 :                 node->reader = NULL;
     249             :             }
     250             :         }
     251             : 
     252             :         /* allow leader to participate if enabled or no choice */
     253         162 :         if (parallel_leader_participation || node->nreaders == 0)
     254         156 :             node->need_to_scan_locally = true;
     255         162 :         node->initialized = true;
     256             :     }
     257             : 
     258             :     /*
     259             :      * Reset per-tuple memory context to free any expression evaluation
     260             :      * storage allocated in the previous tuple cycle.
     261             :      */
     262      254186 :     econtext = node->ps.ps_ExprContext;
     263      254186 :     ResetExprContext(econtext);
     264             : 
     265             :     /*
     266             :      * Get next tuple, either from one of our workers, or by running the plan
     267             :      * ourselves.
     268             :      */
     269      254186 :     slot = gather_merge_getnext(node);
     270      254186 :     if (TupIsNull(slot))
     271         126 :         return NULL;
     272             : 
     273             :     /* If no projection is required, we're done. */
     274      254060 :     if (node->ps.ps_ProjInfo == NULL)
     275      254060 :         return slot;
     276             : 
     277             :     /*
     278             :      * Form the result tuple using ExecProject(), and return it.
     279             :      */
     280           0 :     econtext->ecxt_outertuple = slot;
     281           0 :     return ExecProject(node->ps.ps_ProjInfo);
     282             : }
     283             : 
     284             : /* ----------------------------------------------------------------
     285             :  *      ExecEndGatherMerge
     286             :  *
     287             :  *      frees any storage allocated through C routines.
     288             :  * ----------------------------------------------------------------
     289             :  */
     290             : void
     291         330 : ExecEndGatherMerge(GatherMergeState *node)
     292             : {
     293         330 :     ExecEndNode(outerPlanState(node));  /* let children clean up first */
     294         330 :     ExecShutdownGatherMerge(node);
     295         330 : }
     296             : 
     297             : /* ----------------------------------------------------------------
     298             :  *      ExecShutdownGatherMerge
     299             :  *
     300             :  *      Destroy the setup for parallel workers including parallel context.
     301             :  * ----------------------------------------------------------------
     302             :  */
     303             : void
     304         462 : ExecShutdownGatherMerge(GatherMergeState *node)
     305             : {
     306         462 :     ExecShutdownGatherMergeWorkers(node);
     307             : 
     308             :     /* Now destroy the parallel context. */
     309         462 :     if (node->pei != NULL)
     310             :     {
     311         132 :         ExecParallelCleanup(node->pei);
     312         132 :         node->pei = NULL;
     313             :     }
     314         462 : }
     315             : 
     316             : /* ----------------------------------------------------------------
     317             :  *      ExecShutdownGatherMergeWorkers
     318             :  *
     319             :  *      Stop all the parallel workers.
     320             :  * ----------------------------------------------------------------
     321             :  */
     322             : static void
     323         510 : ExecShutdownGatherMergeWorkers(GatherMergeState *node)
     324             : {
     325         510 :     if (node->pei != NULL)
     326         162 :         ExecParallelFinish(node->pei);
     327             : 
     328             :     /* Flush local copy of reader array */
     329         510 :     if (node->reader)
     330         150 :         pfree(node->reader);
     331         510 :     node->reader = NULL;
     332         510 : }
     333             : 
     334             : /* ----------------------------------------------------------------
     335             :  *      ExecReScanGatherMerge
     336             :  *
     337             :  *      Prepare to re-scan the result of a GatherMerge.
     338             :  * ----------------------------------------------------------------
     339             :  */
     340             : void
     341          48 : ExecReScanGatherMerge(GatherMergeState *node)
     342             : {
     343          48 :     GatherMerge *gm = (GatherMerge *) node->ps.plan;
     344          48 :     PlanState  *outerPlan = outerPlanState(node);
     345             : 
     346             :     /* Make sure any existing workers are gracefully shut down */
     347          48 :     ExecShutdownGatherMergeWorkers(node);
     348             : 
     349             :     /* Free any unused tuples, so we don't leak memory across rescans */
     350          48 :     gather_merge_clear_tuples(node);
     351             : 
     352             :     /* Mark node so that shared state will be rebuilt at next call */
     353          48 :     node->initialized = false;
     354          48 :     node->gm_initialized = false;
     355             : 
     356             :     /*
     357             :      * Set child node's chgParam to tell it that the next scan might deliver a
     358             :      * different set of rows within the leader process.  (The overall rowset
     359             :      * shouldn't change, but the leader process's subset might; hence nodes
     360             :      * between here and the parallel table scan node mustn't optimize on the
     361             :      * assumption of an unchanging rowset.)
     362             :      */
     363          48 :     if (gm->rescan_param >= 0)
     364          48 :         outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
     365             :                                              gm->rescan_param);
     366             : 
     367             :     /*
     368             :      * If chgParam of subnode is not null then plan will be re-scanned by
     369             :      * first ExecProcNode.  Note: because this does nothing if we have a
     370             :      * rescan_param, it's currently guaranteed that parallel-aware child nodes
     371             :      * will not see a ReScan call until after they get a ReInitializeDSM call.
     372             :      * That ordering might not be something to rely on, though.  A good rule
     373             :      * of thumb is that ReInitializeDSM should reset only shared state, ReScan
     374             :      * should reset only local state, and anything that depends on both of
     375             :      * those steps being finished must wait until the first ExecProcNode call.
     376             :      */
     377          48 :     if (outerPlan->chgParam == NULL)
     378           0 :         ExecReScan(outerPlan);
     379          48 : }
     380             : 
     381             : /*
     382             :  * Set up the data structures that we'll need for Gather Merge.
     383             :  *
     384             :  * We allocate these once on the basis of gm->num_workers, which is an
     385             :  * upper bound for the number of workers we'll actually have.  During
     386             :  * a rescan, we reset the structures to empty.  This approach simplifies
     387             :  * not leaking memory across rescans.
     388             :  *
     389             :  * In the gm_slots[] array, index 0 is for the leader, and indexes 1 to n
     390             :  * are for workers.  The values placed into gm_heap correspond to indexes
     391             :  * in gm_slots[].  The gm_tuple_buffers[] array, however, is indexed from
     392             :  * 0 to n-1; it has no entry for the leader.
     393             :  */
     394             : static void
     395         330 : gather_merge_setup(GatherMergeState *gm_state)
     396             : {
     397         330 :     GatherMerge *gm = castNode(GatherMerge, gm_state->ps.plan);
     398         330 :     int         nreaders = gm->num_workers;
     399             :     int         i;
     400             : 
     401             :     /*
     402             :      * Allocate gm_slots for the number of workers + one more slot for leader.
     403             :      * Slot 0 is always for the leader.  Leader always calls ExecProcNode() to
     404             :      * read the tuple, and then stores it directly into its gm_slots entry.
     405             :      * For other slots, code below will call ExecInitExtraTupleSlot() to
     406             :      * create a slot for the worker's results.  Note that during any single
     407             :      * scan, we might have fewer than num_workers available workers, in which
     408             :      * case the extra array entries go unused.
     409             :      */
     410         330 :     gm_state->gm_slots = (TupleTableSlot **)
     411         330 :         palloc0((nreaders + 1) * sizeof(TupleTableSlot *));
     412             : 
     413             :     /* Allocate the tuple slot and tuple array for each worker */
     414         330 :     gm_state->gm_tuple_buffers = (GMReaderTupleBuffer *)
     415         330 :         palloc0(nreaders * sizeof(GMReaderTupleBuffer));
     416             : 
     417        1218 :     for (i = 0; i < nreaders; i++)
     418             :     {
     419             :         /* Allocate the tuple array with length MAX_TUPLE_STORE */
     420        1776 :         gm_state->gm_tuple_buffers[i].tuple =
     421         888 :             (MinimalTuple *) palloc0(sizeof(MinimalTuple) * MAX_TUPLE_STORE);
     422             : 
     423             :         /* Initialize tuple slot for worker */
     424         888 :         gm_state->gm_slots[i + 1] =
     425         888 :             ExecInitExtraTupleSlot(gm_state->ps.state, gm_state->tupDesc,
     426             :                                    &TTSOpsMinimalTuple);
     427             :     }
     428             : 
     429             :     /* Allocate the resources for the merge */
     430         330 :     gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
     431             :                                             heap_compare_slots,
     432             :                                             gm_state);
     433         330 : }
     434             : 
     435             : /*
     436             :  * Initialize the Gather Merge.
     437             :  *
     438             :  * Reset data structures to ensure they're empty.  Then pull at least one
     439             :  * tuple from leader + each worker (or set its "done" indicator), and set up
     440             :  * the heap.
     441             :  */
     442             : static void
     443         162 : gather_merge_init(GatherMergeState *gm_state)
     444             : {
     445         162 :     int         nreaders = gm_state->nreaders;
     446         162 :     bool        nowait = true;
     447             :     int         i;
     448             : 
     449             :     /* Assert that gather_merge_setup made enough space */
     450             :     Assert(nreaders <= castNode(GatherMerge, gm_state->ps.plan)->num_workers);
     451             : 
     452             :     /* Reset leader's tuple slot to empty */
     453         162 :     gm_state->gm_slots[0] = NULL;
     454             : 
     455             :     /* Reset the tuple slot and tuple array for each worker */
     456         576 :     for (i = 0; i < nreaders; i++)
     457             :     {
     458             :         /* Reset tuple array to empty */
     459         414 :         gm_state->gm_tuple_buffers[i].nTuples = 0;
     460         414 :         gm_state->gm_tuple_buffers[i].readCounter = 0;
     461             :         /* Reset done flag to not-done */
     462         414 :         gm_state->gm_tuple_buffers[i].done = false;
     463             :         /* Ensure output slot is empty */
     464         414 :         ExecClearTuple(gm_state->gm_slots[i + 1]);
     465             :     }
     466             : 
     467             :     /* Reset binary heap to empty */
     468         162 :     binaryheap_reset(gm_state->gm_heap);
     469             : 
     470             :     /*
     471             :      * First, try to read a tuple from each worker (including leader) in
     472             :      * nowait mode.  After this, if not all workers were able to produce a
     473             :      * tuple (or a "done" indication), then re-read from remaining workers,
     474             :      * this time using wait mode.  Add all live readers (those producing at
     475             :      * least one tuple) to the heap.
     476             :      */
     477         308 : reread:
     478        1436 :     for (i = 0; i <= nreaders; i++)
     479             :     {
     480        1128 :         CHECK_FOR_INTERRUPTS();
     481             : 
     482             :         /* skip this source if already known done */
     483        1948 :         if ((i == 0) ? gm_state->need_to_scan_locally :
     484         820 :             !gm_state->gm_tuple_buffers[i - 1].done)
     485             :         {
     486        1104 :             if (TupIsNull(gm_state->gm_slots[i]))
     487             :             {
     488             :                 /* Don't have a tuple yet, try to get one */
     489         960 :                 if (gather_merge_readnext(gm_state, i, nowait))
     490         170 :                     binaryheap_add_unordered(gm_state->gm_heap,
     491             :                                              Int32GetDatum(i));
     492             :             }
     493             :             else
     494             :             {
     495             :                 /*
     496             :                  * We already got at least one tuple from this worker, but
     497             :                  * might as well see if it has any more ready by now.
     498             :                  */
     499         144 :                 load_tuple_array(gm_state, i);
     500             :             }
     501             :         }
     502             :     }
     503             : 
     504             :     /* need not recheck leader, since nowait doesn't matter for it */
     505         722 :     for (i = 1; i <= nreaders; i++)
     506             :     {
     507         560 :         if (!gm_state->gm_tuple_buffers[i - 1].done &&
     508         156 :             TupIsNull(gm_state->gm_slots[i]))
     509             :         {
     510         146 :             nowait = false;
     511         146 :             goto reread;
     512             :         }
     513             :     }
     514             : 
     515             :     /* Now heapify the heap. */
     516         162 :     binaryheap_build(gm_state->gm_heap);
     517             : 
     518         162 :     gm_state->gm_initialized = true;
     519         162 : }
     520             : 
     521             : /*
     522             :  * Clear out the tuple table slot, and any unused pending tuples,
     523             :  * for each gather merge input.
     524             :  */
     525             : static void
     526         174 : gather_merge_clear_tuples(GatherMergeState *gm_state)
     527             : {
     528             :     int         i;
     529             : 
     530         648 :     for (i = 0; i < gm_state->nreaders; i++)
     531             :     {
     532         474 :         GMReaderTupleBuffer *tuple_buffer = &gm_state->gm_tuple_buffers[i];
     533             : 
     534         474 :         while (tuple_buffer->readCounter < tuple_buffer->nTuples)
     535           0 :             pfree(tuple_buffer->tuple[tuple_buffer->readCounter++]);
     536             : 
     537         474 :         ExecClearTuple(gm_state->gm_slots[i + 1]);
     538             :     }
     539         174 : }
     540             : 
     541             : /*
     542             :  * Read the next tuple for gather merge.
     543             :  *
     544             :  * Fetch the sorted tuple out of the heap.
     545             :  */
     546             : static TupleTableSlot *
     547      254186 : gather_merge_getnext(GatherMergeState *gm_state)
     548             : {
     549             :     int         i;
     550             : 
     551      254186 :     if (!gm_state->gm_initialized)
     552             :     {
     553             :         /*
     554             :          * First time through: pull the first tuple from each participant, and
     555             :          * set up the heap.
     556             :          */
     557         162 :         gather_merge_init(gm_state);
     558             :     }
     559             :     else
     560             :     {
     561             :         /*
     562             :          * Otherwise, pull the next tuple from whichever participant we
     563             :          * returned from last time, and reinsert that participant's index into
     564             :          * the heap, because it might now compare differently against the
     565             :          * other elements of the heap.
     566             :          */
     567      254024 :         i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
     568             : 
     569      254024 :         if (gather_merge_readnext(gm_state, i, false))
     570      253890 :             binaryheap_replace_first(gm_state->gm_heap, Int32GetDatum(i));
     571             :         else
     572             :         {
     573             :             /* reader exhausted, remove it from heap */
     574         134 :             (void) binaryheap_remove_first(gm_state->gm_heap);
     575             :         }
     576             :     }
     577             : 
     578      254186 :     if (binaryheap_empty(gm_state->gm_heap))
     579             :     {
     580             :         /* All the queues are exhausted, and so is the heap */
     581         126 :         gather_merge_clear_tuples(gm_state);
     582         126 :         return NULL;
     583             :     }
     584             :     else
     585             :     {
     586             :         /* Return next tuple from whichever participant has the leading one */
     587      254060 :         i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
     588      254060 :         return gm_state->gm_slots[i];
     589             :     }
     590             : }
     591             : 
     592             : /*
     593             :  * Read tuple(s) for given reader in nowait mode, and load into its tuple
     594             :  * array, until we have MAX_TUPLE_STORE of them or would have to block.
     595             :  */
     596             : static void
     597         168 : load_tuple_array(GatherMergeState *gm_state, int reader)
     598             : {
     599             :     GMReaderTupleBuffer *tuple_buffer;
     600             :     int         i;
     601             : 
     602             :     /* Don't do anything if this is the leader. */
     603         168 :     if (reader == 0)
     604         140 :         return;
     605             : 
     606          28 :     tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
     607             : 
     608             :     /* If there's nothing in the array, reset the counters to zero. */
     609          28 :     if (tuple_buffer->nTuples == tuple_buffer->readCounter)
     610          24 :         tuple_buffer->nTuples = tuple_buffer->readCounter = 0;
     611             : 
     612             :     /* Try to fill additional slots in the array. */
     613         228 :     for (i = tuple_buffer->nTuples; i < MAX_TUPLE_STORE; i++)
     614             :     {
     615             :         MinimalTuple tuple;
     616             : 
     617         214 :         tuple = gm_readnext_tuple(gm_state,
     618             :                                   reader,
     619             :                                   true,
     620             :                                   &tuple_buffer->done);
     621         214 :         if (!tuple)
     622          14 :             break;
     623         200 :         tuple_buffer->tuple[i] = tuple;
     624         200 :         tuple_buffer->nTuples++;
     625             :     }
     626             : }
     627             : 
     628             : /*
     629             :  * Store the next tuple for a given reader into the appropriate slot.
     630             :  *
     631             :  * Returns true if successful, false if not (either reader is exhausted,
     632             :  * or we didn't want to wait for a tuple).  Sets done flag if reader
     633             :  * is found to be exhausted.
     634             :  */
     635             : static bool
     636      254984 : gather_merge_readnext(GatherMergeState *gm_state, int reader, bool nowait)
     637             : {
     638             :     GMReaderTupleBuffer *tuple_buffer;
     639             :     MinimalTuple tup;
     640             : 
     641             :     /*
     642             :      * If we're being asked to generate a tuple from the leader, then we just
     643             :      * call ExecProcNode as normal to produce one.
     644             :      */
     645      254984 :     if (reader == 0)
     646             :     {
     647      253956 :         if (gm_state->need_to_scan_locally)
     648             :         {
     649      253956 :             PlanState  *outerPlan = outerPlanState(gm_state);
     650             :             TupleTableSlot *outerTupleSlot;
     651      253956 :             EState     *estate = gm_state->ps.state;
     652             : 
     653             :             /* Install our DSA area while executing the plan. */
     654      253956 :             estate->es_query_dsa = gm_state->pei ? gm_state->pei->area : NULL;
     655      253956 :             outerTupleSlot = ExecProcNode(outerPlan);
     656      253956 :             estate->es_query_dsa = NULL;
     657             : 
     658      253956 :             if (!TupIsNull(outerTupleSlot))
     659             :             {
     660      253836 :                 gm_state->gm_slots[0] = outerTupleSlot;
     661      253836 :                 return true;
     662             :             }
     663             :             /* need_to_scan_locally serves as "done" flag for leader */
     664         120 :             gm_state->need_to_scan_locally = false;
     665             :         }
     666         120 :         return false;
     667             :     }
     668             : 
     669             :     /* Otherwise, check the state of the relevant tuple buffer. */
     670        1028 :     tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
     671             : 
     672        1028 :     if (tuple_buffer->nTuples > tuple_buffer->readCounter)
     673             :     {
     674             :         /* Return any tuple previously read that is still buffered. */
     675         200 :         tup = tuple_buffer->tuple[tuple_buffer->readCounter++];
     676             :     }
     677         828 :     else if (tuple_buffer->done)
     678             :     {
     679             :         /* Reader is known to be exhausted. */
     680          14 :         return false;
     681             :     }
     682             :     else
     683             :     {
     684             :         /* Read and buffer next tuple. */
     685         814 :         tup = gm_readnext_tuple(gm_state,
     686             :                                 reader,
     687             :                                 nowait,
     688             :                                 &tuple_buffer->done);
     689         814 :         if (!tup)
     690         790 :             return false;
     691             : 
     692             :         /*
     693             :          * Attempt to read more tuples in nowait mode and store them in the
     694             :          * pending-tuple array for the reader.
     695             :          */
     696          24 :         load_tuple_array(gm_state, reader);
     697             :     }
     698             : 
     699             :     Assert(tup);
     700             : 
     701             :     /* Build the TupleTableSlot for the given tuple */
     702         224 :     ExecStoreMinimalTuple(tup,  /* tuple to store */
     703         224 :                           gm_state->gm_slots[reader],    /* slot in which to
     704             :                                                          * store the tuple */
     705             :                           true);    /* pfree tuple when done with it */
     706             : 
     707         224 :     return true;
     708             : }
     709             : 
     710             : /*
     711             :  * Attempt to read a tuple from given worker.
     712             :  */
     713             : static MinimalTuple
     714        1028 : gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait,
     715             :                   bool *done)
     716             : {
     717             :     TupleQueueReader *reader;
     718             :     MinimalTuple tup;
     719             : 
     720             :     /* Check for async events, particularly messages from workers. */
     721        1028 :     CHECK_FOR_INTERRUPTS();
     722             : 
     723             :     /*
     724             :      * Attempt to read a tuple.
     725             :      *
     726             :      * Note that TupleQueueReaderNext will just return NULL for a worker which
     727             :      * fails to initialize.  We'll treat that worker as having produced no
     728             :      * tuples; WaitForParallelWorkersToFinish will error out when we get
     729             :      * there.
     730             :      */
     731        1028 :     reader = gm_state->reader[nreader - 1];
     732        1028 :     tup = TupleQueueReaderNext(reader, nowait, done);
     733             : 
     734             :     /*
     735             :      * Since we'll be buffering these across multiple calls, we need to make a
     736             :      * copy.
     737             :      */
     738        1028 :     return tup ? heap_copy_minimal_tuple(tup) : NULL;
     739             : }
     740             : 
     741             : /*
     742             :  * We have one slot for each item in the heap array.  We use SlotNumber
     743             :  * to store slot indexes.  This doesn't actually provide any formal
     744             :  * type-safety, but it makes the code more self-documenting.
     745             :  */
     746             : typedef int32 SlotNumber;
     747             : 
     748             : /*
     749             :  * Compare the tuples in the two given slots.
     750             :  */
     751             : static int32
     752         344 : heap_compare_slots(Datum a, Datum b, void *arg)
     753             : {
     754         344 :     GatherMergeState *node = (GatherMergeState *) arg;
     755         344 :     SlotNumber  slot1 = DatumGetInt32(a);
     756         344 :     SlotNumber  slot2 = DatumGetInt32(b);
     757             : 
     758         344 :     TupleTableSlot *s1 = node->gm_slots[slot1];
     759         344 :     TupleTableSlot *s2 = node->gm_slots[slot2];
     760             :     int         nkey;
     761             : 
     762             :     Assert(!TupIsNull(s1));
     763             :     Assert(!TupIsNull(s2));
     764             : 
     765         520 :     for (nkey = 0; nkey < node->gm_nkeys; nkey++)
     766             :     {
     767         344 :         SortSupport sortKey = node->gm_sortkeys + nkey;
     768         344 :         AttrNumber  attno = sortKey->ssup_attno;
     769             :         Datum       datum1,
     770             :                     datum2;
     771             :         bool        isNull1,
     772             :                     isNull2;
     773             :         int         compare;
     774             : 
     775         344 :         datum1 = slot_getattr(s1, attno, &isNull1);
     776         344 :         datum2 = slot_getattr(s2, attno, &isNull2);
     777             : 
     778         344 :         compare = ApplySortComparator(datum1, isNull1,
     779             :                                       datum2, isNull2,
     780             :                                       sortKey);
     781         344 :         if (compare != 0)
     782             :         {
     783         168 :             INVERT_COMPARE_RESULT(compare);
     784         168 :             return compare;
     785             :         }
     786             :     }
     787         176 :     return 0;
     788             : }

Generated by: LCOV version 1.14