LCOV - code coverage report
Current view: top level - src/backend/executor - nodeMergeAppend.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 106 111 95.5 %
Date: 2020-06-05 20:07:15 Functions: 5 5 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * nodeMergeAppend.c
       4             :  *    routines to handle MergeAppend nodes.
       5             :  *
       6             :  * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *    src/backend/executor/nodeMergeAppend.c
      12             :  *
      13             :  *-------------------------------------------------------------------------
      14             :  */
      15             : /* INTERFACE ROUTINES
      16             :  *      ExecInitMergeAppend     - initialize the MergeAppend node
      17             :  *      ExecMergeAppend         - retrieve the next tuple from the node
      18             :  *      ExecEndMergeAppend      - shut down the MergeAppend node
      19             :  *      ExecReScanMergeAppend   - rescan the MergeAppend node
      20             :  *
      21             :  *   NOTES
      22             :  *      A MergeAppend node contains a list of one or more subplans.
      23             :  *      These are each expected to deliver tuples that are sorted according
      24             :  *      to a common sort key.  The MergeAppend node merges these streams
      25             :  *      to produce output sorted the same way.
      26             :  *
      27             :  *      MergeAppend nodes don't make use of their left and right
      28             :  *      subtrees, rather they maintain a list of subplans so
      29             :  *      a typical MergeAppend node looks like this in the plan tree:
      30             :  *
      31             :  *                 ...
      32             :  *                 /
      33             :  *              MergeAppend---+------+------+--- nil
      34             :  *              /   \         |      |      |
      35             :  *            nil   nil      ...    ...    ...
      36             :  *                               subplans
      37             :  */
      38             : 
      39             : #include "postgres.h"
      40             : 
      41             : #include "executor/execdebug.h"
      42             : #include "executor/execPartition.h"
      43             : #include "executor/nodeMergeAppend.h"
      44             : #include "lib/binaryheap.h"
      45             : #include "miscadmin.h"
      46             : 
      47             : /*
      48             :  * We have one slot for each item in the heap array.  We use SlotNumber
      49             :  * to store slot indexes.  This doesn't actually provide any formal
      50             :  * type-safety, but it makes the code more self-documenting.
      51             :  */
      52             : typedef int32 SlotNumber;
      53             : 
      54             : static TupleTableSlot *ExecMergeAppend(PlanState *pstate);
      55             : static int  heap_compare_slots(Datum a, Datum b, void *arg);
      56             : 
      57             : 
      58             : /* ----------------------------------------------------------------
      59             :  *      ExecInitMergeAppend
      60             :  *
      61             :  *      Begin all of the subscans of the MergeAppend node.
      62             :  * ----------------------------------------------------------------
      63             :  */
      64             : MergeAppendState *
      65         262 : ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
      66             : {
      67         262 :     MergeAppendState *mergestate = makeNode(MergeAppendState);
      68             :     PlanState **mergeplanstates;
      69             :     Bitmapset  *validsubplans;
      70             :     int         nplans;
      71             :     int         i,
      72             :                 j;
      73             : 
      74             :     /* check for unsupported flags */
      75             :     Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
      76             : 
      77             :     /*
      78             :      * create new MergeAppendState for our node
      79             :      */
      80         262 :     mergestate->ps.plan = (Plan *) node;
      81         262 :     mergestate->ps.state = estate;
      82         262 :     mergestate->ps.ExecProcNode = ExecMergeAppend;
      83             : 
      84             :     /* If run-time partition pruning is enabled, then set that up now */
      85         262 :     if (node->part_prune_info != NULL)
      86             :     {
      87             :         PartitionPruneState *prunestate;
      88             : 
      89             :         /* We may need an expression context to evaluate partition exprs */
      90          32 :         ExecAssignExprContext(estate, &mergestate->ps);
      91             : 
      92          32 :         prunestate = ExecCreatePartitionPruneState(&mergestate->ps,
      93          32 :                                                    node->part_prune_info);
      94          32 :         mergestate->ms_prune_state = prunestate;
      95             : 
      96             :         /* Perform an initial partition prune, if required. */
      97          32 :         if (prunestate->do_initial_prune)
      98             :         {
      99             :             /* Determine which subplans survive initial pruning */
     100          28 :             validsubplans = ExecFindInitialMatchingSubPlans(prunestate,
     101          28 :                                                             list_length(node->mergeplans));
     102             : 
     103          28 :             nplans = bms_num_members(validsubplans);
     104             :         }
     105             :         else
     106             :         {
     107             :             /* We'll need to initialize all subplans */
     108           4 :             nplans = list_length(node->mergeplans);
     109             :             Assert(nplans > 0);
     110           4 :             validsubplans = bms_add_range(NULL, 0, nplans - 1);
     111             :         }
     112             : 
     113             :         /*
     114             :          * When no run-time pruning is required and there's at least one
     115             :          * subplan, we can fill as_valid_subplans immediately, preventing
     116             :          * later calls to ExecFindMatchingSubPlans.
     117             :          */
     118          32 :         if (!prunestate->do_exec_prune && nplans > 0)
     119          16 :             mergestate->ms_valid_subplans = bms_add_range(NULL, 0, nplans - 1);
     120             :     }
     121             :     else
     122             :     {
     123         230 :         nplans = list_length(node->mergeplans);
     124             : 
     125             :         /*
     126             :          * When run-time partition pruning is not enabled we can just mark all
     127             :          * subplans as valid; they must also all be initialized.
     128             :          */
     129             :         Assert(nplans > 0);
     130         230 :         mergestate->ms_valid_subplans = validsubplans =
     131         230 :             bms_add_range(NULL, 0, nplans - 1);
     132         230 :         mergestate->ms_prune_state = NULL;
     133             :     }
     134             : 
     135         262 :     mergeplanstates = (PlanState **) palloc(nplans * sizeof(PlanState *));
     136         262 :     mergestate->mergeplans = mergeplanstates;
     137         262 :     mergestate->ms_nplans = nplans;
     138             : 
     139         262 :     mergestate->ms_slots = (TupleTableSlot **) palloc0(sizeof(TupleTableSlot *) * nplans);
     140         262 :     mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots,
     141             :                                               mergestate);
     142             : 
     143             :     /*
     144             :      * Miscellaneous initialization
     145             :      *
     146             :      * MergeAppend nodes do have Result slots, which hold pointers to tuples,
     147             :      * so we have to initialize them.  FIXME
     148             :      */
     149         262 :     ExecInitResultTupleSlotTL(&mergestate->ps, &TTSOpsVirtual);
     150             : 
     151             :     /* node returns slots from each of its subnodes, therefore not fixed */
     152         262 :     mergestate->ps.resultopsset = true;
     153         262 :     mergestate->ps.resultopsfixed = false;
     154             : 
     155             :     /*
     156             :      * call ExecInitNode on each of the valid plans to be executed and save
     157             :      * the results into the mergeplanstates array.
     158             :      */
     159         262 :     j = 0;
     160         262 :     i = -1;
     161        1010 :     while ((i = bms_next_member(validsubplans, i)) >= 0)
     162             :     {
     163         748 :         Plan       *initNode = (Plan *) list_nth(node->mergeplans, i);
     164             : 
     165         748 :         mergeplanstates[j++] = ExecInitNode(initNode, estate, eflags);
     166             :     }
     167             : 
     168         262 :     mergestate->ps.ps_ProjInfo = NULL;
     169             : 
     170             :     /*
     171             :      * initialize sort-key information
     172             :      */
     173         262 :     mergestate->ms_nkeys = node->numCols;
     174         262 :     mergestate->ms_sortkeys = palloc0(sizeof(SortSupportData) * node->numCols);
     175             : 
     176         564 :     for (i = 0; i < node->numCols; i++)
     177             :     {
     178         302 :         SortSupport sortKey = mergestate->ms_sortkeys + i;
     179             : 
     180         302 :         sortKey->ssup_cxt = CurrentMemoryContext;
     181         302 :         sortKey->ssup_collation = node->collations[i];
     182         302 :         sortKey->ssup_nulls_first = node->nullsFirst[i];
     183         302 :         sortKey->ssup_attno = node->sortColIdx[i];
     184             : 
     185             :         /*
     186             :          * It isn't feasible to perform abbreviated key conversion, since
     187             :          * tuples are pulled into mergestate's binary heap as needed.  It
     188             :          * would likely be counter-productive to convert tuples into an
     189             :          * abbreviated representation as they're pulled up, so opt out of that
     190             :          * additional optimization entirely.
     191             :          */
     192         302 :         sortKey->abbreviate = false;
     193             : 
     194         302 :         PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
     195             :     }
     196             : 
     197             :     /*
     198             :      * initialize to show we have not run the subplans yet
     199             :      */
     200         262 :     mergestate->ms_initialized = false;
     201             : 
     202         262 :     return mergestate;
     203             : }
     204             : 
     205             : /* ----------------------------------------------------------------
     206             :  *     ExecMergeAppend
     207             :  *
     208             :  *      Handles iteration over multiple subplans.
     209             :  * ----------------------------------------------------------------
     210             :  */
     211             : static TupleTableSlot *
     212        8956 : ExecMergeAppend(PlanState *pstate)
     213             : {
     214        8956 :     MergeAppendState *node = castNode(MergeAppendState, pstate);
     215             :     TupleTableSlot *result;
     216             :     SlotNumber  i;
     217             : 
     218        8956 :     CHECK_FOR_INTERRUPTS();
     219             : 
     220        8956 :     if (!node->ms_initialized)
     221             :     {
     222             :         /* Nothing to do if all subplans were pruned */
     223         142 :         if (node->ms_nplans == 0)
     224          12 :             return ExecClearTuple(node->ps.ps_ResultTupleSlot);
     225             : 
     226             :         /*
     227             :          * If we've yet to determine the valid subplans then do so now.  If
     228             :          * run-time pruning is disabled then the valid subplans will always be
     229             :          * set to all subplans.
     230             :          */
     231         130 :         if (node->ms_valid_subplans == NULL)
     232           4 :             node->ms_valid_subplans =
     233           4 :                 ExecFindMatchingSubPlans(node->ms_prune_state);
     234             : 
     235             :         /*
     236             :          * First time through: pull the first tuple from each valid subplan,
     237             :          * and set up the heap.
     238             :          */
     239         130 :         i = -1;
     240         510 :         while ((i = bms_next_member(node->ms_valid_subplans, i)) >= 0)
     241             :         {
     242         380 :             node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
     243         380 :             if (!TupIsNull(node->ms_slots[i]))
     244         302 :                 binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
     245             :         }
     246         130 :         binaryheap_build(node->ms_heap);
     247         130 :         node->ms_initialized = true;
     248             :     }
     249             :     else
     250             :     {
     251             :         /*
     252             :          * Otherwise, pull the next tuple from whichever subplan we returned
     253             :          * from last time, and reinsert the subplan index into the heap,
     254             :          * because it might now compare differently against the existing
     255             :          * elements of the heap.  (We could perhaps simplify the logic a bit
     256             :          * by doing this before returning from the prior call, but it's better
     257             :          * to not pull tuples until necessary.)
     258             :          */
     259        8814 :         i = DatumGetInt32(binaryheap_first(node->ms_heap));
     260        8814 :         node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
     261        8814 :         if (!TupIsNull(node->ms_slots[i]))
     262        8628 :             binaryheap_replace_first(node->ms_heap, Int32GetDatum(i));
     263             :         else
     264         186 :             (void) binaryheap_remove_first(node->ms_heap);
     265             :     }
     266             : 
     267        8944 :     if (binaryheap_empty(node->ms_heap))
     268             :     {
     269             :         /* All the subplans are exhausted, and so is the heap */
     270          94 :         result = ExecClearTuple(node->ps.ps_ResultTupleSlot);
     271             :     }
     272             :     else
     273             :     {
     274        8850 :         i = DatumGetInt32(binaryheap_first(node->ms_heap));
     275        8850 :         result = node->ms_slots[i];
     276             :     }
     277             : 
     278        8944 :     return result;
     279             : }
     280             : 
     281             : /*
     282             :  * Compare the tuples in the two given slots.
     283             :  */
     284             : static int32
     285        9192 : heap_compare_slots(Datum a, Datum b, void *arg)
     286             : {
     287        9192 :     MergeAppendState *node = (MergeAppendState *) arg;
     288        9192 :     SlotNumber  slot1 = DatumGetInt32(a);
     289        9192 :     SlotNumber  slot2 = DatumGetInt32(b);
     290             : 
     291        9192 :     TupleTableSlot *s1 = node->ms_slots[slot1];
     292        9192 :     TupleTableSlot *s2 = node->ms_slots[slot2];
     293             :     int         nkey;
     294             : 
     295             :     Assert(!TupIsNull(s1));
     296             :     Assert(!TupIsNull(s2));
     297             : 
     298        9532 :     for (nkey = 0; nkey < node->ms_nkeys; nkey++)
     299             :     {
     300        9192 :         SortSupport sortKey = node->ms_sortkeys + nkey;
     301        9192 :         AttrNumber  attno = sortKey->ssup_attno;
     302             :         Datum       datum1,
     303             :                     datum2;
     304             :         bool        isNull1,
     305             :                     isNull2;
     306             :         int         compare;
     307             : 
     308        9192 :         datum1 = slot_getattr(s1, attno, &isNull1);
     309        9192 :         datum2 = slot_getattr(s2, attno, &isNull2);
     310             : 
     311        9192 :         compare = ApplySortComparator(datum1, isNull1,
     312             :                                       datum2, isNull2,
     313             :                                       sortKey);
     314        9192 :         if (compare != 0)
     315             :         {
     316        8852 :             INVERT_COMPARE_RESULT(compare);
     317        8852 :             return compare;
     318             :         }
     319             :     }
     320         340 :     return 0;
     321             : }
     322             : 
     323             : /* ----------------------------------------------------------------
     324             :  *      ExecEndMergeAppend
     325             :  *
     326             :  *      Shuts down the subscans of the MergeAppend node.
     327             :  *
     328             :  *      Returns nothing of interest.
     329             :  * ----------------------------------------------------------------
     330             :  */
     331             : void
     332         262 : ExecEndMergeAppend(MergeAppendState *node)
     333             : {
     334             :     PlanState **mergeplans;
     335             :     int         nplans;
     336             :     int         i;
     337             : 
     338             :     /*
     339             :      * get information from the node
     340             :      */
     341         262 :     mergeplans = node->mergeplans;
     342         262 :     nplans = node->ms_nplans;
     343             : 
     344             :     /*
     345             :      * shut down each of the subscans
     346             :      */
     347        1010 :     for (i = 0; i < nplans; i++)
     348         748 :         ExecEndNode(mergeplans[i]);
     349         262 : }
     350             : 
     351             : void
     352          12 : ExecReScanMergeAppend(MergeAppendState *node)
     353             : {
     354             :     int         i;
     355             : 
     356             :     /*
     357             :      * If any PARAM_EXEC Params used in pruning expressions have changed, then
     358             :      * we'd better unset the valid subplans so that they are reselected for
     359             :      * the new parameter values.
     360             :      */
     361          12 :     if (node->ms_prune_state &&
     362           0 :         bms_overlap(node->ps.chgParam,
     363           0 :                     node->ms_prune_state->execparamids))
     364             :     {
     365           0 :         bms_free(node->ms_valid_subplans);
     366           0 :         node->ms_valid_subplans = NULL;
     367             :     }
     368             : 
     369          36 :     for (i = 0; i < node->ms_nplans; i++)
     370             :     {
     371          24 :         PlanState  *subnode = node->mergeplans[i];
     372             : 
     373             :         /*
     374             :          * ExecReScan doesn't know about my subplans, so I have to do
     375             :          * changed-parameter signaling myself.
     376             :          */
     377          24 :         if (node->ps.chgParam != NULL)
     378          24 :             UpdateChangedParamSet(subnode, node->ps.chgParam);
     379             : 
     380             :         /*
     381             :          * If chgParam of subnode is not null then plan will be re-scanned by
     382             :          * first ExecProcNode.
     383             :          */
     384          24 :         if (subnode->chgParam == NULL)
     385           0 :             ExecReScan(subnode);
     386             :     }
     387          12 :     binaryheap_reset(node->ms_heap);
     388          12 :     node->ms_initialized = false;
     389          12 : }

Generated by: LCOV version 1.13