PostgreSQL Source Code git master
Data Structures | Macros | Typedefs | Functions
nodeGatherMerge.c File Reference
#include "postgres.h"
#include "access/htup_details.h"
#include "executor/executor.h"
#include "executor/execParallel.h"
#include "executor/nodeGatherMerge.h"
#include "executor/tqueue.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "optimizer/optimizer.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 TupleTableSlot *  ExecGatherMerge (PlanState *pstate)
 
static int32  heap_compare_slots (Datum a, Datum b, void *arg)
 
static TupleTableSlot *  gather_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)
 
GatherMergeState *  ExecInitGatherMerge (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 32 of file nodeGatherMerge.c.

Typedef Documentation

◆ GMReaderTupleBuffer

typedef struct GMReaderTupleBuffer GMReaderTupleBuffer

◆ SlotNumber

typedef int32 SlotNumber

Definition at line 747 of file nodeGatherMerge.c.

Function Documentation

◆ ExecEndGatherMerge()

void ExecEndGatherMerge ( GatherMergeState *  node )

Definition at line 292 of file nodeGatherMerge.c.

293{
294 ExecEndNode(outerPlanState(node)); /* let children clean up first */
296}
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:562
#define outerPlanState(node)
Definition: execnodes.h:1255
void ExecShutdownGatherMerge(GatherMergeState *node)

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

Referenced by ExecEndNode().

◆ ExecGatherMerge()

static TupleTableSlot * ExecGatherMerge ( PlanState *  pstate )
static

Definition at line 184 of file nodeGatherMerge.c.

185{
187 TupleTableSlot *slot;
188 ExprContext *econtext;
189
191
192 /*
193 * As with Gather, we don't launch workers until this node is actually
194 * executed.
195 */
196 if (!node->initialized)
197 {
198 EState *estate = node->ps.state;
199 GatherMerge *gm = castNode(GatherMerge, node->ps.plan);
200
201 /*
202 * Sometimes we might have to run without parallelism; but if parallel
203 * mode is active then we can try to fire up some workers.
204 */
205 if (gm->num_workers > 0 && estate->es_use_parallel_mode)
206 {
207 ParallelContext *pcxt;
208
209 /* Initialize, or re-initialize, shared state needed by workers. */
210 if (!node->pei)
212 estate,
213 gm->initParam,
214 gm->num_workers,
215 node->tuples_needed);
216 else
218 node->pei,
219 gm->initParam);
220
221 /* Try to launch workers. */
222 pcxt = node->pei->pcxt;
224 /* We save # workers launched for the benefit of EXPLAIN */
226
227 /*
228 * Count number of workers originally wanted and actually
229 * launched.
230 */
233
234 /* Set up tuple queue readers to read the results. */
235 if (pcxt->nworkers_launched > 0)
236 {
238 /* Make a working array showing the active readers */
239 node->nreaders = pcxt->nworkers_launched;
240 node->reader = (TupleQueueReader **)
241 palloc(node->nreaders * sizeof(TupleQueueReader *));
242 memcpy(node->reader, node->pei->reader,
243 node->nreaders * sizeof(TupleQueueReader *));
244 }
245 else
246 {
247 /* No workers? Then never mind. */
248 node->nreaders = 0;
249 node->reader = NULL;
250 }
251 }
252
253 /* allow leader to participate if enabled or no choice */
254 if (parallel_leader_participation || node->nreaders == 0)
255 node->need_to_scan_locally = true;
256 node->initialized = true;
257 }
258
259 /*
260 * Reset per-tuple memory context to free any expression evaluation
261 * storage allocated in the previous tuple cycle.
262 */
263 econtext = node->ps.ps_ExprContext;
264 ResetExprContext(econtext);
265
266 /*
267 * Get next tuple, either from one of our workers, or by running the plan
268 * ourselves.
269 */
270 slot = gather_merge_getnext(node);
271 if (TupIsNull(slot))
272 return NULL;
273
274 /* If no projection is required, we're done. */
275 if (node->ps.ps_ProjInfo == NULL)
276 return slot;
277
278 /*
279 * Form the result tuple using ExecProject(), and return it.
280 */
281 econtext->ecxt_outertuple = slot;
282 return ExecProject(node->ps.ps_ProjInfo);
283}
void LaunchParallelWorkers(ParallelContext *pcxt)
Definition: parallel.c:573
void ExecParallelReinitialize(PlanState *planstate, ParallelExecutorInfo *pei, Bitmapset *sendParams)
Definition: execParallel.c:917
void ExecParallelCreateReaders(ParallelExecutorInfo *pei)
Definition: execParallel.c:891
ParallelExecutorInfo * ExecInitParallelPlan(PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed)
Definition: execParallel.c:600
static TupleTableSlot * ExecProject(ProjectionInfo *projInfo)
Definition: executor.h:480
#define ResetExprContext(econtext)
Definition: executor.h:647
void * palloc(Size size)
Definition: mcxt.c:1365
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
static TupleTableSlot * gather_merge_getnext(GatherMergeState *gm_state)
#define castNode(_type_, nodeptr)
Definition: nodes.h:182
bool parallel_leader_participation
Definition: planner.c:70
Definition: execnodes.h:655
int es_parallel_workers_to_launch
Definition: execnodes.h:746
bool es_use_parallel_mode
Definition: execnodes.h:744
int es_parallel_workers_launched
Definition: execnodes.h:748
TupleTableSlot * ecxt_outertuple
Definition: execnodes.h:277
struct ParallelExecutorInfo * pei
Definition: execnodes.h:2771
bool initialized
Definition: execnodes.h:2763
int nworkers_launched
Definition: execnodes.h:2774
struct TupleQueueReader ** reader
Definition: execnodes.h:2777
int64 tuples_needed
Definition: execnodes.h:2766
bool need_to_scan_locally
Definition: execnodes.h:2765
PlanState ps
Definition: execnodes.h:2762
Bitmapset * initParam
Definition: plannodes.h:1382
int num_workers
Definition: plannodes.h:1356
int nworkers_launched
Definition: parallel.h:37
int nworkers_to_launch
Definition: parallel.h:36
ParallelContext * pcxt
Definition: execParallel.h:27
struct TupleQueueReader ** reader
Definition: execParallel.h:37
Plan * plan
Definition: execnodes.h:1159
EState * state
Definition: execnodes.h:1161
ExprContext * ps_ExprContext
Definition: execnodes.h:1198
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:1199
#define TupIsNull(slot)
Definition: tuptable.h:309

