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