From 2ee78b44a088b8f9e7b4fa0f1d05a7c89e9f169e Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Wed, 7 Oct 2015 12:43:22 -0400
Subject: [PATCH 13/14] Modify tqueue infrastructure to support transient
 record types.

Commit 4a4e6893aa080b9094dadbe0e65f8a75fee41ac6, which introduced this
mechanism, failed to account for the fact that the RECORD pseudo-type
uses transient typmods that are only meaningful within a single
backend.  Transferring such tuples without modification between two
cooperating backends does not work.  This commit installs a system
for passing the tuple descriptors over the same shm_mq being used to
send the tuples themselves.  The two sides might not assign the same
transient typmod to any given tuple descriptor, so we must also
substitute the appropriate receiver-side typmod for the one used by
the sender.  That adds some CPU overhead, but still seems better than
being unable to pass records between cooperating parallel processes.
---
 src/backend/executor/nodeGather.c |   1 +
 src/backend/executor/tqueue.c     | 492 +++++++++++++++++++++++++++++++++++---
 src/include/executor/tqueue.h     |   4 +-
 3 files changed, 467 insertions(+), 30 deletions(-)

diff --git a/src/backend/executor/nodeGather.c b/src/backend/executor/nodeGather.c
index 7e2272f..bf62eee 100644
--- a/src/backend/executor/nodeGather.c
+++ b/src/backend/executor/nodeGather.c
@@ -207,6 +207,7 @@ gather_getnext(GatherState *gatherstate)
 
 			/* wait only if local scan is done */
 			tup = TupleQueueFunnelNext(gatherstate->funnel,
+									   slot->tts_tupleDescriptor,
 									   gatherstate->need_to_scan_locally,
 									   &done);
 			if (done)
diff --git a/src/backend/executor/tqueue.c b/src/backend/executor/tqueue.c
index 67143d3..53b69e0 100644
--- a/src/backend/executor/tqueue.c
+++ b/src/backend/executor/tqueue.c
@@ -21,23 +21,55 @@
 #include "postgres.h"
 
 #include "access/htup_details.h"
+#include "catalog/pg_type.h"
 #include "executor/tqueue.h"
+#include "funcapi.h"
+#include "lib/stringinfo.h"
 #include "miscadmin.h"
+#include "utils/array.h"
+#include "utils/memutils.h"
+#include "utils/typcache.h"
 
 typedef struct
 {
 	DestReceiver pub;
 	shm_mq_handle *handle;
+	MemoryContext	tmpcontext;
+	HTAB	   *recordhtab;
+	char		mode;
 }	TQueueDestReceiver;
 
+typedef struct RecordTypemodMap
+{
+	int			remotetypmod;
+	int			localtypmod;
+} RecordTypemodMap;
+
 struct TupleQueueFunnel
 {
 	int			nqueues;
 	int			maxqueues;
 	int			nextqueue;
 	shm_mq_handle **queue;
+	char	   *mode;
+	HTAB	   *typmodmap;
 };
 
+#define		TUPLE_QUEUE_MODE_CONTROL			'c'
+#define		TUPLE_QUEUE_MODE_DATA				'd'
+
+static void tqueueWalkRecord(TQueueDestReceiver *tqueue, Datum value);
+static void tqueueWalkRecordArray(TQueueDestReceiver *tqueue, Datum value);
+static void TupleQueueHandleControlMessage(TupleQueueFunnel *funnel,
+			Size nbytes, char *data);
+static HeapTuple TupleQueueHandleDataMessage(TupleQueueFunnel *funnel,
+							TupleDesc tupledesc, Size nbytes,
+							HeapTupleHeader data);
+static HeapTuple TupleQueueRemapTuple(TupleQueueFunnel *funnel,
+					 TupleDesc tupledesc, HeapTuple tuple);
+static Datum TupleQueueRemapRecord(TupleQueueFunnel *funnel, Datum value);
+static Datum TupleQueueRemapRecordArray(TupleQueueFunnel *funnel, Datum value);
+
 /*
  * Receive a tuple.
  */