References castNode, CHECK_FOR_INTERRUPTS, ExprContext::ecxt_outertuple, EState::es_parallel_workers_launched, EState::es_parallel_workers_to_launch, 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, ParallelContext::nworkers_to_launch, 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 68 of file nodeGatherMerge.c.

69{
70 GatherMergeState *gm_state;
71 Plan *outerNode;
72 TupleDesc tupDesc;
73
74 /* Gather merge node doesn't have innerPlan node. */
75 Assert(innerPlan(node) == NULL);
76
77 /*
78 * create state structure
79 */
80 gm_state = makeNode(GatherMergeState);
81 gm_state->ps.plan = (Plan *) node;
82 gm_state->ps.state = estate;
84
85 gm_state->initialized = false;
86 gm_state->gm_initialized = false;
87 gm_state->tuples_needed = -1;
88
89 /*
90 * Miscellaneous initialization
91 *
92 * create expression context for node
93 */
94 ExecAssignExprContext(estate, &gm_state->ps);
95
96 /*
97 * GatherMerge doesn't support checking a qual (it's always more efficient
98 * to do it in the child node).
99 */
100 Assert(!node->plan.qual);
101
102 /*
103 * now initialize outer plan
104 */
105 outerNode = outerPlan(node);
106 outerPlanState(gm_state) = ExecInitNode(outerNode, estate, eflags);
107
108 /*
109 * Leader may access ExecProcNode result directly (if
110 * need_to_scan_locally), or from workers via tuple queue. So we can't
111 * trivially rely on the slot type being fixed for expressions evaluated
112 * within this node.
113 */
114 gm_state->ps.outeropsset = true;
115 gm_state->ps.outeropsfixed = false;
116
117 /*
118 * Store the tuple descriptor into gather merge state, so we can use it
119 * while initializing the gather merge slots.
120 */
121 tupDesc = ExecGetResultType(outerPlanState(gm_state));
122 gm_state->tupDesc = tupDesc;
123
124 /*
125 * Initialize result type and projection.
126 */
127 ExecInitResultTypeTL(&gm_state->ps);
129
130 /*
131 * Without projections result slot type is not trivially known, see
132 * comment above.
133 */
134 if (gm_state->ps.ps_ProjInfo == NULL)
135 {
136 gm_state->ps.resultopsset = true;
137 gm_state->ps.resultopsfixed = false;
138 }
139
140 /*
141 * initialize sort-key information
142 */
143 if (node->numCols)
144 {
145 int i;
146
147 gm_state->gm_nkeys = node->numCols;
148 gm_state->gm_sortkeys =
149 palloc0(sizeof(SortSupportData) * node->numCols);
150
151 for (i = 0; i < node->numCols; i++)
152 {
153 SortSupport sortKey = gm_state->gm_sortkeys + i;
154
156 sortKey->ssup_collation = node->collations[i];
157 sortKey->ssup_nulls_first = node->nullsFirst[i];
158 sortKey->ssup_attno = node->sortColIdx[i];
159
160 /*
161 * We don't perform abbreviated key conversion here, for the same
162 * reasons that it isn't used in MergeAppend
163 */
164 sortKey->abbreviate = false;
165
166 PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
167 }
168 }
169
170 /* Now allocate the workspace for gather merge */
171 gather_merge_setup(gm_state);
172
173 return gm_state;
174}
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:142
void ExecInitResultTypeTL(PlanState *planstate)
Definition: execTuples.c:1944
TupleDesc ExecGetResultType(PlanState *planstate)
Definition: execUtils.c:495
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:485
void ExecConditionalAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc, int varno)
Definition: execUtils.c:603
Assert(PointerIsAligned(start, uint64))
i
int i
Definition: isn.c:77
void * palloc0(Size size)
Definition: mcxt.c:1395
MemoryContext CurrentMemoryContext
Definition: mcxt.c:160
static void gather_merge_setup(GatherMergeState *gm_state)
static TupleTableSlot * ExecGatherMerge(PlanState *pstate)
#define makeNode(_type_)
Definition: nodes.h:161
#define innerPlan(node)
Definition: plannodes.h:251
#define outerPlan(node)
Definition: plannodes.h:252
#define OUTER_VAR
Definition: primnodes.h:243
void PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
Definition: sortsupport.c:134
TupleDesc tupDesc
Definition: execnodes.h:2768
SortSupport gm_sortkeys
Definition: execnodes.h:2770
bool gm_initialized
Definition: execnodes.h:2764
int numCols
Definition: plannodes.h:1364
Plan plan
Definition: plannodes.h:1353
bool outeropsset
Definition: execnodes.h:1242
bool resultopsset
Definition: execnodes.h:1244
bool outeropsfixed
Definition: execnodes.h:1238
bool resultopsfixed
Definition: execnodes.h:1240
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:1165
Definition: plannodes.h:177
List * qual
Definition: plannodes.h:222
AttrNumber ssup_attno
Definition: sortsupport.h:81
bool ssup_nulls_first
Definition: sortsupport.h:75
Oid ssup_collation
Definition: sortsupport.h:67
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 342 of file nodeGatherMerge.c.

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}
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:815
void ExecReScan(PlanState *node)
Definition: execAmi.c:77
static void gather_merge_clear_tuples(GatherMergeState *gm_state)
static void ExecShutdownGatherMergeWorkers(GatherMergeState *node)
int rescan_param
Definition: plannodes.h:1359

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 305 of file nodeGatherMerge.c.

