Function ParallelQueryMain

src/backend/executor/execParallel.c:1396–1501

* Main entrypoint for parallel query worker processes. * * We reach this function from ParallelWorkerMain, so the setup necessary to * create a sensible parallel environment has already been done; * ParallelWorkerMain worries about stuff like the transaction state, combo * CID mappings, and GUC values, so we don't need to deal with any of that * here. * * Our job is to deal with concerns s

 * the shm_toc.
 */
void
ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
{
	FixedParallelExecutorState *fpes;
	BufferUsage *buffer_usage;
	WalUsage   *wal_usage;
	DestReceiver *receiver;
	QueryDesc  *queryDesc;
	SharedExecutorInstrumentation *instrumentation;
	SharedJitInstrumentation *jit_instrumentation;
	int			instrument_options = 0;
	void	   *area_space;
	dsa_area   *area;
	ParallelWorkerContext pwcxt;

	/* Get fixed-size state. */
	fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);

	/* Set up DestReceiver, SharedExecutorInstrumentation, and QueryDesc. */
	receiver = ExecParallelGetReceiver(seg, toc);
	instrumentation = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, true);
	if (instrumentation != NULL)
		instrument_options = instrumentation->instrument_options;
	jit_instrumentation = shm_toc_lookup(toc, PARALLEL_KEY_JIT_INSTRUMENTATION,
										 true);
	queryDesc = ExecParallelGetQueryDesc(toc, receiver, instrument_options);

	/* Setting debug_query_string for individual workers */
	debug_query_string = queryDesc->sourceText;

	/* Report workers' query and queryId for monitoring purposes */
	pgstat_report_activity(STATE_RUNNING, debug_query_string);

	/* Attach to the dynamic shared memory area. */
	area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
	area = dsa_attach_in_place(area_space, seg);

	/* Start up the executor */
	queryDesc->plannedstmt->jitFlags = fpes->jit_flags;
	ExecutorStart(queryDesc, fpes->eflags);

	/* Special executor initialization steps for parallel workers */
	queryDesc->planstate->state->es_query_dsa = area;
	if (DsaPointerIsValid(fpes->param_exec))
	{
		char	   *paramexec_space;

		paramexec_space = dsa_get_address(area, fpes->param_exec);
		RestoreParamExecParams(paramexec_space, queryDesc->estate);

	}
	pwcxt.toc = toc;
	pwcxt.seg = seg;
	ExecParallelInitializeWorker(queryDesc->planstate, &pwcxt);

	/* Pass down any tuple bound */
	ExecSetTupleBound(fpes->tuples_needed, queryDesc->planstate);