PostgreSQL Source Code  git master
nodeGatherMerge.c File Reference
#include "postgres.h"
#include "access/relscan.h"
#include "access/xact.h"
#include "executor/execdebug.h"
#include "executor/execParallel.h"
#include "executor/nodeGatherMerge.h"
#include "executor/nodeSubplan.h"
#include "executor/tqueue.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "utils/memutils.h"
#include "utils/rel.h"
Include dependency graph for nodeGatherMerge.c:

Go to the source code of this file.

Data Structures

struct  GMReaderTupleBuffer
 

Macros

#define MAX_TUPLE_STORE   10
 

Typedefs

typedef struct GMReaderTupleBuffer GMReaderTupleBuffer
 
typedef int32 SlotNumber
 

Functions

static TupleTableSlotExecGatherMerge (PlanState *pstate)
 
static int32 heap_compare_slots (Datum a, Datum b, void *arg)
 
static TupleTableSlotgather_merge_getnext (GatherMergeState *gm_state)
 
static MinimalTuple gm_readnext_tuple (GatherMergeState *gm_state, int nreader, bool nowait, bool *done)
 
static void ExecShutdownGatherMergeWorkers (GatherMergeState *node)
 
static void gather_merge_setup (GatherMergeState *gm_state)
 
static void gather_merge_init (GatherMergeState *gm_state)
 
static void gather_merge_clear_tuples (GatherMergeState *gm_state)
 
static bool gather_merge_readnext (GatherMergeState *gm_state, int reader, bool nowait)
 
static void load_tuple_array (GatherMergeState *gm_state, int reader)
 
GatherMergeStateExecInitGatherMerge (GatherMerge *node, EState *estate, int eflags)
 
void ExecEndGatherMerge (GatherMergeState *node)
 
void ExecShutdownGatherMerge (GatherMergeState *node)
 
void ExecReScanGatherMerge (GatherMergeState *node)
 

Macro Definition Documentation

◆ MAX_TUPLE_STORE

#define MAX_TUPLE_STORE   10

Definition at line 36 of file nodeGatherMerge.c.

Typedef Documentation

◆ GMReaderTupleBuffer

◆ SlotNumber

typedef int32 SlotNumber

Definition at line 744 of file nodeGatherMerge.c.

Function Documentation

◆ ExecEndGatherMerge()

void ExecEndGatherMerge ( GatherMergeState node)

Definition at line 289 of file nodeGatherMerge.c.

290 {
291  ExecEndNode(outerPlanState(node)); /* let children clean up first */
293 }
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:557
#define outerPlanState(node)
Definition: execnodes.h:1132
void ExecShutdownGatherMerge(GatherMergeState *node)

References ExecEndNode(), ExecShutdownGatherMerge(), and outerPlanState.

Referenced by ExecEndNode().

◆ ExecGatherMerge()

static TupleTableSlot * ExecGatherMerge ( PlanState pstate)
static

Definition at line 188 of file nodeGatherMerge.c.

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)
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 }
void LaunchParallelWorkers(ParallelContext *pcxt)
Definition: parallel.c:551
ParallelExecutorInfo * ExecInitParallelPlan(PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed)
Definition: execParallel.c:589
void ExecParallelReinitialize(PlanState *planstate, ParallelExecutorInfo *pei, Bitmapset *sendParams)
Definition: execParallel.c:899
void ExecParallelCreateReaders(ParallelExecutorInfo *pei)
Definition: execParallel.c:873
static TupleTableSlot * ExecProject(ProjectionInfo *projInfo)
Definition: executor.h:375
#define ResetExprContext(econtext)
Definition: executor.h:543
void * palloc(Size size)
Definition: mcxt.c:1226
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
static TupleTableSlot * gather_merge_getnext(GatherMergeState *gm_state)
#define castNode(_type_, nodeptr)
Definition: nodes.h:197
bool parallel_leader_participation
Definition: planner.c:74
bool es_use_parallel_mode
Definition: execnodes.h:692
TupleTableSlot * ecxt_outertuple
Definition: execnodes.h:252
struct ParallelExecutorInfo * pei
Definition: execnodes.h:2620
struct TupleQueueReader ** reader
Definition: execnodes.h:2626
bool need_to_scan_locally
Definition: execnodes.h:2614
PlanState ps
Definition: execnodes.h:2611
Bitmapset * initParam
Definition: plannodes.h:1184
int num_workers
Definition: plannodes.h:1158
int nworkers_launched
Definition: parallel.h:38
ParallelContext * pcxt
Definition: execParallel.h:27
struct TupleQueueReader ** reader
Definition: execParallel.h:37
Plan * plan
Definition: execnodes.h:1036
EState * state
Definition: execnodes.h:1038
ExprContext * ps_ExprContext
Definition: execnodes.h:1075
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:1076
#define TupIsNull(slot)
Definition: tuptable.h:299

