src/backend/executor/nodeRecursiveunion.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * nodeRecursiveunion.c
   4  *        routines to handle RecursiveUnion nodes.
   5  *
   6  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL$
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "executor/execdebug.h"
  18 #include "executor/nodeRecursiveunion.h"
  19 #include "miscadmin.h"
  20 #include "utils/memutils.h"
  21
  22
  23 /*
  24  * To implement UNION (without ALL), we need a hashtable that stores tuples
  25  * already seen.  The hash key is computed from the grouping columns.
  26  */
  27 typedef struct RUHashEntryData *RUHashEntry;
  28
  29 typedef struct RUHashEntryData
  30 {
  31         TupleHashEntryData shared;      /* common header for hash table entries */
  32 } RUHashEntryData;
  33
  34
  35 /*
  36  * Initialize the hash table to empty.
  37  */
  38 static void
  39 build_hash_table(RecursiveUnionState *rustate)
  40 {
  41         RecursiveUnion     *node = (RecursiveUnion *) rustate->ps.plan;
  42
  43         Assert(node->numCols > 0);
  44         Assert(node->numGroups > 0);
  45
  46         rustate->hashtable = BuildTupleHashTable(node->numCols,
  47                                                                                          node->dupColIdx,
  48                                                                                          rustate->eqfunctions,
  49                                                                                          rustate->hashfunctions,
  50                                                                                          node->numGroups,
  51                                                                                          sizeof(RUHashEntryData),
  52                                                                                          rustate->tableContext,
  53                                                                                          rustate->tempContext);
  54 }
  55
  56
  57 /* ----------------------------------------------------------------
  58  *              ExecRecursiveUnion(node)
  59  *
  60  *              Scans the recursive query sequentially and returns the next
  61  *      qualifying tuple.
  62  *
  63  * 1. evaluate non recursive term and assign the result to RT
  64  *
  65  * 2. execute recursive terms
  66  *
  67  * 2.1 WT := RT
  68  * 2.2 while WT is not empty repeat 2.3 to 2.6. if WT is empty returns RT
  69  * 2.3 replace the name of recursive term with WT
  70  * 2.4 evaluate the recursive term and store into WT
  71  * 2.5 append WT to RT
  72  * 2.6 go back to 2.2
  73  * ----------------------------------------------------------------
  74  */
  75 TupleTableSlot *
  76 ExecRecursiveUnion(RecursiveUnionState *node)
  77 {
  78         PlanState  *outerPlan = outerPlanState(node);
  79         PlanState  *innerPlan = innerPlanState(node);
  80         RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;
  81         TupleTableSlot *slot;
  82         RUHashEntry entry;
  83         bool            isnew;
  84
  85         /* 1. Evaluate non-recursive term */
  86         if (!node->recursing)
  87         {
  88                 for (;;)
  89                 {
  90                         slot = ExecProcNode(outerPlan);
  91                         if (TupIsNull(slot))
  92                                 break;
  93                         if (plan->numCols > 0)
  94                         {
  95                                 /* Find or build hashtable entry for this tuple's group */
  96                                 entry = (RUHashEntry)
  97                                         LookupTupleHashEntry(node->hashtable, slot, &isnew);
  98                                 /* Must reset temp context after each hashtable lookup */
  99                                 MemoryContextReset(node->tempContext);
 100                                 /* Ignore tuple if already seen */
 101                                 if (!isnew)
 102                                         continue;
 103                         }
 104                         /* Each non-duplicate tuple goes to the working table ... */
 105                         tuplestore_puttupleslot(node->working_table, slot);
 106                         /* ... and to the caller */
 107                         return slot;
 108                 }
 109                 node->recursing = true;
 110         }
 111
 112         /* 2. Execute recursive term */
 113         for (;;)
 114         {
 115                 slot = ExecProcNode(innerPlan);
 116                 if (TupIsNull(slot))
 117                 {
 118                         /* Done if there's nothing in the intermediate table */
 119                         if (node->intermediate_empty)
 120                                 break;
 121
 122                         /* done with old working table ... */
 123                         tuplestore_end(node->working_table);
 124
 125                         /* intermediate table becomes working table */
 126                         node->working_table = node->intermediate_table;
 127
 128                         /* create new empty intermediate table */
 129                         node->intermediate_table = tuplestore_begin_heap(false, false,
 130                                                                                                                          work_mem);
 131                         node->intermediate_empty = true;
 132
 133                         /* reset the recursive term */
 134                         innerPlan->chgParam = bms_add_member(innerPlan->chgParam,
 135                                                                                                  plan->wtParam);
 136
 137                         /* and continue fetching from recursive term */
 138                         continue;
 139                 }
 140
 141                 if (plan->numCols > 0)
 142                 {
 143                         /* Find or build hashtable entry for this tuple's group */
 144                         entry = (RUHashEntry)
 145                                 LookupTupleHashEntry(node->hashtable, slot, &isnew);
 146                         /* Must reset temp context after each hashtable lookup */
 147                         MemoryContextReset(node->tempContext);
 148                         /* Ignore tuple if already seen */
 149                         if (!isnew)
 150                                 continue;
 151                 }
 152
 153                 /* Else, tuple is good; stash it in intermediate table ... */
 154                 node->intermediate_empty = false;
 155                 tuplestore_puttupleslot(node->intermediate_table, slot);
 156                 /* ... and return it */
 157                 return slot;
 158         }
 159
 160         return NULL;
 161 }
 162
 163 /* ----------------------------------------------------------------
 164  *              ExecInitRecursiveUnionScan
 165  * ----------------------------------------------------------------
 166  */
 167 RecursiveUnionState *
 168 ExecInitRecursiveUnion(RecursiveUnion *node, EState *estate, int eflags)
 169 {
 170         RecursiveUnionState *rustate;
 171         ParamExecData *prmdata;
 172
 173         /* check for unsupported flags */
 174         Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
 175
 176         /*
 177          * create state structure
 178          */
 179         rustate = makeNode(RecursiveUnionState);
 180         rustate->ps.plan = (Plan *) node;
 181         rustate->ps.state = estate;
 182
 183         rustate->eqfunctions = NULL;
 184         rustate->hashfunctions = NULL;
 185         rustate->hashtable = NULL;
 186         rustate->tempContext = NULL;
 187         rustate->tableContext = NULL;
 188
 189         /* initialize processing state */
 190         rustate->recursing = false;
 191         rustate->intermediate_empty = true;
 192         rustate->working_table = tuplestore_begin_heap(false, false, work_mem);
 193         rustate->intermediate_table = tuplestore_begin_heap(false, false, work_mem);
 194
 195         /*
 196          * If hashing, we need a per-tuple memory context for comparisons, and a
 197          * longer-lived context to store the hash table.  The table can't just be
 198          * kept in the per-query context because we want to be able to throw it
 199          * away when rescanning.
 200          */
 201         if (node->numCols > 0)
 202         {
 203                 rustate->tempContext =
 204                         AllocSetContextCreate(CurrentMemoryContext,
 205                                                                   "RecursiveUnion",
 206                                                                   ALLOCSET_DEFAULT_MINSIZE,
 207                                                                   ALLOCSET_DEFAULT_INITSIZE,
 208                                                                   ALLOCSET_DEFAULT_MAXSIZE);
 209                 rustate->tableContext =
 210                         AllocSetContextCreate(CurrentMemoryContext,
 211                                                                   "RecursiveUnion hash table",
 212                                                                   ALLOCSET_DEFAULT_MINSIZE,
 213                                                                   ALLOCSET_DEFAULT_INITSIZE,
 214                                                                   ALLOCSET_DEFAULT_MAXSIZE);
 215         }
 216
 217         /*
 218          * Make the state structure available to descendant WorkTableScan nodes
 219          * via the Param slot reserved for it.
 220          */
 221         prmdata = &(estate->es_param_exec_vals[node->wtParam]);
 222         Assert(prmdata->execPlan == NULL);
 223         prmdata->value = PointerGetDatum(rustate);
 224         prmdata->isnull = false;
 225
 226         /*
 227          * Miscellaneous initialization
 228          *
 229          * RecursiveUnion plans don't have expression contexts because they never
 230          * call ExecQual or ExecProject.
 231          */
 232         Assert(node->plan.qual == NIL);
 233
 234 #define RECURSIVEUNION_NSLOTS 1
 235
 236         /*
 237          * RecursiveUnion nodes still have Result slots, which hold pointers to
 238          * tuples, so we have to initialize them.
 239          */
 240         ExecInitResultTupleSlot(estate, &rustate->ps);
 241
 242         /*
 243          * Initialize result tuple type and projection info.  (Note: we have
 244          * to set up the result type before initializing child nodes, because
 245          * nodeWorktablescan.c expects it to be valid.)
 246          */
 247         ExecAssignResultTypeFromTL(&rustate->ps);
 248         rustate->ps.ps_ProjInfo = NULL;
 249
 250         /*
 251          * initialize child nodes
 252          */
 253         outerPlanState(rustate) = ExecInitNode(outerPlan(node), estate, eflags);
 254         innerPlanState(rustate) = ExecInitNode(innerPlan(node), estate, eflags);
 255
 256         /*
 257          * If hashing, precompute fmgr lookup data for inner loop, and create
 258          * the hash table.
 259          */
 260         if (node->numCols > 0)
 261         {
 262                 execTuplesHashPrepare(node->numCols,
 263                                                           node->dupOperators,
 264                                                           &rustate->eqfunctions,
 265                                                           &rustate->hashfunctions);
 266                 build_hash_table(rustate);
 267         }
 268
 269         return rustate;
 270 }
 271
 272 int
 273 ExecCountSlotsRecursiveUnion(RecursiveUnion *node)
 274 {
 275         return ExecCountSlotsNode(outerPlan(node)) +
 276                 ExecCountSlotsNode(innerPlan(node)) +
 277                 RECURSIVEUNION_NSLOTS;
 278 }
 279
 280 /* ----------------------------------------------------------------
 281  *              ExecEndRecursiveUnionScan
 282  *
 283  *              frees any storage allocated through C routines.
 284  * ----------------------------------------------------------------
 285  */
 286 void
 287 ExecEndRecursiveUnion(RecursiveUnionState *node)
 288 {
 289         /* Release tuplestores */
 290         tuplestore_end(node->working_table);
 291         tuplestore_end(node->intermediate_table);
 292
 293         /* free subsidiary stuff including hashtable */
 294         if (node->tempContext)
 295                 MemoryContextDelete(node->tempContext);
 296         if (node->tableContext)
 297                 MemoryContextDelete(node->tableContext);
 298
 299         /*
 300          * clean out the upper tuple table
 301          */
 302         ExecClearTuple(node->ps.ps_ResultTupleSlot);
 303
 304         /*
 305          * close down subplans
 306          */
 307         ExecEndNode(outerPlanState(node));
 308         ExecEndNode(innerPlanState(node));
 309 }
 310
 311 /* ----------------------------------------------------------------
 312  *              ExecRecursiveUnionReScan
 313  *
 314  *              Rescans the relation.
 315  * ----------------------------------------------------------------
 316  */
 317 void
 318 ExecRecursiveUnionReScan(RecursiveUnionState *node, ExprContext *exprCtxt)
 319 {
 320         PlanState  *outerPlan = outerPlanState(node);
 321         PlanState  *innerPlan = innerPlanState(node);
 322         RecursiveUnion *plan = (RecursiveUnion *) node->ps.plan;
 323
 324         /*
 325          * Set recursive term's chgParam to tell it that we'll modify the
 326          * working table and therefore it has to rescan.
 327          */
 328         innerPlan->chgParam = bms_add_member(innerPlan->chgParam, plan->wtParam);
 329
 330         /*
 331          * if chgParam of subnode is not null then plan will be re-scanned by
 332          * first ExecProcNode.  Because of above, we only have to do this to
 333          * the non-recursive term.
 334          */
 335         if (outerPlan->chgParam == NULL)
 336                 ExecReScan(outerPlan, exprCtxt);
 337
 338         /* Release any hashtable storage */
 339         if (node->tableContext)
 340                 MemoryContextResetAndDeleteChildren(node->tableContext);
 341
 342         /* And rebuild empty hashtable if needed */
 343         if (plan->numCols > 0)
 344                 build_hash_table(node);
 345
 346         /* reset processing state */
 347         node->recursing = false;
 348         node->intermediate_empty = true;
 349         tuplestore_clear(node->working_table);
 350         tuplestore_clear(node->intermediate_table);
 351 }