1 /*-------------------------------------------------------------------------
2  *
3  * nodeGatherMerge.c
4  *		Scan a plan in multiple workers, and do order-preserving merge.
5  *
6  * Portions Copyright (c) 1996-2021, 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"
21 #include "executor/nodeGatherMerge.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 	MinimalTuple *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 */
52 } GMReaderTupleBuffer;
53 
54 static TupleTableSlot *ExecGatherMerge(PlanState *pstate);
55 static int32 heap_compare_slots(Datum a, Datum b, void *arg);
56 static TupleTableSlot *gather_merge_getnext(GatherMergeState *gm_state);
57 static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader,
58 									  bool nowait, bool *done);
59 static void ExecShutdownGatherMergeWorkers(GatherMergeState *node);
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  */
71 GatherMergeState *
ExecInitGatherMerge(GatherMerge * node,EState * estate,int eflags)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 
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 }
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 *
ExecGatherMerge(PlanState * pstate)188 ExecGatherMerge(PlanState *pstate)
189 {
190 	GatherMergeState *node = castNode(GatherMergeState, pstate);
191 	TupleTableSlot *slot;
192 	ExprContext *econtext;
193 
194 	CHECK_FOR_INTERRUPTS();
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
221 				ExecParallelReinitialize(node->ps.lefttree,
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 			{
234 				ExecParallelCreateReaders(node->pei);
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
ExecEndGatherMerge(GatherMergeState * node)289 ExecEndGatherMerge(GatherMergeState *node)
290 {
291 	ExecEndNode(outerPlanState(node));	/* let children clean up first */
292 	ExecShutdownGatherMerge(node);
293 	ExecFreeExprContext(&node->ps);
294 	if (node->ps.ps_ResultTupleSlot)
295 		ExecClearTuple(node->ps.ps_ResultTupleSlot);
296 }
297 
298 /* ----------------------------------------------------------------
299  *		ExecShutdownGatherMerge
300  *
301  *		Destroy the setup for parallel workers including parallel context.
302  * ----------------------------------------------------------------
303  */
304 void
ExecShutdownGatherMerge(GatherMergeState * node)305 ExecShutdownGatherMerge(GatherMergeState *node)
306 {
307 	ExecShutdownGatherMergeWorkers(node);
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
ExecShutdownGatherMergeWorkers(GatherMergeState * node)324 ExecShutdownGatherMergeWorkers(GatherMergeState *node)
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
ExecReScanGatherMerge(GatherMergeState * node)342 ExecReScanGatherMerge(GatherMergeState *node)
343 {
344 	GatherMerge *gm = (GatherMerge *) node->ps.plan;
345 	PlanState  *outerPlan = outerPlanState(node);
346 
347 	/* Make sure any existing workers are gracefully shut down */
348 	ExecShutdownGatherMergeWorkers(node);
349 
350 	/* Free any unused tuples, so we don't leak memory across rescans */
351 	gather_merge_clear_tuples(node);
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)
379 		ExecReScan(outerPlan);
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
gather_merge_setup(GatherMergeState * gm_state)396 gather_merge_setup(GatherMergeState *gm_state)
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 =
422 			(MinimalTuple *) palloc0(sizeof(MinimalTuple) * MAX_TUPLE_STORE);
423 
424 		/* Initialize tuple slot for worker */
425 		gm_state->gm_slots[i + 1] =
426 			ExecInitExtraTupleSlot(gm_state->ps.state, gm_state->tupDesc,
427 								   &TTSOpsMinimalTuple);
428 	}
429 
430 	/* Allocate the resources for the merge */
431 	gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
432 											heap_compare_slots,
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
gather_merge_init(GatherMergeState * gm_state)444 gather_merge_init(GatherMergeState *gm_state)
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 	{
481 		CHECK_FOR_INTERRUPTS();
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))
491 					binaryheap_add_unordered(gm_state->gm_heap,
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
gather_merge_clear_tuples(GatherMergeState * gm_state)527 gather_merge_clear_tuples(GatherMergeState *gm_state)
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 }
541 
542 /*
543  * Read the next tuple for gather merge.
544  *
545  * Fetch the sorted tuple out of the heap.
546  */
547 static TupleTableSlot *
gather_merge_getnext(GatherMergeState * gm_state)548 gather_merge_getnext(GatherMergeState *gm_state)
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))
571 			binaryheap_replace_first(gm_state->gm_heap, Int32GetDatum(i));
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
load_tuple_array(GatherMergeState * gm_state,int reader)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 	{
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 }
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
gather_merge_readnext(GatherMergeState * gm_state,int reader,bool nowait)637 gather_merge_readnext(GatherMergeState *gm_state, int reader, bool nowait)
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 		{
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 (!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 }
710 
711 /*
712  * Attempt to read a tuple from given worker.
713  */
714 static MinimalTuple
gm_readnext_tuple(GatherMergeState * gm_state,int nreader,bool nowait,bool * done)715 gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait,
716 				  bool *done)
717 {
718 	TupleQueueReader *reader;
719 	MinimalTuple tup;
720 
721 	/* Check for async events, particularly messages from workers. */
722 	CHECK_FOR_INTERRUPTS();
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) : NULL;
740 }
741 
742 /*
743  * We have one slot for each item in the heap array.  We use SlotNumber
744  * to store slot indexes.  This doesn't actually provide any formal
745  * type-safety, but it makes the code more self-documenting.
746  */
747 typedef int32 SlotNumber;
748 
749 /*
750  * Compare the tuples in the two given slots.
751  */
752 static int32
heap_compare_slots(Datum a,Datum b,void * arg)753 heap_compare_slots(Datum a, Datum b, void *arg)
754 {
755 	GatherMergeState *node = (GatherMergeState *) arg;
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 
763 	Assert(!TupIsNull(s1));
764 	Assert(!TupIsNull(s2));
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 		{
784 			INVERT_COMPARE_RESULT(compare);
785 			return compare;
786 		}
787 	}
788 	return 0;
789 }
790