References castNode, CHECK_FOR_INTERRUPTS, ExprContext::ecxt_outertuple, EState::es_use_parallel_mode, ExecInitParallelPlan(), ExecParallelCreateReaders(), ExecParallelReinitialize(), ExecProject(), gather_merge_getnext(), GatherMergeState::initialized, GatherMerge::initParam, LaunchParallelWorkers(), GatherMergeState::need_to_scan_locally, GatherMergeState::nreaders, GatherMerge::num_workers, ParallelContext::nworkers_launched, GatherMergeState::nworkers_launched, outerPlanState, palloc(), parallel_leader_participation, ParallelExecutorInfo::pcxt, GatherMergeState::pei, PlanState::plan, GatherMergeState::ps, PlanState::ps_ExprContext, PlanState::ps_ProjInfo, ParallelExecutorInfo::reader, GatherMergeState::reader, ResetExprContext, PlanState::state, TupIsNull, and GatherMergeState::tuples_needed.

Referenced by ExecInitGatherMerge().

◆ ExecInitGatherMerge()

GatherMergeState* ExecInitGatherMerge ( GatherMerge node,
EState estate,
int  eflags 
)

Definition at line 72 of file nodeGatherMerge.c.

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 
170  PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
171  }
172  }
173 
174  /* Now allocate the workspace for gather merge */
175  gather_merge_setup(gm_state);
176 
177  return gm_state;
178 }
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:142
void ExecInitResultTypeTL(PlanState *planstate)
Definition: execTuples.c:1756
TupleDesc ExecGetResultType(PlanState *planstate)
Definition: execUtils.c:498
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:488
void ExecConditionalAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc, int varno)
Definition: execUtils.c:563
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void * palloc0(Size size)
Definition: mcxt.c:1257
MemoryContext CurrentMemoryContext
Definition: mcxt.c:135
static void gather_merge_setup(GatherMergeState *gm_state)
static TupleTableSlot * ExecGatherMerge(PlanState *pstate)
#define makeNode(_type_)
Definition: nodes.h:176
#define innerPlan(node)
Definition: plannodes.h:181
#define outerPlan(node)
Definition: plannodes.h:182
#define OUTER_VAR
Definition: primnodes.h:223
void PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
Definition: sortsupport.c:135
TupleDesc tupDesc
Definition: execnodes.h:2617
SortSupport gm_sortkeys
Definition: execnodes.h:2619
bool outeropsset
Definition: execnodes.h:1119
bool resultopsset
Definition: execnodes.h:1121
bool outeropsfixed
Definition: execnodes.h:1115
bool resultopsfixed
Definition: execnodes.h:1117
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:1042
List * qual
Definition: plannodes.h:153
AttrNumber ssup_attno
Definition: sortsupport.h:81
bool ssup_nulls_first
Definition: sortsupport.h:75
MemoryContext ssup_cxt
Definition: sortsupport.h:66

References SortSupportData::abbreviate, Assert(), CurrentMemoryContext, ExecAssignExprContext(), ExecConditionalAssignProjectionInfo(), ExecGatherMerge(), ExecGetResultType(), ExecInitNode(), ExecInitResultTypeTL(), PlanState::ExecProcNode, gather_merge_setup(), GatherMergeState::gm_initialized, GatherMergeState::gm_nkeys, GatherMergeState::gm_sortkeys, i, GatherMergeState::initialized, innerPlan, makeNode, GatherMerge::numCols, OUTER_VAR, PlanState::outeropsfixed, PlanState::outeropsset, outerPlan, outerPlanState, palloc0(), PlanState::plan, GatherMerge::plan, PrepareSortSupportFromOrderingOp(), GatherMergeState::ps, PlanState::ps_ProjInfo, Plan::qual, PlanState::resultopsfixed, PlanState::resultopsset, SortSupportData::ssup_attno, SortSupportData::ssup_collation, SortSupportData::ssup_cxt, SortSupportData::ssup_nulls_first, PlanState::state, GatherMergeState::tupDesc, and GatherMergeState::tuples_needed.

