PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
proto.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * proto.c
4  * logical replication protocol functions
5  *
6  * Copyright (c) 2015, PostgreSQL Global Development Group
7  *
8  * IDENTIFICATION
9  * src/backend/replication/logical/proto.c
10  *
11  *-------------------------------------------------------------------------
12  */
13 #include "postgres.h"
14 
15 #include "access/sysattr.h"
16 #include "catalog/pg_namespace.h"
17 #include "catalog/pg_type.h"
18 #include "libpq/pqformat.h"
20 #include "utils/builtins.h"
21 #include "utils/lsyscache.h"
22 #include "utils/syscache.h"
23 
24 /*
25  * Protocol message flags.
26  */
27 #define LOGICALREP_IS_REPLICA_IDENTITY 1
28 
29 static void logicalrep_write_attrs(StringInfo out, Relation rel);
30 static void logicalrep_write_tuple(StringInfo out, Relation rel,
31  HeapTuple tuple);
32 
35 
36 static void logicalrep_write_namespace(StringInfo out, Oid nspid);
37 static const char *logicalrep_read_namespace(StringInfo in);
38 
39 /*
40  * Write BEGIN to the output stream.
41  */
42 void
44 {
45  pq_sendbyte(out, 'B'); /* BEGIN */
46 
47  /* fixed fields */
48  pq_sendint64(out, txn->final_lsn);
49  pq_sendint64(out, txn->commit_time);
50  pq_sendint(out, txn->xid, 4);
51 }
52 
53 /*
54  * Read transaction BEGIN from the stream.
55  */
56 void
58 {
59  /* read fields */
60  begin_data->final_lsn = pq_getmsgint64(in);
61  if (begin_data->final_lsn == InvalidXLogRecPtr)
62  elog(ERROR, "final_lsn not set in begin message");
63  begin_data->committime = pq_getmsgint64(in);
64  begin_data->xid = pq_getmsgint(in, 4);
65 }
66 
67 
68 /*
69  * Write COMMIT to the output stream.
70  */
71 void
73  XLogRecPtr commit_lsn)
74 {
75  uint8 flags = 0;
76 
77  pq_sendbyte(out, 'C'); /* sending COMMIT */
78 
79  /* send the flags field (unused for now) */
80  pq_sendbyte(out, flags);
81 
82  /* send fields */
83  pq_sendint64(out, commit_lsn);
84  pq_sendint64(out, txn->end_lsn);
85  pq_sendint64(out, txn->commit_time);
86 }
87 
88 /*
89  * Read transaction COMMIT from the stream.
90  */
91 void
93 {
94  /* read flags (unused for now) */
95  uint8 flags = pq_getmsgbyte(in);
96 
97  if (flags != 0)
98  elog(ERROR, "unrecognized flags %u in commit message", flags);
99 
100  /* read fields */
101  commit_data->commit_lsn = pq_getmsgint64(in);
102  commit_data->end_lsn = pq_getmsgint64(in);
103  commit_data->committime = pq_getmsgint64(in);
104 }
105 
106 /*
107  * Write ORIGIN to the output stream.
108  */
109 void
110 logicalrep_write_origin(StringInfo out, const char *origin,
111  XLogRecPtr origin_lsn)
112 {
113  pq_sendbyte(out, 'O'); /* ORIGIN */
114 
115  /* fixed fields */
116  pq_sendint64(out, origin_lsn);
117 
118  /* origin string */
119  pq_sendstring(out, origin);
120 }
121 
122 /*
123  * Read ORIGIN from the output stream.
124  */
125 char *
127 {
128  /* fixed fields */
129  *origin_lsn = pq_getmsgint64(in);
130 
131  /* return origin */
132  return pstrdup(pq_getmsgstring(in));
133 }
134 
135 /*
136  * Write INSERT to the output stream.
137  */
138 void
140 {
141  pq_sendbyte(out, 'I'); /* action INSERT */
142 
143  Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
144  rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
145  rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);
146 
147  /* use Oid as relation identifier */
148  pq_sendint(out, RelationGetRelid(rel), 4);
149 
150  pq_sendbyte(out, 'N'); /* new tuple follows */
151  logicalrep_write_tuple(out, rel, newtuple);
152 }
153 
154 /*
155  * Read INSERT from stream.
156  *
157  * Fills the new tuple.
158  */
161 {
162  char action;
163  LogicalRepRelId relid;
164 
165  /* read the relation id */
166  relid = pq_getmsgint(in, 4);
167 
168  action = pq_getmsgbyte(in);
169  if (action != 'N')
170  elog(ERROR, "expected new tuple but got %d",
171  action);
172 
173  logicalrep_read_tuple(in, newtup);
174 
175  return relid;
176 }
177 
178 /*
179  * Write UPDATE to the output stream.
180  */
181 void
183  HeapTuple newtuple)
184 {
185  pq_sendbyte(out, 'U'); /* action UPDATE */
186 
187  Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
188  rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
189  rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);
190 
191  /* use Oid as relation identifier */
192  pq_sendint(out, RelationGetRelid(rel), 4);
193 
194  if (oldtuple != NULL)
195  {
196  if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
197  pq_sendbyte(out, 'O'); /* old tuple follows */
198  else
199  pq_sendbyte(out, 'K'); /* old key follows */
200  logicalrep_write_tuple(out, rel, oldtuple);
201  }
202 
203  pq_sendbyte(out, 'N'); /* new tuple follows */
204  logicalrep_write_tuple(out, rel, newtuple);
205 }
206 
207 /*
208  * Read UPDATE from stream.
209  */
211 logicalrep_read_update(StringInfo in, bool *has_oldtuple,
212  LogicalRepTupleData *oldtup,
213  LogicalRepTupleData *newtup)
214 {
215  char action;
216  LogicalRepRelId relid;
217 
218  /* read the relation id */
219  relid = pq_getmsgint(in, 4);
220 
221  /* read and verify action */
222  action = pq_getmsgbyte(in);
223  if (action != 'K' && action != 'O' && action != 'N')
224  elog(ERROR, "expected action 'N', 'O' or 'K', got %c",
225  action);
226 
227  /* check for old tuple */
228  if (action == 'K' || action == 'O')
229  {
230  logicalrep_read_tuple(in, oldtup);
231  *has_oldtuple = true;
232 
233  action = pq_getmsgbyte(in);
234  }
235  else
236  *has_oldtuple = false;
237 
238  /* check for new tuple */
239  if (action != 'N')
240  elog(ERROR, "expected action 'N', got %c",
241  action);
242 
243  logicalrep_read_tuple(in, newtup);
244 
245  return relid;
246 }
247 
248 /*
249  * Write DELETE to the output stream.
250  */
251 void
253 {
254  Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
255  rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
256  rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);
257 
258  pq_sendbyte(out, 'D'); /* action DELETE */
259 
260  /* use Oid as relation identifier */
261  pq_sendint(out, RelationGetRelid(rel), 4);
262 
263  if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
264  pq_sendbyte(out, 'O'); /* old tuple follows */
265  else
266  pq_sendbyte(out, 'K'); /* old key follows */
267 
268  logicalrep_write_tuple(out, rel, oldtuple);
269 }
270 
271 /*
272  * Read DELETE from stream.
273  *
274  * Fills the old tuple.
275  */
278 {
279  char action;
280  LogicalRepRelId relid;
281 
282  /* read the relation id */
283  relid = pq_getmsgint(in, 4);
284 
285  /* read and verify action */
286  action = pq_getmsgbyte(in);
287  if (action != 'K' && action != 'O')
288  elog(ERROR, "expected action 'O' or 'K', got %c", action);
289 
290  logicalrep_read_tuple(in, oldtup);
291 
292  return relid;
293 }
294 
295 /*
296  * Write relation description to the output stream.
297  */
298 void
300 {
301  char *relname;
302 
303  pq_sendbyte(out, 'R'); /* sending RELATION */
304 
305  /* use Oid as relation identifier */
306  pq_sendint(out, RelationGetRelid(rel), 4);
307 
308  /* send qualified relation name */
310  relname = RelationGetRelationName(rel);
311  pq_sendstring(out, relname);
312 
313  /* send replica identity */
314  pq_sendbyte(out, rel->rd_rel->relreplident);
315 
316  /* send the attribute info */
317  logicalrep_write_attrs(out, rel);
318 }
319 
320 /*
321  * Read the relation info from stream and return as LogicalRepRelation.
322  */
325 {
327 
328  rel->remoteid = pq_getmsgint(in, 4);
329 
330  /* Read relation name from stream */
332  rel->relname = pstrdup(pq_getmsgstring(in));
333 
334  /* Read the replica identity. */
335  rel->replident = pq_getmsgbyte(in);
336 
337  /* Get attribute description */
338  logicalrep_read_attrs(in, rel);
339 
340  return rel;
341 }
342 
343 /*
344  * Write type info to the output stream.
345  *
346  * This function will always write base type info.
347  */
348 void
350 {
351  Oid basetypoid = getBaseType(typoid);
352  HeapTuple tup;
353  Form_pg_type typtup;
354 
355  pq_sendbyte(out, 'Y'); /* sending TYPE */
356 
357  tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(basetypoid));
358  if (!HeapTupleIsValid(tup))
359  elog(ERROR, "cache lookup failed for type %u", basetypoid);
360  typtup = (Form_pg_type) GETSTRUCT(tup);
361 
362  /* use Oid as relation identifier */
363  pq_sendint(out, typoid, 4);
364 
365  /* send qualified type name */
366  logicalrep_write_namespace(out, typtup->typnamespace);
367  pq_sendstring(out, NameStr(typtup->typname));
368 
369  ReleaseSysCache(tup);
370 }
371 
372 /*
373  * Read type info from the output stream.
374  */
375 void
377 {
378  ltyp->remoteid = pq_getmsgint(in, 4);
379 
380  /* Read type name from stream */
382  ltyp->typname = pstrdup(pq_getmsgstring(in));
383 }
384 
385 /*
386  * Write a tuple to the outputstream, in the most efficient format possible.
387  */
388 static void
390 {
391  TupleDesc desc;
393  bool isnull[MaxTupleAttributeNumber];
394  int i;
395  uint16 nliveatts = 0;
396 
397  desc = RelationGetDescr(rel);
398 
399  for (i = 0; i < desc->natts; i++)
400  {
401  if (desc->attrs[i]->attisdropped)
402  continue;
403  nliveatts++;
404  }
405  pq_sendint(out, nliveatts, 2);
406 
407  /* try to allocate enough memory from the get-go */
408  enlargeStringInfo(out, tuple->t_len +
409  nliveatts * (1 + 4));
410 
411  heap_deform_tuple(tuple, desc, values, isnull);
412 
413  /* Write the values */
414  for (i = 0; i < desc->natts; i++)
415  {
416  HeapTuple typtup;
417  Form_pg_type typclass;
418  Form_pg_attribute att = desc->attrs[i];
419  char *outputstr;
420 
421  /* skip dropped columns */
422  if (att->attisdropped)
423  continue;
424 
425  if (isnull[i])
426  {
427  pq_sendbyte(out, 'n'); /* null column */
428  continue;
429  }
430  else if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(values[i]))
431  {
432  pq_sendbyte(out, 'u'); /* unchanged toast column */
433  continue;
434  }
435 
436  typtup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(att->atttypid));
437  if (!HeapTupleIsValid(typtup))
438  elog(ERROR, "cache lookup failed for type %u", att->atttypid);
439  typclass = (Form_pg_type) GETSTRUCT(typtup);
440 
441  pq_sendbyte(out, 't'); /* 'text' data follows */
442 
443  outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
444  pq_sendcountedtext(out, outputstr, strlen(outputstr), false);
445  pfree(outputstr);
446 
447  ReleaseSysCache(typtup);
448  }
449 }
450 
451 /*
452  * Read tuple in remote format from stream.
453  *
454  * The returned tuple points into the input stringinfo.
455  */
456 static void
458 {
459  int i;
460  int natts;
461 
462  /* Get number of attributes */
463  natts = pq_getmsgint(in, 2);
464 
465  memset(tuple->changed, 0, sizeof(tuple->changed));
466 
467  /* Read the data */
468  for (i = 0; i < natts; i++)
469  {
470  char kind;
471 
472  kind = pq_getmsgbyte(in);
473 
474  switch (kind)
475  {
476  case 'n': /* null */
477  tuple->values[i] = NULL;
478  tuple->changed[i] = true;
479  break;
480  case 'u': /* unchanged column */
481  /* we don't receive the value of an unchanged column */
482  tuple->values[i] = NULL;
483  break;
484  case 't': /* text formatted value */
485  {
486  int len;
487 
488  tuple->changed[i] = true;
489 
490  len = pq_getmsgint(in, 4); /* read length */
491 
492  /* and data */
493  tuple->values[i] = palloc(len + 1);
494  pq_copymsgbytes(in, tuple->values[i], len);
495  tuple->values[i][len] = '\0';
496  }
497  break;
498  default:
499  elog(ERROR, "unrecognized data representation type '%c'", kind);
500  }
501  }
502 }
503 
504 /*
505  * Write relation attributes to the stream.
506  */
507 static void
509 {
510  TupleDesc desc;
511  int i;
512  uint16 nliveatts = 0;
513  Bitmapset *idattrs = NULL;
514  bool replidentfull;
515 
516  desc = RelationGetDescr(rel);
517 
518  /* send number of live attributes */
519  for (i = 0; i < desc->natts; i++)
520  {
521  if (desc->attrs[i]->attisdropped)
522  continue;
523  nliveatts++;
524  }
525  pq_sendint(out, nliveatts, 2);
526 
527  /* fetch bitmap of REPLICATION IDENTITY attributes */
528  replidentfull = (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL);
529  if (!replidentfull)
530  idattrs = RelationGetIndexAttrBitmap(rel,
532 
533  /* send the attributes */
534  for (i = 0; i < desc->natts; i++)
535  {
536  Form_pg_attribute att = desc->attrs[i];
537  uint8 flags = 0;
538 
539  if (att->attisdropped)
540  continue;
541 
542  /* REPLICA IDENTITY FULL means all columns are sent as part of key. */
543  if (replidentfull ||
545  idattrs))
547 
548  pq_sendbyte(out, flags);
549 
550  /* attribute name */
551  pq_sendstring(out, NameStr(att->attname));
552 
553  /* attribute type id */
554  pq_sendint(out, (int) att->atttypid, sizeof(att->atttypid));
555 
556  /* attribute mode */
557  pq_sendint(out, att->atttypmod, sizeof(att->atttypmod));
558  }
559 
560  bms_free(idattrs);
561 }
562 
563 /*
564  * Read relation attribute names from the stream.
565  */
566 static void
568 {
569  int i;
570  int natts;
571  char **attnames;
572  Oid *atttyps;
573  Bitmapset *attkeys = NULL;
574 
575  natts = pq_getmsgint(in, 2);
576  attnames = palloc(natts * sizeof(char *));
577  atttyps = palloc(natts * sizeof(Oid));
578 
579  /* read the attributes */
580  for (i = 0; i < natts; i++)
581  {
582  uint8 flags;
583 
584  /* Check for replica identity column */
585  flags = pq_getmsgbyte(in);
586  if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
587  attkeys = bms_add_member(attkeys, i);
588 
589  /* attribute name */
590  attnames[i] = pstrdup(pq_getmsgstring(in));
591 
592  /* attribute type id */
593  atttyps[i] = (Oid) pq_getmsgint(in, 4);
594 
595  /* we ignore attribute mode for now */
596  (void) pq_getmsgint(in, 4);
597  }
598 
599  rel->attnames = attnames;
600  rel->atttyps = atttyps;
601  rel->attkeys = attkeys;
602  rel->natts = natts;
603 }
604 
605 /*
606  * Write the namespace name or empty string for pg_catalog (to save space).
607  */
608 static void
610 {
611  if (nspid == PG_CATALOG_NAMESPACE)
612  pq_sendbyte(out, '\0');
613  else
614  {
615  char *nspname = get_namespace_name(nspid);
616 
617  if (nspname == NULL)
618  elog(ERROR, "cache lookup failed for namespace %u",
619  nspid);
620 
621  pq_sendstring(out, nspname);
622  }
623 }
624 
625 /*
626  * Read the namespace name while treating empty string as pg_catalog.
627  */
628 static const char *
630 {
631  const char *nspname = pq_getmsgstring(in);
632 
633  if (nspname[0] == '\0')
634  nspname = "pg_catalog";
635 
636  return nspname;
637 }
TimestampTz commit_time
void logicalrep_write_typ(StringInfo out, Oid typoid)
Definition: proto.c:349
#define VARATT_IS_EXTERNAL_ONDISK(PTR)
Definition: postgres.h:315
TransactionId xid
Definition: logicalproto.h:69
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup)
Definition: proto.c:277
char * typname
Definition: logicalproto.h:60
void pq_sendbyte(StringInfo buf, int byt)
Definition: pqformat.c:105
#define MaxTupleAttributeNumber
Definition: htup_details.h:33
#define RelationGetDescr(relation)
Definition: rel.h:428
void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data)
Definition: proto.c:57
const char * pq_getmsgstring(StringInfo msg)
Definition: pqformat.c:621
char * pstrdup(const char *in)
Definition: mcxt.c:1077
static void logicalrep_write_namespace(StringInfo out, Oid nspid)
Definition: proto.c:609
LogicalRepRelation * logicalrep_read_rel(StringInfo in)
Definition: proto.c:324
Form_pg_attribute * attrs
Definition: tupdesc.h:74
unsigned char uint8
Definition: c.h:266
#define REPLICA_IDENTITY_DEFAULT
Definition: pg_class.h:175
void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
Definition: proto.c:72
#define FirstLowInvalidHeapAttributeNumber
Definition: sysattr.h:28
void pq_sendstring(StringInfo buf, const char *str)
Definition: pqformat.c:186
void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn)
Definition: proto.c:110
FormData_pg_type * Form_pg_type
Definition: pg_type.h:233
Form_pg_class rd_rel
Definition: rel.h:114
unsigned int Oid
Definition: postgres_ext.h:31
Bitmapset * attkeys
Definition: logicalproto.h:52
int natts
Definition: tupdesc.h:73
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:156
LogicalRepRelId remoteid
Definition: logicalproto.h:45
#define LOGICALREP_IS_REPLICA_IDENTITY
Definition: proto.c:27
LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
Definition: proto.c:160
unsigned short uint16
Definition: c.h:267
void pfree(void *pointer)
Definition: mcxt.c:950
#define REPLICA_IDENTITY_FULL
Definition: pg_class.h:179
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
void logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
Definition: proto.c:376
uint32 t_len
Definition: htup.h:64
char * get_namespace_name(Oid nspid)
Definition: lsyscache.c:3033
bool changed[MaxTupleAttributeNumber]
Definition: logicalproto.h:36
static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
Definition: proto.c:457
void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple)
Definition: proto.c:252
static void logicalrep_write_attrs(StringInfo out, Relation rel)
Definition: proto.c:508
void enlargeStringInfo(StringInfo str, int needed)
Definition: stringinfo.c:245
#define RelationGetRelationName(relation)
Definition: rel.h:436
FormData_pg_attribute * Form_pg_attribute
Definition: pg_attribute.h:187
XLogRecPtr final_lsn
void pq_sendcountedtext(StringInfo buf, const char *str, int slen, bool countincludesself)
Definition: pqformat.c:131
static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
Definition: proto.c:567
#define PG_CATALOG_NAMESPACE
Definition: pg_namespace.h:71
void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple)
Definition: proto.c:139
void logicalrep_write_rel(StringInfo out, Relation rel)
Definition: proto.c:299
void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn)
Definition: proto.c:43
XLogRecPtr final_lsn
Definition: logicalproto.h:67
uintptr_t Datum
Definition: postgres.h:372
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1117
TransactionId xid
static const char * logicalrep_read_namespace(StringInfo in)
Definition: proto.c:629
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:432
void bms_free(Bitmapset *a)
Definition: bitmapset.c:201
void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
Definition: proto.c:182
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
void pq_copymsgbytes(StringInfo msg, char *buf, int datalen)
Definition: pqformat.c:570
XLogRecPtr end_lsn
LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
Definition: proto.c:211
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:698
void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data)
Definition: proto.c:92
char * values[MaxTupleAttributeNumber]
Definition: logicalproto.h:34
void heap_deform_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *values, bool *isnull)
Definition: heaptuple.c:933
static Datum values[MAXATTR]
Definition: bootstrap.c:163
char * OidOutputFunctionCall(Oid functionId, Datum val)
Definition: fmgr.c:1747
void * palloc(Size size)
Definition: mcxt.c:849
void pq_sendint(StringInfo buf, int i, int b)
Definition: pqformat.c:236
int i
int64 pq_getmsgint64(StringInfo msg)
Definition: pqformat.c:486
Bitmapset * RelationGetIndexAttrBitmap(Relation relation, IndexAttrBitmapKind attrKind)
Definition: relcache.c:4847
#define NameStr(name)
Definition: c.h:499
char * logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn)
Definition: proto.c:126
unsigned int pq_getmsgint(StringInfo msg, int b)
Definition: pqformat.c:448
#define elog
Definition: elog.h:219
TimestampTz committime
Definition: logicalproto.h:68
#define REPLICA_IDENTITY_INDEX
Definition: pg_class.h:185
XLogRecPtr commit_lsn
Definition: logicalproto.h:74
Oid getBaseType(Oid typid)
Definition: lsyscache.c:2271
bool bms_is_member(int x, const Bitmapset *a)
Definition: bitmapset.c:420
void pq_sendint64(StringInfo buf, int64 i)
Definition: pqformat.c:271
#define RelationGetRelid(relation)
Definition: rel.h:416
static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple)
Definition: proto.c:389
TimestampTz committime
Definition: logicalproto.h:76
char * nspname
Definition: logicalproto.h:59
uint32 LogicalRepRelId
Definition: logicalproto.h:39
#define RelationGetNamespace(relation)
Definition: rel.h:443