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