/
githubmirror
/
postgres
Обзор
Документация
Войти
/
githubmirror
/
postgres
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/backend/executor/nodeAppend.c
1 282 строки
36 KB
Etsuro Fujita
Drain pending asynchronous requests during ExecReScanAppend.
07 авг 2026, 11:30
07 авг 2026, 11:30
f57df20
Код
Авторство
О чём код?
/*------------------------------------------------------------------------- * * nodeAppend.c * routines to handle append nodes. * * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * * IDENTIFICATION * src/backend/executor/nodeAppend.c * *------------------------------------------------------------------------- */ /* * INTERFACE ROUTINES * ExecInitAppend - initialize the append node * ExecAppend - retrieve the next tuple from the node * ExecEndAppend - shut down the append node * ExecReScanAppend - rescan the append node * * NOTES * Each append node contains a list of one or more subplans which * must be iteratively processed (forwards or backwards). * Tuples are retrieved by executing the 'whichplan'th subplan * until the subplan stops returning tuples, at which point that * plan is shut down and the next started up. * * Append nodes don't make use of their left and right * subtrees, rather they maintain a list of subplans so * a typical append node looks like this in the plan tree: * * ... * / * Append -------+------+------+--- nil * / \ | | | * nil nil ... ... ... * subplans * * Append nodes are currently used for unions, and to support * inheritance queries, where several relations need to be scanned. * For example, in our standard person/student/employee/student-emp * example, where student and employee inherit from person * and student-emp inherits from student and employee, the * query: * * select name from person * * generates the plan: * * | * Append -------+-------+--------+--------+ * / \ | | | | * nil nil Scan Scan Scan Scan * | | | | * person employee student student-emp */ #include "postgres.h" #include "executor/execAsync.h" #include "executor/execPartition.h" #include "executor/executor.h" #include "executor/nodeAppend.h" #include "miscadmin.h" #include "pgstat.h" #include "storage/latch.h" #include "storage/lwlock.h" #include "utils/wait_event.h" /* Shared state for parallel-aware Append. */ struct ParallelAppendState { LWLock pa_lock; /* mutual exclusion to choose next subplan */ int pa_next_plan; /* next plan to choose by any worker */ /* * pa_finished[i] should be true if no more workers should select subplan * i. for a non-partial plan, this should be set to true as soon as a * worker selects the plan; for a partial plan, it remains false until * some worker executes the plan to completion. */ bool pa_finished[FLEXIBLE_ARRAY_MEMBER]; }; #define INVALID_SUBPLAN_INDEX -1 #define EVENT_BUFFER_SIZE 16 static TupleTableSlot *ExecAppend(PlanState *pstate); static bool choose_next_subplan_locally(AppendState *node); static bool choose_next_subplan_for_leader(AppendState *node); static bool choose_next_subplan_for_worker(AppendState *node); static void mark_invalid_subplans_as_finished(AppendState *node); static void ExecAppendAsyncBegin(AppendState *node); static bool ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result); static bool ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result); static void ExecAppendAsyncEventWait(AppendState *node); static void ExecAppendAsyncReset(AppendState *node); static void classify_matching_subplans(AppendState *node); /* ---------------------------------------------------------------- * ExecInitAppend * * Begin all of the subscans of the append node. * * (This is potentially wasteful, since the entire result of the * append node may not be scanned, but this way all of the * structures get allocated in the executor's top level memory * block instead of that of the call to ExecAppend.) * ---------------------------------------------------------------- */ AppendState * ExecInitAppend(Append *node, EState *estate, int eflags) { AppendState *appendstate = makeNode(AppendState); PlanState **appendplanstates; const TupleTableSlotOps *appendops; Bitmapset *validsubplans; Bitmapset *asyncplans; int nplans; int nasyncplans; int firstvalid; int i, j; /* check for unsupported flags */ Assert(!(eflags & EXEC_FLAG_MARK)); /* * create new AppendState for our append node */ appendstate->ps.plan = (Plan *) node; appendstate->ps.state = estate; appendstate->ps.ExecProcNode = ExecAppend; /* Let choose_next_subplan_* function handle setting the first subplan */ appendstate->as_whichplan = INVALID_SUBPLAN_INDEX; appendstate->as_syncdone = false; appendstate->as_begun = false; /* If run-time partition pruning is enabled, then set that up now */ if (node->part_prune_index >= 0) { PartitionPruneState *prunestate; /* * Set up pruning data structure. This also initializes the set of * subplans to initialize (validsubplans) by taking into account the * result of performing initial pruning if any. */ prunestate = ExecInitPartitionExecPruning(&appendstate->ps, list_length(node->appendplans), node->part_prune_index, node->apprelids, &validsubplans); appendstate->as_prune_state = prunestate; nplans = bms_num_members(validsubplans); /* * When no run-time pruning is required and there's at least one * subplan, we can fill as_valid_subplans immediately, preventing * later calls to ExecFindMatchingSubPlans. */ if (!prunestate->do_exec_prune && nplans > 0) { appendstate->as_valid_subplans = bms_add_range(NULL, 0, nplans - 1); appendstate->as_valid_subplans_identified = true; } } else { nplans = list_length(node->appendplans); /* * When run-time partition pruning is not enabled we can just mark all * subplans as valid; they must also all be initialized. */ Assert(nplans > 0); appendstate->as_valid_subplans = validsubplans = bms_add_range(NULL, 0, nplans - 1); appendstate->as_valid_subplans_identified = true; appendstate->as_prune_state = NULL; } appendplanstates = (PlanState **) palloc(nplans * sizeof(PlanState *)); /* * call ExecInitNode on each of the valid plans to be executed and save * the results into the appendplanstates array. * * While at it, find out the first valid partial plan. */ j = 0; asyncplans = NULL; nasyncplans = 0; firstvalid = nplans; i = -1; while ((i = bms_next_member(validsubplans, i)) >= 0) { Plan *initNode = (Plan *) list_nth(node->appendplans, i); /* * Record async subplans. When executing EvalPlanQual, we treat them * as sync ones; don't do this when initializing an EvalPlanQual plan * tree. */ if (initNode->async_capable && estate->es_epq_active == NULL) { asyncplans = bms_add_member(asyncplans, j); nasyncplans++; } /* * Record the lowest appendplans index which is a valid partial plan. */ if (i >= node->first_partial_plan && j < firstvalid) firstvalid = j; appendplanstates[j++] = ExecInitNode(initNode, estate, eflags); } appendstate->as_first_partial_plan = firstvalid; appendstate->appendplans = appendplanstates; appendstate->as_nplans = nplans; /* * Initialize Append's result tuple type and slot. If the child plans all * produce the same fixed slot type, we can use that slot type; otherwise * make a virtual slot. (Note that the result slot itself is used only to * return a null tuple at end of execution; real tuples are returned to * the caller in the children's own result slots. What we are doing here * is allowing the parent plan node to optimize if the Append will return * only one kind of slot.) */ appendops = ExecGetCommonSlotOps(appendplanstates, j); if (appendops != NULL) { ExecInitResultTupleSlotTL(&appendstate->ps, appendops); } else { ExecInitResultTupleSlotTL(&appendstate->ps, &TTSOpsVirtual); /* show that the output slot type is not fixed */ appendstate->ps.resultopsset = true; appendstate->ps.resultopsfixed = false; } /* Initialize async state */ appendstate->as_asyncplans = asyncplans; appendstate->as_nasyncplans = nasyncplans; appendstate->as_asyncrequests = NULL; appendstate->as_asyncresults = NULL; appendstate->as_nasyncresults = 0; appendstate->as_nasyncremain = 0; appendstate->as_needrequest = NULL; appendstate->as_eventset = NULL; appendstate->as_valid_asyncplans = NULL; if (nasyncplans > 0) { appendstate->as_asyncrequests = (AsyncRequest **) palloc0(nplans * sizeof(AsyncRequest *)); i = -1; while ((i = bms_next_member(asyncplans, i)) >= 0) { AsyncRequest *areq; areq = palloc_object(AsyncRequest); areq->requestor = (PlanState *) appendstate; areq->requestee = appendplanstates[i]; areq->request_index = i; areq->callback_pending = false; areq->request_complete = false; areq->result = NULL; appendstate->as_asyncrequests[i] = areq; } appendstate->as_asyncresults = (TupleTableSlot **) palloc0(nasyncplans * sizeof(TupleTableSlot *)); if (appendstate->as_valid_subplans_identified) classify_matching_subplans(appendstate); } /* * Miscellaneous initialization */ appendstate->ps.ps_ProjInfo = NULL; /* For parallel query, this will be overridden later. */ appendstate->choose_next_subplan = choose_next_subplan_locally; return appendstate; } /* ---------------------------------------------------------------- * ExecAppend * * Handles iteration over multiple subplans. * ---------------------------------------------------------------- */ static TupleTableSlot * ExecAppend(PlanState *pstate) { AppendState *node = castNode(AppendState, pstate); TupleTableSlot *result; /* * If this is the first call after Init or ReScan, we need to do the * initialization work. */ if (!node->as_begun) { Assert(node->as_whichplan == INVALID_SUBPLAN_INDEX); Assert(!node->as_syncdone); /* Nothing to do if there are no subplans */ if (node->as_nplans == 0) return ExecClearTuple(node->ps.ps_ResultTupleSlot); /* If there are any async subplans, begin executing them. */ if (node->as_nasyncplans > 0) ExecAppendAsyncBegin(node); /* * If no sync subplan has been chosen, we must choose one before * proceeding. */ if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0) return ExecClearTuple(node->ps.ps_ResultTupleSlot); Assert(node->as_syncdone || (node->as_whichplan >= 0 && node->as_whichplan < node->as_nplans)); /* And we're initialized. */ node->as_begun = true; } for (;;) { PlanState *subnode; CHECK_FOR_INTERRUPTS(); /* * try to get a tuple from an async subplan if any */ if (node->as_syncdone || !bms_is_empty(node->as_needrequest)) { if (ExecAppendAsyncGetNext(node, &result)) return result; Assert(!node->as_syncdone); Assert(bms_is_empty(node->as_needrequest)); } /* * figure out which sync subplan we are currently processing */ Assert(node->as_whichplan >= 0 && node->as_whichplan < node->as_nplans); subnode = node->appendplans[node->as_whichplan]; /* * get a tuple from the subplan */ result = ExecProcNode(subnode); if (!TupIsNull(result)) { /* * If the subplan gave us something then return it as-is. We do * NOT make use of the result slot that was set up in * ExecInitAppend; there's no need for it. */ return result; } /* * wait or poll for async events if any. We do this before checking * for the end of iteration, because it might drain the remaining * async subplans. */ if (node->as_nasyncremain > 0) ExecAppendAsyncEventWait(node); /* choose new sync subplan; if no sync/async subplans, we're done */ if (!node->choose_next_subplan(node) && node->as_nasyncremain == 0) return ExecClearTuple(node->ps.ps_ResultTupleSlot); } } /* ---------------------------------------------------------------- * ExecEndAppend * * Shuts down the subscans of the append node. * * Returns nothing of interest. * ---------------------------------------------------------------- */ void ExecEndAppend(AppendState *node) { PlanState **appendplans; int nplans; int i; /* * get information from the node */ appendplans = node->appendplans; nplans = node->as_nplans; /* * shut down each of the subscans */ for (i = 0; i < nplans; i++) ExecEndNode(appendplans[i]); } void ExecReScanAppend(AppendState *node) { int nasyncplans = node->as_nasyncplans; int i; /* If there are any async subplans, reset async requests made for them. */ if (nasyncplans > 0) ExecAppendAsyncReset(node); /* * If any PARAM_EXEC Params used in pruning expressions have changed, then * we'd better unset the valid subplans so that they are reselected for * the new parameter values. */ if (node->as_prune_state && bms_overlap(node->ps.chgParam, node->as_prune_state->execparamids)) { node->as_valid_subplans_identified = false; bms_free(node->as_valid_subplans); node->as_valid_subplans = NULL; bms_free(node->as_valid_asyncplans); node->as_valid_asyncplans = NULL; } for (i = 0; i < node->as_nplans; i++) { PlanState *subnode = node->appendplans[i]; /* * ExecReScan doesn't know about my subplans, so I have to do * changed-parameter signaling myself. */ if (node->ps.chgParam != NULL) UpdateChangedParamSet(subnode, node->ps.chgParam); /* * If chgParam of subnode is not null then plan will be re-scanned by * first ExecProcNode or by first ExecAsyncRequest. */ if (subnode->chgParam == NULL) ExecReScan(subnode); } /* Let choose_next_subplan_* function handle setting the first subplan */ node->as_whichplan = INVALID_SUBPLAN_INDEX; node->as_syncdone = false; node->as_begun = false; } /* ---------------------------------------------------------------- * Parallel Append Support * ---------------------------------------------------------------- */ /* ---------------------------------------------------------------- * ExecAppendEstimate * * Compute the amount of space we'll need in the parallel * query DSM, and inform pcxt->estimator about our needs. * ---------------------------------------------------------------- */ void ExecAppendEstimate(AppendState *node, ParallelContext *pcxt) { node->pstate_len = add_size(offsetof(ParallelAppendState, pa_finished), sizeof(bool) * node->as_nplans); shm_toc_estimate_chunk(&pcxt->estimator, node->pstate_len); shm_toc_estimate_keys(&pcxt->estimator, 1); } /* ---------------------------------------------------------------- * ExecAppendInitializeDSM * * Set up shared state for Parallel Append. * ---------------------------------------------------------------- */ void ExecAppendInitializeDSM(AppendState *node, ParallelContext *pcxt) { ParallelAppendState *pstate; pstate = shm_toc_allocate(pcxt->toc, node->pstate_len); memset(pstate, 0, node->pstate_len); LWLockInitialize(&pstate->pa_lock, LWTRANCHE_PARALLEL_APPEND); shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id, pstate); node->as_pstate = pstate; node->choose_next_subplan = choose_next_subplan_for_leader; } /* ---------------------------------------------------------------- * ExecAppendReInitializeDSM * * Reset shared state before beginning a fresh scan. * ---------------------------------------------------------------- */ void ExecAppendReInitializeDSM(AppendState *node, ParallelContext *pcxt) { ParallelAppendState *pstate = node->as_pstate; pstate->pa_next_plan = 0; memset(pstate->pa_finished, 0, sizeof(bool) * node->as_nplans); } /* ---------------------------------------------------------------- * ExecAppendInitializeWorker * * Copy relevant information from TOC into planstate, and initialize * whatever is required to choose and execute the optimal subplan. * ---------------------------------------------------------------- */ void ExecAppendInitializeWorker(AppendState *node, ParallelWorkerContext *pwcxt) { node->as_pstate = shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false); node->choose_next_subplan = choose_next_subplan_for_worker; } /* ---------------------------------------------------------------- * choose_next_subplan_locally * * Choose next sync subplan for a non-parallel-aware Append, * returning false if there are no more. * ---------------------------------------------------------------- */ static bool choose_next_subplan_locally(AppendState *node) { int whichplan = node->as_whichplan; int nextplan; /* We should never be called when there are no subplans */ Assert(node->as_nplans > 0); /* Nothing to do if syncdone */ if (node->as_syncdone) return false; /* * If first call then have the bms member function choose the first valid * sync subplan by initializing whichplan to -1. If there happen to be no * valid sync subplans then the bms member function will handle that by * returning a negative number which will allow us to exit returning a * false value. */ if (whichplan == INVALID_SUBPLAN_INDEX) { if (node->as_nasyncplans > 0) { /* We'd have filled as_valid_subplans already */ Assert(node->as_valid_subplans_identified); } else if (!node->as_valid_subplans_identified) { node->as_valid_subplans = ExecFindMatchingSubPlans(node->as_prune_state, false, NULL); node->as_valid_subplans_identified = true; } whichplan = -1; } /* Ensure whichplan is within the expected range */ Assert(whichplan >= -1 && whichplan <= node->as_nplans); if (ScanDirectionIsForward(node->ps.state->es_direction)) nextplan = bms_next_member(node->as_valid_subplans, whichplan); else nextplan = bms_prev_member(node->as_valid_subplans, whichplan); if (nextplan < 0) { /* Set as_syncdone if in async mode */ if (node->as_nasyncplans > 0) node->as_syncdone = true; return false; } node->as_whichplan = nextplan; return true; } /* ---------------------------------------------------------------- * choose_next_subplan_for_leader * * Try to pick a plan which doesn't commit us to doing much * work locally, so that as much work as possible is done in * the workers. Cheapest subplans are at the end. * ---------------------------------------------------------------- */ static bool choose_next_subplan_for_leader(AppendState *node) { ParallelAppendState *pstate = node->as_pstate; /* Backward scan is not supported by parallel-aware plans */ Assert(ScanDirectionIsForward(node->ps.state->es_direction)); /* We should never be called when there are no subplans */ Assert(node->as_nplans > 0); LWLockAcquire(&pstate->pa_lock, LW_EXCLUSIVE); if (node->as_whichplan != INVALID_SUBPLAN_INDEX) { /* Mark just-completed subplan as finished. */ node->as_pstate->pa_finished[node->as_whichplan] = true; } else { /* Start with last subplan. */ node->as_whichplan = node->as_nplans - 1; /* * If we've yet to determine the valid subplans then do so now. If * run-time pruning is disabled then the valid subplans will always be * set to all subplans. */ if (!node->as_valid_subplans_identified) { node->as_valid_subplans = ExecFindMatchingSubPlans(node->as_prune_state, false, NULL); node->as_valid_subplans_identified = true; /* * Mark each invalid plan as finished to allow the loop below to * select the first valid subplan. */ mark_invalid_subplans_as_finished(node); } } /* Loop until we find a subplan to execute. */ while (pstate->pa_finished[node->as_whichplan]) { if (node->as_whichplan == 0) { pstate->pa_next_plan = INVALID_SUBPLAN_INDEX; node->as_whichplan = INVALID_SUBPLAN_INDEX; LWLockRelease(&pstate->pa_lock); return false; } /* * We needn't pay attention to as_valid_subplans here as all invalid * plans have been marked as finished. */ node->as_whichplan--; } /* If non-partial, immediately mark as finished. */ if (node->as_whichplan < node->as_first_partial_plan) node->as_pstate->pa_finished[node->as_whichplan] = true; LWLockRelease(&pstate->pa_lock); return true; } /* ---------------------------------------------------------------- * choose_next_subplan_for_worker * * Choose next subplan for a parallel-aware Append, returning * false if there are no more. * * We start from the first plan and advance through the list; * when we get back to the end, we loop back to the first * partial plan. This assigns the non-partial plans first in * order of descending cost and then spreads out the workers * as evenly as possible across the remaining partial plans. * ---------------------------------------------------------------- */ static bool choose_next_subplan_for_worker(AppendState *node) { ParallelAppendState *pstate = node->as_pstate; /* Backward scan is not supported by parallel-aware plans */ Assert(ScanDirectionIsForward(node->ps.state->es_direction)); /* We should never be called when there are no subplans */ Assert(node->as_nplans > 0); LWLockAcquire(&pstate->pa_lock, LW_EXCLUSIVE); /* Mark just-completed subplan as finished. */ if (node->as_whichplan != INVALID_SUBPLAN_INDEX) node->as_pstate->pa_finished[node->as_whichplan] = true; /* * If we've yet to determine the valid subplans then do so now. If * run-time pruning is disabled then the valid subplans will always be set * to all subplans. */ else if (!node->as_valid_subplans_identified) { node->as_valid_subplans = ExecFindMatchingSubPlans(node->as_prune_state, false, NULL); node->as_valid_subplans_identified = true; mark_invalid_subplans_as_finished(node); } /* If all the plans are already done, we have nothing to do */ if (pstate->pa_next_plan == INVALID_SUBPLAN_INDEX) { LWLockRelease(&pstate->pa_lock); return false; } /* Save the plan from which we are starting the search. */ node->as_whichplan = pstate->pa_next_plan; /* Loop until we find a valid subplan to execute. */ while (pstate->pa_finished[pstate->pa_next_plan]) { int nextplan; nextplan = bms_next_member(node->as_valid_subplans, pstate->pa_next_plan); if (nextplan >= 0) { /* Advance to the next valid plan. */ pstate->pa_next_plan = nextplan; } else if (node->as_whichplan > node->as_first_partial_plan) { /* * Try looping back to the first valid partial plan, if there is * one. If there isn't, arrange to bail out below. */ nextplan = bms_next_member(node->as_valid_subplans, node->as_first_partial_plan - 1); pstate->pa_next_plan = nextplan < 0 ? node->as_whichplan : nextplan; } else { /* * At last plan, and either there are no partial plans or we've * tried them all. Arrange to bail out. */ pstate->pa_next_plan = node->as_whichplan; } if (pstate->pa_next_plan == node->as_whichplan) { /* We've tried everything! */ pstate->pa_next_plan = INVALID_SUBPLAN_INDEX; LWLockRelease(&pstate->pa_lock); return false; } } /* Pick the plan we found, and advance pa_next_plan one more time. */ node->as_whichplan = pstate->pa_next_plan; pstate->pa_next_plan = bms_next_member(node->as_valid_subplans, pstate->pa_next_plan); /* * If there are no more valid plans then try setting the next plan to the * first valid partial plan. */ if (pstate->pa_next_plan < 0) { int nextplan = bms_next_member(node->as_valid_subplans, node->as_first_partial_plan - 1); if (nextplan >= 0) pstate->pa_next_plan = nextplan; else { /* * There are no valid partial plans, and we already chose the last * non-partial plan; so flag that there's nothing more for our * fellow workers to do. */ pstate->pa_next_plan = INVALID_SUBPLAN_INDEX; } } /* If non-partial, immediately mark as finished. */ if (node->as_whichplan < node->as_first_partial_plan) node->as_pstate->pa_finished[node->as_whichplan] = true; LWLockRelease(&pstate->pa_lock); return true; } /* * mark_invalid_subplans_as_finished * Marks the ParallelAppendState's pa_finished as true for each invalid * subplan. * * This function should only be called for parallel Append with run-time * pruning enabled. */ static void mark_invalid_subplans_as_finished(AppendState *node) { int i; /* Only valid to call this while in parallel Append mode */ Assert(node->as_pstate); /* Shouldn't have been called when run-time pruning is not enabled */ Assert(node->as_prune_state); /* Nothing to do if all plans are valid */ if (bms_num_members(node->as_valid_subplans) == node->as_nplans) return; /* Mark all non-valid plans as finished */ for (i = 0; i < node->as_nplans; i++) { if (!bms_is_member(i, node->as_valid_subplans)) node->as_pstate->pa_finished[i] = true; } } /* ---------------------------------------------------------------- * Asynchronous Append Support * ---------------------------------------------------------------- */ /* ---------------------------------------------------------------- * ExecAppendAsyncBegin * * Begin executing designed async-capable subplans. * ---------------------------------------------------------------- */ static void ExecAppendAsyncBegin(AppendState *node) { int i; /* Backward scan is not supported by async-aware Appends. */ Assert(ScanDirectionIsForward(node->ps.state->es_direction)); /* We should never be called when there are no subplans */ Assert(node->as_nplans > 0); /* We should never be called when there are no async subplans. */ Assert(node->as_nasyncplans > 0); /* If we've yet to determine the valid subplans then do so now. */ if (!node->as_valid_subplans_identified) { node->as_valid_subplans = ExecFindMatchingSubPlans(node->as_prune_state, false, NULL); node->as_valid_subplans_identified = true; classify_matching_subplans(node); } /* Initialize state variables. */ node->as_syncdone = bms_is_empty(node->as_valid_subplans); node->as_nasyncremain = bms_num_members(node->as_valid_asyncplans); /* Nothing to do if there are no valid async subplans. */ if (node->as_nasyncremain == 0) return; /* Make a request for each of the valid async subplans. */ i = -1; while ((i = bms_next_member(node->as_valid_asyncplans, i)) >= 0) { AsyncRequest *areq = node->as_asyncrequests[i]; Assert(areq->request_index == i); Assert(!areq->callback_pending); /* Do the actual work. */ ExecAsyncRequest(areq); } } /* ---------------------------------------------------------------- * ExecAppendAsyncGetNext * * Get the next tuple from any of the asynchronous subplans. * ---------------------------------------------------------------- */ static bool ExecAppendAsyncGetNext(AppendState *node, TupleTableSlot **result) { *result = NULL; /* We should never be called when there are no valid async subplans. */ Assert(node->as_nasyncremain > 0); /* Request a tuple asynchronously. */ if (ExecAppendAsyncRequest(node, result)) return true; while (node->as_nasyncremain > 0) { CHECK_FOR_INTERRUPTS(); /* Wait or poll for async events. */ ExecAppendAsyncEventWait(node); /* Request a tuple asynchronously. */ if (ExecAppendAsyncRequest(node, result)) return true; /* Break from loop if there's any sync subplan that isn't complete. */ if (!node->as_syncdone) break; } /* * If all sync subplans are complete, we're totally done scanning the * given node. Otherwise, we're done with the asynchronous stuff but must * continue scanning the sync subplans. */ if (node->as_syncdone) { Assert(node->as_nasyncremain == 0); *result = ExecClearTuple(node->ps.ps_ResultTupleSlot); return true; } return false; } /* ---------------------------------------------------------------- * ExecAppendAsyncRequest * * Request a tuple asynchronously. * ---------------------------------------------------------------- */ static bool ExecAppendAsyncRequest(AppendState *node, TupleTableSlot **result) { Bitmapset *needrequest; int i; /* Nothing to do if there are no async subplans needing a new request. */ if (bms_is_empty(node->as_needrequest)) { Assert(node->as_nasyncresults == 0); return false; } /* * If there are any asynchronously-generated results that have not yet * been returned, we have nothing to do; just return one of them. */ if (node->as_nasyncresults > 0) { --node->as_nasyncresults; *result = node->as_asyncresults[node->as_nasyncresults]; return true; } /* Make a new request for each of the async subplans that need it. */ needrequest = node->as_needrequest; node->as_needrequest = NULL; i = -1; while ((i = bms_next_member(needrequest, i)) >= 0) { AsyncRequest *areq = node->as_asyncrequests[i]; /* Do the actual work. */ ExecAsyncRequest(areq); } bms_free(needrequest); /* Return one of the asynchronously-generated results if any. */ if (node->as_nasyncresults > 0) { --node->as_nasyncresults; *result = node->as_asyncresults[node->as_nasyncresults]; return true; } return false; } /* ---------------------------------------------------------------- * ExecAppendAsyncEventWait * * Wait or poll for file descriptor events and fire callbacks. * ---------------------------------------------------------------- */ static void ExecAppendAsyncEventWait(AppendState *node) { int nevents = node->as_nasyncplans + 2; long timeout = node->as_syncdone ? -1 : 0; WaitEvent occurred_event[EVENT_BUFFER_SIZE]; int noccurred; int i; /* We should never be called when there are no valid async subplans. */ Assert(node->as_nasyncremain > 0); Assert(node->as_eventset == NULL); node->as_eventset = CreateWaitEventSet(CurrentResourceOwner, nevents); AddWaitEventToSet(node->as_eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, NULL, NULL); /* Give each waiting subplan a chance to add an event. */ i = -1; while ((i = bms_next_member(node->as_asyncplans, i)) >= 0) { AsyncRequest *areq = node->as_asyncrequests[i]; if (areq->callback_pending) ExecAsyncConfigureWait(areq); } /* * No need for further processing if none of the subplans configured any * events. */ if (GetNumRegisteredWaitEvents(node->as_eventset) == 1) { FreeWaitEventSet(node->as_eventset); node->as_eventset = NULL; return; } /* * Add the process latch to the set, so that we wake up to process the * standard interrupts with CHECK_FOR_INTERRUPTS(). * * NOTE: For historical reasons, it's important that this is added to the * WaitEventSet after the ExecAsyncConfigureWait() calls. Namely, * postgres_fdw calls "GetNumRegisteredWaitEvents(set) == 1" to check if * any other events are in the set. That's a poor design, it's * questionable for postgres_fdw to be doing that in the first place, but * we cannot change it now. The pattern has possibly been copied to other * extensions too. */ AddWaitEventToSet(node->as_eventset, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, NULL); /* Return at most EVENT_BUFFER_SIZE events in one call. */ if (nevents > EVENT_BUFFER_SIZE) nevents = EVENT_BUFFER_SIZE; /* * If the timeout is -1, wait until at least one event occurs. If the * timeout is 0, poll for events, but do not wait at all. */ noccurred = WaitEventSetWait(node->as_eventset, timeout, occurred_event, nevents, WAIT_EVENT_APPEND_READY); FreeWaitEventSet(node->as_eventset); node->as_eventset = NULL; if (noccurred == 0) return; /* Deliver notifications. */ for (i = 0; i < noccurred; i++) { WaitEvent *w = &occurred_event[i]; /* * Each waiting subplan should have registered its wait event with * user_data pointing back to its AsyncRequest. */ if ((w->events & WL_SOCKET_READABLE) != 0) { AsyncRequest *areq = (AsyncRequest *) w->user_data; if (areq->callback_pending) { /* * Mark it as no longer needing a callback. We must do this * before dispatching the callback in case the callback resets * the flag. */ areq->callback_pending = false; /* Do the actual work. */ ExecAsyncNotify(areq); } } /* Handle standard interrupts */ if ((w->events & WL_LATCH_SET) != 0) { ResetLatch(MyLatch); CHECK_FOR_INTERRUPTS(); } } } /* ---------------------------------------------------------------- * ExecAppendAsyncReset * * Reset asynchronous requests made for async-capable subplans. * ---------------------------------------------------------------- */ static void ExecAppendAsyncReset(AppendState *node) { int i; /* We should never be called when there are no async subplans. */ Assert(node->as_nasyncplans > 0); /* * Drain pending async requests if any. We force the as_syncdone flag to * be true so that ExecAppendAsyncEventWait() waits until at least one * event occurs. */ node->as_syncdone = true; for (;;) { bool found = false; /* * When called from ExecAppendAsyncEventWait(), postgres_fdw (and * possibly other FDWs) will skip configuration of events for pending * requests in some cases if as_needrequest isn't empty. To avoid * that, discard results we already have. Note that we need to do * this on every iteration, as the call to that function may produce * new results. */ node->as_nasyncresults = 0; bms_free(node->as_needrequest); node->as_needrequest = NULL; i = -1; while ((i = bms_next_member(node->as_asyncplans, i)) >= 0) { AsyncRequest *areq = node->as_asyncrequests[i]; if (areq->callback_pending) { found = true; break; } } if (!found) break; CHECK_FOR_INTERRUPTS(); /* Wait or poll for async events. */ ExecAppendAsyncEventWait(node); } /* Reset async requests. */ i = -1; while ((i = bms_next_member(node->as_asyncplans, i)) >= 0) { AsyncRequest *areq = node->as_asyncrequests[i]; Assert(!areq->callback_pending); areq->request_complete = false; areq->result = NULL; } /* Reset state variables. */ Assert(node->as_nasyncresults == 0); Assert(node->as_needrequest == NULL); node->as_nasyncremain = 0; } /* ---------------------------------------------------------------- * ExecAsyncAppendResponse * * Receive a response from an asynchronous request we made. * ---------------------------------------------------------------- */ void ExecAsyncAppendResponse(AsyncRequest *areq) { AppendState *node = (AppendState *) areq->requestor; TupleTableSlot *slot = areq->result; /* The result should be a TupleTableSlot or NULL. */ Assert(slot == NULL || IsA(slot, TupleTableSlot)); /* Nothing to do if the request is pending. */ if (!areq->request_complete) { /* The request would have been pending for a callback. */ Assert(areq->callback_pending); return; } /* If the result is NULL or an empty slot, there's nothing more to do. */ if (TupIsNull(slot)) { /* The ending subplan wouldn't have been pending for a callback. */ Assert(!areq->callback_pending); --node->as_nasyncremain; return; } /* Save result so we can return it. */ Assert(node->as_nasyncresults < node->as_nasyncplans); node->as_asyncresults[node->as_nasyncresults++] = slot; /* * Mark the subplan that returned a result as ready for a new request. We * don't launch another one here immediately because it might complete. */ node->as_needrequest = bms_add_member(node->as_needrequest, areq->request_index); } /* ---------------------------------------------------------------- * classify_matching_subplans * * Classify the node's as_valid_subplans into sync ones and * async ones, adjust it to contain sync ones only, and save * async ones in the node's as_valid_asyncplans. * ---------------------------------------------------------------- */ static void classify_matching_subplans(AppendState *node) { Bitmapset *valid_asyncplans; Assert(node->as_valid_subplans_identified); Assert(node->as_valid_asyncplans == NULL); /* Nothing to do if there are no valid subplans. */ if (bms_is_empty(node->as_valid_subplans)) { node->as_syncdone = true; node->as_nasyncremain = 0; return; } /* Nothing to do if there are no valid async subplans. */ if (!bms_overlap(node->as_valid_subplans, node->as_asyncplans)) { node->as_nasyncremain = 0; return; } /* Get valid async subplans. */ valid_asyncplans = bms_intersect(node->as_asyncplans, node->as_valid_subplans); /* Adjust the valid subplans to contain sync subplans only. */ node->as_valid_subplans = bms_del_members(node->as_valid_subplans, valid_asyncplans); /* Save valid async subplans. */ node->as_valid_asyncplans = valid_asyncplans; }