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