Move interrupt checking from ExecProcNode() to executor nodes.
In a followup commit ExecProcNode(), and especially the large switch it contains, will largely be replaced by a function pointer directly to the correct node. The node functions will then get invoked by a thin inline function wrapper. To avoid having to include miscadmin.h in headers - CHECK_FOR_INTERRUPTS() - move the interrupt checks into the individual executor routines. While looking through all executor nodes, I noticed a number of arguably missing interrupt checks, add these too. Author: Andres Freund, Tom Lane Reviewed-By: Tom Lane Discussion: https://postgr.es/m/22833.1490390175@sss.pgh.pa.us
This commit is contained in:
parent
9dea962b3e
commit
d47cfef711
29 changed files with 104 additions and 10 deletions
|
@ -399,8 +399,6 @@ ExecProcNode(PlanState *node)
|
|||
{
|
||||
TupleTableSlot *result;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (node->chgParam != NULL) /* something changed */
|
||||
ExecReScan(node); /* let ReScan handle this */
|
||||
|
||||
|
|
|
@ -677,6 +677,8 @@ fetch_input_tuple(AggState *aggstate)
|
|||
|
||||
if (aggstate->sort_in)
|
||||
{
|
||||
/* make sure we check for interrupts in either path through here */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
if (!tuplesort_gettupleslot(aggstate->sort_in, true, false,
|
||||
aggstate->sort_slot, NULL))
|
||||
return NULL;
|
||||
|
@ -1414,6 +1416,8 @@ process_ordered_aggregate_multi(AggState *aggstate,
|
|||
while (tuplesort_gettupleslot(pertrans->sortstates[aggstate->current_set],
|
||||
true, true, slot1, &newAbbrevVal))
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Extract the first numTransInputs columns as datums to pass to the
|
||||
* transfn. (This will help execTuplesMatch too, so we do it
|
||||
|
@ -2100,6 +2104,8 @@ ExecAgg(AggState *node)
|
|||
{
|
||||
TupleTableSlot *result = NULL;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (!node->agg_done)
|
||||
{
|
||||
/* Dispatch based on strategy */
|
||||
|
@ -2563,6 +2569,8 @@ agg_retrieve_hash_table(AggState *aggstate)
|
|||
TupleTableSlot *hashslot = perhash->hashslot;
|
||||
int i;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Find the next entry in the hash table
|
||||
*/
|
||||
|
|
|
@ -59,6 +59,7 @@
|
|||
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeAppend.h"
|
||||
#include "miscadmin.h"
|
||||
|
||||
static bool exec_append_initialize_next(AppendState *appendstate);
|
||||
|
||||
|
@ -204,6 +205,8 @@ ExecAppend(AppendState *node)
|
|||
PlanState *subnode;
|
||||
TupleTableSlot *result;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* figure out which subplan we are currently processing
|
||||
*/
|
||||
|
|
|
@ -41,6 +41,7 @@
|
|||
#include "access/transam.h"
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeBitmapHeapscan.h"
|
||||
#include "miscadmin.h"
|
||||
#include "pgstat.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "storage/predicate.h"
|
||||
|
@ -192,6 +193,8 @@ BitmapHeapNext(BitmapHeapScanState *node)
|
|||
Page dp;
|
||||
ItemId lp;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Get next page of results if needed
|
||||
*/
|
||||
|
|
|
@ -15,6 +15,7 @@
|
|||
#include "executor/nodeCustom.h"
|
||||
#include "nodes/execnodes.h"
|
||||
#include "nodes/plannodes.h"
|
||||
#include "miscadmin.h"
|
||||
#include "parser/parsetree.h"
|
||||
#include "utils/hsearch.h"
|
||||
#include "utils/memutils.h"
|
||||
|
@ -104,6 +105,8 @@ ExecInitCustomScan(CustomScan *cscan, EState *estate, int eflags)
|
|||
TupleTableSlot *
|
||||
ExecCustomScan(CustomScanState *node)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
Assert(node->methods->ExecCustomScan != NULL);
|
||||
return node->methods->ExecCustomScan(node);
|
||||
}
|
||||
|
|
|
@ -128,6 +128,8 @@ ExecGather(GatherState *node)
|
|||
TupleTableSlot *slot;
|
||||
ExprContext *econtext;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Initialize the parallel context and workers on first execution. We do
|
||||
* this on first execution rather than during node initialization, as it
|
||||
|
@ -247,6 +249,8 @@ gather_getnext(GatherState *gatherstate)
|
|||
|
||||
while (gatherstate->reader != NULL || gatherstate->need_to_scan_locally)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (gatherstate->reader != NULL)
|
||||
{
|
||||
MemoryContext oldContext;
|
||||
|
|
|
@ -164,6 +164,8 @@ ExecGatherMerge(GatherMergeState *node)
|
|||
ExprContext *econtext;
|
||||
int i;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* As with Gather, we don't launch workers until this node is actually
|
||||
* executed.
|
||||
|
@ -393,6 +395,8 @@ gather_merge_init(GatherMergeState *gm_state)
|
|||
reread:
|
||||
for (i = 0; i < nreaders + 1; i++)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (!gm_state->gm_tuple_buffers[i].done &&
|
||||
(TupIsNull(gm_state->gm_slots[i]) ||
|
||||
gm_state->gm_slots[i]->tts_isempty))
|
||||
|
|
|
@ -24,6 +24,7 @@
|
|||
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeGroup.h"
|
||||
#include "miscadmin.h"
|
||||
|
||||
|
||||
/*
|
||||
|
@ -40,6 +41,8 @@ ExecGroup(GroupState *node)
|
|||
TupleTableSlot *firsttupleslot;
|
||||
TupleTableSlot *outerslot;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get state info from node
|
||||
*/
|
||||
|
|
|
@ -810,6 +810,9 @@ ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
|
|||
idx += MAXALIGN(HJTUPLE_OVERHEAD +
|
||||
HJTUPLE_MINTUPLE(hashTuple)->t_len);
|
||||
}
|
||||
|
||||
/* allow this loop to be cancellable */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -1192,6 +1195,9 @@ ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
|
|||
|
||||
hashTuple = hashTuple->next;
|
||||
}
|
||||
|
||||
/* allow this loop to be cancellable */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
@ -92,6 +92,14 @@ ExecHashJoin(HashJoinState *node)
|
|||
*/
|
||||
for (;;)
|
||||
{
|
||||
/*
|
||||
* It's possible to iterate this loop many times before returning a
|
||||
* tuple, in some pathological cases such as needing to move much of
|
||||
* the current batch to a later batch. So let's check for interrupts
|
||||
* each time through.
|
||||
*/
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
switch (node->hj_JoinState)
|
||||
{
|
||||
case HJ_BUILD_HASHTABLE:
|
||||
|
@ -246,13 +254,6 @@ ExecHashJoin(HashJoinState *node)
|
|||
|
||||
case HJ_SCAN_BUCKET:
|
||||
|
||||
/*
|
||||
* We check for interrupts here because this corresponds to
|
||||
* where we'd fetch a row from a child plan node in other join
|
||||
* types.
|
||||
*/
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Scan the selected hash bucket for matches to current outer
|
||||
*/
|
||||
|
|
|
@ -34,6 +34,7 @@
|
|||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeIndexonlyscan.h"
|
||||
#include "executor/nodeIndexscan.h"
|
||||
#include "miscadmin.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "storage/predicate.h"
|
||||
#include "utils/memutils.h"
|
||||
|
@ -117,6 +118,8 @@ IndexOnlyNext(IndexOnlyScanState *node)
|
|||
{
|
||||
HeapTuple tuple = NULL;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* We can skip the heap fetch if the TID references a heap page on
|
||||
* which all tuples are known visible to everybody. In any case,
|
||||
|
|
|
@ -34,6 +34,7 @@
|
|||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeIndexscan.h"
|
||||
#include "lib/pairingheap.h"
|
||||
#include "miscadmin.h"
|
||||
#include "nodes/nodeFuncs.h"
|
||||
#include "optimizer/clauses.h"
|
||||
#include "utils/array.h"
|
||||
|
@ -131,6 +132,8 @@ IndexNext(IndexScanState *node)
|
|||
*/
|
||||
while ((tuple = index_getnext(scandesc, direction)) != NULL)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Store the scanned tuple in the scan tuple slot of the scan state.
|
||||
* Note: we pass 'false' because tuples returned by amgetnext are
|
||||
|
@ -233,6 +236,8 @@ IndexNextWithReorder(IndexScanState *node)
|
|||
|
||||
for (;;)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Check the reorder queue first. If the topmost tuple in the queue
|
||||
* has an ORDER BY value smaller than (or equal to) the value last
|
||||
|
@ -299,6 +304,8 @@ next_indextuple:
|
|||
{
|
||||
/* Fails recheck, so drop it and loop back for another */
|
||||
InstrCountFiltered2(node, 1);
|
||||
/* allow this loop to be cancellable */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
goto next_indextuple;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -23,6 +23,7 @@
|
|||
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeLimit.h"
|
||||
#include "miscadmin.h"
|
||||
#include "nodes/nodeFuncs.h"
|
||||
|
||||
static void recompute_limits(LimitState *node);
|
||||
|
@ -43,6 +44,8 @@ ExecLimit(LimitState *node)
|
|||
TupleTableSlot *slot;
|
||||
PlanState *outerPlan;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get information from the node
|
||||
*/
|
||||
|
|
|
@ -26,6 +26,7 @@
|
|||
#include "executor/executor.h"
|
||||
#include "executor/nodeLockRows.h"
|
||||
#include "foreign/fdwapi.h"
|
||||
#include "miscadmin.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "utils/rel.h"
|
||||
#include "utils/tqual.h"
|
||||
|
@ -44,6 +45,8 @@ ExecLockRows(LockRowsState *node)
|
|||
bool epq_needed;
|
||||
ListCell *lc;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get information from the node
|
||||
*/
|
||||
|
|
|
@ -45,6 +45,8 @@ ExecMaterial(MaterialState *node)
|
|||
bool eof_tuplestore;
|
||||
TupleTableSlot *slot;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get state info from node
|
||||
*/
|
||||
|
|
|
@ -40,8 +40,8 @@
|
|||
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeMergeAppend.h"
|
||||
|
||||
#include "lib/binaryheap.h"
|
||||
#include "miscadmin.h"
|
||||
|
||||
/*
|
||||
* We have one slot for each item in the heap array. We use SlotNumber
|
||||
|
@ -175,6 +175,8 @@ ExecMergeAppend(MergeAppendState *node)
|
|||
TupleTableSlot *result;
|
||||
SlotNumber i;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (!node->ms_initialized)
|
||||
{
|
||||
/*
|
||||
|
|
|
@ -95,6 +95,7 @@
|
|||
#include "access/nbtree.h"
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeMergejoin.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/lsyscache.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
@ -610,6 +611,8 @@ ExecMergeJoin(MergeJoinState *node)
|
|||
bool doFillOuter;
|
||||
bool doFillInner;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get information from node
|
||||
*/
|
||||
|
|
|
@ -1551,6 +1551,8 @@ ExecModifyTable(ModifyTableState *node)
|
|||
HeapTupleData oldtupdata;
|
||||
HeapTuple oldtuple;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* This should NOT get called during EvalPlanQual; we should have passed a
|
||||
* subplan tree to EvalPlanQual, instead. Use a runtime test not just
|
||||
|
|
|
@ -23,6 +23,7 @@
|
|||
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeNestloop.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
||||
|
@ -69,6 +70,8 @@ ExecNestLoop(NestLoopState *node)
|
|||
ExprContext *econtext;
|
||||
ListCell *lc;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get information from the node
|
||||
*/
|
||||
|
|
|
@ -24,6 +24,7 @@
|
|||
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeProjectSet.h"
|
||||
#include "miscadmin.h"
|
||||
#include "nodes/nodeFuncs.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
@ -46,6 +47,8 @@ ExecProjectSet(ProjectSetState *node)
|
|||
PlanState *outerPlan;
|
||||
ExprContext *econtext;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
econtext = node->ps.ps_ExprContext;
|
||||
|
||||
/*
|
||||
|
|
|
@ -75,6 +75,8 @@ ExecRecursiveUnion(RecursiveUnionState *node)
|
|||
TupleTableSlot *slot;
|
||||
bool isnew;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* 1. Evaluate non-recursive term */
|
||||
if (!node->recursing)
|
||||
{
|
||||
|
|
|
@ -47,6 +47,7 @@
|
|||
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeResult.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
||||
|
@ -70,6 +71,8 @@ ExecResult(ResultState *node)
|
|||
PlanState *outerPlan;
|
||||
ExprContext *econtext;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
econtext = node->ps.ps_ExprContext;
|
||||
|
||||
/*
|
||||
|
|
|
@ -47,6 +47,7 @@
|
|||
#include "access/htup_details.h"
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeSetOp.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
||||
|
@ -185,6 +186,8 @@ ExecSetOp(SetOpState *node)
|
|||
SetOp *plannode = (SetOp *) node->ps.plan;
|
||||
TupleTableSlot *resultTupleSlot = node->ps.ps_ResultTupleSlot;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* If the previously-returned tuple needs to be returned more than once,
|
||||
* keep returning it.
|
||||
|
@ -428,6 +431,8 @@ setop_retrieve_hash_table(SetOpState *setopstate)
|
|||
*/
|
||||
while (!setopstate->setop_done)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Find the next entry in the hash table
|
||||
*/
|
||||
|
|
|
@ -43,6 +43,8 @@ ExecSort(SortState *node)
|
|||
Tuplesortstate *tuplesortstate;
|
||||
TupleTableSlot *slot;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get state info from node
|
||||
*/
|
||||
|
|
|
@ -33,6 +33,7 @@
|
|||
#include "executor/executor.h"
|
||||
#include "executor/nodeSubplan.h"
|
||||
#include "nodes/makefuncs.h"
|
||||
#include "miscadmin.h"
|
||||
#include "optimizer/clauses.h"
|
||||
#include "utils/array.h"
|
||||
#include "utils/lsyscache.h"
|
||||
|
@ -65,6 +66,8 @@ ExecSubPlan(SubPlanState *node,
|
|||
{
|
||||
SubPlan *subplan = node->subplan;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Set non-null as default */
|
||||
*isNull = false;
|
||||
|
||||
|
@ -618,6 +621,8 @@ findPartialMatch(TupleHashTable hashtable, TupleTableSlot *slot,
|
|||
InitTupleHashIterator(hashtable, &hashiter);
|
||||
while ((entry = ScanTupleHashTable(hashtable, &hashiter)) != NULL)
|
||||
{
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
ExecStoreMinimalTuple(entry->firstTuple, hashtable->tableslot, false);
|
||||
if (!execTuplesUnequal(slot, hashtable->tableslot,
|
||||
numCols, keyColIdx,
|
||||
|
|
|
@ -440,6 +440,8 @@ tfuncLoadRows(TableFuncScanState *tstate, ExprContext *econtext)
|
|||
ListCell *cell = list_head(tstate->coldefexprs);
|
||||
int colno;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
ExecClearTuple(tstate->ss.ss_ScanTupleSlot);
|
||||
|
||||
/*
|
||||
|
|
|
@ -26,6 +26,7 @@
|
|||
#include "catalog/pg_type.h"
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/nodeTidscan.h"
|
||||
#include "miscadmin.h"
|
||||
#include "optimizer/clauses.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "utils/array.h"
|
||||
|
@ -400,6 +401,8 @@ TidNext(TidScanState *node)
|
|||
node->tss_TidPtr--;
|
||||
else
|
||||
node->tss_TidPtr++;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
@ -35,6 +35,7 @@
|
|||
|
||||
#include "executor/executor.h"
|
||||
#include "executor/nodeUnique.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
|
||||
|
||||
|
@ -50,6 +51,8 @@ ExecUnique(UniqueState *node)
|
|||
TupleTableSlot *slot;
|
||||
PlanState *outerPlan;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* get information from the node
|
||||
*/
|
||||
|
|
|
@ -1594,6 +1594,8 @@ ExecWindowAgg(WindowAggState *winstate)
|
|||
int i;
|
||||
int numfuncs;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (winstate->all_done)
|
||||
return NULL;
|
||||
|
||||
|
@ -2371,6 +2373,9 @@ window_gettupleslot(WindowObject winobj, int64 pos, TupleTableSlot *slot)
|
|||
WindowAggState *winstate = winobj->winstate;
|
||||
MemoryContext oldcontext;
|
||||
|
||||
/* often called repeatedly in a row */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Don't allow passing -1 to spool_tuples here */
|
||||
if (pos < 0)
|
||||
return false;
|
||||
|
|
Loading…
Reference in a new issue