@@ -46,12 +78,178 @@ tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
 {
 	TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
 	HeapTuple	tuple;
+	HeapTupleHeader tup;
+	AttrNumber	i;
 
 	tuple = ExecMaterializeSlot(slot);
+	tup = tuple->t_data;
+
+	/*
+	 * If any of the columns that we're sending back are records, special
+	 * handling is required, because the tuple descriptors are stored in a
+	 * backend-local cache, and the backend receiving data from us need not
+	 * have the same cache contents we do.  We grovel through the tuple,
+	 * find all the transient record types contained therein, and send
+	 * special control messages through the queue so that the receiving
+	 * process can interpret them correctly.
+	 */
+	for (i = 0; i < slot->tts_tupleDescriptor->natts; ++i)
+	{
+		Form_pg_attribute attr = slot->tts_tupleDescriptor->attrs[i];
+		MemoryContext	oldcontext;
+
+		/* Ignore nulls and non-records. */
+		if (slot->tts_isnull[i] || (attr->atttypid != RECORDOID
+			&& attr->atttypid != RECORDARRAYOID))
+			continue;
+
+		/*
+		 * OK, we're going to need to examine this attribute.  We could
+		 * use heap_deform_tuple here, but there's a possibility that the
+		 * slot already constains the deconstructed tuple, in which case
+		 * deforming it again would be needlessly inefficient.
+		 */
+		slot_getallattrs(slot);
+
+		/* Switch to temporary memory context to avoid leaking. */
+		if (tqueue->tmpcontext == NULL)
+			tqueue->tmpcontext =
+				AllocSetContextCreate(TopTransactionContext,
+									  "tqueue temporary context",
+									  ALLOCSET_DEFAULT_MINSIZE,
+									  ALLOCSET_DEFAULT_INITSIZE,
+									  ALLOCSET_DEFAULT_MAXSIZE);
+		oldcontext = MemoryContextSwitchTo(tqueue->tmpcontext);
+		if (attr->atttypid == RECORDOID)
+			tqueueWalkRecord(tqueue, slot->tts_values[i]);
+		else
+			tqueueWalkRecordArray(tqueue, slot->tts_values[i]);
+		MemoryContextSwitchTo(oldcontext);
+
+		/* Clean up anything memory we allocated. */
+		MemoryContextReset(tqueue->tmpcontext);
+	}
+
+	/* If we entered control mode, switch back to data mode. */
+	if (tqueue->mode != TUPLE_QUEUE_MODE_DATA)
+	{
+		tqueue->mode = TUPLE_QUEUE_MODE_DATA;
+		shm_mq_send(tqueue->handle, sizeof(char), &tqueue->mode, false);
+	}
+
+	/* Send the tuple itself. */
 	shm_mq_send(tqueue->handle, tuple->t_len, tuple->t_data, false);
 }
 
 /*
+ * Walk a record and send control messages for transient record types
+ * contained therein.
+ */
+static void
+tqueueWalkRecord(TQueueDestReceiver *tqueue, Datum value)
+{
+	HeapTupleHeader	tup;
+	Oid			typmod;
+	bool		found;
+	TupleDesc	tupledesc;
+	Datum	   *values;
+	bool	   *isnull;
+	HeapTupleData	tdata;
+	AttrNumber	i;
+
+	/* Extract typmod from tuple. */
+	tup = DatumGetHeapTupleHeader(value);
+	typmod = HeapTupleHeaderGetTypMod(tup);
+
+	/* Look up tuple descriptor in typecache. */
+	tupledesc = lookup_rowtype_tupdesc(RECORDOID, typmod);
+
+	/* Initialize hash table if not done yet. */
+	if (tqueue->recordhtab == NULL)
+	{
+		HASHCTL		ctl;
+
+		ctl.keysize = sizeof(int);
+		ctl.entrysize = sizeof(int);
+		ctl.hcxt = TopMemoryContext;
+		tqueue->recordhtab = hash_create("tqueue record hashtable",
+										 100, &ctl, HASH_ELEM | HASH_CONTEXT);
+	}
+
+	/* Have we already seen this record type?  If not, must report it. */
+	hash_search(tqueue->recordhtab, &typmod, HASH_ENTER, &found);
+	if (!found)
+	{
+		StringInfoData	buf;
+
+		/* If message queue is in data mode, switch to control mode. */
+		if (tqueue->mode != TUPLE_QUEUE_MODE_CONTROL)
+		{
+			tqueue->mode = TUPLE_QUEUE_MODE_CONTROL;
+			shm_mq_send(tqueue->handle, sizeof(char), &tqueue->mode, false);
+		}
+
+		/* Assemble a control message. */
+		initStringInfo(&buf);
+		appendBinaryStringInfo(&buf, (char *) &typmod, sizeof(int));
+		appendBinaryStringInfo(&buf, (char *) &tupledesc->natts, sizeof(int));
+		appendBinaryStringInfo(&buf, (char *) &tupledesc->tdhasoid,
+							   sizeof(bool));
+		for (i = 0; i < tupledesc->natts; ++i)
+			appendBinaryStringInfo(&buf, (char *) tupledesc->attrs[i],
+								   sizeof(FormData_pg_attribute));
+
+		/* Send control message. */
+		shm_mq_send(tqueue->handle, buf.len, buf.data, false);
+	}
+
+	/* Deform the tuple so we can check each column within. */
+	values = palloc(tupledesc->natts * sizeof(Datum));
+	isnull = palloc(tupledesc->natts * sizeof(bool));
+	tdata.t_len = HeapTupleHeaderGetDatumLength(tup);
+	ItemPointerSetInvalid(&(tdata.t_self));
+	tdata.t_tableOid = InvalidOid;
+	tdata.t_data = tup;
+	heap_deform_tuple(&tdata, tupledesc, values, isnull);
+
+	/* Recursively check each non-NULL attribute. */
+	for (i = 0; i < tupledesc->natts; ++i)
+	{
+		Form_pg_attribute attr = tupledesc->attrs[i];
+		if (isnull[i])
+			continue;
+		if (attr->atttypid == RECORDOID)
+			tqueueWalkRecord(tqueue, values[i]);
+		if (attr->atttypid == RECORDARRAYOID)
+			tqueueWalkRecordArray(tqueue, values[i]);
+	}
+
+	/* Release reference count acquired by lookup_rowtype_tupdesc. */
+	DecrTupleDescRefCount(tupledesc);
+}
+
+/*
+ * Walk a record and send control messages for transient record types
+ * contained therein.
+ */
+static void
+tqueueWalkRecordArray(TQueueDestReceiver *tqueue, Datum value)
+{
+	ArrayType  *arr = DatumGetArrayTypeP(value);
+	Datum	   *elem_values;
+	bool	   *elem_nulls;
+	int			num_elems;
+	int			i;
+
+	Assert(ARR_ELEMTYPE(arr) == RECORDOID);
+	deconstruct_array(arr, RECORDOID, -1, false, 'd',
+					  &elem_values, &elem_nulls, &num_elems);
+	for (i = 0; i < num_elems; ++i)
+		if (!elem_nulls[i])
+			tqueueWalkRecord(tqueue, elem_values[i]);
+}
+
+/*
  * Prepare to receive tuples from executor.
  */
 static void
