---------------------------------------------------------------- * ExecMotion * ---------------------------------------------------------------- */
| 97 | * ---------------------------------------------------------------- |
| 98 | */ |
| 99 | static TupleTableSlot * |
| 100 | ExecMotion(PlanState *pstate) |
| 101 | { |
| 102 | MotionState *node = castNode(MotionState, pstate); |
| 103 | Motion *motion = (Motion *) node->ps.plan; |
| 104 | |
| 105 | /* |
| 106 | * Check for interrupts. Without this we've seen the scenario before that |
| 107 | * it could be quite slow to cancel a query that selects all the tuples |
| 108 | * from a big distributed table because the motion node on QD has no chance |
| 109 | * of checking the cancel signal. |
| 110 | */ |
| 111 | CHECK_FOR_INTERRUPTS(); |
| 112 | |
| 113 | /* sanity check */ |
| 114 | if (node->stopRequested) |
| 115 | ereport(ERROR, |
| 116 | (errcode(ERRCODE_INTERNAL_ERROR), |
| 117 | errmsg("unexpected internal error"), |
| 118 | errmsg("Already stopped motion node is executed again, data will lost"), |
| 119 | errhint("Likely motion node is incorrectly squelched earlier"))); |
| 120 | |
| 121 | /* |
| 122 | * at the top here we basically decide: -- SENDER vs. RECEIVER and -- |
| 123 | * SORTED vs. UNSORTED |
| 124 | */ |
| 125 | if (node->mstype == MOTIONSTATE_RECV) |
| 126 | { |
| 127 | TupleTableSlot *tuple; |
| 128 | #ifdef MEASURE_MOTION_TIME |
| 129 | struct timeval startTime; |
| 130 | struct timeval stopTime; |
| 131 | |
| 132 | gettimeofday(&startTime, NULL); |
| 133 | #endif |
| 134 | |
| 135 | if (node->ps.state->active_recv_id >= 0) |
| 136 | { |
| 137 | if (node->ps.state->active_recv_id != motion->motionID) |
| 138 | { |
| 139 | /* |
| 140 | * See motion_sanity_walker() for details on how a deadlock |
| 141 | * may occur. |
| 142 | */ |
| 143 | elog(LOG, "DEADLOCK HAZARD: Updating active_motion_id from %d to %d", |
| 144 | node->ps.state->active_recv_id, motion->motionID); |
| 145 | node->ps.state->active_recv_id = motion->motionID; |
| 146 | } |
| 147 | } |
| 148 | else |
| 149 | node->ps.state->active_recv_id = motion->motionID; |
| 150 | |
| 151 | if (motion->sendSorted) |
| 152 | tuple = execMotionSortedReceiver(node); |
| 153 | else |
| 154 | tuple = execMotionUnsortedReceiver(node); |
| 155 | |
| 156 | /* |
nothing calls this directly
no test coverage detected