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