@@ -77,6 +275,12 @@ tqueueShutdownReceiver(DestReceiver *self)
 static void
 tqueueDestroyReceiver(DestReceiver *self)
 {
+	TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
+
+	if (tqueue->tmpcontext != NULL)
+		MemoryContextDelete(tqueue->tmpcontext);
+	if (tqueue->recordhtab != NULL)
+		hash_destroy(tqueue->recordhtab);
 	pfree(self);
 }
 
@@ -96,6 +300,9 @@ CreateTupleQueueDestReceiver(shm_mq_handle *handle)
 	self->pub.rDestroy = tqueueDestroyReceiver;
 	self->pub.mydest = DestTupleQueue;
 	self->handle = handle;
+	self->tmpcontext = NULL;
+	self->recordhtab = NULL;
+	self->mode = TUPLE_QUEUE_MODE_DATA;
 
 	return (DestReceiver *) self;
 }
@@ -110,6 +317,7 @@ CreateTupleQueueFunnel(void)
 
 	funnel->maxqueues = 8;
 	funnel->queue = palloc(funnel->maxqueues * sizeof(shm_mq_handle *));
+	funnel->mode = palloc(funnel->maxqueues * sizeof(char));
 
 	return funnel;
 }
@@ -125,6 +333,9 @@ DestroyTupleQueueFunnel(TupleQueueFunnel *funnel)
 	for (i = 0; i < funnel->nqueues; i++)
 		shm_mq_detach(shm_mq_get_queue(funnel->queue[i]));
 	pfree(funnel->queue);
