Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * nodeGatherMerge.c
4 : * Scan a plan in multiple workers, and do order-preserving merge.
5 : *
6 : * Portions Copyright (c) 1996-2026, 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/htup_details.h"
18 : #include "executor/executor.h"
19 : #include "executor/execParallel.h"
20 : #include "executor/nodeGatherMerge.h"
21 : #include "executor/tqueue.h"
22 : #include "lib/binaryheap.h"
23 : #include "miscadmin.h"
24 : #include "optimizer/optimizer.h"
25 : #include "utils/sortsupport.h"
26 :
27 : /*
28 : * When we read tuples from workers, it's a good idea to read several at once
29 : * for efficiency when possible: this minimizes context-switching overhead.
30 : * But reading too many at a time wastes memory without improving performance.
31 : * We'll read up to MAX_TUPLE_STORE tuples (in addition to the first one).
32 : */
33 : #define MAX_TUPLE_STORE 10
34 :
35 : /*
36 : * Pending-tuple array for each worker. This holds additional tuples that
37 : * we were able to fetch from the worker, but can't process yet. In addition,
38 : * this struct holds the "done" flag indicating the worker is known to have
39 : * no more tuples. (We do not use this struct for the leader; we don't keep
40 : * any pending tuples for the leader, and the need_to_scan_locally flag serves
41 : * as its "done" indicator.)
42 : */
43 : typedef struct GMReaderTupleBuffer
44 : {
45 : MinimalTuple *tuple; /* array of length MAX_TUPLE_STORE */
46 : int nTuples; /* number of tuples currently stored */
47 : int readCounter; /* index of next tuple to extract */
48 : bool done; /* true if reader is known exhausted */
49 : } GMReaderTupleBuffer;
50 :
51 : static TupleTableSlot *ExecGatherMerge(PlanState *pstate);
52 : static int32 heap_compare_slots(Datum a, Datum b, void *arg);
53 : static TupleTableSlot *gather_merge_getnext(GatherMergeState *gm_state);
54 : static MinimalTuple gm_readnext_tuple(GatherMergeState *gm_state, int nreader,
55 : bool nowait, bool *done);
56 : static void ExecShutdownGatherMergeWorkers(GatherMergeState *node);
57 : static void gather_merge_setup(GatherMergeState *gm_state);
58 : static void gather_merge_init(GatherMergeState *gm_state);
59 : static void gather_merge_clear_tuples(GatherMergeState *gm_state);
60 : static bool gather_merge_readnext(GatherMergeState *gm_state, int reader,
61 : bool nowait);
62 : static void load_tuple_array(GatherMergeState *gm_state, int reader);
63 :
64 : /* ----------------------------------------------------------------
65 : * ExecInitGather
66 : * ----------------------------------------------------------------
67 : */
68 : GatherMergeState *
69 197 : ExecInitGatherMerge(GatherMerge *node, EState *estate, int eflags)
70 : {
71 : GatherMergeState *gm_state;
72 : Plan *outerNode;
73 : TupleDesc tupDesc;
74 :
75 : /* Gather merge node doesn't have innerPlan node. */
76 : Assert(innerPlan(node) == NULL);
77 :
78 : /*
79 : * create state structure
80 : */
81 197 : gm_state = makeNode(GatherMergeState);
82 197 : gm_state->ps.plan = (Plan *) node;
83 197 : gm_state->ps.state = estate;
84 197 : gm_state->ps.ExecProcNode = ExecGatherMerge;
85 :
86 197 : gm_state->initialized = false;
87 197 : gm_state->gm_initialized = false;
88 197 : gm_state->tuples_needed = -1;
89 :
90 : /*
91 : * Miscellaneous initialization
92 : *
93 : * create expression context for node
94 : */
95 197 : ExecAssignExprContext(estate, &gm_state->ps);
96 :
97 : /*
98 : * GatherMerge doesn't support checking a qual (it's always more efficient
99 : * to do it in the child node).
100 : */
101 : Assert(!node->plan.qual);
102 :
103 : /*
104 : * now initialize outer plan
105 : */
106 197 : outerNode = outerPlan(node);
107 197 : outerPlanState(gm_state) = ExecInitNode(outerNode, estate, eflags);
108 :
109 : /*
110 : * Leader may access ExecProcNode result directly (if
111 : * need_to_scan_locally), or from workers via tuple queue. So we can't
112 : * trivially rely on the slot type being fixed for expressions evaluated
113 : * within this node.
114 : */
115 197 : gm_state->ps.outeropsset = true;
116 197 : gm_state->ps.outeropsfixed = false;
117 :
118 : /*
119 : * Store the tuple descriptor into gather merge state, so we can use it
120 : * while initializing the gather merge slots.
121 : */
122 197 : tupDesc = ExecGetResultType(outerPlanState(gm_state));
123 197 : gm_state->tupDesc = tupDesc;
124 :
125 : /*
126 : * Initialize result type and projection.
127 : */
128 197 : ExecInitResultTypeTL(&gm_state->ps);
129 197 : ExecConditionalAssignProjectionInfo(&gm_state->ps, tupDesc, OUTER_VAR);
130 :
131 : /*
132 : * Without projections result slot type is not trivially known, see
133 : * comment above.
134 : */
135 197 : if (gm_state->ps.ps_ProjInfo == NULL)
136 : {
137 191 : gm_state->ps.resultopsset = true;
138 191 : gm_state->ps.resultopsfixed = false;
139 : }
140 :
141 : /*
142 : * initialize sort-key information
143 : */
144 197 : if (node->numCols)
145 : {
146 : int i;
147 :
148 197 : gm_state->gm_nkeys = node->numCols;
149 197 : gm_state->gm_sortkeys = palloc0_array(SortSupportData, node->numCols);
150 :
151 469 : for (i = 0; i < node->numCols; i++)
152 : {
153 272 : SortSupport sortKey = gm_state->gm_sortkeys + i;
154 :
155 272 : sortKey->ssup_cxt = CurrentMemoryContext;
156 272 : sortKey->ssup_collation = node->collations[i];
157 272 : sortKey->ssup_nulls_first = node->nullsFirst[i];
158 272 : 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 272 : sortKey->abbreviate = false;
165 :
166 272 : PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
167 : }
168 : }
169 :
170 : /* Now allocate the workspace for gather merge */
171 197 : gather_merge_setup(gm_state);
172 :
173 197 : return gm_state;
174 : }
175 :
176 : /* ----------------------------------------------------------------
177 : * ExecGatherMerge(node)
178 : *
179 : * Scans the relation via multiple workers and returns
180 : * the next qualifying tuple.
181 : * ----------------------------------------------------------------
182 : */
183 : static TupleTableSlot *
184 127583 : ExecGatherMerge(PlanState *pstate)
185 : {
186 127583 : GatherMergeState *node = castNode(GatherMergeState, pstate);
187 : TupleTableSlot *slot;
188 : ExprContext *econtext;
189 :
190 127583 : CHECK_FOR_INTERRUPTS();
191 :
192 : /*
193 : * As with Gather, we don't launch workers until this node is actually
194 : * executed.
195 : */
196 127583 : if (!node->initialized)
197 : {
198 96 : EState *estate = node->ps.state;
199 96 : 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 96 : 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 96 : if (!node->pei)
211 81 : node->pei = ExecInitParallelPlan(outerPlanState(node),
212 : estate,
213 : gm->initParam,
214 : gm->num_workers,
215 : node->tuples_needed);
216 : else
217 15 : ExecParallelReinitialize(outerPlanState(node),
218 15 : node->pei,
219 : gm->initParam);
220 :
221 : /* Try to launch workers. */
222 96 : pcxt = node->pei->pcxt;
223 96 : LaunchParallelWorkers(pcxt);
224 : /* We save # workers launched for the benefit of EXPLAIN */
225 96 : node->nworkers_launched = pcxt->nworkers_launched;
226 :
227 : /*
228 : * Count number of workers originally wanted and actually
229 : * launched.
230 : */
231 96 : estate->es_parallel_workers_to_launch += pcxt->nworkers_to_launch;
232 96 : estate->es_parallel_workers_launched += pcxt->nworkers_launched;
233 :
234 : /* Set up tuple queue readers to read the results. */
235 96 : if (pcxt->nworkers_launched > 0)
236 : {
237 90 : ExecParallelCreateReaders(node->pei);
238 : /* Make a working array showing the active readers */
239 90 : node->nreaders = pcxt->nworkers_launched;
240 90 : node->reader = (TupleQueueReader **)
241 90 : palloc(node->nreaders * sizeof(TupleQueueReader *));
242 90 : memcpy(node->reader, node->pei->reader,
243 90 : node->nreaders * sizeof(TupleQueueReader *));
244 : }
245 : else
246 : {
247 : /* No workers? Then never mind. */
248 6 : node->nreaders = 0;
249 6 : node->reader = NULL;
250 : }
251 : }
252 :
253 : /* allow leader to participate if enabled or no choice */
254 96 : if (parallel_leader_participation || node->nreaders == 0)
255 93 : node->need_to_scan_locally = true;
256 96 : 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 127583 : econtext = node->ps.ps_ExprContext;
264 127583 : ResetExprContext(econtext);
265 :
266 : /*
267 : * Get next tuple, either from one of our workers, or by running the plan
268 : * ourselves.
269 : */
270 127583 : slot = gather_merge_getnext(node);
271 127583 : if (TupIsNull(slot))
272 78 : return NULL;
273 :
274 : /* If no projection is required, we're done. */
275 127505 : if (node->ps.ps_ProjInfo == NULL)
276 127505 : return slot;
277 :
278 : /*
279 : * Form the result tuple using ExecProject(), and return it.
280 : */
281 0 : econtext->ecxt_outertuple = slot;
282 0 : return ExecProject(node->ps.ps_ProjInfo);
283 : }
284 :
285 : /* ----------------------------------------------------------------
286 : * ExecEndGatherMerge
287 : *
288 : * frees any storage allocated through C routines.
289 : * ----------------------------------------------------------------
290 : */
291 : void
292 197 : ExecEndGatherMerge(GatherMergeState *node)
293 : {
294 197 : ExecEndNode(outerPlanState(node)); /* let children clean up first */
295 197 : ExecShutdownGatherMerge(node);
296 197 : }
297 :
298 : /* ----------------------------------------------------------------
299 : * ExecShutdownGatherMerge
300 : *
301 : * Destroy the setup for parallel workers including parallel context.
302 : * ----------------------------------------------------------------
303 : */
304 : void
305 278 : ExecShutdownGatherMerge(GatherMergeState *node)
306 : {
307 278 : ExecShutdownGatherMergeWorkers(node);
308 :
309 : /* Now destroy the parallel context. */
310 278 : if (node->pei != NULL)
311 : {
312 81 : ExecParallelCleanup(node->pei);
313 81 : node->pei = NULL;
314 : }
315 278 : }
316 :
317 : /* ----------------------------------------------------------------
318 : * ExecShutdownGatherMergeWorkers
319 : *
320 : * Stop all the parallel workers.
321 : * ----------------------------------------------------------------
322 : */
323 : static void
324 302 : ExecShutdownGatherMergeWorkers(GatherMergeState *node)
325 : {
326 302 : if (node->pei != NULL)
327 96 : ExecParallelFinish(node->pei);
328 :
329 : /* Flush local copy of reader array */
330 302 : if (node->reader)
331 90 : pfree(node->reader);
332 302 : node->reader = NULL;
333 302 : }
334 :
335 : /* ----------------------------------------------------------------
336 : * ExecReScanGatherMerge
337 : *
338 : * Prepare to re-scan the result of a GatherMerge.
339 : * ----------------------------------------------------------------
340 : */
341 : void
342 24 : ExecReScanGatherMerge(GatherMergeState *node)
343 : {
344 24 : GatherMerge *gm = (GatherMerge *) node->ps.plan;
345 24 : PlanState *outerPlan = outerPlanState(node);
346 :
347 : /* Make sure any existing workers are gracefully shut down */
348 24 : ExecShutdownGatherMergeWorkers(node);
349 :
350 : /* Free any unused tuples, so we don't leak memory across rescans */
351 24 : gather_merge_clear_tuples(node);
352 :
353 : /* Mark node so that shared state will be rebuilt at next call */
354 24 : node->initialized = false;
355 24 : 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 24 : if (gm->rescan_param >= 0)
365 24 : 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 24 : if (outerPlan->chgParam == NULL)
379 0 : ExecReScan(outerPlan);
380 24 : }
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
396 197 : gather_merge_setup(GatherMergeState *gm_state)
397 : {
398 197 : GatherMerge *gm = castNode(GatherMerge, gm_state->ps.plan);
399 197 : 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 197 : gm_state->gm_slots = (TupleTableSlot **)
412 197 : palloc0((nreaders + 1) * sizeof(TupleTableSlot *));
413 :
414 : /* Allocate the tuple slot and tuple array for each worker */
415 197 : gm_state->gm_tuple_buffers = (GMReaderTupleBuffer *)
416 197 : palloc0(nreaders * sizeof(GMReaderTupleBuffer));
417 :
418 697 : for (i = 0; i < nreaders; i++)
419 : {
420 : /* Allocate the tuple array with length MAX_TUPLE_STORE */
421 500 : gm_state->gm_tuple_buffers[i].tuple = palloc0_array(MinimalTuple, MAX_TUPLE_STORE);
422 :
423 : /* Initialize tuple slot for worker */
424 500 : gm_state->gm_slots[i + 1] =
425 500 : ExecInitExtraTupleSlot(gm_state->ps.state, gm_state->tupDesc,
426 : &TTSOpsMinimalTuple);
427 : }
428 :
429 : /* Allocate the resources for the merge */
430 197 : gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
431 : heap_compare_slots,
432 : gm_state);
433 197 : }
434 :
435 : /*
436 : * Initialize the Gather Merge.
437 : *
438 : * Reset data structures to ensure they're empty. Then pull at least one
439 : * tuple from leader + each worker (or set its "done" indicator), and set up
440 : * the heap.
441 : */
442 : static void
443 96 : gather_merge_init(GatherMergeState *gm_state)
444 : {
445 96 : int nreaders = gm_state->nreaders;
446 96 : bool nowait = true;
447 : int i;
448 :
449 : /* Assert that gather_merge_setup made enough space */
450 : Assert(nreaders <= castNode(GatherMerge, gm_state->ps.plan)->num_workers);
451 :
452 : /* Reset leader's tuple slot to empty */
453 96 : gm_state->gm_slots[0] = NULL;
454 :
455 : /* Reset the tuple slot and tuple array for each worker */
456 332 : for (i = 0; i < nreaders; i++)
457 : {
458 : /* Reset tuple array to empty */
459 236 : gm_state->gm_tuple_buffers[i].nTuples = 0;
460 236 : gm_state->gm_tuple_buffers[i].readCounter = 0;
461 : /* Reset done flag to not-done */
462 236 : gm_state->gm_tuple_buffers[i].done = false;
463 : /* Ensure output slot is empty */
464 236 : ExecClearTuple(gm_state->gm_slots[i + 1]);
465 : }
466 :
467 : /* Reset binary heap to empty */
468 96 : binaryheap_reset(gm_state->gm_heap);
469 :
470 : /*
471 : * First, try to read a tuple from each worker (including leader) in
472 : * nowait mode. After this, if not all workers were able to produce a
473 : * tuple (or a "done" indication), then re-read from remaining workers,
474 : * this time using wait mode. Add all live readers (those producing at
475 : * least one tuple) to the heap.
476 : */
477 183 : reread:
478 832 : for (i = 0; i <= nreaders; i++)
479 : {
480 649 : CHECK_FOR_INTERRUPTS();
481 :
482 : /* skip this source if already known done */
483 1115 : if ((i == 0) ? gm_state->need_to_scan_locally :
484 466 : !gm_state->gm_tuple_buffers[i - 1].done)
485 : {
486 633 : if (TupIsNull(gm_state->gm_slots[i]))
487 : {
488 : /* Don't have a tuple yet, try to get one */
489 669 : if (gather_merge_readnext(gm_state, i, nowait))
490 126 : binaryheap_add_unordered(gm_state->gm_heap,
491 : Int32GetDatum(i));
492 : }
493 : else
494 : {
495 : /*
496 : * We already got at least one tuple from this worker, but
497 : * might as well see if it has any more ready by now.
498 : */
499 90 : load_tuple_array(gm_state, i);
500 : }
501 : }
502 : }
503 :
504 : /* need not recheck leader, since nowait doesn't matter for it */
505 427 : for (i = 1; i <= nreaders; i++)
506 : {
507 331 : if (!gm_state->gm_tuple_buffers[i - 1].done &&
508 116 : TupIsNull(gm_state->gm_slots[i]))
509 : {
510 87 : nowait = false;
511 87 : goto reread;
512 : }
513 : }
514 :
515 : /* Now heapify the heap. */
516 96 : binaryheap_build(gm_state->gm_heap);
517 :
518 96 : gm_state->gm_initialized = true;
519 96 : }
520 :
521 : /*
522 : * Clear out the tuple table slot, and any unused pending tuples,
523 : * for each gather merge input.
524 : */
525 : static void
526 102 : gather_merge_clear_tuples(GatherMergeState *gm_state)
527 : {
528 : int i;
529 :
530 368 : for (i = 0; i < gm_state->nreaders; i++)
531 : {
532 266 : GMReaderTupleBuffer *tuple_buffer = &gm_state->gm_tuple_buffers[i];
533 :
534 266 : while (tuple_buffer->readCounter < tuple_buffer->nTuples)
535 0 : pfree(tuple_buffer->tuple[tuple_buffer->readCounter++]);
536 :
537 266 : ExecClearTuple(gm_state->gm_slots[i + 1]);
538 : }
539 102 : }
540 :
541 : /*
542 : * Read the next tuple for gather merge.
543 : *
544 : * Fetch the sorted tuple out of the heap.
545 : */
546 : static TupleTableSlot *
547 127583 : gather_merge_getnext(GatherMergeState *gm_state)
548 : {
549 : int i;
550 :
551 127583 : if (!gm_state->gm_initialized)
552 : {
553 : /*
554 : * First time through: pull the first tuple from each participant, and
555 : * set up the heap.
556 : */
557 96 : gather_merge_init(gm_state);
558 : }
559 : else
560 : {
561 : /*
562 : * Otherwise, pull the next tuple from whichever participant we
563 : * returned from last time, and reinsert that participant's index into
564 : * the heap, because it might now compare differently against the
565 : * other elements of the heap.
566 : */
567 127487 : i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
568 :
569 127487 : if (gather_merge_readnext(gm_state, i, false))
570 127379 : binaryheap_replace_first(gm_state->gm_heap, Int32GetDatum(i));
571 : else
572 : {
573 : /* reader exhausted, remove it from heap */
574 108 : (void) binaryheap_remove_first(gm_state->gm_heap);
575 : }
576 : }
577 :
578 127583 : if (binaryheap_empty(gm_state->gm_heap))
579 : {
580 : /* All the queues are exhausted, and so is the heap */
581 78 : gather_merge_clear_tuples(gm_state);
582 78 : return NULL;
583 : }
584 : else
585 : {
586 : /* Return next tuple from whichever participant has the leading one */
587 127505 : i = DatumGetInt32(binaryheap_first(gm_state->gm_heap));
588 127505 : return gm_state->gm_slots[i];
589 : }
590 : }
591 :
592 : /*
593 : * Read tuple(s) for given reader in nowait mode, and load into its tuple
594 : * array, until we have MAX_TUPLE_STORE of them or would have to block.
595 : */
596 : static void
597 498 : load_tuple_array(GatherMergeState *gm_state, int reader)
598 : {
599 : GMReaderTupleBuffer *tuple_buffer;
600 : int i;
601 :
602 : /* Don't do anything if this is the leader. */
603 498 : if (reader == 0)
604 84 : return;
605 :
606 414 : tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
607 :
608 : /* If there's nothing in the array, reset the counters to zero. */
609 414 : if (tuple_buffer->nTuples == tuple_buffer->readCounter)
610 408 : tuple_buffer->nTuples = tuple_buffer->readCounter = 0;
611 :
612 : /* Try to fill additional slots in the array. */
613 4320 : for (i = tuple_buffer->nTuples; i < MAX_TUPLE_STORE; i++)
614 : {
615 : MinimalTuple tuple;
616 :
617 3940 : tuple = gm_readnext_tuple(gm_state,
618 : reader,
619 : true,
620 : &tuple_buffer->done);
621 3940 : if (!tuple)
622 34 : break;
623 3906 : tuple_buffer->tuple[i] = tuple;
624 3906 : tuple_buffer->nTuples++;
625 : }
626 : }
627 :
628 : /*
629 : * Store the next tuple for a given reader into the appropriate slot.
630 : *
631 : * Returns true if successful, false if not (either reader is exhausted,
632 : * or we didn't want to wait for a tuple). Sets done flag if reader
633 : * is found to be exhausted.
634 : */
635 : static bool
636 128030 : gather_merge_readnext(GatherMergeState *gm_state, int reader, bool nowait)
637 : {
638 : GMReaderTupleBuffer *tuple_buffer;
639 : MinimalTuple tup;
640 :
641 : /*
642 : * If we're being asked to generate a tuple from the leader, then we just
643 : * call ExecProcNode as normal to produce one.
644 : */
645 128030 : if (reader == 0)
646 : {
647 123266 : if (gm_state->need_to_scan_locally)
648 : {
649 123266 : PlanState *outerPlan = outerPlanState(gm_state);
650 : TupleTableSlot *outerTupleSlot;
651 123266 : EState *estate = gm_state->ps.state;
652 :
653 : /* Install our DSA area while executing the plan. */
654 123266 : estate->es_query_dsa = gm_state->pei ? gm_state->pei->area : NULL;
655 123266 : outerTupleSlot = ExecProcNode(outerPlan);
656 123266 : estate->es_query_dsa = NULL;
657 :
658 123266 : if (!TupIsNull(outerTupleSlot))
659 : {
660 123191 : gm_state->gm_slots[0] = outerTupleSlot;
661 123191 : return true;
662 : }
663 : /* need_to_scan_locally serves as "done" flag for leader */
664 75 : gm_state->need_to_scan_locally = false;
665 : }
666 75 : return false;
667 : }
668 :
669 : /* Otherwise, check the state of the relevant tuple buffer. */
670 4764 : tuple_buffer = &gm_state->gm_tuple_buffers[reader - 1];
671 :
672 4764 : if (tuple_buffer->nTuples > tuple_buffer->readCounter)
673 : {
674 : /* Return any tuple previously read that is still buffered. */
675 3906 : tup = tuple_buffer->tuple[tuple_buffer->readCounter++];
676 : }
677 858 : else if (tuple_buffer->done)
678 : {
679 : /* Reader is known to be exhausted. */
680 32 : return false;
681 : }
682 : else
683 : {
684 : /* Read and buffer next tuple. */
685 826 : tup = gm_readnext_tuple(gm_state,
686 : reader,
687 : nowait,
688 : &tuple_buffer->done);
689 826 : if (!tup)
690 418 : return false;
691 :
692 : /*
693 : * Attempt to read more tuples in nowait mode and store them in the
694 : * pending-tuple array for the reader.
695 : */
696 408 : load_tuple_array(gm_state, reader);
697 : }
698 :
699 : Assert(tup);
700 :
701 : /* Build the TupleTableSlot for the given tuple */
702 4314 : ExecStoreMinimalTuple(tup, /* tuple to store */
703 4314 : gm_state->gm_slots[reader], /* slot in which to
704 : * store the tuple */
705 : true); /* pfree tuple when done with it */
706 :
707 4314 : return true;
708 : }
709 :
710 : /*
711 : * Attempt to read a tuple from given worker.
712 : */
713 : static MinimalTuple
714 4766 : gm_readnext_tuple(GatherMergeState *gm_state, int nreader, bool nowait,
715 : bool *done)
716 : {
717 : TupleQueueReader *reader;
718 : MinimalTuple tup;
719 :
720 : /* Check for async events, particularly messages from workers. */
721 4766 : CHECK_FOR_INTERRUPTS();
722 :
723 : /*
724 : * Attempt to read a tuple.
725 : *
726 : * Note that TupleQueueReaderNext will just return NULL for a worker which
727 : * fails to initialize. We'll treat that worker as having produced no
728 : * tuples; WaitForParallelWorkersToFinish will error out when we get
729 : * there.
730 : */
731 4766 : reader = gm_state->reader[nreader - 1];
732 4766 : tup = TupleQueueReaderNext(reader, nowait, done);
733 :
734 : /*
735 : * Since we'll be buffering these across multiple calls, we need to make a
736 : * copy.
737 : */
738 4766 : return tup ? heap_copy_minimal_tuple(tup, 0) : NULL;
739 : }
740 :
741 : /*
742 : * We have one slot for each item in the heap array. We use SlotNumber
743 : * to store slot indexes. This doesn't actually provide any formal
744 : * type-safety, but it makes the code more self-documenting.
745 : */
746 : typedef int32 SlotNumber;
747 :
748 : /*
749 : * Compare the tuples in the two given slots.
750 : */
751 : static int32
752 21255 : heap_compare_slots(Datum a, Datum b, void *arg)
753 : {
754 21255 : GatherMergeState *node = (GatherMergeState *) arg;
755 21255 : SlotNumber slot1 = DatumGetInt32(a);
756 21255 : SlotNumber slot2 = DatumGetInt32(b);
757 :
758 21255 : TupleTableSlot *s1 = node->gm_slots[slot1];
759 21255 : TupleTableSlot *s2 = node->gm_slots[slot2];
760 : int nkey;
761 :
762 : Assert(!TupIsNull(s1));
763 : Assert(!TupIsNull(s2));
764 :
765 21922 : for (nkey = 0; nkey < node->gm_nkeys; nkey++)
766 : {
767 21255 : SortSupport sortKey = node->gm_sortkeys + nkey;
768 21255 : AttrNumber attno = sortKey->ssup_attno;
769 : Datum datum1,
770 : datum2;
771 : bool isNull1,
772 : isNull2;
773 : int compare;
774 :
775 21255 : datum1 = slot_getattr(s1, attno, &isNull1);
776 21255 : datum2 = slot_getattr(s2, attno, &isNull2);
777 :
778 21255 : compare = ApplySortComparator(datum1, isNull1,
779 : datum2, isNull2,
780 : sortKey);
781 21255 : if (compare != 0)
782 : {
783 20588 : INVERT_COMPARE_RESULT(compare);
784 20588 : return compare;
785 : }
786 : }
787 667 : return 0;
788 : }
|
