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-2024, PostgreSQL Global Development Group
10  *
11  * IDENTIFICATION
12  * src/backend/replication/logical/logicalfuncs.c
13  *-------------------------------------------------------------------------
14  */
15 
16 #include "postgres.h"
17 
18 #include <unistd.h>
19 
20 #include "access/xlogrecovery.h"
21 #include "access/xlogutils.h"
22 #include "catalog/pg_type.h"
23 #include "fmgr.h"
24 #include "funcapi.h"
25 #include "mb/pg_wchar.h"
26 #include "miscadmin.h"
27 #include "nodes/makefuncs.h"
28 #include "replication/decode.h"
29 #include "replication/logical.h"
30 #include "replication/message.h"
31 #include "utils/array.h"
32 #include "utils/builtins.h"
33 #include "utils/inval.h"
34 #include "utils/memutils.h"
35 #include "utils/pg_lsn.h"
36 #include "utils/regproc.h"
37 #include "utils/resowner.h"
38 
39 /* Private data for writing out data */
40 typedef struct DecodingOutputState
41 {
47 
48 /*
49  * Prepare for an output plugin write.
50  */
51 static void
53  bool last_write)
54 {
55  resetStringInfo(ctx->out);
56 }
57 
58 /*
59  * Perform output plugin write into tuplestore.
60  */
61 static void
63  bool last_write)
64 {
65  Datum values[3];
66  bool nulls[3];
68 
69  /* SQL Datums can only be of a limited length... */
70  if (ctx->out->len > MaxAllocSize - VARHDRSZ)
71  elog(ERROR, "too much output for sql interface");
72 
74 
75  memset(nulls, 0, sizeof(nulls));
76  values[0] = LSNGetDatum(lsn);
77  values[1] = TransactionIdGetDatum(xid);
78 
79  /*
80  * Assert ctx->out is in database encoding when we're writing textual
81  * output.
82  */
83  if (!p->binary_output)
85  ctx->out->data, ctx->out->len,
86  false));
87 
88  /* ick, but cstring_to_text_with_len works for bytea perfectly fine */
90 
92  p->returned_rows++;
93 }
94 
95 /*
96  * Helper function for the various SQL callable logical decoding functions.
97  */
98 static Datum
99 pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
100 {
101  Name name;
102  XLogRecPtr upto_lsn;
103  int32 upto_nchanges;
104  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
105  MemoryContext per_query_ctx;
106  MemoryContext oldcontext;
107  XLogRecPtr end_of_wal;
108  XLogRecPtr wait_for_wal_lsn;
110  ResourceOwner old_resowner = CurrentResourceOwner;
111  ArrayType *arr;
112  Size ndim;
113  List *options = NIL;
115 
117 
119 
120  if (PG_ARGISNULL(0))
121  ereport(ERROR,
122  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
123  errmsg("slot name must not be null")));
124  name = PG_GETARG_NAME(0);
125 
126  if (PG_ARGISNULL(1))
127  upto_lsn = InvalidXLogRecPtr;
128  else
129  upto_lsn = PG_GETARG_LSN(1);
130 
131  if (PG_ARGISNULL(2))
132  upto_nchanges = InvalidXLogRecPtr;
133  else
134  upto_nchanges = PG_GETARG_INT32(2);
135 
136  if (PG_ARGISNULL(3))
137  ereport(ERROR,
138  (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
139  errmsg("options array must not be null")));
140  arr = PG_GETARG_ARRAYTYPE_P(3);
141 
142  /* state to write output to */
143  p = palloc0(sizeof(DecodingOutputState));
144 
145  p->binary_output = binary;
146 
147  per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
148  oldcontext = MemoryContextSwitchTo(per_query_ctx);
149 
150  /* Deconstruct options array */
151  ndim = ARR_NDIM(arr);
152  if (ndim > 1)
153  {
154  ereport(ERROR,
155  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
156  errmsg("array must be one-dimensional")));
157  }
158  else if (array_contains_nulls(arr))
159  {
160  ereport(ERROR,
161  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
162  errmsg("array must not contain nulls")));
163  }
164  else if (ndim == 1)
165  {
166  int nelems;
167  Datum *datum_opts;
168  int i;
169 
170  Assert(ARR_ELEMTYPE(arr) == TEXTOID);
171 
172  deconstruct_array_builtin(arr, TEXTOID, &datum_opts, NULL, &nelems);
173 
174  if (nelems % 2 != 0)
175  ereport(ERROR,
176  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
177  errmsg("array must have even number of elements")));
178 
179  for (i = 0; i < nelems; i += 2)
180  {
181  char *optname = TextDatumGetCString(datum_opts[i]);
182  char *opt = TextDatumGetCString(datum_opts[i + 1]);
183 
184  options = lappend(options, makeDefElem(optname, (Node *) makeString(opt), -1));
185  }
186  }
187 
188  InitMaterializedSRF(fcinfo, 0);
189  p->tupstore = rsinfo->setResult;
190  p->tupdesc = rsinfo->setDesc;
191 
192  /*
193  * Compute the current end-of-wal.
194  */
195  if (!RecoveryInProgress())
196  end_of_wal = GetFlushRecPtr(NULL);
197  else
198  end_of_wal = GetXLogReplayRecPtr(NULL);
199 
201 
202  PG_TRY();
203  {
204  /* restart at slot's confirmed_flush */
206  options,
207  false,
208  XL_ROUTINE(.page_read = read_local_xlog_page,
209  .segment_open = wal_segment_open,
210  .segment_close = wal_segment_close),
212  LogicalOutputWrite, NULL);
213 
214  MemoryContextSwitchTo(oldcontext);
215 
216  /*
217  * Check whether the output plugin writes textual output if that's
218  * what we need.
219  */
220  if (!binary &&
221  ctx->options.output_type !=OUTPUT_PLUGIN_TEXTUAL_OUTPUT)
222  ereport(ERROR,
223  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
224  errmsg("logical decoding output plugin \"%s\" produces binary output, but function \"%s\" expects textual data",
226  format_procedure(fcinfo->flinfo->fn_oid))));
227 
228  /*
229  * Wait for specified streaming replication standby servers (if any)
230  * to confirm receipt of WAL up to wait_for_wal_lsn.
231  */
232  if (XLogRecPtrIsInvalid(upto_lsn))
233  wait_for_wal_lsn = end_of_wal;
234  else
235  wait_for_wal_lsn = Min(upto_lsn, end_of_wal);
236 
237  WaitForStandbyConfirmation(wait_for_wal_lsn);
238 
239  ctx->output_writer_private = p;
240 
241  /*
242  * Decoding of WAL must start at restart_lsn so that the entirety of
243  * xacts that committed after the slot's confirmed_flush can be
244  * accumulated into reorder buffers.
245  */
247 
248  /* invalidate non-timetravel entries */
250 
251  /* Decode until we run out of records */
252  while (ctx->reader->EndRecPtr < end_of_wal)
253  {
254  XLogRecord *record;
255  char *errm = NULL;
256 
257  record = XLogReadRecord(ctx->reader, &errm);
258  if (errm)
259  elog(ERROR, "could not find record for logical decoding: %s", errm);
260 
261  /*
262  * The {begin_txn,change,commit_txn}_wrapper callbacks above will
263  * store the description into our tuplestore.
264  */
265  if (record != NULL)
266  LogicalDecodingProcessRecord(ctx, ctx->reader);
267 
268  /* check limits */
269  if (upto_lsn != InvalidXLogRecPtr &&
270  upto_lsn <= ctx->reader->EndRecPtr)
271  break;
272  if (upto_nchanges != 0 &&
273  upto_nchanges <= p->returned_rows)
274  break;
276  }
277 
278  /*
279  * Logical decoding could have clobbered CurrentResourceOwner during
280  * transaction management, so restore the executor's value. (This is
281  * a kluge, but it's not worth cleaning up right now.)
282  */
283  CurrentResourceOwner = old_resowner;
284 
285  /*
286  * Next time, start where we left off. (Hunting things, the family
287  * business..)
288  */
289  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr && confirm)
290  {
291  LogicalConfirmReceivedLocation(ctx->reader->EndRecPtr);
292 
293  /*
294  * If only the confirmed_flush_lsn has changed the slot won't get
295  * marked as dirty by the above. Callers on the walsender
296  * interface are expected to keep track of their own progress and
297  * don't need it written out. But SQL-interface users cannot
298  * specify their own start positions and it's harder for them to
299  * keep track of their progress, so we should make more of an
300  * effort to save it for them.
301  *
302  * Dirty the slot so it's written out at the next checkpoint.
303  * We'll still lose its position on crash, as documented, but it's
304  * better than always losing the position even on clean restart.
305  */
307  }
308 
309  /* free context, call shutdown callback */
310  FreeDecodingContext(ctx);
311 
314  }
315  PG_CATCH();
316  {
317  /* clear all timetravel entries */
319 
320  PG_RE_THROW();
321  }
322  PG_END_TRY();
323 
324  return (Datum) 0;
325 }
326 
327 /*
328  * SQL function returning the changestream as text, consuming the data.
329  */
330 Datum
332 {
333  return pg_logical_slot_get_changes_guts(fcinfo, true, false);
334 }
335 
336 /*
337  * SQL function returning the changestream as text, only peeking ahead.
338  */
339 Datum
341 {
342  return pg_logical_slot_get_changes_guts(fcinfo, false, false);
343 }
344 
345 /*
346  * SQL function returning the changestream in binary, consuming the data.
347  */
348 Datum
350 {
351  return pg_logical_slot_get_changes_guts(fcinfo, true, true);
352 }
353 
354 /*
355  * SQL function returning the changestream in binary, only peeking ahead.
356  */
357 Datum
359 {
360  return pg_logical_slot_get_changes_guts(fcinfo, false, true);
361 }
362 
363 
364 /*
365  * SQL function for writing logical decoding message into WAL.
366  */
367 Datum
369 {
370  bool transactional = PG_GETARG_BOOL(0);
371  char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(1));
373  bool flush = PG_GETARG_BOOL(3);
374  XLogRecPtr lsn;
375 
377  transactional, flush);
378  PG_RETURN_LSN(lsn);
379 }
380 
381 Datum
383 {
384  /* bytea and text are compatible */
385  return pg_logical_emit_message_bytea(fcinfo);
386 }
#define ARR_NDIM(a)
Definition: array.h:290
#define PG_GETARG_ARRAYTYPE_P(n)
Definition: array.h:263
#define ARR_ELEMTYPE(a)
Definition: array.h:292
bool array_contains_nulls(ArrayType *array)
Definition: arrayfuncs.c:3748
void deconstruct_array_builtin(ArrayType *array, Oid elmtype, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3678
static Datum values[MAXATTR]
Definition: bootstrap.c:152
#define TextDatumGetCString(d)
Definition: builtins.h:98
#define NameStr(name)
Definition: c.h:746
#define Min(x, y)
Definition: c.h:1004
signed int int32
Definition: c.h:494
#define VARHDRSZ
Definition: c.h:692
#define Assert(condition)
Definition: c.h:858
uint32 TransactionId
Definition: c.h:652
size_t Size
Definition: c.h:605
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:88
int errcode(int sqlerrcode)
Definition: elog.c:857
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define PG_RE_THROW()
Definition: elog.h:411
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define ERROR
Definition: elog.h:39
#define PG_CATCH(...)
Definition: elog.h:380
#define elog(elevel,...)
Definition: elog.h:224
#define ereport(elevel,...)
Definition: elog.h:149
#define PG_GETARG_BYTEA_PP(n)
Definition: fmgr.h:308
#define PG_GETARG_TEXT_PP(n)
Definition: fmgr.h:309
#define PG_ARGISNULL(n)
Definition: fmgr.h:209
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define PG_GETARG_INT32(n)
Definition: fmgr.h:269
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
void InvalidateSystemCaches(void)
Definition: inval.c:792
int i
Definition: isn.c:73
List * lappend(List *list, void *datum)
Definition: list.c:339
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1834
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:693
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:109
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:495
static void LogicalOutputPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:52
Datum pg_logical_slot_get_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:331
static void LogicalOutputWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
Definition: logicalfuncs.c:62
Datum pg_logical_emit_message_bytea(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:368
Datum pg_logical_slot_get_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:349
Datum pg_logical_emit_message_text(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:382
Datum pg_logical_slot_peek_binary_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:358
Datum pg_logical_slot_peek_changes(PG_FUNCTION_ARGS)
Definition: logicalfuncs.c:340
struct DecodingOutputState DecodingOutputState
static Datum pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool binary)
Definition: logicalfuncs.c:99
DefElem * makeDefElem(char *name, Node *arg, int location)
Definition: makefuncs.c:564
int GetDatabaseEncoding(void)
Definition: mbutils.c:1261
bool pg_verify_mbstr(int encoding, const char *mbstr, int len, bool noError)
Definition: mbutils.c:1566
void * palloc0(Size size)
Definition: mcxt.c:1346
#define MaxAllocSize
Definition: memutils.h:40
XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional, bool flush)
Definition: message.c:43
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
@ OUTPUT_PLUGIN_TEXTUAL_OUTPUT
Definition: output_plugin.h:20
const void * data
#define NIL
Definition: pg_list.h:68
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:33
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
#define PG_RETURN_LSN(x)
Definition: pg_lsn.h:34
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
static Datum TransactionIdGetDatum(TransactionId X)
Definition: postgres.h:272
uintptr_t Datum
Definition: postgres.h:64
MemoryContextSwitchTo(old_ctx)
char * format_procedure(Oid procedure_oid)
Definition: regproc.c:299
ResourceOwner CurrentResourceOwner
Definition: resowner.c:165
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1010
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:540
ReplicationSlot * MyReplicationSlot
Definition: slot.c:138
void CheckSlotPermissions(void)
Definition: slot.c:1384
void ReplicationSlotRelease(void)
Definition: slot.c:652
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
Definition: slot.c:2740
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:78
Tuplestorestate * tupstore
Definition: logicalfuncs.c:42
MemoryContext ecxt_per_query_memory
Definition: execnodes.h:262
Oid fn_oid
Definition: fmgr.h:59
fmNodePtr resultinfo
Definition: fmgr.h:89
FmgrInfo * flinfo
Definition: fmgr.h:87
Definition: pg_list.h:54
void * output_writer_private
Definition: logical.h:81
StringInfo out
Definition: logical.h:71
Definition: nodes.h:129
XLogRecPtr restart_lsn
Definition: slot.h:93
ReplicationSlotPersistentData data
Definition: slot.h:178
ExprContext * econtext
Definition: execnodes.h:332
TupleDesc setDesc
Definition: execnodes.h:340
Tuplestorestate * setResult
Definition: execnodes.h:339
Definition: c.h:741
Definition: c.h:687
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
Definition: tuplestore.c:750
String * makeString(char *str)
Definition: value.c:63
#define VARDATA_ANY(PTR)
Definition: varatt.h:324
#define VARSIZE_ANY_EXHDR(PTR)
Definition: varatt.h:317
char * text_to_cstring(const text *t)
Definition: varlena.c:217
text * cstring_to_text_with_len(const char *s, int len)
Definition: varlena.c:196
const char * name
bool RecoveryInProgress(void)
Definition: xlog.c:6290
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6455
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:389
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:231
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:842
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:817
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:861