PostgreSQL Source Code  git master
nodeAppend.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * nodeAppend.c
4  * routines to handle append nodes.
5  *
6  * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
7  * Portions Copyright (c) 1994, Regents of the University of California
8  *
9  *
10  * IDENTIFICATION
11  * src/backend/executor/nodeAppend.c
12  *
13  *-------------------------------------------------------------------------
14  */
15 /* INTERFACE ROUTINES
16  * ExecInitAppend - initialize the append node
17  * ExecAppend - retrieve the next tuple from the node
18  * ExecEndAppend - shut down the append node
19  * ExecReScanAppend - rescan the append node
20  *
21  * NOTES
22  * Each append node contains a list of one or more subplans which
23  * must be iteratively processed (forwards or backwards).
24  * Tuples are retrieved by executing the 'whichplan'th subplan
25  * until the subplan stops returning tuples, at which point that
26  * plan is shut down and the next started up.
27  *
28  * Append nodes don't make use of their left and right
29  * subtrees, rather they maintain a list of subplans so
30  * a typical append node looks like this in the plan tree:
31  *
32  * ...
33  * /
34  * Append -------+------+------+--- nil
35  * / \ | | |
36  * nil nil ... ... ...
37  * subplans
38  *
39  * Append nodes are currently used for unions, and to support
40  * inheritance queries, where several relations need to be scanned.
41  * For example, in our standard person/student/employee/student-emp
42  * example, where student and employee inherit from person
43  * and student-emp inherits from student and employee, the
44  * query:
45  *
46  * select name from person
47  *
48  * generates the plan:
49  *
50  * |
51  * Append -------+-------+--------+--------+
52  * / \ | | | |
53  * nil nil Scan Scan Scan Scan
54  * | | | |
55  * person employee student student-emp
56  */
57 
58 #include "postgres.h"
59 
60 #include "executor/execdebug.h"
61 #include "executor/nodeAppend.h"
62 #include "miscadmin.h"
63 
64 /* Shared state for parallel-aware Append. */
66 {
67  LWLock pa_lock; /* mutual exclusion to choose next subplan */
68  int pa_next_plan; /* next plan to choose by any worker */
69 
70  /*
71  * pa_finished[i] should be true if no more workers should select subplan
72  * i. for a non-partial plan, this should be set to true as soon as a
73  * worker selects the plan; for a partial plan, it remains false until
74  * some worker executes the plan to completion.
75  */
76  bool pa_finished[FLEXIBLE_ARRAY_MEMBER];
77 };
78 
79 #define INVALID_SUBPLAN_INDEX -1
80 
81 static TupleTableSlot *ExecAppend(PlanState *pstate);
82 static bool choose_next_subplan_locally(AppendState *node);
85 
86 /* ----------------------------------------------------------------
87  * ExecInitAppend
88  *
89  * Begin all of the subscans of the append node.
90  *
91  * (This is potentially wasteful, since the entire result of the
92  * append node may not be scanned, but this way all of the
93  * structures get allocated in the executor's top level memory
94  * block instead of that of the call to ExecAppend.)
95  * ----------------------------------------------------------------
96  */
98 ExecInitAppend(Append *node, EState *estate, int eflags)
99 {
100  AppendState *appendstate = makeNode(AppendState);
101  PlanState **appendplanstates;
102  int nplans;
103  int i;
104  ListCell *lc;
105 
106  /* check for unsupported flags */
107  Assert(!(eflags & EXEC_FLAG_MARK));
108 
109  /*
110  * Lock the non-leaf tables in the partition tree controlled by this node.
111  * It's a no-op for non-partitioned parent tables.
112  */
114 
115  /*
116  * Set up empty vector of subplan states
117  */
118  nplans = list_length(node->appendplans);
119 
120  appendplanstates = (PlanState **) palloc0(nplans * sizeof(PlanState *));
121 
122  /*
123  * create new AppendState for our append node
124  */
125  appendstate->ps.plan = (Plan *) node;
126  appendstate->ps.state = estate;
127  appendstate->ps.ExecProcNode = ExecAppend;
128  appendstate->appendplans = appendplanstates;
129  appendstate->as_nplans = nplans;
130 
131  /*
132  * Miscellaneous initialization
133  *
134  * Append plans don't have expression contexts because they never call
135  * ExecQual or ExecProject.
136  */
137 
138  /*
139  * append nodes still have Result slots, which hold pointers to tuples, so
140  * we have to initialize them.
141  */
142  ExecInitResultTupleSlot(estate, &appendstate->ps);
143 
144  /*
145  * call ExecInitNode on each of the plans to be executed and save the
146  * results into the array "appendplans".
147  */
148  i = 0;
149  foreach(lc, node->appendplans)
150  {
151  Plan *initNode = (Plan *) lfirst(lc);
152 
153  appendplanstates[i] = ExecInitNode(initNode, estate, eflags);
154  i++;
155  }
156 
157  /*
158  * initialize output tuple type
159  */
160  ExecAssignResultTypeFromTL(&appendstate->ps);
161  appendstate->ps.ps_ProjInfo = NULL;
162 
163  /*
164  * Parallel-aware append plans must choose the first subplan to execute by
165  * looking at shared memory, but non-parallel-aware append plans can
166  * always start with the first subplan.
167  */
168  appendstate->as_whichplan =
169  appendstate->ps.plan->parallel_aware ? INVALID_SUBPLAN_INDEX : 0;
170 
171  /* If parallel-aware, this will be overridden later. */
173 
174  return appendstate;
175 }
176 
177 /* ----------------------------------------------------------------
178  * ExecAppend
179  *
180  * Handles iteration over multiple subplans.
181  * ----------------------------------------------------------------
182  */
183 static TupleTableSlot *
185 {
186  AppendState *node = castNode(AppendState, pstate);
187 
188  /* If no subplan has been chosen, we must choose one before proceeding. */
189  if (node->as_whichplan == INVALID_SUBPLAN_INDEX &&
190  !node->choose_next_subplan(node))
191  return ExecClearTuple(node->ps.ps_ResultTupleSlot);
192 
193  for (;;)
194  {
195  PlanState *subnode;
196  TupleTableSlot *result;
197 
199 
200  /*
201  * figure out which subplan we are currently processing
202  */
203  Assert(node->as_whichplan >= 0 && node->as_whichplan < node->as_nplans);
204  subnode = node->appendplans[node->as_whichplan];
205 
206  /*
207  * get a tuple from the subplan
208  */
209  result = ExecProcNode(subnode);
210 
211  if (!TupIsNull(result))
212  {
213  /*
214  * If the subplan gave us something then return it as-is. We do
215  * NOT make use of the result slot that was set up in
216  * ExecInitAppend; there's no need for it.
217  */
218  return result;
219  }
220 
221  /* choose new subplan; if none, we're done */
222  if (!node->choose_next_subplan(node))
223  return ExecClearTuple(node->ps.ps_ResultTupleSlot);
224  }
225 }
226 
227 /* ----------------------------------------------------------------
228  * ExecEndAppend
229  *
230  * Shuts down the subscans of the append node.
231  *
232  * Returns nothing of interest.
233  * ----------------------------------------------------------------
234  */
235 void
237 {
238  PlanState **appendplans;
239  int nplans;
240  int i;
241 
242  /*
243  * get information from the node
244  */
245  appendplans = node->appendplans;
246  nplans = node->as_nplans;
247 
248  /*
249  * shut down each of the subscans
250  */
251  for (i = 0; i < nplans; i++)
252  ExecEndNode(appendplans[i]);
253 }
254 
255 void
257 {
258  int i;
259 
260  for (i = 0; i < node->as_nplans; i++)
261  {
262  PlanState *subnode = node->appendplans[i];
263 
264  /*
265  * ExecReScan doesn't know about my subplans, so I have to do
266  * changed-parameter signaling myself.
267  */
268  if (node->ps.chgParam != NULL)
269  UpdateChangedParamSet(subnode, node->ps.chgParam);
270 
271  /*
272  * If chgParam of subnode is not null then plan will be re-scanned by
273  * first ExecProcNode.
274  */
275  if (subnode->chgParam == NULL)
276  ExecReScan(subnode);
277  }
278 
279  node->as_whichplan =
281 }
282 
283 /* ----------------------------------------------------------------
284  * Parallel Append Support
285  * ----------------------------------------------------------------
286  */
287 
288 /* ----------------------------------------------------------------
289  * ExecAppendEstimate
290  *
291  * Compute the amount of space we'll need in the parallel
292  * query DSM, and inform pcxt->estimator about our needs.
293  * ----------------------------------------------------------------
294  */
295 void
297  ParallelContext *pcxt)
298 {
299  node->pstate_len =
301  sizeof(bool) * node->as_nplans);
302 
304  shm_toc_estimate_keys(&pcxt->estimator, 1);
305 }
306 
307 
308 /* ----------------------------------------------------------------
309  * ExecAppendInitializeDSM
310  *
311  * Set up shared state for Parallel Append.
312  * ----------------------------------------------------------------
313  */
314 void
316  ParallelContext *pcxt)
317 {
318  ParallelAppendState *pstate;
319 
320  pstate = shm_toc_allocate(pcxt->toc, node->pstate_len);
321  memset(pstate, 0, node->pstate_len);
323  shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id, pstate);
324 
325  node->as_pstate = pstate;
327 }
328 
329 /* ----------------------------------------------------------------
330  * ExecAppendReInitializeDSM
331  *
332  * Reset shared state before beginning a fresh scan.
333  * ----------------------------------------------------------------
334  */
335 void
337 {
338  ParallelAppendState *pstate = node->as_pstate;
339 
340  pstate->pa_next_plan = 0;
341  memset(pstate->pa_finished, 0, sizeof(bool) * node->as_nplans);
342 }
343 
344 /* ----------------------------------------------------------------
345  * ExecAppendInitializeWorker
346  *
347  * Copy relevant information from TOC into planstate, and initialize
348  * whatever is required to choose and execute the optimal subplan.
349  * ----------------------------------------------------------------
350  */
351 void
353 {
354  node->as_pstate = shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
356 }
357 
358 /* ----------------------------------------------------------------
359  * choose_next_subplan_locally
360  *
361  * Choose next subplan for a non-parallel-aware Append,
362  * returning false if there are no more.
363  * ----------------------------------------------------------------
364  */
365 static bool
367 {
368  int whichplan = node->as_whichplan;
369 
370  /* We should never see INVALID_SUBPLAN_INDEX in this case. */
371  Assert(whichplan >= 0 && whichplan <= node->as_nplans);
372 
374  {
375  if (whichplan >= node->as_nplans - 1)
376  return false;
377  node->as_whichplan++;
378  }
379  else
380  {
381  if (whichplan <= 0)
382  return false;
383  node->as_whichplan--;
384  }
385 
386  return true;
387 }
388 
389 /* ----------------------------------------------------------------
390  * choose_next_subplan_for_leader
391  *
392  * Try to pick a plan which doesn't commit us to doing much
393  * work locally, so that as much work as possible is done in
394  * the workers. Cheapest subplans are at the end.
395  * ----------------------------------------------------------------
396  */
397 static bool
399 {
400  ParallelAppendState *pstate = node->as_pstate;
401  Append *append = (Append *) node->ps.plan;
402 
403  /* Backward scan is not supported by parallel-aware plans */
405 
407 
408  if (node->as_whichplan != INVALID_SUBPLAN_INDEX)
409  {
410  /* Mark just-completed subplan as finished. */
411  node->as_pstate->pa_finished[node->as_whichplan] = true;
412  }
413  else
414  {
415  /* Start with last subplan. */
416  node->as_whichplan = node->as_nplans - 1;
417  }
418 
419  /* Loop until we find a subplan to execute. */
420  while (pstate->pa_finished[node->as_whichplan])
421  {
422  if (node->as_whichplan == 0)
423  {
426  LWLockRelease(&pstate->pa_lock);
427  return false;
428  }
429  node->as_whichplan--;
430  }
431 
432  /* If non-partial, immediately mark as finished. */
433  if (node->as_whichplan < append->first_partial_plan)
434  node->as_pstate->pa_finished[node->as_whichplan] = true;
435 
436  LWLockRelease(&pstate->pa_lock);
437 
438  return true;
439 }
440 
441 /* ----------------------------------------------------------------
442  * choose_next_subplan_for_worker
443  *
444  * Choose next subplan for a parallel-aware Append, returning
445  * false if there are no more.
446  *
447  * We start from the first plan and advance through the list;
448  * when we get back to the end, we loop back to the first
449  * nonpartial plan. This assigns the non-partial plans first
450  * in order of descending cost and then spreads out the
451  * workers as evenly as possible across the remaining partial
452  * plans.
453  * ----------------------------------------------------------------
454  */
455 static bool
457 {
458  ParallelAppendState *pstate = node->as_pstate;
459  Append *append = (Append *) node->ps.plan;
460 
461  /* Backward scan is not supported by parallel-aware plans */
463 
465 
466  /* Mark just-completed subplan as finished. */
467  if (node->as_whichplan != INVALID_SUBPLAN_INDEX)
468  node->as_pstate->pa_finished[node->as_whichplan] = true;
469 
470  /* If all the plans are already done, we have nothing to do */
471  if (pstate->pa_next_plan == INVALID_SUBPLAN_INDEX)
472  {
473  LWLockRelease(&pstate->pa_lock);
474  return false;
475  }
476 
477  /* Loop until we find a subplan to execute. */
478  while (pstate->pa_finished[pstate->pa_next_plan])
479  {
480  if (pstate->pa_next_plan < node->as_nplans - 1)
481  {
482  /* Advance to next plan. */
483  pstate->pa_next_plan++;
484  }
485  else if (append->first_partial_plan < node->as_nplans)
486  {
487  /* Loop back to first partial plan. */
488  pstate->pa_next_plan = append->first_partial_plan;
489  }
490  else
491  {
492  /* At last plan, no partial plans, arrange to bail out. */
493  pstate->pa_next_plan = node->as_whichplan;
494  }
495 
496  if (pstate->pa_next_plan == node->as_whichplan)
497  {
498  /* We've tried everything! */
500  LWLockRelease(&pstate->pa_lock);
501  return false;
502  }
503  }
504 
505  /* Pick the plan we found, and advance pa_next_plan one more time. */
506  node->as_whichplan = pstate->pa_next_plan++;
507  if (pstate->pa_next_plan >= node->as_nplans)
508  {
509  if (append->first_partial_plan < node->as_nplans)
510  pstate->pa_next_plan = append->first_partial_plan;
511  else
512  {
513  /*
514  * We have only non-partial plans, and we already chose the last
515  * one; so arrange for the other workers to immediately bail out.
516  */
518  }
519  }
520 
521  /* If non-partial, immediately mark as finished. */
522  if (node->as_whichplan < append->first_partial_plan)
523  node->as_pstate->pa_finished[node->as_whichplan] = true;
524 
525  LWLockRelease(&pstate->pa_lock);
526 
527  return true;
528 }
Size pstate_len
Definition: execnodes.h:1017
void ExecLockNonLeafAppendTables(List *partitioned_rels, EState *estate)
Definition: execUtils.c:913
Definition: lwlock.h:32
void ExecAppendInitializeWorker(AppendState *node, ParallelWorkerContext *pwcxt)
Definition: nodeAppend.c:352
static TupleTableSlot * ExecAppend(PlanState *pstate)
Definition: nodeAppend.c:184
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:885
#define ScanDirectionIsForward(direction)
Definition: sdir.h:55
#define castNode(_type_, nodeptr)
Definition: nodes.h:581
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:539
void ExecAppendReInitializeDSM(AppendState *node, ParallelContext *pcxt)
Definition: nodeAppend.c:336
shm_toc_estimator estimator
Definition: parallel.h:41
static bool choose_next_subplan_locally(AppendState *node)
Definition: nodeAppend.c:366
void ExecReScan(PlanState *node)
Definition: execAmi.c:76
TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: execTuples.c:439
int plan_node_id
Definition: plannodes.h:143
AppendState * ExecInitAppend(Append *node, EState *estate, int eflags)
Definition: nodeAppend.c:98
EState * state
Definition: execnodes.h:852
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
ScanDirection es_direction
Definition: execnodes.h:429
void ExecAssignResultTypeFromTL(PlanState *planstate)
Definition: execUtils.c:448
PlanState ps
Definition: execnodes.h:1012
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1722
List * appendplans
Definition: plannodes.h:250
TupleTableSlot * ps_ResultTupleSlot
Definition: execnodes.h:883
ParallelAppendState * as_pstate
Definition: execnodes.h:1016
void ExecInitResultTupleSlot(EState *estate, PlanState *planstate)
Definition: execTuples.c:832
void ExecEndAppend(AppendState *node)
Definition: nodeAppend.c:236
int first_partial_plan
Definition: plannodes.h:251
bool(* choose_next_subplan)(AppendState *)
Definition: execnodes.h:1018
bool parallel_aware
Definition: plannodes.h:137
#define TupIsNull(slot)
Definition: tuptable.h:138
List * partitioned_rels
Definition: plannodes.h:249
Bitmapset * chgParam
Definition: execnodes.h:878
static bool choose_next_subplan_for_worker(AppendState *node)
Definition: nodeAppend.c:456
#define INVALID_SUBPLAN_INDEX
Definition: nodeAppend.c:79
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:674
void * palloc0(Size size)
Definition: mcxt.c:864
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:856
void ExecReScanAppend(AppendState *node)
Definition: nodeAppend.c:256
Size add_size(Size s1, Size s2)
Definition: shmem.c:475
static TupleTableSlot * ExecProcNode(PlanState *node)
Definition: executor.h:237
bool pa_finished[FLEXIBLE_ARRAY_MEMBER]
Definition: nodeAppend.c:76
Plan * plan
Definition: execnodes.h:850
void UpdateChangedParamSet(PlanState *node, Bitmapset *newchg)
Definition: execUtils.c:758
void ExecAppendInitializeDSM(AppendState *node, ParallelContext *pcxt)
Definition: nodeAppend.c:315
void ExecAppendEstimate(AppendState *node, ParallelContext *pcxt)
Definition: nodeAppend.c:296
#define makeNode(_type_)
Definition: nodes.h:560
#define Assert(condition)
Definition: c.h:680
#define lfirst(lc)
Definition: pg_list.h:106
#define EXEC_FLAG_MARK
Definition: executor.h:61
static bool choose_next_subplan_for_leader(AppendState *node)
Definition: nodeAppend.c:398
int as_whichplan
Definition: execnodes.h:1015
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
static int list_length(const List *l)
Definition: pg_list.h:89
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1118
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
PlanState ** appendplans
Definition: execnodes.h:1013
int i
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:139
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
#define offsetof(type, field)
Definition: c.h:603
shm_toc * toc
Definition: parallel.h:44