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