Referenced by ExecInitNode().

◆ ExecReScanGatherMerge()

void ExecReScanGatherMerge ( GatherMergeState node)

Definition at line 339 of file nodeGatherMerge.c.

340 {
341  GatherMerge *gm = (GatherMerge *) node->ps.plan;
343 
344  /* Make sure any existing workers are gracefully shut down */
346 
347  /* Free any unused tuples, so we don't leak memory across rescans */
349 
350  /* Mark node so that shared state will be rebuilt at next call */
351  node->initialized = false;
352  node->gm_initialized = false;
353 
354  /*
355  * Set child node's chgParam to tell it that the next scan might deliver a
356  * different set of rows within the leader process. (The overall rowset
357  * shouldn't change, but the leader process's subset might; hence nodes
358  * between here and the parallel table scan node mustn't optimize on the
359  * assumption of an unchanging rowset.)
360  */
361  if (gm->rescan_param >= 0)
362  outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
363  gm->rescan_param);
364 
365  /*
366  * If chgParam of subnode is not null then plan will be re-scanned by
367  * first ExecProcNode. Note: because this does nothing if we have a
368  * rescan_param, it's currently guaranteed that parallel-aware child nodes
369  * will not see a ReScan call until after they get a ReInitializeDSM call.
370  * That ordering might not be something to rely on, though. A good rule
371  * of thumb is that ReInitializeDSM should reset only shared state, ReScan
372  * should reset only local state, and anything that depends on both of
373  * those steps being finished must wait until the first ExecProcNode call.
374  */
375  if (outerPlan->chgParam == NULL)
377 }
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:753
void ExecReScan(PlanState *node)
Definition: execAmi.c:78
static void gather_merge_clear_tuples(GatherMergeState *gm_state)
static void ExecShutdownGatherMergeWorkers(GatherMergeState *node)
int rescan_param
Definition: plannodes.h:1161

References bms_add_member(), ExecReScan(), ExecShutdownGatherMergeWorkers(), gather_merge_clear_tuples(), GatherMergeState::gm_initialized, GatherMergeState::initialized, outerPlan, outerPlanState, PlanState::plan, GatherMergeState::ps, and GatherMerge::rescan_param.

Referenced by ExecReScan().

◆ ExecShutdownGatherMerge()

void ExecShutdownGatherMerge ( GatherMergeState node)

Definition at line 302 of file nodeGatherMerge.c.

303 {
305 
306  /* Now destroy the parallel context. */
307  if (node->pei != NULL)
308  {
309  ExecParallelCleanup(node->pei);
310  node->pei = NULL;
311  }
312 }
void ExecParallelCleanup(ParallelExecutorInfo *pei)

References ExecParallelCleanup(), ExecShutdownGatherMergeWorkers(), and GatherMergeState::pei.

Referenced by ExecEndGatherMerge(), and ExecShutdownNode_walker().

◆ ExecShutdownGatherMergeWorkers()

static void ExecShutdownGatherMergeWorkers ( GatherMergeState node)
static

Definition at line 321 of file nodeGatherMerge.c.

322 {
323  if (node->pei != NULL)
324  ExecParallelFinish(node->pei);
325 
326  /* Flush local copy of reader array */
327  if (node->reader)
328  pfree(node->reader);
329  node->reader = NULL;
330 }
void ExecParallelFinish(ParallelExecutorInfo *pei)
void pfree(void *pointer)
Definition: mcxt.c:1456

References ExecParallelFinish(), GatherMergeState::pei, pfree(), and GatherMergeState::reader.

Referenced by ExecReScanGatherMerge(), and ExecShutdownGatherMerge().

◆ gather_merge_clear_tuples()

static void gather_merge_clear_tuples ( GatherMergeState gm_state)
static

Definition at line 524 of file nodeGatherMerge.c.

525 {
526  int i;
527 
528  for (i = 0; i < gm_state->nreaders; i++)
529  {
530  GMReaderTupleBuffer *tuple_buffer = &gm_state->gm_tuple_buffers[i];
531 
532  while (tuple_buffer->readCounter < tuple_buffer->nTuples)
533  pfree(tuple_buffer->tuple[tuple_buffer->readCounter++]);
534 
535  ExecClearTuple(gm_state->gm_slots[i + 1]);
536  }
537 }
MinimalTuple * tuple
struct GMReaderTupleBuffer * gm_tuple_buffers
Definition: execnodes.h:2627
TupleTableSlot ** gm_slots
Definition: execnodes.h:2625
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: tuptable.h:432

