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