PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
tqueue.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * tqueue.c
4  * Use shm_mq to send & receive tuples between parallel backends
5  *
6  * Most of the complexity in this module arises from transient RECORD types,
7  * which all have type RECORDOID and are distinguished by typmod numbers
8  * that are managed per-backend (see src/backend/utils/cache/typcache.c).
9  * The sender's set of RECORD typmod assignments probably doesn't match the
10  * receiver's. To deal with this, we make the sender send a description
11  * of each transient RECORD type appearing in the data it sends. The
12  * receiver finds or creates a matching type in its own typcache, and then
13  * maps the sender's typmod for that type to its own typmod.
14  *
15  * A DestReceiver of type DestTupleQueue, which is a TQueueDestReceiver
16  * under the hood, writes tuples from the executor to a shm_mq. If
17  * necessary, it also writes control messages describing transient
18  * record types used within the tuple.
19  *
20  * A TupleQueueReader reads tuples, and control messages if any are sent,
21  * from a shm_mq and returns the tuples. If transient record types are
22  * in use, it registers those types locally based on the control messages
23  * and rewrites the typmods sent by the remote side to the corresponding
24  * local record typmods.
25  *
26  * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
27  * Portions Copyright (c) 1994, Regents of the University of California
28  *
29  * IDENTIFICATION
30  * src/backend/executor/tqueue.c
31  *
32  *-------------------------------------------------------------------------
33  */
34 
35 #include "postgres.h"
36 
37 #include "access/htup_details.h"
38 #include "catalog/pg_type.h"
39 #include "executor/tqueue.h"
40 #include "funcapi.h"
41 #include "lib/stringinfo.h"
42 #include "miscadmin.h"
43 #include "utils/array.h"
44 #include "utils/lsyscache.h"
45 #include "utils/memutils.h"
46 #include "utils/rangetypes.h"
47 #include "utils/syscache.h"
48 #include "utils/typcache.h"
49 
50 
51 /*
52  * The data transferred through the shm_mq is divided into messages.
53  * One-byte messages are mode-switch messages, telling the receiver to switch
54  * between "control" and "data" modes. (We always start up in "data" mode.)
55  * Otherwise, when in "data" mode, each message is a tuple. When in "control"
56  * mode, each message defines one transient-typmod-to-tupledesc mapping to
57  * let us interpret future tuples. Both of those cases certainly require
58  * more than one byte, so no confusion is possible.
59  */
60 #define TUPLE_QUEUE_MODE_CONTROL 'c' /* mode-switch message contents */
61 #define TUPLE_QUEUE_MODE_DATA 'd'
62 
63 /*
64  * Both the sender and receiver build trees of TupleRemapInfo nodes to help
65  * them identify which (sub) fields of transmitted tuples are composite and
66  * may thus need remap processing. We might need to look within arrays and
67  * ranges, not only composites, to find composite sub-fields. A NULL
68  * TupleRemapInfo pointer indicates that it is known that the described field
69  * is not composite and has no composite substructure.
70  *
71  * Note that we currently have to look at each composite field at runtime,
72  * even if we believe it's of a named composite type (i.e., not RECORD).
73  * This is because we allow the actual value to be a compatible transient
74  * RECORD type. That's grossly inefficient, and it would be good to get
75  * rid of the requirement, but it's not clear what would need to change.
76  *
77  * Also, we allow the top-level tuple structure, as well as the actual
78  * structure of composite subfields, to change from one tuple to the next
79  * at runtime. This may well be entirely historical, but it's mostly free
80  * to support given the previous requirement; and other places in the system
81  * also permit this, so it's not entirely clear if we could drop it.
82  */
83 
84 typedef enum
85 {
86  TQUEUE_REMAP_ARRAY, /* array */
87  TQUEUE_REMAP_RANGE, /* range */
88  TQUEUE_REMAP_RECORD /* composite type, named or transient */
90 
92 
93 typedef struct ArrayRemapInfo
94 {
95  int16 typlen; /* array element type's storage properties */
96  bool typbyval;
97  char typalign;
98  TupleRemapInfo *element_remap; /* array element type's remap info */
100 
101 typedef struct RangeRemapInfo
102 {
103  TypeCacheEntry *typcache; /* range type's typcache entry */
104  TupleRemapInfo *bound_remap; /* range bound type's remap info */
106 
107 typedef struct RecordRemapInfo
108 {
109  /* Original (remote) type ID info last seen for this composite field */
112  /* Local RECORD typmod, or -1 if unset; not used on sender side */
114  /* If no fields of the record require remapping, these are NULL: */
115  TupleDesc tupledesc; /* copy of record's tupdesc */
116  TupleRemapInfo **field_remap; /* each field's remap info */
118 
120 {
122  union
123  {
127  } u;
128 };
129 
130 /*
131  * DestReceiver object's private contents
132  *
133  * queue and tupledesc are pointers to data supplied by DestReceiver's caller.
134  * The recordhtab and remap info are owned by the DestReceiver and are kept
135  * in mycontext. tmpcontext is a tuple-lifespan context to hold cruft
136  * created while traversing each tuple to find record subfields.
137  */
138 typedef struct TQueueDestReceiver
139 {
140  DestReceiver pub; /* public fields */
141  shm_mq_handle *queue; /* shm_mq to send to */
142  MemoryContext mycontext; /* context containing TQueueDestReceiver */
143  MemoryContext tmpcontext; /* per-tuple context, if needed */
144  HTAB *recordhtab; /* table of transmitted typmods, if needed */
145  char mode; /* current message mode */
146  TupleDesc tupledesc; /* current top-level tuple descriptor */
147  TupleRemapInfo **field_remapinfo; /* current top-level remap info */
149 
150 /*
151  * Hash table entries for mapping remote to local typmods.
152  */
153 typedef struct RecordTypmodMap
154 {
155  int32 remotetypmod; /* hash key (must be first!) */
158 
159 /*
160  * TupleQueueReader object's private contents
161  *
162  * queue and tupledesc are pointers to data supplied by reader's caller.
163  * The typmodmap and remap info are owned by the TupleQueueReader and
164  * are kept in mycontext.
165  *
166  * "typedef struct TupleQueueReader TupleQueueReader" is in tqueue.h
167  */
169 {
170  shm_mq_handle *queue; /* shm_mq to receive from */
171  MemoryContext mycontext; /* context containing TupleQueueReader */
172  HTAB *typmodmap; /* RecordTypmodMap hashtable, if needed */
173  char mode; /* current message mode */
174  TupleDesc tupledesc; /* current top-level tuple descriptor */
175  TupleRemapInfo **field_remapinfo; /* current top-level remap info */
176 };
177 
178 /* Local function prototypes */
179 static void TQExamine(TQueueDestReceiver *tqueue,
180  TupleRemapInfo *remapinfo,
181  Datum value);
182 static void TQExamineArray(TQueueDestReceiver *tqueue,
183  ArrayRemapInfo *remapinfo,
184  Datum value);
185 static void TQExamineRange(TQueueDestReceiver *tqueue,
186  RangeRemapInfo *remapinfo,
187  Datum value);
188 static void TQExamineRecord(TQueueDestReceiver *tqueue,
189  RecordRemapInfo *remapinfo,
190  Datum value);
191 static void TQSendRecordInfo(TQueueDestReceiver *tqueue, int32 typmod,
192  TupleDesc tupledesc);
194  Size nbytes, char *data);
196  Size nbytes, HeapTupleHeader data);
198  TupleDesc tupledesc,
199  TupleRemapInfo **field_remapinfo,
200  HeapTuple tuple);
201 static Datum TQRemap(TupleQueueReader *reader, TupleRemapInfo *remapinfo,
202  Datum value, bool *changed);
203 static Datum TQRemapArray(TupleQueueReader *reader, ArrayRemapInfo *remapinfo,
204  Datum value, bool *changed);
205 static Datum TQRemapRange(TupleQueueReader *reader, RangeRemapInfo *remapinfo,
206  Datum value, bool *changed);
207 static Datum TQRemapRecord(TupleQueueReader *reader, RecordRemapInfo *remapinfo,
208  Datum value, bool *changed);
209 static TupleRemapInfo *BuildTupleRemapInfo(Oid typid, MemoryContext mycontext);
210 static TupleRemapInfo *BuildArrayRemapInfo(Oid elemtypid,
211  MemoryContext mycontext);
212 static TupleRemapInfo *BuildRangeRemapInfo(Oid rngtypid,
213  MemoryContext mycontext);
214 static TupleRemapInfo **BuildFieldRemapInfo(TupleDesc tupledesc,
215  MemoryContext mycontext);
216 
217 
218 /*
219  * Receive a tuple from a query, and send it to the designated shm_mq.
220  *
221  * Returns TRUE if successful, FALSE if shm_mq has been detached.
222  */
223 static bool
225 {
226  TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
227  TupleDesc tupledesc = slot->tts_tupleDescriptor;
228  HeapTuple tuple;
229  shm_mq_result result;
230 
231  /*
232  * If first time through, compute remapping info for the top-level fields.
233  * On later calls, if the tupledesc has changed, set up for the new
234  * tupledesc. (This is a strange test both because the executor really
235  * shouldn't change the tupledesc, and also because it would be unsafe if
236  * the old tupledesc could be freed and a new one allocated at the same
237  * address. But since some very old code in printtup.c uses a similar
238  * approach, we adopt it here as well.)
239  *
240  * Here and elsewhere in this module, when replacing remapping info we
241  * pfree the top-level object because that's easy, but we don't bother to
242  * recursively free any substructure. This would lead to query-lifespan
243  * memory leaks if the mapping info actually changed frequently, but since
244  * we don't expect that to happen, it doesn't seem worth expending code to
245  * prevent it.
246  */
247  if (tqueue->tupledesc != tupledesc)
248  {
249  /* Is it worth trying to free substructure of the remap tree? */
250  if (tqueue->field_remapinfo != NULL)
251  pfree(tqueue->field_remapinfo);
252  tqueue->field_remapinfo = BuildFieldRemapInfo(tupledesc,
253  tqueue->mycontext);
254  tqueue->tupledesc = tupledesc;
255  }
256 
257  /*
258  * When, because of the types being transmitted, no record typmod mapping
259  * can be needed, we can skip a good deal of work.
260  */
261  if (tqueue->field_remapinfo != NULL)
262  {
263  TupleRemapInfo **remapinfo = tqueue->field_remapinfo;
264  int i;
265  MemoryContext oldcontext = NULL;
266 
267  /* Deform the tuple so we can examine fields, if not done already. */
268  slot_getallattrs(slot);
269 
270  /* Iterate over each attribute and search it for transient typmods. */
271  for (i = 0; i < tupledesc->natts; i++)
272  {
273  /* Ignore nulls and types that don't need special handling. */
274  if (slot->tts_isnull[i] || remapinfo[i] == NULL)
275  continue;
276 
277  /* Switch to temporary memory context to avoid leaking. */
278  if (oldcontext == NULL)
279  {
280  if (tqueue->tmpcontext == NULL)
281  tqueue->tmpcontext =
283  "tqueue sender temp context",
285  oldcontext = MemoryContextSwitchTo(tqueue->tmpcontext);
286  }
287 
288  /* Examine the value. */
289  TQExamine(tqueue, remapinfo[i], slot->tts_values[i]);
290  }
291 
292  /* If we used the temp context, reset it and restore prior context. */
293  if (oldcontext != NULL)
294  {
295  MemoryContextSwitchTo(oldcontext);
297  }
298 
299  /* If we entered control mode, switch back to data mode. */
300  if (tqueue->mode != TUPLE_QUEUE_MODE_DATA)
301  {
302  tqueue->mode = TUPLE_QUEUE_MODE_DATA;
303  shm_mq_send(tqueue->queue, sizeof(char), &tqueue->mode, false);
304  }
305  }
306 
307  /* Send the tuple itself. */
308  tuple = ExecMaterializeSlot(slot);
309  result = shm_mq_send(tqueue->queue, tuple->t_len, tuple->t_data, false);
310 
311  /* Check for failure. */
312  if (result == SHM_MQ_DETACHED)
313  return false;
314  else if (result != SHM_MQ_SUCCESS)
315  ereport(ERROR,
316  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
317  errmsg("could not send tuple to shared-memory queue")));
318 
319  return true;
320 }
321 
322 /*
323  * Examine the given datum and send any necessary control messages for
324  * transient record types contained in it.
325  *
326  * remapinfo is previously-computed remapping info about the datum's type.
327  *
328  * This function just dispatches based on the remap class.
329  */
330 static void
332 {
333  /* This is recursive, so it could be driven to stack overflow. */
335 
336  switch (remapinfo->remapclass)
337  {
338  case TQUEUE_REMAP_ARRAY:
339  TQExamineArray(tqueue, &remapinfo->u.arr, value);
340  break;
341  case TQUEUE_REMAP_RANGE:
342  TQExamineRange(tqueue, &remapinfo->u.rng, value);
343  break;
344  case TQUEUE_REMAP_RECORD:
345  TQExamineRecord(tqueue, &remapinfo->u.rec, value);
346  break;
347  }
348 }
349 
350 /*
351  * Examine a record datum and send any necessary control messages for
352  * transient record types contained in it.
353  */
354 static void
356  Datum value)
357 {
358  HeapTupleHeader tup;
359  Oid typid;
360  int32 typmod;
361  TupleDesc tupledesc;
362 
363  /* Extract type OID and typmod from tuple. */
364  tup = DatumGetHeapTupleHeader(value);
365  typid = HeapTupleHeaderGetTypeId(tup);
366  typmod = HeapTupleHeaderGetTypMod(tup);
367 
368  /*
369  * If first time through, or if this isn't the same composite type as last
370  * time, consider sending a control message, and then look up the
371  * necessary information for examining the fields.
372  */
373  if (typid != remapinfo->rectypid || typmod != remapinfo->rectypmod)
374  {
375  /* Free any old data. */
376  if (remapinfo->tupledesc != NULL)
377  FreeTupleDesc(remapinfo->tupledesc);
378  /* Is it worth trying to free substructure of the remap tree? */
379  if (remapinfo->field_remap != NULL)
380  pfree(remapinfo->field_remap);
381 
382  /* Look up tuple descriptor in typcache. */
383  tupledesc = lookup_rowtype_tupdesc(typid, typmod);
384 
385  /*
386  * If this is a transient record type, send the tupledesc in a control
387  * message. (TQSendRecordInfo is smart enough to do this only once
388  * per typmod.)
389  */
390  if (typid == RECORDOID)
391  TQSendRecordInfo(tqueue, typmod, tupledesc);
392 
393  /* Figure out whether fields need recursive processing. */
394  remapinfo->field_remap = BuildFieldRemapInfo(tupledesc,
395  tqueue->mycontext);
396  if (remapinfo->field_remap != NULL)
397  {
398  /*
399  * We need to inspect the record contents, so save a copy of the
400  * tupdesc. (We could possibly just reference the typcache's
401  * copy, but then it's problematic when to release the refcount.)
402  */
403  MemoryContext oldcontext = MemoryContextSwitchTo(tqueue->mycontext);
404 
405  remapinfo->tupledesc = CreateTupleDescCopy(tupledesc);
406  MemoryContextSwitchTo(oldcontext);
407  }
408  else
409  {
410  /* No fields of the record require remapping. */
411  remapinfo->tupledesc = NULL;
412  }
413  remapinfo->rectypid = typid;
414  remapinfo->rectypmod = typmod;
415 
416  /* Release reference count acquired by lookup_rowtype_tupdesc. */
417  DecrTupleDescRefCount(tupledesc);
418  }
419 
420  /*
421  * If field remapping is required, deform the tuple and examine each
422  * field.
423  */
424  if (remapinfo->field_remap != NULL)
425  {
426  Datum *values;
427  bool *isnull;
428  HeapTupleData tdata;
429  int i;
430 
431  /* Deform the tuple so we can check each column within. */
432  tupledesc = remapinfo->tupledesc;
433  values = (Datum *) palloc(tupledesc->natts * sizeof(Datum));
434  isnull = (bool *) palloc(tupledesc->natts * sizeof(bool));
436  ItemPointerSetInvalid(&(tdata.t_self));
437  tdata.t_tableOid = InvalidOid;
438  tdata.t_data = tup;
439  heap_deform_tuple(&tdata, tupledesc, values, isnull);
440 
441  /* Recursively check each interesting non-NULL attribute. */
442  for (i = 0; i < tupledesc->natts; i++)
443  {
444  if (!isnull[i] && remapinfo->field_remap[i])
445  TQExamine(tqueue, remapinfo->field_remap[i], values[i]);
446  }
447 
448  /* Need not clean up, since we're in a short-lived context. */
449  }
450 }
451 
452 /*
453  * Examine an array datum and send any necessary control messages for
454  * transient record types contained in it.
455  */
456 static void
458  Datum value)
459 {
460  ArrayType *arr = DatumGetArrayTypeP(value);
461  Oid typid = ARR_ELEMTYPE(arr);
462  Datum *elem_values;
463  bool *elem_nulls;
464  int num_elems;
465  int i;
466 
467  /* Deconstruct the array. */
468  deconstruct_array(arr, typid, remapinfo->typlen,
469  remapinfo->typbyval, remapinfo->typalign,
470  &elem_values, &elem_nulls, &num_elems);
471 
472  /* Examine each element. */
473  for (i = 0; i < num_elems; i++)
474  {
475  if (!elem_nulls[i])
476  TQExamine(tqueue, remapinfo->element_remap, elem_values[i]);
477  }
478 }
479 
480 /*
481  * Examine a range datum and send any necessary control messages for
482  * transient record types contained in it.
483  */
484 static void
486  Datum value)
487 {
491  bool empty;
492 
493  /* Extract the lower and upper bounds. */
494  range_deserialize(remapinfo->typcache, range, &lower, &upper, &empty);
495 
496  /* Nothing to do for an empty range. */
497  if (empty)
498  return;
499 
500  /* Examine each bound, if present. */
501  if (!upper.infinite)
502  TQExamine(tqueue, remapinfo->bound_remap, upper.val);
503  if (!lower.infinite)
504  TQExamine(tqueue, remapinfo->bound_remap, lower.val);
505 }
506 
507 /*
508  * Send tuple descriptor information for a transient typmod, unless we've
509  * already done so previously.
510  */
511 static void
513 {
515  bool found;
516  int i;
517 
518  /* Initialize hash table if not done yet. */
519  if (tqueue->recordhtab == NULL)
520  {
521  HASHCTL ctl;
522 
523  MemSet(&ctl, 0, sizeof(ctl));
524  /* Hash table entries are just typmods */
525  ctl.keysize = sizeof(int32);
526  ctl.entrysize = sizeof(int32);
527  ctl.hcxt = tqueue->mycontext;
528  tqueue->recordhtab = hash_create("tqueue sender record type hashtable",
529  100, &ctl,
531  }
532 
533  /* Have we already seen this record type? If not, must report it. */
534  hash_search(tqueue->recordhtab, &typmod, HASH_ENTER, &found);
535  if (found)
536  return;
537 
538  elog(DEBUG3, "sending tqueue control message for record typmod %d", typmod);
539 
540  /* If message queue is in data mode, switch to control mode. */
541  if (tqueue->mode != TUPLE_QUEUE_MODE_CONTROL)
542  {
543  tqueue->mode = TUPLE_QUEUE_MODE_CONTROL;
544  shm_mq_send(tqueue->queue, sizeof(char), &tqueue->mode, false);
545  }
546 
547  /* Assemble a control message. */
548  initStringInfo(&buf);
549  appendBinaryStringInfo(&buf, (char *) &typmod, sizeof(int32));
550  appendBinaryStringInfo(&buf, (char *) &tupledesc->natts, sizeof(int));
551  appendBinaryStringInfo(&buf, (char *) &tupledesc->tdhasoid, sizeof(bool));
552  for (i = 0; i < tupledesc->natts; i++)
553  {
554  appendBinaryStringInfo(&buf, (char *) tupledesc->attrs[i],
555  sizeof(FormData_pg_attribute));
556  }
557 
558  /* Send control message. */
559  shm_mq_send(tqueue->queue, buf.len, buf.data, false);
560 
561  /* We assume it's OK to leak buf because we're in a short-lived context. */
562 }
563 
564 /*
565  * Prepare to receive tuples from executor.
566  */
567 static void
568 tqueueStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)
569 {
570  /* do nothing */
571 }
572 
573 /*
574  * Clean up at end of an executor run
575  */
576 static void
578 {
579  TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
580 
582 }
583 
584 /*
585  * Destroy receiver when done with it
586  */
587 static void
589 {
590  TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
591 
592  if (tqueue->tmpcontext != NULL)
594  if (tqueue->recordhtab != NULL)
595  hash_destroy(tqueue->recordhtab);
596  /* Is it worth trying to free substructure of the remap tree? */
597  if (tqueue->field_remapinfo != NULL)
598  pfree(tqueue->field_remapinfo);
599  pfree(self);
600 }
601 
602 /*
603  * Create a DestReceiver that writes tuples to a tuple queue.
604  */
605 DestReceiver *
607 {
608  TQueueDestReceiver *self;
609 
610  self = (TQueueDestReceiver *) palloc0(sizeof(TQueueDestReceiver));
611 
612  self->pub.receiveSlot = tqueueReceiveSlot;
613  self->pub.rStartup = tqueueStartupReceiver;
614  self->pub.rShutdown = tqueueShutdownReceiver;
615  self->pub.rDestroy = tqueueDestroyReceiver;
616  self->pub.mydest = DestTupleQueue;
617  self->queue = handle;
618  self->mycontext = CurrentMemoryContext;
619  self->tmpcontext = NULL;
620  self->recordhtab = NULL;
621  self->mode = TUPLE_QUEUE_MODE_DATA;
622  /* Top-level tupledesc is not known yet */
623  self->tupledesc = NULL;
624  self->field_remapinfo = NULL;
625 
626  return (DestReceiver *) self;
627 }
628 
629 /*
630  * Create a tuple queue reader.
631  */
634 {
635  TupleQueueReader *reader = palloc0(sizeof(TupleQueueReader));
636 
637  reader->queue = handle;
639  reader->typmodmap = NULL;
640  reader->mode = TUPLE_QUEUE_MODE_DATA;
641  reader->tupledesc = tupledesc;
642  reader->field_remapinfo = BuildFieldRemapInfo(tupledesc, reader->mycontext);
643 
644  return reader;
645 }
646 
647 /*
648  * Destroy a tuple queue reader.
649  */
650 void
652 {
654  if (reader->typmodmap != NULL)
655  hash_destroy(reader->typmodmap);
656  /* Is it worth trying to free substructure of the remap tree? */
657  if (reader->field_remapinfo != NULL)
658  pfree(reader->field_remapinfo);
659  pfree(reader);
660 }
661 
662 /*
663  * Fetch a tuple from a tuple queue reader.
664  *
665  * The return value is NULL if there are no remaining tuples or if
666  * nowait = true and no tuple is ready to return. *done, if not NULL,
667  * is set to true when there are no remaining tuples and otherwise to false.
668  *
669  * The returned tuple, if any, is allocated in CurrentMemoryContext.
670  * That should be a short-lived (tuple-lifespan) context, because we are
671  * pretty cavalier about leaking memory in that context if we have to do
672  * tuple remapping.
673  *
674  * Even when shm_mq_receive() returns SHM_MQ_WOULD_BLOCK, this can still
675  * accumulate bytes from a partially-read message, so it's useful to call
676  * this with nowait = true even if nothing is returned.
677  */
678 HeapTuple
679 TupleQueueReaderNext(TupleQueueReader *reader, bool nowait, bool *done)
680 {
681  shm_mq_result result;
682 
683  if (done != NULL)
684  *done = false;
685 
686  for (;;)
687  {
688  Size nbytes;
689  void *data;
690 
691  /* Attempt to read a message. */
692  result = shm_mq_receive(reader->queue, &nbytes, &data, nowait);
693 
694  /* If queue is detached, set *done and return NULL. */
695  if (result == SHM_MQ_DETACHED)
696  {
697  if (done != NULL)
698  *done = true;
699  return NULL;
700  }
701 
702  /* In non-blocking mode, bail out if no message ready yet. */
703  if (result == SHM_MQ_WOULD_BLOCK)
704  return NULL;
705  Assert(result == SHM_MQ_SUCCESS);
706 
707  /*
708  * We got a message (see message spec at top of file). Process it.
709  */
710  if (nbytes == 1)
711  {
712  /* Mode switch message. */
713  reader->mode = ((char *) data)[0];
714  }
715  else if (reader->mode == TUPLE_QUEUE_MODE_DATA)
716  {
717  /* Tuple data. */
718  return TupleQueueHandleDataMessage(reader, nbytes, data);
719  }
720  else if (reader->mode == TUPLE_QUEUE_MODE_CONTROL)
721  {
722  /* Control message, describing a transient record type. */
723  TupleQueueHandleControlMessage(reader, nbytes, data);
724  }
725  else
726  elog(ERROR, "unrecognized tqueue mode: %d", (int) reader->mode);
727  }
728 }
729 
730 /*
731  * Handle a data message - that is, a tuple - from the remote side.
732  */
733 static HeapTuple
735  Size nbytes,
736  HeapTupleHeader data)
737 {
738  HeapTupleData htup;
739 
740  /*
741  * Set up a dummy HeapTupleData pointing to the data from the shm_mq
742  * (which had better be sufficiently aligned).
743  */
745  htup.t_tableOid = InvalidOid;
746  htup.t_len = nbytes;
747  htup.t_data = data;
748 
749  /*
750  * Either just copy the data into a regular palloc'd tuple, or remap it,
751  * as required.
752  */
753  return TQRemapTuple(reader,
754  reader->tupledesc,
755  reader->field_remapinfo,
756  &htup);
757 }
758 
759 /*
760  * Copy the given tuple, remapping any transient typmods contained in it.
761  */
762 static HeapTuple
764  TupleDesc tupledesc,
765  TupleRemapInfo **field_remapinfo,
766  HeapTuple tuple)
767 {
768  Datum *values;
769  bool *isnull;
770  bool changed = false;
771  int i;
772 
773  /*
774  * If no remapping is necessary, just copy the tuple into a single
775  * palloc'd chunk, as caller will expect.
776  */
777  if (field_remapinfo == NULL)
778  return heap_copytuple(tuple);
779 
780  /* Deform tuple so we can remap record typmods for individual attrs. */
781  values = (Datum *) palloc(tupledesc->natts * sizeof(Datum));
782  isnull = (bool *) palloc(tupledesc->natts * sizeof(bool));
783  heap_deform_tuple(tuple, tupledesc, values, isnull);
784 
785  /* Recursively process each interesting non-NULL attribute. */
786  for (i = 0; i < tupledesc->natts; i++)
787  {
788  if (isnull[i] || field_remapinfo[i] == NULL)
789  continue;
790  values[i] = TQRemap(reader, field_remapinfo[i], values[i], &changed);
791  }
792 
793  /* Reconstruct the modified tuple, if anything was modified. */
794  if (changed)
795  return heap_form_tuple(tupledesc, values, isnull);
796  else
797  return heap_copytuple(tuple);
798 }
799 
800 /*
801  * Process the given datum and replace any transient record typmods
802  * contained in it. Set *changed to TRUE if we actually changed the datum.
803  *
804  * remapinfo is previously-computed remapping info about the datum's type.
805  *
806  * This function just dispatches based on the remap class.
807  */
808 static Datum
810  Datum value, bool *changed)
811 {
812  /* This is recursive, so it could be driven to stack overflow. */
814 
815  switch (remapinfo->remapclass)
816  {
817  case TQUEUE_REMAP_ARRAY:
818  return TQRemapArray(reader, &remapinfo->u.arr, value, changed);
819 
820  case TQUEUE_REMAP_RANGE:
821  return TQRemapRange(reader, &remapinfo->u.rng, value, changed);
822 
823  case TQUEUE_REMAP_RECORD:
824  return TQRemapRecord(reader, &remapinfo->u.rec, value, changed);
825  }
826 
827  elog(ERROR, "unrecognized tqueue remap class: %d",
828  (int) remapinfo->remapclass);
829  return (Datum) 0;
830 }
831 
832 /*
833  * Process the given array datum and replace any transient record typmods
834  * contained in it. Set *changed to TRUE if we actually changed the datum.
835  */
836 static Datum
838  Datum value, bool *changed)
839 {
840  ArrayType *arr = DatumGetArrayTypeP(value);
841  Oid typid = ARR_ELEMTYPE(arr);
842  bool element_changed = false;
843  Datum *elem_values;
844  bool *elem_nulls;
845  int num_elems;
846  int i;
847 
848  /* Deconstruct the array. */
849  deconstruct_array(arr, typid, remapinfo->typlen,
850  remapinfo->typbyval, remapinfo->typalign,
851  &elem_values, &elem_nulls, &num_elems);
852 
853  /* Remap each element. */
854  for (i = 0; i < num_elems; i++)
855  {
856  if (!elem_nulls[i])
857  elem_values[i] = TQRemap(reader,
858  remapinfo->element_remap,
859  elem_values[i],
860  &element_changed);
861  }
862 
863  if (element_changed)
864  {
865  /* Reconstruct and return the array. */
866  *changed = true;
867  arr = construct_md_array(elem_values, elem_nulls,
868  ARR_NDIM(arr), ARR_DIMS(arr), ARR_LBOUND(arr),
869  typid, remapinfo->typlen,
870  remapinfo->typbyval, remapinfo->typalign);
871  return PointerGetDatum(arr);
872  }
873 
874  /* Else just return the value as-is. */
875  return value;
876 }
877 
878 /*
879  * Process the given range datum and replace any transient record typmods
880  * contained in it. Set *changed to TRUE if we actually changed the datum.
881  */
882 static Datum
884  Datum value, bool *changed)
885 {
887  bool bound_changed = false;
890  bool empty;
891 
892  /* Extract the lower and upper bounds. */
893  range_deserialize(remapinfo->typcache, range, &lower, &upper, &empty);
894 
895  /* Nothing to do for an empty range. */
896  if (empty)
897  return value;
898 
899  /* Remap each bound, if present. */
900  if (!upper.infinite)
901  upper.val = TQRemap(reader, remapinfo->bound_remap,
902  upper.val, &bound_changed);
903  if (!lower.infinite)
904  lower.val = TQRemap(reader, remapinfo->bound_remap,
905  lower.val, &bound_changed);
906 
907  if (bound_changed)
908  {
909  /* Reserialize. */
910  *changed = true;
911  range = range_serialize(remapinfo->typcache, &lower, &upper, empty);
912  return RangeTypeGetDatum(range);
913  }
914 
915  /* Else just return the value as-is. */
916  return value;
917 }
918 
919 /*
920  * Process the given record datum and replace any transient record typmods
921  * contained in it. Set *changed to TRUE if we actually changed the datum.
922  */
923 static Datum
925  Datum value, bool *changed)
926 {
927  HeapTupleHeader tup;
928  Oid typid;
929  int32 typmod;
930  bool changed_typmod;
931  TupleDesc tupledesc;
932 
933  /* Extract type OID and typmod from tuple. */
934  tup = DatumGetHeapTupleHeader(value);
935  typid = HeapTupleHeaderGetTypeId(tup);
936  typmod = HeapTupleHeaderGetTypMod(tup);
937 
938  /*
939  * If first time through, or if this isn't the same composite type as last
940  * time, identify the required typmod mapping, and then look up the
941  * necessary information for processing the fields.
942  */
943  if (typid != remapinfo->rectypid || typmod != remapinfo->rectypmod)
944  {
945  /* Free any old data. */
946  if (remapinfo->tupledesc != NULL)
947  FreeTupleDesc(remapinfo->tupledesc);
948  /* Is it worth trying to free substructure of the remap tree? */
949  if (remapinfo->field_remap != NULL)
950  pfree(remapinfo->field_remap);
951 
952  /* If transient record type, look up matching local typmod. */
953  if (typid == RECORDOID)
954  {
955  RecordTypmodMap *mapent;
956 
957  Assert(reader->typmodmap != NULL);
958  mapent = hash_search(reader->typmodmap, &typmod,
959  HASH_FIND, NULL);
960  if (mapent == NULL)
961  elog(ERROR, "tqueue received unrecognized remote typmod %d",
962  typmod);
963  remapinfo->localtypmod = mapent->localtypmod;
964  }
965  else
966  remapinfo->localtypmod = -1;
967 
968  /* Look up tuple descriptor in typcache. */
969  tupledesc = lookup_rowtype_tupdesc(typid, remapinfo->localtypmod);
970 
971  /* Figure out whether fields need recursive processing. */
972  remapinfo->field_remap = BuildFieldRemapInfo(tupledesc,
973  reader->mycontext);
974  if (remapinfo->field_remap != NULL)
975  {
976  /*
977  * We need to inspect the record contents, so save a copy of the
978  * tupdesc. (We could possibly just reference the typcache's
979  * copy, but then it's problematic when to release the refcount.)
980  */
981  MemoryContext oldcontext = MemoryContextSwitchTo(reader->mycontext);
982 
983  remapinfo->tupledesc = CreateTupleDescCopy(tupledesc);
984  MemoryContextSwitchTo(oldcontext);
985  }
986  else
987  {
988  /* No fields of the record require remapping. */
989  remapinfo->tupledesc = NULL;
990  }
991  remapinfo->rectypid = typid;
992  remapinfo->rectypmod = typmod;
993 
994  /* Release reference count acquired by lookup_rowtype_tupdesc. */
995  DecrTupleDescRefCount(tupledesc);
996  }
997 
998  /* If transient record, replace remote typmod with local typmod. */
999  if (typid == RECORDOID && typmod != remapinfo->localtypmod)
1000  {
1001  typmod = remapinfo->localtypmod;
1002  changed_typmod = true;
1003  }
1004  else
1005  changed_typmod = false;
1006 
1007  /*
1008  * If we need to change the typmod, or if there are any potentially
1009  * remappable fields, replace the tuple.
1010  */
1011  if (changed_typmod || remapinfo->field_remap != NULL)
1012  {
1013  HeapTupleData htup;
1014  HeapTuple atup;
1015 
1016  /* For now, assume we always need to change the tuple in this case. */
1017  *changed = true;
1018 
1019  /* Copy tuple, possibly remapping contained fields. */
1021  htup.t_tableOid = InvalidOid;
1023  htup.t_data = tup;
1024  atup = TQRemapTuple(reader,
1025  remapinfo->tupledesc,
1026  remapinfo->field_remap,
1027  &htup);
1028 
1029  /* Apply the correct labeling for a local Datum. */
1030  HeapTupleHeaderSetTypeId(atup->t_data, typid);
1031  HeapTupleHeaderSetTypMod(atup->t_data, typmod);
1033 
1034  /* And return the results. */
1035  return HeapTupleHeaderGetDatum(atup->t_data);
1036  }
1037 
1038  /* Else just return the value as-is. */
1039  return value;
1040 }
1041 
1042 /*
1043  * Handle a control message from the tuple queue reader.
1044  *
1045  * Control messages are sent when the remote side is sending tuples that
1046  * contain transient record types. We need to arrange to bless those
1047  * record types locally and translate between remote and local typmods.
1048  */
1049 static void
1051  char *data)
1052 {
1053  int32 remotetypmod;
1054  int natts;
1055  bool hasoid;
1056  Size offset = 0;
1057  Form_pg_attribute *attrs;
1058  TupleDesc tupledesc;
1059  RecordTypmodMap *mapent;
1060  bool found;
1061  int i;
1062 
1063  /* Extract remote typmod. */
1064  memcpy(&remotetypmod, &data[offset], sizeof(int32));
1065  offset += sizeof(int32);
1066 
1067  /* Extract attribute count. */
1068  memcpy(&natts, &data[offset], sizeof(int));
1069  offset += sizeof(int);
1070 
1071  /* Extract hasoid flag. */
1072  memcpy(&hasoid, &data[offset], sizeof(bool));
1073  offset += sizeof(bool);
1074 
1075  /* Extract attribute details. The tupledesc made here is just transient. */
1076  attrs = palloc(natts * sizeof(Form_pg_attribute));
1077  for (i = 0; i < natts; i++)
1078  {
1079  attrs[i] = palloc(sizeof(FormData_pg_attribute));
1080  memcpy(attrs[i], &data[offset], sizeof(FormData_pg_attribute));
1081  offset += sizeof(FormData_pg_attribute);
1082  }
1083 
1084  /* We should have read the whole message. */
1085  Assert(offset == nbytes);
1086 
1087  /* Construct TupleDesc, and assign a local typmod. */
1088  tupledesc = CreateTupleDesc(natts, hasoid, attrs);
1089  tupledesc = BlessTupleDesc(tupledesc);
1090 
1091  /* Create mapping hashtable if it doesn't exist already. */
1092  if (reader->typmodmap == NULL)
1093  {
1094  HASHCTL ctl;
1095 
1096  MemSet(&ctl, 0, sizeof(ctl));
1097  ctl.keysize = sizeof(int32);
1098  ctl.entrysize = sizeof(RecordTypmodMap);
1099  ctl.hcxt = reader->mycontext;
1100  reader->typmodmap = hash_create("tqueue receiver record type hashtable",
1101  100, &ctl,
1103  }
1104 
1105  /* Create map entry. */
1106  mapent = hash_search(reader->typmodmap, &remotetypmod, HASH_ENTER,
1107  &found);
1108  if (found)
1109  elog(ERROR, "duplicate tqueue control message for typmod %d",
1110  remotetypmod);
1111  mapent->localtypmod = tupledesc->tdtypmod;
1112 
1113  elog(DEBUG3, "tqueue mapping remote typmod %d to local typmod %d",
1114  remotetypmod, mapent->localtypmod);
1115 }
1116 
1117 /*
1118  * Build remap info for the specified data type, storing it in mycontext.
1119  * Returns NULL if neither the type nor any subtype could require remapping.
1120  */
1121 static TupleRemapInfo *
1123 {
1124  HeapTuple tup;
1125  Form_pg_type typ;
1126 
1127  /* This is recursive, so it could be driven to stack overflow. */
1129 
1130 restart:
1131  tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid));
1132  if (!HeapTupleIsValid(tup))
1133  elog(ERROR, "cache lookup failed for type %u", typid);
1134  typ = (Form_pg_type) GETSTRUCT(tup);
1135 
1136  /* Look through domains to underlying base type. */
1137  if (typ->typtype == TYPTYPE_DOMAIN)
1138  {
1139  typid = typ->typbasetype;
1140  ReleaseSysCache(tup);
1141  goto restart;
1142  }
1143 
1144  /* If it's a true array type, deal with it that way. */
1145  if (OidIsValid(typ->typelem) && typ->typlen == -1)
1146  {
1147  typid = typ->typelem;
1148  ReleaseSysCache(tup);
1149  return BuildArrayRemapInfo(typid, mycontext);
1150  }
1151 
1152  /* Similarly, deal with ranges appropriately. */
1153  if (typ->typtype == TYPTYPE_RANGE)
1154  {
1155  ReleaseSysCache(tup);
1156  return BuildRangeRemapInfo(typid, mycontext);
1157  }
1158 
1159  /*
1160  * If it's a composite type (including RECORD), set up for remapping. We
1161  * don't attempt to determine the status of subfields here, since we do
1162  * not have enough information yet; just mark everything invalid.
1163  */
1164  if (typ->typtype == TYPTYPE_COMPOSITE || typid == RECORDOID)
1165  {
1166  TupleRemapInfo *remapinfo;
1167 
1168  remapinfo = (TupleRemapInfo *)
1169  MemoryContextAlloc(mycontext, sizeof(TupleRemapInfo));
1170  remapinfo->remapclass = TQUEUE_REMAP_RECORD;
1171  remapinfo->u.rec.rectypid = InvalidOid;
1172  remapinfo->u.rec.rectypmod = -1;
1173  remapinfo->u.rec.localtypmod = -1;
1174  remapinfo->u.rec.tupledesc = NULL;
1175  remapinfo->u.rec.field_remap = NULL;
1176  ReleaseSysCache(tup);
1177  return remapinfo;
1178  }
1179 
1180  /* Nothing else can possibly need remapping attention. */
1181  ReleaseSysCache(tup);
1182  return NULL;
1183 }
1184 
1185 static TupleRemapInfo *
1187 {
1188  TupleRemapInfo *remapinfo;
1189  TupleRemapInfo *element_remapinfo;
1190 
1191  /* See if element type requires remapping. */
1192  element_remapinfo = BuildTupleRemapInfo(elemtypid, mycontext);
1193  /* If not, the array doesn't either. */
1194  if (element_remapinfo == NULL)
1195  return NULL;
1196  /* OK, set up to remap the array. */
1197  remapinfo = (TupleRemapInfo *)
1198  MemoryContextAlloc(mycontext, sizeof(TupleRemapInfo));
1199  remapinfo->remapclass = TQUEUE_REMAP_ARRAY;
1200  get_typlenbyvalalign(elemtypid,
1201  &remapinfo->u.arr.typlen,
1202  &remapinfo->u.arr.typbyval,
1203  &remapinfo->u.arr.typalign);
1204  remapinfo->u.arr.element_remap = element_remapinfo;
1205  return remapinfo;
1206 }
1207 
1208 static TupleRemapInfo *
1210 {
1211  TupleRemapInfo *remapinfo;
1212  TupleRemapInfo *bound_remapinfo;
1213  TypeCacheEntry *typcache;
1214 
1215  /*
1216  * Get range info from the typcache. We assume this pointer will stay
1217  * valid for the duration of the query.
1218  */
1219  typcache = lookup_type_cache(rngtypid, TYPECACHE_RANGE_INFO);
1220  if (typcache->rngelemtype == NULL)
1221  elog(ERROR, "type %u is not a range type", rngtypid);
1222 
1223  /* See if range bound type requires remapping. */
1224  bound_remapinfo = BuildTupleRemapInfo(typcache->rngelemtype->type_id,
1225  mycontext);
1226  /* If not, the range doesn't either. */
1227  if (bound_remapinfo == NULL)
1228  return NULL;
1229  /* OK, set up to remap the range. */
1230  remapinfo = (TupleRemapInfo *)
1231  MemoryContextAlloc(mycontext, sizeof(TupleRemapInfo));
1232  remapinfo->remapclass = TQUEUE_REMAP_RANGE;
1233  remapinfo->u.rng.typcache = typcache;
1234  remapinfo->u.rng.bound_remap = bound_remapinfo;
1235  return remapinfo;
1236 }
1237 
1238 /*
1239  * Build remap info for fields of the type described by the given tupdesc.
1240  * Returns an array of TupleRemapInfo pointers, or NULL if no field
1241  * requires remapping. Data is allocated in mycontext.
1242  */
1243 static TupleRemapInfo **
1245 {
1246  TupleRemapInfo **remapinfo;
1247  bool noop = true;
1248  int i;
1249 
1250  /* Recursively determine the remapping status of each field. */
1251  remapinfo = (TupleRemapInfo **)
1252  MemoryContextAlloc(mycontext,
1253  tupledesc->natts * sizeof(TupleRemapInfo *));
1254  for (i = 0; i < tupledesc->natts; i++)
1255  {
1256  Form_pg_attribute attr = tupledesc->attrs[i];
1257 
1258  if (attr->attisdropped)
1259  {
1260  remapinfo[i] = NULL;
1261  continue;
1262  }
1263  remapinfo[i] = BuildTupleRemapInfo(attr->atttypid, mycontext);
1264  if (remapinfo[i] != NULL)
1265  noop = false;
1266  }
1267 
1268  /* If no fields require remapping, report that by returning NULL. */
1269  if (noop)
1270  {
1271  pfree(remapinfo);
1272  remapinfo = NULL;
1273  }
1274 
1275  return remapinfo;
1276 }
ArrayRemapInfo arr
Definition: tqueue.c:124
signed short int16
Definition: c.h:252
HeapTuple heap_copytuple(HeapTuple tuple)
Definition: heaptuple.c:608
#define TYPTYPE_DOMAIN
Definition: pg_type.h:710
static struct @76 value
void hash_destroy(HTAB *hashp)
Definition: dynahash.c:793
static void tqueueDestroyReceiver(DestReceiver *self)
Definition: tqueue.c:588
TupleRemapClass
Definition: tqueue.c:84
TupleDesc CreateTupleDescCopy(TupleDesc tupdesc)
Definition: tupdesc.c:141
void DestroyTupleQueueReader(TupleQueueReader *reader)
Definition: tqueue.c:651
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:200
static void TQSendRecordInfo(TQueueDestReceiver *tqueue, int32 typmod, TupleDesc tupledesc)
Definition: tqueue.c:512
#define HeapTupleHeaderSetTypeId(tup, typeid)
Definition: htup_details.h:450
#define TYPECACHE_RANGE_INFO
Definition: typcache.h:121
#define RangeTypeGetDatum(X)
Definition: rangetypes.h:73
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
MemoryContext tmpcontext
Definition: tqueue.c:143
bool tdhasoid
Definition: tupdesc.h:79
static void tqueueStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)
Definition: tqueue.c:568
TypeCacheEntry * typcache
Definition: tqueue.c:103
static HeapTuple TupleQueueHandleDataMessage(TupleQueueReader *reader, Size nbytes, HeapTupleHeader data)
Definition: tqueue.c:734
#define HASH_CONTEXT
Definition: hsearch.h:93
static void TQExamineRecord(TQueueDestReceiver *tqueue, RecordRemapInfo *remapinfo, Datum value)
Definition: tqueue.c:355
#define HASH_ELEM
Definition: hsearch.h:87
TupleDesc lookup_rowtype_tupdesc(Oid type_id, int32 typmod)
Definition: typcache.c:1245
MemoryContext hcxt
Definition: hsearch.h:78
shm_mq_handle * queue
Definition: tqueue.c:170
#define DEBUG3
Definition: elog.h:23
Datum lower(PG_FUNCTION_ARGS)
Definition: oracle_compat.c:43
#define TYPTYPE_COMPOSITE
Definition: pg_type.h:709
void get_typlenbyvalalign(Oid typid, int16 *typlen, bool *typbyval, char *typalign)
Definition: lsyscache.c:1989
#define PointerGetDatum(X)
Definition: postgres.h:564
int32 localtypmod
Definition: tqueue.c:156
TupleDesc tupledesc
Definition: tqueue.c:146
Form_pg_attribute * attrs
Definition: tupdesc.h:74
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
#define DatumGetRangeType(X)
Definition: rangetypes.h:71
Datum val
Definition: rangetypes.h:62
Size entrysize
Definition: hsearch.h:73
HeapTuple TupleQueueReaderNext(TupleQueueReader *reader, bool nowait, bool *done)
Definition: tqueue.c:679
int errcode(int sqlerrcode)
Definition: elog.c:575
#define MemSet(start, val, len)
Definition: c.h:852
Datum * tts_values
Definition: tuptable.h:125
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:135
static bool tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
Definition: tqueue.c:224
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:692
static TupleRemapInfo * BuildTupleRemapInfo(Oid typid, MemoryContext mycontext)
Definition: tqueue.c:1122
Datum upper(PG_FUNCTION_ARGS)
Definition: oracle_compat.c:74
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:885
FormData_pg_type * Form_pg_type
Definition: pg_type.h:233
TupleDesc CreateTupleDesc(int natts, bool hasoid, Form_pg_attribute *attrs)
Definition: tupdesc.c:112
unsigned int Oid
Definition: postgres_ext.h:31
static void TQExamineArray(TQueueDestReceiver *tqueue, ArrayRemapInfo *remapinfo, Datum value)
Definition: tqueue.c:457
#define OidIsValid(objectId)
Definition: c.h:533
#define DatumGetHeapTupleHeader(X)
Definition: fmgr.h:254
int natts
Definition: tupdesc.h:73
int32 rectypmod
Definition: tqueue.c:111
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:149
int32 tdtypmod
Definition: tupdesc.h:78
char bool
Definition: c.h:199
signed int int32
Definition: c.h:253
TupleDesc tupledesc
Definition: tqueue.c:174
#define HeapTupleHeaderSetDatumLength(tup, len)
Definition: htup_details.h:442
void shm_mq_detach(shm_mq *mq)
Definition: shm_mq.c:778
char typalign
Definition: tqueue.c:97
TupleRemapClass remapclass
Definition: tqueue.c:121
#define ARR_LBOUND(a)
Definition: array.h:277
HeapTupleHeader t_data
Definition: htup.h:67
#define HeapTupleHeaderGetTypMod(tup)
Definition: htup_details.h:455
Definition: dynahash.c:193
HTAB * recordhtab
Definition: tqueue.c:144
void pfree(void *pointer)
Definition: mcxt.c:992
#define ObjectIdGetDatum(X)
Definition: postgres.h:515
#define ERROR
Definition: elog.h:43
static void TQExamineRange(TQueueDestReceiver *tqueue, RangeRemapInfo *remapinfo, Datum value)
Definition: tqueue.c:485
#define ARR_DIMS(a)
Definition: array.h:275
MemoryContext mycontext
Definition: tqueue.c:142
ItemPointerData t_self
Definition: htup.h:65
HTAB * typmodmap
Definition: tqueue.c:172
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:145
TupleDesc BlessTupleDesc(TupleDesc tupdesc)
Definition: execTuples.c:1031
static Datum TQRemapRecord(TupleQueueReader *reader, RecordRemapInfo *remapinfo, Datum value, bool *changed)
Definition: tqueue.c:924
uint32 t_len
Definition: htup.h:64
static Datum TQRemap(TupleQueueReader *reader, TupleRemapInfo *remapinfo, Datum value, bool *changed)
Definition: tqueue.c:809
static char * buf
Definition: pg_test_fsync.c:65
TupleRemapInfo * bound_remap
Definition: tqueue.c:104
bool * tts_isnull
Definition: tuptable.h:126
void check_stack_depth(void)
Definition: postgres.c:3096
bool typbyval
Definition: tqueue.c:96
static TupleRemapInfo ** BuildFieldRemapInfo(TupleDesc tupledesc, MemoryContext mycontext)
Definition: tqueue.c:1244
struct RecordTypmodMap RecordTypmodMap
FormData_pg_attribute * Form_pg_attribute
Definition: pg_attribute.h:184
Oid t_tableOid
Definition: htup.h:66
#define RECORDOID
Definition: pg_type.h:668
MemoryContext CurrentMemoryContext
Definition: mcxt.c:37
static struct cvec * range(struct vars *v, chr a, chr b, int cases)
Definition: regc_locale.c:416
#define ereport(elevel, rest)
Definition: elog.h:122
RangeType * range_serialize(TypeCacheEntry *typcache, RangeBound *lower, RangeBound *upper, bool empty)
Definition: rangetypes.c:1513
void slot_getallattrs(TupleTableSlot *slot)
Definition: heaptuple.c:1239
struct TypeCacheEntry * rngelemtype
Definition: typcache.h:84
static void TQExamine(TQueueDestReceiver *tqueue, TupleRemapInfo *remapinfo, Datum value)
Definition: tqueue.c:331
int32 remotetypmod
Definition: tqueue.c:155
struct RecordRemapInfo RecordRemapInfo
#define TYPTYPE_RANGE
Definition: pg_type.h:713
void initStringInfo(StringInfo str)
Definition: stringinfo.c:65
void range_deserialize(TypeCacheEntry *typcache, RangeType *range, RangeBound *lower, RangeBound *upper, bool *empty)
Definition: rangetypes.c:1642
struct ArrayRemapInfo ArrayRemapInfo
TupleDesc tts_tupleDescriptor
Definition: tuptable.h:121
#define HASH_BLOBS
Definition: hsearch.h:88
TupleRemapInfo ** field_remapinfo
Definition: tqueue.c:175
MemoryContext AllocSetContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
Definition: aset.c:440
void * palloc0(Size size)
Definition: mcxt.c:920
HTAB * hash_create(const char *tabname, long nelem, HASHCTL *info, int flags)
Definition: dynahash.c:301
uintptr_t Datum
Definition: postgres.h:374
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1083
#define HeapTupleHeaderSetTypMod(tup, typmod)
Definition: htup_details.h:460
#define HeapTupleHeaderGetTypeId(tup)
Definition: htup_details.h:445
Size keysize
Definition: hsearch.h:72
MemoryContext mycontext
Definition: tqueue.c:171
TypeCacheEntry * lookup_type_cache(Oid type_id, int flags)
Definition: typcache.c:191
#define InvalidOid
Definition: postgres_ext.h:36
FormData_pg_attribute
Definition: pg_attribute.h:168
shm_mq_result
Definition: shm_mq.h:36
static Datum TQRemapArray(TupleQueueReader *reader, ArrayRemapInfo *remapinfo, Datum value, bool *changed)
Definition: tqueue.c:837
union TupleRemapInfo::@21 u
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
shm_mq_handle * queue
Definition: tqueue.c:141
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:670
void DecrTupleDescRefCount(TupleDesc tupdesc)
Definition: tupdesc.c:334
static HeapTuple TQRemapTuple(TupleQueueReader *reader, TupleDesc tupledesc, TupleRemapInfo **field_remapinfo, HeapTuple tuple)
Definition: tqueue.c:763
size_t Size
Definition: c.h:352
shm_mq_result shm_mq_send(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait)
Definition: shm_mq.c:321
static void tqueueShutdownReceiver(DestReceiver *self)
Definition: tqueue.c:577
shm_mq * shm_mq_get_queue(shm_mq_handle *mqh)
Definition: shm_mq.c:802
static TupleRemapInfo * BuildArrayRemapInfo(Oid elemtypid, MemoryContext mycontext)
Definition: tqueue.c:1186
bool infinite
Definition: rangetypes.h:63
HeapTuple ExecMaterializeSlot(TupleTableSlot *slot)
Definition: execTuples.c:725
#define ARR_NDIM(a)
Definition: array.h:271
TupleRemapInfo * element_remap
Definition: tqueue.c:98
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:266
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
Definition: heaptuple.c:935
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3475
#define TUPLE_QUEUE_MODE_CONTROL
Definition: tqueue.c:60
static Datum values[MAXATTR]
Definition: bootstrap.c:162
struct TQueueDestReceiver TQueueDestReceiver
static TupleRemapInfo * BuildRangeRemapInfo(Oid rngtypid, MemoryContext mycontext)
Definition: tqueue.c:1209
static void TupleQueueHandleControlMessage(TupleQueueReader *reader, Size nbytes, char *data)
Definition: tqueue.c:1050
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:131
TupleRemapInfo ** field_remapinfo
Definition: tqueue.c:147
Datum HeapTupleHeaderGetDatum(HeapTupleHeader tuple)
Definition: execTuples.c:1201
RangeRemapInfo rng
Definition: tqueue.c:125
void * palloc(Size size)
Definition: mcxt.c:891
int errmsg(const char *fmt,...)
Definition: elog.c:797
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:749
TupleRemapInfo ** field_remap
Definition: tqueue.c:116
int i
TupleQueueReader * CreateTupleQueueReader(shm_mq_handle *handle, TupleDesc tupledesc)
Definition: tqueue.c:633
#define TUPLE_QUEUE_MODE_DATA
Definition: tqueue.c:61
DestReceiver pub
Definition: tqueue.c:140
#define elog
Definition: elog.h:219
DestReceiver * CreateTupleQueueDestReceiver(shm_mq_handle *handle)
Definition: tqueue.c:606
RecordRemapInfo rec
Definition: tqueue.c:126
ArrayType * construct_md_array(Datum *elems, bool *nulls, int ndims, int *dims, int *lbs, Oid elmtype, int elmlen, bool elmbyval, char elmalign)
Definition: arrayfuncs.c:3340
shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
Definition: shm_mq.c:518
struct RangeRemapInfo RangeRemapInfo
TupleDesc tupledesc
Definition: tqueue.c:115
#define ARR_ELEMTYPE(a)
Definition: array.h:273
static Datum TQRemapRange(TupleQueueReader *reader, RangeRemapInfo *remapinfo, Datum value, bool *changed)
Definition: tqueue.c:883
void appendBinaryStringInfo(StringInfo str, const char *data, int datalen)
Definition: stringinfo.c:240
int32 localtypmod
Definition: tqueue.c:113
int16 typlen
Definition: tqueue.c:95
#define HeapTupleHeaderGetDatumLength(tup)
Definition: htup_details.h:439
#define DatumGetArrayTypeP(X)
Definition: array.h:242