References ExecClearTuple(), GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, i, GatherMergeState::nreaders, GMReaderTupleBuffer::nTuples, pfree(), GMReaderTupleBuffer::readCounter, and GMReaderTupleBuffer::tuple.

Referenced by ExecReScanGatherMerge(), and gather_merge_getnext().

◆ gather_merge_getnext()

static TupleTableSlot * gather_merge_getnext ( GatherMergeState gm_state)
static

Definition at line 545 of file nodeGatherMerge.c.

546 {
547  int i;
548 
549  if (!gm_state->gm_initialized)
550  {
551  /*
552  * First time through: pull the first tuple from each participant, and
553  * set up the heap.
554  */
555  gather_merge_init(gm_state);
556  }
557  else
558  {
559  /*
560  * Otherwise, pull the next tuple from whichever participant we
561  * returned from last time, and reinsert that participant's index into
562  * the heap, because it might now compare differently against the
563  * other elements of the heap.
564  */
565  i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
566 
567  if (gather_merge_readnext(gm_state, i, false))
569  else
570  {
571  /* reader exhausted, remove it from heap */
572  (void) binaryheap_remove_first(gm_state->gm_heap);
573  }
574  }
575 
576  if (binaryheap_empty(gm_state->gm_heap))
577  {
578  /* All the queues are exhausted, and so is the heap */
579  gather_merge_clear_tuples(gm_state);
580  return NULL;
581  }
582  else
583  {
584  /* Return next tuple from whichever participant has the leading one */
585  i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
586  return gm_state->gm_slots[i];
587  }
588 }
void binaryheap_replace_first(binaryheap *heap, bh_node_type d)
Definition: binaryheap.c:255
bh_node_type binaryheap_first(binaryheap *heap)
Definition: binaryheap.c:177
bh_node_type binaryheap_remove_first(binaryheap *heap)
Definition: binaryheap.c:192
#define binaryheap_empty(h)
Definition: binaryheap.h:65
static void gather_merge_init(GatherMergeState *gm_state)
static bool gather_merge_readnext(GatherMergeState *gm_state, int reader, bool nowait)
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:212
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:202
struct binaryheap * gm_heap
Definition: execnodes.h:2628

References binaryheap_empty, binaryheap_first(), binaryheap_remove_first(), binaryheap_replace_first(), DatumGetInt32(), gather_merge_clear_tuples(), gather_merge_init(), gather_merge_readnext(), GatherMergeState::gm_heap, GatherMergeState::gm_initialized, GatherMergeState::gm_slots, i, and Int32GetDatum().

Referenced by ExecGatherMerge().

◆ gather_merge_init()

static void gather_merge_init ( GatherMergeState gm_state)
static

Definition at line 441 of file nodeGatherMerge.c.