306{
308
309 /* Now destroy the parallel context. */
310 if (node->pei != NULL)
311 {
313 node->pei = NULL;
314 }
315}
void ExecParallelCleanup(ParallelExecutorInfo *pei)
Definition: execParallel.c:1210

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

Referenced by ExecEndGatherMerge(), and ExecShutdownNode_walker().

◆ ExecShutdownGatherMergeWorkers()

static void ExecShutdownGatherMergeWorkers ( GatherMergeState *  node )
static

Definition at line 324 of file nodeGatherMerge.c.

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}
void ExecParallelFinish(ParallelExecutorInfo *pei)
Definition: execParallel.c:1157
void pfree(void *pointer)
Definition: mcxt.c:1594

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 527 of file nodeGatherMerge.c.

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 pfree(tuple_buffer->tuple[tuple_buffer->readCounter++]);
537
538 ExecClearTuple(gm_state->gm_slots[i + 1]);
539 }
540}
MinimalTuple * tuple
struct GMReaderTupleBuffer * gm_tuple_buffers
Definition: execnodes.h:2778
TupleTableSlot ** gm_slots
Definition: execnodes.h:2776
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: tuptable.h:457

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 548 of file nodeGatherMerge.c.

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 */
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 */
583 return NULL;
584 }
585 else
586 {
587 /* Return next tuple from whichever participant has the leading one */
589 return gm_state->gm_slots[i];
590 }
591}
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:222
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:212
struct binaryheap * gm_heap
Definition: execnodes.h:2779

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 444 of file nodeGatherMerge.c.

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 */
478reread:
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))
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}
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 637 of file nodeGatherMerge.c.

