---------------------------------------------------------------- * ExecGather(node) * * Scans the relation via multiple workers and returns * the next qualifying tuple. * ---------------------------------------------------------------- */
| 139 | * ---------------------------------------------------------------- |
| 140 | */ |
| 141 | static TupleTableSlot * |
| 142 | ExecGather(PlanState *pstate) |
| 143 | { |
| 144 | GatherState *node = castNode(GatherState, pstate); |
| 145 | TupleTableSlot *slot; |
| 146 | ExprContext *econtext; |
| 147 | |
| 148 | CHECK_FOR_INTERRUPTS(); |
| 149 | |
| 150 | /* |
| 151 | * Initialize the parallel context and workers on first execution. We do |
| 152 | * this on first execution rather than during node initialization, as it |
| 153 | * needs to allocate a large dynamic segment, so it is better to do it |
| 154 | * only if it is really needed. |
| 155 | */ |
| 156 | if (!node->initialized) |
| 157 | { |
| 158 | EState *estate = node->ps.state; |
| 159 | Gather *gather = (Gather *) node->ps.plan; |
| 160 | |
| 161 | /* |
| 162 | * Sometimes we might have to run without parallelism; but if parallel |
| 163 | * mode is active then we can try to fire up some workers. |
| 164 | */ |
| 165 | if (gather->num_workers > 0 && estate->es_use_parallel_mode) |
| 166 | { |
| 167 | ParallelContext *pcxt; |
| 168 | |
| 169 | /* Initialize, or re-initialize, shared state needed by workers. */ |
| 170 | if (!node->pei) |
| 171 | node->pei = ExecInitParallelPlan(node->ps.lefttree, |
| 172 | estate, |
| 173 | gather->initParam, |
| 174 | gather->num_workers, |
| 175 | node->tuples_needed); |
| 176 | else |
| 177 | ExecParallelReinitialize(node->ps.lefttree, |
| 178 | node->pei, |
| 179 | gather->initParam); |
| 180 | |
| 181 | /* |
| 182 | * Register backend workers. We might not get as many as we |
| 183 | * requested, or indeed any at all. |
| 184 | */ |
| 185 | pcxt = node->pei->pcxt; |
| 186 | LaunchParallelWorkers(pcxt); |
| 187 | /* We save # workers launched for the benefit of EXPLAIN */ |
| 188 | node->nworkers_launched = pcxt->nworkers_launched; |
| 189 | |
| 190 | /* Set up tuple queue readers to read the results. */ |
| 191 | if (pcxt->nworkers_launched > 0) |
| 192 | { |
| 193 | ExecParallelCreateReaders(node->pei); |
| 194 | /* Make a working array showing the active readers */ |
| 195 | node->nreaders = pcxt->nworkers_launched; |
| 196 | node->reader = (TupleQueueReader **) |
| 197 | palloc(node->nreaders * sizeof(TupleQueueReader *)); |
| 198 | memcpy(node->reader, node->pei->reader, |
nothing calls this directly
no test coverage detected