442 {
443  int nreaders = gm_state->nreaders;
444  bool nowait = true;
445  int i;
446 
447  /* Assert that gather_merge_setup made enough space */
448  Assert(nreaders <= castNode(GatherMerge, gm_state->ps.plan)->num_workers);
449 
450  /* Reset leader's tuple slot to empty */
451  gm_state->gm_slots[0] = NULL;
452 
453  /* Reset the tuple slot and tuple array for each worker */
454  for (i = 0; i < nreaders; i++)
455  {
456  /* Reset tuple array to empty */
457  gm_state->gm_tuple_buffers[i].nTuples = 0;
458  gm_state->gm_tuple_buffers[i].readCounter = 0;
459  /* Reset done flag to not-done */
460  gm_state->gm_tuple_buffers[i].done = false;
461  /* Ensure output slot is empty */
462  ExecClearTuple(gm_state->gm_slots[i + 1]);
463  }
464 
465  /* Reset binary heap to empty */
466  binaryheap_reset(gm_state->gm_heap);
467 
468  /*
469  * First, try to read a tuple from each worker (including leader) in
470  * nowait mode. After this, if not all workers were able to produce a
471  * tuple (or a "done" indication), then re-read from remaining workers,
472  * this time using wait mode. Add all live readers (those producing at
473  * least one tuple) to the heap.
474  */
475 reread:
476  for (i = 0; i <= nreaders; i++)
477  {
479 
480  /* skip this source if already known done */
481  if ((i == 0) ? gm_state->need_to_scan_locally :
482  !gm_state->gm_tuple_buffers[i - 1].done)
483  {
484  if (TupIsNull(gm_state->gm_slots[i]))
485  {
486  /* Don't have a tuple yet, try to get one */
487  if (gather_merge_readnext(gm_state, i, nowait))
489  Int32GetDatum(i));
490  }
491  else
492  {
493  /*
494  * We already got at least one tuple from this worker, but
495  * might as well see if it has any more ready by now.
496  */
497  load_tuple_array(gm_state, i);
498  }
499  }
500  }
501 
502  /* need not recheck leader, since nowait doesn't matter for it */
503  for (i = 1; i <= nreaders; i++)
504  {
505  if (!gm_state->gm_tuple_buffers[i - 1].done &&
506  TupIsNull(gm_state->gm_slots[i]))
507  {
508  nowait = false;
509  goto reread;
510  }
511  }
512 
513  /* Now heapify the heap. */
514  binaryheap_build(gm_state->gm_heap);
515 
516  gm_state->gm_initialized = true;
517 }
void binaryheap_build(binaryheap *heap)
Definition: binaryheap.c:138
void binaryheap_reset(binaryheap *heap)
Definition: binaryheap.c:63
void binaryheap_add_unordered(binaryheap *heap, bh_node_type d)
Definition: binaryheap.c:116
static void load_tuple_array(GatherMergeState *gm_state, int reader)

References Assert(), binaryheap_add_unordered(), binaryheap_build(), binaryheap_reset(), castNode, CHECK_FOR_INTERRUPTS, GMReaderTupleBuffer::done, ExecClearTuple(), gather_merge_readnext(), GatherMergeState::gm_heap, GatherMergeState::gm_initialized, GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, i, Int32GetDatum(), load_tuple_array(), GatherMergeState::need_to_scan_locally, GatherMergeState::nreaders, GMReaderTupleBuffer::nTuples, PlanState::plan, GatherMergeState::ps, GMReaderTupleBuffer::readCounter, and TupIsNull.

Referenced by gather_merge_getnext().

◆ gather_merge_readnext()

static bool gather_merge_readnext ( GatherMergeState gm_state,
int  reader,
bool  nowait 
)
static

Definition at line 634 of file nodeGatherMerge.c.

635 {
636  GMReaderTupleBuffer *tuple_buffer;
637  MinimalTuple tup;
638 
639  /*
640  * If we're being asked to generate a tuple from the leader, then we just
641  * call ExecProcNode as normal to produce one.
642  */
643  if (reader == 0)
644  {
645  if (gm_state->need_to_scan_locally)
646  {
647  PlanState *outerPlan = outerPlanState(gm_state);
648  TupleTableSlot *outerTupleSlot;
649  EState *estate = gm_state->ps.state;
650 
651  /* Install our DSA area while executing the plan. */
652  estate->es_query_dsa = gm_state->pei ? gm_state->pei->area : NULL;
653  outerTupleSlot = ExecProcNode(outerPlan);
654  estate->es_query_dsa = NULL;
655 
656  if (!TupIsNull(outerTupleSlot))
657  {
658  gm_state->gm_slots[0] = outerTupleSlot;
659  return true;
660  }
661  /* need_to_scan_locally serves as "done" flag for leader */
662  gm_state->need_to_scan_locally = false;
663  }
664  return false;
665  }
666 
667  /* Otherwise, check the state of the relevant tuple buffer. */
668  tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
669 
670  if (tuple_buffer->nTuples > tuple_buffer->readCounter)
671  {
672  /* Return any tuple previously read that is still buffered. */
673  tup = tuple_buffer->tuple[tuple_buffer->readCounter++];
674  }
675  else if (tuple_buffer->done)
676  {
677  /* Reader is known to be exhausted. */
678  return false;
679  }
680  else
681  {
682  /* Read and buffer next tuple. */
683  tup = gm_readnext_tuple(gm_state,
684  reader,
685  nowait,
686  &tuple_buffer->done);
687  if (!tup)
688  return false;
689 
690  /*
691  * Attempt to read more tuples in nowait mode and store them in the
692  * pending-tuple array for the reader.
693  */
694  load_tuple_array(gm_state, reader);
695  }
696 
697  Assert(tup);
698 
699  /* Build the TupleTableSlot for the given tuple */
700  ExecStoreMinimalTuple(tup, /* tuple to store */
701  gm_state->gm_slots[reader], /* slot in which to
702  * store the tuple */
703  true); /* pfree tuple when done with it */
704 
705  return true;
706 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1447
static TupleTableSlot * ExecProcNode(PlanState *node)
Definition: executor.h:268
static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait, bool *done)
struct dsa_area * es_query_dsa
Definition: execnodes.h:695

