granicus.if.org Git - postgresql/blob - src/backend/executor/nodeMergeAppend.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * nodeMergeAppend.c
   4  *        routines to handle MergeAppend nodes.
   5  *
   6  * Portions Copyright (c) 1996-2013, 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/nodeMergeAppend.h"
  43
  44 #include "lib/binaryheap.h"
  45
  46 /*
  47  * We have one slot for each item in the heap array.  We use SlotNumber
  48  * to store slot indexes.  This doesn't actually provide any formal
  49  * type-safety, but it makes the code more self-documenting.
  50  */
  51 typedef int32 SlotNumber;
  52
  53 static int      heap_compare_slots(Datum a, Datum b, void *arg);
  54
  55
  56 /* ----------------------------------------------------------------
  57  *              ExecInitMergeAppend
  58  *
  59  *              Begin all of the subscans of the MergeAppend node.
  60  * ----------------------------------------------------------------
  61  */
  62 MergeAppendState *
  63 ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
  64 {
  65         MergeAppendState *mergestate = makeNode(MergeAppendState);
  66         PlanState **mergeplanstates;
  67         int                     nplans;
  68         int                     i;
  69         ListCell   *lc;
  70
  71         /* check for unsupported flags */
  72         Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
  73
  74         /*
  75          * Set up empty vector of subplan states
  76          */
  77         nplans = list_length(node->mergeplans);
  78
  79         mergeplanstates = (PlanState **) palloc0(nplans * sizeof(PlanState *));
  80
  81         /*
  82          * create new MergeAppendState for our node
  83          */
  84         mergestate->ps.plan = (Plan *) node;
  85         mergestate->ps.state = estate;
  86         mergestate->mergeplans = mergeplanstates;
  87         mergestate->ms_nplans = nplans;
  88
  89         mergestate->ms_slots = (TupleTableSlot **) palloc0(sizeof(TupleTableSlot *) * nplans);
  90         mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots,
  91                                                                                           mergestate);
  92
  93         /*
  94          * Miscellaneous initialization
  95          *
  96          * MergeAppend plans don't have expression contexts because they never
  97          * call ExecQual or ExecProject.
  98          */
  99
 100         /*
 101          * MergeAppend nodes do have Result slots, which hold pointers to tuples,
 102          * so we have to initialize them.
 103          */
 104         ExecInitResultTupleSlot(estate, &mergestate->ps);
 105
 106         /*
 107          * call ExecInitNode on each of the plans to be executed and save the
 108          * results into the array "mergeplans".
 109          */
 110         i = 0;
 111         foreach(lc, node->mergeplans)
 112         {
 113                 Plan       *initNode = (Plan *) lfirst(lc);
 114
 115                 mergeplanstates[i] = ExecInitNode(initNode, estate, eflags);
 116                 i++;
 117         }
 118
 119         /*
 120          * initialize output tuple type
 121          */
 122         ExecAssignResultTypeFromTL(&mergestate->ps);
 123         mergestate->ps.ps_ProjInfo = NULL;
 124
 125         /*
 126          * initialize sort-key information
 127          */
 128         mergestate->ms_nkeys = node->numCols;
 129         mergestate->ms_sortkeys = palloc0(sizeof(SortSupportData) * node->numCols);
 130
 131         for (i = 0; i < node->numCols; i++)
 132         {
 133                 SortSupport sortKey = mergestate->ms_sortkeys + i;
 134
 135                 sortKey->ssup_cxt = CurrentMemoryContext;
 136                 sortKey->ssup_collation = node->collations[i];
 137                 sortKey->ssup_nulls_first = node->nullsFirst[i];
 138                 sortKey->ssup_attno = node->sortColIdx[i];
 139
 140                 PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
 141         }
 142
 143         /*
 144          * initialize to show we have not run the subplans yet
 145          */
 146         mergestate->ms_initialized = false;
 147
 148         return mergestate;
 149 }
 150
 151 /* ----------------------------------------------------------------
 152  *         ExecMergeAppend
 153  *
 154  *              Handles iteration over multiple subplans.
 155  * ----------------------------------------------------------------
 156  */
 157 TupleTableSlot *
 158 ExecMergeAppend(MergeAppendState *node)
 159 {
 160         TupleTableSlot *result;
 161         SlotNumber      i;
 162
 163         if (!node->ms_initialized)
 164         {
 165                 /*
 166                  * First time through: pull the first tuple from each subplan, and set
 167                  * up the heap.
 168                  */
 169                 for (i = 0; i < node->ms_nplans; i++)
 170                 {
 171                         node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
 172                         if (!TupIsNull(node->ms_slots[i]))
 173                                 binaryheap_add_unordered(node->ms_heap, Int32GetDatum(i));
 174                 }
 175                 binaryheap_build(node->ms_heap);
 176                 node->ms_initialized = true;
 177         }
 178         else
 179         {
 180                 /*
 181                  * Otherwise, pull the next tuple from whichever subplan we returned
 182                  * from last time, and reinsert the subplan index into the heap,
 183                  * because it might now compare differently against the existing
 184                  * elements of the heap.  (We could perhaps simplify the logic a bit
 185                  * by doing this before returning from the prior call, but it's better
 186                  * to not pull tuples until necessary.)
 187                  */
 188                 i = DatumGetInt32(binaryheap_first(node->ms_heap));
 189                 node->ms_slots[i] = ExecProcNode(node->mergeplans[i]);
 190                 if (!TupIsNull(node->ms_slots[i]))
 191                         binaryheap_replace_first(node->ms_heap, Int32GetDatum(i));
 192                 else
 193                         (void) binaryheap_remove_first(node->ms_heap);
 194         }
 195
 196         if (binaryheap_empty(node->ms_heap))
 197         {
 198                 /* All the subplans are exhausted, and so is the heap */
 199                 result = ExecClearTuple(node->ps.ps_ResultTupleSlot);
 200         }
 201         else
 202         {
 203                 i = DatumGetInt32(binaryheap_first(node->ms_heap));
 204                 result = node->ms_slots[i];
 205         }
 206
 207         return result;
 208 }
 209
 210 /*
 211  * Compare the tuples in the two given slots.
 212  */
 213 static int32
 214 heap_compare_slots(Datum a, Datum b, void *arg)
 215 {
 216         MergeAppendState *node = (MergeAppendState *) arg;
 217         SlotNumber      slot1 = DatumGetInt32(a);
 218         SlotNumber      slot2 = DatumGetInt32(b);
 219
 220         TupleTableSlot *s1 = node->ms_slots[slot1];
 221         TupleTableSlot *s2 = node->ms_slots[slot2];
 222         int                     nkey;
 223
 224         Assert(!TupIsNull(s1));
 225         Assert(!TupIsNull(s2));
 226
 227         for (nkey = 0; nkey < node->ms_nkeys; nkey++)
 228         {
 229                 SortSupport sortKey = node->ms_sortkeys + nkey;
 230                 AttrNumber      attno = sortKey->ssup_attno;
 231                 Datum           datum1,
 232                                         datum2;
 233                 bool            isNull1,
 234                                         isNull2;
 235                 int                     compare;
 236
 237                 datum1 = slot_getattr(s1, attno, &isNull1);
 238                 datum2 = slot_getattr(s2, attno, &isNull2);
 239
 240                 compare = ApplySortComparator(datum1, isNull1,
 241                                                                           datum2, isNull2,
 242                                                                           sortKey);
 243                 if (compare != 0)
 244                         return -compare;
 245         }
 246         return 0;
 247 }
 248
 249 /* ----------------------------------------------------------------
 250  *              ExecEndMergeAppend
 251  *
 252  *              Shuts down the subscans of the MergeAppend node.
 253  *
 254  *              Returns nothing of interest.
 255  * ----------------------------------------------------------------
 256  */
 257 void
 258 ExecEndMergeAppend(MergeAppendState *node)
 259 {
 260         PlanState **mergeplans;
 261         int                     nplans;
 262         int                     i;
 263
 264         /*
 265          * get information from the node
 266          */
 267         mergeplans = node->mergeplans;
 268         nplans = node->ms_nplans;
 269
 270         /*
 271          * shut down each of the subscans
 272          */
 273         for (i = 0; i < nplans; i++)
 274                 ExecEndNode(mergeplans[i]);
 275 }
 276
 277 void
 278 ExecReScanMergeAppend(MergeAppendState *node)
 279 {
 280         int                     i;
 281
 282         for (i = 0; i < node->ms_nplans; i++)
 283         {
 284                 PlanState  *subnode = node->mergeplans[i];
 285
 286                 /*
 287                  * ExecReScan doesn't know about my subplans, so I have to do
 288                  * changed-parameter signaling myself.
 289                  */
 290                 if (node->ps.chgParam != NULL)
 291                         UpdateChangedParamSet(subnode, node->ps.chgParam);
 292
 293                 /*
 294                  * If chgParam of subnode is not null then plan will be re-scanned by
 295                  * first ExecProcNode.
 296                  */
 297                 if (subnode->chgParam == NULL)
 298                         ExecReScan(subnode);
 299         }
 300         node->ms_initialized = false;
 301 }