+	pfree(funnel->mode);
+	if (funnel->typmodmap != NULL)
+		hash_destroy(funnel->typmodmap);
 	pfree(funnel);
 }
 
@@ -134,12 +345,6 @@ DestroyTupleQueueFunnel(TupleQueueFunnel *funnel)
 void
 RegisterTupleQueueOnFunnel(TupleQueueFunnel *funnel, shm_mq_handle *handle)
 {
-	if (funnel->nqueues < funnel->maxqueues)
-	{
-		funnel->queue[funnel->nqueues++] = handle;
-		return;
-	}
-
 	if (funnel->nqueues >= funnel->maxqueues)
 	{
 		int			newsize = funnel->nqueues * 2;
@@ -148,10 +353,12 @@ RegisterTupleQueueOnFunnel(TupleQueueFunnel *funnel, shm_mq_handle *handle)
 
 		funnel->queue = repalloc(funnel->queue,
 								 newsize * sizeof(shm_mq_handle *));
+		funnel->mode = repalloc(funnel->mode, newsize * sizeof(bool));
 		funnel->maxqueues = newsize;
 	}
 
-	funnel->queue[funnel->nqueues++] = handle;
+	funnel->queue[funnel->nqueues] = handle;
+	funnel->mode[funnel->nqueues++] = TUPLE_QUEUE_MODE_DATA;
 }
 
 /*
@@ -172,7 +379,8 @@ RegisterTupleQueueOnFunnel(TupleQueueFunnel *funnel, shm_mq_handle *handle)
  * any other case.
  */
 HeapTuple