References ParallelExecutorInfo::area, Assert(), GMReaderTupleBuffer::done, EState::es_query_dsa, ExecProcNode(), ExecStoreMinimalTuple(), gm_readnext_tuple(), GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, load_tuple_array(), GatherMergeState::need_to_scan_locally, GMReaderTupleBuffer::nTuples, outerPlan, outerPlanState, GatherMergeState::pei, GatherMergeState::ps, GMReaderTupleBuffer::readCounter, PlanState::state, TupIsNull, and GMReaderTupleBuffer::tuple.

Referenced by gather_merge_getnext(), and gather_merge_init().

◆ gather_merge_setup()

static void gather_merge_setup ( GatherMergeState gm_state)
static

Definition at line 393 of file nodeGatherMerge.c.

394 {
395  GatherMerge *gm = castNode(GatherMerge, gm_state->ps.plan);
396  int nreaders = gm->num_workers;
397  int i;
398 
399  /*
400  * Allocate gm_slots for the number of workers + one more slot for leader.
401  * Slot 0 is always for the leader. Leader always calls ExecProcNode() to
402  * read the tuple, and then stores it directly into its gm_slots entry.
403  * For other slots, code below will call ExecInitExtraTupleSlot() to
404  * create a slot for the worker's results. Note that during any single
405  * scan, we might have fewer than num_workers available workers, in which
406  * case the extra array entries go unused.
407  */
408  gm_state->gm_slots = (TupleTableSlot **)
409  palloc0((nreaders + 1) * sizeof(TupleTableSlot *));
410 
411  /* Allocate the tuple slot and tuple array for each worker */
412  gm_state->gm_tuple_buffers = (GMReaderTupleBuffer *)
413  palloc0(nreaders * sizeof(GMReaderTupleBuffer));
414 
415  for (i = 0; i < nreaders; i++)
416  {
417  /* Allocate the tuple array with length MAX_TUPLE_STORE */
418  gm_state->gm_tuple_buffers[i].tuple =
420 
421  /* Initialize tuple slot for worker */
422  gm_state->gm_slots[i + 1] =
423  ExecInitExtraTupleSlot(gm_state->ps.state, gm_state->tupDesc,
425  }
426 
427  /* Allocate the resources for the merge */
428  gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
430  gm_state);
431 }
binaryheap * binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg)
Definition: binaryheap.c:39
TupleTableSlot * ExecInitExtraTupleSlot(EState *estate, TupleDesc tupledesc, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1832
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:85
static int32 heap_compare_slots(Datum a, Datum b, void *arg)
#define MAX_TUPLE_STORE

References binaryheap_allocate(), castNode, ExecInitExtraTupleSlot(), GatherMergeState::gm_heap, GatherMergeState::gm_slots, GatherMergeState::gm_tuple_buffers, heap_compare_slots(), i, MAX_TUPLE_STORE, GatherMerge::num_workers, palloc0(), PlanState::plan, GatherMergeState::ps, PlanState::state, TTSOpsMinimalTuple, GatherMergeState::tupDesc, and GMReaderTupleBuffer::tuple.

Referenced by ExecInitGatherMerge().

◆ gm_readnext_tuple()

static MinimalTuple gm_readnext_tuple ( GatherMergeState gm_state,
int  nreader,
bool  nowait,
bool done 
)
static

Definition at line 712 of file nodeGatherMerge.c.

714 {
715  TupleQueueReader *reader;
716  MinimalTuple tup;
717 
718  /* Check for async events, particularly messages from workers. */
720 
721  /*
722  * Attempt to read a tuple.
723  *
724  * Note that TupleQueueReaderNext will just return NULL for a worker which
725  * fails to initialize. We'll treat that worker as having produced no
726  * tuples; WaitForParallelWorkersToFinish will error out when we get
727  * there.
728  */
729  reader = gm_state->reader[nreader - 1];
730  tup = TupleQueueReaderNext(reader, nowait, done);
731 
732  /*
733  * Since we'll be buffering these across multiple calls, we need to make a
734  * copy.
735  */
736  return tup ? heap_copy_minimal_tuple(tup) : NULL;
737 }
MinimalTuple heap_copy_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1536
MinimalTuple TupleQueueReaderNext(TupleQueueReader *reader, bool nowait, bool *done)
Definition: tqueue.c:176

References CHECK_FOR_INTERRUPTS, heap_copy_minimal_tuple(), GatherMergeState::reader, and TupleQueueReaderNext().

Referenced by gather_merge_readnext(), and load_tuple_array().

◆ heap_compare_slots()

static int32 heap_compare_slots ( Datum  a,
Datum  b,
void *  arg 
)
static

Definition at line 750 of file nodeGatherMerge.c.

751 {
753  SlotNumber slot1 = DatumGetInt32(a);
754  SlotNumber slot2 = DatumGetInt32(b);
755 
756  TupleTableSlot *s1 = node->gm_slots[slot1];
757  TupleTableSlot *s2 = node->gm_slots[slot2];
758  int nkey;
759 
760  Assert(!TupIsNull(s1));
761  Assert(!TupIsNull(s2));
762 
763  for (nkey = 0; nkey < node->gm_nkeys; nkey++)
764  {
765  SortSupport sortKey = node->gm_sortkeys + nkey;
766  AttrNumber attno = sortKey->ssup_attno;
767  Datum datum1,
768  datum2;
769  bool isNull1,
770  isNull2;
771  int compare;
772 
773  datum1 = slot_getattr(s1, attno, &isNull1);
774  datum2 = slot_getattr(s2, attno, &isNull2);
775 
776  compare = ApplySortComparator(datum1, isNull1,
777  datum2, isNull2,
778  sortKey);
779  if (compare != 0)
780  {
782  return compare;
783  }
784  }
785  return 0;
786 }
int16 AttrNumber
Definition: attnum.h:21
#define INVERT_COMPARE_RESULT(var)
Definition: c.h:1119
static int compare(const void *arg1, const void *arg2)
Definition: geqo_pool.c:145
int b
Definition: isn.c:70
int a
Definition: isn.c:69
int32 SlotNumber
void * arg
uintptr_t Datum
Definition: postgres.h:64
char * s1
char * s2
static int ApplySortComparator(Datum datum1, bool isNull1, Datum datum2, bool isNull2, SortSupport ssup)
Definition: sortsupport.h:200
static Datum slot_getattr(TupleTableSlot *slot, int attnum, bool *isnull)
Definition: tuptable.h:388

References a, ApplySortComparator(), arg, Assert(), b, compare(), DatumGetInt32(), GatherMergeState::gm_nkeys, GatherMergeState::gm_slots, GatherMergeState::gm_sortkeys, INVERT_COMPARE_RESULT, s1, s2, slot_getattr(), SortSupportData::ssup_attno, and TupIsNull.

Referenced by gather_merge_setup().

◆ load_tuple_array()

static void load_tuple_array ( GatherMergeState gm_state,
int  reader 
)
static

Definition at line 595 of file nodeGatherMerge.c.

596 {
597  GMReaderTupleBuffer *tuple_buffer;
598  int i;
599 
600  /* Don't do anything if this is the leader. */
601  if (reader == 0)
602  return;
603 
604  tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
605 
606  /* If there's nothing in the array, reset the counters to zero. */
607  if (tuple_buffer->nTuples == tuple_buffer->readCounter)
608  tuple_buffer->nTuples = tuple_buffer->readCounter = 0;
609 
610  /* Try to fill additional slots in the array. */
611  for (i = tuple_buffer->nTuples; i < MAX_TUPLE_STORE; i++)
612  {
613  MinimalTuple tuple;
614 
615  tuple = gm_readnext_tuple(gm_state,
616  reader,
617  true,
618  &tuple_buffer->done);
619  if (!tuple)
620  break;
621  tuple_buffer->tuple[i] = tuple;
622  tuple_buffer->nTuples++;
623  }
624 }

References GMReaderTupleBuffer::done, gm_readnext_tuple(), GatherMergeState::gm_tuple_buffers, i, MAX_TUPLE_STORE, GMReaderTupleBuffer::nTuples, GMReaderTupleBuffer::readCounter, and GMReaderTupleBuffer::tuple.

Referenced by gather_merge_init(), and gather_merge_readnext().