Function execMotionSortedReceiver

src/backend/executor/nodeMotion.c:432–653

Sorted receiver using binary heap */

/* Sorted receiver using binary heap */
static TupleTableSlot *
execMotionSortedReceiver(MotionState *node)
{
	TupleTableSlot *slot;
	binaryheap *hp = node->tupleheap;
	MinimalTuple inputTuple;
	Motion	   *motion = (Motion *) node->ps.plan;
	EState	   *estate = node->ps.state;

	AssertState(motion->motionType == MOTIONTYPE_GATHER &&
				motion->sendSorted &&
				hp != NULL);

	/* Notify senders and return EOS if caller doesn't want any more data. */
	if (node->stopRequested)
	{

		SendStopMessage(node->ps.state->motionlayer_context,
						node->ps.state->interconnect_context,
						motion->motionID);
		return NULL;
	}

	if (estate->interconnect_context == NULL)
	{
		if (!estate->es_interconnect_is_setup && estate->dispatcherState &&
			!estate->es_got_eos)
		{
			/*
			 * We could only possibly get here in the following scenario:
			 * 1. We are QD gracefully aborting a transaction.
			 * 2. We have torn down the interconnect of the current slice.
			 * 3. Since an error has happened, we no longer need to finish fetching
			 * all the tuples, hence squelching the executor subtree.
			 * 4. We are in the process of ExecSquelchShareInputScan(), and the
			 * Shared Scan has this Motion below it.
			 *
			 * NB: if you need to change this, see also execMotionUnsortedReceiver()
			 */
			ereport(NOTICE,
					(errmsg("An ERROR must have happened. Stopping a Shared Scan.")));
			return NULL;
		}
		else
			ereport(ERROR, (errmsg("Interconnect is down unexpectedly.")));
	}

	/*
	 * On first call, fill the priority queue with each sender's first tuple.
	 */
	if (!node->tupleheapReady)
	{
		MinimalTuple inputTuple;
		binaryheap *hp = node->tupleheap;
		Motion	   *motion = (Motion *) node->ps.plan;
		int			iSegIdx;
		ListCell   *lcProcess;
		ExecSlice  *sendSlice = &node->ps.state->es_sliceTable->slices[motion->motionID];