MCPcopy Create free account
hub / github.com/apache/cloudberry / ExecGatherMerge

Function ExecGatherMerge

src/backend/executor/nodeGatherMerge.c:187–280  ·  view source on GitHub ↗

---------------------------------------------------------------- * ExecGatherMerge(node) * * Scans the relation via multiple workers and returns * the next qualifying tuple. * ---------------------------------------------------------------- */

Source from the content-addressed store, hash-verified

185 * ----------------------------------------------------------------
186 */
187static TupleTableSlot *
188ExecGatherMerge(PlanState *pstate)
189{
190 GatherMergeState *node = castNode(GatherMergeState, pstate);
191 TupleTableSlot *slot;
192 ExprContext *econtext;
193
194 CHECK_FOR_INTERRUPTS();
195
196 /*
197 * As with Gather, we don't launch workers until this node is actually
198 * executed.
199 */
200 if (!node->initialized)
201 {
202 EState *estate = node->ps.state;
203 GatherMerge *gm = castNode(GatherMerge, node->ps.plan);
204
205 /*
206 * Sometimes we might have to run without parallelism; but if parallel
207 * mode is active then we can try to fire up some workers.
208 */
209 if (gm->num_workers > 0 && estate->es_use_parallel_mode)
210 {
211 ParallelContext *pcxt;
212
213 /* Initialize, or re-initialize, shared state needed by workers. */
214 if (!node->pei)
215 node->pei = ExecInitParallelPlan(node->ps.lefttree,
216 estate,
217 gm->initParam,
218 gm->num_workers,
219 node->tuples_needed);
220 else
221 ExecParallelReinitialize(node->ps.lefttree,
222 node->pei,
223 gm->initParam);
224
225 /* Try to launch workers. */
226 pcxt = node->pei->pcxt;
227 LaunchParallelWorkers(pcxt);
228 /* We save # workers launched for the benefit of EXPLAIN */
229 node->nworkers_launched = pcxt->nworkers_launched;
230
231 /* Set up tuple queue readers to read the results. */
232 if (pcxt->nworkers_launched > 0)
233 {
234 ExecParallelCreateReaders(node->pei);
235 /* Make a working array showing the active readers */
236 node->nreaders = pcxt->nworkers_launched;
237 node->reader = (TupleQueueReader **)
238 palloc(node->nreaders * sizeof(TupleQueueReader *));
239 memcpy(node->reader, node->pei->reader,
240 node->nreaders * sizeof(TupleQueueReader *));
241 }
242 else
243 {
244 /* No workers? Then never mind. */

Callers

nothing calls this directly

Calls 7

ExecInitParallelPlanFunction · 0.85
ExecParallelReinitializeFunction · 0.85
LaunchParallelWorkersFunction · 0.85
gather_merge_getnextFunction · 0.85
ExecProjectFunction · 0.85
pallocFunction · 0.50

Tested by

no test coverage detected