638{
639 GMReaderTupleBuffer *tuple_buffer;
640 MinimalTuple 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 {
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 (!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(tup);
701
702 /* Build the TupleTableSlot for the given tuple */
703 ExecStoreMinimalTuple(tup, /* tuple to store */
704 gm_state->gm_slots[reader], /* slot in which to
705 * store the tuple */
706 true); /* pfree tuple when done with it */
707
708 return true;
709}
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1635
static TupleTableSlot * ExecProcNode(PlanState *node)
Definition: executor.h:311
static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait, bool *done)
struct dsa_area * es_query_dsa
Definition: execnodes.h:752
dsa_area * area
Definition: execParallel.h:32

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 396 of file nodeGatherMerge.c.

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 */
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,
428 }
429
430 /* Allocate the resources for the merge */
431 gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
433 gm_state);
434}
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:2020
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:86
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 715 of file nodeGatherMerge.c.

717{
718 TupleQueueReader *reader;
719 MinimalTuple 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 /*
736 * Since we'll be buffering these across multiple calls, we need to make a
737 * copy.
738 */
739 return tup ? heap_copy_minimal_tuple(tup, 0) : NULL;
740}
MinimalTuple heap_copy_minimal_tuple(MinimalTuple mtup, Size extra)
Definition: heaptuple.c:1542
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 753 of file nodeGatherMerge.c.

754{
756 SlotNumber slot1 = DatumGetInt32(a);
757 SlotNumber slot2 = DatumGetInt32(b);
758
759 TupleTableSlot *s1 = node->gm_slots[slot1];
760 TupleTableSlot *s2 = node->gm_slots[slot2];
761 int nkey;
762
765
766 for (nkey = 0; nkey < node->gm_nkeys; nkey++)
767 {
768 SortSupport sortKey = node->gm_sortkeys + nkey;
769 AttrNumber attno = sortKey->ssup_attno;
770 Datum datum1,
771 datum2;
772 bool isNull1,
773 isNull2;
774 int compare;
775
776 datum1 = slot_getattr(s1, attno, &isNull1);
777 datum2 = slot_getattr(s2, attno, &isNull2);
778
779 compare = ApplySortComparator(datum1, isNull1,
780 datum2, isNull2,
781 sortKey);
782 if (compare != 0)
783 {
785 return compare;
786 }
787 }
788 return 0;
789}
int16 AttrNumber
Definition: attnum.h:21
#define INVERT_COMPARE_RESULT(var)
Definition: c.h:1105
static int compare(const void *arg1, const void *arg2)
Definition: geqo_pool.c:145
b
int b
Definition: isn.c:74
a
int a
Definition: isn.c:73
int32 SlotNumber
void * arg
uint64_t Datum
Definition: postgres.h:70
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:398

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 598 of file nodeGatherMerge.c.

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 {
616 MinimalTuple tuple;
617
618 tuple = gm_readnext_tuple(gm_state,
619 reader,
620 true,
621 &tuple_buffer->done);
622 if (!tuple)
623 break;
624 tuple_buffer->tuple[i] = tuple;
625 tuple_buffer->nTuples++;
626 }
627}

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().

AltStyle によって変換されたページ (->オリジナル) /