* A crufty confusing part of the current code is how contentId is used within * the motion structures and then how that gets translated to targetRoutes by * this motion nodes. * * WARNING: There are ALOT of assumptions in here about how the motion node * instructions are encoded into motion and stuff. * * There are 3 types of sending that can happen here: * * FIXED - sending to a single
| 1178 | * |
| 1179 | */ |
| 1180 | void |
| 1181 | doSendTuple(Motion *motion, MotionState *node, TupleTableSlot *outerTupleSlot) |
| 1182 | { |
| 1183 | int16 targetRoute = 0; |
| 1184 | SendReturnCode sendRC = STOP_SENDING; |
| 1185 | ExprContext *econtext = node->ps.ps_ExprContext; |
| 1186 | int parallel_workers; |
| 1187 | int i; |
| 1188 | |
| 1189 | int parentIndex = node->ps.state->currentSliceId; |
| 1190 | ExecSlice *recvSlice = &node->ps.state->es_sliceTable->slices[parentIndex]; |
| 1191 | parallel_workers = recvSlice->parallel_workers; |
| 1192 | Assert(parallel_workers != 0); |
| 1193 | |
| 1194 | /* We got a tuple from the child-plan. */ |
| 1195 | node->numTuplesFromChild++; |
| 1196 | |
| 1197 | if (motion->motionType == MOTIONTYPE_GATHER || |
| 1198 | motion->motionType == MOTIONTYPE_GATHER_SINGLE) |
| 1199 | { |
| 1200 | /* |
| 1201 | * Actually, since we can only send to a single output segment |
| 1202 | * here, we are guaranteed that we only have a single targetRoute |
| 1203 | * setup that we could possibly send to. So we can cheat and just |
| 1204 | * fix the targetRoute to 0 (the 1st route). |
| 1205 | */ |
| 1206 | targetRoute = 0; |
| 1207 | |
| 1208 | } |
| 1209 | else if (motion->motionType == MOTIONTYPE_BROADCAST) |
| 1210 | { |
| 1211 | targetRoute = BROADCAST_SEGIDX; |
| 1212 | } |
| 1213 | else if (motion->motionType == MOTIONTYPE_BROADCAST_WORKERS) |
| 1214 | { |
| 1215 | int numSegments = recvSlice->planNumSegments; |
| 1216 | |
| 1217 | Assert(numSegments != 0); |
| 1218 | Assert(numSegments % parallel_workers == 0); |
| 1219 | |
| 1220 | for (i = 0; i < numSegments / parallel_workers; i++) |
| 1221 | { |
| 1222 | targetRoute = i * parallel_workers + random() % parallel_workers; |
| 1223 | |
| 1224 | CheckAndSendRecordCache(node->ps.state->motionlayer_context, |
| 1225 | node->ps.state->interconnect_context, |
| 1226 | motion->motionID, |
| 1227 | targetRoute); |
| 1228 | sendRC = SendTuple(node->ps.state->motionlayer_context, |
| 1229 | node->ps.state->interconnect_context, |
| 1230 | motion->motionID, |
| 1231 | outerTupleSlot, |
| 1232 | targetRoute); |
| 1233 | |
| 1234 | Assert(sendRC == SEND_COMPLETE || sendRC == STOP_SENDING); |
| 1235 | |
| 1236 | if (sendRC == STOP_SENDING) |
| 1237 | break; |
no test coverage detected