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