-TupleQueueFunnelNext(TupleQueueFunnel *funnel, bool nowait, bool *done)
+TupleQueueFunnelNext(TupleQueueFunnel *funnel, TupleDesc tupledesc,
+					 bool nowait, bool *done)
 {
 	int			waitpos = funnel->nextqueue;
 
@@ -190,6 +398,7 @@ TupleQueueFunnelNext(TupleQueueFunnel *funnel, bool nowait, bool *done)
 	for (;;)
 	{
 		shm_mq_handle *mqh = funnel->queue[funnel->nextqueue];
+		char	   *modep = &funnel->mode[funnel->nextqueue];
 		shm_mq_result result;
 		Size		nbytes;
 		void	   *data;
@@ -198,15 +407,10 @@ TupleQueueFunnelNext(TupleQueueFunnel *funnel, bool nowait, bool *done)
 		result = shm_mq_receive(mqh, &nbytes, &data, true);
 
 		/*
-		 * Normally, we advance funnel->nextqueue to the next queue at this
-		 * point, but if we're pointing to a queue that we've just discovered
-		 * is detached, then forget that queue and leave the pointer where it
-		 * is until the number of remaining queues fall below that pointer and
-		 * at that point make the pointer point to the first queue.
+		 * If this queue has been detached, forget about it and shift the
+		 * remmaining queues downward in the array.
 		 */
-		if (result != SHM_MQ_DETACHED)
-			funnel->nextqueue = (funnel->nextqueue + 1) % funnel->nqueues;
-		else
+		if (result == SHM_MQ_DETACHED)
 		{
 			--funnel->nqueues;
 			if (funnel->nqueues == 0)
@@ -230,21 +434,32 @@ TupleQueueFunnelNext(TupleQueueFunnel *funnel, bool nowait, bool *done)
 			continue;
 		}
 
+		/* Advance nextqueue pointer to next queue in round-robin fashion. */
+		funnel->nextqueue = (funnel->nextqueue + 1) % funnel->nqueues;
+
 		/* If we got a message, return it. */
 		if (result == SHM_MQ_SUCCESS)
 		{
-			HeapTupleData htup;
-
-			/*
-			 * The tuple data we just read from the queue is only valid until
-			 * we again attempt to read from it.  Copy the tuple into a single
-			 * palloc'd chunk as callers will expect.
-			 */
-			ItemPointerSetInvalid(&htup.t_self);
-			htup.t_tableOid = InvalidOid;
-			htup.t_len = nbytes;
-			htup.t_data = data;
-			return heap_copytuple(&htup);
+			if (nbytes == 1)
+			{
+				/* Mode switch message. */
+				*modep = ((char *) data)[0];
+				continue;
+			}
+			else if (*modep == TUPLE_QUEUE_MODE_DATA)
+			{
+				/* Tuple data. */
+				return TupleQueueHandleDataMessage(funnel, tupledesc,
+												   nbytes, data);
+			}
+			else if (*modep == TUPLE_QUEUE_MODE_CONTROL)
+			{
+				/* Control message, describing a transient record type. */
+				TupleQueueHandleControlMessage(funnel, nbytes, data);
+				continue;
+			}
+			else
+				elog(ERROR, "invalid mode: %d", (int) *modep);
 		}
 
 		/*
@@ -262,3 +477,224 @@ TupleQueueFunnelNext(TupleQueueFunnel *funnel, bool nowait, bool *done)
 		}
 	}
 }
+
+/*
+ * Handle a data message - that is, a tuple - from the tuple queue funnel.
+ */
+static HeapTuple
+TupleQueueHandleDataMessage(TupleQueueFunnel *funnel, TupleDesc tupledesc,
+							Size nbytes, HeapTupleHeader data)
+{
+	HeapTupleData htup;
+
+	ItemPointerSetInvalid(&htup.t_self);
+	htup.t_tableOid = InvalidOid;
+	htup.t_len = nbytes;
+	htup.t_data = data;
+
+	/* If necessary, remap record typmods. */
+	if (funnel->typmodmap != NULL)
+	{
+		HeapTuple	newtuple;
+
+		newtuple = TupleQueueRemapTuple(funnel, tupledesc, &htup);
+		if (newtuple != NULL)
+			return newtuple;
+	}
+
+	/*
+	 * Otherwise, just copy the tuple into a single palloc'd chunk, as
+	 * callers will expect.
+	 */
+	return heap_copytuple(&htup);
+}
+
+/*
+ * Remap tuple typmods per control information received from remote side.
+ */
+static HeapTuple
+TupleQueueRemapTuple(TupleQueueFunnel *funnel, TupleDesc tupledesc,
+					 HeapTuple tuple)
+{
+	Datum	   *values;
+	bool	   *isnull;
+	bool		dirty = false;
+	int			i;
+
+	/* Deform tuple so we can remap record typmods for individual attrs. */
+	values = palloc(tupledesc->natts * sizeof(Datum));
+	isnull = palloc(tupledesc->natts * sizeof(bool));
+	heap_deform_tuple(tuple, tupledesc, values, isnull);
+
+	/* Recursively check each non-NULL attribute. */
+	for (i = 0; i < tupledesc->natts; ++i)
+	{
+		Form_pg_attribute attr = tupledesc->attrs[i];
+
+		if (isnull[i])
+			continue;
+
+		if (attr->atttypid == RECORDOID)
+		{
+			values[i] = TupleQueueRemapRecord(funnel, values[i]);
+			dirty = true;
+		}
+
+
+		if (attr->atttypid == RECORDARRAYOID)
+		{
+			values[i] = TupleQueueRemapRecordArray(funnel, values[i]);
+			dirty = true;
+		}
+	}
+
+	/* If we didn't need to change anything, just return NULL. */
+	if (!dirty)
+		return NULL;
+
+	/* Reform the modified tuple. */
+	return heap_form_tuple(tupledesc, values, isnull);
+}
+
+static Datum
+TupleQueueRemapRecord(TupleQueueFunnel *funnel, Datum value)
+{
+	HeapTupleHeader	tup;
+	int				remotetypmod;
+	RecordTypemodMap *mapent;
+	TupleDesc		atupledesc;
+	HeapTupleData	htup;
+	HeapTuple		atup;
+
+	tup = DatumGetHeapTupleHeader(value);
+
+	/* Map remote typmod to local typmod and get tupledesc. */
+	remotetypmod = HeapTupleHeaderGetTypMod(tup);
+	Assert(funnel->typmodmap != NULL);
+	mapent = hash_search(funnel->typmodmap, &remotetypmod,
+						 HASH_FIND, NULL);
+	if (mapent == NULL)
+		elog(ERROR, "found unrecognized remote typmod %d",
+			 mapent->remotetypmod);
+	atupledesc = lookup_rowtype_tupdesc(RECORDOID, mapent->localtypmod);
+
+	/* Recursively process contents of record. */
+	ItemPointerSetInvalid(&htup.t_self);
+	htup.t_tableOid = InvalidOid;
+	htup.t_len = HeapTupleHeaderGetDatumLength(tup);
+	htup.t_data = tup;
+	atup = TupleQueueRemapTuple(funnel, atupledesc, &htup);
+
+	/* Release reference count acquired by lookup_rowtype_tupdesc. */
+	DecrTupleDescRefCount(atupledesc);
+
+	/*
+	 * Even if none of the attributes inside this tuple are records that
+	 * require typmod remapping, we still need to change the typmod on
+	 * the record itself.  However, we can do that by copying the tuple
+	 * rather than reforming it.
+	 */
+	if (atup == NULL)
+	{
+		atup = heap_copytuple(&htup);
+		HeapTupleHeaderSetTypMod(atup->t_data, mapent->localtypmod);
+	}
+
+	return HeapTupleHeaderGetDatum(atup->t_data);
+}
+
+static Datum
+TupleQueueRemapRecordArray(TupleQueueFunnel *funnel, Datum value)
+{
+	ArrayType  *arr = DatumGetArrayTypeP(value);
+	Datum	   *elem_values;
+	bool	   *elem_nulls;
+	int			num_elems;
+	int			i;
+
+	Assert(ARR_ELEMTYPE(arr) == RECORDOID);
+	deconstruct_array(arr, RECORDOID, -1, false, 'd',
+					  &elem_values, &elem_nulls, &num_elems);
+	for (i = 0; i < num_elems; ++i)
+		if (!elem_nulls[i])
+			elem_values[i] = TupleQueueRemapRecord(funnel, elem_values[i]);
+	arr = construct_md_array(elem_values, elem_nulls,
+							 ARR_NDIM(arr), ARR_DIMS(arr), ARR_LBOUND(arr),
+							 RECORDOID,
+							 -1, false, 'd');
+	return PointerGetDatum(arr);
+}
+
+/*
+ * Handle a control message from the tuple queue funnel.
+ *
+ * Control messages are sent when the remote side is sending tuples that
+ * contain transient record types.  We need to arrange to bless those
+ * record types locally and translate between remote and local typmods.
+ */
+static void
+TupleQueueHandleControlMessage(TupleQueueFunnel *funnel, Size nbytes,
+							   char *data)
+{
+	int		natts;
+	int		remotetypmod;
+	bool	hasoid;
+	char   *buf = data;
+	int		rc = 0;
+	int		i;
+	Form_pg_attribute *attrs;
+	MemoryContext	oldcontext;
+	TupleDesc	tupledesc;
+	RecordTypemodMap *mapent;
+	bool	found;
+
+	/* Extract remote typmod. */
+	memcpy(&remotetypmod, &buf[rc], sizeof(int));
+	rc += sizeof(int);
+
+	/* Extract attribute count. */
+	memcpy(&natts, &buf[rc], sizeof(int));
+	rc += sizeof(int);
+
+	/* Extract hasoid flag. */
+	memcpy(&hasoid, &buf[rc], sizeof(bool));
+	rc += sizeof(bool);
+
+	/* Extract attribute details. */
+	oldcontext = MemoryContextSwitchTo(CurTransactionContext);
+	attrs = palloc(natts * sizeof(Form_pg_attribute));
+	for (i = 0; i < natts; ++i)
+	{
+		attrs[i] = palloc(sizeof(FormData_pg_attribute));
+		memcpy(attrs[i], &buf[rc], sizeof(FormData_pg_attribute));
+		rc += sizeof(FormData_pg_attribute);
+	}
+	MemoryContextSwitchTo(oldcontext);
+
+	/* We should have read the whole message. */
+	Assert(rc == nbytes);
+
+	/* Construct TupleDesc. */
+	tupledesc = CreateTupleDesc(natts, hasoid, attrs);
+	tupledesc = BlessTupleDesc(tupledesc);
+
+	/* Create map if it doesn't exist already. */
+	if (funnel->typmodmap == NULL)
+	{
+		HASHCTL		ctl;
+
+		ctl.keysize = sizeof(int);
+		ctl.entrysize = sizeof(RecordTypemodMap);
+		ctl.hcxt = CurTransactionContext;
+		funnel->typmodmap = hash_create("typmodmap hashtable",
+							 100, &ctl, HASH_ELEM | HASH_CONTEXT);
+	}
+
+	/* Create map entry. */
+	mapent = hash_search(funnel->typmodmap, &remotetypmod, HASH_ENTER,
+						 &found);
+	if (found)
+		elog(ERROR, "duplicate message for typmod %d",
+			 remotetypmod);
+	mapent->localtypmod = tupledesc->tdtypmod;
+}
diff --git a/src/include/executor/tqueue.h b/src/include/executor/tqueue.h
index 6f8eb73..59f35c7 100644
--- a/src/include/executor/tqueue.h
+++ b/src/include/executor/tqueue.h
@@ -25,7 +25,7 @@ typedef struct TupleQueueFunnel TupleQueueFunnel;
 extern TupleQueueFunnel *CreateTupleQueueFunnel(void);
 extern void DestroyTupleQueueFunnel(TupleQueueFunnel *funnel);
 extern void RegisterTupleQueueOnFunnel(TupleQueueFunnel *, shm_mq_handle *);
-extern HeapTuple TupleQueueFunnelNext(TupleQueueFunnel *, bool nowait,
-					 bool *done);
+extern HeapTuple TupleQueueFunnelNext(TupleQueueFunnel *, TupleDesc tupledesc,
+					 bool nowait, bool *done);
 
 #endif   /* TQUEUE_H */
-- 
2.3.8 (Apple Git-58)

