59 #define PARALLEL_KEY_EXECUTOR_FIXED UINT64CONST(0xE000000000000001) 60 #define PARALLEL_KEY_PLANNEDSTMT UINT64CONST(0xE000000000000002) 61 #define PARALLEL_KEY_PARAMLISTINFO UINT64CONST(0xE000000000000003) 62 #define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xE000000000000004) 63 #define PARALLEL_KEY_TUPLE_QUEUE UINT64CONST(0xE000000000000005) 64 #define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xE000000000000006) 65 #define PARALLEL_KEY_DSA UINT64CONST(0xE000000000000007) 66 #define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xE000000000000008) 67 #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009) 68 #define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xE00000000000000A) 70 #define PARALLEL_TUPLE_QUEUE_SIZE 65536 108 #define GetInstrumentationArray(sei) \ 109 (AssertVariableIsOfTypeMacro(sei, SharedExecutorInstrumentation *), \ 110 (Instrumentation *) (((char *) sei) + sei->instrument_offset)) 232 if (planstate == NULL)
246 if (planstate->plan->parallel_aware)
251 if (planstate->plan->parallel_aware)
256 if (planstate->plan->parallel_aware)
261 if (planstate->plan->parallel_aware)
266 if (planstate->plan->parallel_aware)
271 if (planstate->plan->parallel_aware)
276 if (planstate->plan->parallel_aware)
314 Size sz =
sizeof(int);
336 typLen =
sizeof(
Datum);
371 memcpy(start_address, &nparams,
sizeof(
int));
372 start_address +=
sizeof(int);
387 memcpy(start_address, ¶mid,
sizeof(
int));
388 start_address +=
sizeof(int);
396 typLen =
sizeof(
Datum);
416 memcpy(&nparams, start_address,
sizeof(
int));
417 start_address +=
sizeof(int);
419 for (i = 0; i < nparams; i++)
424 memcpy(¶mid, start_address,
sizeof(
int));
425 start_address +=
sizeof(int);
442 if (planstate == NULL)
470 if (planstate->plan->parallel_aware)
475 if (planstate->plan->parallel_aware)
480 if (planstate->plan->parallel_aware)
485 if (planstate->plan->parallel_aware)
490 if (planstate->plan->parallel_aware)
495 if (planstate->plan->parallel_aware)
500 if (planstate->plan->parallel_aware)
569 (
Size) PARALLEL_TUPLE_QUEUE_SIZE);
599 char *paramlistinfo_space;
605 int paramlistinfo_len;
606 int instrumentation_len = 0;
607 int jit_instrumentation_len = 0;
608 int instrument_offset = 0;
655 pstmt_len = strlen(pstmt_data) + 1;
698 instrumentation_len =
701 instrumentation_len =
MAXALIGN(instrumentation_len);
702 instrument_offset = instrumentation_len;
703 instrumentation_len +=
712 jit_instrumentation_len =
750 memcpy(pstmt_space, pstmt_data, pstmt_len);
792 for (i = 0; i < nworkers * e.
nnodes; ++
i)
801 jit_instrumentation_len);
803 memset(jit_instrumentation->
jit_instr, 0,
806 jit_instrumentation);
816 if (pcxt->
seg != NULL)
859 elog(
ERROR,
"inconsistent count of PlanState nodes");
884 for (i = 0; i < nworkers; i++)
950 if (planstate == NULL)
964 if (planstate->plan->parallel_aware)
969 if (planstate->plan->parallel_aware)
974 if (planstate->plan->parallel_aware)
979 if (planstate->plan->parallel_aware)
983 if (planstate->plan->parallel_aware)
988 if (planstate->plan->parallel_aware)
993 if (planstate->plan->parallel_aware)
1031 elog(
ERROR,
"plan node %d not found", plan_node_id);
1036 for (n = 0; n < instrumentation->
num_workers; ++n)
1140 for (i = 0; i < nworkers; i++)
1152 for (i = 0; i < nworkers; i++)
1165 for (i = 0; i < nworkers; i++)
1196 if (pei->
area != NULL)
1201 if (pei->
pcxt != NULL)
1231 int instrument_options)
1254 receiver, paramLI, NULL, instrument_options);
1281 elog(
ERROR,
"plan node %d not found", plan_node_id);
1290 Assert(ParallelWorkerNumber < instrumentation->num_workers);
1305 if (planstate == NULL)
1315 if (planstate->plan->parallel_aware)
1320 if (planstate->plan->parallel_aware)
1325 if (planstate->plan->parallel_aware)
1330 if (planstate->plan->parallel_aware)
1334 if (planstate->plan->parallel_aware)
1339 if (planstate->plan->parallel_aware)
1344 if (planstate->plan->parallel_aware)
1404 int instrument_options = 0;
1415 if (instrumentation != NULL)
1439 char *paramexec_space;
1477 &wal_usage[ParallelWorkerNumber]);
1480 if (instrumentation != NULL)
1485 if (queryDesc->
estate->
es_jit && jit_instrumentation != NULL)
1487 Assert(ParallelWorkerNumber < jit_instrumentation->num_workers);
void ExecForeignScanInitializeWorker(ForeignScanState *node, ParallelWorkerContext *pwcxt)
void ExecResultCacheRetrieveInstrumentation(ResultCacheState *node)
void ExecIndexScanInitializeDSM(IndexScanState *node, ParallelContext *pcxt)
struct dsa_area * es_query_dsa
void ExecSetParamPlanMulti(const Bitmapset *params, ExprContext *econtext)
void DestroyTupleQueueReader(TupleQueueReader *reader)
void ExecAppendInitializeWorker(AppendState *node, ParallelWorkerContext *pwcxt)
WorkerInstrumentation * worker_instrument
struct ExecParallelEstimateContext ExecParallelEstimateContext
int plan_node_id[FLEXIBLE_ARRAY_MEMBER]
void ExecCustomScanInitializeWorker(CustomScanState *node, ParallelWorkerContext *pwcxt)
Instrumentation * instrument
#define InvalidDsaPointer
ParallelContext * CreateParallelContext(const char *library_name, const char *function_name, int nworkers)
struct JitContext * es_jit
void ExecSortEstimate(SortState *node, ParallelContext *pcxt)
void ExecParallelFinish(ParallelExecutorInfo *pei)
#define PARALLEL_KEY_TUPLE_QUEUE
void ExecCustomScanReInitializeDSM(CustomScanState *node, ParallelContext *pcxt)
void FreeQueryDesc(QueryDesc *qdesc)
void shm_mq_detach(shm_mq_handle *mqh)
void InstrAggNode(Instrumentation *dst, Instrumentation *add)
struct SharedJitInstrumentation * jit_instrumentation
static bool ExecParallelReportInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation)
void ExecAppendReInitializeDSM(AppendState *node, ParallelContext *pcxt)
shm_toc_estimator estimator
void ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *cxt)
struct JitInstrumentation * es_jit_worker_instr
void ExecSortInitializeWorker(SortState *node, ParallelWorkerContext *pwcxt)
int bms_next_member(const Bitmapset *a, int prevbit)
void ExecutorStart(QueryDesc *queryDesc, int eflags)
dsa_area * dsa_attach_in_place(void *place, dsm_segment *segment)
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Snapshot GetActiveSnapshot(void)
PlannedStmt * es_plannedstmt
#define FLEXIBLE_ARRAY_MEMBER
void * stringToNode(const char *str)
size_t dsa_minimum_size(void)
static char * ExecSerializePlan(Plan *plan, EState *estate)
void ExecIndexScanReInitializeDSM(IndexScanState *node, ParallelContext *pcxt)
static shm_mq_handle ** ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize)
void ExecIndexScanInitializeWorker(IndexScanState *node, ParallelWorkerContext *pwcxt)
void ExecHashRetrieveInstrumentation(HashState *node)
#define shm_toc_estimate_chunk(e, sz)
void InstrEndParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
Size EstimateParamListSpace(ParamListInfo paramLI)
#define OidIsValid(objectId)
JitInstrumentation jit_instr[FLEXIBLE_ARRAY_MEMBER]
void ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt)
static bool ExecParallelReInitializeDSM(PlanState *planstate, ParallelContext *pcxt)
static dsa_pointer SerializeParamExecParams(EState *estate, Bitmapset *params, dsa_area *area)
void ExecSetTupleBound(int64 tuples_needed, PlanState *child_node)
void ExecForeignScanReInitializeDSM(ForeignScanState *node, ParallelContext *pcxt)
static Oid list_nth_oid(const List *list, int n)
void InstrEndLoop(Instrumentation *instr)
SharedExecutorInstrumentation * instrumentation
void ExecutorEnd(QueryDesc *queryDesc)
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
#define GetPerTupleExprContext(estate)
void WaitForParallelWorkersToFinish(ParallelContext *pcxt)
#define PARALLEL_KEY_INSTRUMENTATION
void DestroyParallelContext(ParallelContext *pcxt)
const char * es_sourceText
void ExecSortInitializeDSM(SortState *node, ParallelContext *pcxt)
ParallelWorkerInfo * worker
ParamExecData * es_param_exec_vals
void pfree(void *pointer)
MemoryContext es_query_cxt
void ExecSeqScanInitializeDSM(SeqScanState *node, ParallelContext *pcxt)
static bool ExecParallelInitializeDSM(PlanState *node, ExecParallelInitializeDSMContext *d)
void ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
void ExecBitmapHeapInitializeDSM(BitmapHeapScanState *node, ParallelContext *pcxt)
#define PARALLEL_KEY_WAL_USAGE
BufferUsage * buffer_usage
void ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count, bool execute_once)
static bool ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
void ExecIndexOnlyScanEstimate(IndexOnlyScanState *node, ParallelContext *pcxt)
QueryDesc * CreateQueryDesc(PlannedStmt *plannedstmt, const char *sourceText, Snapshot snapshot, Snapshot crosscheck_snapshot, DestReceiver *dest, ParamListInfo params, QueryEnvironment *queryEnv, int instrument_options)
#define PARALLEL_KEY_PARAMLISTINFO
shm_mq * shm_mq_create(void *address, Size size)
#define lfirst_node(type, lc)
int bms_num_members(const Bitmapset *a)
void ExecParallelCreateReaders(ParallelExecutorInfo *pei)
#define PARALLEL_KEY_PLANNEDSTMT
ParamListInfo RestoreParamList(char **start_address)
static bool ExecParallelRetrieveInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation)
struct TupleQueueReader ** reader
void ExecAggEstimate(AggState *node, ParallelContext *pcxt)
struct ExecParallelInitializeDSMContext ExecParallelInitializeDSMContext
void dsa_detach(dsa_area *area)
BackgroundWorkerHandle * bgwhandle
#define PARALLEL_TUPLE_QUEUE_SIZE
static Size EstimateParamExecSpace(EState *estate, Bitmapset *params)
void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt)
#define PARALLEL_KEY_EXECUTOR_FIXED
void ExecResultCacheInitializeDSM(ResultCacheState *node, ParallelContext *pcxt)
static bool ExecParallelEstimate(PlanState *node, ExecParallelEstimateContext *e)
dsa_area * dsa_create_in_place(void *place, size_t size, int tranche_id, dsm_segment *segment)
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
#define PARALLEL_KEY_BUFFER_USAGE
void InstrAccumParallelQuery(BufferUsage *bufusage, WalUsage *walusage)
#define IsParallelWorker()
struct SharedJitInstrumentation * worker_jit_instrument
void ExecAggRetrieveInstrumentation(AggState *node)
void ExecutorFinish(QueryDesc *queryDesc)
void ExecBitmapHeapReInitializeDSM(BitmapHeapScanState *node, ParallelContext *pcxt)
List * lappend(List *list, void *datum)
#define PARALLEL_KEY_JIT_INSTRUMENTATION
Size datumEstimateSpace(Datum value, bool isnull, bool typByVal, int typLen)
bool bms_is_empty(const Bitmapset *a)
const char * debug_query_string
void InitializeParallelDSM(ParallelContext *pcxt)
void ExecHashJoinEstimate(HashJoinState *state, ParallelContext *pcxt)
TupleQueueReader * CreateTupleQueueReader(shm_mq_handle *handle)
void ExecIncrementalSortInitializeWorker(IncrementalSortState *node, ParallelWorkerContext *pwcxt)
void InstrStartParallelQuery(void)
void InstrInit(Instrumentation *instr, int instrument_options)
void ExecIncrementalSortInitializeDSM(IncrementalSortState *node, ParallelContext *pcxt)
static void ExecParallelRetrieveJitInstrumentation(PlanState *planstate, SharedJitInstrumentation *shared_jit)
Size mul_size(Size s1, Size s2)
Instrumentation instrument[FLEXIBLE_ARRAY_MEMBER]
void * palloc0(Size size)
Datum datumRestore(char **start_address, bool *isnull)
void ExecHashEstimate(HashState *node, ParallelContext *pcxt)
Size add_size(Size s1, Size s2)
uint64 pgstat_get_my_query_id(void)
void InstrJitAgg(JitInstrumentation *dst, JitInstrumentation *add)
void ExecForeignScanInitializeDSM(ForeignScanState *node, ParallelContext *pcxt)
void ReinitializeParallelDSM(ParallelContext *pcxt)
void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt)
void shm_mq_set_handle(shm_mq_handle *mqh, BackgroundWorkerHandle *handle)
void * MemoryContextAllocZero(MemoryContext context, Size size)
void ExecCustomScanInitializeDSM(CustomScanState *node, ParallelContext *pcxt)
void ExecAppendInitializeDSM(AppendState *node, ParallelContext *pcxt)
void(* rDestroy)(DestReceiver *self)
void ExecAppendEstimate(AppendState *node, ParallelContext *pcxt)
void ExecIncrementalSortRetrieveInstrumentation(IncrementalSortState *node)
void ExecParallelCleanup(ParallelExecutorInfo *pei)
void SerializeParamList(ParamListInfo paramLI, char **start_address)
#define Assert(condition)
Bitmapset * rewindPlanIDs
struct FixedParallelExecutorState FixedParallelExecutorState
void ExecForeignScanEstimate(ForeignScanState *node, ParallelContext *pcxt)
void ExecResultCacheInitializeWorker(ResultCacheState *node, ParallelWorkerContext *pwcxt)
void ExecSeqScanEstimate(SeqScanState *node, ParallelContext *pcxt)
#define shm_toc_estimate_keys(e, cnt)
void get_typlenbyval(Oid typid, int16 *typlen, bool *typbyval)
void ExecParallelReinitialize(PlanState *planstate, ParallelExecutorInfo *pei, Bitmapset *sendParams)
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
void ExecIndexOnlyScanInitializeWorker(IndexOnlyScanState *node, ParallelWorkerContext *pwcxt)
void ExecIndexOnlyScanReInitializeDSM(IndexOnlyScanState *node, ParallelContext *pcxt)
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
void ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt)
#define DsaPointerIsValid(x)
void ExecSeqScanReInitializeDSM(SeqScanState *node, ParallelContext *pcxt)
void ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
void ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt)
void dsa_free(dsa_area *area, dsa_pointer dp)
struct JitInstrumentation JitInstrumentation
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
void pgstat_report_activity(BackendState state, const char *cmd_str)
void * MemoryContextAlloc(MemoryContext context, Size size)
void ExecBitmapHeapEstimate(BitmapHeapScanState *node, ParallelContext *pcxt)
#define PARALLEL_KEY_QUERY_TEXT
void datumSerialize(Datum value, bool isnull, bool typByVal, int typLen, char **start_address)
char * nodeToString(const void *obj)
void ExecCustomScanEstimate(CustomScanState *node, ParallelContext *pcxt)
ParamListInfo es_param_list_info
void ExecIncrementalSortEstimate(IncrementalSortState *node, ParallelContext *pcxt)
void ExecIndexOnlyScanInitializeDSM(IndexOnlyScanState *node, ParallelContext *pcxt)
ParallelExecutorInfo * ExecInitParallelPlan(PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed)
static DestReceiver * ExecParallelGetReceiver(dsm_segment *seg, shm_toc *toc)
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
static void RestoreParamExecParams(char *start_address, EState *estate)
void ExecSortRetrieveInstrumentation(SortState *node)
DestReceiver * CreateTupleQueueDestReceiver(shm_mq_handle *handle)
PlannedStmt * plannedstmt
void ExecBitmapHeapInitializeWorker(BitmapHeapScanState *node, ParallelWorkerContext *pwcxt)
void ExecIndexScanEstimate(IndexScanState *node, ParallelContext *pcxt)
SharedExecutorInstrumentation * instrumentation
void ExecSeqScanInitializeWorker(SeqScanState *node, ParallelWorkerContext *pwcxt)
bool planstate_tree_walker(PlanState *planstate, bool(*walker)(), void *context)
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
static QueryDesc * ExecParallelGetQueryDesc(shm_toc *toc, DestReceiver *receiver, int instrument_options)
#define offsetof(type, field)
void ExecResultCacheEstimate(ResultCacheState *node, ParallelContext *pcxt)
#define GetInstrumentationArray(sei)
#define dsa_allocate(area, size)