PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
logicalfuncs.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * logicalfuncs.c
4  *
5  * Support functions for using logical decoding and management of
6  * logical replication slots via SQL.
7  *
8  *
9  * Copyright (c) 2012-2017, PostgreSQL Global Development Group
10  *
11  * IDENTIFICATION
12  * src/backend/replication/logicalfuncs.c
13  *-------------------------------------------------------------------------
14  */
15 
16 #include "postgres.h"
17 
18 #include <unistd.h>
19 
20 #include "fmgr.h"
21 #include "funcapi.h"
22 #include "miscadmin.h"
23 
24 #include "access/xlog_internal.h"
25 #include "access/xlogutils.h"
26 
27 #include "access/xact.h"
28 
29 #include "catalog/pg_type.h"
30 
31 #include "nodes/makefuncs.h"
32 
33 #include "mb/pg_wchar.h"
34 
35 #include "utils/array.h"
36 #include "utils/builtins.h"
37 #include "utils/inval.h"
38 #include "utils/memutils.h"
39 #include "utils/pg_lsn.h"
40 #include "utils/regproc.h"
41 #include "utils/resowner.h"
42 #include "utils/lsyscache.h"
43 
44 #include "replication/decode.h"
45 #include "replication/logical.h"
47 #include "replication/message.h"
48 
49 #include "storage/fd.h"
50 
51 /* private date for writing out data */
52 typedef struct DecodingOutputState
53 {
59 
60 /*
61  * Prepare for an output plugin write.
62  */
63 static void
65  bool last_write)
66 {
67  resetStringInfo(ctx->out);
68 }
69 
70 /*
71  * Perform output plugin write into tuplestore.
72  */
73 static void
75  bool last_write)
76 {
77  Datum values[3];
78  bool nulls[3];
80 
81  /* SQL Datums can only be of a limited length... */
82  if (ctx->out->len > MaxAllocSize - VARHDRSZ)
83  elog(ERROR, "too much output for sql interface");
84 
86 
87  memset(nulls, 0, sizeof(nulls));
88  values[0] = LSNGetDatum(lsn);
89  values[1] = TransactionIdGetDatum(xid);
90 
91  /*
92  * Assert ctx->out is in database encoding when we're writing textual
93  * output.
94  */
95  if (!p->binary_output)
97  ctx->out->data, ctx->out->len,
98  false));
99 
100  /* ick, but cstring_to_text_with_len works for bytea perfectly fine */
101  values[2] = PointerGetDatum(
102  cstring_to_text_with_len(ctx->out->data, ctx->out->len));
103 
104  tuplestore_putvalues(p->tupstore, p->tupdesc, values, nulls);
105  p->returned_rows++;
106 }
107 
108 static void
110 {
111  if (!superuser() && !has_rolreplication(GetUserId()))
112  ereport(ERROR,
113  (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
114  (errmsg("must be superuser or replication role to use replication slots"))));
115 }
116 
117 int
119  int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
120 {
121  return read_local_xlog_page(state, targetPagePtr, reqLen,
122  targetRecPtr, cur_page, pageTLI);
123 }
124 
125 /*
126  * Helper function for the various SQL callable logical decoding functions.
127  */
128 static Datum
129 pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
130 {
131  Name name;
132  XLogRecPtr upto_lsn;
133  int32 upto_nchanges;
134  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
135  MemoryContext per_query_ctx;
136  MemoryContext oldcontext;
137  XLogRecPtr end_of_wal;
138  XLogRecPtr startptr;
140  ResourceOwner old_resowner = CurrentResourceOwner;
141  ArrayType *arr;
142  Size ndim;
143  List *options = NIL;
145 
147 
149 
150  if (PG_ARGISNULL(0))
151  ereport(ERROR,
152  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
153  errmsg("slot name must not be null")));
154  name = PG_GETARG_NAME(0);
155 
156  if (PG_ARGISNULL(1))
157  upto_lsn = InvalidXLogRecPtr;
158  else
159  upto_lsn = PG_GETARG_LSN(1);
160 
161  if (PG_ARGISNULL(2))
162  upto_nchanges = InvalidXLogRecPtr;
163  else
164  upto_nchanges = PG_GETARG_INT32(2);
165 
166  if (PG_ARGISNULL(3))
167  ereport(ERROR,
168  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
169  errmsg("options array must not be null")));
170  arr = PG_GETARG_ARRAYTYPE_P(3);
171 
172  /* check to see if caller supports us returning a tuplestore */
173  if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
174  ereport(ERROR,
175  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
176  errmsg("set-valued function called in context that cannot accept a set")));
177  if (!(rsinfo->allowedModes & SFRM_Materialize))
178  ereport(ERROR,
179  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
180  errmsg("materialize mode required, but it is not allowed in this context")));
181 
182  /* state to write output to */
183  p = palloc0(sizeof(DecodingOutputState));
184 
185  p->binary_output = binary;
186 
187  /* Build a tuple descriptor for our result type */
188  if (get_call_result_type(fcinfo, NULL, &p->tupdesc) != TYPEFUNC_COMPOSITE)
189  elog(ERROR, "return type must be a row type");
190 
191  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
192  oldcontext = MemoryContextSwitchTo(per_query_ctx);
193 
194  /* Deconstruct options array */
195  ndim = ARR_NDIM(arr);
196  if (ndim > 1)
197  {
198  ereport(ERROR,
199  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
200  errmsg("array must be one-dimensional")));
201  }
202  else if (array_contains_nulls(arr))
203  {
204  ereport(ERROR,
205  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
206  errmsg("array must not contain nulls")));
207  }
208  else if (ndim == 1)
209  {
210  int nelems;
211  Datum *datum_opts;
212  int i;
213 
214  Assert(ARR_ELEMTYPE(arr) == TEXTOID);
215 
216  deconstruct_array(arr, TEXTOID, -1, false, 'i',
217  &datum_opts, NULL, &nelems);
218 
219  if (nelems % 2 != 0)
220  ereport(ERROR,
221  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
222  errmsg("array must have even number of elements")));
223 
224  for (i = 0; i < nelems; i += 2)
225  {
226  char *name = TextDatumGetCString(datum_opts[i]);
227  char *opt = TextDatumGetCString(datum_opts[i + 1]);
228 
229  options = lappend(options, makeDefElem(name, (Node *) makeString(opt), -1));
230  }
231  }
232 
233  p->tupstore = tuplestore_begin_heap(true, false, work_mem);
234  rsinfo->returnMode = SFRM_Materialize;
235  rsinfo->setResult = p->tupstore;
236  rsinfo->setDesc = p->tupdesc;
237 
238  /*
239  * Compute the current end-of-wal and maintain ThisTimeLineID.
240  * RecoveryInProgress() will update ThisTimeLineID on promotion.
241  */
242  if (!RecoveryInProgress())
243  end_of_wal = GetFlushRecPtr();
244  else
245  end_of_wal = GetXLogReplayRecPtr(&ThisTimeLineID);
246 
248 
249  PG_TRY();
250  {
251  /* restart at slot's confirmed_flush */
253  options,
257 
258  MemoryContextSwitchTo(oldcontext);
259 
260  /*
261  * Check whether the output plugin writes textual output if that's
262  * what we need.
263  */
264  if (!binary &&
265  ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
266  ereport(ERROR,
267  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
268  errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
270  format_procedure(fcinfo->flinfo->fn_oid))));
271 
272  ctx->output_writer_private = p;
273 
274  /*
275  * Decoding of WAL must start at restart_lsn so that the entirety of
276  * xacts that committed after the slot's confirmed_flush can be
277  * accumulated into reorder buffers.
278  */
279  startptr = MyReplicationSlot->data.restart_lsn;
280 
282 
283  /* invalidate non-timetravel entries */
285 
286  /* Decode until we run out of records */
287  while ((startptr != InvalidXLogRecPtr && startptr < end_of_wal) ||
288  (ctx->reader->EndRecPtr != InvalidXLogRecPtr && ctx->reader->EndRecPtr < end_of_wal))
289  {
290  XLogRecord *record;
291  char *errm = NULL;
292 
293  record = XLogReadRecord(ctx->reader, startptr, &errm);
294  if (errm)
295  elog(ERROR, "%s", errm);
296 
297  /*
298  * Now that we've set up the xlog reader state, subsequent calls
299  * pass InvalidXLogRecPtr to say "continue from last record"
300  */
301  startptr = InvalidXLogRecPtr;
302 
303  /*
304  * The {begin_txn,change,commit_txn}_wrapper callbacks above will
305  * store the description into our tuplestore.
306  */
307  if (record != NULL)
308  LogicalDecodingProcessRecord(ctx, ctx->reader);
309 
310  /* check limits */
311  if (upto_lsn != InvalidXLogRecPtr &&
312  upto_lsn <= ctx->reader->EndRecPtr)
313  break;
314  if (upto_nchanges != 0 &&
315  upto_nchanges <= p->returned_rows)
316  break;
318  }
319 
320  tuplestore_donestoring(tupstore);
321 
322  CurrentResourceOwner = old_resowner;
323 
324  /*
325  * Next time, start where we left off. (Hunting things, the family
326  * business..)
327  */
328  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
329  {
330  LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);
331 
332  /*
333  * If only the confirmed_flush_lsn has changed the slot won't get
334  * marked as dirty by the above. Callers on the walsender
335  * interface are expected to keep track of their own progress and
336  * don't need it written out. But SQL-interface users cannot
337  * specify their own start positions and it's harder for them to
338  * keep track of their progress, so we should make more of an
339  * effort to save it for them.
340  *
341  * Dirty the slot so it's written out at the next checkpoint.
342  * We'll still lose its position on crash, as documented, but it's
343  * better than always losing the position even on clean restart.
344  */
346  }
347 
348  /* free context, call shutdown callback */
349  FreeDecodingContext(ctx);
350 
353  }
354  PG_CATCH();
355  {
356  /* clear all timetravel entries */
358 
359  PG_RE_THROW();
360  }
361  PG_END_TRY();
362 
363  return (Datum) 0;
364 }
365 
366 /*
367  * SQL function returning the changestream as text, consuming the data.
368  */
369 Datum
371 {
372  return pg_logical_slot_get_changes_guts(fcinfo, true, false);
373 }
374 
375 /*
376  * SQL function returning the changestream as text, only peeking ahead.
377  */
378 Datum
380 {
381  return pg_logical_slot_get_changes_guts(fcinfo, false, false);
382 }
383 
384 /*
385  * SQL function returning the changestream in binary, consuming the data.
386  */
387 Datum
389 {
390  return pg_logical_slot_get_changes_guts(fcinfo, true, true);
391 }
392 
393 /*
394  * SQL function returning the changestream in binary, only peeking ahead.
395  */
396 Datum
398 {
399  return pg_logical_slot_get_changes_guts(fcinfo, false, true);
400 }
401 
402 
403 /*
404  * SQL function for writing logical decoding message into WAL.
405  */
406 Datum
408 {
409  bool transactional = PG_GETARG_BOOL(0);
410  char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
411  bytea *data = PG_GETARG_BYTEA_PP(2);
412  XLogRecPtr lsn;
413 
414  lsn = LogLogicalMessage(prefix, VARDATA_ANY(data), VARSIZE_ANY_EXHDR(data),
415  transactional);
416  PG_RETURN_LSN(lsn);
417 }
418 
419 Datum
421 {
422  /* bytea and text are compatible */
423  return pg_logical_emit_message_bytea(fcinfo);
424 }
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, Datum *values, bool *isnull)
Definition: tuplestore.c:750
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:69
#define PG_GETARG_INT32(n)
Definition: fmgr.h:234
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define IsA(nodeptr, _type_)
Definition: nodes.h:560
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:211
#define VARDATA_ANY(PTR)
Definition: postgres.h:347
uint32 TimeLineID
Definition: xlogdefs.h:45
uint32 TransactionId
Definition: c.h:397
int logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
Definition: logicalfuncs.c:118
Oid GetUserId(void)
Definition: miscinit.c:283
#define TEXTOID
Definition: pg_type.h:324
#define PointerGetDatum(X)
Definition: postgres.h:562
#define VARHDRSZ
Definition: c.h:445
ResourceOwner CurrentResourceOwner
Definition: resowner.c:138
static void check_permissions(void)
Definition: logicalfuncs.c:109
#define tuplestore_donestoring(state)
Definition: tuplestore.h:60
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:407
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
Definition: nodes.h:509
int errcode(int sqlerrcode)
Definition: elog.c:575
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
bool superuser(void)
Definition: superuser.c:47
Datum pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:388
Tuplestorestate * tupstore
Definition: logicalfuncs.c:54
XLogRecPtr GetFlushRecPtr(void)
Definition: xlog.c:8221
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:239
ReplicationSlotPersistentData data
Definition: slot.h:115
bool RecoveryInProgress(void)
Definition: xlog.c:7872
DefElem * makeDefElem(char *name, Node *arg, int location)
Definition: makefuncs.c:544
void InvalidateSystemCaches(void)
Definition: inval.c:641
XLogRecord * XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
Definition: xlogreader.c:193
signed int int32
Definition: c.h:256
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:273
#define PG_RETURN_LSN(x)
Definition: pg_lsn.h:25
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
Definition: xlogutils.c:903
FmgrInfo * flinfo
Definition: fmgr.h:79
#define PG_GETARG_ARRAYTYPE_P(n)
Definition: array.h:244
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:129
#define ERROR
Definition: elog.h:43
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:93
struct DecodingOutputState DecodingOutputState
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11088
Definition: c.h:493
text * cstring_to_text_with_len(const char *s, int len)
Definition: varlena.c:161
fmNodePtr resultinfo
Definition: fmgr.h:81
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:62
void ReplicationSlotRelease(void)
Definition: slot.c:373
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:64
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
Definition: wchar.c:1877
#define ereport(elevel, rest)
Definition: elog.h:122
List * lappend(List *list, void *datum)
Definition: list.c:128
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional)
Definition: message.c:51
#define MaxAllocSize
Definition: memutils.h:40
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
#define TextDatumGetCString(d)
Definition: builtins.h:92
#define TransactionIdGetDatum(X)
Definition: postgres.h:527
void * palloc0(Size size)
Definition: mcxt.c:878
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:24
uintptr_t Datum
Definition: postgres.h:372
int GetDatabaseEncoding(void)
Definition: mbutils.c:1015
int work_mem
Definition: globals.c:113
TimeLineID ThisTimeLineID
Definition: xlog.c:179
Oid fn_oid
Definition: fmgr.h:59
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, XLogPageReadCB read_page, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:343
int allowedModes
Definition: execnodes.h:268
char * format_procedure(Oid procedure_oid)
Definition: regproc.c:323
Datum pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:370
SetFunctionReturnMode returnMode
Definition: execnodes.h:270
#define PG_CATCH()
Definition: elog.h:293
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define PG_ARGISNULL(n)
Definition: fmgr.h:174
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
Definition: regguts.h:298
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:475
XLogRecPtr restart_lsn
Definition: slot.h:68
void ReplicationSlotAcquire(const char *name)
Definition: slot.c:326
size_t Size
Definition: c.h:356
#define PG_GETARG_BYTEA_PP(n)
Definition: fmgr.h:272
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:464
#define PG_RE_THROW()
Definition: elog.h:314
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:914
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:202
#define ARR_NDIM(a)
Definition: array.h:271
const char * name
Definition: encode.c:521
Tuplestorestate * setResult
Definition: execnodes.h:273
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3475
static Datum values[MAXATTR]
Definition: bootstrap.c:163
char * text_to_cstring(const text *t)
Definition: varlena.c:182
ExprContext * econtext
Definition: execnodes.h:266
TupleDesc setDesc
Definition: execnodes.h:274
#define VARSIZE_ANY_EXHDR(PTR)
Definition: postgres.h:340
int errmsg(const char *fmt,...)
Definition: elog.c:797
StringInfo out
Definition: logical.h:66
int i
#define NameStr(name)
Definition: c.h:499
Datum pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:379
Definition: c.h:439
#define PG_FUNCTION_ARGS
Definition: fmgr.h:158
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:100
#define elog
Definition: elog.h:219
#define PG_TRY()
Definition: elog.h:284
bool array_contains_nulls(ArrayType *array)
Definition: arrayfuncs.c:3542
Definition: pg_list.h:45
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:76
#define ARR_ELEMTYPE(a)
Definition: array.h:273
void * output_writer_private
Definition: logical.h:76
#define PG_END_TRY()
Definition: elog.h:300
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:74
void ReplicationSlotMarkDirty(void)
Definition: slot.c:595
#define PG_GETARG_NAME(n)
Definition: fmgr.h:243
Datum pg_logical_emit_message_text(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:420
ResourceOwner ResourceOwnerCreate(ResourceOwner parent, const char *name)
Definition: resowner.c:416
Datum pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:397