Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeAppend.c
4 : : * routines to handle append nodes.
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/backend/executor/nodeAppend.c
12 : : *
13 : : *-------------------------------------------------------------------------
14 : : */
15 : : /*
16 : : * INTERFACE ROUTINES
17 : : * ExecInitAppend - initialize the append node
18 : : * ExecAppend - retrieve the next tuple from the node
19 : : * ExecEndAppend - shut down the append node
20 : : * ExecReScanAppend - rescan the append node
21 : : *
22 : : * NOTES
23 : : * Each append node contains a list of one or more subplans which
24 : : * must be iteratively processed (forwards or backwards).
25 : : * Tuples are retrieved by executing the 'whichplan'th subplan
26 : : * until the subplan stops returning tuples, at which point that
27 : : * plan is shut down and the next started up.
28 : : *
29 : : * Append nodes don't make use of their left and right
30 : : * subtrees, rather they maintain a list of subplans so
31 : : * a typical append node looks like this in the plan tree:
32 : : *
33 : : * ...
34 : : * /
35 : : * Append -------+------+------+--- nil
36 : : * / \ | | |
37 : : * nil nil ... ... ...
38 : : * subplans
39 : : *
40 : : * Append nodes are currently used for unions, and to support
41 : : * inheritance queries, where several relations need to be scanned.
42 : : * For example, in our standard person/student/employee/student-emp
43 : : * example, where student and employee inherit from person
44 : : * and student-emp inherits from student and employee, the
45 : : * query:
46 : : *
47 : : * select name from person
48 : : *
49 : : * generates the plan:
50 : : *
51 : : * |
52 : : * Append -------+-------+--------+--------+
53 : : * / \ | | | |
54 : : * nil nil Scan Scan Scan Scan
55 : : * | | | |
56 : : * person employee student student-emp
57 : : */
58 : :
59 : : #include "postgres.h"
60 : :
61 : : #include "executor/execAsync.h"
62 : : #include "executor/execPartition.h"
63 : : #include "executor/executor.h"
64 : : #include "executor/nodeAppend.h"
65 : : #include "miscadmin.h"
66 : : #include "pgstat.h"
67 : : #include "storage/latch.h"
68 : : #include "storage/lwlock.h"
69 : : #include "utils/wait_event.h"
70 : :
71 : : /* Shared state for parallel-aware Append. */
72 : : struct ParallelAppendState
73 : : {
74 : : LWLock pa_lock; /* mutual exclusion to choose next subplan */
75 : : int pa_next_plan; /* next plan to choose by any worker */
76 : :
77 : : /*
78 : : * pa_finished[i] should be true if no more workers should select subplan
79 : : * i. for a non-partial plan, this should be set to true as soon as a
80 : : * worker selects the plan; for a partial plan, it remains false until
81 : : * some worker executes the plan to completion.
82 : : */
83 : : bool pa_finished[FLEXIBLE_ARRAY_MEMBER];
84 : : };
85 : :
86 : : #define INVALID_SUBPLAN_INDEX -1
87 : : #define EVENT_BUFFER_SIZE 16
88 : :
89 : : static TupleTableSlot *ExecAppend(PlanState *pstate);
90 : : static bool choose_next_subplan_locally(AppendState *node);
91 : : static bool choose_next_subplan_for_leader(AppendState *node);
92 : : static bool choose_next_subplan_for_worker(AppendState *node);
93 : : static void mark_invalid_subplans_as_finished(AppendState *node);
94 : : static void ExecAppendAsyncBegin(AppendState *node);
95 : : static bool ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result);
96 : : static bool ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result);
97 : : static void ExecAppendAsyncEventWait(AppendState *node);
98 : : static void ExecAppendAsyncReset(AppendState *node);
99 : : static void classify_matching_subplans(AppendState *node);
100 : :
101 : : /* ----------------------------------------------------------------
102 : : * ExecInitAppend
103 : : *
104 : : * Begin all of the subscans of the append node.
105 : : *
106 : : * (This is potentially wasteful, since the entire result of the
107 : : * append node may not be scanned, but this way all of the
108 : : * structures get allocated in the executor's top level memory
109 : : * block instead of that of the call to ExecAppend.)
110 : : * ----------------------------------------------------------------
111 : : */
112 : : AppendState *
113 : 12683 : ExecInitAppend(Append *node, EState *estate, int eflags)
114 : : {
115 : 12683 : AppendState *appendstate = makeNode(AppendState);
116 : : PlanState **appendplanstates;
117 : : const TupleTableSlotOps *appendops;
118 : : Bitmapset *validsubplans;
119 : : Bitmapset *asyncplans;
120 : : int nplans;
121 : : int nasyncplans;
122 : : int firstvalid;
123 : : int i,
124 : : j;
125 : :
126 : : /* check for unsupported flags */
127 : : Assert(!(eflags & EXEC_FLAG_MARK));
128 : :
129 : : /*
130 : : * create new AppendState for our append node
131 : : */
132 : 12683 : appendstate->ps.plan = (Plan *) node;
133 : 12683 : appendstate->ps.state = estate;
134 : 12683 : appendstate->ps.ExecProcNode = ExecAppend;
135 : :
136 : : /* Let choose_next_subplan_* function handle setting the first subplan */
137 : 12683 : appendstate->as_whichplan = INVALID_SUBPLAN_INDEX;
138 : 12683 : appendstate->as_syncdone = false;
139 : 12683 : appendstate->as_begun = false;
140 : :
141 : : /* If run-time partition pruning is enabled, then set that up now */
142 [ + + ]: 12683 : if (node->part_prune_index >= 0)
143 : : {
144 : : PartitionPruneState *prunestate;
145 : :
146 : : /*
147 : : * Set up pruning data structure. This also initializes the set of
148 : : * subplans to initialize (validsubplans) by taking into account the
149 : : * result of performing initial pruning if any.
150 : : */
151 : 497 : prunestate = ExecInitPartitionExecPruning(&appendstate->ps,
152 : 497 : list_length(node->appendplans),
153 : : node->part_prune_index,
154 : : node->apprelids,
155 : : &validsubplans);
156 : 497 : appendstate->as_prune_state = prunestate;
157 : 497 : nplans = bms_num_members(validsubplans);
158 : :
159 : : /*
160 : : * When no run-time pruning is required and there's at least one
161 : : * subplan, we can fill as_valid_subplans immediately, preventing
162 : : * later calls to ExecFindMatchingSubPlans.
163 : : */
164 [ + + + + ]: 497 : if (!prunestate->do_exec_prune && nplans > 0)
165 : : {
166 : 182 : appendstate->as_valid_subplans = bms_add_range(NULL, 0, nplans - 1);
167 : 182 : appendstate->as_valid_subplans_identified = true;
168 : : }
169 : : }
170 : : else
171 : : {
172 : 12186 : nplans = list_length(node->appendplans);
173 : :
174 : : /*
175 : : * When run-time partition pruning is not enabled we can just mark all
176 : : * subplans as valid; they must also all be initialized.
177 : : */
178 : : Assert(nplans > 0);
179 : 12186 : appendstate->as_valid_subplans = validsubplans =
180 : 12186 : bms_add_range(NULL, 0, nplans - 1);
181 : 12186 : appendstate->as_valid_subplans_identified = true;
182 : 12186 : appendstate->as_prune_state = NULL;
183 : : }
184 : :
185 : 12683 : appendplanstates = (PlanState **) palloc(nplans *
186 : : sizeof(PlanState *));
187 : :
188 : : /*
189 : : * call ExecInitNode on each of the valid plans to be executed and save
190 : : * the results into the appendplanstates array.
191 : : *
192 : : * While at it, find out the first valid partial plan.
193 : : */
194 : 12683 : j = 0;
195 : 12683 : asyncplans = NULL;
196 : 12683 : nasyncplans = 0;
197 : 12683 : firstvalid = nplans;
198 : 12683 : i = -1;
199 [ + + ]: 51815 : while ((i = bms_next_member(validsubplans, i)) >= 0)
200 : : {
201 : 39132 : Plan *initNode = (Plan *) list_nth(node->appendplans, i);
202 : :
203 : : /*
204 : : * Record async subplans. When executing EvalPlanQual, we treat them
205 : : * as sync ones; don't do this when initializing an EvalPlanQual plan
206 : : * tree.
207 : : */
208 [ + + + - ]: 39132 : if (initNode->async_capable && estate->es_epq_active == NULL)
209 : : {
210 : 105 : asyncplans = bms_add_member(asyncplans, j);
211 : 105 : nasyncplans++;
212 : : }
213 : :
214 : : /*
215 : : * Record the lowest appendplans index which is a valid partial plan.
216 : : */
217 [ + + + + ]: 39132 : if (i >= node->first_partial_plan && j < firstvalid)
218 : 334 : firstvalid = j;
219 : :
220 : 39132 : appendplanstates[j++] = ExecInitNode(initNode, estate, eflags);
221 : : }
222 : :
223 : 12683 : appendstate->as_first_partial_plan = firstvalid;
224 : 12683 : appendstate->appendplans = appendplanstates;
225 : 12683 : appendstate->as_nplans = nplans;
226 : :
227 : : /*
228 : : * Initialize Append's result tuple type and slot. If the child plans all
229 : : * produce the same fixed slot type, we can use that slot type; otherwise
230 : : * make a virtual slot. (Note that the result slot itself is used only to
231 : : * return a null tuple at end of execution; real tuples are returned to
232 : : * the caller in the children's own result slots. What we are doing here
233 : : * is allowing the parent plan node to optimize if the Append will return
234 : : * only one kind of slot.)
235 : : */
236 : 12683 : appendops = ExecGetCommonSlotOps(appendplanstates, j);
237 [ + + ]: 12683 : if (appendops != NULL)
238 : : {
239 : 12029 : ExecInitResultTupleSlotTL(&appendstate->ps, appendops);
240 : : }
241 : : else
242 : : {
243 : 654 : ExecInitResultTupleSlotTL(&appendstate->ps, &TTSOpsVirtual);
244 : : /* show that the output slot type is not fixed */
245 : 654 : appendstate->ps.resultopsset = true;
246 : 654 : appendstate->ps.resultopsfixed = false;
247 : : }
248 : :
249 : : /* Initialize async state */
250 : 12683 : appendstate->as_asyncplans = asyncplans;
251 : 12683 : appendstate->as_nasyncplans = nasyncplans;
252 : 12683 : appendstate->as_asyncrequests = NULL;
253 : 12683 : appendstate->as_asyncresults = NULL;
254 : 12683 : appendstate->as_nasyncresults = 0;
255 : 12683 : appendstate->as_nasyncremain = 0;
256 : 12683 : appendstate->as_needrequest = NULL;
257 : 12683 : appendstate->as_eventset = NULL;
258 : 12683 : appendstate->as_valid_asyncplans = NULL;
259 : :
260 [ + + ]: 12683 : if (nasyncplans > 0)
261 : : {
262 : 51 : appendstate->as_asyncrequests = (AsyncRequest **)
263 : 51 : palloc0(nplans * sizeof(AsyncRequest *));
264 : :
265 : 51 : i = -1;
266 [ + + ]: 156 : while ((i = bms_next_member(asyncplans, i)) >= 0)
267 : : {
268 : : AsyncRequest *areq;
269 : :
270 : 105 : areq = palloc_object(AsyncRequest);
271 : 105 : areq->requestor = (PlanState *) appendstate;
272 : 105 : areq->requestee = appendplanstates[i];
273 : 105 : areq->request_index = i;
274 : 105 : areq->callback_pending = false;
275 : 105 : areq->request_complete = false;
276 : 105 : areq->result = NULL;
277 : :
278 : 105 : appendstate->as_asyncrequests[i] = areq;
279 : : }
280 : :
281 : 51 : appendstate->as_asyncresults = (TupleTableSlot **)
282 : 51 : palloc0(nasyncplans * sizeof(TupleTableSlot *));
283 : :
284 [ + + ]: 51 : if (appendstate->as_valid_subplans_identified)
285 : 44 : classify_matching_subplans(appendstate);
286 : : }
287 : :
288 : : /*
289 : : * Miscellaneous initialization
290 : : */
291 : :
292 : 12683 : appendstate->ps.ps_ProjInfo = NULL;
293 : :
294 : : /* For parallel query, this will be overridden later. */
295 : 12683 : appendstate->choose_next_subplan = choose_next_subplan_locally;
296 : :
297 : 12683 : return appendstate;
298 : : }
299 : :
300 : : /* ----------------------------------------------------------------
301 : : * ExecAppend
302 : : *
303 : : * Handles iteration over multiple subplans.
304 : : * ----------------------------------------------------------------
305 : : */
306 : : static TupleTableSlot *
307 : 1708959 : ExecAppend(PlanState *pstate)
308 : : {
309 : 1708959 : AppendState *node = castNode(AppendState, pstate);
310 : : TupleTableSlot *result;
311 : :
312 : : /*
313 : : * If this is the first call after Init or ReScan, we need to do the
314 : : * initialization work.
315 : : */
316 [ + + ]: 1708959 : if (!node->as_begun)
317 : : {
318 : : Assert(node->as_whichplan == INVALID_SUBPLAN_INDEX);
319 : : Assert(!node->as_syncdone);
320 : :
321 : : /* Nothing to do if there are no subplans */
322 [ + + ]: 24887 : if (node->as_nplans == 0)
323 : 39 : return ExecClearTuple(node->ps.ps_ResultTupleSlot);
324 : :
325 : : /* If there are any async subplans, begin executing them. */
326 [ + + ]: 24848 : if (node->as_nasyncplans > 0)
327 : 41 : ExecAppendAsyncBegin(node);
328 : :
329 : : /*
330 : : * If no sync subplan has been chosen, we must choose one before
331 : : * proceeding.
332 : : */
333 [ + + + + ]: 24848 : if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0)
334 : 2157 : return ExecClearTuple(node->ps.ps_ResultTupleSlot);
335 : :
336 : : Assert(node->as_syncdone ||
337 : : (node->as_whichplan >= 0 &&
338 : : node->as_whichplan < node->as_nplans));
339 : :
340 : : /* And we're initialized. */
341 : 22691 : node->as_begun = true;
342 : : }
343 : :
344 : : for (;;)
345 : 32100 : {
346 : : PlanState *subnode;
347 : :
348 [ + + ]: 1738863 : CHECK_FOR_INTERRUPTS();
349 : :
350 : : /*
351 : : * try to get a tuple from an async subplan if any
352 : : */
353 [ + + - + ]: 1738863 : if (node->as_syncdone || !bms_is_empty(node->as_needrequest))
354 : : {
355 [ + - ]: 6142 : if (ExecAppendAsyncGetNext(node, &result))
356 : 6141 : return result;
357 : : Assert(!node->as_syncdone);
358 : : Assert(bms_is_empty(node->as_needrequest));
359 : : }
360 : :
361 : : /*
362 : : * figure out which sync subplan we are currently processing
363 : : */
364 : : Assert(node->as_whichplan >= 0 && node->as_whichplan < node->as_nplans);
365 : 1732721 : subnode = node->appendplans[node->as_whichplan];
366 : :
367 : : /*
368 : : * get a tuple from the subplan
369 : : */
370 : 1732721 : result = ExecProcNode(subnode);
371 : :
372 [ + + + + ]: 1732687 : if (!TupIsNull(result))
373 : : {
374 : : /*
375 : : * If the subplan gave us something then return it as-is. We do
376 : : * NOT make use of the result slot that was set up in
377 : : * ExecInitAppend; there's no need for it.
378 : : */
379 : 1678321 : return result;
380 : : }
381 : :
382 : : /*
383 : : * wait or poll for async events if any. We do this before checking
384 : : * for the end of iteration, because it might drain the remaining
385 : : * async subplans.
386 : : */
387 [ + + ]: 54366 : if (node->as_nasyncremain > 0)
388 : 17 : ExecAppendAsyncEventWait(node);
389 : :
390 : : /* choose new sync subplan; if no sync/async subplans, we're done */
391 [ + + + + ]: 54366 : if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0)
392 : 22266 : return ExecClearTuple(node->ps.ps_ResultTupleSlot);
393 : : }
394 : : }
395 : :
396 : : /* ----------------------------------------------------------------
397 : : * ExecEndAppend
398 : : *
399 : : * Shuts down the subscans of the append node.
400 : : *
401 : : * Returns nothing of interest.
402 : : * ----------------------------------------------------------------
403 : : */
404 : : void
405 : 12482 : ExecEndAppend(AppendState *node)
406 : : {
407 : : PlanState **appendplans;
408 : : int nplans;
409 : : int i;
410 : :
411 : : /*
412 : : * get information from the node
413 : : */
414 : 12482 : appendplans = node->appendplans;
415 : 12482 : nplans = node->as_nplans;
416 : :
417 : : /*
418 : : * shut down each of the subscans
419 : : */
420 [ + + ]: 51134 : for (i = 0; i < nplans; i++)
421 : 38652 : ExecEndNode(appendplans[i]);
422 : 12482 : }
423 : :
424 : : void
425 : 16317 : ExecReScanAppend(AppendState *node)
426 : : {
427 : 16317 : int nasyncplans = node->as_nasyncplans;
428 : : int i;
429 : :
430 : : /* If there are any async subplans, reset async requests made for them. */
431 [ + + ]: 16317 : if (nasyncplans > 0)
432 : 21 : ExecAppendAsyncReset(node);
433 : :
434 : : /*
435 : : * If any PARAM_EXEC Params used in pruning expressions have changed, then
436 : : * we'd better unset the valid subplans so that they are reselected for
437 : : * the new parameter values.
438 : : */
439 [ + + + + ]: 18501 : if (node->as_prune_state &&
440 : 2184 : bms_overlap(node->ps.chgParam,
441 : 2184 : node->as_prune_state->execparamids))
442 : : {
443 : 2182 : node->as_valid_subplans_identified = false;
444 : 2182 : bms_free(node->as_valid_subplans);
445 : 2182 : node->as_valid_subplans = NULL;
446 : 2182 : bms_free(node->as_valid_asyncplans);
447 : 2182 : node->as_valid_asyncplans = NULL;
448 : : }
449 : :
450 [ + + ]: 64380 : for (i = 0; i < node->as_nplans; i++)
451 : : {
452 : 48063 : PlanState *subnode = node->appendplans[i];
453 : :
454 : : /*
455 : : * ExecReScan doesn't know about my subplans, so I have to do
456 : : * changed-parameter signaling myself.
457 : : */
458 [ + + ]: 48063 : if (node->ps.chgParam != NULL)
459 : 37591 : UpdateChangedParamSet(subnode, node->ps.chgParam);
460 : :
461 : : /*
462 : : * If chgParam of subnode is not null then plan will be re-scanned by
463 : : * first ExecProcNode or by first ExecAsyncRequest.
464 : : */
465 [ + + ]: 48063 : if (subnode->chgParam == NULL)
466 : 18212 : ExecReScan(subnode);
467 : : }
468 : :
469 : : /* Let choose_next_subplan_* function handle setting the first subplan */
470 : 16317 : node->as_whichplan = INVALID_SUBPLAN_INDEX;
471 : 16317 : node->as_syncdone = false;
472 : 16317 : node->as_begun = false;
473 : 16317 : }
474 : :
475 : : /* ----------------------------------------------------------------
476 : : * Parallel Append Support
477 : : * ----------------------------------------------------------------
478 : : */
479 : :
480 : : /* ----------------------------------------------------------------
481 : : * ExecAppendEstimate
482 : : *
483 : : * Compute the amount of space we'll need in the parallel
484 : : * query DSM, and inform pcxt->estimator about our needs.
485 : : * ----------------------------------------------------------------
486 : : */
487 : : void
488 : 108 : ExecAppendEstimate(AppendState *node,
489 : : ParallelContext *pcxt)
490 : : {
491 : 108 : node->pstate_len =
492 : 108 : add_size(offsetof(ParallelAppendState, pa_finished),
493 : 108 : sizeof(bool) * node->as_nplans);
494 : :
495 : 108 : shm_toc_estimate_chunk(&pcxt->estimator, node->pstate_len);
496 : 108 : shm_toc_estimate_keys(&pcxt->estimator, 1);
497 : 108 : }
498 : :
499 : :
500 : : /* ----------------------------------------------------------------
501 : : * ExecAppendInitializeDSM
502 : : *
503 : : * Set up shared state for Parallel Append.
504 : : * ----------------------------------------------------------------
505 : : */
506 : : void
507 : 108 : ExecAppendInitializeDSM(AppendState *node,
508 : : ParallelContext *pcxt)
509 : : {
510 : : ParallelAppendState *pstate;
511 : :
512 : 108 : pstate = shm_toc_allocate(pcxt->toc, node->pstate_len);
513 : 108 : memset(pstate, 0, node->pstate_len);
514 : 108 : LWLockInitialize(&pstate->pa_lock, LWTRANCHE_PARALLEL_APPEND);
515 : 108 : shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id, pstate);
516 : :
517 : 108 : node->as_pstate = pstate;
518 : 108 : node->choose_next_subplan = choose_next_subplan_for_leader;
519 : 108 : }
520 : :
521 : : /* ----------------------------------------------------------------
522 : : * ExecAppendReInitializeDSM
523 : : *
524 : : * Reset shared state before beginning a fresh scan.
525 : : * ----------------------------------------------------------------
526 : : */
527 : : void
528 : 0 : ExecAppendReInitializeDSM(AppendState *node, ParallelContext *pcxt)
529 : : {
530 : 0 : ParallelAppendState *pstate = node->as_pstate;
531 : :
532 : 0 : pstate->pa_next_plan = 0;
533 : 0 : memset(pstate->pa_finished, 0, sizeof(bool) * node->as_nplans);
534 : 0 : }
535 : :
536 : : /* ----------------------------------------------------------------
537 : : * ExecAppendInitializeWorker
538 : : *
539 : : * Copy relevant information from TOC into planstate, and initialize
540 : : * whatever is required to choose and execute the optimal subplan.
541 : : * ----------------------------------------------------------------
542 : : */
543 : : void
544 : 241 : ExecAppendInitializeWorker(AppendState *node, ParallelWorkerContext *pwcxt)
545 : : {
546 : 241 : node->as_pstate = shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
547 : 241 : node->choose_next_subplan = choose_next_subplan_for_worker;
548 : 241 : }
549 : :
550 : : /* ----------------------------------------------------------------
551 : : * choose_next_subplan_locally
552 : : *
553 : : * Choose next sync subplan for a non-parallel-aware Append,
554 : : * returning false if there are no more.
555 : : * ----------------------------------------------------------------
556 : : */
557 : : static bool
558 : 77930 : choose_next_subplan_locally(AppendState *node)
559 : : {
560 : 77930 : int whichplan = node->as_whichplan;
561 : : int nextplan;
562 : :
563 : : /* We should never be called when there are no subplans */
564 : : Assert(node->as_nplans > 0);
565 : :
566 : : /* Nothing to do if syncdone */
567 [ + + ]: 77930 : if (node->as_syncdone)
568 : 22 : return false;
569 : :
570 : : /*
571 : : * If first call then have the bms member function choose the first valid
572 : : * sync subplan by initializing whichplan to -1. If there happen to be no
573 : : * valid sync subplans then the bms member function will handle that by
574 : : * returning a negative number which will allow us to exit returning a
575 : : * false value.
576 : : */
577 [ + + ]: 77908 : if (whichplan == INVALID_SUBPLAN_INDEX)
578 : : {
579 [ + + ]: 24512 : if (node->as_nasyncplans > 0)
580 : : {
581 : : /* We'd have filled as_valid_subplans already */
582 : : Assert(node->as_valid_subplans_identified);
583 : : }
584 [ + + ]: 24493 : else if (!node->as_valid_subplans_identified)
585 : : {
586 : 2255 : node->as_valid_subplans =
587 : 2255 : ExecFindMatchingSubPlans(node->as_prune_state, false, NULL);
588 : 2255 : node->as_valid_subplans_identified = true;
589 : : }
590 : :
591 : 24512 : whichplan = -1;
592 : : }
593 : :
594 : : /* Ensure whichplan is within the expected range */
595 : : Assert(whichplan >= -1 && whichplan <= node->as_nplans);
596 : :
597 [ + + ]: 77908 : if (ScanDirectionIsForward(node->ps.state->es_direction))
598 : 77896 : nextplan = bms_next_member(node->as_valid_subplans, whichplan);
599 : : else
600 : 12 : nextplan = bms_prev_member(node->as_valid_subplans, whichplan);
601 : :
602 [ + + ]: 77908 : if (nextplan < 0)
603 : : {
604 : : /* Set as_syncdone if in async mode */
605 [ + + ]: 24126 : if (node->as_nasyncplans > 0)
606 : 17 : node->as_syncdone = true;
607 : 24126 : return false;
608 : : }
609 : :
610 : 53782 : node->as_whichplan = nextplan;
611 : :
612 : 53782 : return true;
613 : : }
614 : :
615 : : /* ----------------------------------------------------------------
616 : : * choose_next_subplan_for_leader
617 : : *
618 : : * Try to pick a plan which doesn't commit us to doing much
619 : : * work locally, so that as much work as possible is done in
620 : : * the workers. Cheapest subplans are at the end.
621 : : * ----------------------------------------------------------------
622 : : */
623 : : static bool
624 : 953 : choose_next_subplan_for_leader(AppendState *node)
625 : : {
626 : 953 : ParallelAppendState *pstate = node->as_pstate;
627 : :
628 : : /* Backward scan is not supported by parallel-aware plans */
629 : : Assert(ScanDirectionIsForward(node->ps.state->es_direction));
630 : :
631 : : /* We should never be called when there are no subplans */
632 : : Assert(node->as_nplans > 0);
633 : :
634 : 953 : LWLockAcquire(&pstate->pa_lock, LW_EXCLUSIVE);
635 : :
636 [ + + ]: 953 : if (node->as_whichplan != INVALID_SUBPLAN_INDEX)
637 : : {
638 : : /* Mark just-completed subplan as finished. */
639 : 857 : node->as_pstate->pa_finished[node->as_whichplan] = true;
640 : : }
641 : : else
642 : : {
643 : : /* Start with last subplan. */
644 : 96 : node->as_whichplan = node->as_nplans - 1;
645 : :
646 : : /*
647 : : * If we've yet to determine the valid subplans then do so now. If
648 : : * run-time pruning is disabled then the valid subplans will always be
649 : : * set to all subplans.
650 : : */
651 [ + + ]: 96 : if (!node->as_valid_subplans_identified)
652 : : {
653 : 16 : node->as_valid_subplans =
654 : 16 : ExecFindMatchingSubPlans(node->as_prune_state, false, NULL);
655 : 16 : node->as_valid_subplans_identified = true;
656 : :
657 : : /*
658 : : * Mark each invalid plan as finished to allow the loop below to
659 : : * select the first valid subplan.
660 : : */
661 : 16 : mark_invalid_subplans_as_finished(node);
662 : : }
663 : : }
664 : :
665 : : /* Loop until we find a subplan to execute. */
666 [ + + ]: 1753 : while (pstate->pa_finished[node->as_whichplan])
667 : : {
668 [ + + ]: 896 : if (node->as_whichplan == 0)
669 : : {
670 : 96 : pstate->pa_next_plan = INVALID_SUBPLAN_INDEX;
671 : 96 : node->as_whichplan = INVALID_SUBPLAN_INDEX;
672 : 96 : LWLockRelease(&pstate->pa_lock);
673 : 96 : return false;
674 : : }
675 : :
676 : : /*
677 : : * We needn't pay attention to as_valid_subplans here as all invalid
678 : : * plans have been marked as finished.
679 : : */
680 : 800 : node->as_whichplan--;
681 : : }
682 : :
683 : : /* If non-partial, immediately mark as finished. */
684 [ + + ]: 857 : if (node->as_whichplan < node->as_first_partial_plan)
685 : 121 : node->as_pstate->pa_finished[node->as_whichplan] = true;
686 : :
687 : 857 : LWLockRelease(&pstate->pa_lock);
688 : :
689 : 857 : return true;
690 : : }
691 : :
692 : : /* ----------------------------------------------------------------
693 : : * choose_next_subplan_for_worker
694 : : *
695 : : * Choose next subplan for a parallel-aware Append, returning
696 : : * false if there are no more.
697 : : *
698 : : * We start from the first plan and advance through the list;
699 : : * when we get back to the end, we loop back to the first
700 : : * partial plan. This assigns the non-partial plans first in
701 : : * order of descending cost and then spreads out the workers
702 : : * as evenly as possible across the remaining partial plans.
703 : : * ----------------------------------------------------------------
704 : : */
705 : : static bool
706 : 331 : choose_next_subplan_for_worker(AppendState *node)
707 : : {
708 : 331 : ParallelAppendState *pstate = node->as_pstate;
709 : :
710 : : /* Backward scan is not supported by parallel-aware plans */
711 : : Assert(ScanDirectionIsForward(node->ps.state->es_direction));
712 : :
713 : : /* We should never be called when there are no subplans */
714 : : Assert(node->as_nplans > 0);
715 : :
716 : 331 : LWLockAcquire(&pstate->pa_lock, LW_EXCLUSIVE);
717 : :
718 : : /* Mark just-completed subplan as finished. */
719 [ + + ]: 331 : if (node->as_whichplan != INVALID_SUBPLAN_INDEX)
720 : 113 : node->as_pstate->pa_finished[node->as_whichplan] = true;
721 : :
722 : : /*
723 : : * If we've yet to determine the valid subplans then do so now. If
724 : : * run-time pruning is disabled then the valid subplans will always be set
725 : : * to all subplans.
726 : : */
727 [ + + ]: 218 : else if (!node->as_valid_subplans_identified)
728 : : {
729 : 16 : node->as_valid_subplans =
730 : 16 : ExecFindMatchingSubPlans(node->as_prune_state, false, NULL);
731 : 16 : node->as_valid_subplans_identified = true;
732 : :
733 : 16 : mark_invalid_subplans_as_finished(node);
734 : : }
735 : :
736 : : /* If all the plans are already done, we have nothing to do */
737 [ + + ]: 331 : if (pstate->pa_next_plan == INVALID_SUBPLAN_INDEX)
738 : : {
739 : 194 : LWLockRelease(&pstate->pa_lock);
740 : 194 : return false;
741 : : }
742 : :
743 : : /* Save the plan from which we are starting the search. */
744 : 137 : node->as_whichplan = pstate->pa_next_plan;
745 : :
746 : : /* Loop until we find a valid subplan to execute. */
747 [ + + ]: 253 : while (pstate->pa_finished[pstate->pa_next_plan])
748 : : {
749 : : int nextplan;
750 : :
751 : 140 : nextplan = bms_next_member(node->as_valid_subplans,
752 : : pstate->pa_next_plan);
753 [ + + ]: 140 : if (nextplan >= 0)
754 : : {
755 : : /* Advance to the next valid plan. */
756 : 104 : pstate->pa_next_plan = nextplan;
757 : : }
758 [ + + ]: 36 : else if (node->as_whichplan > node->as_first_partial_plan)
759 : : {
760 : : /*
761 : : * Try looping back to the first valid partial plan, if there is
762 : : * one. If there isn't, arrange to bail out below.
763 : : */
764 : 24 : nextplan = bms_next_member(node->as_valid_subplans,
765 : 24 : node->as_first_partial_plan - 1);
766 : 24 : pstate->pa_next_plan =
767 [ - + ]: 24 : nextplan < 0 ? node->as_whichplan : nextplan;
768 : : }
769 : : else
770 : : {
771 : : /*
772 : : * At last plan, and either there are no partial plans or we've
773 : : * tried them all. Arrange to bail out.
774 : : */
775 : 12 : pstate->pa_next_plan = node->as_whichplan;
776 : : }
777 : :
778 [ + + ]: 140 : if (pstate->pa_next_plan == node->as_whichplan)
779 : : {
780 : : /* We've tried everything! */
781 : 24 : pstate->pa_next_plan = INVALID_SUBPLAN_INDEX;
782 : 24 : LWLockRelease(&pstate->pa_lock);
783 : 24 : return false;
784 : : }
785 : : }
786 : :
787 : : /* Pick the plan we found, and advance pa_next_plan one more time. */
788 : 113 : node->as_whichplan = pstate->pa_next_plan;
789 : 113 : pstate->pa_next_plan = bms_next_member(node->as_valid_subplans,
790 : : pstate->pa_next_plan);
791 : :
792 : : /*
793 : : * If there are no more valid plans then try setting the next plan to the
794 : : * first valid partial plan.
795 : : */
796 [ + + ]: 113 : if (pstate->pa_next_plan < 0)
797 : : {
798 : 19 : int nextplan = bms_next_member(node->as_valid_subplans,
799 : 19 : node->as_first_partial_plan - 1);
800 : :
801 [ + - ]: 19 : if (nextplan >= 0)
802 : 19 : pstate->pa_next_plan = nextplan;
803 : : else
804 : : {
805 : : /*
806 : : * There are no valid partial plans, and we already chose the last
807 : : * non-partial plan; so flag that there's nothing more for our
808 : : * fellow workers to do.
809 : : */
810 : 0 : pstate->pa_next_plan = INVALID_SUBPLAN_INDEX;
811 : : }
812 : : }
813 : :
814 : : /* If non-partial, immediately mark as finished. */
815 [ + + ]: 113 : if (node->as_whichplan < node->as_first_partial_plan)
816 : 19 : node->as_pstate->pa_finished[node->as_whichplan] = true;
817 : :
818 : 113 : LWLockRelease(&pstate->pa_lock);
819 : :
820 : 113 : return true;
821 : : }
822 : :
823 : : /*
824 : : * mark_invalid_subplans_as_finished
825 : : * Marks the ParallelAppendState's pa_finished as true for each invalid
826 : : * subplan.
827 : : *
828 : : * This function should only be called for parallel Append with run-time
829 : : * pruning enabled.
830 : : */
831 : : static void
832 : 32 : mark_invalid_subplans_as_finished(AppendState *node)
833 : : {
834 : : int i;
835 : :
836 : : /* Only valid to call this while in parallel Append mode */
837 : : Assert(node->as_pstate);
838 : :
839 : : /* Shouldn't have been called when run-time pruning is not enabled */
840 : : Assert(node->as_prune_state);
841 : :
842 : : /* Nothing to do if all plans are valid */
843 [ - + ]: 32 : if (bms_num_members(node->as_valid_subplans) == node->as_nplans)
844 : 0 : return;
845 : :
846 : : /* Mark all non-valid plans as finished */
847 [ + + ]: 108 : for (i = 0; i < node->as_nplans; i++)
848 : : {
849 [ + + ]: 76 : if (!bms_is_member(i, node->as_valid_subplans))
850 : 32 : node->as_pstate->pa_finished[i] = true;
851 : : }
852 : : }
853 : :
854 : : /* ----------------------------------------------------------------
855 : : * Asynchronous Append Support
856 : : * ----------------------------------------------------------------
857 : : */
858 : :
859 : : /* ----------------------------------------------------------------
860 : : * ExecAppendAsyncBegin
861 : : *
862 : : * Begin executing designed async-capable subplans.
863 : : * ----------------------------------------------------------------
864 : : */
865 : : static void
866 : 41 : ExecAppendAsyncBegin(AppendState *node)
867 : : {
868 : : int i;
869 : :
870 : : /* Backward scan is not supported by async-aware Appends. */
871 : : Assert(ScanDirectionIsForward(node->ps.state->es_direction));
872 : :
873 : : /* We should never be called when there are no subplans */
874 : : Assert(node->as_nplans > 0);
875 : :
876 : : /* We should never be called when there are no async subplans. */
877 : : Assert(node->as_nasyncplans > 0);
878 : :
879 : : /* If we've yet to determine the valid subplans then do so now. */
880 [ + + ]: 41 : if (!node->as_valid_subplans_identified)
881 : : {
882 : 6 : node->as_valid_subplans =
883 : 6 : ExecFindMatchingSubPlans(node->as_prune_state, false, NULL);
884 : 6 : node->as_valid_subplans_identified = true;
885 : :
886 : 6 : classify_matching_subplans(node);
887 : : }
888 : :
889 : : /* Initialize state variables. */
890 : 41 : node->as_syncdone = bms_is_empty(node->as_valid_subplans);
891 : 41 : node->as_nasyncremain = bms_num_members(node->as_valid_asyncplans);
892 : :
893 : : /* Nothing to do if there are no valid async subplans. */
894 [ - + ]: 41 : if (node->as_nasyncremain == 0)
895 : 0 : return;
896 : :
897 : : /* Make a request for each of the valid async subplans. */
898 : 41 : i = -1;
899 [ + + ]: 121 : while ((i = bms_next_member(node->as_valid_asyncplans, i)) >= 0)
900 : : {
901 : 80 : AsyncRequest *areq = node->as_asyncrequests[i];
902 : :
903 : : Assert(areq->request_index == i);
904 : : Assert(!areq->callback_pending);
905 : :
906 : : /* Do the actual work. */
907 : 80 : ExecAsyncRequest(areq);
908 : : }
909 : : }
910 : :
911 : : /* ----------------------------------------------------------------
912 : : * ExecAppendAsyncGetNext
913 : : *
914 : : * Get the next tuple from any of the asynchronous subplans.
915 : : * ----------------------------------------------------------------
916 : : */
917 : : static bool
918 : 6142 : ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result)
919 : : {
920 : 6142 : *result = NULL;
921 : :
922 : : /* We should never be called when there are no valid async subplans. */
923 : : Assert(node->as_nasyncremain > 0);
924 : :
925 : : /* Request a tuple asynchronously. */
926 [ + + ]: 6142 : if (ExecAppendAsyncRequest(node, result))
927 : 6034 : return true;
928 : :
929 [ + + ]: 152 : while (node->as_nasyncremain > 0)
930 : : {
931 [ - + ]: 121 : CHECK_FOR_INTERRUPTS();
932 : :
933 : : /* Wait or poll for async events. */
934 : 121 : ExecAppendAsyncEventWait(node);
935 : :
936 : : /* Request a tuple asynchronously. */
937 [ + + ]: 120 : if (ExecAppendAsyncRequest(node, result))
938 : 76 : return true;
939 : :
940 : : /* Break from loop if there's any sync subplan that isn't complete. */
941 [ - + ]: 44 : if (!node->as_syncdone)
942 : 0 : break;
943 : : }
944 : :
945 : : /*
946 : : * If all sync subplans are complete, we're totally done scanning the
947 : : * given node. Otherwise, we're done with the asynchronous stuff but must
948 : : * continue scanning the sync subplans.
949 : : */
950 [ + - ]: 31 : if (node->as_syncdone)
951 : : {
952 : : Assert(node->as_nasyncremain == 0);
953 : 31 : *result = ExecClearTuple(node->ps.ps_ResultTupleSlot);
954 : 31 : return true;
955 : : }
956 : :
957 : 0 : return false;
958 : : }
959 : :
960 : : /* ----------------------------------------------------------------
961 : : * ExecAppendAsyncRequest
962 : : *
963 : : * Request a tuple asynchronously.
964 : : * ----------------------------------------------------------------
965 : : */
966 : : static bool
967 : 6262 : ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result)
968 : : {
969 : : Bitmapset *needrequest;
970 : : int i;
971 : :
972 : : /* Nothing to do if there are no async subplans needing a new request. */
973 [ + + ]: 6262 : if (bms_is_empty(node->as_needrequest))
974 : : {
975 : : Assert(node->as_nasyncresults == 0);
976 : 72 : return false;
977 : : }
978 : :
979 : : /*
980 : : * If there are any asynchronously-generated results that have not yet
981 : : * been returned, we have nothing to do; just return one of them.
982 : : */
983 [ + + ]: 6190 : if (node->as_nasyncresults > 0)
984 : : {
985 : 789 : --node->as_nasyncresults;
986 : 789 : *result = node->as_asyncresults[node->as_nasyncresults];
987 : 789 : return true;
988 : : }
989 : :
990 : : /* Make a new request for each of the async subplans that need it. */
991 : 5401 : needrequest = node->as_needrequest;
992 : 5401 : node->as_needrequest = NULL;
993 : 5401 : i = -1;
994 [ + + ]: 11504 : while ((i = bms_next_member(needrequest, i)) >= 0)
995 : : {
996 : 6103 : AsyncRequest *areq = node->as_asyncrequests[i];
997 : :
998 : : /* Do the actual work. */
999 : 6103 : ExecAsyncRequest(areq);
1000 : : }
1001 : 5401 : bms_free(needrequest);
1002 : :
1003 : : /* Return one of the asynchronously-generated results if any. */
1004 [ + + ]: 5401 : if (node->as_nasyncresults > 0)
1005 : : {
1006 : 5321 : --node->as_nasyncresults;
1007 : 5321 : *result = node->as_asyncresults[node->as_nasyncresults];
1008 : 5321 : return true;
1009 : : }
1010 : :
1011 : 80 : return false;
1012 : : }
1013 : :
1014 : : /* ----------------------------------------------------------------
1015 : : * ExecAppendAsyncEventWait
1016 : : *
1017 : : * Wait or poll for file descriptor events and fire callbacks.
1018 : : * ----------------------------------------------------------------
1019 : : */
1020 : : static void
1021 : 140 : ExecAppendAsyncEventWait(AppendState *node)
1022 : : {
1023 : 140 : int nevents = node->as_nasyncplans + 2;
1024 [ + + ]: 140 : long timeout = node->as_syncdone ? -1 : 0;
1025 : : WaitEvent occurred_event[EVENT_BUFFER_SIZE];
1026 : : int noccurred;
1027 : : int i;
1028 : :
1029 : : /* We should never be called when there are no valid async subplans. */
1030 : : Assert(node->as_nasyncremain > 0);
1031 : :
1032 : : Assert(node->as_eventset == NULL);
1033 : 140 : node->as_eventset = CreateWaitEventSet(CurrentResourceOwner, nevents);
1034 : 140 : AddWaitEventToSet(node->as_eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
1035 : : NULL, NULL);
1036 : :
1037 : : /* Give each waiting subplan a chance to add an event. */
1038 : 140 : i = -1;
1039 [ + + ]: 433 : while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
1040 : : {
1041 : 294 : AsyncRequest *areq = node->as_asyncrequests[i];
1042 : :
1043 [ + + ]: 294 : if (areq->callback_pending)
1044 : 238 : ExecAsyncConfigureWait(areq);
1045 : : }
1046 : :
1047 : : /*
1048 : : * No need for further processing if none of the subplans configured any
1049 : : * events.
1050 : : */
1051 [ + + ]: 139 : if (GetNumRegisteredWaitEvents(node->as_eventset) == 1)
1052 : : {
1053 : 1 : FreeWaitEventSet(node->as_eventset);
1054 : 1 : node->as_eventset = NULL;
1055 : 7 : return;
1056 : : }
1057 : :
1058 : : /*
1059 : : * Add the process latch to the set, so that we wake up to process the
1060 : : * standard interrupts with CHECK_FOR_INTERRUPTS().
1061 : : *
1062 : : * NOTE: For historical reasons, it's important that this is added to the
1063 : : * WaitEventSet after the ExecAsyncConfigureWait() calls. Namely,
1064 : : * postgres_fdw calls "GetNumRegisteredWaitEvents(set) == 1" to check if
1065 : : * any other events are in the set. That's a poor design, it's
1066 : : * questionable for postgres_fdw to be doing that in the first place, but
1067 : : * we cannot change it now. The pattern has possibly been copied to other
1068 : : * extensions too.
1069 : : */
1070 : 138 : AddWaitEventToSet(node->as_eventset, WL_LATCH_SET, PGINVALID_SOCKET,
1071 : : MyLatch, NULL);
1072 : :
1073 : : /* Return at most EVENT_BUFFER_SIZE events in one call. */
1074 [ - + ]: 138 : if (nevents > EVENT_BUFFER_SIZE)
1075 : 0 : nevents = EVENT_BUFFER_SIZE;
1076 : :
1077 : : /*
1078 : : * If the timeout is -1, wait until at least one event occurs. If the
1079 : : * timeout is 0, poll for events, but do not wait at all.
1080 : : */
1081 : 138 : noccurred = WaitEventSetWait(node->as_eventset, timeout, occurred_event,
1082 : : nevents, WAIT_EVENT_APPEND_READY);
1083 : 138 : FreeWaitEventSet(node->as_eventset);
1084 : 138 : node->as_eventset = NULL;
1085 [ + + ]: 138 : if (noccurred == 0)
1086 : 6 : return;
1087 : :
1088 : : /* Deliver notifications. */
1089 [ + + ]: 290 : for (i = 0; i < noccurred; i++)
1090 : : {
1091 : 158 : WaitEvent *w = &occurred_event[i];
1092 : :
1093 : : /*
1094 : : * Each waiting subplan should have registered its wait event with
1095 : : * user_data pointing back to its AsyncRequest.
1096 : : */
1097 [ + - ]: 158 : if ((w->events & WL_SOCKET_READABLE) != 0)
1098 : : {
1099 : 158 : AsyncRequest *areq = (AsyncRequest *) w->user_data;
1100 : :
1101 [ + - ]: 158 : if (areq->callback_pending)
1102 : : {
1103 : : /*
1104 : : * Mark it as no longer needing a callback. We must do this
1105 : : * before dispatching the callback in case the callback resets
1106 : : * the flag.
1107 : : */
1108 : 158 : areq->callback_pending = false;
1109 : :
1110 : : /* Do the actual work. */
1111 : 158 : ExecAsyncNotify(areq);
1112 : : }
1113 : : }
1114 : :
1115 : : /* Handle standard interrupts */
1116 [ - + ]: 158 : if ((w->events & WL_LATCH_SET) != 0)
1117 : : {
1118 : 0 : ResetLatch(MyLatch);
1119 [ # # ]: 0 : CHECK_FOR_INTERRUPTS();
1120 : : }
1121 : : }
1122 : : }
1123 : :
1124 : : /* ----------------------------------------------------------------
1125 : : * ExecAppendAsyncReset
1126 : : *
1127 : : * Reset asynchronous requests made for async-capable subplans.
1128 : : * ----------------------------------------------------------------
1129 : : */
1130 : : static void
1131 : 21 : ExecAppendAsyncReset(AppendState *node)
1132 : : {
1133 : : int i;
1134 : :
1135 : : /* We should never be called when there are no async subplans. */
1136 : : Assert(node->as_nasyncplans > 0);
1137 : :
1138 : : /*
1139 : : * Drain pending async requests if any. We force the as_syncdone flag to
1140 : : * be true so that ExecAppendAsyncEventWait() waits until at least one
1141 : : * event occurs.
1142 : : */
1143 : 21 : node->as_syncdone = true;
1144 : : for (;;)
1145 : 2 : {
1146 : 23 : bool found = false;
1147 : :
1148 : : /*
1149 : : * When called from ExecAppendAsyncEventWait(), postgres_fdw (and
1150 : : * possibly other FDWs) will skip configuration of events for pending
1151 : : * requests in some cases if as_needrequest isn't empty. To avoid
1152 : : * that, discard results we already have. Note that we need to do
1153 : : * this on every iteration, as the call to that function may produce
1154 : : * new results.
1155 : : */
1156 : 23 : node->as_nasyncresults = 0;
1157 : 23 : bms_free(node->as_needrequest);
1158 : 23 : node->as_needrequest = NULL;
1159 : :
1160 : 23 : i = -1;
1161 [ + + ]: 71 : while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
1162 : : {
1163 : 50 : AsyncRequest *areq = node->as_asyncrequests[i];
1164 : :
1165 [ + + ]: 50 : if (areq->callback_pending)
1166 : : {
1167 : 2 : found = true;
1168 : 2 : break;
1169 : : }
1170 : : }
1171 [ + + ]: 23 : if (!found)
1172 : 21 : break;
1173 : :
1174 [ - + ]: 2 : CHECK_FOR_INTERRUPTS();
1175 : :
1176 : : /* Wait or poll for async events. */
1177 : 2 : ExecAppendAsyncEventWait(node);
1178 : : }
1179 : :
1180 : : /* Reset async requests. */
1181 : 21 : i = -1;
1182 [ + + ]: 67 : while ((i = bms_next_member(node->as_asyncplans, i)) >= 0)
1183 : : {
1184 : 46 : AsyncRequest *areq = node->as_asyncrequests[i];
1185 : :
1186 : : Assert(!areq->callback_pending);
1187 : 46 : areq->request_complete = false;
1188 : 46 : areq->result = NULL;
1189 : : }
1190 : :
1191 : : /* Reset state variables. */
1192 : : Assert(node->as_nasyncresults == 0);
1193 : : Assert(node->as_needrequest == NULL);
1194 : 21 : node->as_nasyncremain = 0;
1195 : 21 : }
1196 : :
1197 : : /* ----------------------------------------------------------------
1198 : : * ExecAsyncAppendResponse
1199 : : *
1200 : : * Receive a response from an asynchronous request we made.
1201 : : * ----------------------------------------------------------------
1202 : : */
1203 : : void
1204 : 6346 : ExecAsyncAppendResponse(AsyncRequest *areq)
1205 : : {
1206 : 6346 : AppendState *node = (AppendState *) areq->requestor;
1207 : 6346 : TupleTableSlot *slot = areq->result;
1208 : :
1209 : : /* The result should be a TupleTableSlot or NULL. */
1210 : : Assert(slot == NULL || IsA(slot, TupleTableSlot));
1211 : :
1212 : : /* Nothing to do if the request is pending. */
1213 [ + + ]: 6346 : if (!areq->request_complete)
1214 : : {
1215 : : /* The request would have been pending for a callback. */
1216 : : Assert(areq->callback_pending);
1217 : 168 : return;
1218 : : }
1219 : :
1220 : : /* If the result is NULL or an empty slot, there's nothing more to do. */
1221 [ + + - + ]: 6178 : if (TupIsNull(slot))
1222 : : {
1223 : : /* The ending subplan wouldn't have been pending for a callback. */
1224 : : Assert(!areq->callback_pending);
1225 : 62 : --node->as_nasyncremain;
1226 : 62 : return;
1227 : : }
1228 : :
1229 : : /* Save result so we can return it. */
1230 : : Assert(node->as_nasyncresults < node->as_nasyncplans);
1231 : 6116 : node->as_asyncresults[node->as_nasyncresults++] = slot;
1232 : :
1233 : : /*
1234 : : * Mark the subplan that returned a result as ready for a new request. We
1235 : : * don't launch another one here immediately because it might complete.
1236 : : */
1237 : 6116 : node->as_needrequest = bms_add_member(node->as_needrequest,
1238 : : areq->request_index);
1239 : : }
1240 : :
1241 : : /* ----------------------------------------------------------------
1242 : : * classify_matching_subplans
1243 : : *
1244 : : * Classify the node's as_valid_subplans into sync ones and
1245 : : * async ones, adjust it to contain sync ones only, and save
1246 : : * async ones in the node's as_valid_asyncplans.
1247 : : * ----------------------------------------------------------------
1248 : : */
1249 : : static void
1250 : 50 : classify_matching_subplans(AppendState *node)
1251 : : {
1252 : : Bitmapset *valid_asyncplans;
1253 : :
1254 : : Assert(node->as_valid_subplans_identified);
1255 : : Assert(node->as_valid_asyncplans == NULL);
1256 : :
1257 : : /* Nothing to do if there are no valid subplans. */
1258 [ - + ]: 50 : if (bms_is_empty(node->as_valid_subplans))
1259 : : {
1260 : 0 : node->as_syncdone = true;
1261 : 0 : node->as_nasyncremain = 0;
1262 : 0 : return;
1263 : : }
1264 : :
1265 : : /* Nothing to do if there are no valid async subplans. */
1266 [ - + ]: 50 : if (!bms_overlap(node->as_valid_subplans, node->as_asyncplans))
1267 : : {
1268 : 0 : node->as_nasyncremain = 0;
1269 : 0 : return;
1270 : : }
1271 : :
1272 : : /* Get valid async subplans. */
1273 : 50 : valid_asyncplans = bms_intersect(node->as_asyncplans,
1274 : 50 : node->as_valid_subplans);
1275 : :
1276 : : /* Adjust the valid subplans to contain sync subplans only. */
1277 : 50 : node->as_valid_subplans = bms_del_members(node->as_valid_subplans,
1278 : : valid_asyncplans);
1279 : :
1280 : : /* Save valid async subplans. */
1281 : 50 : node->as_valid_asyncplans = valid_asyncplans;
1282 : : }
|
