Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeGather.c
4 : : * Support routines for scanning a plan via multiple workers.
5 : : *
6 : : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * A Gather executor launches parallel workers to run multiple copies of a
10 : : * plan. It can also run the plan itself, if the workers are not available
11 : : * or have not started up yet. It then merges all of the results it produces
12 : : * and the results from the workers into a single output stream. Therefore,
13 : : * it will normally be used with a plan where running multiple copies of the
14 : : * same plan does not produce duplicate output, such as parallel-aware
15 : : * SeqScan.
16 : : *
17 : : * Alternatively, a Gather node can be configured to use just one worker
18 : : * and the single-copy flag can be set. In this case, the Gather node will
19 : : * run the plan in one worker and will not execute the plan itself. In
20 : : * this case, it simply returns whatever tuples were returned by the worker.
21 : : * If a worker cannot be obtained, then it will run the plan itself and
22 : : * return the results. Therefore, a plan used with a single-copy Gather
23 : : * node need not be parallel-aware.
24 : : *
25 : : * IDENTIFICATION
26 : : * src/backend/executor/nodeGather.c
27 : : *
28 : : *-------------------------------------------------------------------------
29 : : */
30 : :
31 : : #include "postgres.h"
32 : :
33 : : #include "executor/execParallel.h"
34 : : #include "executor/executor.h"
35 : : #include "executor/nodeGather.h"
36 : : #include "executor/tqueue.h"
37 : : #include "miscadmin.h"
38 : : #include "optimizer/optimizer.h"
39 : : #include "utils/wait_event.h"
40 : :
41 : :
42 : : static TupleTableSlot *ExecGather(PlanState *pstate);
43 : : static TupleTableSlot *gather_getnext(GatherState *gatherstate);
44 : : static MinimalTuple gather_readnext(GatherState *gatherstate);
45 : : static void ExecShutdownGatherWorkers(GatherState *node);
46 : :
47 : :
48 : : /* ----------------------------------------------------------------
49 : : * ExecInitGather
50 : : * ----------------------------------------------------------------
51 : : */
52 : : GatherState *
3629 rhaas@postgresql.org 53 :CBC 544 : ExecInitGather(Gather *node, EState *estate, int eflags)
54 : : {
55 : : GatherState *gatherstate;
56 : : Plan *outerNode;
57 : : TupleDesc tupDesc;
58 : :
59 : : /* Gather node doesn't have innerPlan node. */
60 [ - + ]: 544 : Assert(innerPlan(node) == NULL);
61 : :
62 : : /*
63 : : * create state structure
64 : : */
65 : 544 : gatherstate = makeNode(GatherState);
66 : 544 : gatherstate->ps.plan = (Plan *) node;
67 : 544 : gatherstate->ps.state = estate;
2973 andres@anarazel.de 68 : 544 : gatherstate->ps.ExecProcNode = ExecGather;
69 : :
2929 tgl@sss.pgh.pa.us 70 : 544 : gatherstate->initialized = false;
2852 rhaas@postgresql.org 71 : 544 : gatherstate->need_to_scan_locally =
72 [ + + + + ]: 544 : !node->single_copy && parallel_leader_participation;
2930 73 : 544 : gatherstate->tuples_needed = -1;
74 : :
75 : : /*
76 : : * Miscellaneous initialization
77 : : *
78 : : * create expression context for node
79 : : */
3629 80 : 544 : ExecAssignExprContext(estate, &gatherstate->ps);
81 : :
82 : : /*
83 : : * now initialize outer plan
84 : : */
3601 85 : 544 : outerNode = outerPlan(node);
86 : 544 : outerPlanState(gatherstate) = ExecInitNode(outerNode, estate, eflags);
2759 andres@anarazel.de 87 : 544 : tupDesc = ExecGetResultType(outerPlanState(gatherstate));
88 : :
89 : : /*
90 : : * Leader may access ExecProcNode result directly (if
91 : : * need_to_scan_locally), or from workers via tuple queue. So we can't
92 : : * trivially rely on the slot type being fixed for expressions evaluated
93 : : * within this node.
94 : : */
2487 95 : 544 : gatherstate->ps.outeropsset = true;
96 : 544 : gatherstate->ps.outeropsfixed = false;
97 : :
98 : : /*
99 : : * Initialize result type and projection.
100 : : */
2493 101 : 544 : ExecInitResultTypeTL(&gatherstate->ps);
2759 102 : 544 : ExecConditionalAssignProjectionInfo(&gatherstate->ps, tupDesc, OUTER_VAR);
103 : :
104 : : /*
105 : : * Without projections result slot type is not trivially known, see
106 : : * comment above.
107 : : */
2487 108 [ + + ]: 544 : if (gatherstate->ps.ps_ProjInfo == NULL)
109 : : {
110 : 505 : gatherstate->ps.resultopsset = true;
111 : 505 : gatherstate->ps.resultopsfixed = false;
112 : : }
113 : :
114 : : /*
115 : : * Initialize funnel slot to same tuple descriptor as outer plan.
116 : : */
117 : 544 : gatherstate->funnel_slot = ExecInitExtraTupleSlot(estate, tupDesc,
118 : : &TTSOpsMinimalTuple);
119 : :
120 : : /*
121 : : * Gather doesn't support checking a qual (it's always more efficient to
122 : : * do it in the child node).
123 : : */
2759 124 [ - + ]: 544 : Assert(!node->plan.qual);
125 : :
3629 rhaas@postgresql.org 126 : 544 : return gatherstate;
127 : : }
128 : :
129 : : /* ----------------------------------------------------------------
130 : : * ExecGather(node)
131 : : *
132 : : * Scans the relation via multiple workers and returns
133 : : * the next qualifying tuple.
134 : : * ----------------------------------------------------------------
135 : : */
136 : : static TupleTableSlot *
2973 andres@anarazel.de 137 : 1594351 : ExecGather(PlanState *pstate)
138 : : {
139 : 1594351 : GatherState *node = castNode(GatherState, pstate);
140 : : TupleTableSlot *slot;
141 : : ExprContext *econtext;
142 : :
2965 143 [ + + ]: 1594351 : CHECK_FOR_INTERRUPTS();
144 : :
145 : : /*
146 : : * Initialize the parallel context and workers on first execution. We do
147 : : * this on first execution rather than during node initialization, as it
148 : : * needs to allocate a large dynamic segment, so it is better to do it
149 : : * only if it is really needed.
150 : : */
3613 rhaas@postgresql.org 151 [ + + ]: 1594351 : if (!node->initialized)
152 : : {
3629 153 : 429 : EState *estate = node->ps.state;
3613 154 : 429 : Gather *gather = (Gather *) node->ps.plan;
155 : :
156 : : /*
157 : : * Sometimes we might have to run without parallelism; but if parallel
158 : : * mode is active then we can try to fire up some workers.
159 : : */
2871 160 [ + - + + ]: 429 : if (gather->num_workers > 0 && estate->es_use_parallel_mode)
161 : : {
162 : : ParallelContext *pcxt;
163 : :
164 : : /* Initialize, or re-initialize, shared state needed by workers. */
3599 165 [ + + ]: 408 : if (!node->pei)
1157 tgl@sss.pgh.pa.us 166 : 294 : node->pei = ExecInitParallelPlan(outerPlanState(node),
167 : : estate,
168 : : gather->initParam,
169 : : gather->num_workers,
170 : : node->tuples_needed);
171 : : else
172 : 114 : ExecParallelReinitialize(outerPlanState(node),
2851 rhaas@postgresql.org 173 : 114 : node->pei,
174 : : gather->initParam);
175 : :
176 : : /*
177 : : * Register backend workers. We might not get as many as we
178 : : * requested, or indeed any at all.
179 : : */
3592 180 : 408 : pcxt = node->pei->pcxt;
181 : 408 : LaunchParallelWorkers(pcxt);
182 : : /* We save # workers launched for the benefit of EXPLAIN */
3431 183 : 408 : node->nworkers_launched = pcxt->nworkers_launched;
184 : :
185 : : /*
186 : : * Count number of workers originally wanted and actually
187 : : * launched.
188 : : */
332 michael@paquier.xyz 189 : 408 : estate->es_parallel_workers_to_launch += pcxt->nworkers_to_launch;
190 : 408 : estate->es_parallel_workers_launched += pcxt->nworkers_launched;
191 : :
192 : : /* Set up tuple queue readers to read the results. */
3473 rhaas@postgresql.org 193 [ + + ]: 408 : if (pcxt->nworkers_launched > 0)
194 : : {
2914 andres@anarazel.de 195 : 405 : ExecParallelCreateReaders(node->pei);
196 : : /* Make a working array showing the active readers */
2927 tgl@sss.pgh.pa.us 197 : 405 : node->nreaders = pcxt->nworkers_launched;
198 : 405 : node->reader = (TupleQueueReader **)
199 : 405 : palloc(node->nreaders * sizeof(TupleQueueReader *));
200 : 405 : memcpy(node->reader, node->pei->reader,
201 : 405 : node->nreaders * sizeof(TupleQueueReader *));
202 : : }
203 : : else
204 : : {
205 : : /* No workers? Then never mind. */
206 : 3 : node->nreaders = 0;
207 : 3 : node->reader = NULL;
208 : : }
209 : 408 : node->nextreader = 0;
210 : : }
211 : :
212 : : /* Run plan locally if no workers or enabled and not single-copy. */
213 : 858 : node->need_to_scan_locally = (node->nreaders == 0)
2852 rhaas@postgresql.org 214 [ + + + + : 429 : || (!gather->single_copy && parallel_leader_participation);
+ + ]
3613 215 : 429 : node->initialized = true;
216 : : }
217 : :
218 : : /*
219 : : * Reset per-tuple memory context to free any expression evaluation
220 : : * storage allocated in the previous tuple cycle.
221 : : */
3601 222 : 1594351 : econtext = node->ps.ps_ExprContext;
223 : 1594351 : ResetExprContext(econtext);
224 : :
225 : : /*
226 : : * Get next tuple, either from one of our workers, or by running the plan
227 : : * ourselves.
228 : : */
3149 tgl@sss.pgh.pa.us 229 : 1594351 : slot = gather_getnext(node);
230 [ + - + + ]: 1594345 : if (TupIsNull(slot))
231 : 405 : return NULL;
232 : :
233 : : /* If no projection is required, we're done. */
2842 rhaas@postgresql.org 234 [ + + ]: 1593940 : if (node->ps.ps_ProjInfo == NULL)
235 : 1593922 : return slot;
236 : :
237 : : /*
238 : : * Form the result tuple using ExecProject(), and return it.
239 : : */
3149 tgl@sss.pgh.pa.us 240 : 18 : econtext->ecxt_outertuple = slot;
241 : 18 : return ExecProject(node->ps.ps_ProjInfo);
242 : : }
243 : :
244 : : /* ----------------------------------------------------------------
245 : : * ExecEndGather
246 : : *
247 : : * frees any storage allocated through C routines.
248 : : * ----------------------------------------------------------------
249 : : */
250 : : void
3629 rhaas@postgresql.org 251 : 538 : ExecEndGather(GatherState *node)
252 : : {
3034 bruce@momjian.us 253 : 538 : ExecEndNode(outerPlanState(node)); /* let children clean up first */
3629 rhaas@postgresql.org 254 : 538 : ExecShutdownGather(node);
255 : 538 : }
256 : :
257 : : /*
258 : : * Read the next tuple. We might fetch a tuple from one of the tuple queues
259 : : * using gather_readnext, or if no tuple queue contains a tuple and the
260 : : * single_copy flag is not set, we might generate one locally instead.
261 : : */
262 : : static TupleTableSlot *
263 : 1594351 : gather_getnext(GatherState *gatherstate)
264 : : {
3601 265 : 1594351 : PlanState *outerPlan = outerPlanState(gatherstate);
266 : : TupleTableSlot *outerTupleSlot;
267 : 1594351 : TupleTableSlot *fslot = gatherstate->funnel_slot;
268 : : MinimalTuple tup;
269 : :
2927 tgl@sss.pgh.pa.us 270 [ + + + + ]: 3189326 : while (gatherstate->nreaders > 0 || gatherstate->need_to_scan_locally)
271 : : {
2965 andres@anarazel.de 272 [ - + ]: 1594570 : CHECK_FOR_INTERRUPTS();
273 : :
2927 tgl@sss.pgh.pa.us 274 [ + + ]: 1594570 : if (gatherstate->nreaders > 0)
275 : : {
3592 rhaas@postgresql.org 276 : 1439021 : tup = gather_readnext(gatherstate);
277 : :
3629 278 [ + + ]: 1439015 : if (HeapTupleIsValid(tup))
279 : : {
1578 tgl@sss.pgh.pa.us 280 : 1043713 : ExecStoreMinimalTuple(tup, /* tuple to store */
281 : : fslot, /* slot to store the tuple */
282 : : false); /* don't pfree tuple */
3601 rhaas@postgresql.org 283 : 1043713 : return fslot;
284 : : }
285 : : }
286 : :
3629 287 [ + + ]: 550851 : if (gatherstate->need_to_scan_locally)
288 : : {
2690 tgl@sss.pgh.pa.us 289 : 550511 : EState *estate = gatherstate->ps.state;
290 : :
291 : : /* Install our DSA area while executing the plan. */
2819 rhaas@postgresql.org 292 : 550511 : estate->es_query_dsa =
293 [ + + ]: 550511 : gatherstate->pei ? gatherstate->pei->area : NULL;
3629 294 : 550511 : outerTupleSlot = ExecProcNode(outerPlan);
2819 295 : 550511 : estate->es_query_dsa = NULL;
296 : :
3629 297 [ + + + + ]: 550511 : if (!TupIsNull(outerTupleSlot))
298 : 550227 : return outerTupleSlot;
299 : :
300 : 284 : gatherstate->need_to_scan_locally = false;
301 : : }
302 : : }
303 : :
3601 304 : 405 : return ExecClearTuple(fslot);
305 : : }
306 : :
307 : : /*
308 : : * Attempt to read a tuple from one of our parallel workers.
309 : : */
310 : : static MinimalTuple
3592 311 : 1439021 : gather_readnext(GatherState *gatherstate)
312 : : {
3326 tgl@sss.pgh.pa.us 313 : 1439021 : int nvisited = 0;
314 : :
315 : : for (;;)
3592 rhaas@postgresql.org 316 : 1055314 : {
317 : : TupleQueueReader *reader;
318 : : MinimalTuple tup;
319 : : bool readerdone;
320 : :
321 : : /* Check for async events, particularly messages from workers. */
3323 tgl@sss.pgh.pa.us 322 [ + + ]: 2494335 : CHECK_FOR_INTERRUPTS();
323 : :
324 : : /*
325 : : * Attempt to read a tuple, but don't block if none is available.
326 : : *
327 : : * Note that TupleQueueReaderNext will just return NULL for a worker
328 : : * which fails to initialize. We'll treat that worker as having
329 : : * produced no tuples; WaitForParallelWorkersToFinish will error out
330 : : * when we get there.
331 : : */
3197 rhaas@postgresql.org 332 [ - + ]: 2494329 : Assert(gatherstate->nextreader < gatherstate->nreaders);
3592 333 : 2494329 : reader = gatherstate->reader[gatherstate->nextreader];
334 : 2494329 : tup = TupleQueueReaderNext(reader, true, &readerdone);
335 : :
336 : : /*
337 : : * If this reader is done, remove it from our working array of active
338 : : * readers. If all readers are done, we're outta here.
339 : : */
340 [ + + ]: 2494329 : if (readerdone)
341 : : {
3326 tgl@sss.pgh.pa.us 342 [ - + ]: 1054 : Assert(!tup);
3592 rhaas@postgresql.org 343 : 1054 : --gatherstate->nreaders;
344 [ + + ]: 1054 : if (gatherstate->nreaders == 0)
345 : : {
2591 akapila@postgresql.o 346 : 399 : ExecShutdownGatherWorkers(gatherstate);
3592 rhaas@postgresql.org 347 : 1439015 : return NULL;
348 : : }
3326 tgl@sss.pgh.pa.us 349 : 655 : memmove(&gatherstate->reader[gatherstate->nextreader],
350 : 655 : &gatherstate->reader[gatherstate->nextreader + 1],
351 : : sizeof(TupleQueueReader *)
352 : 655 : * (gatherstate->nreaders - gatherstate->nextreader));
353 [ + + ]: 655 : if (gatherstate->nextreader >= gatherstate->nreaders)
354 : 220 : gatherstate->nextreader = 0;
3592 rhaas@postgresql.org 355 : 655 : continue;
356 : : }
357 : :
358 : : /* If we got a tuple, return it. */
359 [ + + ]: 2493275 : if (tup)
360 : 1043713 : return tup;
361 : :
362 : : /*
363 : : * Advance nextreader pointer in round-robin fashion. Note that we
364 : : * only reach this code if we weren't able to get a tuple from the
365 : : * current worker. We used to advance the nextreader pointer after
366 : : * every tuple, but it turns out to be much more efficient to keep
367 : : * reading from the same queue until that would require blocking.
368 : : */
3326 tgl@sss.pgh.pa.us 369 : 1449562 : gatherstate->nextreader++;
370 [ + + ]: 1449562 : if (gatherstate->nextreader >= gatherstate->nreaders)
371 : 398150 : gatherstate->nextreader = 0;
372 : :
373 : : /* Have we visited every (surviving) TupleQueueReader? */
374 : 1449562 : nvisited++;
375 [ + + ]: 1449562 : if (nvisited >= gatherstate->nreaders)
376 : : {
377 : : /*
378 : : * If (still) running plan locally, return NULL so caller can
379 : : * generate another tuple from the local copy of the plan.
380 : : */
3592 rhaas@postgresql.org 381 [ + + ]: 397861 : if (gatherstate->need_to_scan_locally)
382 : 394903 : return NULL;
383 : :
384 : : /* Nothing to do except wait for developments. */
2479 tmunro@postgresql.or 385 : 2958 : (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
386 : : WAIT_EVENT_EXECUTE_GATHER);
3592 rhaas@postgresql.org 387 : 2958 : ResetLatch(MyLatch);
3326 tgl@sss.pgh.pa.us 388 : 2958 : nvisited = 0;
389 : : }
390 : : }
391 : : }
392 : :
393 : : /* ----------------------------------------------------------------
394 : : * ExecShutdownGatherWorkers
395 : : *
396 : : * Stop all the parallel workers.
397 : : * ----------------------------------------------------------------
398 : : */
399 : : static void
3599 rhaas@postgresql.org 400 : 1402 : ExecShutdownGatherWorkers(GatherState *node)
401 : : {
3613 402 [ + + ]: 1402 : if (node->pei != NULL)
403 : 801 : ExecParallelFinish(node->pei);
404 : :
405 : : /* Flush local copy of reader array */
2927 tgl@sss.pgh.pa.us 406 [ + + ]: 1402 : if (node->reader)
407 : 399 : pfree(node->reader);
408 : 1402 : node->reader = NULL;
3599 rhaas@postgresql.org 409 : 1402 : }
410 : :
411 : : /* ----------------------------------------------------------------
412 : : * ExecShutdownGather
413 : : *
414 : : * Destroy the setup for parallel workers including parallel context.
415 : : * ----------------------------------------------------------------
416 : : */
417 : : void
418 : 853 : ExecShutdownGather(GatherState *node)
419 : : {
420 : 853 : ExecShutdownGatherWorkers(node);
421 : :
422 : : /* Now destroy the parallel context. */
423 [ + + ]: 853 : if (node->pei != NULL)
424 : : {
3613 425 : 288 : ExecParallelCleanup(node->pei);
426 : 288 : node->pei = NULL;
427 : : }
3629 428 : 853 : }
429 : :
430 : : /* ----------------------------------------------------------------
431 : : * Join Support
432 : : * ----------------------------------------------------------------
433 : : */
434 : :
435 : : /* ----------------------------------------------------------------
436 : : * ExecReScanGather
437 : : *
438 : : * Prepare to re-scan the result of a Gather.
439 : : * ----------------------------------------------------------------
440 : : */
441 : : void
442 : 150 : ExecReScanGather(GatherState *node)
443 : : {
2929 tgl@sss.pgh.pa.us 444 : 150 : Gather *gather = (Gather *) node->ps.plan;
445 : 150 : PlanState *outerPlan = outerPlanState(node);
446 : :
447 : : /* Make sure any existing workers are gracefully shut down */
3599 rhaas@postgresql.org 448 : 150 : ExecShutdownGatherWorkers(node);
449 : :
450 : : /* Mark node so that shared state will be rebuilt at next call */
3613 451 : 150 : node->initialized = false;
452 : :
453 : : /*
454 : : * Set child node's chgParam to tell it that the next scan might deliver a
455 : : * different set of rows within the leader process. (The overall rowset
456 : : * shouldn't change, but the leader process's subset might; hence nodes
457 : : * between here and the parallel table scan node mustn't optimize on the
458 : : * assumption of an unchanging rowset.)
459 : : */
2929 tgl@sss.pgh.pa.us 460 [ + - ]: 150 : if (gather->rescan_param >= 0)
461 : 150 : outerPlan->chgParam = bms_add_member(outerPlan->chgParam,
462 : : gather->rescan_param);
463 : :
464 : : /*
465 : : * If chgParam of subnode is not null then plan will be re-scanned by
466 : : * first ExecProcNode. Note: because this does nothing if we have a
467 : : * rescan_param, it's currently guaranteed that parallel-aware child nodes
468 : : * will not see a ReScan call until after they get a ReInitializeDSM call.
469 : : * That ordering might not be something to rely on, though. A good rule
470 : : * of thumb is that ReInitializeDSM should reset only shared state, ReScan
471 : : * should reset only local state, and anything that depends on both of
472 : : * those steps being finished must wait until the first ExecProcNode call.
473 : : */
474 [ - + ]: 150 : if (outerPlan->chgParam == NULL)
2929 tgl@sss.pgh.pa.us 475 :UBC 0 : ExecReScan(outerPlan);
3629 rhaas@postgresql.org 476 :CBC 150 : }
|
