57#define PARALLEL_KEY_EXECUTOR_FIXED UINT64CONST(0xE000000000000001)
58#define PARALLEL_KEY_PLANNEDSTMT UINT64CONST(0xE000000000000002)
59#define PARALLEL_KEY_PARAMLISTINFO UINT64CONST(0xE000000000000003)
60#define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xE000000000000004)
61#define PARALLEL_KEY_TUPLE_QUEUE UINT64CONST(0xE000000000000005)
62#define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xE000000000000006)
63#define PARALLEL_KEY_DSA UINT64CONST(0xE000000000000007)
64#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xE000000000000008)
65#define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
66#define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xE00000000000000A)
68#define PARALLEL_TUPLE_QUEUE_SIZE 65536
106#define GetInstrumentationArray(sei) \
107 (AssertVariableIsOfTypeMacro(sei, SharedExecutorInstrumentation *), \
108 (Instrumentation *) (((char *) sei) + sei->instrument_offset))
162 foreach(lc,
plan->targetlist)
166 tle->resjunk =
false;
231 if (planstate == NULL)
244 case T_IndexScanState:
245 if (planstate->plan->parallel_aware)
249 case T_IndexOnlyScanState:
250 if (planstate->plan->parallel_aware)
254 case T_ForeignScanState:
255 if (planstate->plan->parallel_aware)
260 if (planstate->plan->parallel_aware)
264 case T_CustomScanState:
265 if (planstate->plan->parallel_aware)
269 case T_BitmapHeapScanState:
270 if (planstate->plan->parallel_aware)
274 case T_HashJoinState:
275 if (planstate->plan->parallel_aware)
287 case T_IncrementalSortState:
313 Size sz =
sizeof(int);
335 typLen =
sizeof(
Datum);
370 memcpy(start_address, &nparams,
sizeof(
int));
371 start_address +=
sizeof(int);
386 memcpy(start_address, ¶mid,
sizeof(
int));
387 start_address +=
sizeof(int);
395 typLen =
sizeof(
Datum);
415 memcpy(&nparams, start_address,
sizeof(
int));
416 start_address +=
sizeof(int);
418 for (
i = 0;
i < nparams;
i++)
423 memcpy(¶mid, start_address,
sizeof(
int));
424 start_address +=
sizeof(int);
441 if (planstate == NULL)
468 case T_IndexScanState:
469 if (planstate->plan->parallel_aware)
473 case T_IndexOnlyScanState:
474 if (planstate->plan->parallel_aware)
478 case T_ForeignScanState:
479 if (planstate->plan->parallel_aware)
484 if (planstate->plan->parallel_aware)
488 case T_CustomScanState:
489 if (planstate->plan->parallel_aware)
493 case T_BitmapHeapScanState:
494 if (planstate->plan->parallel_aware)
498 case T_HashJoinState:
499 if (planstate->plan->parallel_aware)
511 case T_IncrementalSortState:
598 char *paramlistinfo_space;
604 int paramlistinfo_len;
605 int instrumentation_len = 0;
606 int jit_instrumentation_len = 0;
607 int instrument_offset = 0;
654 pstmt_len = strlen(pstmt_data) + 1;
697 instrumentation_len =
699 sizeof(int) *
e.nnodes;
700 instrumentation_len =
MAXALIGN(instrumentation_len);
701 instrument_offset = instrumentation_len;
702 instrumentation_len +=
711 jit_instrumentation_len =
756 memcpy(pstmt_space, pstmt_data, pstmt_len);
798 for (
i = 0;
i < nworkers *
e.nnodes; ++
i)
807 jit_instrumentation_len);
809 memset(jit_instrumentation->
jit_instr, 0,
812 jit_instrumentation);
822 if (pcxt->
seg != NULL)
865 elog(
ERROR,
"inconsistent count of PlanState nodes");
890 for (
i = 0;
i < nworkers;
i++)
956 if (planstate == NULL)
969 case T_IndexScanState:
970 if (planstate->plan->parallel_aware)
974 case T_IndexOnlyScanState:
975 if (planstate->plan->parallel_aware)
979 case T_ForeignScanState:
980 if (planstate->plan->parallel_aware)
985 if (planstate->plan->parallel_aware)
988 case T_CustomScanState:
989 if (planstate->plan->parallel_aware)
993 case T_BitmapHeapScanState:
994 if (planstate->plan->parallel_aware)
998 case T_HashJoinState:
999 if (planstate->plan->parallel_aware)
1005 case T_IncrementalSortState:
1006 case T_MemoizeState:
1037 elog(
ERROR,
"plan node %d not found", plan_node_id);
1042 for (n = 0; n < instrumentation->
num_workers; ++n)
1067 case T_IncrementalSortState:
1076 case T_MemoizeState:
1079 case T_BitmapHeapScanState:
1149 for (
i = 0;
i < nworkers;
i++)
1161 for (
i = 0;
i < nworkers;
i++)
1174 for (
i = 0;
i < nworkers;
i++)
1205 if (pei->
area != NULL)
1210 if (pei->
pcxt != NULL)
1240 int instrument_options)
1263 receiver, paramLI, NULL, instrument_options);
1290 elog(
ERROR,
"plan node %d not found", plan_node_id);
1299 Assert(ParallelWorkerNumber < instrumentation->num_workers);
1314 if (planstate == NULL)
1319 case T_SeqScanState:
1323 case T_IndexScanState:
1324 if (planstate->plan->parallel_aware)
1328 case T_IndexOnlyScanState:
1329 if (planstate->plan->parallel_aware)
1333 case T_ForeignScanState:
1334 if (planstate->plan->parallel_aware)
1339 if (planstate->plan->parallel_aware)
1342 case T_CustomScanState:
1343 if (planstate->plan->parallel_aware)
1347 case T_BitmapHeapScanState:
1348 if (planstate->plan->parallel_aware)
1352 case T_HashJoinState:
1353 if (planstate->plan->parallel_aware)
1365 case T_IncrementalSortState:
1374 case T_MemoizeState:
1412 int instrument_options = 0;
1423 if (instrumentation != NULL)
1447 char *paramexec_space;
1486 if (instrumentation != NULL)
1491 if (queryDesc->
estate->
es_jit && jit_instrumentation != NULL)
1493 Assert(ParallelWorkerNumber < jit_instrumentation->num_workers);
void InitializeParallelDSM(ParallelContext *pcxt)
void WaitForParallelWorkersToFinish(ParallelContext *pcxt)
void ReinitializeParallelDSM(ParallelContext *pcxt)
void DestroyParallelContext(ParallelContext *pcxt)
ParallelContext * CreateParallelContext(const char *library_name, const char *function_name, int nworkers)
uint64 pgstat_get_my_query_id(void)
void pgstat_report_activity(BackendState state, const char *cmd_str)
int bms_next_member(const Bitmapset *a, int prevbit)
int bms_num_members(const Bitmapset *a)
#define Assert(condition)
#define FLEXIBLE_ARRAY_MEMBER
#define OidIsValid(objectId)
Datum datumRestore(char **start_address, bool *isnull)
void datumSerialize(Datum value, bool isnull, bool typByVal, int typLen, char **start_address)
Size datumEstimateSpace(Datum value, bool isnull, bool typByVal, int typLen)
dsa_area * dsa_attach_in_place(void *place, dsm_segment *segment)
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
void dsa_detach(dsa_area *area)
void dsa_free(dsa_area *area, dsa_pointer dp)
size_t dsa_minimum_size(void)
#define dsa_create_in_place(place, size, tranch_id, segment)
#define dsa_allocate(area, size)
#define InvalidDsaPointer
#define DsaPointerIsValid(x)
void ExecutorEnd(QueryDesc *queryDesc)
void ExecutorFinish(QueryDesc *queryDesc)
void ExecutorStart(QueryDesc *queryDesc, int eflags)
void ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count)
#define PARALLEL_KEY_BUFFER_USAGE
static bool ExecParallelReInitializeDSM(PlanState *planstate, ParallelContext *pcxt)
#define PARALLEL_KEY_JIT_INSTRUMENTATION
struct ExecParallelEstimateContext ExecParallelEstimateContext
#define PARALLEL_KEY_PARAMLISTINFO
#define PARALLEL_TUPLE_QUEUE_SIZE
static QueryDesc * ExecParallelGetQueryDesc(shm_toc *toc, DestReceiver *receiver, int instrument_options)
static bool ExecParallelRetrieveInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation)
static dsa_pointer SerializeParamExecParams(EState *estate, Bitmapset *params, dsa_area *area)
void ExecParallelCleanup(ParallelExecutorInfo *pei)
struct ExecParallelInitializeDSMContext ExecParallelInitializeDSMContext
#define PARALLEL_KEY_INSTRUMENTATION
static DestReceiver * ExecParallelGetReceiver(dsm_segment *seg, shm_toc *toc)
void ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
static shm_mq_handle ** ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize)
#define PARALLEL_KEY_PLANNEDSTMT
static bool ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e)
#define GetInstrumentationArray(sei)
void ExecParallelReinitialize(PlanState *planstate, ParallelExecutorInfo *pei, Bitmapset *sendParams)
static bool ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
void ExecParallelCreateReaders(ParallelExecutorInfo *pei)
#define PARALLEL_KEY_TUPLE_QUEUE
#define PARALLEL_KEY_EXECUTOR_FIXED
static char * ExecSerializePlan(Plan *plan, EState *estate)
ParallelExecutorInfo * ExecInitParallelPlan(PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed)
struct FixedParallelExecutorState FixedParallelExecutorState
#define PARALLEL_KEY_QUERY_TEXT
static Size EstimateParamExecSpace(EState *estate, Bitmapset *params)
void ExecParallelFinish(ParallelExecutorInfo *pei)
static bool ExecParallelReportInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation)
#define PARALLEL_KEY_WAL_USAGE
static void ExecParallelRetrieveJitInstrumentation(PlanState *planstate, SharedJitInstrumentation *shared_jit)
static bool ExecParallelInitializeDSM(PlanState *planstate, ExecParallelInitializeDSMContext *d)
static void RestoreParamExecParams(char *start_address, EState *estate)
void ExecSetTupleBound(int64 tuples_needed, PlanState *child_node)
#define GetPerTupleExprContext(estate)
#define IsParallelWorker()
void InstrAccumParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
void InstrEndLoop(Instrumentation *instr)
void InstrAggNode(Instrumentation *dst, Instrumentation *add)
void InstrEndParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
void InstrStartParallelQuery(void)
void InstrInit(Instrumentation *instr, int instrument_options)
void InstrJitAgg(JitInstrumentation *dst, JitInstrumentation *add)
struct JitInstrumentation JitInstrumentation
List * lappend(List *list, void *datum)
void get_typlenbyval(Oid typid, int16 *typlen, bool *typbyval)
@ LWTRANCHE_PARALLEL_QUERY_DSA
void * MemoryContextAlloc(MemoryContext context, Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
void pfree(void *pointer)
void * palloc0(Size size)
void ExecAggEstimate(AggState *node, ParallelContext *pcxt)
void ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt)
void ExecAggRetrieveInstrumentation(AggState *node)
void ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt)
void ExecAppendReInitializeDSM(AppendState *node, ParallelContext *pcxt)
void ExecAppendInitializeWorker(AppendState *node, ParallelWorkerContext *pwcxt)
void ExecAppendInitializeDSM(AppendState *node, ParallelContext *pcxt)
void ExecAppendEstimate(AppendState *node, ParallelContext *pcxt)
void ExecBitmapHeapInitializeWorker(BitmapHeapScanState *node, ParallelWorkerContext *pwcxt)
void ExecBitmapHeapEstimate(BitmapHeapScanState *node, ParallelContext *pcxt)
void ExecBitmapHeapRetrieveInstrumentation(BitmapHeapScanState *node)
void ExecBitmapHeapInitializeDSM(BitmapHeapScanState *node, ParallelContext *pcxt)
void ExecBitmapHeapReInitializeDSM(BitmapHeapScanState *node, ParallelContext *pcxt)
void ExecCustomScanInitializeDSM(CustomScanState *node, ParallelContext *pcxt)
void ExecCustomScanEstimate(CustomScanState *node, ParallelContext *pcxt)
void ExecCustomScanReInitializeDSM(CustomScanState *node, ParallelContext *pcxt)
void ExecCustomScanInitializeWorker(CustomScanState *node, ParallelWorkerContext *pwcxt)
void ExecForeignScanInitializeDSM(ForeignScanState *node, ParallelContext *pcxt)
void ExecForeignScanReInitializeDSM(ForeignScanState *node, ParallelContext *pcxt)
void ExecForeignScanEstimate(ForeignScanState *node, ParallelContext *pcxt)
void ExecForeignScanInitializeWorker(ForeignScanState *node, ParallelWorkerContext *pwcxt)
#define planstate_tree_walker(ps, w, c)
void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt)
void ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt)
void ExecHashEstimate(HashState *node, ParallelContext *pcxt)
void ExecHashRetrieveInstrumentation(HashState *node)
void ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
void ExecHashJoinEstimate(HashJoinState *state, ParallelContext *pcxt)
void ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt)
void ExecIncrementalSortEstimate(IncrementalSortState *node, ParallelContext *pcxt)
void ExecIncrementalSortInitializeDSM(IncrementalSortState *node, ParallelContext *pcxt)
void ExecIncrementalSortRetrieveInstrumentation(IncrementalSortState *node)
void ExecIncrementalSortInitializeWorker(IncrementalSortState *node, ParallelWorkerContext *pwcxt)
void ExecIndexOnlyScanEstimate(IndexOnlyScanState *node, ParallelContext *pcxt)
void ExecIndexOnlyScanInitializeWorker(IndexOnlyScanState *node, ParallelWorkerContext *pwcxt)
void ExecIndexOnlyScanReInitializeDSM(IndexOnlyScanState *node, ParallelContext *pcxt)
void ExecIndexOnlyScanInitializeDSM(IndexOnlyScanState *node, ParallelContext *pcxt)
void ExecIndexScanEstimate(IndexScanState *node, ParallelContext *pcxt)
void ExecIndexScanReInitializeDSM(IndexScanState *node, ParallelContext *pcxt)
void ExecIndexScanInitializeDSM(IndexScanState *node, ParallelContext *pcxt)
void ExecIndexScanInitializeWorker(IndexScanState *node, ParallelWorkerContext *pwcxt)
void ExecMemoizeInitializeDSM(MemoizeState *node, ParallelContext *pcxt)
void ExecMemoizeEstimate(MemoizeState *node, ParallelContext *pcxt)
void ExecMemoizeRetrieveInstrumentation(MemoizeState *node)
void ExecMemoizeInitializeWorker(MemoizeState *node, ParallelWorkerContext *pwcxt)
void ExecSeqScanReInitializeDSM(SeqScanState *node, ParallelContext *pcxt)
void ExecSeqScanInitializeWorker(SeqScanState *node, ParallelWorkerContext *pwcxt)
void ExecSeqScanInitializeDSM(SeqScanState *node, ParallelContext *pcxt)
void ExecSeqScanEstimate(SeqScanState *node, ParallelContext *pcxt)
void ExecSortInitializeWorker(SortState *node, ParallelWorkerContext *pwcxt)
void ExecSortEstimate(SortState *node, ParallelContext *pcxt)
void ExecSortInitializeDSM(SortState *node, ParallelContext *pcxt)
void ExecSortRetrieveInstrumentation(SortState *node)
void ExecSetParamPlanMulti(const Bitmapset *params, ExprContext *econtext)
char * nodeToString(const void *obj)
Size EstimateParamListSpace(ParamListInfo paramLI)
void SerializeParamList(ParamListInfo paramLI, char **start_address)
ParamListInfo RestoreParamList(char **start_address)
#define lfirst_node(type, lc)
static Oid list_nth_oid(const List *list, int n)
const char * debug_query_string
void FreeQueryDesc(QueryDesc *qdesc)
QueryDesc * CreateQueryDesc(PlannedStmt *plannedstmt, const char *sourceText, Snapshot snapshot, Snapshot crosscheck_snapshot, DestReceiver *dest, ParamListInfo params, QueryEnvironment *queryEnv, int instrument_options)
MemoryContextSwitchTo(old_ctx)
void * stringToNode(const char *str)
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
shm_mq * shm_mq_create(void *address, Size size)
void shm_mq_set_handle(shm_mq_handle *mqh, BackgroundWorkerHandle *handle)
void shm_mq_detach(shm_mq_handle *mqh)
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
#define shm_toc_estimate_chunk(e, sz)
#define shm_toc_estimate_keys(e, cnt)
Size add_size(Size s1, Size s2)
Size mul_size(Size s1, Size s2)
static pg_noinline void Size size
Snapshot GetActiveSnapshot(void)
struct dsa_area * es_query_dsa
struct JitContext * es_jit
PlannedStmt * es_plannedstmt
struct JitInstrumentation * es_jit_worker_instr
ParamExecData * es_param_exec_vals
ParamListInfo es_param_list_info
MemoryContext es_query_cxt
const char * es_sourceText
SharedExecutorInstrumentation * instrumentation
shm_toc_estimator estimator
ParallelWorkerInfo * worker
struct SharedJitInstrumentation * jit_instrumentation
BufferUsage * buffer_usage
SharedExecutorInstrumentation * instrumentation
struct TupleQueueReader ** reader
BackgroundWorkerHandle * bgwhandle
struct SharedJitInstrumentation * worker_jit_instrument
Instrumentation * instrument
WorkerInstrumentation * worker_instrument
Bitmapset * rewindPlanIDs
PlannedStmt * plannedstmt
int plan_node_id[FLEXIBLE_ARRAY_MEMBER]
JitInstrumentation jit_instr[FLEXIBLE_ARRAY_MEMBER]
Instrumentation instrument[FLEXIBLE_ARRAY_MEMBER]
void(* rDestroy)(DestReceiver *self)
DestReceiver * CreateTupleQueueDestReceiver(shm_mq_handle *handle)
TupleQueueReader * CreateTupleQueueReader(shm_mq_handle *handle)
void DestroyTupleQueueReader(TupleQueueReader *reader)