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

Function doSendTuple

src/backend/executor/nodeMotion.c:1180–1346  ·  view source on GitHub ↗

* 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

Source from the content-addressed store, hash-verified

1178 *
1179 */
1180void
1181doSendTuple(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;

Callers 1

execMotionSenderFunction · 0.85

Calls 11

randomFunction · 0.85
CheckAndSendRecordCacheFunction · 0.85
SendTupleFunction · 0.85
initStringInfoFunction · 0.85
appendStringInfoFunction · 0.85
formatTupleFunction · 0.85
list_lengthFunction · 0.85
slot_getattrFunction · 0.85
Int32GetDatumFunction · 0.85
evalHashKeyFunction · 0.70
pfreeFunction · 0.50

Tested by

no test coverage detected